lnbits-legend/lnbits/core/models.py

import json
import hashlib
from ecdsa import SECP256k1, SigningKey  # type: ignore
from typing import List, NamedTuple, Optional, Dict
from sqlite3 import Row

from lnbits.settings import WALLET


class User(NamedTuple):
    id: str
    email: str
    extensions: List[str] = []
    wallets: List["Wallet"] = []
    password: Optional[str] = None

    @property
    def wallet_ids(self) -> List[str]:
        return [wallet.id for wallet in self.wallets]

    def get_wallet(self, wallet_id: str) -> Optional["Wallet"]:
        w = [wallet for wallet in self.wallets if wallet.id == wallet_id]
        return w[0] if w else None


class Wallet(NamedTuple):
    id: str
    name: str
    user: str
    adminkey: str
    inkey: str
    balance_msat: int

    @property
    def balance(self) -> int:
        return self.balance_msat // 1000

    @property
    def lnurlauth_key(self) -> SigningKey:
        return SigningKey.from_string(
            hashlib.sha256(self.id.encode("utf-8")).digest(),
            curve=SECP256k1,
            hashfunc=hashlib.sha256,
        )

    def get_payment(self, payment_hash: str) -> Optional["Payment"]:
        from .crud import get_wallet_payment

        return get_wallet_payment(self.id, payment_hash)

    def get_payments(
        self,
        *,
        complete: bool = True,
        pending: bool = False,
        outgoing: bool = True,
        incoming: bool = True,
        exclude_uncheckable: bool = False
    ) -> List["Payment"]:
        from .crud import get_wallet_payments

        return get_wallet_payments(
            self.id,
            complete=complete,
            pending=pending,
            outgoing=outgoing,
            incoming=incoming,
            exclude_uncheckable=exclude_uncheckable,
        )


class Payment(NamedTuple):
    checking_id: str
    pending: bool
    amount: int
    fee: int
    memo: str
    time: int
    bolt11: str
    preimage: str
    payment_hash: str
    extra: Dict
    wallet_id: str

    @classmethod
    def from_row(cls, row: Row):
        return cls(
            checking_id=row["checking_id"],
            payment_hash=row["hash"] or "0" * 64,
            bolt11=row["bolt11"] or "",
            preimage=row["preimage"] or "0" * 64,
            extra=json.loads(row["extra"] or "{}"),
            pending=row["pending"],
            amount=row["amount"],
            fee=row["fee"],
            memo=row["memo"],
            time=row["time"],
            wallet_id=row["wallet"],
        )

    @property
    def tag(self) -> Optional[str]:
        return self.extra.get("tag")

    @property
    def msat(self) -> int:
        return self.amount

    @property
    def sat(self) -> int:
        return self.amount // 1000

    @property
    def is_in(self) -> bool:
        return self.amount > 0

    @property
    def is_out(self) -> bool:
        return self.amount < 0

    @property
    def is_uncheckable(self) -> bool:
        return self.checking_id.startswith("temp_") or self.checking_id.startswith("internal_")

    def set_pending(self, pending: bool) -> None:
        from .crud import update_payment_status

        update_payment_status(self.checking_id, pending)

    def check_pending(self) -> None:
        if self.is_uncheckable:
            return

        if self.is_out:
            pending = WALLET.get_payment_status(self.checking_id)
        else:
            pending = WALLET.get_invoice_status(self.checking_id)

        self.set_pending(pending.pending)

    def delete(self) -> None:
        from .crud import delete_payment

        delete_payment(self.checking_id)