routstr-core/routstr/core/db.py

import asyncio
import hashlib
import os
import pathlib
import sqlite3
import time
import uuid
from contextlib import asynccontextmanager
from enum import Enum
from typing import AsyncGenerator

from alembic import command
from alembic.config import Config
from alembic.util.exc import CommandError
from sqlalchemy import Index, UniqueConstraint, case, delete, event, or_, text
from sqlalchemy.engine import make_url
from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.ext.asyncio.engine import create_async_engine
from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
from sqlmodel.ext.asyncio.session import AsyncSession

from .logging import get_logger
from .settings import settings

logger = get_logger(__name__)

DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///keys.db")


def create_db_engine(database_url: str = DATABASE_URL) -> AsyncEngine:
    """Build and instrument an async engine from environment-only settings."""
    url = make_url(database_url)
    backend = url.get_backend_name()
    is_sqlite = backend == "sqlite"
    is_memory_sqlite = is_sqlite and url.database in {None, "", ":memory:"}
    pool_pre_ping = settings.database_pool_pre_ping or not is_sqlite
    options: dict[str, int | float | bool] = {"pool_pre_ping": pool_pre_ping}
    connect_args: dict[str, object] = {}
    if is_sqlite and not is_memory_sqlite:
        connect_args["timeout"] = settings.database_busy_timeout
    if not is_memory_sqlite:
        options.update(
            pool_size=settings.database_pool_size,
            max_overflow=settings.database_max_overflow,
            pool_timeout=settings.database_pool_timeout,
            pool_recycle=settings.database_pool_recycle,
        )

    logger.info(
        "Database pool configured",
        extra={
            "database_url_backend": backend,
            "in_memory_sqlite": is_memory_sqlite,
            **options,
            "connect_args": connect_args,
        },
    )
    created_engine = create_async_engine(
        database_url, echo=False, connect_args=connect_args, **options
    )
    hold_warn_seconds = settings.database_pool_hold_warn_seconds

    def record_pool_checkout(
        dbapi_connection: object, connection_record: object, proxy: object
    ) -> None:
        connection_record.info["routstr_checked_out_at"] = time.monotonic()  # type: ignore[attr-defined]

    def record_pool_checkin(
        dbapi_connection: object, connection_record: object
    ) -> None:
        checked_out_at = connection_record.info.pop(  # type: ignore[attr-defined]
            "routstr_checked_out_at", None
        )
        if checked_out_at is None:
            return
        held_seconds = time.monotonic() - checked_out_at
        if held_seconds >= hold_warn_seconds:
            logger.warning(
                "Database connection held longer than threshold",
                extra={
                    "held_seconds": round(held_seconds, 3),
                    "threshold_seconds": hold_warn_seconds,
                    "pool_status": created_engine.pool.status(),
                },
            )

    event.listen(created_engine.sync_engine, "checkout", record_pool_checkout)
    event.listen(created_engine.sync_engine, "checkin", record_pool_checkin)
    return created_engine


engine = create_db_engine()


class ApiKey(SQLModel, table=True):  # type: ignore
    __tablename__ = "api_keys"

    hashed_key: str = Field(primary_key=True)
    balance: int = Field(default=0, description="Balance in millisatoshis (msats)")
    reserved_balance: int = Field(
        default=0, description="Reserved balance in millisatoshis (msats)"
    )
    reserved_at: int | None = Field(
        default=None,
        description=(
            "Unix timestamp of the most recent balance reservation. Used to "
            "detect and release stale reservations (e.g. after client "
            "disconnects). NULL when no reservation has been made yet."
        ),
    )
    refund_address: str | None = Field(
        default=None,
        description="Lightning address to refund remaining balance after key expires",
    )
    key_expiry_time: int | None = Field(
        default=None,
        description="Unix-timestamp after which the cashu-token's balance gets refunded to the refund_address",
    )
    total_spent: int = Field(
        default=0, description="Total spent in millisatoshis (msats)"
    )
    total_requests: int = Field(default=0)
    created_at: int | None = Field(
        default_factory=lambda: int(time.time()),
        nullable=True,
        description=(
            "Unix timestamp when the key was created. Nullable: keys created "
            "before this column existed have no value and sort last."
        ),
    )
    refund_mint_url: str | None = Field(
        default=None,
        description="URL of the mint used to create the cashu-token",
    )
    refund_currency: str | None = Field(
        default=None,
        description="Currency of the cashu-token",
    )
    validity_date: int | None = Field(
        default=None,
        description="Unix timestamp after which the key is no longer valid",
    )

    @property
    def total_balance(self) -> int:
        return self.balance - self.reserved_balance


async def _transition_stale_reservation(
    session: AsyncSession, reservation_id: str, cutoff: int
) -> bool:
    """Mark one reservation released iff its lease is still older than cutoff.

    ``created_at`` doubles as the heartbeat lease timestamp, so the guard must
    be part of this update: a reservation renewed between the sweeper's select
    and this transition is in flight and must survive.
    """
    transition = await session.exec(  # type: ignore[call-overload]
        update(ReservationRelease)
        .where(col(ReservationRelease.id) == reservation_id)
        .where(col(ReservationRelease.status) == "active")
        .where(
            (col(ReservationRelease.created_at) < cutoff)
            | (col(ReservationRelease.expires_at) <= int(time.time()))
        )
        .values(status="released")
    )
    return bool(transition.rowcount == 1)


async def _release_legacy_aggregate(
    session: AsyncSession,
    key_hash: str,
    observed_reserved: int,
    observed_reserved_at: int | None,
) -> bool:
    """Zero one legacy aggregate reservation iff it is exactly as observed.

    A new reservation committing between the sweeper's read and this update
    changes ``reserved_balance``/``reserved_at`` in the same transaction that
    creates its durable row, so this compare-and-swap fails instead of erasing
    the newcomer's reserved funds.
    """
    if observed_reserved <= 0:
        return False
    reserved_at_guard = (
        col(ApiKey.reserved_at).is_(None)
        if observed_reserved_at is None
        else col(ApiKey.reserved_at) == observed_reserved_at
    )
    result = await session.exec(  # type: ignore[call-overload]
        update(ApiKey)
        .where(col(ApiKey.hashed_key) == key_hash)
        .where(col(ApiKey.reserved_balance) == observed_reserved)
        .where(reserved_at_guard)
        .values(reserved_balance=0, reserved_at=None)
    )
    return bool(result.rowcount == 1)


async def release_stale_reservations(
    session: AsyncSession,
    max_age_seconds: int,
    *,
    key_hash: str | None = None,
) -> int:
    """Release stale durable reservations without touching newer reservations."""
    cutoff = int(time.time()) - max_age_seconds
    query = (
        select(ReservationRelease)
        .where(col(ReservationRelease.status) == "active")
        .where(
            (col(ReservationRelease.created_at) < cutoff)
            | (col(ReservationRelease.expires_at) <= int(time.time()))
        )
    )
    if key_hash is not None:
        query = query.where(
            or_(
                col(ReservationRelease.key_hash) == key_hash,
                col(ReservationRelease.billing_key_hash) == key_hash,
            )
        )
    # Capture primitives: a repair rollback below would expire ORM instances.
    reservation_rows = [
        (r.id, r.key_hash, r.billing_key_hash, r.reserved_msats)
        for r in (await session.exec(query)).all()
    ]
    released = 0

    for res_id, res_key_hash, res_billing_hash, res_msats in reservation_rows:
        if not await _transition_stale_reservation(session, res_id, cutoff):
            continue

        values = {
            "reserved_balance": col(ApiKey.reserved_balance) - res_msats,
            "reserved_at": case(
                (
                    col(ApiKey.reserved_balance) - res_msats > 0,
                    col(ApiKey.reserved_at),
                ),
                else_=None,
            ),
        }
        parent_result = await session.exec(  # type: ignore[call-overload]
            update(ApiKey)
            .where(col(ApiKey.hashed_key) == res_billing_hash)
            .where(col(ApiKey.reserved_balance) >= res_msats)
            .values(**values)
        )
        aggregates_ok = parent_result.rowcount == 1
        if aggregates_ok and res_billing_hash != res_key_hash:
            child_result = await session.exec(  # type: ignore[call-overload]
                update(ApiKey)
                .where(col(ApiKey.hashed_key) == res_key_hash)
                .where(col(ApiKey.reserved_balance) >= res_msats)
                .values(**values)
            )
            aggregates_ok = child_result.rowcount == 1

        if not aggregates_ok:
            # The aggregates no longer hold this reservation's msats — the
            # durable row is corrupt. Repair by terminalizing it WITHOUT
            # subtracting uncertain aggregates (legacy cleanup below reconciles
            # any stale remainder) and keep sweeping the rest of the batch:
            # one corrupt row must not poison all stale cleanup.
            await session.rollback()
            if await _transition_stale_reservation(session, res_id, cutoff):
                await session.commit()
                released += 1
                logger.error(
                    "Released corrupt stale reservation without aggregate subtraction",
                    extra={
                        "reservation_id": res_id,
                        "billing_key_hash": res_billing_hash[:8] + "...",
                        "reserved_msats": res_msats,
                    },
                )
            continue
        # Commit each release on its own so a later corrupt record's rollback
        # cannot discard the healthy releases already processed in this batch.
        await session.commit()
        released += 1

    # Rolling upgrades can leave aggregate reservations created before durable
    # reservation rows existed. Release only stale aggregates that have no active
    # durable owner; targeted refund cleanup also heals legacy NULL timestamps.
    legacy_query = select(ApiKey).where(col(ApiKey.reserved_balance) > 0)
    if key_hash is None:
        legacy_query = legacy_query.where(col(ApiKey.reserved_at).is_not(None)).where(
            col(ApiKey.reserved_at) < cutoff
        )
    else:
        legacy_query = legacy_query.where(col(ApiKey.hashed_key) == key_hash).where(
            or_(col(ApiKey.reserved_at).is_(None), col(ApiKey.reserved_at) < cutoff)
        )

    for legacy_key in (await session.exec(legacy_query)).all():
        observed_reserved = legacy_key.reserved_balance
        observed_reserved_at = legacy_key.reserved_at
        active_owner = (
            await session.exec(
                select(ReservationRelease.id)
                .where(col(ReservationRelease.status) == "active")
                .where(
                    or_(
                        col(ReservationRelease.key_hash) == legacy_key.hashed_key,
                        col(ReservationRelease.billing_key_hash)
                        == legacy_key.hashed_key,
                    )
                )
                .limit(1)
            )
        ).first()
        if active_owner is not None:
            continue
        if await _release_legacy_aggregate(
            session, legacy_key.hashed_key, observed_reserved, observed_reserved_at
        ):
            released += 1

    await session.commit()
    if released:
        logger.warning(
            "Released stale reservations",
            extra={
                "released_reservations": released,
                "max_age_seconds": max_age_seconds,
            },
        )
    return released


async def prune_dead_api_keys(session: AsyncSession, min_age_seconds: int) -> int:
    """Delete dead API keys; return the count removed.

    Dead = 0 balance/reservation/spend/requests, older than the grace
    period, no invoice that could still settle. Cashu rows are
    unlinked (not deleted) first to keep the audit trail.
    """
    now = int(time.time())
    cutoff = now - min_age_seconds

    # An expired invoice stays creditable for the grace window, and crediting it
    # after its target key is gone strands the payment at the mint.
    settleable_invoice = (
        select(LightningInvoice.id)
        .where(col(LightningInvoice.api_key_hash) == col(ApiKey.hashed_key))
        .where(
            col(LightningInvoice.status).in_(("pending", "settlement_pending"))
            | (
                (col(LightningInvoice.status) == "expired")
                & (
                    col(LightningInvoice.expires_at)
                    > now - INVOICE_EXPIRY_GRACE_SECONDS
                )
            )
        )
    ).exists()

    has_refund_claim = (
        select(Refund.id).where(
            col(Refund.api_key_hashed_key) == col(ApiKey.hashed_key)
        )
    ).exists()

    eligible_hashes = (
        select(ApiKey.hashed_key)
        .where(col(ApiKey.balance) == 0)
        .where(col(ApiKey.reserved_balance) == 0)
        .where(col(ApiKey.total_spent) == 0)
        .where(col(ApiKey.total_requests) == 0)
        .where((col(ApiKey.created_at).is_(None)) | (col(ApiKey.created_at) < cutoff))
        .where(~settleable_invoice)
        # refunds holds a non-null FK to the key.
        .where(~has_refund_claim)
    )

    # Unlink transactions rather than cascade-deleting them, so the financial
    # audit trail survives. The eligibility predicate is re-evaluated inside both
    # statements so a key that gained balance mid-run is left untouched.
    await session.exec(  # type: ignore[call-overload]
        update(CashuTransaction)
        .where(col(CashuTransaction.api_key_hashed_key).in_(eligible_hashes))
        .values(api_key_hashed_key=None)
    )

    result = await session.exec(  # type: ignore[call-overload]
        delete(ApiKey).where(col(ApiKey.hashed_key).in_(eligible_hashes))
    )
    await session.commit()

    pruned = int(result.rowcount or 0)
    logger.info(
        "Pruned dead API keys",
        extra={"pruned_keys": pruned, "min_age_seconds": min_age_seconds},
    )
    return pruned


class ModelRow(SQLModel, table=True):  # type: ignore
    __tablename__ = "models"
    id: str = Field(primary_key=True)
    upstream_provider_id: int = Field(
        primary_key=True, foreign_key="upstream_providers.id", ondelete="CASCADE"
    )
    name: str = Field()
    created: int = Field()
    description: str = Field()
    context_length: int = Field()
    architecture: str = Field()
    pricing: str = Field()
    sats_pricing: str | None = Field(default=None)
    image_pricing: str | None = Field(
        default=None, description="JSON per-image price book, for image models"
    )
    per_request_limits: str | None = Field(default=None)
    top_provider: str | None = Field(default=None)
    canonical_slug: str | None = Field(default=None, description="Canonical model slug")
    alias_ids: str | None = Field(
        default=None, description="JSON array of model alias IDs"
    )
    enabled: bool = Field(default=True, description="Whether this model is enabled")
    forwarded_model_id: str | None = Field(
        default=None,
        description=(
            "Client-facing alias advertised by /v1/models and accepted on "
            "requests in place of id. JSON request bodies sent upstream carry "
            "id (via the provider's transform_model_name), not this value."
        ),
    )
    upstream_provider: "UpstreamProviderRow" = Relationship(back_populates="models")


class ModelPathRow(SQLModel, table=True):  # type: ignore
    """Upstream provider path a model is reachable through.

    Discovery data plus provider-specific model metadata. ``model_id`` is
    intentionally NOT globally unique: it is the client-visible ``/v1/models``
    id (``forwarded_model_id or id``) grouped across every provider that exposes
    the model. A single model
    can therefore have several rows — one per direct provider path plus one per
    OpenRouter sub-provider endpoint.
    """

    __tablename__ = "model_paths"
    __table_args__ = (
        UniqueConstraint(
            "model_id",
            "path",
            "upstream_provider_id",
            name="uq_model_paths_model_path_provider",
        ),
    )
    id: int | None = Field(default=None, primary_key=True)
    # No standalone index on model_id: the unique constraint's autoindex already
    # leads on model_id, so a second index only adds write amplification.
    model_id: str = Field(
        description="Client-visible /v1/models id (forwarded_model_id or id)"
    )
    path: str = Field(
        description=(
            "Opaque selector containing upstream URL, provider ID, model ID, "
            "and optional endpoint tag"
        )
    )
    provider_slug: str = Field(
        description="Public slug of the configured upstream provider"
    )
    provider_type: str = Field(description="Configured upstream provider type")
    endpoint_tag: str | None = Field(
        default=None,
        description="Exact OpenRouter endpoint tag used for request-side selection",
    )
    endpoint_name: str | None = Field(
        default=None, description="Human-readable endpoint display name"
    )
    model_metadata: str = Field(
        default="{}",
        description="JSON model metadata specific to this provider path",
    )
    upstream_provider_id: int = Field(
        index=True,
        foreign_key="upstream_providers.id",
        ondelete="CASCADE",
        description="upstream_providers.id this path was discovered from",
    )
    updated_at: int = Field(
        default=0,
        description="Unix timestamp of the refresh cycle that wrote this row",
    )


# expires_at is our own clock, not the mint's quote expiry, so an expired row
# may still be paid at the mint and must stay creditable for this long after.
INVOICE_EXPIRY_GRACE_SECONDS = 86_400


class LightningInvoice(SQLModel, table=True):  # type: ignore
    __tablename__ = "lightning_invoices"

    id: str = Field(primary_key=True, description="Unique invoice identifier")
    bolt11: str = Field(description="BOLT11 invoice string", unique=True)
    amount_sats: int = Field(description="Amount in satoshis")
    description: str = Field(description="Invoice description")
    payment_hash: str = Field(description="Payment hash for tracking", unique=True)
    status: str = Field(
        default="pending",
        description=(
            "pending, settlement_pending, paid, failed, expired, cancelled, "
            "reconciliation_required"
        ),
    )
    api_key_hash: str | None = Field(
        default=None, description="Associated API key hash for topup operations"
    )
    direction: str = Field(
        default="in", description="in for incoming invoices, out for payouts"
    )
    purpose: str = Field(description="create, topup or payout")
    mint_url: str | None = Field(
        default=None,
        description="Mint URL where the quote was created (fallback tracking)",
    )
    created_at: int = Field(
        default_factory=lambda: int(time.time()), description="Unix timestamp"
    )
    expires_at: int = Field(description="Unix timestamp when invoice expires")
    paid_at: int | None = Field(default=None, description="Unix timestamp when paid")
    validity_date: int | None = Field(
        default=None,
        description="Unix timestamp after which the created key expires",
    )


class CashuTransaction(SQLModel, table=True):  # type: ignore
    __tablename__ = "cashu_transactions"

    id: str = Field(
        primary_key=True,
        default_factory=lambda: uuid.uuid4().hex,
        description="Unique transaction identifier",
    )
    token: str = Field(description="Serialized Cashu token")
    amount: int = Field(description="Amount in the token's unit")
    unit: str = Field(description="Token unit (sat or msat)")
    mint_url: str | None = Field(default=None, description="Mint URL for the token")
    type: str = Field(default="out", description="Transaction type: in or out")
    request_id: str | None = Field(default=None, description="Associated request ID")
    created_at: int = Field(
        default_factory=lambda: int(time.time()),
        description="Unix timestamp",
    )
    collected: bool = Field(default=False)
    swept: bool = Field(default=False)
    sweep_started_at: int | None = Field(
        default=None,
        description="Unix timestamp for a recoverable refund-sweep claim",
    )
    source: str = Field(
        default="x-cashu",
        description="Payment source: x-cashu or apikey",
    )
    api_key_hashed_key: str | None = Field(
        default=None,
        foreign_key="api_keys.hashed_key",
        index=True,
        description="Associated API key hash for wallet history",
    )


REFUND_OPEN_STATUSES = ("pending", "ambiguous")

# Debited from the key but neither paid out nor restored, so still owed.
REFUND_UNRESOLVED_STATUSES = ("pending", "ambiguous", "stuck")

_REFUND_OPEN_PREDICATE = "status IN ('pending', 'ambiguous')"


class Refund(SQLModel, table=True):  # type: ignore
    """One payout claim; the partial unique index allows one open claim per key."""

    __tablename__ = "refunds"
    __table_args__ = (
        Index(
            "ux_refunds_open_per_key",
            "api_key_hashed_key",
            unique=True,
            sqlite_where=text(_REFUND_OPEN_PREDICATE),
            postgresql_where=text(_REFUND_OPEN_PREDICATE),
        ),
    )

    id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex)
    api_key_hashed_key: str = Field(foreign_key="api_keys.hashed_key", index=True)
    method: str = Field(description="Payout method: lightning or cashu")
    destination: str | None = Field(
        default=None, description="Lightning address or LNURL, NULL for cashu"
    )
    amount_msats: int = Field(description="Balance debited when the claim opened")
    unit: str = Field(description="Mint unit the payout is denominated in")
    mint_url: str = Field(description="Mint the payout is drawn from")
    status: str = Field(
        default="pending",
        index=True,
        description="pending, paid, failed, ambiguous, or stuck",
    )
    quote_id: str | None = Field(
        default=None, description="Melt quote id, for reconciling an ambiguous payout"
    )
    token: str | None = Field(default=None, description="Issued cashu token")
    claimed_at: int | None = Field(
        default=None, description="Reconciler lease timestamp"
    )
    created_at: int = Field(default_factory=lambda: int(time.time()))
    updated_at: int = Field(default_factory=lambda: int(time.time()))


# Swap rows the reconciler still owns: the melt was dispatched and its outcome
# or follow-up (mint, credit, token issue) is not final.
SWAP_OPEN_STATUSES = ("melting", "ambiguous", "melted", "minted", "issued")


class CashuSwap(SQLModel, table=True):  # type: ignore
    """Journal of one cross-mint swap, written before any Lightning payment.

    ``in`` swaps melt a token from a mint the operator does not trust into the
    preferred trusted mint and credit an API key. ``out`` swaps melt owner
    proofs on that trusted mint to issue a refund token on the user's own mint.
    Every money movement is recorded here first so a crash or timeout leaves a row the
    reconciler can finish or fail, never an unknown balance.
    """

    __tablename__ = "cashu_swaps"

    id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex)
    direction: str = Field(description="in (token -> primary) or out (refund)")
    status: str = Field(
        default="melting",
        index=True,
        description=(
            "melting, ambiguous, melted, minted, credited, issued, settled, failed"
        ),
    )
    api_key_hashed_key: str | None = Field(
        default=None, foreign_key="api_keys.hashed_key", index=True
    )
    refund_id: str | None = Field(default=None, index=True)
    token_hash: str | None = Field(
        default=None,
        index=True,
        unique=True,
        description="sha256 of the incoming token",
    )
    source_mint: str = Field()
    source_unit: str = Field()
    source_amount: int = Field(description="Gross amount leaving the source mint")
    destination_mint: str = Field()
    destination_unit: str = Field()
    destination_amount: int = Field(description="Net amount minted at the destination")
    fee_reserve: int = Field(default=0)
    input_fees: int = Field(default=0)
    mint_quote_id: str | None = Field(default=None)
    melt_quote_id: str | None = Field(default=None)
    token: str | None = Field(default=None, description="Issued token (out swaps)")
    change_token: str | None = Field(
        default=None,
        description="Unused inbound melt fee reserve returned on the source mint",
    )
    error: str | None = Field(default=None)
    claimed_at: int | None = Field(default=None, description="Reconciler lease")
    created_at: int = Field(default_factory=lambda: int(time.time()))
    updated_at: int = Field(default_factory=lambda: int(time.time()))


async def store_cashu_transaction(
    token: str,
    amount: int,
    unit: str,
    mint_url: str | None = None,
    typ: str = "out",
    request_id: str | None = None,
    collected: bool = False,
    created_at: int | None = None,
    source: str = "x-cashu",
    api_key_hashed_key: str | None = None,
    transaction_id: str | None = None,
    log_failure: bool = True,
) -> bool:
    try:
        async with create_session() as session:
            tx = CashuTransaction(
                id=transaction_id or uuid.uuid4().hex,
                token=token,
                amount=amount,
                unit=unit,
                mint_url=mint_url,
                type=typ,
                request_id=request_id,
                collected=collected,
                created_at=created_at or int(time.time()),
                source=source,
                api_key_hashed_key=api_key_hashed_key,
            )
            session.add(tx)
            await session.commit()
    except Exception:
        if log_failure:
            logger.critical(
                "Failed to store Cashu transaction",
                extra={"type": typ, "request_id": request_id, "source": source},
                exc_info=True,
            )
        raise
    return True


async def _cashu_transaction_exists(transaction_id: str) -> bool:
    async with create_session() as session:
        return await session.get(CashuTransaction, transaction_id) is not None


async def store_cashu_transaction_with_retry(
    token: str,
    amount: int,
    unit: str,
    mint_url: str | None = None,
    typ: str = "out",
    request_id: str | None = None,
    collected: bool = False,
    created_at: int | None = None,
    source: str = "x-cashu",
    api_key_hashed_key: str | None = None,
    max_attempts: int = 3,
) -> bool:
    """Retry a critical Cashu transaction write with bounded backoff."""
    transaction_id = hashlib.sha256(f"{typ}\0{token}".encode()).hexdigest()
    last_error: Exception | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            return await store_cashu_transaction(
                token=token,
                amount=amount,
                unit=unit,
                mint_url=mint_url,
                typ=typ,
                request_id=request_id,
                collected=collected,
                created_at=created_at,
                source=source,
                api_key_hashed_key=api_key_hashed_key,
                transaction_id=transaction_id,
                log_failure=False,
            )
        except IntegrityError as error:
            try:
                if await _cashu_transaction_exists(transaction_id):
                    return True
            except Exception as lookup_error:
                last_error = lookup_error
            else:
                last_error = error
        except Exception as error:
            last_error = error

        if last_error is not None:
            if attempt == max_attempts:
                break
            delay = 0.25 * (2 ** (attempt - 1))
            logger.warning(
                "Cashu transaction storage failed; retrying",
                extra={
                    "type": typ,
                    "request_id": request_id,
                    "attempt": attempt,
                    "max_attempts": max_attempts,
                    "retry_delay_seconds": delay,
                },
            )
            await asyncio.sleep(delay)

    logger.critical(
        "Cashu transaction storage failed after bounded retries",
        extra={
            "type": typ,
            "request_id": request_id,
            "attempts": max_attempts,
            "error": str(last_error),
        },
    )
    if last_error is None:
        raise RuntimeError("Cashu transaction storage failed without an exception")
    raise last_error


class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
    __tablename__ = "upstream_providers"
    __table_args__ = (
        UniqueConstraint(
            "base_url", "api_key", name="uq_upstream_providers_base_url_api_key"
        ),
        {"sqlite_autoincrement": True},
    )
    id: int | None = Field(default=None, primary_key=True)
    slug: str | None = Field(
        default=None,
        unique=True,
        index=True,
        description="Stable external slug used for updates via API key.",
    )
    provider_type: str = Field(
        description="Provider type: custom, openai, anthropic, azure, openrouter, etc."
    )
    base_url: str = Field(description="Base URL of the upstream API")
    api_key: str = Field(description="API key for the upstream provider")
    api_version: str | None = Field(
        default=None, description="API version for Azure OpenAI"
    )
    enabled: bool = Field(default=True, description="Whether this provider is enabled")
    provider_fee: float = Field(
        default=1.01, description="Provider fee multiplier (default 1%)"
    )
    provider_settings: str | None = Field(
        default=None, description="JSON string for provider-specific settings"
    )
    models: list["ModelRow"] = Relationship(
        back_populates="upstream_provider",
        sa_relationship_kwargs={"cascade": "all, delete-orphan"},
    )


class ReservationRelease(SQLModel, table=True):  # type: ignore
    __tablename__ = "reservation_releases"
    __table_args__ = (
        Index("ix_reservation_releases_status_created_at", "status", "created_at"),
    )

    id: str = Field(primary_key=True)
    key_hash: str = Field(index=True)
    billing_key_hash: str = Field(index=True)
    reserved_msats: int
    started_at: int | None = Field(default=None)
    expires_at: int | None = Field(default=None, index=True)
    status: str = Field(default="active")
    created_at: int = Field(default_factory=lambda: int(time.time()))


class RoutstrFee(SQLModel, table=True):  # type: ignore
    __tablename__ = "routstr_fees"
    id: int = Field(default=1, primary_key=True)
    accumulated_msats: int = Field(default=0)
    total_paid_msats: int = Field(default=0)
    last_paid_at: int | None = Field(default=None)
    payout_in_progress_msats: int = Field(default=0)
    payout_started_at: int | None = Field(default=None)
    payout_quote_id: str | None = Field(default=None)
    payout_mint_url: str | None = Field(default=None)
    payout_unit: str | None = Field(default=None)


class NsecState(str, Enum):
    """Ownership state of the node's nsec — an explicit 3-state machine.

    The single ``encrypted_nsec`` column cannot distinguish "never migrated" from
    "intentionally cleared" (both leave it empty), which let a cleared identity be
    resurrected from a stale legacy ``NSEC``. This names the three states so the
    bootstrap branches on ownership rather than inferring it:

    * ``legacy`` — the vault has not taken ownership; a plaintext ``NSEC`` (env or
      old settings blob) may still exist and should be migrated in once.
    * ``encrypted`` — the vault owns a ciphertext; decrypt it, never re-read env.
    * ``cleared`` — the vault owns it but the operator emptied it; stay empty,
      never re-import from a stale legacy copy.
    """

    legacy = "legacy"
    encrypted = "encrypted"
    cleared = "cleared"


class Secret(SQLModel, table=True):  # type: ignore
    """Node-level secrets, stored encrypted/hashed at rest (singleton, id=1).

    The asymmetric column names document the encoding: ``_hash`` is one-way
    (scrypt, verify only) while ``encrypted_`` is reversible (Fernet). Per-provider
    upstream keys live on ``upstream_providers``, not here. See ``routstr.core.vault``.
    """

    __tablename__ = "secrets"
    id: int = Field(default=1, primary_key=True)
    admin_password_hash: str | None = Field(default=None)
    encrypted_nsec: str | None = Field(default=None)
    nsec_state: NsecState = Field(default=NsecState.legacy)
    updated_at: int | None = Field(default=None)


class CliToken(SQLModel, table=True):  # type: ignore
    """Long-lived authorization token for CLI/agent use against admin endpoints."""

    __tablename__ = "cli_tokens"
    id: str = Field(primary_key=True, default_factory=lambda: uuid.uuid4().hex)
    token: str = Field(unique=True, index=True, description="Bearer token value")
    name: str = Field(description="Human-readable label for this token")
    created_at: int = Field(default_factory=lambda: int(time.time()))
    last_used_at: int | None = Field(default=None)
    expires_at: int | None = Field(
        default=None, description="Optional expiry unix timestamp; null = never expires"
    )


async def accumulate_routstr_fee(session: AsyncSession, amount_msats: int) -> None:
    stmt = (
        update(RoutstrFee)
        .where(col(RoutstrFee.id) == 1)
        .values(accumulated_msats=RoutstrFee.accumulated_msats + amount_msats)
    )
    result = await session.exec(stmt)  # type: ignore[call-overload]
    if result.rowcount == 0:
        session.add(RoutstrFee(id=1, accumulated_msats=amount_msats))
    await session.commit()


async def get_routstr_fee(session: AsyncSession) -> RoutstrFee:
    fee = await session.get(RoutstrFee, 1)
    if fee is None:
        fee = RoutstrFee(id=1, accumulated_msats=0, total_paid_msats=0)
        session.add(fee)
        await session.commit()
        await session.refresh(fee)
    return fee


async def get_secret(session: AsyncSession) -> Secret:
    secret = await session.get(Secret, 1)
    if secret is None:
        secret = Secret(id=1)
        session.add(secret)
        try:
            await session.commit()
        except IntegrityError:
            # Another worker created the singleton row between our read and
            # insert (multiple workers booting against one shared DB). Roll back
            # and read the row they committed instead of failing startup.
            await session.rollback()
            secret = await session.get(Secret, 1)
            if secret is None:
                raise
            return secret
        await session.refresh(secret)
    return secret


async def set_admin_password(session: AsyncSession, password: str) -> None:
    """Store the admin password as a one-way hash on the Secret singleton."""
    from .vault import hash_password

    secret = await get_secret(session)
    secret.admin_password_hash = hash_password(password)
    secret.updated_at = int(time.time())
    session.add(secret)
    await session.commit()


async def set_nsec(session: AsyncSession, nsec: str) -> None:
    """Store the node's nsec, Fernet-encrypted, on the Secret singleton.

    An empty string clears it (the node then holds no Nostr identity and signs
    no events). Either way the vault now owns the nsec, so the state moves off
    ``legacy``: a cleared identity (``cleared``) must not be resurrected from a
    stale legacy ``NSEC`` on the next boot.
    """
    from .vault import encrypt

    secret = await get_secret(session)
    secret.encrypted_nsec = encrypt(nsec) if nsec else None
    secret.nsec_state = NsecState.encrypted if nsec else NsecState.cleared
    secret.updated_at = int(time.time())
    session.add(secret)
    await session.commit()


async def reset_routstr_fee(
    session: AsyncSession,
    paid_msats: int,
    quote_id: str,
    mint_url: str,
    unit: str,
) -> bool:
    """Checkpoint a fee payout and its reconciliation metadata before dispatch."""
    stmt = (
        update(RoutstrFee)
        .where(col(RoutstrFee.id) == 1)
        .where(col(RoutstrFee.payout_in_progress_msats) == 0)
        .where(col(RoutstrFee.accumulated_msats) >= paid_msats)
        .values(
            accumulated_msats=RoutstrFee.accumulated_msats - paid_msats,
            payout_in_progress_msats=paid_msats,
            payout_started_at=int(time.time()),
            payout_quote_id=quote_id,
            payout_mint_url=mint_url,
            payout_unit=unit,
        )
    )
    result = await session.exec(stmt)  # type: ignore[call-overload]
    await session.commit()
    return result.rowcount == 1


async def restore_routstr_fee_payout(
    session: AsyncSession,
    paid_msats: int,
    quote_id: str,
    mint_url: str,
    unit: str,
) -> bool:
    """Return the matching unresolved payout to the accumulated fee balance."""
    stmt = (
        update(RoutstrFee)
        .where(col(RoutstrFee.id) == 1)
        .where(col(RoutstrFee.payout_in_progress_msats) == paid_msats)
        .where(col(RoutstrFee.payout_quote_id) == quote_id)
        .where(col(RoutstrFee.payout_mint_url) == mint_url)
        .where(col(RoutstrFee.payout_unit) == unit)
        .values(
            accumulated_msats=RoutstrFee.accumulated_msats + paid_msats,
            payout_in_progress_msats=0,
            payout_started_at=None,
            payout_quote_id=None,
            payout_mint_url=None,
            payout_unit=None,
        )
    )
    result = await session.exec(stmt)  # type: ignore[call-overload]
    await session.commit()
    return result.rowcount == 1


async def complete_routstr_fee_payout(
    session: AsyncSession,
    paid_msats: int,
    quote_id: str,
    mint_url: str,
    unit: str,
) -> bool:
    """Mark the matching checkpoint complete after external payment succeeds."""
    stmt = (
        update(RoutstrFee)
        .where(col(RoutstrFee.id) == 1)
        .where(col(RoutstrFee.payout_in_progress_msats) == paid_msats)
        .where(col(RoutstrFee.payout_quote_id) == quote_id)
        .where(col(RoutstrFee.payout_mint_url) == mint_url)
        .where(col(RoutstrFee.payout_unit) == unit)
        .values(
            payout_in_progress_msats=0,
            payout_started_at=None,
            payout_quote_id=None,
            payout_mint_url=None,
            payout_unit=None,
            total_paid_msats=RoutstrFee.total_paid_msats + paid_msats,
            last_paid_at=int(time.time()),
        )
    )
    result = await session.exec(stmt)  # type: ignore[call-overload]
    await session.commit()
    return result.rowcount == 1


async def record_lightning_payout(
    session: AsyncSession,
    *,
    quote_id: str,
    bolt11: str,
    amount_sats: int,
    mint_url: str,
    destination: str,
) -> None:
    """Record a dispatched payout so it shows up in the Lightning history."""
    session.add(
        LightningInvoice(
            id=uuid.uuid4().hex,
            bolt11=bolt11,
            amount_sats=amount_sats,
            description=f"Payout to {destination}",
            payment_hash=quote_id,
            status="pending",
            direction="out",
            purpose="payout",
            mint_url=mint_url,
            # Payouts settle or fail at the mint; they never expire on our side.
            expires_at=int(time.time()),
        )
    )
    await session.commit()


async def settle_lightning_payout(
    session: AsyncSession,
    quote_id: str,
    *,
    status: str,
    amount_sats: int | None = None,
) -> None:
    result = await session.exec(
        select(LightningInvoice)
        .where(col(LightningInvoice.payment_hash) == quote_id)
        .where(col(LightningInvoice.direction) == "out")
    )
    payout = result.first()
    if payout is None:
        logger.warning(
            "No Lightning payout history row for quote",
            extra={"quote_id": quote_id, "status": status},
        )
        return
    payout.status = status
    if status == "paid":
        payout.paid_at = int(time.time())
        if amount_sats is not None:
            payout.amount_sats = amount_sats
    session.add(payout)
    await session.commit()


UNSETTLED_PAYOUT_STATUSES = ("pending", "reconciliation_required")


async def list_unsettled_lightning_payouts(
    session: AsyncSession, mint_url: str, *, created_before: int
) -> list[LightningInvoice]:
    """Payout rows whose mint outcome was never written back to history."""
    result = await session.exec(
        select(LightningInvoice)
        .where(col(LightningInvoice.direction) == "out")
        .where(col(LightningInvoice.mint_url) == mint_url)
        .where(col(LightningInvoice.status).in_(UNSETTLED_PAYOUT_STATUSES))
        .where(col(LightningInvoice.created_at) < created_before)
        .order_by(col(LightningInvoice.created_at))
    )
    return list(result.all())


async def total_user_liability(db_session: AsyncSession) -> int:
    """Return all outstanding user funds in millisatoshis.

    Key balances and unresolved refunds are summed in one statement so a
    claim opened between two reads cannot be missed by both.
    """
    key_balances = select(func.coalesce(func.sum(ApiKey.balance), 0)).scalar_subquery()
    unresolved_refunds = (
        select(func.coalesce(func.sum(Refund.amount_msats), 0))
        .where(col(Refund.status).in_(REFUND_UNRESOLVED_STATUSES))
        .scalar_subquery()
    )
    result = await db_session.exec(select(key_balances + unresolved_refunds))
    return int(result.one() or 0)


async def user_liability_for_mint_and_unit(
    db_session: AsyncSession, mint_url: str, unit: str
) -> int:
    """Return outstanding user funds that refund from one mint and unit, in msats.

    Single statement, for the same atomicity reason as ``total_user_liability``.
    """
    key_balances = (
        select(func.coalesce(func.sum(ApiKey.balance), 0))
        .where(
            col(ApiKey.refund_mint_url) == mint_url,
            col(ApiKey.refund_currency) == unit,
        )
        .scalar_subquery()
    )
    unresolved_refunds = (
        select(func.coalesce(func.sum(Refund.amount_msats), 0))
        .where(
            col(Refund.status).in_(REFUND_UNRESOLVED_STATUSES),
            col(Refund.mint_url) == mint_url,
            col(Refund.unit) == unit,
        )
        .scalar_subquery()
    )
    result = await db_session.exec(select(key_balances + unresolved_refunds))
    return int(result.one() or 0)


async def balance_for_mint_and_unit(
    db_session: AsyncSession, mint_url: str, unit: str
) -> int:
    """Return the user liability for one mint and unit in millisatoshis."""
    result = await db_session.exec(
        select(func.sum(ApiKey.balance)).where(
            col(ApiKey.refund_mint_url) == mint_url,
            col(ApiKey.refund_currency) == unit,
        )
    )
    return int(result.one() or 0)


async def balances_by_mint_and_unit(
    db_session: AsyncSession, mint_urls: list[str], units: list[str]
) -> dict[tuple[str, str], int]:
    """Return requested user liabilities grouped by mint and unit."""
    if not mint_urls or not units:
        return {}
    query = (
        select(
            col(ApiKey.refund_mint_url),
            col(ApiKey.refund_currency),
            func.sum(ApiKey.balance),
        )
        .where(
            col(ApiKey.refund_mint_url).in_(mint_urls),
            col(ApiKey.refund_currency).in_(units),
        )
        .group_by(col(ApiKey.refund_mint_url), col(ApiKey.refund_currency))
    )
    result = await db_session.exec(query)
    return {
        (mint_url, unit): int(balance or 0)
        for mint_url, unit, balance in result.all()
        if mint_url is not None and unit is not None
    }


async def init_db() -> None:
    """Initializes the database and creates tables if they don't exist."""
    async with engine.begin() as conn:
        if DATABASE_URL.startswith("sqlite"):
            await conn.exec_driver_sql("PRAGMA journal_mode=WAL")
        await conn.run_sync(SQLModel.metadata.create_all)


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session


@asynccontextmanager
async def create_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session


def fix_cashu_migrations() -> None:
    """
    Fixes Cashu wallet migrations that are not idempotent.
    This specifically addresses the 'duplicate column name: public_keys' error
    in the keysets table of Cashu's internal SQLite databases.
    """
    project_root = pathlib.Path(__file__).resolve().parents[2]
    wallet_dir = project_root / ".wallet"

    if not wallet_dir.exists() or not wallet_dir.is_dir():
        return

    logger.info("Checking Cashu wallet databases for migration idempotency")

    for db_file in wallet_dir.glob("*.sqlite3"):
        try:
            conn = sqlite3.connect(db_file)
            cursor = conn.cursor()

            # Check if keysets table exists
            cursor.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='keysets'"
            )
            if not cursor.fetchone():
                conn.close()
                continue

            # Check if public_keys column exists
            cursor.execute("PRAGMA table_info(keysets)")
            columns = [info[1] for info in cursor.fetchall()]

            if "public_keys" not in columns:
                logger.info(f"Adding missing public_keys column to {db_file.name}")
                cursor.execute("ALTER TABLE keysets ADD COLUMN public_keys TEXT")
                conn.commit()

            conn.close()
        except Exception as e:
            logger.warning(f"Could not check/fix Cashu database {db_file}: {e}")


def _clear_alembic_version() -> None:
    """Clear the alembic_version table so stamp/upgrade can proceed."""
    sync_url = DATABASE_URL.replace("+aiosqlite", "")
    from sqlalchemy import create_engine, text

    eng = create_engine(sync_url)
    with eng.begin() as conn:
        conn.execute(text("DELETE FROM alembic_version"))
    eng.dispose()


def run_migrations() -> None:
    """Run Alembic migrations programmatically."""
    try:
        # Run Cashu migration fix first
        fix_cashu_migrations()

        # Get the path to the alembic.ini file
        project_root = pathlib.Path(__file__).resolve().parents[2]
        alembic_ini_path = project_root / "alembic.ini"

        if not alembic_ini_path.exists():
            raise FileNotFoundError(
                f"Alembic configuration file not found at {alembic_ini_path}"
            )

        # Create Alembic config object
        alembic_cfg = Config(str(alembic_ini_path))

        # Set the database URL in the config
        alembic_cfg.set_main_option("sqlalchemy.url", DATABASE_URL)

        try:
            command.upgrade(alembic_cfg, "head")
        except CommandError as e:
            if "Can't locate revision" in str(e):
                logger.warning(
                    "Database stamped with unknown revision (likely from another branch). "
                    "Re-stamping to current head.",
                    extra={"error": str(e)},
                )
                _clear_alembic_version()
                command.stamp(alembic_cfg, "head")
            else:
                raise
        except OperationalError as e:
            if "duplicate column name" in str(e).lower():
                logger.warning(
                    "Migration hit a column that already exists (likely added via "
                    "create_all on another branch). Stamping to current head.",
                    extra={"error": str(e)},
                )
                _clear_alembic_version()
                command.stamp(alembic_cfg, "head")
            else:
                raise

        logger.info("Database migrations completed successfully")

    except Exception as e:
        logger.error(
            "Database migration failed",
            extra={"error": str(e), "error_type": type(e).__name__},
        )
        raise