diff --git a/somba/db/migrations/versions/0004_ledger_intent_created_at.py b/somba/db/migrations/versions/0004_ledger_intent_created_at.py new file mode 100644 index 0000000..38fbbfb --- /dev/null +++ b/somba/db/migrations/versions/0004_ledger_intent_created_at.py @@ -0,0 +1,28 @@ +"""Add created_at to ledger_intents (for the pending-intent age alert). + +Revision ID: 0004 +Revises: 0003 +Create Date: 2026-06-30 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "0004" +down_revision = "0003" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Nullable so SQLite's ADD COLUMN accepts it; the ORM stamps new rows. + op.add_column( + "ledger_intents", + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("ledger_intents", "created_at") diff --git a/somba/db/models.py b/somba/db/models.py index 3275af8..d1d8b21 100644 --- a/somba/db/models.py +++ b/somba/db/models.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import date, datetime +from datetime import date, datetime, timezone from enum import Enum from sqlalchemy import BigInteger, Boolean, Date, DateTime, Enum as SAEnum, ForeignKey, Integer, JSON, String, Text, UniqueConstraint, func @@ -236,6 +236,11 @@ class LedgerIntent(Base): order_reference: Mapped[str] = mapped_column(String(255), unique=True, nullable=False) amount: Mapped[int] = mapped_column(BigInteger, nullable=False) status: Mapped[LedgerIntentStatus] = mapped_column(_enum(LedgerIntentStatus), nullable=False, default=LedgerIntentStatus.pending) + created_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + default=lambda: datetime.now(timezone.utc), + ) class LedgerSettlementSource(str, Enum): diff --git a/somba/observability/__init__.py b/somba/observability/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/somba/observability/alerts.py b/somba/observability/alerts.py new file mode 100644 index 0000000..45da3d3 --- /dev/null +++ b/somba/observability/alerts.py @@ -0,0 +1,104 @@ +"""Observability alerts: surface the correctness risks the PRD demands. + +- Pending ledger intents older than 10 min -> "zero silent lost charges". +- payment_uncertain count stuck for 15 min -> reconciliation worker is down. + +Both are LOG-based: they emit an error-level line for a monitor/log drain to +fire on, and never mutate state. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from somba.db.models import ( + LedgerIntent, + LedgerIntentStatus, + Subscription, + SubscriptionStatus, +) + +log = logging.getLogger(__name__) + +PENDING_INTENT_MAX_AGE_MINUTES = 10 +PAYMENT_UNCERTAIN_STUCK_MINUTES = 15 + + +def _as_utc(dt: datetime) -> datetime: + """Treat a naive datetime (SQLite) as UTC so age math is tz-safe.""" + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + +def check_pending_intents( + db: Session, + max_age_minutes: int = PENDING_INTENT_MAX_AGE_MINUTES, + now: datetime | None = None, +) -> int: + """Alert for every ledger intent stuck pending past max_age_minutes. + + Returns the count of stale intents. PRD rule: every intent must reach a + terminal status; one pending >10m means a charge may be silently lost. + """ + now = now or datetime.now(timezone.utc) + pending = db.scalars( + select(LedgerIntent).where(LedgerIntent.status == LedgerIntentStatus.pending) + ) + + stale = 0 + for intent in pending: + if intent.created_at is None: + continue + age_min = (now - _as_utc(intent.created_at)).total_seconds() / 60 + if age_min >= max_age_minutes: + stale += 1 + log.error( + "ALERT unmatched_intent: id=%d sub=%d order_ref=%s pending %.1f min (>=%d)", + intent.id, intent.subscription_id, intent.order_reference, age_min, max_age_minutes, + ) + if stale: + log.error("ALERT: %d ledger intent(s) pending >= %d min", stale, max_age_minutes) + return stale + + +class PaymentUncertainMonitor: + """Stateful monitor: alerts when the payment_uncertain count is stuck. + + A non-zero count that doesn't change for >= stuck_after_minutes means the + reconciliation worker isn't draining payment_uncertain — i.e. it's down. + Holds state across scheduler ticks, so keep ONE instance. + """ + + def __init__(self, stuck_after_minutes: int = PAYMENT_UNCERTAIN_STUCK_MINUTES) -> None: + self._stuck_after = stuck_after_minutes + self._last_count: int | None = None + self._last_changed: datetime | None = None + + def check(self, db: Session, now: datetime | None = None) -> bool: + """One observation. Returns True if it fired the stuck alert.""" + now = now or datetime.now(timezone.utc) + count = db.scalar( + select(func.count()) + .select_from(Subscription) + .where(Subscription.status == SubscriptionStatus.payment_uncertain) + ) + + if count != self._last_count: + # The count moved — recon is making progress. Reset the clock. + self._last_count = count + self._last_changed = now + return False + + if count and self._last_changed is not None: + stuck_min = (now - self._last_changed).total_seconds() / 60 + if stuck_min >= self._stuck_after: + log.error( + "ALERT payment_uncertain_stuck: %d subscription(s) unchanged for %.1f min " + "(>=%d) — reconciliation worker may be down", + count, stuck_min, self._stuck_after, + ) + return True + return False diff --git a/somba/scheduler/runner.py b/somba/scheduler/runner.py index c318a94..105be61 100644 --- a/somba/scheduler/runner.py +++ b/somba/scheduler/runner.py @@ -8,9 +8,13 @@ from somba.db.session import SessionLocal from somba.scheduler.billing_sweep import emit_due_billing_events +from somba.observability.alerts import check_pending_intents, PaymentUncertainMonitor + log = logging.getLogger(__name__) +# One long-lived monitor — it tracks the payment_uncertain count across ticks. +_payment_uncertain_monitor = PaymentUncertainMonitor() def _billing_sweep_tick() -> None: """One scheduler tick: emit billing.due for everything due now.""" @@ -23,6 +27,27 @@ def _billing_sweep_tick() -> None: finally: db.close() +def _pending_intents_tick() -> None: + """Alert on ledger intents stuck pending past the threshold.""" + db = SessionLocal() + try: + check_pending_intents(db) + except Exception: # noqa: BLE001 + log.exception("pending-intent alert tick failed") + finally: + db.close() + + +def _payment_uncertain_tick() -> None: + """Alert if the payment_uncertain count is stuck (recon worker down).""" + db = SessionLocal() + try: + _payment_uncertain_monitor.check(db) + except Exception: # noqa: BLE001 + log.exception("payment_uncertain monitor tick failed") + finally: + db.close() + def build_scheduler() -> BlockingScheduler: scheduler = BlockingScheduler(timezone="UTC") @@ -34,6 +59,23 @@ def build_scheduler() -> BlockingScheduler: max_instances=1, # never run two sweeps at once coalesce=True, # if ticks were missed, run once on resume, not N times ) + scheduler.add_job( + _pending_intents_tick, + trigger="interval", + minutes=1, + id="pending_intent_alert", + max_instances=1, + coalesce=True, + ) + scheduler.add_job( + _payment_uncertain_tick, + trigger="interval", + minutes=5, + id="payment_uncertain_alert", + max_instances=1, + coalesce=True, + ) + return scheduler diff --git a/tests/integration/test_charge_correctness.py b/tests/integration/test_charge_correctness.py new file mode 100644 index 0000000..699fc29 --- /dev/null +++ b/tests/integration/test_charge_correctness.py @@ -0,0 +1,114 @@ +"""Correctness tests: double-charge impossibility and intent-before-Nomba (crash sim). + +Guards the PRD's hardest invariants: + - Zero double charges (billing lock + intent status + charge-attempt key). + - RPO=0 on the charge path: the ledger intent is durably written BEFORE the + Nomba call, so a crash mid-charge loses nothing and stays recoverable. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import func, select + +from somba.db.models import ( + ChargeAttempt, + LedgerIntent, + LedgerIntentStatus, + SubscriptionStatus, +) +from somba.nomba.client import NombaChargeResult, NombaChargeStatus +from somba.workers.charge import worker as charge_worker +from somba.workers.charge.worker import execute_pending, run + +UTC = timezone.utc +NOW = datetime(2026, 6, 28, 12, 0, 0, tzinfo=UTC) +PAST = NOW - timedelta(hours=1) + + +def _due_sub_with_mandate(db, make_plan, make_customer, make_subscription, merchant): + plan = make_plan(merchant, amount=10_000) + customer = make_customer(merchant) + customer.mandate_id = "mandate_test" # Phase 2 needs a mandate to charge + db.commit() + return make_subscription( + merchant, customer, plan, + status=SubscriptionStatus.active, + next_bill_date=PAST, + current_period_start=PAST - timedelta(days=30), + current_period_end=PAST, + ) + + +def _count(db, model) -> int: + return db.scalar(select(func.count()).select_from(model)) + + +def test_double_charge_impossible_on_repeated_full_run( + db, merchant_and_token, make_plan, make_customer, make_subscription, monkeypatch +): + """Running sweep + execute TWICE charges exactly once.""" + merchant, _ = merchant_and_token + _due_sub_with_mandate(db, make_plan, make_customer, make_subscription, merchant) + + calls = {"n": 0} + + def fake_debit(*, mandate_id, amount_kobo, base_url=None): + calls["n"] += 1 + return NombaChargeResult( + status=NombaChargeStatus.succeeded, + transaction_id="txn_1", + failure_reason=None, + response_code=None, + ) + + monkeypatch.setattr(charge_worker.nomba_client, "debit_mandate", fake_debit) + + # Fire the entire billing flow twice (scheduler + worker both running twice). + for _ in range(2): + run(db, cutoff=NOW) + execute_pending(db, now=NOW) + + assert calls["n"] == 1, f"Nomba was debited {calls['n']}x — double charge!" + assert _count(db, LedgerIntent) == 1 # billing lock -> one intent per period + assert _count(db, ChargeAttempt) == 1 # intent went terminal -> not re-charged + + +def test_intent_written_before_nomba_call_survives_crash( + db, merchant_and_token, make_plan, make_customer, make_subscription, monkeypatch +): + """A crash AT the Nomba call leaves the intent durably pending (RPO=0).""" + merchant, _ = merchant_and_token + sub = _due_sub_with_mandate(db, make_plan, make_customer, make_subscription, merchant) + + # Phase 1 writes + commits the intent BEFORE any Nomba call happens. + run(db, cutoff=NOW) + intent = db.scalar(select(LedgerIntent).where(LedgerIntent.subscription_id == sub.id)) + assert intent is not None + assert intent.status == LedgerIntentStatus.pending + + # Crash exactly at the Nomba call. + called = {"n": 0} + + def boom(*, mandate_id, amount_kobo, base_url=None): + called["n"] += 1 + raise RuntimeError("simulated crash during Nomba call") + + monkeypatch.setattr(charge_worker.nomba_client, "debit_mandate", boom) + + with pytest.raises(RuntimeError): + execute_pending(db, now=NOW) + + # We reached the Nomba seam — proving the intent was written before the call. + assert called["n"] == 1 + + # The intent survived the crash, still pending -> recoverable on restart. + db.expire_all() + survivor = db.scalar(select(LedgerIntent).where(LedgerIntent.subscription_id == sub.id)) + assert survivor is not None + assert survivor.status == LedgerIntentStatus.pending + + # No charge attempt recorded — the crash happened before _handle_success/_failure. + assert _count(db, ChargeAttempt) == 0