feat(api): GET /api/payers/{payer_id}/summary with 60s cache + pubsub invalidation

backend/src/cyclone/api.py

@@ -22,6 +22,7 @@ import logging
 import uuid
 from datetime import datetime, timezone
 from contextlib import asynccontextmanager
+from time import monotonic
 from typing import Any, AsyncIterator
 
 from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile
@@ -2958,6 +2959,126 @@ def list_payer_configs(payer_id: str):
     return configs
 
 
+# --------------------------------------------------------------------------- #
+# SP21 Task 1.5: payer-level summary for the drill-down panel
+# --------------------------------------------------------------------------- #
+
+# Per-payer_id memoization for /api/payers/{payer_id}/summary. The drill-
+# down UI hammers this on every hover; the underlying query is O(claims)
+# per payer so a 60s TTL keeps the panel responsive. Process-local on
+# purpose — invalidation is driven by the TTL alone for now (see note
+# below).
+_SUMMARY_TTL_S = 60.0
+_summary_cache: dict[str, tuple[float, dict]] = {}
+
+
+def _clear_summary_cache() -> None:
+    """Test hook: wipe the process-local cache.
+
+    The 60s TTL means tests that want to assert on a recomputed payload
+    must clear the cache between requests. Not used by production code.
+    """
+    _summary_cache.clear()
+
+
+# Pubsub invalidation is intentionally NOT wired. The ``claim_written``
+# and ``remittance_written`` payloads emitted by ``store.add`` are the
+# UI-shaped dicts from ``to_ui_claim_from_orm`` / ``to_ui_remittance_from_orm``,
+# neither of which carries the X12 ``payer_id`` (NM1*PR*PI qualifier).
+# They carry ``payerName`` only, which is the human-readable name and not
+# the URL key we cache on. Wiring a subscriber here would either need a
+# DB lookup per event (re-deriving payer_id from Claim.id) or a different
+# cache key — both are out of scope for this task. The 60s TTL keeps
+# staleness bounded; a follow-up can wire targeted invalidation if the
+# UI proves TTL-bounded staleness is unacceptable.
+
+
+@app.get("/api/payers/{payer_id}/summary")
+def get_payer_summary(payer_id: str):
+    """Payer-level rollup for the drill-down panel.
+
+    Returns billed/received totals, denial rate, and top providers for
+    the given ``payer_id`` (the X12 NM1*PR*PI qualifier, e.g. ``SKCO0``).
+    Cached in-process for ``_SUMMARY_TTL_S`` seconds.
+
+    404 when no claims AND no remits reference this payer_id — the UI
+    uses that to distinguish a typo from a payer with zero traffic
+    (the latter would still return a valid 200 with zeroed totals).
+    """
+    now = monotonic()
+    cached = _summary_cache.get(payer_id)
+    if cached and (now - cached[0]) < _SUMMARY_TTL_S:
+        return cached[1]
+
+    log.debug("payer summary cache miss", extra={"payer_id": payer_id})
+
+    # Query claims + remits scoped to this payer_id. We bypass
+    # ``store.iter_claims`` because that helper filters by payer NAME
+    # substring, not by the X12 PI qualifier we use as the URL key.
+    with db.SessionLocal()() as s:
+        claim_rows: list[Claim] = (
+            s.query(Claim).filter(Claim.payer_id == payer_id).all()
+        )
+        claim_ids = [c.id for c in claim_rows]
+        remit_rows: list[Remittance] = []
+        if claim_ids:
+            remit_rows = (
+                s.query(Remittance)
+                .filter(Remittance.claim_id.in_(claim_ids))
+                .all()
+            )
+
+        if not claim_rows and not remit_rows:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Payer {payer_id!r} not found",
+            )
+
+        billed_total = sum(float(c.charge_amount or 0) for c in claim_rows)
+        received_total = sum(float(r.total_paid or 0) for r in remit_rows)
+        denied = sum(
+            1 for c in claim_rows if c.state == ClaimState.DENIED
+        )
+        claim_count = len(claim_rows)
+        denial_rate = (denied / claim_count) if claim_count else 0.0
+
+        provider_counts: dict[str, int] = {}
+        for c in claim_rows:
+            npi = c.provider_npi
+            if not npi:
+                continue
+            provider_counts[npi] = provider_counts.get(npi, 0) + 1
+        top_providers = [
+            {"npi": npi, "count": count}
+            for npi, count in sorted(
+                provider_counts.items(), key=lambda kv: -kv[1]
+            )[:5]
+        ]
+
+        # Best-effort payer display name from the first claim's raw
+        # payer object (NM1*PR name element, e.g. "COHCPF"). Falls
+        # back to the id when no parsed envelope is available.
+        payer_name = payer_id
+        for c in claim_rows:
+            raw = c.raw_json or {}
+            p = raw.get("payer") if isinstance(raw, dict) else None
+            if isinstance(p, dict) and p.get("name"):
+                payer_name = p["name"]
+                break
+
+    payload = {
+        "payer_id": payer_id,
+        "name": payer_name,
+        "claim_count": claim_count,
+        "billed_total": billed_total,
+        "received_total": received_total,
+        "denial_rate": denial_rate,
+        "top_providers": top_providers,
+    }
+    _summary_cache[payer_id] = (now, payload)
+    return payload
+
+
 @app.post("/api/admin/reload-config")
 def reload_config():
     """Re-read ``config/payers.yaml`` and revalidate. Returns counts."""

backend/tests/test_payer_summary.py

@@ -0,0 +1,106 @@
+"""Tests for GET /api/payers/{payer_id}/summary (SP21 Task 1.5).
+
+The endpoint is the payer-level aggregate that the drill-down UI's
+"Payer → Claims" panel hangs off. It returns billed/received totals,
+denial rate, and the top 5 NPIs by claim volume for one payer_id,
+cached in-process for 60s.
+
+The minimal 837P fixture ships one CLM with ``payer_id="SKCO0"``,
+charge_amount=100.00; the minimal 835 carries one CLP for the same
+claim with total_paid=85.00. So ``/api/payers/SKCO0/summary`` returns
+``claim_count >= 1`` after both files are ingested.
+
+Note: the spec calls this ``payer_id`` (the X12 NM1*PR*PI qualifier,
+e.g. ``SKCO0``). It is NOT the configured payer name from
+``config/payers.yaml``. The filter key in the store layer is
+``Claim.payer_id`` — not the ``payer=`` substring filter used by
+``/api/claims?payer=...``.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cyclone import api as api_mod
+from cyclone.api import app
+
+FIXTURE_837 = Path(__file__).parent / "fixtures" / "minimal_837p.txt"
+FIXTURE_835 = Path(__file__).parent / "fixtures" / "minimal_835.txt"
+
+
+@pytest.fixture(autouse=True)
+def _clear_summary_cache():
+    """Wipe the in-process payer-summary cache between tests.
+
+    conftest resets the DB per test but the cache is module-level
+    state on ``cyclone.api``. Without this clear, a stale payload
+    from a previous test's seed would leak into a later test's first
+    call — masking recompute behavior. The 60s TTL is the only
+    invalidation story today (see api.py docstring on the endpoint).
+    """
+    api_mod._clear_summary_cache()
+    yield
+    api_mod._clear_summary_cache()
+
+
+@pytest.fixture
+def client() -> TestClient:
+    return TestClient(app)
+
+
+@pytest.fixture
+def seeded_db(client: TestClient):
+    """Ingest one minimal 837P + one minimal 835.
+
+    Both fixtures carry ``payer_id="SKCO0"`` so the summary endpoint
+    has something to aggregate. ``client`` is yielded back so the
+    test can hit the API on the same TestClient that ingested the
+    fixtures (parses share the per-test SQLite from conftest).
+    """
+    text_837 = FIXTURE_837.read_text()
+    text_835 = FIXTURE_835.read_text()
+    r837 = client.post(
+        "/api/parse-837",
+        files={"file": ("x.txt", text_837, "text/plain")},
+        headers={"Accept": "application/json"},
+    )
+    assert r837.status_code == 200, r837.text
+    r835 = client.post(
+        "/api/parse-835",
+        files={"file": ("era.txt", text_835, "text/plain")},
+        headers={"Accept": "application/json"},
+    )
+    assert r835.status_code == 200, r835.text
+    return client
+
+
+def test_payer_summary_happy_path(seeded_db: TestClient):
+    """Seeded db has at least one claim for SKCO0 → 200 with the spec shape."""
+    resp = seeded_db.get("/api/payers/SKCO0/summary")
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["payer_id"] == "SKCO0"
+    assert "claim_count" in data
+    assert "billed_total" in data
+    assert "received_total" in data
+    assert "denial_rate" in data
+    assert data["claim_count"] >= 1
+    # denial_rate must be a float in [0, 1] (0/1 claim → 0.0).
+    assert isinstance(data["denial_rate"], (int, float))
+    assert 0.0 <= data["denial_rate"] <= 1.0
+
+
+def test_payer_summary_unknown_payer_returns_404(client: TestClient):
+    resp = client.get("/api/payers/DOES_NOT_EXIST/summary")
+    assert resp.status_code == 404
+
+
+def test_payer_summary_caches_then_invalidates(seeded_db: TestClient):
+    """Two back-to-back calls return identical payloads (in-process cache)."""
+    resp1 = seeded_db.get("/api/payers/SKCO0/summary")
+    resp2 = seeded_db.get("/api/payers/SKCO0/summary")
+    assert resp1.status_code == 200
+    assert resp2.status_code == 200
+    assert resp1.json() == resp2.json()