From 423e2cba73ea76207ad2c070a7ca4b10e6088766 Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Sun, 26 Jul 2026 02:30:18 +0200 Subject: [PATCH] ppq-auto-topup --- routstr/core/admin.py | 162 +++- routstr/upstream/auto_topup.py | 869 ++++++++++++++++-- routstr/upstream/ppqai.py | 2 +- routstr/wallet.py | 204 +++- .../integration/test_ppq_auto_topup_claim.py | 516 +++++++++++ tests/unit/test_auto_topup.py | 363 +++++++- tests/unit/test_wallet.py | 180 +++- ui/components/provider-card.tsx | 132 ++- ui/components/provider-form-fields.tsx | 16 + .../providers/PPQAutoTopupSettings.tsx | 136 +++ ui/lib/api/services/admin.ts | 50 + 11 files changed, 2536 insertions(+), 94 deletions(-) create mode 100644 tests/integration/test_ppq_auto_topup_claim.py create mode 100644 ui/components/providers/PPQAutoTopupSettings.tsx diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 66a1d288..d8e259c0 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -863,6 +863,32 @@ class UpstreamProviderUpdateBySlug(BaseModel): provider_settings: dict | None = None +async def _active_ppq_claim_in_session( + session: AsyncSession, provider_id: int +) -> bool: + """Check for an active claim inside the caller's transaction. + + Must share the transaction of whatever destructive write it is guarding — + a check in its own session leaves a window for a worker to create the + claim between the check and the commit. + """ + from ..upstream.auto_topup import _ppq_state_id_for_provider + + claim = await session.get( + CashuTransaction, _ppq_state_id_for_provider(provider_id) + ) + return claim is not None and not claim.collected and not claim.swept + + +def _require_valid_auto_topup(provider_type: str, settings: dict | None) -> None: + """Reject auto top-up settings the worker would later refuse to act on.""" + from ..upstream.auto_topup import validate_auto_topup_settings + + problem = validate_auto_topup_settings(provider_type, settings) + if problem is not None: + raise HTTPException(status_code=400, detail=problem) + + async def _apply_provider_update( session: AsyncSession, provider: UpstreamProviderRow, @@ -874,6 +900,23 @@ async def _apply_provider_update( await _ensure_unique_slug(session, validated, exclude_id=provider.id) provider.slug = validated + if ( + payload.provider_type is not None + and payload.provider_type != provider.provider_type + and provider.provider_type == "ppqai" + and provider.id is not None + and await _active_ppq_claim_in_session(session, provider.id) + ): + # Changing the type would orphan the claim: the PPQ endpoints refuse + # non-ppqai providers, so nobody could ever inspect or release it. + raise HTTPException( + status_code=409, + detail=( + "This provider has an active PPQ auto top-up claim. Release " + "it before changing the provider type" + ), + ) + if payload.provider_type is not None: provider.provider_type = payload.provider_type if payload.base_url is not None: @@ -886,6 +929,22 @@ async def _apply_provider_update( provider.enabled = payload.enabled if payload.provider_fee is not None: provider.provider_fee = payload.provider_fee + + # Validate against the effective type and effective settings: a type + # change without new settings must not leave stored settings that the + # worker will refuse, and new settings must fit the new type. + effective_settings = payload.provider_settings + if effective_settings is None and payload.provider_type is not None: + try: + effective_settings = ( + json.loads(provider.provider_settings) + if provider.provider_settings + else None + ) + except (json.JSONDecodeError, TypeError): + effective_settings = None + if effective_settings is not None: + _require_valid_auto_topup(provider.provider_type, effective_settings) if payload.provider_settings is not None: provider.provider_settings = json.dumps(payload.provider_settings) @@ -925,6 +984,8 @@ async def create_upstream_provider( else: slug = await allocate_unique_provider_slug(session, payload.provider_type) + _require_valid_auto_topup(payload.provider_type, payload.provider_settings) + provider = UpstreamProviderRow( slug=slug, provider_type=payload.provider_type, @@ -1013,6 +1074,25 @@ async def delete_upstream_provider(provider_id: str) -> dict[str, object]: async with create_session() as session: provider = await _get_upstream_provider_by_ref(session, provider_id) deleted_id = _provider_pk(provider) + + # Checked inside the delete transaction: the worker's claim creation + # re-reads the provider inside its own transaction, so these two + # writes serialise — either the claim lands first and this 409s, or + # the delete lands first and the worker refuses to claim. + if provider.provider_type == "ppqai" and await _active_ppq_claim_in_session( + session, deleted_id + ): + # Deleting now would orphan the claim and any funds it tracks: + # the PPQ endpoints 404 without the provider row, so the claim + # could never again be inspected or released. + raise HTTPException( + status_code=409, + detail=( + "This provider has an active PPQ auto top-up claim. " + "Resolve and release it before deleting the provider" + ), + ) + await session.delete(provider) await session.commit() await reinitialize_upstreams() @@ -1621,6 +1701,83 @@ async def get_log_dates_api(request: Request) -> dict[str, object]: return {"dates": dates} +_PPQ_RELEASE_ERRORS = { + "no_active_claim": "No active PPQ claim to release", + "stale_state": ( + "The claim changed since it was reviewed; reload and check again" + ), + "payment_in_flight": ( + "A Lightning payment is still in flight for this claim. Wait for it to " + "finish or expire before releasing" + ), + "claim_changed": ( + "The claim changed while the release was being applied; reload and " + "check again" + ), +} + + +class ReleasePPQAutoTopupRequest(BaseModel): + confirmed_safe_to_retry: bool + # Echoes the state_token the admin reviewed — the claim's full versioned + # state, not just its operation id. Any change since the review (a new + # attempt, a phase change, a renewed lease) fails the match, so the + # release cannot land on a state the admin never saw. + state_token: str | None = None + + +async def _require_ppq_provider(provider_id: int) -> UpstreamProviderRow: + async with create_session() as session: + provider = await session.get(UpstreamProviderRow, provider_id) + if provider is None: + raise HTTPException(status_code=404, detail="Provider not found") + if provider.provider_type != "ppqai": + raise HTTPException(status_code=400, detail="Provider is not PPQ") + return provider + + +@admin_router.get( + "/api/upstream-providers/{provider_id}/ppq-auto-topup", + dependencies=[Depends(require_admin_api)], +) +async def get_ppq_auto_topup_api(provider_id: int) -> dict[str, object]: + await _require_ppq_provider(provider_id) + from ..upstream.auto_topup import get_ppq_auto_topup_state + + return {"ok": True, **await get_ppq_auto_topup_state(provider_id)} + + +@admin_router.post( + "/api/upstream-providers/{provider_id}/ppq-auto-topup/release", + dependencies=[Depends(require_admin_api)], +) +async def release_ppq_auto_topup_api( + provider_id: int, payload: ReleasePPQAutoTopupRequest +) -> dict[str, object]: + await _require_ppq_provider(provider_id) + if not payload.confirmed_safe_to_retry: + raise HTTPException( + status_code=400, + detail="Confirm the Lightning payment outcome is safe before releasing", + ) + + from ..upstream.auto_topup import release_ppq_auto_topup_state + + outcome = await release_ppq_auto_topup_state( + provider_id, state_token=payload.state_token + ) + if not outcome.released: + raise HTTPException( + status_code=409, detail=_PPQ_RELEASE_ERRORS[outcome.reason] + ) + + logger.warning( + "Admin released PPQ auto top-up claim after manual reconciliation", + extra={"provider_id": provider_id, "state_token": payload.state_token}, + ) + return {"ok": True, "released": True} + + @admin_router.get("/api/transactions", dependencies=[Depends(require_admin_api)]) async def get_transactions_api( type: str | None = None, @@ -1633,7 +1790,10 @@ async def get_transactions_api( async with create_session() as session: from sqlmodel import col, func - base = select(CashuTransaction) + base = select(CashuTransaction).where( + (CashuTransaction.source != "ppq_auto_topup") + | (CashuTransaction.source == None) # noqa: E711 + ) if type: base = base.where(CashuTransaction.type == type) if source: diff --git a/routstr/upstream/auto_topup.py b/routstr/upstream/auto_topup.py index a88a28fb..7ca830ff 100644 --- a/routstr/upstream/auto_topup.py +++ b/routstr/upstream/auto_topup.py @@ -1,35 +1,50 @@ import asyncio import json +import math +import time +import typing +import uuid -from sqlmodel import select +from sqlalchemy.exc import IntegrityError +from sqlmodel import col, or_, select, update from ..core import get_logger -from ..core.db import ( - CashuTransaction, - UpstreamProviderRow, - create_session, -) +from ..core.db import CashuTransaction, UpstreamProviderRow, create_session from ..core.db import ( store_cashu_transaction_with_retry as store_cashu_transaction, ) -from ..wallet import send_token +from ..payment.price import sats_usd_price +from ..wallet import ( + Bolt11PaymentNotAttempted, + check_bolt11_payment_status, + execute_bolt11_payment, + prepare_bolt11_payment, + send_token, +) +from .ppqai import PPQAIUpstreamProvider from .routstr import RoutstrUpstreamProvider logger = get_logger(__name__) -# Check every 60 seconds AUTO_TOPUP_INTERVAL_SECONDS = 60 +# Claim lifecycle. "claimed" holds the slot while the invoice is being created +# and priced; nothing has been spent yet, so it is always safe to release. +# "in_flight" means proofs are committed to a mint and the outcome is unknown. +# "reconcile" means the worker gave up and an admin must decide. +PPQ_PHASE_CLAIMED = "claimed" +PPQ_PHASE_IN_FLIGHT = "in_flight" +PPQ_PHASE_RECONCILE = "reconcile" +PPQ_PHASES = frozenset({PPQ_PHASE_CLAIMED, PPQ_PHASE_IN_FLIGHT, PPQ_PHASE_RECONCILE}) +PPQ_SETTLEMENT_ATTEMPTS = 5 +PPQ_SETTLEMENT_POLL_SECONDS = 2 +PPQ_PENDING_TTL_SECONDS = 15 * 60 +PPQ_MAX_INVOICE_PREMIUM = 1.10 +PPQ_MIN_TOPUP_USD = 1 +PPQ_MAX_TOPUP_USD = 500 async def periodic_auto_topup() -> None: - """Background task that monitors Routstr provider balances and auto-tops up when below threshold. - - For each Routstr provider with auto_topup enabled in provider_settings: - 1. Checks the upstream balance via get_balance() - 2. If balance < topup_threshold, creates a cashu token from the configured mint - 3. Sends the token to the upstream provider via topup() - """ - # Wait for initial startup to complete + """Monitor enabled Routstr and PPQ providers and fund low balances.""" await asyncio.sleep(30) logger.info("Auto top-up worker started") @@ -41,19 +56,19 @@ async def periodic_auto_topup() -> None: "Auto top-up cycle failed", extra={"error": str(e), "error_type": type(e).__name__}, ) - await asyncio.sleep(AUTO_TOPUP_INTERVAL_SECONDS) async def _run_auto_topup_cycle() -> None: - """Single cycle: check all eligible providers and top up if needed.""" + """Check all eligible providers once, isolating failures by provider.""" + await _reconcile_all_ppq_claims() + async with create_session() as session: query = select(UpstreamProviderRow).where( - UpstreamProviderRow.provider_type == "routstr", + col(UpstreamProviderRow.provider_type).in_(["routstr", "ppqai"]), UpstreamProviderRow.enabled == True, # noqa: E712 ) - result = await session.exec(query) - providers = result.all() + providers = (await session.exec(query)).all() for row in providers: try: @@ -63,6 +78,7 @@ async def _run_auto_topup_cycle() -> None: "Auto top-up failed for provider", extra={ "provider_id": row.id, + "provider_type": row.provider_type, "base_url": row.base_url, "error": str(e), "error_type": type(e).__name__, @@ -70,62 +86,156 @@ async def _run_auto_topup_cycle() -> None: ) -async def _check_and_topup(row: UpstreamProviderRow) -> None: - """Check a single provider's balance and top up if below threshold.""" - # Parse provider settings +async def _reconcile_all_ppq_claims() -> None: + """Reconcile every active PPQ claim, ignoring top-up eligibility. + + A claim tracks money already committed, so it must keep reconciling even + after the provider is disabled, auto top-up is switched off, or the API + key is removed — none of which change what happened to the payment. Only + new top-ups depend on eligibility. + """ + async with create_session() as session: + rows = ( + await session.exec( + select(UpstreamProviderRow).where( + col(UpstreamProviderRow.provider_type) == "ppqai" + ) + ) + ).all() + + for row in rows: + try: + if row.id is None: + continue + state = await get_ppq_auto_topup_state(row.id) + if not state.get("active"): + continue + # Without an API key the PPQ side cannot be polled, but mint + # reconciliation still can — pass provider as None. + provider = ( + PPQAIUpstreamProvider.from_db_row(row) if row.api_key else None + ) + await _reconcile_ppq_state(row, provider) + except Exception as e: + logger.error( + "PPQ claim reconciliation failed", + extra={"provider_id": row.id, "error": str(e)}, + ) + + +def _invalid_number(value: object, *, integer: bool = False) -> bool: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return True + try: + # JSON accepts arbitrarily large integers; float() raises + # OverflowError past ~1e308 rather than returning inf. + as_float = float(value) + except OverflowError: + return True + if not math.isfinite(as_float) or value <= 0: + return True + return integer and not as_float.is_integer() + + +def validate_auto_topup_settings( + provider_type: str, settings: dict | None +) -> str | None: + """Return why these auto top-up settings are unusable, or None if they are. + + Shared by the admin write path and the worker so a configuration that the + UI accepted cannot silently fail to top up hours later. Settings with + auto top-up switched off are always valid. + """ + if not settings or not settings.get("auto_topup"): + return None + + if _invalid_number(settings.get("topup_threshold")): + return "Auto top-up threshold must be a positive number" + if _invalid_number(settings.get("topup_amount_limit"), integer=True): + return "Auto top-up amount must be a positive whole number" + + if provider_type == "routstr": + mint_url = settings.get("topup_mint_url") + if not isinstance(mint_url, str) or not mint_url.strip(): + return "Auto top-up requires a mint URL" + elif provider_type == "ppqai": + amount = int(settings["topup_amount_limit"]) + if not PPQ_MIN_TOPUP_USD <= amount <= PPQ_MAX_TOPUP_USD: + return ( + f"PPQ auto top-up amount must be between {PPQ_MIN_TOPUP_USD} " + f"and {PPQ_MAX_TOPUP_USD} USD" + ) + return None + + +def _get_auto_topup_settings(row: UpstreamProviderRow) -> dict | None: settings: dict = {} if row.provider_settings: try: settings = json.loads(row.provider_settings) except (json.JSONDecodeError, TypeError): - return + return None - if not settings.get("auto_topup"): + if not isinstance(settings, dict) or not settings.get("auto_topup"): + return None + + # Re-checked here even though the admin API validates on write: rows + # predate that check, and the database is not the only way in. + problem = validate_auto_topup_settings(row.provider_type, settings) + if problem is not None: + logger.warning( + "Auto top-up enabled but its configuration is invalid", + extra={"provider_id": row.id, "problem": problem}, + ) + return None + return settings + + +async def _check_and_topup(row: UpstreamProviderRow) -> None: + """Dispatch one provider row to its funding strategy.""" + settings = _get_auto_topup_settings(row) + if settings is None or not row.api_key: return - threshold = settings.get("topup_threshold") - amount = settings.get("topup_amount_limit") - mint_url = settings.get("topup_mint_url") + if row.provider_type == "routstr": + await _check_and_topup_routstr(row, settings) + elif row.provider_type == "ppqai": + await _check_and_topup_ppq(row, settings) - if not threshold or not amount or not mint_url: + +async def _check_and_topup_routstr( + row: UpstreamProviderRow, settings: dict +) -> None: + threshold = float(settings["topup_threshold"]) + amount = int(settings["topup_amount_limit"]) + mint_url = settings.get("topup_mint_url") + if not isinstance(mint_url, str) or not mint_url: logger.warning( - "Auto top-up enabled but missing configuration", - extra={ - "provider_id": row.id, - "has_threshold": bool(threshold), - "has_amount": bool(amount), - "has_mint": bool(mint_url), - }, + "Routstr auto top-up enabled but no mint is configured", + extra={"provider_id": row.id}, ) return - if not row.api_key: - return - - # Instantiate provider and check balance provider = RoutstrUpstreamProvider.from_db_row(row) if provider is None: return balance = await provider.get_balance() - if balance is None: logger.warning( - "Could not fetch balance for auto top-up", + "Could not fetch Routstr balance for auto top-up", extra={"provider_id": row.id, "base_url": row.base_url}, ) return - - if balance >= threshold * 1000: + if balance >= threshold: return - # Balance is below threshold - create token and top up logger.info( - "Auto top-up triggered", + "Routstr auto top-up triggered", extra={ "provider_id": row.id, - "balance": balance, - "threshold": threshold, - "topup_amount": amount, + "balance_sats": balance, + "threshold_sats": threshold, + "topup_sats": amount, "mint_url": mint_url, }, ) @@ -134,13 +244,8 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: token = await send_token(amount, "sat", mint_url) except Exception as e: logger.error( - "Failed to create cashu token for auto top-up", - extra={ - "provider_id": row.id, - "amount": amount, - "mint_url": mint_url, - "error": str(e), - }, + "Failed to create Cashu token for Routstr auto top-up", + extra={"provider_id": row.id, "mint_url": mint_url, "error": str(e)}, ) return @@ -156,47 +261,635 @@ async def _check_and_topup(row: UpstreamProviderRow) -> None: ) except Exception: logger.critical( - "Aborting auto top-up because its cashu token could not be persisted", + "Aborting Routstr auto top-up because its token was not persisted", extra={"provider_id": row.id, "mint_url": mint_url}, ) return result = await provider.topup(token) - if "error" in result: logger.error( - "Auto top-up upstream call failed", - extra={ - "provider_id": row.id, - "error": result["error"], - }, + "Routstr auto top-up call failed", + extra={"provider_id": row.id, "error": result["error"]}, ) - else: - async with create_session() as session: - transaction = ( - await session.exec( - select(CashuTransaction).where( - CashuTransaction.token == token, - CashuTransaction.type == "out", - CashuTransaction.source == "auto_topup", - ) - ) - ).first() - if transaction is None: - logger.critical( - "Completed auto top-up transaction is missing from the database", - extra={"provider_id": row.id, "mint_url": mint_url}, - ) - else: - transaction.collected = True - session.add(transaction) - await session.commit() + return - logger.info( - "Auto top-up completed successfully", + await _mark_routstr_topup_collected(row, token, mint_url) + logger.info( + "Routstr auto top-up completed", + extra={"provider_id": row.id, "amount_sats": amount}, + ) + + +async def _mark_routstr_topup_collected( + row: UpstreamProviderRow, token: str, mint_url: str +) -> None: + async with create_session() as session: + transaction = ( + await session.exec( + select(CashuTransaction).where( + CashuTransaction.token == token, + CashuTransaction.type == "out", + CashuTransaction.source == "auto_topup", + ) + ) + ).first() + if transaction is None: + logger.critical( + "Completed Routstr auto top-up is missing from the database", + extra={"provider_id": row.id, "mint_url": mint_url}, + ) + return + transaction.collected = True + session.add(transaction) + await session.commit() + + +def _ppq_state_id(row: UpstreamProviderRow) -> str: + if row.id is None: + raise ValueError("PPQ auto top-up requires a persisted provider row") + return _ppq_state_id_for_provider(row.id) + + +def _ppq_state_id_for_provider(provider_id: int | str) -> str: + return f"ppq-auto-topup-{provider_id}" + + +class PPQClaim(typing.NamedTuple): + operation_id: str + # Worker lease, not the BOLT11 invoice expiry: the invoice can expire + # while melt() is still running, and only the lease says whether the + # owning worker can still be alive. + lease_expires_at: int + phase: str + invoice_id: str + # Cashu melt quote id, "none" until a payment plan exists. This is what + # lets an ambiguous payment be reconciled against the mint later. + quote_id: str + + +def _ppq_request_id( + operation_id: str, + lease_expires_at: int, + phase: str, + invoice_id: str, + quote_id: str = "none", +) -> str: + return f"ppq:{operation_id}:{lease_expires_at}:{phase}:{invoice_id}:{quote_id}" + + +def _parse_ppq_request_id(request_id: str | None) -> PPQClaim | None: + parts = (request_id or "").split(":", 5) + if len(parts) != 6 or parts[0] != "ppq" or parts[3] not in PPQ_PHASES: + return None + try: + lease_expires_at = int(parts[2]) + except (TypeError, ValueError): + return None + return PPQClaim(parts[1], lease_expires_at, parts[3], parts[4], parts[5]) + + +def _ppq_claim_is_releasable(claim: PPQClaim | None, now: float) -> bool: + """Whether an admin may sweep this claim without risking a double payment. + + A claim in ``in_flight`` is owned by a worker that is somewhere between + reserving proofs and hearing back from the mint. Releasing it there frees + the next cycle to pay a second invoice, which is exactly what the claim + exists to prevent — so it is releasable only once its lease has passed, + which means the owning worker died rather than that it is still working. + """ + if claim is None: + # Corrupt state cannot be reasoned about and has no operation id to + # fence on. Leaving it unreleasable would strand the provider forever. + return True + if claim.phase == PPQ_PHASE_IN_FLIGHT: + return now >= claim.lease_expires_at + return True + + +async def get_ppq_auto_topup_state(provider_id: int) -> dict[str, object]: + """Return admin-safe state for a provider's durable PPQ claim.""" + async with create_session() as session: + transaction = await session.get( + CashuTransaction, _ppq_state_id_for_provider(provider_id) + ) + if transaction is None or transaction.collected or transaction.swept: + return {"active": False} + + claim = _parse_ppq_request_id(transaction.request_id) + return { + "active": True, + # The exact version of the claim the admin is looking at. A release + # must echo it back verbatim: the operation id alone is stable across + # phase changes, so it cannot distinguish "the state I reviewed" from + # "the same attempt after its payment turned ambiguous". + "state_token": transaction.request_id, + "operation_id": claim.operation_id if claim else None, + "phase": claim.phase if claim else None, + "releasable": _ppq_claim_is_releasable(claim, time.time()), + "expires_at": claim.lease_expires_at if claim else None, + "invoice_id": ( + claim.invoice_id if claim and claim.invoice_id != "pending" else None + ), + "created_at": transaction.created_at, + "amount": transaction.amount, + "unit": transaction.unit, + "mint_url": transaction.mint_url, + "malformed": claim is None, + } + + +class PPQReleaseOutcome(typing.NamedTuple): + released: bool + reason: str + + +async def release_ppq_auto_topup_state( + provider_id: int, *, state_token: str | None +) -> PPQReleaseOutcome: + """Force-release an active claim after an admin reconciles payment status. + + ``state_token`` is the ``state_token`` the caller read from + :func:`get_ppq_auto_topup_state` — the claim's full ``request_id``. Two + things are enforced with it. The claim must not be inside an unexpired + ``in_flight`` lease, because a worker between reserving proofs and hearing + from the mint still owns the outcome. And the row must be byte-identical + to the one the caller reviewed: the update fences on the whole token, so + any phase change, lease renewal, or new attempt since the review fails the + write instead of sweeping a state the admin never saw. + """ + state_id = _ppq_state_id_for_provider(provider_id) + async with create_session() as session: + transaction = await session.get(CashuTransaction, state_id) + + if transaction is None or transaction.collected or transaction.swept: + return PPQReleaseOutcome(False, "no_active_claim") + + if transaction.request_id != state_token: + return PPQReleaseOutcome(False, "stale_state") + + claim = _parse_ppq_request_id(transaction.request_id) + if not _ppq_claim_is_releasable(claim, time.time()): + return PPQReleaseOutcome(False, "payment_in_flight") + + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == state_id, + col(CashuTransaction.source) == "ppq_auto_topup", + # Fence on the token itself, not the row read above: a change + # between the read and this write must lose the race. + col(CashuTransaction.request_id) == state_token, + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values(swept=True) + ) + await session.commit() + if (getattr(result, "rowcount", 0) or 0) == 1: + return PPQReleaseOutcome(True, "released") + return PPQReleaseOutcome(False, "claim_changed") + + +async def _set_ppq_state_terminal( + row: UpstreamProviderRow, + operation_id: str, + *, + collected: bool, + swept: bool, +) -> bool: + """Finish a PPQ attempt only if this worker still owns the claim.""" + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == _ppq_state_id(row), + col(CashuTransaction.request_id).like(f"ppq:{operation_id}:%"), + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values(collected=collected, swept=swept) + ) + await session.commit() + return (getattr(result, "rowcount", 0) or 0) == 1 + + +async def _reconcile_ppq_state( + row: UpstreamProviderRow, provider: PPQAIUpstreamProvider | None +) -> bool: + """Return True while a prior PPQ attempt must suppress a new payment. + + ``provider`` may be ``None`` when the API key is gone: PPQ settlement + cannot be polled then, but mint-side reconciliation still runs. + """ + async with create_session() as session: + transaction = await session.get(CashuTransaction, _ppq_state_id(row)) + if transaction is None or transaction.collected or transaction.swept: + return False + + claim = _parse_ppq_request_id(transaction.request_id) + if claim is None: + logger.critical( + "Malformed PPQ auto top-up state; suppressing duplicate payment", + extra={"provider_id": row.id}, + ) + return True + + if claim.invoice_id != "pending": + if provider is not None and await provider.check_topup_status( + claim.invoice_id + ): + if not await _set_ppq_state_terminal( + row, claim.operation_id, collected=True, swept=False + ): + # An admin release won the race against a settlement that + # turned out to have succeeded. The next cycle may pay again; + # the balance check is the only remaining guard, so shout. + logger.critical( + "PPQ invoice settled but its claim was already released; " + "a duplicate top-up is possible on the next cycle", + extra={ + "provider_id": row.id, + "invoice_id": claim.invoice_id, + }, + ) + return True + # PPQ has not credited the invoice. Ask the mint what became of the + # melt — the durable reconciliation path for a payment whose worker + # died or whose melt call never returned. cashu settles the wallet + # database as a side effect: "unpaid" releases the reserved proofs. + if ( + claim.quote_id != "none" + and transaction.mint_url + and time.time() >= claim.lease_expires_at + ): + status = await check_bolt11_payment_status( + transaction.mint_url, transaction.unit, claim.quote_id + ) + if status == "unpaid": + # Provably never paid, funds recovered — safe to retry. + released = await _set_ppq_state_terminal( + row, claim.operation_id, collected=False, swept=True + ) + if released: + logger.warning( + "PPQ auto top-up melt was never paid; claim released", + extra={ + "provider_id": row.id, + "invoice_id": claim.invoice_id, + }, + ) + return not released + # "paid" means the mint paid but PPQ has not credited yet: keep + # waiting on PPQ. "pending"/"unknown" stay locked for the admin. + return True + if time.time() < claim.lease_expires_at: + return True + + released = await _set_ppq_state_terminal( + row, claim.operation_id, collected=False, swept=True + ) + return not released + + +async def _ppq_provider_is_claimable( + session: object, provider_id: int | None +) -> bool: + """Re-read the provider inside the claim transaction. + + SQLite serialises write transactions, so checking here — rather than + trusting the row the cycle loaded earlier — means a concurrent provider + deletion or type change either commits before us (we see it and refuse) + or after us (its own claim check sees our claim and refuses). Without + this the worker could create a claim for a provider that no longer + exists, orphaning it forever. + """ + if provider_id is None: + return False + current = await session.get(UpstreamProviderRow, provider_id) # type: ignore[attr-defined] + return current is not None and current.provider_type == "ppqai" + + +async def _claim_ppq_topup(row: UpstreamProviderRow) -> str | None: + """Acquire a durable, ownership-fenced per-provider claim.""" + state_id = _ppq_state_id(row) + operation_id = uuid.uuid4().hex + expires_at = int(time.time()) + PPQ_PENDING_TTL_SECONDS + request_id = _ppq_request_id( + operation_id, expires_at, PPQ_PHASE_CLAIMED, "pending" + ) + + async with create_session() as session: + if not await _ppq_provider_is_claimable(session, row.id): + return None + existing = await session.get(CashuTransaction, state_id) + if existing is not None: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == state_id, + or_( + CashuTransaction.collected == True, # noqa: E712 + CashuTransaction.swept == True, # noqa: E712 + ), + ) + .values( + token="pending", + amount=0, + unit="sat", + mint_url=None, + request_id=request_id, + collected=False, + swept=False, + created_at=int(time.time()), + ) + ) + await session.commit() + if (getattr(result, "rowcount", 0) or 0) != 1: + return None + return operation_id + + try: + async with create_session() as session: + # Same fencing as the update path: the provider must still exist + # inside the transaction that creates the claim. + if not await _ppq_provider_is_claimable(session, row.id): + return None + session.add( + CashuTransaction( + id=state_id, + token="pending", + amount=0, + unit="sat", + type="out", + request_id=request_id, + collected=False, + source="ppq_auto_topup", + ) + ) + await session.commit() + except IntegrityError: + return None + return operation_id + + +async def _record_ppq_invoice( + row: UpstreamProviderRow, + operation_id: str, + *, + invoice: str, + invoice_id: str, + quote_id: str, + amount: int, + unit: str, + mint_url: str, +) -> int: + """Move the claim to in_flight and return its fresh worker lease. + + The lease is minted here, at the start of the payment, and deliberately + not derived from the BOLT11 invoice's expiry: the invoice can expire + while melt() is still running, and the lease answers a different question + — can the worker that owns this claim still be alive? + """ + lease_expires_at = int(time.time()) + PPQ_PENDING_TTL_SECONDS + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == _ppq_state_id(row), + col(CashuTransaction.request_id).like( + f"ppq:{operation_id}:%:{PPQ_PHASE_CLAIMED}:pending:none" + ), + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values( + token=invoice, + # Moves the claim to in_flight: from here the proofs are + # committed and an admin may not release it until the lease + # runs out. + request_id=_ppq_request_id( + operation_id, + lease_expires_at, + PPQ_PHASE_IN_FLIGHT, + invoice_id, + quote_id, + ), + amount=amount, + unit=unit, + mint_url=mint_url, + ) + ) + await session.commit() + if (getattr(result, "rowcount", 0) or 0) != 1: + raise RuntimeError("PPQ auto top-up claim ownership was lost") + return lease_expires_at + + +async def _mark_ppq_reconcile( + row: UpstreamProviderRow, + operation_id: str, + lease_expires_at: int, + invoice_id: str, + quote_id: str, +) -> None: + """Move an in_flight claim to reconcile so an admin may release it. + + Without this an ambiguous payment stays in_flight, and in_flight is only + releasable once its lease expires — the admin would have to wait out the + lease before they could act on an alert that already fired. + """ + async with create_session() as session: + result = await session.exec( # type: ignore[call-overload] + update(CashuTransaction) + .where( + col(CashuTransaction.id) == _ppq_state_id(row), + col(CashuTransaction.request_id) + == _ppq_request_id( + operation_id, + lease_expires_at, + PPQ_PHASE_IN_FLIGHT, + invoice_id, + quote_id, + ), + col(CashuTransaction.collected) == False, # noqa: E712 + col(CashuTransaction.swept) == False, # noqa: E712 + ) + .values( + request_id=_ppq_request_id( + operation_id, + lease_expires_at, + PPQ_PHASE_RECONCILE, + invoice_id, + quote_id, + ) + ) + ) + await session.commit() + if (getattr(result, "rowcount", 0) or 0) != 1: + logger.warning( + "Could not flag the PPQ claim for reconciliation; " + "it is no longer owned by this attempt", + extra={"provider_id": row.id, "invoice_id": invoice_id}, + ) + + +async def _check_and_topup_ppq(row: UpstreamProviderRow, settings: dict) -> None: + threshold_usd = float(settings["topup_threshold"]) + amount_usd = int(settings["topup_amount_limit"]) + provider = PPQAIUpstreamProvider.from_db_row(row) + if provider is None or await _reconcile_ppq_state(row, provider): + return + + balance = await provider.get_balance() + if balance is None or not math.isfinite(balance) or balance < 0: + logger.warning( + "Could not fetch a valid PPQ balance for auto top-up", + extra={"provider_id": row.id}, + ) + return + if balance >= threshold_usd: + return + operation_id = await _claim_ppq_topup(row) + if operation_id is None: + return + + logger.info( + "PPQ auto top-up triggered", + extra={ + "provider_id": row.id, + "balance_usd": balance, + "threshold_usd": threshold_usd, + "topup_usd": amount_usd, + }, + ) + + try: + topup = await provider.initiate_topup(amount_usd) + if topup.currency.upper() != "USD" or topup.amount != amount_usd: + raise ValueError("PPQ top-up response amount or currency does not match") + + now = int(time.time()) + # The invoice's own expiry is a pre-payment sanity check only; the + # claim's lease is minted separately in _record_ppq_invoice. + invoice_expires_at = topup.expires_at or now + PPQ_PENDING_TTL_SECONDS + if invoice_expires_at > 10**12: + invoice_expires_at //= 1000 + if invoice_expires_at <= now: + raise ValueError("PPQ returned an expired Lightning invoice") + + plan = await prepare_bolt11_payment(topup.payment_request) + max_invoice_sats = math.ceil( + amount_usd / sats_usd_price() * PPQ_MAX_INVOICE_PREMIUM + ) + if plan.maximum_spend_sats > max_invoice_sats: + raise ValueError("PPQ Lightning invoice exceeds the USD spending cap") + + lease_expires_at = await _record_ppq_invoice( + row, + operation_id, + invoice=topup.payment_request, + invoice_id=topup.invoice_id, + quote_id=str(plan.quote.quote), + amount=int(plan.quote.amount + plan.quote.fee_reserve), + unit=plan.unit, + mint_url=plan.mint_url, + ) + except Exception: + # Nothing has been paid yet, so the claim can be handed back. If the + # release does not land, the claim is no longer ours to reason about. + if not await _set_ppq_state_terminal( + row, operation_id, collected=False, swept=True + ): + logger.warning( + "Could not release the PPQ auto top-up claim after a " + "pre-payment failure; it is owned by another attempt", + extra={"provider_id": row.id}, + ) + raise + + try: + paid_amount, mint_url, unit = await execute_bolt11_payment(plan) + except Bolt11PaymentNotAttempted: + # The mint's own answer rules out a settlement and any reserved proofs + # were handed back, so this claim is safe to retry next cycle. + if not await _set_ppq_state_terminal( + row, operation_id, collected=False, swept=True + ): + logger.warning( + "Could not release the PPQ auto top-up claim after a payment " + "that was never attempted; it is owned by another attempt", + extra={"provider_id": row.id}, + ) + logger.warning( + "PPQ Lightning payment was not attempted; claim released for retry", + extra={"provider_id": row.id, "invoice_id": topup.invoice_id}, + exc_info=True, + ) + raise + except Exception: + await _mark_ppq_reconcile( + row, + operation_id, + lease_expires_at, + topup.invoice_id, + str(plan.quote.quote), + ) + logger.critical( + "PPQ auto top-up payment outcome is ambiguous; claim remains locked until admin reconciliation", extra={ "provider_id": row.id, - "amount": amount, - "new_balance_approx": balance + amount, + "invoice_id": topup.invoice_id, + "admin_action": f"POST /admin/api/upstream-providers/{row.id}/ppq-auto-topup/release", }, + exc_info=True, ) + raise + + settled = False + for attempt in range(PPQ_SETTLEMENT_ATTEMPTS): + if await provider.check_topup_status(topup.invoice_id): + settled = True + break + if attempt + 1 < PPQ_SETTLEMENT_ATTEMPTS: + await asyncio.sleep(PPQ_SETTLEMENT_POLL_SECONDS) + + if not settled: + # The payment left the wallet, so the claim must stay. Flag it for + # reconciliation: _reconcile_ppq_state keeps polling PPQ, and an admin + # can step in without waiting out the lease. + await _mark_ppq_reconcile( + row, + operation_id, + lease_expires_at, + topup.invoice_id, + str(plan.quote.quote), + ) + logger.critical( + "PPQ Lightning payment completed but credit settlement is unconfirmed", + extra={"provider_id": row.id, "invoice_id": topup.invoice_id}, + ) + return + + if not await _set_ppq_state_terminal( + row, operation_id, collected=True, swept=False + ): + # Something else finished this claim while the payment was in flight — + # an admin release, most likely. The next cycle is now free to pay + # again, so surface it rather than completing quietly. + logger.critical( + "PPQ auto top-up settled but its claim was already released; " + "a duplicate top-up is possible on the next cycle", + extra={"provider_id": row.id, "invoice_id": topup.invoice_id}, + ) + logger.info( + "PPQ auto top-up completed", + extra={ + "provider_id": row.id, + "topup_usd": amount_usd, + "cashu_paid": paid_amount, + "cashu_unit": unit, + "mint_url": mint_url, + }, + ) diff --git a/routstr/upstream/ppqai.py b/routstr/upstream/ppqai.py index 6cc26cb2..50ab4532 100644 --- a/routstr/upstream/ppqai.py +++ b/routstr/upstream/ppqai.py @@ -441,7 +441,7 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider): """ data = await self.check_balance() balance = data.get("balance") - if isinstance(balance, (int, float)): + if isinstance(balance, (int, float)) and not isinstance(balance, bool): return float(balance) return None diff --git a/routstr/wallet.py b/routstr/wallet.py index dd92d913..08eb07a8 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -3,10 +3,11 @@ import re import socket import time import typing +from dataclasses import dataclass from typing import TypedDict import httpx -from cashu.core.base import Proof, Token +from cashu.core.base import MeltQuote, Proof, Token from cashu.core.mint_info import MintInfo as _CashuMintInfo from cashu.wallet.helpers import deserialize_token_from_string from cashu.wallet.wallet import Wallet @@ -282,6 +283,203 @@ async def send_token(amount: int, unit: str, mint_url: str | None = None) -> str return token +class Bolt11PaymentNotAttempted(Exception): + """The invoice was definitively not paid, so the attempt can be retried. + + Raised only where the mint's own answer rules out a settlement: coin + selection never reached ``melt``, or ``melt`` returned an explicit unpaid + state. Any proofs reserved along the way are released before this is + raised. + """ + + +class Bolt11PaymentAmbiguous(Exception): + """The payment may or may not have settled, so it must not be retried. + + Raised when ``melt`` errored, timed out, or came back pending. The selected + proofs stay reserved: the mint may still complete the payment with them, + and spending them elsewhere would be a double spend. + """ + + +@dataclass +class Bolt11PaymentPlan: + invoice: str + wallet: Wallet + proofs: list[Proof] + quote: MeltQuote + mint_url: str + unit: str + + @property + def invoice_amount_sats(self) -> int: + amount = int(self.quote.amount) + return amount if self.unit == "sat" else (amount + 999) // 1000 + + @property + def maximum_spend_sats(self) -> int: + maximum = ( + int(self.quote.amount) + + int(self.quote.fee_reserve) + + int(self.wallet.get_fees_for_proofs(self.proofs)) + ) + return maximum if self.unit == "sat" else (maximum + 999) // 1000 + + +async def prepare_bolt11_payment(invoice: str) -> Bolt11PaymentPlan: + """Choose the sufficiently funded configured mint with most balance. + + Candidate discovery only reads balances and melt quotes. Coin selection, + which may split proofs, is deferred until the winning mint is known. + """ + mint_urls = list(dict.fromkeys([*settings.cashu_mints, settings.primary_mint])) + candidates: list[tuple[int, Wallet, list[Proof], MeltQuote, str, str]] = [] + + for mint_url in mint_urls: + if not mint_url: + continue + for unit in ("sat", "msat"): + try: + wallet = await get_wallet(mint_url, unit) + proofs = get_proofs_per_mint_and_unit( + wallet, mint_url, unit, not_reserved=True + ) + proofs = await slow_filter_spend_proofs(proofs, wallet) + balance = sum(proof.amount for proof in proofs) + if balance <= 0: + continue + + quote = await wallet.melt_quote(invoice=invoice) + # select_to_send runs with include_fees=True, so the input fee + # has to be part of sufficiency too. Without it a mint passes + # this filter and then fails coin selection. + required = ( + quote.amount + + quote.fee_reserve + + wallet.get_fees_for_proofs(proofs) + ) + if balance < required: + continue + balance_msats = balance * 1000 if unit == "sat" else balance + candidates.append( + (balance_msats, wallet, proofs, quote, mint_url, unit) + ) + except Exception as e: + logger.debug( + "Cashu mint cannot fund BOLT11 invoice", + extra={"mint_url": mint_url, "unit": unit, "error": str(e)}, + ) + + if not candidates: + raise ValueError("No configured Cashu mint has enough balance to pay invoice") + + _, wallet, proofs, quote, mint_url, unit = max( + candidates, key=lambda item: item[0] + ) + return Bolt11PaymentPlan(invoice, wallet, proofs, quote, mint_url, unit) + + +async def execute_bolt11_payment(plan: Bolt11PaymentPlan) -> tuple[int, str, str]: + """Execute a prepared payment, separating retryable from ambiguous failure. + + Raises ``Bolt11PaymentNotAttempted`` when the invoice provably did not + settle, and ``Bolt11PaymentAmbiguous`` when the outcome is unknown. Callers + may safely retry the first and must never retry the second. + """ + # Select unreserved, mirroring send_token: a selection failure must not + # strand proofs that were never handed to the mint. + try: + selected, _ = await plan.wallet.select_to_send( + plan.proofs, + plan.quote.amount + plan.quote.fee_reserve, + set_reserved=False, + include_fees=True, + ) + except Exception as e: + raise Bolt11PaymentNotAttempted(f"Coin selection failed: {e}") from e + + await plan.wallet.set_reserved_for_send(selected, reserved=True) + + try: + result = await asyncio.wait_for( + plan.wallet.melt( + proofs=selected, + invoice=plan.invoice, + fee_reserve_sat=plan.quote.fee_reserve, + quote_id=plan.quote.quote, + ), + timeout=60, + ) + except Exception as e: + # The mint may still be settling with these proofs, so they must stay + # reserved — but cashu's melt() un-reserves them itself on a mint + # transport error, the exact ambiguous case. Re-reserve with the melt + # quote id, not as a send: get_melt_quote() finds the proofs to settle + # by melt_id, so a send-style reservation would strand them — paid + # proofs never invalidated, unpaid ones never released. + try: + await plan.wallet.set_reserved_for_melt( + selected, reserved=True, quote_id=plan.quote.quote + ) + except Exception: + logger.critical( + "Could not re-reserve proofs after an ambiguous melt", + extra={"mint_url": plan.mint_url, "quote_id": plan.quote.quote}, + ) + raise Bolt11PaymentAmbiguous(f"Cashu melt did not return: {e}") from e + + raw_state = getattr(result, "state", None) + state = str(raw_state).lower().rsplit(".", 1)[-1] if raw_state is not None else "" + if state == "paid" or getattr(result, "paid", None) is True: + change = getattr(result, "change", None) or [] + paid = sum(proof.amount for proof in selected) - sum( + int(item.amount) for item in change + ) + return paid, plan.mint_url, plan.unit + + if state == "unpaid": + # The mint is telling us it did not pay, so the proofs are ours again. + await plan.wallet.set_reserved_for_send(selected, reserved=False) + raise Bolt11PaymentNotAttempted("Cashu mint reported the melt as unpaid") + + raise Bolt11PaymentAmbiguous( + f"Cashu melt did not reach a final state: {state or 'unknown'}" + ) + + +async def pay_bolt11_invoice(invoice: str) -> tuple[int, str, str]: + """Prepare and pay a BOLT11 invoice from the best-funded Cashu mint.""" + return await execute_bolt11_payment(await prepare_bolt11_payment(invoice)) + + +async def check_bolt11_payment_status( + mint_url: str, unit: str, quote_id: str +) -> str: + """Ask the mint what became of an earlier melt attempt. + + Returns ``"paid"``, ``"unpaid"``, ``"pending"``, or ``"unknown"``. This is + the durable reconciliation path for an ambiguous payment: cashu's + ``get_melt_quote`` also settles the wallet database — invalidating the + proofs on ``paid`` and releasing their reservation on ``unpaid`` — so a + caller that sees ``"unpaid"`` may safely retry with the same funds. + """ + try: + wallet = await get_wallet(mint_url, unit) + quote = await wallet.get_melt_quote(quote_id) + except Exception as e: + logger.warning( + "Could not query the mint for a melt quote's status", + extra={"mint_url": mint_url, "quote_id": quote_id, "error": str(e)}, + ) + return "unknown" + if quote is None: + return "unknown" + state = str(getattr(quote, "state", "")).lower().rsplit(".", 1)[-1] + if state in ("paid", "unpaid", "pending"): + return state + return "unknown" + + # A foreign mint's fee_reserve is a non-binding estimate (NUT-05): the mint may # demand more when re-quoting or at melt execution. Instead of padding the # estimate with a safety buffer (which strands the margin at the foreign mint @@ -999,6 +1197,10 @@ async def _refund_sweep_once(cutoff: int) -> None: db.CashuTransaction.type == "out", db.CashuTransaction.collected == False, # noqa: E712 db.CashuTransaction.swept == False, # noqa: E712 + # NULL != 'x' is NULL in SQL, so legacy rows with no source would + # drop out of the sweep without the explicit IS NULL arm. + (db.CashuTransaction.source != "ppq_auto_topup") + | (db.CashuTransaction.source == None), # noqa: E711 db.CashuTransaction.created_at < cutoff, ) results = await session.exec(stmt) diff --git a/tests/integration/test_ppq_auto_topup_claim.py b/tests/integration/test_ppq_auto_topup_claim.py new file mode 100644 index 00000000..0b338bf0 --- /dev/null +++ b/tests/integration/test_ppq_auto_topup_claim.py @@ -0,0 +1,516 @@ +"""Real-database tests for the PPQ auto top-up claim lifecycle. + +These exercise the claim against actual SQL rather than mocked sessions, +because the guarantees under test are all about what the database will and +will not let two concurrent writers do. +""" + +import time +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from sqlmodel import select + +from routstr.core.db import CashuTransaction, create_session +from routstr.upstream.auto_topup import ( + PPQ_PHASE_CLAIMED, + PPQ_PHASE_IN_FLIGHT, + PPQ_PHASE_RECONCILE, + _claim_ppq_topup, + _ppq_request_id, + _ppq_state_id_for_provider, + _record_ppq_invoice, + _set_ppq_state_terminal, + get_ppq_auto_topup_state, + release_ppq_auto_topup_state, +) + +pytestmark = pytest.mark.asyncio + + +def _row(provider_id: int = 1) -> MagicMock: + row = MagicMock() + row.id = provider_id + return row + + +async def _seed_provider(provider_id: int = 1, slug: str = "ppq") -> None: + """Claim creation is fenced on the provider row existing; seed it.""" + from routstr.core.db import UpstreamProviderRow + + async with create_session() as session: + session.add( + UpstreamProviderRow( + id=provider_id, + slug=slug, + provider_type="ppqai", + base_url="https://api.ppq.ai", + api_key="secret", + enabled=True, + ) + ) + await session.commit() + + +async def _state_row(provider_id: int = 1) -> CashuTransaction | None: + async with create_session() as session: + return await session.get( + CashuTransaction, _ppq_state_id_for_provider(provider_id) + ) + + +async def _seed_claim( + provider_id: int, + phase: str, + invoice_id: str, + lease_expires_at: int, + quote_id: str = "quote-1", +) -> str: + """Seed a claim row and return its state token (the full request_id).""" + token = _ppq_request_id( + "operation-1", lease_expires_at, phase, invoice_id, quote_id + ) + async with create_session() as session: + session.add( + CashuTransaction( + id=_ppq_state_id_for_provider(provider_id), + token="lnbc-invoice", + amount=102, + unit="sat", + type="out", + request_id=token, + mint_url="https://mint.test", + collected=False, + source="ppq_auto_topup", + ) + ) + await session.commit() + return token + + +async def test_second_claim_is_refused_while_the_first_is_active( + patched_db_engine: Any, +) -> None: + await _seed_provider() + assert await _claim_ppq_topup(_row()) is not None + # The whole point of the claim: a concurrent cycle must not get one. + assert await _claim_ppq_topup(_row()) is None + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + + +async def test_claim_is_reusable_once_the_previous_attempt_finished( + patched_db_engine: Any, +) -> None: + await _seed_provider() + first = await _claim_ppq_topup(_row()) + assert first is not None + assert await _set_ppq_state_terminal( + _row(), first, collected=True, swept=False + ) + + second = await _claim_ppq_topup(_row()) + assert second is not None and second != first + + +async def test_recording_the_invoice_moves_the_claim_in_flight( + patched_db_engine: Any, +) -> None: + await _seed_provider() + operation_id = await _claim_ppq_topup(_row()) + assert operation_id is not None + + state = await get_ppq_auto_topup_state(1) + assert state["phase"] == PPQ_PHASE_CLAIMED + assert state["releasable"] is True + assert state["invoice_id"] is None + + lease = await _record_ppq_invoice( + _row(), + operation_id, + invoice="lnbc-invoice", + invoice_id="invoice-1", + quote_id="quote-1", + amount=102, + unit="sat", + mint_url="https://mint.test", + ) + assert lease > int(time.time()) + + state = await get_ppq_auto_topup_state(1) + assert state["phase"] == PPQ_PHASE_IN_FLIGHT + assert state["invoice_id"] == "invoice-1" + # A payment is committed to a mint, so an admin must not sweep it. + assert state["releasable"] is False + # The raw BOLT11 invoice must never reach the admin API. + assert "token" not in state + + +async def test_release_refuses_an_in_flight_claim(patched_db_engine: Any) -> None: + token = await _seed_claim( + 1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900 + ) + + outcome = await release_ppq_auto_topup_state(1, state_token=token) + + assert outcome.released is False + assert outcome.reason == "payment_in_flight" + row = await _state_row() + assert row is not None and row.swept is False + + +async def test_release_refuses_a_stale_state_token(patched_db_engine: Any) -> None: + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900) + + outcome = await release_ppq_auto_topup_state(1, state_token="ppq:stale:token") + + assert outcome.released is False + assert outcome.reason == "stale_state" + row = await _state_row() + assert row is not None and row.swept is False + + +async def test_release_accepts_a_reconcile_claim(patched_db_engine: Any) -> None: + token = await _seed_claim( + 1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900 + ) + + outcome = await release_ppq_auto_topup_state(1, state_token=token) + + assert outcome.released is True + row = await _state_row() + assert row is not None and row.swept is True + + +async def test_expired_in_flight_claim_becomes_releasable( + patched_db_engine: Any, +) -> None: + # A worker that died mid-payment must not lock the provider forever. + token = await _seed_claim( + 1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) - 1 + ) + + assert (await get_ppq_auto_topup_state(1))["releasable"] is True + outcome = await release_ppq_auto_topup_state(1, state_token=token) + assert outcome.released is True + + +async def test_release_reports_no_active_claim_once_swept( + patched_db_engine: Any, +) -> None: + token = await _seed_claim( + 1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900 + ) + assert (await release_ppq_auto_topup_state(1, state_token=token)).released + + outcome = await release_ppq_auto_topup_state(1, state_token=token) + assert outcome.released is False + assert outcome.reason == "no_active_claim" + + +async def test_terminal_write_fails_after_the_claim_was_released( + patched_db_engine: Any, +) -> None: + """The symptom an admin release leaves behind for the owning worker.""" + token = await _seed_claim( + 1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900 + ) + assert (await release_ppq_auto_topup_state(1, state_token=token)).released + + assert ( + await _set_ppq_state_terminal( + _row(), "operation-1", collected=True, swept=False + ) + is False + ) + + +async def test_ppq_claim_rows_are_excluded_from_the_admin_transaction_list( + patched_db_engine: Any, +) -> None: + from routstr.core.admin import get_transactions_api + + await _claim_ppq_topup(_row()) + async with create_session() as session: + session.add( + CashuTransaction( + id="real-transaction", + token="cashuAreal", + amount=50, + unit="sat", + type="out", + source="x-cashu", + ) + ) + await session.commit() + + result = await get_transactions_api() + + ids = {t["id"] for t in result["transactions"]} # type: ignore[index,union-attr] + assert "real-transaction" in ids + assert _ppq_state_id_for_provider(1) not in ids + + +async def test_reconcile_settles_a_recorded_invoice(patched_db_engine: Any) -> None: + from routstr.upstream.auto_topup import _reconcile_ppq_state + + await _seed_claim(1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900) + provider = MagicMock() + provider.check_topup_status = AsyncMock(return_value=True) + + # Still suppresses this cycle, but the claim is now finished. + assert await _reconcile_ppq_state(_row(), provider) is True + + row = await _state_row() + assert row is not None and row.collected is True + + +async def test_stale_token_from_before_a_phase_change_cannot_release( + patched_db_engine: Any, +) -> None: + """The blocker scenario: admin reviews `claimed`, payment turns ambiguous. + + The operation id is identical in both states, so an id-based fence would + let the stale confirmation land. The full state token must not. + """ + await _seed_provider() + operation_id = await _claim_ppq_topup(_row()) + assert operation_id is not None + reviewed = await get_ppq_auto_topup_state(1) + assert reviewed["phase"] == PPQ_PHASE_CLAIMED + + # Worker records the invoice: same operation, new phase, proofs committed. + await _record_ppq_invoice( + _row(), + operation_id, + invoice="lnbc-invoice", + invoice_id="invoice-1", + quote_id="quote-1", + amount=102, + unit="sat", + mint_url="https://mint.test", + ) + + outcome = await release_ppq_auto_topup_state( + 1, state_token=str(reviewed["state_token"]) + ) + assert outcome.released is False + assert outcome.reason == "stale_state" + row = await _state_row() + assert row is not None and row.swept is False + + +async def test_concurrent_claims_only_one_wins(patched_db_engine: Any) -> None: + import asyncio + + await _seed_provider() + + results = await asyncio.gather( + *(_claim_ppq_topup(_row()) for _ in range(5)), return_exceptions=True + ) + winners = [r for r in results if isinstance(r, str)] + assert len(winners) == 1 + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert len(rows) == 1 + + +async def test_reconcile_releases_claim_when_mint_reports_unpaid( + patched_db_engine: Any, +) -> None: + from routstr.upstream.auto_topup import _reconcile_ppq_state + + # Lease expired, PPQ never credited: only the mint's own "unpaid" answer + # may hand the claim back. + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1) + provider = MagicMock() + provider.check_topup_status = AsyncMock(return_value=False) + + with patch( + "routstr.upstream.auto_topup.check_bolt11_payment_status", + AsyncMock(return_value="unpaid"), + ) as status: + suppressed = await _reconcile_ppq_state(_row(), provider) + + status.assert_awaited_once_with("https://mint.test", "sat", "quote-1") + assert suppressed is False + row = await _state_row() + assert row is not None and row.swept is True + + +async def test_reconcile_keeps_claim_when_mint_answer_is_not_final( + patched_db_engine: Any, +) -> None: + from routstr.upstream.auto_topup import _reconcile_ppq_state + + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1) + provider = MagicMock() + provider.check_topup_status = AsyncMock(return_value=False) + + for answer in ("paid", "pending", "unknown"): + with patch( + "routstr.upstream.auto_topup.check_bolt11_payment_status", + AsyncMock(return_value=answer), + ): + assert await _reconcile_ppq_state(_row(), provider) is True + row = await _state_row() + assert row is not None and row.swept is False, answer + + +async def test_release_endpoint_maps_refusals_to_409(patched_db_engine: Any) -> None: + from fastapi import HTTPException + + from routstr.core.admin import ( + ReleasePPQAutoTopupRequest, + release_ppq_auto_topup_api, + ) + + provider_row = MagicMock() + provider_row.provider_type = "ppqai" + + token = await _seed_claim( + 1, PPQ_PHASE_IN_FLIGHT, "invoice-1", int(time.time()) + 900 + ) + + with patch( + "routstr.core.admin._require_ppq_provider", + AsyncMock(return_value=provider_row), + ): + with pytest.raises(HTTPException) as excinfo: + await release_ppq_auto_topup_api( + 1, + ReleasePPQAutoTopupRequest( + confirmed_safe_to_retry=True, state_token=token + ), + ) + assert excinfo.value.status_code == 409 + assert "in flight" in excinfo.value.detail + + with pytest.raises(HTTPException) as excinfo: + await release_ppq_auto_topup_api( + 1, + ReleasePPQAutoTopupRequest( + confirmed_safe_to_retry=True, state_token="ppq:wrong" + ), + ) + assert excinfo.value.status_code == 409 + assert "changed since" in excinfo.value.detail + + +async def test_provider_delete_is_blocked_by_an_active_claim( + patched_db_engine: Any, +) -> None: + from fastapi import HTTPException + + from routstr.core.admin import delete_upstream_provider + from routstr.core.db import UpstreamProviderRow + + async with create_session() as session: + session.add( + UpstreamProviderRow( + id=1, + slug="ppq", + provider_type="ppqai", + base_url="https://api.ppq.ai", + api_key="secret", + enabled=True, + ) + ) + await session.commit() + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900) + + with pytest.raises(HTTPException) as excinfo: + await delete_upstream_provider("1") + assert excinfo.value.status_code == 409 + + # Provider must still exist. + async with create_session() as session: + assert await session.get(UpstreamProviderRow, 1) is not None + + +async def test_claim_is_refused_when_the_provider_row_is_gone( + patched_db_engine: Any, +) -> None: + """The worker's half of the delete race: no provider row, no claim.""" + assert await _claim_ppq_topup(_row()) is None + + async with create_session() as session: + rows = (await session.exec(select(CashuTransaction))).all() + assert rows == [] + + +async def test_claim_is_refused_after_a_provider_type_change( + patched_db_engine: Any, +) -> None: + from routstr.core.db import UpstreamProviderRow + + await _seed_provider() + async with create_session() as session: + provider = await session.get(UpstreamProviderRow, 1) + assert provider is not None + provider.provider_type = "openai" + session.add(provider) + await session.commit() + + assert await _claim_ppq_topup(_row()) is None + + +async def test_disabled_provider_with_claim_still_reconciles( + patched_db_engine: Any, +) -> None: + """A claim tracks committed money; eligibility must not stop reconciling.""" + from routstr.core.db import UpstreamProviderRow + from routstr.upstream.auto_topup import _reconcile_all_ppq_claims + + await _seed_provider() + async with create_session() as session: + provider = await session.get(UpstreamProviderRow, 1) + assert provider is not None + provider.enabled = False + session.add(provider) + await session.commit() + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) + 900) + + ppq = MagicMock() + ppq.check_topup_status = AsyncMock(return_value=True) + with patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=ppq, + ): + await _reconcile_all_ppq_claims() + + row = await _state_row() + assert row is not None and row.collected is True + + +async def test_claim_without_api_key_still_reconciles_via_the_mint( + patched_db_engine: Any, +) -> None: + from routstr.core.db import UpstreamProviderRow + from routstr.upstream.auto_topup import _reconcile_all_ppq_claims + + await _seed_provider() + async with create_session() as session: + provider = await session.get(UpstreamProviderRow, 1) + assert provider is not None + provider.api_key = "" + session.add(provider) + await session.commit() + # Lease expired, so the mint may be consulted. + await _seed_claim(1, PPQ_PHASE_RECONCILE, "invoice-1", int(time.time()) - 1) + + with patch( + "routstr.upstream.auto_topup.check_bolt11_payment_status", + AsyncMock(return_value="unpaid"), + ) as status: + await _reconcile_all_ppq_claims() + + # No API key: PPQ was never polled, but the mint was, and its definitive + # "unpaid" released the claim. + status.assert_awaited_once() + row = await _state_row() + assert row is not None and row.swept is True diff --git a/tests/unit/test_auto_topup.py b/tests/unit/test_auto_topup.py index 2adb13e6..9bb6babf 100644 --- a/tests/unit/test_auto_topup.py +++ b/tests/unit/test_auto_topup.py @@ -4,7 +4,24 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from routstr.core.db import CashuTransaction -from routstr.upstream.auto_topup import _check_and_topup +from routstr.upstream.auto_topup import ( + _check_and_topup, + _parse_ppq_request_id, + validate_auto_topup_settings, +) +from routstr.upstream.ppqai import PPQAIUpstreamProvider + + +def test_ppq_claim_parser_rejects_invalid_expiry() -> None: + assert _parse_ppq_request_id("ppq:operation:not-a-timestamp:invoice") is None + + +@pytest.mark.asyncio +async def test_ppq_balance_rejects_boolean_api_value() -> None: + provider = PPQAIUpstreamProvider("secret") + provider.check_balance = AsyncMock(return_value={"balance": False}) # type: ignore[method-assign] + + assert await provider.get_balance() is None def _row() -> MagicMock: @@ -12,6 +29,7 @@ def _row() -> MagicMock: row.id = "provider-1" row.base_url = "https://provider.test" row.api_key = "secret" + row.provider_type = "routstr" row.provider_settings = json.dumps( { "auto_topup": True, @@ -141,3 +159,346 @@ async def test_auto_topup_does_not_send_untracked_token() -> None: ): await _check_and_topup(_row()) provider.topup.assert_not_awaited() + + +def _ppq_row() -> MagicMock: + row = MagicMock() + row.id = "ppq-provider-1" + row.base_url = "https://api.ppq.ai" + row.api_key = "secret" + row.provider_type = "ppqai" + row.provider_settings = json.dumps( + { + "auto_topup": True, + "topup_threshold": 5.0, + "topup_amount_limit": 10, + } + ) + return row + + +@pytest.mark.asyncio +async def test_ppq_auto_topup_pays_invoice_and_confirms_settlement() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=2.5) + provider.initiate_topup = AsyncMock( + return_value=MagicMock( + invoice_id="invoice-1", + payment_request="lnbc-invoice", + amount=10, + currency="USD", + expires_at=None, + ) + ) + provider.check_topup_status = AsyncMock(return_value=True) + plan = MagicMock() + plan.invoice_amount_sats = 100 + plan.maximum_spend_sats = 102 + plan.quote.amount = 100 + plan.quote.fee_reserve = 2 + plan.mint_url = "https://mint-rich.test" + plan.unit = "sat" + row = _ppq_row() + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=False), + ), + patch( + "routstr.upstream.auto_topup._claim_ppq_topup", + AsyncMock(return_value="operation-1"), + ), + patch( + "routstr.upstream.auto_topup.prepare_bolt11_payment", + AsyncMock(return_value=plan), + ) as prepare, + patch( + "routstr.upstream.auto_topup.execute_bolt11_payment", + AsyncMock(return_value=(101, "https://mint-rich.test", "sat")), + ) as execute, + patch( + "routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock() + ) as record, + patch( + "routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock() + ) as terminal, + patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001), + ): + await _check_and_topup(row) + + provider.initiate_topup.assert_awaited_once_with(10) + prepare.assert_awaited_once_with("lnbc-invoice") + execute.assert_awaited_once_with(plan) + record.assert_awaited_once() + provider.check_topup_status.assert_awaited_once_with("invoice-1") + terminal.assert_awaited_once_with( + row, "operation-1", collected=True, swept=False + ) + + +@pytest.mark.asyncio +async def test_ppq_ambiguous_melt_keeps_claim_and_emits_critical_alert() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=2.5) + provider.initiate_topup = AsyncMock( + return_value=MagicMock( + invoice_id="invoice-1", + payment_request="lnbc-invoice", + amount=10, + currency="USD", + expires_at=None, + ) + ) + plan = MagicMock(maximum_spend_sats=102, mint_url="https://mint.test", unit="sat") + plan.quote.amount = 100 + plan.quote.fee_reserve = 2 + row = _ppq_row() + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=False), + ), + patch( + "routstr.upstream.auto_topup._claim_ppq_topup", + AsyncMock(return_value="operation-1"), + ), + patch( + "routstr.upstream.auto_topup.prepare_bolt11_payment", + AsyncMock(return_value=plan), + ), + patch( + "routstr.upstream.auto_topup.execute_bolt11_payment", + AsyncMock(side_effect=TimeoutError("ambiguous melt")), + ), + patch( + "routstr.upstream.auto_topup._record_ppq_invoice", + AsyncMock(return_value=2_000_000_000), + ), + patch( + "routstr.upstream.auto_topup._mark_ppq_reconcile", AsyncMock() + ) as reconcile_mark, + patch( + "routstr.upstream.auto_topup._set_ppq_state_terminal", AsyncMock() + ) as terminal, + patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001), + patch("routstr.upstream.auto_topup.logger.critical") as critical, + ): + with pytest.raises(TimeoutError, match="ambiguous melt"): + await _check_and_topup(row) + + # The claim is never released — it moves to reconcile for the admin. + terminal.assert_not_awaited() + reconcile_mark.assert_awaited_once() + critical.assert_called_once() + assert "admin reconciliation" in critical.call_args.args[0] + + +@pytest.mark.asyncio +async def test_ppq_auto_topup_skips_when_balance_meets_threshold() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=5.0) + provider.initiate_topup = AsyncMock() + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=False), + ), + ): + await _check_and_topup(_ppq_row()) + + provider.initiate_topup.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_ppq_pending_attempt_suppresses_duplicate_topup() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock() + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=True), + ), + ): + await _check_and_topup(_ppq_row()) + + provider.get_balance.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_ppq_auto_topup_rejects_non_finite_balance() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=float("nan")) + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=False), + ), + patch("routstr.upstream.auto_topup._claim_ppq_topup", AsyncMock()) as claim, + ): + await _check_and_topup(_ppq_row()) + + claim.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_routstr_threshold_is_compared_in_sats() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=100) + provider.topup = AsyncMock() + + with ( + patch( + "routstr.upstream.auto_topup.RoutstrUpstreamProvider.from_db_row", + return_value=provider, + ), + patch("routstr.upstream.auto_topup.send_token", AsyncMock()) as send, + ): + await _check_and_topup(_row()) + + send.assert_not_awaited() + provider.topup.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_settled_topup_alerts_when_its_claim_was_already_released() -> None: + provider = MagicMock() + provider.get_balance = AsyncMock(return_value=2.5) + provider.initiate_topup = AsyncMock( + return_value=MagicMock( + invoice_id="invoice-1", + payment_request="lnbc-invoice", + amount=10, + currency="USD", + expires_at=None, + ) + ) + provider.check_topup_status = AsyncMock(return_value=True) + plan = MagicMock() + plan.maximum_spend_sats = 102 + plan.quote.amount = 100 + plan.quote.fee_reserve = 2 + plan.mint_url = "https://mint-rich.test" + plan.unit = "sat" + + with ( + patch( + "routstr.upstream.auto_topup.PPQAIUpstreamProvider.from_db_row", + return_value=provider, + ), + patch( + "routstr.upstream.auto_topup._reconcile_ppq_state", + AsyncMock(return_value=False), + ), + patch( + "routstr.upstream.auto_topup._claim_ppq_topup", + AsyncMock(return_value="operation-1"), + ), + patch( + "routstr.upstream.auto_topup.prepare_bolt11_payment", + AsyncMock(return_value=plan), + ), + patch( + "routstr.upstream.auto_topup.execute_bolt11_payment", + AsyncMock(return_value=(101, "https://mint-rich.test", "sat")), + ), + patch("routstr.upstream.auto_topup._record_ppq_invoice", AsyncMock()), + patch( + "routstr.upstream.auto_topup._set_ppq_state_terminal", + AsyncMock(return_value=False), + ), + patch("routstr.upstream.auto_topup.sats_usd_price", return_value=0.001), + patch("routstr.upstream.auto_topup.logger") as log, + ): + await _check_and_topup(_ppq_row()) + + assert any( + "claim was already released" in call.args[0] + for call in log.critical.call_args_list + ) + + +@pytest.mark.parametrize( + ("provider_type", "settings", "expected"), + [ + ("ppqai", {"auto_topup": False, "topup_threshold": -1}, None), + ("ppqai", {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 10}, None), + ( + "ppqai", + {"auto_topup": True, "topup_threshold": None, "topup_amount_limit": 10}, + "threshold", + ), + ( + "ppqai", + {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 0.5}, + "whole number", + ), + ( + "ppqai", + {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 5000}, + "between", + ), + ( + "ppqai", + {"auto_topup": True, "topup_threshold": True, "topup_amount_limit": 10}, + "threshold", + ), + ( + "routstr", + {"auto_topup": True, "topup_threshold": 5, "topup_amount_limit": 10}, + "mint URL", + ), + ( + "routstr", + { + "auto_topup": True, + "topup_threshold": 5, + "topup_amount_limit": 10, + "topup_mint_url": "https://mint.test", + }, + None, + ), + ], +) +def test_auto_topup_settings_validation( + provider_type: str, settings: dict, expected: str | None +) -> None: + problem = validate_auto_topup_settings(provider_type, settings) + if expected is None: + assert problem is None + else: + assert problem is not None and expected in problem + + +def test_auto_topup_settings_validation_survives_huge_json_integers() -> None: + # json.loads happily produces integers past float range; float() raises + # OverflowError there instead of returning inf. + problem = validate_auto_topup_settings( + "ppqai", + {"auto_topup": True, "topup_threshold": 10**400, "topup_amount_limit": 10}, + ) + assert problem is not None and "threshold" in problem diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index 61506b4a..be9c1465 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -1,19 +1,24 @@ import base64 import json import socket -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import AsyncMock, MagicMock, Mock, patch import httpx import pytest from routstr.core.db import ApiKey from routstr.wallet import ( + Bolt11PaymentAmbiguous, + Bolt11PaymentNotAttempted, MintConnectionError, TokenConsumedError, classify_redemption_error, credit_balance, + execute_bolt11_payment, get_balance, is_mint_connection_error, + pay_bolt11_invoice, + prepare_bolt11_payment, recieve_token, send, send_token, @@ -1389,3 +1394,176 @@ async def test_swap_melt_transport_error_raises_mint_connection_error() -> None: await swap_to_primary_mint(mock_token, mock_token_wallet) assert mock_token_wallet.melt.call_count == 1 + + +@pytest.mark.asyncio +async def test_pay_bolt11_invoice_uses_sufficient_mint_with_highest_balance() -> None: + from routstr.core.settings import settings + + low_wallet = MagicMock() + high_wallet = MagicMock() + low_proof = MagicMock(amount=200) + high_proof = MagicMock(amount=500) + low_wallet.proofs = [low_proof] + high_wallet.proofs = [high_proof] + + for wallet in (low_wallet, high_wallet): + wallet.melt_quote = AsyncMock( + return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1") + ) + wallet.get_fees_for_proofs = Mock(return_value=0) + wallet.select_to_send = AsyncMock( + side_effect=lambda proofs, *args, **kwargs: (proofs, 0) + ) + wallet.melt = AsyncMock( + return_value=MagicMock( + state="PAID", change=[MagicMock(amount=399)] + ) + ) + wallet.set_reserved_for_send = AsyncMock() + + async def get_wallet(mint_url: str, unit: str = "sat") -> MagicMock: + if unit == "msat": + raise ValueError("unit unsupported") + return high_wallet if mint_url == "https://high.test" else low_wallet + + with ( + patch.object( + settings, + "cashu_mints", + ["https://low.test", "https://high.test"], + ), + patch.object(settings, "primary_mint", "https://low.test"), + patch("routstr.wallet.get_wallet", side_effect=get_wallet), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + side_effect=lambda wallet, *args, **kwargs: wallet.proofs, + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + side_effect=lambda proofs, wallet: proofs, + ), + ): + amount, mint_url, unit = await pay_bolt11_invoice("lnbc-invoice") + + assert (amount, mint_url, unit) == (101, "https://high.test", "sat") + high_wallet.melt.assert_awaited_once() + low_wallet.melt.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_execute_bolt11_payment_rejects_unpaid_melt_state() -> None: + plan = MagicMock() + plan.proofs = [MagicMock(amount=110)] + plan.quote.amount = 100 + plan.quote.fee_reserve = 10 + plan.quote.quote = "quote-1" + plan.invoice = "lnbc-invoice" + plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0)) + plan.wallet.set_reserved_for_send = AsyncMock() + plan.wallet.melt = AsyncMock(return_value=MagicMock(state="UNPAID", change=[])) + + with pytest.raises(Bolt11PaymentNotAttempted): + await execute_bolt11_payment(plan) + + # An explicit unpaid answer means the proofs are ours again. + plan.wallet.set_reserved_for_send.assert_awaited_with(plan.proofs, reserved=False) + + +@pytest.mark.asyncio +async def test_execute_bolt11_payment_accepts_legacy_paid_response() -> None: + plan = MagicMock() + plan.proofs = [MagicMock(amount=110)] + plan.quote.amount = 100 + plan.quote.fee_reserve = 10 + plan.quote.quote = "quote-1" + plan.invoice = "lnbc-invoice" + plan.mint_url = "https://mint.test" + plan.unit = "sat" + plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0)) + plan.wallet.set_reserved_for_send = AsyncMock() + plan.wallet.melt = AsyncMock( + return_value=MagicMock(state=None, paid=True, change=[]) + ) + + assert await execute_bolt11_payment(plan) == ( + 110, + "https://mint.test", + "sat", + ) + + +@pytest.mark.asyncio +async def test_execute_bolt11_payment_keeps_proofs_reserved_when_melt_errors() -> None: + plan = MagicMock() + plan.proofs = [MagicMock(amount=110)] + plan.quote.amount = 100 + plan.quote.fee_reserve = 10 + plan.quote.quote = "quote-1" + plan.invoice = "lnbc-invoice" + plan.wallet.select_to_send = AsyncMock(return_value=(plan.proofs, 0)) + plan.wallet.set_reserved_for_send = AsyncMock() + plan.wallet.set_reserved_for_melt = AsyncMock() + plan.wallet.melt = AsyncMock(side_effect=TimeoutError("no answer")) + + with pytest.raises(Bolt11PaymentAmbiguous): + await execute_bolt11_payment(plan) + + # The mint may still settle with these proofs. cashu's own melt() + # un-reserves them on a mint transport error, so the ambiguous path must + # re-reserve — and it must do so with the melt quote id, because + # get_melt_quote() finds the proofs to settle by melt_id. + plan.wallet.set_reserved_for_melt.assert_awaited_once_with( + plan.proofs, reserved=True, quote_id="quote-1" + ) + + +@pytest.mark.asyncio +async def test_execute_bolt11_payment_does_not_reserve_when_selection_fails() -> None: + plan = MagicMock() + plan.proofs = [MagicMock(amount=110)] + plan.quote.amount = 100 + plan.quote.fee_reserve = 10 + plan.wallet.select_to_send = AsyncMock(side_effect=ValueError("insufficient")) + plan.wallet.set_reserved_for_send = AsyncMock() + plan.wallet.melt = AsyncMock() + + with pytest.raises(Bolt11PaymentNotAttempted): + await execute_bolt11_payment(plan) + + plan.wallet.set_reserved_for_send.assert_not_awaited() + plan.wallet.melt.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_prepare_bolt11_payment_counts_input_fees_in_sufficiency() -> None: + from routstr.core.settings import settings + + wallet = MagicMock() + wallet.proofs = [MagicMock(amount=105)] + wallet.melt_quote = AsyncMock( + return_value=MagicMock(amount=100, fee_reserve=2, quote="quote-1") + ) + # Balance covers amount + fee_reserve (102) but not the 5 sat input fee. + wallet.get_fees_for_proofs = Mock(return_value=5) + + async def get_wallet(mint_url: str, unit: str = "sat") -> MagicMock: + if unit == "msat": + raise ValueError("unit unsupported") + return wallet + + with ( + patch.object(settings, "cashu_mints", ["https://only.test"]), + patch.object(settings, "primary_mint", "https://only.test"), + patch("routstr.wallet.get_wallet", side_effect=get_wallet), + patch( + "routstr.wallet.get_proofs_per_mint_and_unit", + side_effect=lambda wallet, *args, **kwargs: wallet.proofs, + ), + patch( + "routstr.wallet.slow_filter_spend_proofs", + side_effect=lambda proofs, wallet: proofs, + ), + pytest.raises(ValueError, match="enough balance"), + ): + await prepare_bolt11_payment("lnbc-invoice") diff --git a/ui/components/provider-card.tsx b/ui/components/provider-card.tsx index 851dda85..3cdce61d 100644 --- a/ui/components/provider-card.tsx +++ b/ui/components/provider-card.tsx @@ -1,3 +1,5 @@ +import { AdminService } from '@/lib/api/services/admin'; +import type { PPQAutoTopupState } from '@/lib/api/services/admin'; import type { AdminModel, ProviderModels, @@ -20,12 +22,15 @@ import { Trash2, Key, RotateCcw, + AlertTriangle, + Unlock, + Loader2, } from 'lucide-react'; import { ProviderBalance } from '@/components/provider-balance'; import { ProviderModelsPanel } from '@/components/provider-models-panel'; import { RoutstrCreateKeySection } from '@/components/providers/RoutstrCreateKeySection'; import { RoutstrProviderService } from '@/lib/api/services/routstr-provider'; -import { useMutation, useQueryClient } from '@tanstack/react-query'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { useState } from 'react'; import { toast } from 'sonner'; import { cn } from '@/lib/utils'; @@ -36,6 +41,16 @@ import { DialogHeader, DialogTitle, } from '@/components/ui/dialog'; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, +} from '@/components/ui/alert-dialog'; interface ProviderCardProps { provider: UpstreamProvider; @@ -77,8 +92,60 @@ export function ProviderCard({ }: ProviderCardProps) { const queryClient = useQueryClient(); const [isKeyModalOpen, setIsKeyModalOpen] = useState(false); + const [isReleaseDialogOpen, setIsReleaseDialogOpen] = useState(false); + // Snapshot of the claim as it looked when the admin opened the dialog. + // The mutation sends this token, never the live query data: a background + // refetch must not swap in a state the admin never reviewed. + const [reviewedState, setReviewedState] = useState( + null + ); const hasDetails = Boolean(provider.api_version) || isExpanded; const isRoutstr = provider.provider_type === 'routstr'; + const isPPQ = provider.provider_type === 'ppqai'; + + const { data: ppqAutoTopupState } = useQuery({ + queryKey: ['ppq-auto-topup-state', provider.id], + queryFn: () => AdminService.getPPQAutoTopupState(provider.id), + enabled: isPPQ, + refetchInterval: 30000, + }); + + // A claim the server will not let us release: a worker is between reserving + // proofs and hearing back from the mint, and sweeping it would let the next + // cycle pay a second invoice. + const isPPQPaymentInFlight = + Boolean(ppqAutoTopupState?.active) && + ppqAutoTopupState?.releasable === false; + + const openReleaseDialog = () => { + setReviewedState(ppqAutoTopupState ?? null); + setIsReleaseDialogOpen(true); + }; + + const releasePPQMutation = useMutation({ + mutationFn: () => + AdminService.releasePPQAutoTopup( + provider.id, + reviewedState?.state_token ?? null + ), + onSuccess: () => { + queryClient.invalidateQueries({ + queryKey: ['ppq-auto-topup-state', provider.id], + }); + setIsReleaseDialogOpen(false); + setReviewedState(null); + toast.success('PPQ auto top-up claim released'); + }, + onError: (error: Error) => { + // A 409 usually means the claim changed since it was reviewed. + queryClient.invalidateQueries({ + queryKey: ['ppq-auto-topup-state', provider.id], + }); + setIsReleaseDialogOpen(false); + setReviewedState(null); + toast.error(`Failed to release PPQ claim: ${error.message}`); + }, + }); const refundMutation = useMutation({ mutationFn: () => RoutstrProviderService.refundBalance(provider.id), @@ -113,6 +180,26 @@ export function ProviderCard({ > {provider.enabled ? 'Enabled' : 'Disabled'} + {ppqAutoTopupState?.active && ( + + {isPPQPaymentInFlight ? ( + + ) : ( + + )} + {isPPQPaymentInFlight + ? 'Paying invoice' + : 'Auto top-up needs review'} + + )} {provider.base_url} @@ -153,6 +240,19 @@ export function ProviderCard({ )} + {isPPQ && ppqAutoTopupState?.active && !isPPQPaymentInFlight && ( + + )} + {isRoutstr && provider.api_key && (