Sulkta-OSS/cardano-checkout-py
3
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

v0.2: refactor monitor + scheduler around InvoiceStore Protocol

Browse source
This commit is contained in:
Sulkta 2026-04-23 19:55:28 -07:00
parent 409005415e
commit fb1e1b73a6
6 changed files with 952 additions and 596 deletions
Download patch file Download diff file Expand all files Collapse all files

561
cardano_checkout/scheduler.py
View file

@ -1,465 +1,170 @@
"""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
That's the *full* SDK job surface. The subscription-level + grace-period
jobs that the original the host app scheduler shipped are the host app-specific
(they touch ``Company``, ``Subscription``, ``SubscriptionPayment`` models
that are merchant-specific) — those live in
:mod:`cardano_checkout.hostapp_compat` so the host app can still import the
exact wrappers it has always used, without polluting the generic SDK.
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()
"""
Cardano Payment Monitoring Scheduler
APScheduler integration that runs five recurring jobs:
- check_pending_payments — every 15 seconds
- reprice_expired_payments — every 60 seconds
- _check_subscription_payments — every 60 seconds
- _reprice_subscription_payments — every 6 hours
- _enforce_grace_period — daily at 06:00 UTC
from __future__ import annotations
Usage in main.py lifespan:
from services.cardano_scheduler import start_cardano_scheduler, stop_cardano_scheduler
@asynccontextmanager
async def lifespan(app: FastAPI):
await start_cardano_scheduler()
yield
await stop_cardano_scheduler()
"""
import logging
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from dataclasses import dataclass, field
from typing import Optional
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy import select
from database import async_session_maker
from services.cardano_monitor import (
_check_address_utxos,
_evaluate_payment,
check_pending_payments,
reprice_expired_payments,
from cardano_checkout.monitor import (
DEFAULT_MAX_REPRICINGS,
DEFAULT_PAYMENT_WINDOW_MINUTES,
KOIOS_URL,
check_pending_invoices,
reprice_expired_invoices,
)
from services.cardano_price import convert_usd_to_lovelace, get_ada_usd_price
from cardano_checkout.store import InvoiceStore
logger = logging.getLogger(__name__)
_scheduler: Optional[AsyncIOScheduler] = None
@dataclass
class InvoiceScheduler:
"""APScheduler harness around the monitor loop.
# =============================================================================
# Scheduled job wrappers — invoice payments
# =============================================================================
async def _job_check_pending() -> None:
"""Scheduler wrapper for check_pending_payments."""
try:
async with async_session_maker() as db:
await check_pending_payments(db)
except Exception:
logger.exception("[cardano-scheduler] check_pending_payments job failed")
async def _job_reprice_expired() -> None:
"""Scheduler wrapper for reprice_expired_payments."""
try:
async with async_session_maker() as db:
await reprice_expired_payments(db)
except Exception:
logger.exception("[cardano-scheduler] reprice_expired_payments job failed")
# =============================================================================
# Scheduled job wrappers — subscription payments
# =============================================================================
async def _job_check_subscription_payments() -> None:
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 — identical to the host app's production cadence.
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.
"""
Poll Koios for UTXOs at awaiting_payment subscription addresses.
Reuses _check_address_utxos and _evaluate_payment from cardano_monitor.
On confirmation, advances subscription status to 'active' and updates
company.subscription_tier to match the subscription's tier.
"""
from models import Company, Subscription, SubscriptionPayment
store: InvoiceStore
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)
try:
async with async_session_maker() as db:
now = datetime.now(timezone.utc)
result = await db.execute(
select(SubscriptionPayment).where(
SubscriptionPayment.status.in_(["awaiting_payment", "underpaid"]),
SubscriptionPayment.expires_at > now,
)
async def _job_check_pending(self) -> None:
try:
await check_pending_invoices(
self.store, koios_url=self.koios_url, limit=self.limit
)
payments = result.scalars().all()
if not payments:
return
logger.debug(
"[cardano-scheduler] Checking %d subscription payment(s)", len(payments)
except Exception:
logger.exception(
"[cardano-scheduler] check_pending_invoices job failed"
)
for sp in payments:
try:
utxos = await _check_address_utxos(sp.address)
# _evaluate_payment expects a CardanoPayment-like object.
# SubscriptionPayment has the same fields it needs.
new_status, raw_lovelace, total_value, received_assets, tx_hash = (
await _evaluate_payment(sp, utxos)
)
# Map monitor statuses to subscription payment statuses
status_map = {
"pending": "awaiting_payment",
"confirmed": "confirmed",
"overpaid": "overpaid",
"underpaid": "underpaid",
}
mapped_status = status_map.get(new_status, new_status)
if mapped_status == sp.status and raw_lovelace == 0:
continue
sp.received_lovelace = raw_lovelace
sp.total_value_lovelace = total_value
sp.received_assets = received_assets
if tx_hash:
sp.tx_hash = tx_hash
if mapped_status != sp.status:
old_status = sp.status
sp.status = mapped_status
if mapped_status in ("confirmed", "overpaid"):
sp.confirmed_at = now
# Advance subscription + company tier
if sp.subscription_id:
sub_result = await db.execute(
select(Subscription).where(
Subscription.id == sp.subscription_id
)
)
sub = sub_result.scalar_one_or_none()
if sub:
sub.status = "active"
sub.updated_at = now
# Apply any pending tier downgrade
if sub.pending_tier and sp.period_end and sp.period_end <= now.date():
sub.tier = sub.pending_tier
sub.pending_tier = None
sub.pending_tier_at = None
company_result = await db.execute(
select(Company).where(Company.id == sp.company_id)
)
company = company_result.scalar_one_or_none()
if company:
# Get current sub tier
sub_result2 = await db.execute(
select(Subscription).where(
Subscription.company_id == sp.company_id
)
)
sub2 = sub_result2.scalar_one_or_none()
if sub2:
company.subscription_tier = sub2.tier
company.subscription_status = "active"
logger.info(
"[cardano-scheduler] sub_payment #%d company_id=%d: %s -> %s"
" (%.6f ADA received)",
sp.id,
sp.company_id,
old_status,
mapped_status,
raw_lovelace / 1_000_000,
)
except Exception as e:
logger.exception(
"[cardano-scheduler] Error checking sub_payment #%d: %s", sp.id, e
)
await db.commit()
except Exception:
logger.exception("[cardano-scheduler] _check_subscription_payments job failed")
async def _job_reprice_subscription_payments() -> None:
"""
Reprice awaiting_payment subscription records whose 24-hour window has expired.
Fetches the current ADA price, recalculates expected_lovelace, resets expires_at
to now + 24 hours, and increments repriced_count. Gives up after 3 repricings.
"""
from models import SubscriptionPayment
try:
async with async_session_maker() as db:
now = datetime.now(timezone.utc)
result = await db.execute(
select(SubscriptionPayment).where(
SubscriptionPayment.status == "awaiting_payment",
SubscriptionPayment.expires_at <= now,
SubscriptionPayment.repriced_count < 3,
)
async def _job_reprice_expired(self) -> None:
try:
await reprice_expired_invoices(
self.store,
window_minutes=self.payment_window_minutes,
max_repricings=self.max_repricings,
limit=self.limit,
)
payments = result.scalars().all()
if not payments:
return
logger.info(
"[cardano-scheduler] Repricing %d subscription payment(s)", len(payments)
except Exception:
logger.exception(
"[cardano-scheduler] reprice_expired_invoices job failed"
)
ada_price = await get_ada_usd_price()
if ada_price <= 0:
logger.warning(
"[cardano-scheduler] Cannot reprice subscriptions — ADA price unavailable"
)
return
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
new_expires_at = now + timedelta(hours=24)
self._scheduler = AsyncIOScheduler()
for sp in payments:
try:
total_usd = float(sp.expected_usd or 0)
if total_usd <= 0:
sp.status = "expired"
logger.warning(
"[cardano-scheduler] sub_payment #%d has no expected_usd — expired",
sp.id,
)
continue
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,
)
new_lovelace = await convert_usd_to_lovelace(total_usd)
if new_lovelace == 0:
logger.warning(
"[cardano-scheduler] sub_payment #%d: lovelace conversion returned 0, skipping",
sp.id,
)
continue
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,
)
old_lovelace = sp.expected_lovelace
sp.expected_lovelace = new_lovelace
sp.ada_price_usd = Decimal(str(round(ada_price, 4)))
sp.expires_at = new_expires_at
sp.repriced_count += 1
self._scheduler.start()
logger.info(
"[cardano-scheduler] Repriced sub_payment #%d: %d -> %d lovelace"
" (ADA=$%.4f, reprice #%d)",
sp.id,
old_lovelace or 0,
new_lovelace,
ada_price,
sp.repriced_count,
)
logger.info(
"[cardano-scheduler] Started — check_pending every %ds, reprice_expired every %ds",
self.check_interval_seconds,
self.reprice_interval_seconds,
)
except Exception as e:
logger.exception(
"[cardano-scheduler] Error repricing sub_payment #%d: %s", sp.id, e
)
# Mark max-repriced payments as expired
expired_result = await db.execute(
select(SubscriptionPayment).where(
SubscriptionPayment.status == "awaiting_payment",
SubscriptionPayment.expires_at <= now,
SubscriptionPayment.repriced_count >= 3,
)
)
for sp in expired_result.scalars().all():
sp.status = "expired"
logger.info(
"[cardano-scheduler] sub_payment #%d expired after %d repricings",
sp.id,
sp.repriced_count,
)
await db.commit()
except Exception:
logger.exception("[cardano-scheduler] _reprice_subscription_payments job failed")
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")
async def _job_enforce_grace_period() -> None:
"""
Daily enforcement of subscription grace periods (runs at 06:00 UTC).
# ---------------------------------------------------------------------------
# Backwards-compatible free-function API
# ---------------------------------------------------------------------------
#
# Early adopters may have imported ``start_cardano_scheduler`` / ``stop_cardano_scheduler``
# directly. Provide those as thin wrappers around a module-level default instance.
# Using the InvoiceScheduler class is preferred for anything nontrivial.
Rules:
- If due_date has passed and payment is not confirmed: mark subscription past_due
- If grace_deadline has passed and payment still not confirmed: suspend subscription
and update company.subscription_status accordingly
"""
from models import Company, Subscription, SubscriptionPayment
try:
async with async_session_maker() as db:
today = datetime.now(timezone.utc).date()
# Find subscriptions that are active/past_due and have an overdue payment
overdue_result = await db.execute(
select(SubscriptionPayment).where(
SubscriptionPayment.status.in_(["awaiting_payment", "underpaid", "expired"]),
SubscriptionPayment.due_date < today,
)
)
overdue_payments = overdue_result.scalars().all()
for sp in overdue_payments:
try:
sub_result = await db.execute(
select(Subscription).where(
Subscription.company_id == sp.company_id
)
)
sub = sub_result.scalar_one_or_none()
if not sub or sub.status in ("cancelled", "suspended"):
continue
company_result = await db.execute(
select(Company).where(Company.id == sp.company_id)
)
company = company_result.scalar_one_or_none()
# Grace deadline passed → suspend
if sp.grace_deadline and today > sp.grace_deadline:
if sub.status != "suspended":
sub.status = "suspended"
sub.updated_at = datetime.now(timezone.utc)
if company:
company.subscription_status = "suspended"
logger.info(
"[cardano-scheduler] company_id=%d subscription suspended"
" (grace deadline %s passed)",
sp.company_id,
sp.grace_deadline,
)
# Due date passed but within grace → mark past_due
elif sub.status == "active":
sub.status = "past_due"
sub.updated_at = datetime.now(timezone.utc)
if company:
company.subscription_status = "past_due"
logger.info(
"[cardano-scheduler] company_id=%d subscription past_due"
" (due_date %s passed)",
sp.company_id,
sp.due_date,
)
except Exception as e:
logger.exception(
"[cardano-scheduler] Error enforcing grace period for sub_payment #%d: %s",
sp.id,
e,
)
await db.commit()
except Exception:
logger.exception("[cardano-scheduler] _enforce_grace_period job failed")
_default: Optional[InvoiceScheduler] = None
# =============================================================================
# Lifecycle
# =============================================================================
async def start_cardano_scheduler() -> None:
"""
Start the Cardano payment monitoring scheduler.
Registers five jobs:
- check_pending_payments: every 15 seconds
- reprice_expired_payments: every 60 seconds
- check_subscription_payments: every 60 seconds
- reprice_subscription_payments: every 6 hours
- enforce_grace_period: daily at 06:00 UTC
Safe to call multiple times — skips if already running.
"""
global _scheduler
if _scheduler and _scheduler.running:
logger.debug("[cardano-scheduler] Already running — skipping start")
return
_scheduler = AsyncIOScheduler()
_scheduler.add_job(
_job_check_pending,
trigger=IntervalTrigger(seconds=15),
id="cardano_check_pending",
name="Cardano: Check Pending Payments",
replace_existing=True,
max_instances=1,
coalesce=True,
)
_scheduler.add_job(
_job_reprice_expired,
trigger=IntervalTrigger(seconds=60),
id="cardano_reprice_expired",
name="Cardano: Reprice Expired Payments",
replace_existing=True,
max_instances=1,
coalesce=True,
)
_scheduler.add_job(
_job_check_subscription_payments,
trigger=IntervalTrigger(seconds=60),
id="cardano_check_sub_payments",
name="Cardano: Check Subscription Payments",
replace_existing=True,
max_instances=1,
coalesce=True,
)
_scheduler.add_job(
_job_reprice_subscription_payments,
trigger=IntervalTrigger(hours=6),
id="cardano_reprice_sub_payments",
name="Cardano: Reprice Subscription Payments",
replace_existing=True,
max_instances=1,
coalesce=True,
)
_scheduler.add_job(
_job_enforce_grace_period,
trigger=CronTrigger(hour=6, minute=0, timezone="UTC"),
id="cardano_enforce_grace",
name="Cardano: Enforce Subscription Grace Periods",
replace_existing=True,
max_instances=1,
coalesce=True,
)
_scheduler.start()
logger.info(
"[cardano-scheduler] Started — check_pending every 15s, reprice_expired every 60s,"
" check_sub_payments every 60s, reprice_sub_payments every 6h,"
" enforce_grace_period daily at 06:00 UTC"
)
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:
"""
Gracefully stop the Cardano scheduler.
Call this in the FastAPI lifespan shutdown block.
"""
global _scheduler
if _scheduler:
_scheduler.shutdown(wait=False)
_scheduler = None
logger.info("[cardano-scheduler] Stopped")
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