Ledger alerts and charge correctness tests

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")

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):

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

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
 
 

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