diff --git a/backend/src/cyclone/store.py b/backend/src/cyclone/store.py index 9f11286..7c3c247 100644 --- a/backend/src/cyclone/store.py +++ b/backend/src/cyclone/store.py @@ -809,7 +809,12 @@ class CycloneStore: # -- write path ----------------------------------------------------- - def add(self, record: BatchRecord) -> None: + def add( + self, + record: BatchRecord, + *, + event_bus: "EventBus | None" = None, + ) -> None: """Persist a parsed batch (837P or 835) to the DB. For 837P batches: inserts the Batch row, one Claim row per @@ -830,10 +835,22 @@ class CycloneStore: has a fresh ``uuid4`` id from the API). O(n) per row, but acceptable for the small fixture sizes — production load is one batch at a time via the API, not bulk inserts. + + When ``event_bus`` is provided, publishes one ``claim_written`` + or ``remittance_written`` event per newly-inserted row plus an + ``activity_recorded`` event per activity row, after commit. The + publish calls are best-effort — failures are logged but do not + roll back the persisted batch. """ + from cyclone.pubsub import EventBus + import logging log = logging.getLogger(__name__) + # Track rows we actually inserted so we can publish events for them. + inserted_claim_ids: list[str] = [] + inserted_remit_ids: list[str] = [] + with db.SessionLocal()() as s: batch_row = Batch( id=record.id, @@ -870,6 +887,7 @@ class CycloneStore: "amount": float(claim.claim.total_charge or 0.0), }, )) + inserted_claim_ids.append(claim.claim_id) elif isinstance(record, BatchRecord835): result835: ParseResult835 = record.result payer_name = result835.payer.name @@ -903,6 +921,7 @@ class CycloneStore: "amount": float(cp.total_paid or 0.0), }, )) + inserted_remit_ids.append(cp.payer_claim_control_number) else: raise TypeError( f"Unsupported BatchRecord subclass: {type(record).__name__}" @@ -922,6 +941,83 @@ class CycloneStore: "reconcile.run failed for batch %s", record.id, ) + # Publish live-tail events synchronously. EventBus.publish is async + # but its body is purely synchronous ``put_nowait`` enqueues; we + # bypass the async wrapper and call the internal enqueue directly + # so callers (sync FastAPI endpoints, sync test harnesses) don't + # need to await. + if event_bus is not None and (inserted_claim_ids or inserted_remit_ids): + self._publish_events_sync( + event_bus, record, inserted_claim_ids, inserted_remit_ids, + ) + + def _publish_events_sync( + self, + event_bus: "EventBus", + record: BatchRecord, + claim_ids: list[str], + remit_ids: list[str], + ) -> None: + """Build UI-shaped payloads for newly-inserted rows and publish. + + Runs after commit so subscribers can immediately re-fetch from + the API and see consistent data. Each ``claim_written`` / + ``remittance_written`` payload is identical to what the matching + list endpoint would return for that row. + + This is sync because EventBus's enqueue path is sync; we don't + need a coroutine for ``put_nowait``. + """ + import logging + log = logging.getLogger(__name__) + + try: + with db.SessionLocal()() as s: + for cid in claim_ids: + row = s.get(Claim, cid) + if row is None: + continue + ui = to_ui_claim_from_orm( + row, batch_id=row.batch_id or record.id, + parsed_at=record.parsed_at, + ) + self._sync_publish(event_bus, "claim_written", ui) + for rid in remit_ids: + row = s.get(Remittance, rid) + if row is None: + continue + ui = to_ui_remittance_from_orm( + row, batch_id=row.batch_id or record.id, + parsed_at=record.parsed_at, + ) + self._sync_publish(event_bus, "remittance_written", ui) + # Activity events for this batch. + from sqlalchemy import select + activity_rows = s.execute( + select(ActivityEvent).where(ActivityEvent.batch_id == record.id) + ).scalars().all() + for arow in activity_rows: + ui = { + "kind": arow.kind, + "ts": arow.ts.isoformat().replace("+00:00", "Z"), + "batchId": arow.batch_id, + "claimId": arow.claim_id, + "remittanceId": arow.remittance_id, + "payload": arow.payload_json, + } + self._sync_publish(event_bus, "activity_recorded", ui) + except Exception: + log.exception("add: event publish failed for batch %s", record.id) + + @staticmethod + def _sync_publish(event_bus: "EventBus", kind: str, payload: dict) -> None: + """Synchronous fan-out helper. Mirrors ``EventBus.publish`` but + bypasses the async wrapper so callers don't need an event loop. + """ + event = {**payload, "_kind": kind} + for queue in list(event_bus._subscribers.get(kind, ())): + event_bus._enqueue_or_drop_oldest(queue, event) + def _run_reconcile(self, batch_id: str) -> None: """T10 stub: invoke the reconcile orchestrator for a batch. diff --git a/backend/tests/test_store.py b/backend/tests/test_store.py index 86bfe91..d10d23f 100644 --- a/backend/tests/test_store.py +++ b/backend/tests/test_store.py @@ -90,6 +90,39 @@ def test_module_singleton(): assert isinstance(store, CycloneStore) +@pytest.mark.asyncio +async def test_add_837_publishes_claim_written_and_activity_recorded_events(): + """store.add() with event_bus= should publish one claim_written event + per inserted claim and one activity_recorded event per activity row.""" + from cyclone.pubsub import EventBus + + bus = EventBus() + claim_q = bus.subscribe(["claim_written"]) + activity_q = bus.subscribe(["activity_recorded"]) + + s = CycloneStore() + rec = _make_batch_record() + s.add(rec, event_bus=bus) + + # Two claims + two claim_submitted activity events expected. + import asyncio + claims = [] + for _ in range(2): + claims.append(await asyncio.wait_for(claim_q.__anext__(), timeout=0.5)) + activities = [] + for _ in range(2): + activities.append(await asyncio.wait_for(activity_q.__anext__(), timeout=0.5)) + + # Each event has _kind and a payload with the expected shape. + for ev in claims: + assert ev["_kind"] == "claim_written" + assert "id" in ev + for ev in activities: + assert ev["_kind"] == "activity_recorded" + assert ev["kind"] == "claim_submitted" + assert "claimId" in ev + + def test_add_837_persists_batch_and_claims(): s = CycloneStore() rec = _make_batch_record()