diff --git a/requirements.txt b/requirements.txt index 18dd761..d653187 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,4 @@ psycopg2-binary==2.9.10 email-validator==2.3.0 confluent-kafka==2.6.1 apscheduler==3.11.0 +httpx==0.28.1 diff --git a/somba/api/app.py b/somba/api/app.py index b0c5e7f..8f1576c 100644 --- a/somba/api/app.py +++ b/somba/api/app.py @@ -13,6 +13,7 @@ from somba.api.errors import APIError, error_response from somba.api.events import router as events_router from somba.api.invoices import router as invoices_router +from somba.api.metrics import router as metrics_router from somba.api.middleware.auth import get_current_merchant from somba.api.middleware.idempotency import IdempotencyMiddleware from somba.api.plans import router as plans_router @@ -30,6 +31,7 @@ app.include_router(subscriptions_router) app.include_router(invoices_router) app.include_router(events_router) +app.include_router(metrics_router) @app.on_event("startup") diff --git a/somba/api/metrics.py b/somba/api/metrics.py new file mode 100644 index 0000000..3481539 --- /dev/null +++ b/somba/api/metrics.py @@ -0,0 +1,64 @@ +"""Operational metrics endpoint: key billing health counters for the merchant.""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from somba.api.middleware.auth import get_current_merchant +from somba.db.models import ( + LedgerIntent, + LedgerIntentStatus, + Merchant, + OutboxEvent, + OutboxEventStatus, + Subscription, + SubscriptionStatus, +) +from somba.db.session import get_db + +router = APIRouter(tags=["metrics"]) + + +@router.get("/v1/metrics") +def get_metrics( + db: Session = Depends(get_db), + merchant: Merchant = Depends(get_current_merchant), +) -> dict: + """Return key operational health counters for the authenticated merchant.""" + mid = merchant.id + + pending_intents = db.scalar( + select(func.count()).select_from(LedgerIntent).where( + LedgerIntent.status == LedgerIntentStatus.pending, + LedgerIntent.merchant_id == mid, + ) + ) + payment_uncertain = db.scalar( + select(func.count()).select_from(Subscription).where( + Subscription.status == SubscriptionStatus.payment_uncertain, + Subscription.merchant_id == mid, + ) + ) + outbox_pending = db.scalar( + select(func.count()).select_from(OutboxEvent).where( + OutboxEvent.status == OutboxEventStatus.pending, + OutboxEvent.merchant_id == mid, + ) + ) + active_subs = db.scalar( + select(func.count()).select_from(Subscription).where( + Subscription.status == SubscriptionStatus.active, + Subscription.merchant_id == mid, + ) + ) + + return { + "metrics": { + "pending_intents": pending_intents, + "payment_uncertain_subscriptions": payment_uncertain, + "outbox_pending_events": outbox_pending, + "active_subscriptions": active_subs, + } + } diff --git a/somba/scheduler/reconciliation_triggers.py b/somba/scheduler/reconciliation_triggers.py index 2c9a7f4..dd581b9 100644 --- a/somba/scheduler/reconciliation_triggers.py +++ b/somba/scheduler/reconciliation_triggers.py @@ -1,2 +1,34 @@ -"""Reconciliation trigger scheduler placeholder.""" +"""Reconciliation scheduler ticks: periodic sweep + verify pass.""" +from __future__ import annotations + +import logging + +from somba.db.session import SessionLocal +from somba.workers.reconcile import sweep, verify_pass + +log = logging.getLogger(__name__) + + +def reconcile_sweep_tick() -> None: + """Fetch recent Nomba transactions and match against pending intents.""" + db = SessionLocal() + try: + resolved = sweep.run(db) + log.info("reconcile.sweep tick: resolved=%d", resolved) + except Exception: + log.exception("reconcile sweep tick failed") + finally: + db.close() + + +def verify_pass_tick() -> None: + """Query Nomba to resolve payment_uncertain subscriptions.""" + db = SessionLocal() + try: + resolved = verify_pass.run(db) + log.info("verify_pass tick: resolved=%d", resolved) + except Exception: + log.exception("verify_pass tick failed") + finally: + db.close() diff --git a/somba/scheduler/runner.py b/somba/scheduler/runner.py index 105be61..894f4f8 100644 --- a/somba/scheduler/runner.py +++ b/somba/scheduler/runner.py @@ -8,6 +8,7 @@ from somba.db.session import SessionLocal from somba.scheduler.billing_sweep import emit_due_billing_events +from somba.scheduler.reconciliation_triggers import reconcile_sweep_tick, verify_pass_tick from somba.observability.alerts import check_pending_intents, PaymentUncertainMonitor @@ -75,7 +76,23 @@ def build_scheduler() -> BlockingScheduler: max_instances=1, coalesce=True, ) - + scheduler.add_job( + reconcile_sweep_tick, + trigger="interval", + minutes=5, + id="reconcile_sweep", + max_instances=1, + coalesce=True, + ) + scheduler.add_job( + verify_pass_tick, + trigger="interval", + minutes=5, + id="verify_pass", + max_instances=1, + coalesce=True, + ) + return scheduler diff --git a/tests/integration/test_smoke.py b/tests/integration/test_smoke.py new file mode 100644 index 0000000..c0991e7 --- /dev/null +++ b/tests/integration/test_smoke.py @@ -0,0 +1,118 @@ +"""Golden path smoke test: full subscription billing lifecycle end-to-end. + +Exercises the complete happy path: + 1. Create plan + customer + subscription via HTTP API + 2. Run billing sweep (charge worker phase 1) — writes LedgerIntent + 3. Run execute_pending (charge worker phase 2) with mocked Nomba success + 4. Assert invoice is paid + 5. Assert charge.succeeded event is in outbox + 6. Assert /v1/metrics reflects the active subscription +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +from sqlalchemy import select + +from somba.db.models import ( + Customer, + Invoice, + InvoiceStatus, + OutboxEvent, + OutboxEventStatus, +) +from somba.nomba.client import NombaChargeResult, NombaChargeStatus +from somba.workers.charge.worker import execute_pending +from somba.workers.charge.worker import run as billing_run + + +def _mock_debit_success(**kwargs) -> NombaChargeResult: + return NombaChargeResult( + status=NombaChargeStatus.succeeded, + transaction_id="txn-smoke-001", + failure_reason=None, + response_code="00", + raw={}, + ) + + +def test_full_billing_lifecycle(api_client, merchant_and_token, db): + """Create plan → customer → subscription, sweep, charge, verify paid invoice.""" + merchant, token = merchant_and_token + auth = {"Authorization": f"Bearer {token}"} + + # 1. Create plan (no trial — next_bill_date = now on subscription creation) + plan_resp = api_client.post( + "/v1/plans", + json={"name": "Monthly Basic", "amount": 5000, "interval": "month", "currency": "NGN"}, + headers={**auth, "Idempotency-Key": "smoke-plan-1"}, + ) + assert plan_resp.status_code == 201, plan_resp.json() + plan_id = plan_resp.json()["plan"]["id"] + + # 2. Create customer + cust_resp = api_client.post( + "/v1/customers", + json={"email": "smoke@test.com", "name": "Smoke User"}, + headers={**auth, "Idempotency-Key": "smoke-cust-1"}, + ) + assert cust_resp.status_code == 201, cust_resp.json() + cust_id = cust_resp.json()["customer"]["id"] + + # Attach mandate directly — the customer API doesn't expose this field yet + customer = db.get(Customer, cust_id) + customer.mandate_id = "mandate-smoke-1" + db.commit() + + # 3. Create subscription — no trial, so status=active, next_bill_date=now + sub_resp = api_client.post( + "/v1/subscriptions", + json={"customer_id": cust_id, "plan_id": plan_id}, + headers={**auth, "Idempotency-Key": "smoke-sub-1"}, + ) + assert sub_resp.status_code == 201, sub_resp.json() + sub_data = sub_resp.json()["subscription"] + sub_id = sub_data["id"] + assert sub_data["status"] == "active" + + # 4. Billing sweep: find due subscriptions → write LedgerIntent + cutoff = datetime.now(tz=timezone.utc) + timedelta(seconds=5) + intents_written = billing_run(db, cutoff=cutoff) + assert intents_written == 1, f"Expected 1 intent written, got {intents_written}" + + # 5. Execute pending intents — mock Nomba to confirm success + with patch("somba.nomba.client.debit_mandate", side_effect=_mock_debit_success): + processed = execute_pending(db, now=cutoff) + assert processed == 1 + + # 6. Invoice is now paid + invoice = db.scalar( + select(Invoice).where( + Invoice.subscription_id == sub_id, + Invoice.status == InvoiceStatus.paid, + ) + ) + assert invoice is not None, "Invoice should be paid after a successful charge" + assert invoice.amount == 5000 + + # 7. Outbox contains charge.succeeded (pending relay to Kafka) + charge_event = db.scalar( + select(OutboxEvent).where( + OutboxEvent.event_type == "charge.succeeded", + OutboxEvent.aggregate_id == str(sub_id), + ) + ) + assert charge_event is not None, "Expected charge.succeeded in outbox" + assert charge_event.status == OutboxEventStatus.pending + + # 8. /v1/metrics reflects the state + metrics_resp = api_client.get("/v1/metrics", headers=auth) + assert metrics_resp.status_code == 200 + m = metrics_resp.json()["metrics"] + assert m["active_subscriptions"] >= 1 + # subscription.created + charge.succeeded are both pending relay + assert m["outbox_pending_events"] >= 2 + assert m["pending_intents"] == 0 # intent was resolved by the charge + assert m["payment_uncertain_subscriptions"] == 0