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

import os
import random
from binascii import hexlify, unhexlify
from typing import List, Optional, Union

from embit import bip32, bip39, ec, script
from embit.networks import NETWORKS
from loguru import logger

from lnbits.helpers import urlsafe_short_hash

from . import db
from .core.base import Invoice
from .models import Cashu, Pegs, Promises, Proof


async def create_cashu(wallet_id: str, data: Cashu) -> Cashu:
    cashu_id = urlsafe_short_hash()

    entropy = bytes([random.getrandbits(8) for i in range(16)])
    mnemonic = bip39.mnemonic_from_bytes(entropy)
    seed = bip39.mnemonic_to_seed(mnemonic)
    root = bip32.HDKey.from_seed(seed, version=NETWORKS["main"]["xprv"])

    bip44_xprv = root.derive("m/44h/1h/0h")
    bip44_xpub = bip44_xprv.to_public()

    await db.execute(
        """
        INSERT INTO cashu.cashu (id, wallet, name, tickershort, fraction, maxsats, coins, prvkey, pubkey)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            cashu_id,
            wallet_id,
            data.name,
            data.tickershort,
            data.fraction,
            data.maxsats,
            data.coins,
            bip44_xprv.to_base58(),
            bip44_xpub.to_base58(),
        ),
    )

    cashu = await get_cashu(cashu_id)
    assert cashu, "Newly created cashu couldn't be retrieved"
    return cashu


async def update_cashu_keys(cashu_id, wif: str = None) -> Optional[Cashu]:
    entropy = bytes([random.getrandbits(8) for i in range(16)])
    mnemonic = bip39.mnemonic_from_bytes(entropy)
    seed = bip39.mnemonic_to_seed(mnemonic)
    root = bip32.HDKey.from_seed(seed, version=NETWORKS["main"]["xprv"])

    bip44_xprv = root.derive("m/44h/1h/0h")
    bip44_xpub = bip44_xprv.to_public()

    await db.execute(
        "UPDATE cashu.cashu SET prv = ?, pub = ? WHERE id = ?",
        bip44_xprv.to_base58(),
        bip44_xpub.to_base58(),
        cashu_id,
    )
    row = await db.fetchone("SELECT * FROM cashu.cashu WHERE id = ?", (cashu_id,))
    return Cashu(**row) if row else None


async def get_cashu(cashu_id) -> Optional[Cashu]:
    row = await db.fetchone("SELECT * FROM cashu.cashu WHERE id = ?", (cashu_id,))
    return Cashu(**row) if row else None


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

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

    return [Cashu(**row) for row in rows]


async def delete_cashu(cashu_id) -> None:
    await db.execute("DELETE FROM cashu.cashu WHERE id = ?", (cashu_id,))


##########################################
###############MINT STUFF#################
##########################################


async def store_promises(
    amounts: List[int], B_s: List[str], C_s: List[str], cashu_id: str
):
    for amount, B_, C_ in zip(amounts, B_s, C_s):
        await store_promise(amount, B_, C_, cashu_id)


async def store_promise(amount: int, B_: str, C_: str, cashu_id: str):
    promise_id = urlsafe_short_hash()

    await db.execute(
        """
        INSERT INTO cashu.promises
          (id, amount, B_b, C_b, cashu_id)
        VALUES (?, ?, ?, ?, ?)
        """,
        (promise_id, amount, str(B_), str(C_), cashu_id),
    )


async def get_promises(cashu_id) -> Optional[Cashu]:
    row = await db.fetchall(
        "SELECT * FROM cashu.promises WHERE cashu_id = ?", (cashu_id,)
    )
    return Promises(**row) if row else None


async def get_proofs_used(cashu_id):
    rows = await db.fetchall(
        "SELECT secret from cashu.proofs_used WHERE cashu_id = ?", (cashu_id,)
    )
    return [row[0] for row in rows]


async def invalidate_proof(cashu_id: str, proof: Proof):
    invalidate_proof_id = urlsafe_short_hash()
    await db.execute(
        """
        INSERT INTO cashu.proofs_used
          (id, amount, C, secret, cashu_id)
        VALUES (?, ?, ?, ?, ?)
        """,
        (invalidate_proof_id, proof.amount, str(proof.C), str(proof.secret), cashu_id),
    )


########################################
############ MINT INVOICES #############
########################################


async def store_lightning_invoice(cashu_id: str, invoice: Invoice):
    await db.execute(
        """
        INSERT INTO cashu.invoices
          (cashu_id, amount, pr, hash, issued)
        VALUES (?, ?, ?, ?, ?)
        """,
        (
            cashu_id,
            invoice.amount,
            invoice.pr,
            invoice.hash,
            invoice.issued,
        ),
    )


async def get_lightning_invoice(cashu_id: str, hash: str):
    row = await db.fetchone(
        """
        SELECT * from cashu.invoices
        WHERE cashu_id =? AND hash = ?
        """,
        (
            cashu_id,
            hash,
        ),
    )
    return Invoice.from_row(row)


async def update_lightning_invoice(cashu_id: str, hash: str, issued: bool):
    await db.execute(
        "UPDATE cashu.invoices SET issued = ? WHERE cashu_id = ? AND hash = ?",
        (
            issued,
            cashu_id,
            hash,
        ),
    )