routstr-core/routstr/upstream/auto_topup.py

import asyncio
import json
import math
import time
import typing
import uuid

from sqlalchemy.exc import IntegrityError
from sqlmodel import col, or_, select, update
from sqlmodel.ext.asyncio.session import AsyncSession

from ..core import get_logger
from ..core.db import (
    CashuTransaction,
    UpstreamProviderRow,
    create_session,
)
from ..payment.price import sats_usd_price
from ..wallet import (
    Bolt11PaymentAmbiguous,
    Bolt11PaymentNotAttempted,
    check_bolt11_payment_status,
    execute_bolt11_payment,
    maximum_owner_cashu_balance_sats,
    prepare_bolt11_payment,
    release_token_reservation,
    send_token_from_owner_locked,
    token_mint_url,
    wallet_operation_guard,
)
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_SETTLED_COOLDOWN_SECONDS = 5 * 60
PPQ_MAX_INVOICE_PREMIUM = 1.10
PPQ_MIN_TOPUP_USD = 1
PPQ_MAX_TOPUP_USD = 500
# Rolling 24h ceiling on total PPQ auto top-up spend across all providers.
# Independent of PPQ's own balance endpoint: if that endpoint is buggy or
# compromised and keeps reporting a below-threshold balance, this cap bounds
# the damage instead of letting the worker drain the owner's mint funds one
# per-transaction-capped payment at a time.
PPQ_MAX_DAILY_TOPUP_USD = 300
# Routstr-to-Routstr claim lifecycle. "claimed" holds the slot while the token
# is being minted; nothing has left the wallet yet. "sent" means a bearer token
# was handed to the peer and only the peer's balance can say whether it landed.
# "backoff" holds the failure count between attempts, and "halted" stops the
# provider entirely until an admin releases it.
ROUTSTR_PHASE_CLAIMED = "claimed"
ROUTSTR_PHASE_SENT = "sent"
ROUTSTR_PHASE_BACKOFF = "backoff"
ROUTSTR_PHASE_HALTED = "halted"
ROUTSTR_PHASES = frozenset(
    {
        ROUTSTR_PHASE_CLAIMED,
        ROUTSTR_PHASE_SENT,
        ROUTSTR_PHASE_BACKOFF,
        ROUTSTR_PHASE_HALTED,
    }
)
ROUTSTR_PENDING_TTL_SECONDS = 15 * 60
ROUTSTR_BACKOFF_BASE_SECONDS = 15 * 60
ROUTSTR_MAX_TOPUP_FAILURES = 3
ROUTSTR_MIN_TOPUP_SATS = 1
ROUTSTR_MAX_TOPUP_SATS = 1_000_000
# Rolling 24h ceiling on total Routstr auto top-up spend across all peers. The
# per-attempt claim already stops a peer from being paid twice for the same
# uncredited token; this bounds the total even when every attempt is credited
# and the peer simply keeps reporting a below-threshold balance.
ROUTSTR_MAX_DAILY_TOPUP_SATS = 2_000_000


async def periodic_auto_topup() -> None:
    """Background task that monitors Routstr and PPQ provider balances."""
    # Wait for initial startup to complete
    await asyncio.sleep(30)
    logger.info("Auto top-up worker started")

    while True:
        try:
            await _run_auto_topup_cycle()
        except Exception as e:
            logger.error(
                "Auto top-up cycle failed",
                extra={"error": repr(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."""
    reconciled_ppq_provider_ids = await _reconcile_all_ppq_claims()

    async with create_session() as session:
        query = select(UpstreamProviderRow).where(
            col(UpstreamProviderRow.provider_type).in_(["routstr", "ppqai"]),
            UpstreamProviderRow.enabled == True,  # noqa: E712
        )
        result = await session.exec(query)
        providers = result.all()

    for row in providers:
        # Do not immediately retry a PPQ claim reconciled this cycle: PPQ's
        # balance endpoint may lag its invoice status and cause a duplicate.
        if row.id is not None and row.id in reconciled_ppq_provider_ids:
            continue
        try:
            await _check_and_topup(row)
        except Exception as e:
            logger.error(
                "Auto top-up failed for provider",
                extra={
                    "provider_id": row.id,
                    "base_url": row.base_url,
                    "error": repr(e),
                    "error_type": type(e).__name__,
                },
            )


async def _reconcile_all_ppq_claims() -> set[int]:
    """Reconcile active PPQ claims and return providers suppressed this cycle."""
    async with create_session() as session:
        rows = (
            await session.exec(
                select(UpstreamProviderRow).where(
                    col(UpstreamProviderRow.provider_type) == "ppqai"
                )
            )
        ).all()

    active_provider_ids: set[int] = set()
    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
            active_provider_ids.add(row.id)
            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": repr(e),
                    "error_type": type(e).__name__,
                },
            )
    return active_provider_ids


def _invalid_topup_number(value: object, *, integer: bool = False) -> bool:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return True
    try:
        number = float(value)
    except OverflowError:
        return True
    return (
        not math.isfinite(number)
        or number <= 0
        or (integer and not number.is_integer())
    )


def validate_ppq_auto_topup_settings(settings: dict | None) -> str | None:
    """Return why enabled PPQ auto top-up settings are invalid, if anything."""
    if not settings or not settings.get("auto_topup"):
        return None

    threshold = settings.get("topup_threshold")
    amount = settings.get("topup_amount_limit")
    if _invalid_topup_number(threshold):
        return "PPQ auto top-up threshold must be a positive number"
    if _invalid_topup_number(amount, integer=True):
        return "PPQ auto top-up amount must be a positive whole number"
    amount_usd = int(typing.cast(int | float, amount))
    if not PPQ_MIN_TOPUP_USD <= amount_usd <= 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


_legacy_threshold_hinted: set[int] = set()


def _routstr_threshold_sats(row: UpstreamProviderRow, settings: dict) -> float:
    """Balance, in sats, below which a top-up fires.

    ``topup_threshold_sats`` is used as written. A legacy ``topup_threshold``
    keeps the thousandfold it has always been compared with — reinterpreting it
    as sats would drop an operator's trigger point by a factor of 1000 on
    upgrade and leave the peer to run dry. The hint names the value to migrate
    to, once per provider rather than once per scheduler tick.
    """
    explicit = settings.get("topup_threshold_sats")
    if explicit is not None:
        return float(typing.cast(int | float, explicit))

    threshold_sats = float(typing.cast(int | float, settings["topup_threshold"])) * 1000
    if row.id is not None and row.id not in _legacy_threshold_hinted:
        _legacy_threshold_hinted.add(row.id)
        logger.warning(
            "Routstr auto top-up uses the legacy unitless threshold; set "
            "topup_threshold_sats to state the unit",
            extra={"provider_id": row.id, "threshold_sats": threshold_sats},
        )
    return threshold_sats


def validate_routstr_auto_topup_settings(settings: dict | None) -> str | None:
    """Return why enabled Routstr auto top-up settings are invalid, if anything."""
    if not settings or not settings.get("auto_topup"):
        return None

    threshold_sats = settings.get("topup_threshold_sats")
    legacy_threshold = settings.get("topup_threshold")
    amount = settings.get("topup_amount_limit")
    mint_url = settings.get("topup_mint_url")
    if threshold_sats is not None:
        if _invalid_topup_number(threshold_sats):
            return "Routstr auto top-up threshold must be a positive number of sats"
    elif legacy_threshold is None:
        return "Routstr auto top-up requires a threshold"
    elif _invalid_topup_number(legacy_threshold):
        return "Routstr auto top-up threshold must be a positive number"
    if _invalid_topup_number(amount, integer=True):
        return "Routstr auto top-up amount must be a positive whole number"
    amount_sats = int(typing.cast(int | float, amount))
    if not ROUTSTR_MIN_TOPUP_SATS <= amount_sats <= ROUTSTR_MAX_TOPUP_SATS:
        return (
            f"Routstr auto top-up amount must be between {ROUTSTR_MIN_TOPUP_SATS} "
            f"and {ROUTSTR_MAX_TOPUP_SATS} sats"
        )
    if not isinstance(mint_url, str) or not mint_url.strip():
        return "Routstr auto top-up requires a mint URL"
    return None


async def _check_and_topup_ppq_from_row(row: UpstreamProviderRow) -> None:
    settings: dict = {}
    if row.provider_settings:
        try:
            settings = json.loads(row.provider_settings)
        except (json.JSONDecodeError, TypeError):
            return
    if not isinstance(settings, dict) or not settings.get("auto_topup"):
        return

    problem = validate_ppq_auto_topup_settings(settings)
    if problem is not None:
        logger.warning(
            "PPQ auto top-up enabled but its configuration is invalid",
            extra={"provider_id": row.id, "problem": problem},
        )
        return
    if not row.api_key:
        return
    await _check_and_topup_ppq(row, settings)


async def _check_and_topup(row: UpstreamProviderRow) -> None:
    """Check a single provider's balance and top up if below threshold."""
    if row.provider_type == "ppqai":
        await _check_and_topup_ppq_from_row(row)
        return

    # Parse provider settings
    settings: dict = {}
    if row.provider_settings:
        try:
            settings = json.loads(row.provider_settings)
        except (json.JSONDecodeError, TypeError):
            return

    if not settings.get("auto_topup"):
        return

    problem = validate_routstr_auto_topup_settings(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

    threshold_sats = _routstr_threshold_sats(row, settings)
    amount = int(settings["topup_amount_limit"])
    mint_url = str(settings["topup_mint_url"])

    if not row.api_key:
        return

    # Instantiate provider and check balance
    provider = RoutstrUpstreamProvider.from_db_row(row)
    if provider is None:
        return
    if await _reconcile_routstr_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 balance for auto top-up",
            extra={"provider_id": row.id, "base_url": row.base_url},
        )
        return

    if balance >= threshold_sats:
        return

    spent_24h_sats = await _routstr_spent_last_24h_sats()
    if spent_24h_sats + amount > ROUTSTR_MAX_DAILY_TOPUP_SATS:
        logger.critical(
            "Auto top-up skipped: rolling 24h spend cap reached",
            extra={
                "provider_id": row.id,
                "spent_24h_sats": spent_24h_sats,
                "topup_amount": amount,
                "daily_cap_sats": ROUTSTR_MAX_DAILY_TOPUP_SATS,
            },
        )
        return

    # The balance the peer must report before another token may be sent. Any
    # shortfall is treated as "not credited": the token is a bearer instrument
    # and a peer that took one without crediting it must not be handed another.
    expected_sats = math.floor(balance) + amount
    operation_id = await _claim_routstr_topup(row, expected_sats=expected_sats)
    if operation_id is None:
        return

    # Balance is below threshold - create token and top up
    logger.info(
        "Auto top-up triggered",
        extra={
            "provider_id": row.id,
            "balance": balance,
            "threshold_sats": threshold_sats,
            "topup_amount": amount,
            "mint_url": mint_url,
        },
    )

    try:
        async with wallet_operation_guard():
            # Keep the spend cap and audit mutation in one wallet lock.
            spent_24h_sats = await _routstr_spent_last_24h_sats()
            if spent_24h_sats + amount > ROUTSTR_MAX_DAILY_TOPUP_SATS:
                raise ValueError("Routstr auto top-up daily spend cap reached")
            token = await send_token_from_owner_locked(amount, "sat", mint_url)
            actual_mint_url = token_mint_url(token, mint_url)
            try:
                await _persist_routstr_token_and_mark_sent(
                    row,
                    operation_id,
                    expected_sats=expected_sats,
                    token=token,
                    amount=amount,
                    mint_url=actual_mint_url,
                )
            except Exception:
                logger.critical(
                    "Aborting auto top-up because its token and sent claim "
                    "could not be persisted atomically",
                    extra={"provider_id": row.id, "mint_url": actual_mint_url},
                )
                try:
                    await release_token_reservation(token)
                except Exception as error:
                    logger.critical(
                        "Failed to release untracked auto-topup token",
                        extra={
                            "provider_id": row.id,
                            "mint_url": actual_mint_url,
                            "error": repr(error),
                        },
                    )
                else:
                    logger.warning(
                        "Auto-topup token was released after persistence failed",
                        extra={"provider_id": row.id, "mint_url": actual_mint_url},
                    )
                raise
    except Exception as e:
        logger.warning(
            "Failed to create or persist cashu token for auto top-up",
            extra={
                "provider_id": row.id,
                "amount": amount,
                "mint_url": mint_url,
                "error": repr(e),
                "error_type": type(e).__name__,
            },
        )
        await _release_routstr_claim(row, operation_id)
        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"],
            },
        )
    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()

        logger.info(
            "Auto top-up completed successfully",
            extra={
                "provider_id": row.id,
                "amount": amount,
                "new_balance_approx": balance + amount,
            },
        )


def _routstr_state_id(row: UpstreamProviderRow) -> str:
    if row.id is None:
        raise ValueError("Routstr auto top-up requires a persisted provider row")
    return _routstr_state_id_for_provider(row.id)


def _routstr_state_id_for_provider(provider_id: int | str) -> str:
    return f"routstr-auto-topup-{provider_id}"


class RoutstrClaim(typing.NamedTuple):
    operation_id: str
    # Worker lease for "claimed"/"sent", retry-not-before for "backoff", and
    # meaningless for "halted".
    deadline: int
    phase: str
    # Peer balance in sats that proves this attempt was credited.
    expected_sats: int
    failures: int


def _routstr_request_id(
    operation_id: str,
    deadline: int,
    phase: str,
    expected_sats: int,
    failures: int,
) -> str:
    return f"routstr:{operation_id}:{deadline}:{phase}:{expected_sats}:{failures}"


def _parse_routstr_request_id(request_id: str | None) -> RoutstrClaim | None:
    parts = (request_id or "").split(":", 5)
    if len(parts) != 6 or parts[0] != "routstr" or parts[3] not in ROUTSTR_PHASES:
        return None
    try:
        deadline = int(parts[2])
        expected_sats = int(parts[4])
        failures = int(parts[5])
    except (TypeError, ValueError):
        return None
    return RoutstrClaim(parts[1], deadline, parts[3], expected_sats, failures)


async def _routstr_spent_last_24h_sats() -> int:
    """Total sats committed to Routstr auto top-ups in the last 24 hours.

    Uncollected rows count too: a token whose delivery is unconfirmed is spent
    for capping purposes. Rows marked ``collected=False, swept=True`` record
    tokens that were provably returned to the wallet and are excluded.
    """
    cutoff = int(time.time()) - 24 * 60 * 60
    async with create_session() as session:
        rows = (
            await session.exec(
                select(CashuTransaction.amount, CashuTransaction.unit).where(
                    col(CashuTransaction.source) == "auto_topup",
                    col(CashuTransaction.type) == "out",
                    col(CashuTransaction.created_at) >= cutoff,
                    or_(
                        col(CashuTransaction.collected) == True,  # noqa: E712
                        col(CashuTransaction.swept) == False,  # noqa: E712
                    ),
                )
            )
        ).all()
    return sum(
        amount if unit == "sat" else math.ceil(amount / 1000) for amount, unit in rows
    )


async def _routstr_provider_is_claimable(
    session: AsyncSession, provider_id: int | str | None
) -> bool:
    """Re-read the provider inside the claim transaction.

    Same reasoning as :func:`_ppq_provider_is_claimable`: a provider deleted or
    retyped concurrently must either be visible here or lose the race against
    the claim we are about to write.
    """
    if provider_id is None:
        return False
    current = await session.get(UpstreamProviderRow, provider_id)
    return current is not None and current.provider_type == "routstr"


async def _claim_routstr_topup(
    row: UpstreamProviderRow, *, expected_sats: int
) -> str | None:
    """Acquire the provider's single durable auto top-up slot.

    An expired backoff hands its failure count to the new attempt, so repeated
    non-crediting peers still walk towards the halt instead of resetting the
    counter every cycle.
    """
    state_id = _routstr_state_id(row)
    operation_id = uuid.uuid4().hex
    deadline = int(time.time()) + ROUTSTR_PENDING_TTL_SECONDS

    async with create_session() as session:
        if not await _routstr_provider_is_claimable(session, row.id):
            return None
        existing = await session.get(CashuTransaction, state_id)
        if existing is not None:
            failures = 0
            if not (existing.collected or existing.swept):
                claim = _parse_routstr_request_id(existing.request_id)
                if (
                    claim is None
                    or claim.phase != ROUTSTR_PHASE_BACKOFF
                    or time.time() < claim.deadline
                ):
                    return None
                failures = claim.failures
            result = await session.exec(  # type: ignore[call-overload]
                update(CashuTransaction)
                .where(
                    col(CashuTransaction.id) == state_id,
                    # Fence on the exact row that was read: any concurrent
                    # writer that moved the claim must win instead of us.
                    col(CashuTransaction.request_id) == existing.request_id,
                )
                .values(
                    token="pending",
                    amount=0,
                    unit="sat",
                    mint_url=None,
                    request_id=_routstr_request_id(
                        operation_id,
                        deadline,
                        ROUTSTR_PHASE_CLAIMED,
                        expected_sats,
                        failures,
                    ),
                    collected=False,
                    swept=False,
                    created_at=int(time.time()),
                    source="routstr_auto_topup_claim",
                )
            )
            await session.commit()
            if (getattr(result, "rowcount", 0) or 0) != 1:
                return None
            return operation_id

    try:
        async with create_session() as session:
            if not await _routstr_provider_is_claimable(session, row.id):
                return None
            session.add(
                CashuTransaction(
                    id=state_id,
                    token="pending",
                    amount=0,
                    unit="sat",
                    type="out",
                    request_id=_routstr_request_id(
                        operation_id,
                        deadline,
                        ROUTSTR_PHASE_CLAIMED,
                        expected_sats,
                        0,
                    ),
                    collected=False,
                    source="routstr_auto_topup_claim",
                )
            )
            await session.commit()
    except IntegrityError:
        return None
    return operation_id


async def _advance_routstr_claim(
    row: UpstreamProviderRow,
    operation_id: str,
    *,
    deadline: int,
    phase: str,
    expected_sats: int,
    failures: int,
    token: str | None = None,
    amount: int | None = None,
    mint_url: str | None = None,
) -> bool:
    """Move this worker's claim to another phase, if it still owns it."""
    values: dict[str, object] = {
        "request_id": _routstr_request_id(
            operation_id, deadline, phase, expected_sats, failures
        )
    }
    if token is not None:
        values.update(token=token, amount=amount, mint_url=mint_url)

    async with create_session() as session:
        result = await session.exec(  # type: ignore[call-overload]
            update(CashuTransaction)
            .where(
                col(CashuTransaction.id) == _routstr_state_id(row),
                col(CashuTransaction.request_id).like(f"routstr:{operation_id}:%"),
                col(CashuTransaction.collected) == False,  # noqa: E712
                col(CashuTransaction.swept) == False,  # noqa: E712
            )
            .values(**values)
        )
        await session.commit()
        return (getattr(result, "rowcount", 0) or 0) == 1


async def _set_routstr_state_terminal(
    row: UpstreamProviderRow, operation_id: str, *, collected: bool, swept: bool
) -> bool:
    """Finish an 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) == _routstr_state_id(row),
                col(CashuTransaction.request_id).like(f"routstr:{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 _release_routstr_claim(row: UpstreamProviderRow, operation_id: str) -> None:
    """Hand back a claim whose token never left the wallet."""
    if not await _set_routstr_state_terminal(
        row, operation_id, collected=False, swept=True
    ):
        logger.warning(
            "Could not release the auto top-up claim after a pre-send failure; "
            "it is owned by another attempt",
            extra={"provider_id": row.id},
        )


async def _persist_routstr_token_and_mark_sent(
    row: UpstreamProviderRow,
    operation_id: str,
    *,
    expected_sats: int,
    token: str,
    amount: int,
    mint_url: str,
) -> None:
    state_id = _routstr_state_id(row)
    async with create_session() as session:
        state = await session.get(CashuTransaction, state_id)
        claim = _parse_routstr_request_id(state.request_id if state else None)
        if (
            state is None
            or state.collected
            or state.swept
            or claim is None
            or claim.operation_id != operation_id
            or claim.phase != ROUTSTR_PHASE_CLAIMED
        ):
            raise RuntimeError("Routstr auto top-up claim ownership was lost")

        result = await session.exec(  # type: ignore[call-overload]
            update(CashuTransaction)
            .where(
                col(CashuTransaction.id) == state_id,
                col(CashuTransaction.request_id) == state.request_id,
                col(CashuTransaction.collected) == False,  # noqa: E712
                col(CashuTransaction.swept) == False,  # noqa: E712
            )
            .values(
                request_id=_routstr_request_id(
                    operation_id,
                    int(time.time()) + ROUTSTR_PENDING_TTL_SECONDS,
                    ROUTSTR_PHASE_SENT,
                    expected_sats,
                    claim.failures,
                ),
                token=token,
                amount=amount,
                unit="sat",
                mint_url=mint_url,
            )
        )
        if (getattr(result, "rowcount", 0) or 0) != 1:
            await session.rollback()
            raise RuntimeError("Routstr auto top-up claim ownership was lost")

        session.add(
            CashuTransaction(
                id=uuid.uuid4().hex,
                token=token,
                amount=amount,
                unit="sat",
                mint_url=mint_url,
                type="out",
                collected=False,
                source="auto_topup",
            )
        )
        await session.commit()


async def _current_routstr_claim(row: UpstreamProviderRow) -> RoutstrClaim | None:
    async with create_session() as session:
        transaction = await session.get(CashuTransaction, _routstr_state_id(row))
    if transaction is None:
        return None
    return _parse_routstr_request_id(transaction.request_id)


async def _reconcile_routstr_state(
    row: UpstreamProviderRow, provider: RoutstrUpstreamProvider
) -> bool:
    """Return True while a prior attempt must suppress a new payment."""
    async with create_session() as session:
        transaction = await session.get(CashuTransaction, _routstr_state_id(row))
    if transaction is None or transaction.collected or transaction.swept:
        return False

    claim = _parse_routstr_request_id(transaction.request_id)
    if claim is None:
        logger.critical(
            "Malformed auto top-up state; suppressing duplicate payment",
            extra={"provider_id": row.id},
        )
        return True

    now = time.time()
    if claim.phase == ROUTSTR_PHASE_HALTED:
        return True
    if claim.phase == ROUTSTR_PHASE_BACKOFF:
        return now < claim.deadline
    if claim.phase == ROUTSTR_PHASE_CLAIMED:
        # Nothing left the wallet, so a dead worker's slot is free to reuse.
        if now < claim.deadline:
            return True
        return not await _set_routstr_state_terminal(
            row, claim.operation_id, collected=False, swept=True
        )

    balance = await provider.get_balance()
    if (
        balance is not None
        and math.isfinite(balance)
        and balance >= claim.expected_sats
    ):
        if not await _set_routstr_state_terminal(
            row, claim.operation_id, collected=True, swept=False
        ):
            logger.critical(
                "Auto top-up was credited but its claim was already released; "
                "a duplicate top-up is possible on the next cycle",
                extra={"provider_id": row.id},
            )
        return True
    if now < claim.deadline:
        return True

    failures = claim.failures + 1
    if failures >= ROUTSTR_MAX_TOPUP_FAILURES:
        await _advance_routstr_claim(
            row,
            claim.operation_id,
            deadline=claim.deadline,
            phase=ROUTSTR_PHASE_HALTED,
            expected_sats=claim.expected_sats,
            failures=failures,
        )
        logger.critical(
            "Auto top-up halted: the peer repeatedly failed to credit a token",
            extra={
                "provider_id": row.id,
                "base_url": row.base_url,
                "failures": failures,
                "admin_action": (
                    f"POST /admin/api/upstream-providers/{row.id}"
                    "/routstr-auto-topup/release"
                ),
            },
        )
        return True

    await _advance_routstr_claim(
        row,
        claim.operation_id,
        deadline=int(now) + ROUTSTR_BACKOFF_BASE_SECONDS * 2 ** (failures - 1),
        phase=ROUTSTR_PHASE_BACKOFF,
        expected_sats=claim.expected_sats,
        failures=failures,
    )
    logger.warning(
        "Auto top-up was not credited by the peer; backing off",
        extra={
            "provider_id": row.id,
            "base_url": row.base_url,
            "expected_sats": claim.expected_sats,
            "failures": failures,
        },
    )
    return True


async def get_routstr_auto_topup_state(provider_id: int) -> dict[str, object]:
    """Return admin-safe state for a provider's durable Routstr claim."""
    async with create_session() as session:
        transaction = await session.get(
            CashuTransaction, _routstr_state_id_for_provider(provider_id)
        )
    if transaction is None or transaction.collected or transaction.swept:
        return {"active": False}

    claim = _parse_routstr_request_id(transaction.request_id)
    return {
        "active": True,
        # Echoed back verbatim on release so a claim that moved on since the
        # admin reviewed it fails the write instead of being swept unseen.
        "state_token": transaction.request_id,
        "operation_id": claim.operation_id if claim else None,
        "phase": claim.phase if claim else None,
        "expected_sats": claim.expected_sats if claim else None,
        "failures": claim.failures if claim else None,
        "deadline": claim.deadline if claim else None,
        "created_at": transaction.created_at,
        "amount": transaction.amount,
        "unit": transaction.unit,
        "mint_url": transaction.mint_url,
        "malformed": claim is None,
    }


class RoutstrReleaseOutcome(typing.NamedTuple):
    released: bool
    reason: str


async def release_routstr_auto_topup_state(
    provider_id: int, *, state_token: str | None
) -> RoutstrReleaseOutcome:
    """Clear a halted or stuck claim after an admin reconciles the peer.

    Unlike a Lightning melt there is no in-flight window to protect: the token
    is already with the peer or still in the wallet either way. What the fence
    does protect is the admin's decision — the row must be byte-identical to
    the one they reviewed, so a claim that advanced in the meantime is not
    swept on the strength of stale information.
    """
    state_id = _routstr_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 RoutstrReleaseOutcome(False, "no_active_claim")
    if transaction.request_id != state_token:
        return RoutstrReleaseOutcome(False, "stale_state")

    async with create_session() as session:
        result = await session.exec(  # type: ignore[call-overload]
            update(CashuTransaction)
            .where(
                col(CashuTransaction.id) == state_id,
                col(CashuTransaction.request_id) == state_token,
                col(CashuTransaction.collected) == False,  # noqa: E712
                col(CashuTransaction.swept) == False,  # noqa: E712
            )
            .values(swept=True)
        )
        if (getattr(result, "rowcount", 0) or 0) == 1:
            await session.commit()
            return RoutstrReleaseOutcome(True, "released")
        await session.rollback()
        return RoutstrReleaseOutcome(False, "claim_changed")


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}"


def _ppq_payment_id(operation_id: str) -> str:
    return f"ppq-payment-{operation_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).in_(
                    ["ppq_auto_topup", "ppq_auto_topup_claim"]
                ),
                # 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)
        )
        if (getattr(result, "rowcount", 0) or 0) == 1:
            claim = _parse_ppq_request_id(state_token)
            if claim is not None:
                await session.exec(  # type: ignore[call-overload]
                    update(CashuTransaction)
                    .where(
                        col(CashuTransaction.id) == _ppq_payment_id(claim.operation_id)
                    )
                    .values(swept=True)
                )
            await session.commit()
            return PPQReleaseOutcome(True, "released")
        await session.rollback()
        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,
                created_at=int(time.time()) if collected else CashuTransaction.created_at,
            )
        )
        updated = (getattr(result, "rowcount", 0) or 0) == 1
        if updated:
            await session.exec(  # type: ignore[call-overload]
                update(CashuTransaction)
                .where(col(CashuTransaction.id) == _ppq_payment_id(operation_id))
                .values(collected=collected, swept=swept)
            )
            await session.commit()
        else:
            await session.rollback()
        return updated


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.swept:
        return False
    if transaction.collected:
        return int(time.time()) - transaction.created_at < PPQ_SETTLED_COOLDOWN_SECONDS

    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: AsyncSession, row: UpstreamProviderRow
) -> 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 row.id is None:
        return False
    current = await session.get(UpstreamProviderRow, row.id)
    return bool(
        current is not None
        and current.enabled
        and current.provider_type == "ppqai"
        and current.api_key == row.api_key
        and current.provider_settings == row.provider_settings
    )


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):
            return None
        existing = await session.get(CashuTransaction, state_id)
        if existing is not None:
            if (
                existing.collected
                and not existing.swept
                and int(time.time()) - existing.created_at
                < PPQ_SETTLED_COOLDOWN_SECONDS
            ):
                return 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()),
                    source="ppq_auto_topup_claim",
                )
            )
            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):
                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_claim",
                )
            )
            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,
    amount_usd: 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,
            )
        )
        if (getattr(result, "rowcount", 0) or 0) != 1:
            await session.rollback()
            raise RuntimeError("PPQ auto top-up claim ownership was lost")
        session.add(
            CashuTransaction(
                id=_ppq_payment_id(operation_id),
                # Do not expose the raw BOLT11 through the transaction API.
                # The USD amount is stamped here so the daily spend cap can
                # aggregate what each payment was worth when it was made,
                # independent of later BTC price moves.
                token=f"ppq-invoice:{invoice_id}:usd:{amount_usd}",
                amount=amount,
                unit=unit,
                type="out",
                request_id=invoice_id,
                mint_url=mint_url,
                collected=False,
                swept=False,
                source="ppq_auto_topup",
            )
        )
        await session.commit()
    return lease_expires_at


async def _record_ppq_payment_spent(operation_id: str, paid_amount: int) -> None:
    """Persist the irreversible wallet spend before polling PPQ settlement."""
    async with create_session() as session:
        result = await session.exec(  # type: ignore[call-overload]
            update(CashuTransaction)
            .where(col(CashuTransaction.id) == _ppq_payment_id(operation_id))
            .values(amount=paid_amount)
        )
        await session.commit()
        if (getattr(result, "rowcount", 0) or 0) != 1:
            raise RuntimeError("PPQ payment audit row is missing")


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},
            )


def _ppq_payment_usd(amount: int, unit: str, token: str, price: float) -> float:
    """USD value of one PPQ payment audit row.

    Prefers the USD amount stamped into the token when the payment was
    recorded: converting stored sats at today's price would undercount past
    spend whenever the BTC price has fallen since. Falls back to a current
    price conversion for rows recorded before the stamp existed.
    """
    marker = ":usd:"
    if marker in token:
        try:
            return float(token.rsplit(marker, 1)[1])
        except ValueError:
            pass
    sats = amount if unit == "sat" else math.ceil(amount / 1000)
    return sats * price


async def _ppq_spent_last_24h_usd(price: float) -> float:
    """Total USD committed to PPQ top-ups in the last 24 hours.

    Counts in-flight and ambiguous payments — for spend-cap purposes an
    unresolved payment must be assumed spent — but not rows marked
    ``collected=False, swept=True``, which record payments the mint provably
    never attempted; those must not starve future top-ups for a day.
    """
    cutoff = int(time.time()) - 24 * 60 * 60
    async with create_session() as session:
        rows = (
            await session.exec(
                select(
                    CashuTransaction.amount,
                    CashuTransaction.unit,
                    CashuTransaction.token,
                ).where(
                    col(CashuTransaction.source) == "ppq_auto_topup",
                    col(CashuTransaction.type) == "out",
                    col(CashuTransaction.created_at) >= cutoff,
                    or_(
                        col(CashuTransaction.collected) == True,  # noqa: E712
                        col(CashuTransaction.swept) == False,  # noqa: E712
                    ),
                )
            )
        ).all()
    return sum(
        _ppq_payment_usd(amount, unit, token, price) for amount, unit, token in rows
    )


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

    # Require two low-balance reads before creating an invoice.
    confirmed_balance = await provider.get_balance()
    if (
        confirmed_balance is None
        or not math.isfinite(confirmed_balance)
        or confirmed_balance < 0
        or confirmed_balance >= threshold_usd
    ):
        logger.info(
            "PPQ auto top-up aborted by balance confirmation",
            extra={
                "provider_id": row.id,
                "first_balance_usd": balance,
                "confirmed_balance_usd": confirmed_balance,
                "threshold_usd": threshold_usd,
            },
        )
        return
    balance = confirmed_balance

    # Perform local pricing and owner-funds checks before asking PPQ to create
    # an invoice. The exact mint quote still has to be checked afterward, but
    # predictable local failures should not leave abandoned PPQ invoices.
    price = sats_usd_price()
    minimum_invoice_sats = math.ceil(amount_usd / price)
    max_invoice_sats = math.ceil(minimum_invoice_sats * PPQ_MAX_INVOICE_PREMIUM)
    if await maximum_owner_cashu_balance_sats() < minimum_invoice_sats:
        logger.warning(
            "PPQ auto top-up skipped because no mint has enough owner funds",
            extra={
                "provider_id": row.id,
                "minimum_invoice_sats": minimum_invoice_sats,
            },
        )
        return

    # Cheap early check to avoid claim and invoice churn; the authoritative
    # re-check happens under the wallet guard just before payment, where no
    # concurrent worker can move the total.
    spent_24h_usd = await _ppq_spent_last_24h_usd(price)
    if spent_24h_usd + amount_usd > PPQ_MAX_DAILY_TOPUP_USD:
        logger.critical(
            "PPQ auto top-up skipped: rolling 24h spend cap reached",
            extra={
                "provider_id": row.id,
                "spent_24h_usd": round(spent_24h_usd, 2),
                "topup_usd": amount_usd,
                "daily_cap_usd": PPQ_MAX_DAILY_TOPUP_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")
    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

    # Hold the wallet guard across planning and execution: the plan snapshots
    # live proof state, and another worker process reserving or spending those
    # proofs between the two calls would invalidate the snapshot.
    async with wallet_operation_guard():
        try:
            # Authoritative daily-cap check: the early check above is raceable
            # across worker processes, but here the guard serializes every
            # payment, so the total cannot move between this read and the
            # melt.
            spent_24h_usd = await _ppq_spent_last_24h_usd(price)
            if spent_24h_usd + amount_usd > PPQ_MAX_DAILY_TOPUP_USD:
                logger.critical(
                    "PPQ auto top-up aborted: rolling 24h spend cap reached",
                    extra={
                        "provider_id": row.id,
                        "spent_24h_usd": round(spent_24h_usd, 2),
                        "topup_usd": amount_usd,
                        "daily_cap_usd": PPQ_MAX_DAILY_TOPUP_USD,
                    },
                )
                raise ValueError("PPQ auto top-up daily spend cap reached")

            plan = await prepare_bolt11_payment(topup.payment_request)
            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),
                amount_usd=amount_usd,
                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 Bolt11PaymentAmbiguous:
            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,
                    "invoice_id": topup.invoice_id,
                    "admin_action": f"POST /admin/api/upstream-providers/{row.id}/ppq-auto-topup/release",
                },
                exc_info=True,
            )
            raise
        except BaseException:
            # Cancellation or an unexpected error after execution began is
            # also ambiguous. Preserve the claim before propagating it.
            await asyncio.shield(
                _mark_ppq_reconcile(
                    row,
                    operation_id,
                    lease_expires_at,
                    topup.invoice_id,
                    str(plan.quote.quote),
                )
            )
            logger.critical(
                "Unexpected failure while paying PPQ invoice; payment requires reconciliation",
                extra={"provider_id": row.id, "invoice_id": topup.invoice_id},
                exc_info=True,
            )
            raise

    try:
        # The melt is irreversible now. Persist the actual spend before any
        # fallible PPQ status call so operators retain an audit trail.
        await _record_ppq_payment_spent(operation_id, paid_amount)
        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)
    except BaseException as error:
        await asyncio.shield(
            _mark_ppq_reconcile(
                row,
                operation_id,
                lease_expires_at,
                topup.invoice_id,
                str(plan.quote.quote),
            )
        )
        logger.critical(
            "PPQ Lightning payment completed but settlement polling failed; reconciliation required",
            extra={
                "provider_id": row.id,
                "invoice_id": topup.invoice_id,
                "error": str(error),
                "error_type": type(error).__name__,
            },
            exc_info=True,
        )
        if isinstance(error, asyncio.CancelledError):
            raise
        return

    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,
        },
    )