wowcheckout/api/app/reconciler.py

from __future__ import annotations

import logging
import time
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_DOWN

from sqlalchemy import and_, or_, text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from .btcpay_webhooks import dispatch_btcpay_webhooks
from .config import INVOICE_RECONCILE_INTERVAL_SECONDS, LATE_PAYMENT_LOOKBACK_HOURS
from .db import SessionLocal
from .models import Invoice, InvoiceTransfer, SystemStatus, User
from .monero_service import MoneroWalletService, TransferDetail
from .webhooks import dispatch_webhooks

logger = logging.getLogger(__name__)
MONERO_CONNECTIVITY_STATUS_NAME = "monero_connectivity"


def main() -> None:
    level_name = "INFO"
    try:
        level_name = __import__("os").getenv("LOG_LEVEL", "INFO")
    except Exception:
        level_name = "INFO"
    level = getattr(logging, level_name.upper(), logging.INFO)
    logging.basicConfig(level=level)

    # Silence expected RPC -13 ("No wallet file") spam from the python-monero
    # library. Those happen routinely when close_wallet is called on an idle
    # backend (by _ensure_wallet_open) and are handled by our retry logic;
    # they are not actionable errors and were drowning real log output.
    # Real wallet-rpc problems still bubble up as exceptions which we log
    # explicitly through logger.error / logger.exception.
    logging.getLogger("monero.backends.jsonrpc.wallet").setLevel(logging.CRITICAL)

    while True:
        status_db: Session | None = None
        try:
            status_db = SessionLocal()
            service = MoneroWalletService()
            _safe_update_monero_connectivity_status(status_db, service)
            _safe_update_reconciler_status(
                status_db,
                started_at=datetime.now(timezone.utc),
                completed_at=None,
                error_message=None,
            )
            _reconcile_invoices(service)
            _safe_update_reconciler_status(
                status_db,
                completed_at=datetime.now(timezone.utc),
                error_message=None,
            )
        except Exception as exc:
            logger.exception("Invoice reconcile failed: %s", exc)
            if status_db is None:
                status_db = SessionLocal()
            _safe_update_monero_connectivity_error(status_db)
            _safe_update_reconciler_status(
                status_db,
                error_message=str(exc),
            )
        finally:
            if status_db is not None:
                status_db.close()
        time.sleep(INVOICE_RECONCILE_INTERVAL_SECONDS)


def _reconcile_invoices(service: MoneroWalletService) -> None:
    db: Session = SessionLocal()
    try:
        now = datetime.now(timezone.utc)
        late_cutoff = now - timedelta(hours=max(0, LATE_PAYMENT_LOOKBACK_HOURS))
        invoices = (
            db.query(Invoice)
            .filter(
                or_(
                    Invoice.status.in_(["pending", "payment_detected"]),
                    and_(
                        Invoice.status == "expired",
                        Invoice.expires_at.is_not(None),
                        Invoice.expires_at >= late_cutoff,
                    ),
                )
            )
            .order_by(Invoice.created_at.asc())
            .all()
        )
        user_groups: dict[object, list[Invoice]] = {}
        for invoice in invoices:
            if invoice.user_id is None:
                logger.debug(
                    "Skipping invoice without user",
                    extra={"invoice_id": str(invoice.id)},
                )
                continue
            user_groups.setdefault(invoice.user_id, []).append(invoice)
        logger.debug(
            "Reconciling %d invoices across %d users",
            len(invoices),
            len(user_groups),
        )
        for user_id, user_invoices in user_groups.items():
            user = db.query(User).filter(User.id == user_id).first()
            if user is None:
                logger.debug(
                    "Skipping invoices with missing user",
                    extra={"user_id": str(user_id)},
                )
                continue
            if not user.payment_address or not user.view_key_encrypted:
                logger.debug(
                    "Skipping invoices without payment address",
                    extra={"user_id": str(user.id)},
                )
                continue
            for invoice in user_invoices:
                # Advisory lock prevents duplicate reconciliation during deploys
                lock_id = hash(str(invoice.id)) & 0x7FFFFFFF
                acquired = db.execute(
                    text("SELECT pg_try_advisory_lock(:id)"),
                    {"id": lock_id},
                ).scalar()
                if not acquired:
                    continue
                try:
                    transfers = _get_transfers_with_retry(
                        service, user, invoice.address,
                    )

                    total_atomic = 0
                    max_confirmations = 0
                    for transfer in transfers:
                        if transfer.amount_atomic <= 0:
                            continue
                        total_atomic += transfer.amount_atomic
                        if transfer.confirmations > max_confirmations:
                            max_confirmations = transfer.confirmations
                    logger.debug(
                        "Invoice totals",
                        extra={
                            "invoice_id": str(invoice.id),
                            "received_atomic": total_atomic,
                            "confirmations": max_confirmations,
                        },
                    )
                    now = datetime.now(timezone.utc)
                    previous_confirmations = invoice.confirmations or 0
                    total_changed = invoice.total_paid_atomic != total_atomic
                    confirmations_changed = previous_confirmations != max_confirmations
                    transfers_changed = _sync_invoice_transfers(
                        db,
                        invoice=invoice,
                        transfers=transfers,
                    )
                    if total_changed or confirmations_changed or transfers_changed:
                        if confirmations_changed:
                            invoice.confirmations = max_confirmations
                        if total_changed:
                            invoice.total_paid_atomic = total_atomic
                        db.add(invoice)
                        db.commit()
                    required_atomic = _xmr_to_atomic(invoice.amount_xmr)
                    is_paid = total_atomic >= required_atomic

                    expires_at = invoice.expires_at
                    if expires_at is not None and expires_at.tzinfo is None:
                        expires_at = expires_at.replace(tzinfo=timezone.utc)
                    is_after_expiry = bool(expires_at and now >= expires_at)

                    if total_atomic < required_atomic:
                        logger.debug(
                            "Payment not yet detected",
                            extra={
                                "invoice_id": str(invoice.id),
                                "required_atomic": required_atomic,
                                "received_atomic": total_atomic,
                            },
                        )
                        if invoice.status == "pending" and is_after_expiry:
                            logger.info(
                                "Invoice expired",
                                extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
                            )
                            invoice.status = "expired"
                            db.add(invoice)
                            db.commit()
                            dispatch_webhooks(db, str(user.id), "invoice.expired", invoice)
                            dispatch_btcpay_webhooks(
                                db, str(user.id), "InvoiceExpired", invoice
                            )
                        continue

                    if is_paid and invoice.status in ("pending", "expired"):
                        logger.info(
                            "Invoice marked payment detected",
                            extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
                        )
                        previous_status = invoice.status
                        invoice.status = "payment_detected"
                        if invoice.detected_at is None:
                            invoice.detected_at = now
                        if previous_status == "expired" or (previous_status == "pending" and is_after_expiry):
                            invoice.paid_after_expiry = True
                            if invoice.paid_after_expiry_at is None:
                                invoice.paid_after_expiry_at = now
                        db.add(invoice)
                        db.commit()
                        dispatch_webhooks(db, str(user.id), "invoice.payment_detected", invoice)
                        dispatch_btcpay_webhooks(
                            db, str(user.id), "InvoiceReceivedPayment", invoice
                        )
                        dispatch_btcpay_webhooks(db, str(user.id), "InvoicePaidInFull", invoice)
                        dispatch_btcpay_webhooks(
                            db, str(user.id), "InvoiceProcessing", invoice
                        )
                    if max_confirmations >= invoice.confirmation_target and invoice.status != "confirmed":
                        logger.info(
                            "Invoice confirmed",
                            extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
                        )
                        invoice.status = "confirmed"
                        if invoice.confirmed_at is None:
                            invoice.confirmed_at = now
                        db.add(invoice)
                        db.commit()
                        dispatch_webhooks(db, str(user.id), "invoice.confirmed", invoice)
                        dispatch_btcpay_webhooks(db, str(user.id), "InvoiceSettled", invoice)
                        dispatch_btcpay_webhooks(
                            db, str(user.id), "InvoicePaymentSettled", invoice
                        )
                except Exception as exc:
                    logger.error(
                        "Failed to reconcile invoice: %s",
                        exc,
                        extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
                    )
                    try:
                        db.rollback()
                    except Exception:
                        pass
                finally:
                    db.execute(text("SELECT pg_advisory_unlock(:id)"), {"id": lock_id})
    finally:
        db.close()


def _get_transfers_with_retry(
    service: MoneroWalletService,
    user: User,
    address: str,
    retries: int = 3,
) -> list[TransferDetail]:
    """Call get_transfers_for_address with retry + backoff on RPC errors."""
    delays = [0.5, 1.0, 2.0]
    last_exc: Exception | None = None
    for attempt in range(retries):
        try:
            return service.get_transfers_for_address(user=user, address=address)
        except Exception as exc:
            last_exc = exc
            if attempt < retries - 1:
                time.sleep(delays[attempt])
    raise last_exc  # type: ignore[misc]


def _sync_invoice_transfers(
    db: Session,
    *,
    invoice: Invoice,
    transfers: list[TransferDetail],
) -> bool:
    existing = (
        db.query(InvoiceTransfer)
        .filter(InvoiceTransfer.invoice_id == invoice.id)
        .all()
    )
    existing_by_txid = {transfer.txid: transfer for transfer in existing if transfer.txid}
    seen_txids: set[str] = set()
    changed = False
    for transfer in transfers:
        txid = transfer.txid
        if not txid:
            continue
        seen_txids.add(txid)
        stored = existing_by_txid.get(txid)
        if stored is None:
            db.add(
                InvoiceTransfer(
                    invoice_id=invoice.id,
                    txid=txid,
                    amount_atomic=transfer.amount_atomic,
                    confirmations=transfer.confirmations,
                    timestamp=transfer.timestamp,
                    address=transfer.address,
                )
            )
            changed = True
            continue
        if (
            stored.amount_atomic != transfer.amount_atomic
            or stored.confirmations != transfer.confirmations
            or stored.timestamp != transfer.timestamp
            or stored.address != transfer.address
        ):
            stored.amount_atomic = transfer.amount_atomic
            stored.confirmations = transfer.confirmations
            stored.timestamp = transfer.timestamp
            stored.address = transfer.address
            changed = True
    # Only delete transfers if we got data back — don't wipe records on RPC glitch
    if transfers:
        for stored in existing:
            if stored.txid not in seen_txids:
                db.delete(stored)
                changed = True
    return changed


def _update_reconciler_status(
    db: Session,
    *,
    started_at: datetime | None = None,
    completed_at: datetime | None = None,
    error_message: str | None = None,
) -> None:
    status_row = db.query(SystemStatus).filter(SystemStatus.name == "reconciler").first()
    if status_row is None:
        status_row = SystemStatus(name="reconciler")
    if started_at is not None:
        status_row.last_reconcile_started_at = started_at
    if completed_at is not None:
        status_row.last_reconcile_completed_at = completed_at
    status_row.last_reconcile_error = error_message
    db.add(status_row)
    db.commit()


def _update_monero_connectivity_status(
    db: Session,
    *,
    wallet_rpc: str,
    daemon: str,
    daemon_height: int | None,
) -> None:
    status_row = (
        db.query(SystemStatus)
        .filter(SystemStatus.name == MONERO_CONNECTIVITY_STATUS_NAME)
        .first()
    )
    if status_row is None:
        status_row = SystemStatus(name=MONERO_CONNECTIVITY_STATUS_NAME)
    status_row.wallet_rpc = wallet_rpc
    status_row.daemon = daemon
    status_row.daemon_height = daemon_height
    status_row.checked_at = datetime.now(timezone.utc)
    db.add(status_row)
    db.commit()


def _safe_update_monero_connectivity_status(
    db: Session,
    service: MoneroWalletService,
) -> None:
    try:
        status_payload = service.get_status()
        _update_monero_connectivity_status(
            db,
            wallet_rpc=str(status_payload.get("wallet_rpc", "unreachable")),
            daemon=str(status_payload.get("daemon", "unknown")),
            daemon_height=(
                int(status_payload["daemon_height"])
                if isinstance(status_payload.get("daemon_height"), int)
                else None
            ),
        )
    except Exception:
        db.rollback()
        logger.warning("Unable to persist Monero connectivity status", exc_info=True)


def _safe_update_monero_connectivity_error(db: Session) -> None:
    try:
        _update_monero_connectivity_status(
            db,
            wallet_rpc="unreachable",
            daemon="unknown",
            daemon_height=None,
        )
    except SQLAlchemyError:
        db.rollback()
        logger.warning("Unable to persist Monero connectivity fallback", exc_info=True)


def _safe_update_reconciler_status(
    db: Session,
    *,
    started_at: datetime | None = None,
    completed_at: datetime | None = None,
    error_message: str | None = None,
) -> None:
    try:
        _update_reconciler_status(
            db,
            started_at=started_at,
            completed_at=completed_at,
            error_message=error_message,
        )
    except SQLAlchemyError:
        db.rollback()
        logger.warning("Unable to persist reconciler status heartbeat", exc_info=True)




def _xmr_to_atomic(amount: Decimal) -> int:
    quantized = (Decimal(amount) * Decimal("1000000000000")).to_integral_value(
        rounding=ROUND_DOWN
    )
    return int(quantized)


if __name__ == "__main__":
    main()