WIP nostr nip5 extension

2022-12-20 06:28:43 -06:00
parent c5fdd35078
commit 3ca9edeee6
13 changed files with 1409 additions and 0 deletions
@@ -0,0 +1,162 @@
+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