lnbits-legend/lnbits/extensions/nostrnip5/crud.py

from typing import List, Optional, Union

from lnbits.helpers import urlsafe_short_hash

from . import db
from .models import (
    CreateDomainData,
    Domain,
    Address,
    CreateAddressData,
)


async def get_domain(domain_id: str) -> Optional[Domain]:
    row = await db.fetchone(
        "SELECT * FROM nostrnip5.domains WHERE id = ?", (domain_id,)
    )
    return Domain.from_row(row) if row else None

async def get_domain_by_name(domain: str) -> Optional[Domain]:
    row = await db.fetchone(
        "SELECT * FROM nostrnip5.domains WHERE domain = ?", (domain,)
    )
    return Domain.from_row(row) if row else None

async def get_domains(wallet_ids: Union[str, List[str]]) -> List[Domain]:
    if isinstance(wallet_ids, str):
        wallet_ids = [wallet_ids]

    q = ",".join(["?"] * len(wallet_ids))
    rows = await db.fetchall(
        f"SELECT * FROM nostrnip5.domains WHERE wallet IN ({q})", (*wallet_ids,)
    )

    return [Domain.from_row(row) for row in rows]

async def get_address(domain_id: str, address_id: str) -> Optional[Address]:
    row = await db.fetchone(
        "SELECT * FROM nostrnip5.addresses WHERE domain_id = ? AND id = ?", (domain_id,address_id,)
    )
    return Address.from_row(row) if row else None

async def get_address_by_local_part(domain_id: str, local_part: str) -> Optional[Address]:
    row = await db.fetchone(
        "SELECT * FROM nostrnip5.addresses WHERE domain_id = ? AND local_part = ?", (domain_id,local_part,)
    )
    return Address.from_row(row) if row else None

async def get_addresses(domain_id: str) -> List[Address]:
    rows = await db.fetchall(
        f"SELECT * FROM nostrnip5.addresses WHERE domain_id = ?", (domain_id,)
    )

    return [Address.from_row(row) for row in rows]

async def get_all_addresses(wallet_ids: Union[str, List[str]]) -> List[Address]:
    if isinstance(wallet_ids, str):
        wallet_ids = [wallet_ids]

    q = ",".join(["?"] * len(wallet_ids))
    rows = await db.fetchall(
        f"""
        SELECT a.* 
        FROM nostrnip5.addresses a
        JOIN nostrnip5.domains d ON d.id = a.domain_id
        WHERE d.wallet IN ({q})
        """, 
        (*wallet_ids,)
    )

    return [Address.from_row(row) for row in rows]

async def activate_domain(domain_id: str, address_id: str) -> Address:
    await db.execute(
        """
        UPDATE nostrnip5.addresses
        SET active = true
        WHERE domain_id = ?
        AND id = ?
        """,
        (
            domain_id,
            address_id,
        ),
    )

    address = await get_address(domain_id, address_id)
    assert address, "Newly updated address couldn't be retrieved"
    return address

async def delete_domain(domain_id) -> bool:
    await db.execute(
        """
        DELETE FROM nostrnip5.addresses WHERE domain_id = ?
        """,
        (
            domain_id,
        ),
    )

    await db.execute(
        """
        DELETE FROM nostrnip5.domains WHERE id = ?
        """,
        (
            domain_id,
        ),
    )

    return True

async def delete_address(address_id) -> bool:
    await db.execute(
        """
        DELETE FROM nostrnip5.addresses WHERE id = ?
        """,
        (
            address_id,
        ),
    )

async def create_address_internal(domain_id: str, data: CreateAddressData) -> Address:
    address_id = urlsafe_short_hash()

    await db.execute(
        """
        INSERT INTO nostrnip5.addresses (id, domain_id, local_part, pubkey, active)
        VALUES (?, ?, ?, ?, ?)
        """,
        (
            address_id,
            domain_id,
            data.local_part,
            data.pubkey,
            False,
        ),
    )

    address = await get_address(domain_id, address_id)
    assert address, "Newly created address couldn't be retrieved"
    return address

async def create_domain_internal(wallet_id: str, data: CreateDomainData) -> Domain:
    domain_id = urlsafe_short_hash()

    await db.execute(
        """
        INSERT INTO nostrnip5.domains (id, wallet, currency, amount, domain)
        VALUES (?, ?, ?, ?, ?)
        """,
        (
            domain_id,
            wallet_id,
            data.currency,
            int(data.amount * 100),
            data.domain
        ),
    )

    domain = await get_domain(domain_id)
    assert domain, "Newly created domain couldn't be retrieved"
    return domain