cardano-checkout-py/cardano_checkout/scheduler.py

"""APScheduler integration for the Cardano payment monitoring loop.

The scheduler drives two jobs against a consumer-supplied
:class:`~cardano_checkout.store.InvoiceStore`:

- :func:`cardano_checkout.monitor.check_pending_invoices` — every 15 seconds
- :func:`cardano_checkout.monitor.reprice_expired_invoices` — every 60 seconds

Usage::

    from cardano_checkout.scheduler import InvoiceScheduler
    from my_app.store import MySqlInvoiceStore

    scheduler = InvoiceScheduler(store=MySqlInvoiceStore())
    await scheduler.start()
    # ... app runs ...
    await scheduler.stop()
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Optional

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger

from cardano_checkout.monitor import (
    DEFAULT_MAX_REPRICINGS,
    DEFAULT_PAYMENT_WINDOW_MINUTES,
    KOIOS_URL,
    PriceFn,
    check_pending_invoices,
    reprice_expired_invoices,
)
from cardano_checkout.store import InvoiceStore

logger = logging.getLogger(__name__)


@dataclass
class InvoiceScheduler:
    """APScheduler harness around the monitor loop.

    Attributes:
        store: Persistence backend. Required.
        koios_url: Chain-query endpoint. Override for testnet / custom gateways.
        check_interval_seconds: How often to poll Koios for pending invoices.
            Defaults to 15.
        reprice_interval_seconds: How often to sweep for expired invoices.
            Defaults to 60.
        payment_window_minutes: Re-expiry window when repricing.
        max_repricings: How many times an invoice can reprice before giving up.
        limit: Max invoices examined per poll cycle.
        job_id_prefix: Scheduler job-id namespace. Override if running multiple
            ``InvoiceScheduler`` instances in one process.
    """

    store: InvoiceStore
    price_fn: Optional[PriceFn] = None
    koios_url: str = KOIOS_URL
    check_interval_seconds: int = 15
    reprice_interval_seconds: int = 60
    payment_window_minutes: int = DEFAULT_PAYMENT_WINDOW_MINUTES
    max_repricings: int = DEFAULT_MAX_REPRICINGS
    limit: int = 100
    job_id_prefix: str = "cardano_checkout"
    _scheduler: Optional[AsyncIOScheduler] = field(default=None, init=False, repr=False)

    async def _job_check_pending(self) -> None:
        try:
            await check_pending_invoices(
                self.store, koios_url=self.koios_url, limit=self.limit
            )
        except Exception:
            logger.exception(
                "[cardano-scheduler] check_pending_invoices job failed"
            )

    async def _job_reprice_expired(self) -> None:
        if self.price_fn is None:
            # No oracle wired — skip repricing. Fixed-ADA invoices don't
            # need one.
            return
        try:
            await reprice_expired_invoices(
                self.store,
                price_fn=self.price_fn,
                window_minutes=self.payment_window_minutes,
                max_repricings=self.max_repricings,
                limit=self.limit,
            )
        except Exception:
            logger.exception(
                "[cardano-scheduler] reprice_expired_invoices job failed"
            )

    async def start(self) -> None:
        """Start the scheduler. Safe to call repeatedly."""
        if self._scheduler and self._scheduler.running:
            logger.debug("[cardano-scheduler] Already running — skipping start")
            return

        self._scheduler = AsyncIOScheduler()

        self._scheduler.add_job(
            self._job_check_pending,
            trigger=IntervalTrigger(seconds=self.check_interval_seconds),
            id=f"{self.job_id_prefix}_check_pending",
            name="Cardano: Check Pending Invoices",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )

        self._scheduler.add_job(
            self._job_reprice_expired,
            trigger=IntervalTrigger(seconds=self.reprice_interval_seconds),
            id=f"{self.job_id_prefix}_reprice_expired",
            name="Cardano: Reprice Expired Invoices",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )

        self._scheduler.start()

        logger.info(
            "[cardano-scheduler] Started — check_pending every %ds, reprice_expired every %ds",
            self.check_interval_seconds,
            self.reprice_interval_seconds,
        )

    async def stop(self) -> None:
        """Stop the scheduler. Idempotent."""
        if self._scheduler:
            self._scheduler.shutdown(wait=False)
            self._scheduler = None
            logger.info("[cardano-scheduler] Stopped")


# ---------------------------------------------------------------------------
# Free-function API around a module-level default instance.
# Prefer the InvoiceScheduler class for anything nontrivial.
# ---------------------------------------------------------------------------

_default: Optional[InvoiceScheduler] = None


async def start_cardano_scheduler(
    store: InvoiceStore, **kwargs
) -> InvoiceScheduler:  # pragma: no cover — convenience shim
    """Start the default :class:`InvoiceScheduler` singleton."""
    global _default
    if _default is None:
        _default = InvoiceScheduler(store=store, **kwargs)
    await _default.start()
    return _default


async def stop_cardano_scheduler() -> None:  # pragma: no cover — convenience shim
    """Stop the default :class:`InvoiceScheduler` singleton."""
    global _default
    if _default is not None:
        await _default.stop()
        _default = None