routstr-core/routstr/refund.py

"""Refund claims: one open payout per API key, recorded before it is paid."""

import asyncio
import time
from typing import Any

import httpx
from cashu.wallet.helpers import deserialize_token_from_string
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlmodel import col, func, select, update

from .core.db import (
    REFUND_OPEN_STATUSES,
    REFUND_UNRESOLVED_STATUSES,
    ApiKey,
    AsyncSession,
    Refund,
    create_session,
)
from .core.db import (
    store_cashu_transaction_with_retry as store_cashu_transaction,
)
from .core.logging import get_logger
from .core.settings import settings
from .payment.lnurl import (
    LNURLError,
    MeltOutcomeAmbiguousError,
    MeltUnpaidError,
    get_lnurl_data,
)
from .wallet import (
    check_bolt11_payment_status,
    is_mint_connection_error,
    preferred_trusted_mint,
    resolve_trusted_source_mint,
    send_to_lnurl,
    send_token,
    token_mint_url,
)

logger = get_logger(__name__)

RECONCILE_BATCH_LIMIT = 100


def refund_unit(key: ApiKey) -> str:
    return key.refund_currency or "sat"


def amount_in_unit(amount_msats: int, unit: str) -> int:
    return amount_msats // 1000 if unit == "sat" else amount_msats


def refund_mint(key: ApiKey) -> str:
    """Trusted mint the payout is drawn from.

    A foreign refund mint (a key funded by a swapped-in token) is paid from the
    first configured trusted mint and swapped back; see :func:`swap_destination`.
    """
    if key.refund_mint_url:
        trusted_source = resolve_trusted_source_mint(key.refund_mint_url)
        if trusted_source is not None:
            return trusted_source
    return preferred_trusted_mint()


def swap_destination(key: ApiKey, method: str) -> str | None:
    """The user's own mint to swap a Cashu refund to, if that applies."""
    from .foreign_mint_swap import refund_destination_mint

    if method != "cashu":
        return None
    return refund_destination_mint(key)


async def validate_lightning_destination(destination: str) -> None:
    try:
        await get_lnurl_data(destination)
    except LNURLError as e:
        raise HTTPException(
            status_code=400, detail=f"Invalid lightning destination: {e}"
        )
    except httpx.HTTPError as e:
        raise HTTPException(
            status_code=400, detail=f"Lightning destination unreachable: {e}"
        )


async def open_claim(
    session: AsyncSession,
    key: ApiKey,
    *,
    method: str,
    destination: str | None,
) -> Refund:
    """Zero the balance and insert the claim in one transaction."""
    unit = refund_unit(key)
    # created_at has second resolution; step past the previous claim so the
    # newest claim for a key always sorts first.
    latest = await session.exec(
        select(func.max(col(Refund.created_at))).where(
            Refund.api_key_hashed_key == key.hashed_key
        )
    )
    created_at = max(int(time.time()), (latest.one() or 0) + 1)
    if destination is None:
        # A cashu refund to the user's own foreign mint records that mint as
        # the destination; the payout is still drawn from a trusted mint.
        destination = swap_destination(key, method)
    refund = Refund(
        api_key_hashed_key=key.hashed_key,
        method=method,
        destination=destination,
        amount_msats=key.total_balance,
        unit=unit,
        mint_url=refund_mint(key),
        claimed_at=int(time.time()),
        created_at=created_at,
        updated_at=created_at,
    )
    debit = (
        update(ApiKey)
        .where(col(ApiKey.hashed_key) == key.hashed_key)
        .where(col(ApiKey.balance) == key.balance)
        .where(col(ApiKey.reserved_balance) == key.reserved_balance)
        .values(balance=0, reserved_balance=0, reserved_at=None)
    )
    try:
        debited = await session.exec(debit)  # type: ignore[call-overload]
        if debited.rowcount == 0:
            await session.rollback()
            raise HTTPException(
                status_code=409,
                detail="Balance changed concurrently. Please retry the refund.",
            )
        session.add(refund)
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise refund_in_progress_error()
    return refund


async def _close(
    session: AsyncSession,
    refund: Refund,
    *,
    require_no_quote: bool = False,
    require_no_token: bool = False,
    from_statuses: tuple[str, ...] = REFUND_OPEN_STATUSES,
    **values: object,
) -> bool:
    stmt = (
        update(Refund)
        .where(col(Refund.id) == refund.id)
        .where(col(Refund.status).in_(from_statuses))
    )
    if require_no_quote:
        # A quote recorded since the row was read means a melt may be in flight.
        stmt = stmt.where(col(Refund.quote_id).is_(None))
    if require_no_token:
        stmt = stmt.where(col(Refund.token).is_(None))
    result = await session.exec(  # type: ignore[call-overload]
        stmt.values(claimed_at=None, updated_at=int(time.time()), **values)
    )
    return bool(result.rowcount)


async def renew_lease(session: AsyncSession, refund: Refund) -> None:
    """Push the reconciler lease forward before a slow mint step."""
    await session.exec(  # type: ignore[call-overload]
        update(Refund)
        .where(col(Refund.id) == refund.id)
        .where(col(Refund.status).in_(REFUND_OPEN_STATUSES))
        .values(claimed_at=int(time.time()))
    )
    await session.commit()


async def record_quote(refund: Refund, quote_id: str, mint_url: str) -> None:
    """Store the quote and its mint before the melt is sent; raises if the claim closed.

    Mint fallback can issue the quote on a different mint than the claim's.
    """
    async with create_session() as session:
        result = await session.exec(  # type: ignore[call-overload]
            update(Refund)
            .where(col(Refund.id) == refund.id)
            .where(col(Refund.status).in_(REFUND_OPEN_STATUSES))
            # Renew the lease so the reconciler leaves the payout alone.
            .values(
                quote_id=quote_id,
                mint_url=mint_url,
                claimed_at=int(time.time()),
                updated_at=int(time.time()),
            )
        )
        await session.commit()
    if not result.rowcount:
        raise LNURLError("Refund claim closed before the melt was dispatched")
    refund.quote_id = quote_id
    refund.mint_url = mint_url


async def settle(
    session: AsyncSession,
    refund: Refund,
    *,
    quote_id: str | None = None,
    token: str | None = None,
    mint_url: str | None = None,
) -> bool:
    values: dict[str, Any] = {"status": "paid"}
    if quote_id is not None:
        values["quote_id"] = quote_id
    if token is not None:
        values["token"] = token
    if mint_url is not None:
        values["mint_url"] = mint_url
    # The payout side knows the money moved, so a claim the reconciler gave up
    # on (stuck) is closed as paid too.
    settled = await _close(
        session, refund, from_statuses=REFUND_UNRESOLVED_STATUSES, **values
    )
    await session.commit()
    if not settled:
        logger.warning(
            "refund paid but its claim was already closed",
            extra={"refund_id": refund.id, "prior_status": refund.status},
        )
    return settled


async def release(
    session: AsyncSession, refund: Refund, *, require_no_quote: bool = False
) -> bool:
    """Mark the claim failed and restore the balance."""
    if not await _close(
        session, refund, require_no_quote=require_no_quote, status="failed"
    ):
        # Commit, not rollback: rollback after an ORM UPDATE breaks later loads.
        await session.commit()
        return False
    await session.exec(  # type: ignore[call-overload]
        update(ApiKey)
        .where(col(ApiKey.hashed_key) == refund.api_key_hashed_key)
        .values(balance=col(ApiKey.balance) + refund.amount_msats)
    )
    await session.commit()
    logger.info(
        "refund released; balance restored",
        extra={
            "refund_id": refund.id,
            "key_hash": refund.api_key_hashed_key[:8],
            "restored_msats": refund.amount_msats,
        },
    )
    return True


async def hold(
    session: AsyncSession,
    refund: Refund,
    quote_id: str | None,
    *,
    token: str | None = None,
) -> None:
    """Withhold the balance; the quote or token names what the mint may have paid."""
    values: dict[str, Any] = {"status": "ambiguous", "quote_id": quote_id}
    if token is not None:
        values["token"] = token
        values["mint_url"] = refund.mint_url
    await _close(session, refund, **values)
    await session.commit()


def refund_in_progress_error(refund: Refund | None = None) -> HTTPException:
    """The 409 raised when a key already has an unresolved refund claim."""
    stuck = refund is not None and refund.status == "stuck"
    error: dict[str, str] = {
        "message": (
            "A refund for this key is unresolved and requires operator reconciliation."
            if stuck
            else "A refund for this key is already in progress."
        ),
        "type": "invalid_request_error",
        "code": "refund_unresolved" if stuck else "refund_in_progress",
    }
    if refund is not None:
        error["refund_id"] = refund.id
        error["status"] = refund.status
    return HTTPException(status_code=409, detail={"error": error})


async def _latest_with_status(
    session: AsyncSession, key: ApiKey, statuses: tuple[str, ...]
) -> Refund | None:
    result = await session.exec(
        select(Refund)
        .where(Refund.api_key_hashed_key == key.hashed_key)
        .where(col(Refund.status).in_(statuses))
        .order_by(col(Refund.created_at).desc(), col(Refund.updated_at).desc())
    )
    return result.first()


async def latest_open(session: AsyncSession, key: ApiKey) -> Refund | None:
    """Latest non-terminal (in-flight) claim for the key, if any.

    An open claim means a prior refund already debited the balance and is still
    settling, so the balance reads as zero even though a refund is under way.
    """
    return await _latest_with_status(session, key, REFUND_OPEN_STATUSES)


async def latest_stuck(session: AsyncSession, key: ApiKey) -> Refund | None:
    """Latest claim the reconciler gave up on; needs operator recovery."""
    return await _latest_with_status(session, key, ("stuck",))


async def latest_terminal(session: AsyncSession, key: ApiKey) -> Refund | None:
    """Latest paid refund of either method.

    Cashu tokens are normally served from cashu_transactions, which tracks
    collection and sweeping; the claim row is the fallback when that ledger
    write failed after the token was already issued.
    """
    return await _latest_with_status(session, key, ("paid",))


def _delivered_amount_msats(refund: Refund) -> int:
    """Actual bearer-token value, which can be net of cross-mint fees."""
    if not refund.token:
        return refund.amount_msats
    try:
        token = deserialize_token_from_string(refund.token)
    except Exception:
        logger.error(
            "paid refund token could not be decoded for amount reporting",
            extra={"refund_id": refund.id},
        )
        return refund.amount_msats
    return int(token.amount) * 1000 if str(token.unit) == "sat" else int(token.amount)


def describe(refund: Refund) -> dict[str, str]:
    body: dict[str, str] = {"refund_id": refund.id, "status": refund.status}
    if refund.token:
        body["token"] = refund.token
    if refund.destination:
        body["recipient"] = refund.destination
    delivered_msats = _delivered_amount_msats(refund)
    if refund.unit == "sat":
        body["sats"] = str(delivered_msats // 1000)
    else:
        body["msats"] = str(delivered_msats)
    return body


async def _pay_lightning(session: AsyncSession, refund: Refund) -> bool:
    async def capture_quote(quote: str, mint_url: str) -> None:
        await record_quote(refund, quote, mint_url)

    try:
        await send_to_lnurl(
            amount_in_unit(refund.amount_msats, refund.unit),
            refund.unit,
            refund.mint_url,
            str(refund.destination),
            on_melt_quote=capture_quote,
        )
    except MeltOutcomeAmbiguousError as e:
        await hold(session, refund, refund.quote_id)
        logger.error(
            "refund outcome ambiguous; balance withheld pending reconciliation",
            extra={
                "refund_id": refund.id,
                "error": str(e),
                "key_hash": refund.api_key_hashed_key[:8],
                "quote_id": refund.quote_id,
            },
        )
        raise
    return await settle(session, refund, quote_id=refund.quote_id)


async def _pay_cashu(session: AsyncSession, refund: Refund) -> bool:
    amount = amount_in_unit(refund.amount_msats, refund.unit)
    await renew_lease(session, refund)
    if refund.destination:
        from .foreign_mint_swap import swap_out_for_refund

        return await swap_out_for_refund(session, refund, refund.destination)
    token = await send_token(amount, refund.unit, refund.mint_url)
    # From here the token is bearer money: keep it on the claim so a failed
    # settle withholds the balance instead of restoring it.
    refund.token = token
    refund.mint_url = token_mint_url(token, refund.mint_url)
    return await settle(session, refund, token=token, mint_url=refund.mint_url)


async def _record_cashu_payout(refund: Refund) -> None:
    """Ledger write for an issued token; the claim row already holds the token,
    so a failure here must not fail the request or release the balance."""
    try:
        await store_cashu_transaction(
            token=str(refund.token),
            amount=amount_in_unit(_delivered_amount_msats(refund), refund.unit),
            unit=refund.unit,
            mint_url=refund.mint_url,
            typ="out",
            collected=False,
            source="apikey",
            api_key_hashed_key=refund.api_key_hashed_key,
        )
    except Exception as e:
        logger.error(
            "refund token issued but cashu transaction was not recorded",
            extra={
                "refund_id": refund.id,
                "error": str(e),
                "error_type": type(e).__name__,
                "key_hash": refund.api_key_hashed_key[:8],
            },
        )


def unresolved_refund_error() -> HTTPException:
    return HTTPException(
        status_code=502,
        detail=(
            "Refund was dispatched but its outcome is unconfirmed; the "
            "balance is withheld until reconciliation completes"
        ),
    )


async def _abort(session: AsyncSession, refund: Refund) -> None:
    """Fail the claim, or withhold it once a melt quote or token exists.

    A recorded quote means the mint may already have paid; an issued token is
    already bearer money. In both cases the balance must not be restored.
    """
    if refund.quote_id is None and refund.token is None:
        await release(session, refund)
        return
    await hold(session, refund, refund.quote_id, token=refund.token)
    logger.error(
        "refund failed after its payout was dispatched; balance withheld "
        "pending reconciliation",
        extra={
            "refund_id": refund.id,
            "key_hash": refund.api_key_hashed_key[:8],
            "quote_id": refund.quote_id,
            "has_token": refund.token is not None,
            "mint_url": refund.mint_url,
        },
    )
    raise unresolved_refund_error()


async def execute(session: AsyncSession, refund: Refund) -> dict[str, str]:
    attached_refund = refund
    # Keep payout evidence outside the identity map: a failed flush/commit can
    # expire attached attributes, including the only copy of an issued token.
    refund = Refund(**refund.model_dump())
    try:
        if refund.method == "lightning":
            settled = await _pay_lightning(session, refund)
        else:
            settled = await _pay_cashu(session, refund)
    except MeltOutcomeAmbiguousError:
        # Already held by _pay_lightning; releasing here would pay out twice.
        raise unresolved_refund_error()
    except MeltUnpaidError as e:
        # The mint answered the melt itself with unpaid: proof that nothing was
        # sent, so the balance goes back now rather than after reconciliation.
        await release(session, refund)
        logger.warning(
            "refund melt unpaid at the mint; balance restored",
            extra={
                "refund_id": refund.id,
                "error": str(e),
                "key_hash": refund.api_key_hashed_key[:8],
                "quote_id": refund.quote_id,
            },
        )
        raise HTTPException(
            status_code=503,
            detail="Lightning payment failed at the mint; balance restored. Retry later.",
        )
    except HTTPException:
        await session.rollback()
        await _abort(session, refund)
        raise
    except Exception as e:
        await session.rollback()
        await _abort(session, refund)
        logger.error(
            "refund payout failed",
            extra={
                "refund_id": refund.id,
                "error": str(e),
                "error_type": type(e).__name__,
                "key_hash": refund.api_key_hashed_key[:8],
                "method": refund.method,
                "mint_url": refund.mint_url,
            },
        )
        if is_mint_connection_error(e):
            raise HTTPException(status_code=503, detail="Mint service unavailable")
        raise HTTPException(status_code=500, detail="Refund failed")

    if refund.method == "cashu":
        await _record_cashu_payout(refund)

    if settled:
        refund.status = "paid"
        refund.claimed_at = None
    else:
        # Report the row as it stands rather than a status that was not written.
        await session.refresh(attached_refund)
        refund = attached_refund
    logger.info(
        "refund paid",
        extra={
            "refund_id": refund.id,
            "method": refund.method,
            "amount_msats": refund.amount_msats,
            "key_hash": refund.api_key_hashed_key[:8],
        },
    )
    return describe(refund)


async def _lease(refund_id: str, now: int, lease_cutoff: int) -> bool:
    async with create_session() as session:
        result = await session.exec(  # type: ignore[call-overload]
            update(Refund)
            .where(col(Refund.id) == refund_id)
            .where(col(Refund.status).in_(REFUND_OPEN_STATUSES))
            .where(
                col(Refund.claimed_at).is_(None)
                | (col(Refund.claimed_at) < lease_cutoff)
            )
            .values(claimed_at=now)
        )
        await session.commit()
        return bool(result.rowcount)


async def _reconcile(refund: Refund, now: int) -> None:
    if refund.method != "lightning":
        if refund.token is not None:
            # The token was issued and kept on the claim; the payout is done.
            async with create_session() as session:
                await settle(session, refund)
            return
        if refund.destination and refund.quote_id:
            # A swap to the user's mint: its journal row owns the outcome and
            # the swap reconciler settles or releases this claim from there.
            return
        # No quote to query for cashu; withhold the balance and alert once.
        async with create_session() as session:
            if await _close(session, refund, require_no_token=True, status="stuck"):
                await session.commit()
                logger.critical(
                    "cashu refund stuck; balance withheld, manual reconciliation required",
                    extra={
                        "refund_id": refund.id,
                        "key_hash": refund.api_key_hashed_key[:8],
                        "amount_msats": refund.amount_msats,
                    },
                )
            else:
                await session.commit()
                current = await session.get(Refund, refund.id)
                if current is not None and current.token is not None:
                    await settle(session, current)
        return

    if refund.quote_id is None:
        # Never sent, unless a quote appeared since the row was read.
        async with create_session() as session:
            if not await release(session, refund, require_no_quote=True):
                logger.info(
                    "refund gained a melt quote during reconciliation; left open",
                    extra={"refund_id": refund.id},
                )
        return

    status = await check_bolt11_payment_status(
        refund.mint_url, refund.unit, refund.quote_id
    )
    if status == "paid":
        async with create_session() as session:
            await settle(session, refund)
    elif status == "unpaid":
        # A fresh melt can report unpaid briefly; trust it only after a timeout.
        if refund.updated_at > now - settings.refund_claim_timeout_seconds:
            logger.info(
                "refund unpaid at the mint but too recent to release; waiting",
                extra={"refund_id": refund.id, "updated_at": refund.updated_at},
            )
            return
        async with create_session() as session:
            await release(session, refund)
    else:
        logger.warning(
            "refund still unresolved at the mint",
            extra={"refund_id": refund.id, "melt_status": status},
        )


async def reconcile_once() -> None:
    """Resolve open claims whose lease has lapsed."""
    now = int(time.time())
    lease_cutoff = now - settings.refund_claim_timeout_seconds
    async with create_session() as session:
        result = await session.exec(
            select(Refund)
            .where(col(Refund.status).in_(REFUND_OPEN_STATUSES))
            .where(
                col(Refund.claimed_at).is_(None)
                | (col(Refund.claimed_at) < lease_cutoff)
            )
            .order_by(col(Refund.created_at))
            .limit(RECONCILE_BATCH_LIMIT)
        )
        stale = list(result.all())

    for refund in stale:
        if not await _lease(refund.id, now, lease_cutoff):
            continue
        try:
            await _reconcile(refund, now)
        except Exception as e:
            logger.error(
                "refund reconciliation failed",
                extra={
                    "refund_id": refund.id,
                    "error": str(e),
                    "error_type": type(e).__name__,
                },
            )


async def periodic_refund_reconcile() -> None:
    while True:
        await asyncio.sleep(settings.refund_reconcile_interval_seconds)
        try:
            await reconcile_once()
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.error(
                "refund reconcile loop error",
                extra={"error": str(e), "error_type": type(e).__name__},
            )