withdraw/transport_rpcs.py

"""
Nostr-transport RPC handlers for the withdraw (LNURL-withdraw) extension.

Names mirror the Lightning.Pub `withdraw.*` contract that the lamassu-next
ATM consumes (see ~/dev/shocknet/lamassu-next/packages/lightning/src/client.ts
lines ~301–351). That keeps the lamassu-next-side adapter a pure name
translation — no semantic reshaping.

Auth model (set in `__init__.py:withdraw_start`):
- create / get / update / delete  → AUTH_WALLET; the calling pubkey must
  own the wallet the link is scoped to. *_get / *_update / *_delete also
  verify the link's stored `wallet` matches the caller's wallet id.

`resolve_withdraw_owner` is registered with the core subscription module
under tag `"withdraw"` and extras-key `"withdrawal_link_id"` (matching
where the extension stamps the link id on settlement — see
`views_lnurl.py:144`). That lets `subscribe_payments({tag:"withdraw",
link_id:...})` enforce ownership without core importing this module.
"""

from __future__ import annotations

from lnbits.core.crud.wallets import get_wallets
from lnbits.core.models import Account
from lnbits.core.models.wallets import WalletTypeInfo
from lnbits.core.services.nostr_transport.models import NostrRpcRequest
from shortuuid import uuid

from .crud import (
    create_withdraw_link,
    delete_withdraw_link,
    get_withdraw_link,
    get_withdraw_links,
    update_withdraw_link,
)
from .helpers import create_lnurl_from_baseurl
from .models import CreateWithdrawData, WithdrawLink


async def handle_lnurlw_create_link(
    auth: WalletTypeInfo, request: NostrRpcRequest
) -> dict:
    body = request.body or {}
    data = CreateWithdrawData(**body)
    link = await create_withdraw_link(data, auth.wallet.id)
    return _to_dict(_populate_lnurl(link))


async def handle_lnurlw_get_link(
    auth: WalletTypeInfo, request: NostrRpcRequest
) -> dict:
    link_id = _require_id(request)
    link = await _require_owned_link(link_id, auth.wallet.id)
    return _to_dict(_populate_lnurl(link))


async def handle_lnurlw_update_link(
    auth: WalletTypeInfo, request: NostrRpcRequest
) -> dict:
    link_id = _require_id(request)
    link = await _require_owned_link(link_id, auth.wallet.id)
    body = request.body or {}
    _MUTABLE = {
        "title",
        "min_withdrawable",
        "max_withdrawable",
        "uses",
        "wait_time",
        "is_unique",
        "webhook_url",
        "webhook_headers",
        "webhook_body",
        "custom_url",
        "enabled",
    }
    for k, v in body.items():
        if k in _MUTABLE:
            setattr(link, k, v)
    updated = await update_withdraw_link(link)
    return _to_dict(_populate_lnurl(updated))


async def handle_lnurlw_delete_link(
    auth: WalletTypeInfo, request: NostrRpcRequest
) -> dict:
    link_id = _require_id(request)
    await _require_owned_link(link_id, auth.wallet.id)
    await delete_withdraw_link(link_id)
    return {"ok": True}


async def handle_lnurlw_list_links(auth: Account, request: NostrRpcRequest) -> dict:
    """List withdraw links across all wallets owned by the calling account.
    Useful for ATMs to re-discover their links after a reconnect.

    Body fields:
      - limit: int (0 means no limit; default 0)
      - offset: int (default 0)
    If `request.wallet_id` is set and is one of the caller's wallets,
    narrow to just that wallet.
    """
    body = request.body or {}
    limit = int(body.get("limit") or 0)
    offset = int(body.get("offset") or 0)

    wallets = await get_wallets(auth.id)
    wallet_ids = [w.id for w in wallets]
    if not wallet_ids:
        return {"data": [], "total": 0}
    if request.wallet_id and request.wallet_id in wallet_ids:
        wallet_ids = [request.wallet_id]

    page = await get_withdraw_links(wallet_ids, limit, offset)
    return {
        "data": [_to_dict(_populate_lnurl(link)) for link in page.data],
        "total": page.total,
    }


async def handle_lnurlw_unique_hashes(
    auth: WalletTypeInfo, request: NostrRpcRequest
) -> dict:
    """
    For a `is_unique=True` link, return the per-use `id_unique_hash`
    values that the ATM uses to generate distinct QR codes — one per
    unredeemed slot. Mirrors the formula in `helpers.py:create_lnurl`
    exactly so an ATM never has to re-implement the derivation:

        id_unique_hash = shortuuid.uuid(name=link.id + link.unique_hash + index)

    `link.usescsv` is the canonical list of *unredeemed* slot indexes;
    after a customer claims a slot it gets removed there (see
    `crud.remove_unique_withdraw_link`). The hashes returned here are
    therefore exactly the ones still claimable.

    Response:
      {
        "link_id": str,
        "unique_hash": str,            # base hash
        "is_unique": bool,
        "unredeemed_hashes": [         # one entry per remaining slot
          {"index": str, "id_unique_hash": str}, ...
        ]
      }

    For `is_unique=False` links the list is empty and `unique_hash`
    alone identifies the callback path
    (`/withdraw/api/v1/lnurl/<unique_hash>`). For `is_unique=True`
    each callback path is
    `/withdraw/api/v1/lnurl/<unique_hash>/<id_unique_hash>`.
    """
    link_id = _require_id(request)
    link = await _require_owned_link(link_id, auth.wallet.id)

    unredeemed = []
    if link.is_unique:
        # usescsv is comma-separated; split and skip empties (after the
        # last slot is consumed it becomes the empty string).
        for index_str in [s for s in link.usescsv.split(",") if s.strip()]:
            tohash = link.id + link.unique_hash + index_str
            unredeemed.append(
                {
                    "index": index_str.strip(),
                    "id_unique_hash": uuid(name=tohash),
                }
            )

    return {
        "link_id": link.id,
        "unique_hash": link.unique_hash,
        "is_unique": link.is_unique,
        "unredeemed_hashes": unredeemed,
    }


async def resolve_withdraw_owner(link_id: str) -> str | None:
    """For the core subscription module: link_id -> wallet_id (or None)."""
    link = await get_withdraw_link(link_id)
    return link.wallet if link else None


# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------


def _require_id(request: NostrRpcRequest) -> str:
    body = request.body or {}
    link_id = body.get("id")
    if not link_id:
        raise ValueError("withdraw: body.id is required")
    return str(link_id)


async def _require_owned_link(link_id: str, wallet_id: str):
    link = await get_withdraw_link(link_id)
    if link is None:
        raise ValueError(f"withdraw: link not found: {link_id}")
    if link.wallet != wallet_id:
        raise PermissionError("withdraw: link does not belong to caller's wallet")
    return link


def _populate_lnurl(link: WithdrawLink) -> WithdrawLink:
    """
    Compose `lnurl` / `lnurl_url` from `settings.lnbits_baseurl` so
    nostr-transport responses match the HTTP `views_api` shape, where
    these fields are populated from `request.url_for(...)`. Without
    this, consumers (ATMs, etc.) would have to re-derive the callback
    URL themselves from a separately-provisioned LNbits HTTPS URL —
    duplicating state LNbits already knows. See aiolabs/withdraw#1.
    """
    try:
        encoded = create_lnurl_from_baseurl(link)
        link.lnurl = str(encoded.bech32)
        link.lnurl_url = str(encoded.url)
    except ValueError:
        pass
    return link


def _to_dict(link) -> dict:
    import json

    return json.loads(link.json())