From 8d11b391a0406c3379b28850d9a2d9edad10e52c Mon Sep 17 00:00:00 2001 From: Nora Date: Thu, 2 Jul 2026 08:47:03 -0600 Subject: [PATCH] feat(sp25): add_*_ack publish ack_received / ta1_ack_received / two77ca_ack_received --- backend/src/cyclone/store/acks.py | 78 ++++++++- backend/tests/test_store_acks_publish.py | 210 +++++++++++++++++++++++ 2 files changed, 283 insertions(+), 5 deletions(-) create mode 100644 backend/tests/test_store_acks_publish.py diff --git a/backend/src/cyclone/store/acks.py b/backend/src/cyclone/store/acks.py index ae8d6d0..e9bd81c 100644 --- a/backend/src/cyclone/store/acks.py +++ b/backend/src/cyclone/store/acks.py @@ -4,14 +4,47 @@ Each ACK type has its own ORM row (Ack / Ta1Ack / Two77caAck). The write methods are simple inserts; the list/get methods are simple queries. Fail-soft: persistence errors are logged but do not block parsing. + +SP25: every write path publishes one event on the EventBus so the +Acks page live-tail can subscribe — mirrors the existing +``claim_written`` / ``remittance_written`` / ``activity_recorded`` +publish pattern (see ``cyclone.store.write``). Publish is best-effort: +a failing subscriber MUST NOT roll back the persisted row. """ +from __future__ import annotations + +import logging from datetime import date +from typing import TYPE_CHECKING from cyclone import db from cyclone.db import Ack from . import utcnow +from .ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack +from .write import _sync_publish + +if TYPE_CHECKING: + from cyclone.pubsub import EventBus + +log = logging.getLogger(__name__) + + +def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> None: + """Best-effort publish wrapped in try/except. + + Mirrors ``CycloneStore._publish_events_sync`` — never raises, never + rolls back the persisted row. Falls back to skipping silently when + the bus doesn't implement the private ``_sync_publish`` interface + (e.g. test stubs). + """ + if event_bus is None: + return + try: + _sync_publish(event_bus, kind, payload) + except Exception: # noqa: BLE001 + log.exception("acks store: %s publish failed", kind) # -- 999 ACKs (SP3 P3 T13) ------------------------------------------- @@ -24,6 +57,7 @@ def add_999_ack( received_count: int, ack_code: str, raw_json: dict, + event_bus: "EventBus | None" = None, ) -> db.Ack: """Persist a 999 ACK row and return it. @@ -35,6 +69,11 @@ def add_999_ack( ``raw_json`` is the full ``ParseResult999`` model dump; the detail endpoint surfaces it without re-parsing the original X12 text. + + SP25: when ``event_bus`` is provided, publishes one + ``ack_received`` event (with the full ``to_ui_ack`` row shape) + after the row commits so the Acks page live-tail sees new 999s + the moment they land. """ with db.SessionLocal()() as s: row = Ack( @@ -49,7 +88,14 @@ def add_999_ack( s.add(row) s.commit() s.refresh(row) - return row + row_id = row.id + + if event_bus is not None: + with db.SessionLocal()() as s: + payload = to_ui_ack(s.get(Ack, row_id)) + _safe_publish(event_bus, "ack_received", payload) + + return row def list_acks() -> list[db.Ack]: """Return every 999 ACK row, newest first (auto-increment id desc).""" @@ -79,13 +125,17 @@ def add_ta1_ack( sender_id: str, receiver_id: str, raw_json: dict, + event_bus: "EventBus | None" = None, ) -> db.Ta1Ack: """Persist a TA1 (Interchange Acknowledgment) row and return it. - Mirrors :meth:`add_ack` for the lower-level envelope ack. The + Mirrors :meth:`add_999_ack` for the lower-level envelope ack. The flat columns are promoted out of ``raw_json`` so the list endpoint can sort/filter without a JSON parse; the full ``ParseResultTa1`` stays in ``raw_json`` for the detail endpoint. + + SP25: when ``event_bus`` is provided, publishes one + ``ta1_ack_received`` event after the row commits. """ with db.SessionLocal()() as s: row = db.Ta1Ack( @@ -104,7 +154,14 @@ def add_ta1_ack( s.add(row) s.commit() s.refresh(row) - return row + row_id = row.id + + if event_bus is not None: + with db.SessionLocal()() as s: + payload = to_ui_ta1_ack(s.get(db.Ta1Ack, row_id)) + _safe_publish(event_bus, "ta1_ack_received", payload) + + return row def list_ta1_acks() -> list[db.Ta1Ack]: """Return every TA1 ACK row, newest first (auto-increment id desc). @@ -135,12 +192,16 @@ def add_277ca_ack( paid_count: int, pended_count: int, raw_json: dict, + event_bus: "EventBus | None" = None, ) -> db.Two77caAck: """Persist a 277CA (Claim Acknowledgment) row and return it. - Mirrors :meth:`add_ack` but for the claim-level ack. The + Mirrors :meth:`add_999_ack` but for the claim-level ack. The per-claim status detail stays in ``raw_json``; only the four counts are promoted so the list endpoint stays fast. + + SP25: when ``event_bus`` is provided, publishes one + ``two77ca_ack_received`` event after the row commits. """ with db.SessionLocal()() as s: row = db.Two77caAck( @@ -156,7 +217,14 @@ def add_277ca_ack( s.add(row) s.commit() s.refresh(row) - return row + row_id = row.id + + if event_bus is not None: + with db.SessionLocal()() as s: + payload = to_ui_two77ca_ack(s.get(db.Two77caAck, row_id)) + _safe_publish(event_bus, "two77ca_ack_received", payload) + + return row def list_277ca_acks() -> list[db.Two77caAck]: """Return every 277CA ACK row, newest first (auto-increment id desc).""" diff --git a/backend/tests/test_store_acks_publish.py b/backend/tests/test_store_acks_publish.py new file mode 100644 index 0000000..1fa7023 --- /dev/null +++ b/backend/tests/test_store_acks_publish.py @@ -0,0 +1,210 @@ +"""Tests that the store's ACK add methods publish on the event bus. + +Per SP25, the store is the single write surface; the handlers and +parse endpoints no longer publish. The publish is best-effort — +a failed subscriber MUST NOT roll back the persisted row. +""" +from __future__ import annotations + +import json +from datetime import date, datetime, timezone +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.api import app +from cyclone.pubsub import EventBus +from cyclone.store import store + + +@pytest.fixture +def bus() -> EventBus: + """Fresh EventBus on the app for the duration of one test. + + The autouse conftest already wires an EventBus onto + ``app.state.event_bus`` per-test. We replace it with a fresh + one whose ``max_queue_size`` we control so we can verify the + per-kind subscription is populated. + + Pre-subscribes a queue to every ack kind so publishes land in a + queue we can drain — mirrors what the live-tail endpoints do + via ``bus.subscribe_raw(...)``. + """ + new_bus = EventBus(max_queue_size=64) + new_bus.subscribe_raw(["ack_received"]) + new_bus.subscribe_raw(["ta1_ack_received"]) + new_bus.subscribe_raw(["two77ca_ack_received"]) + app.state.event_bus = new_bus + return new_bus + + +def _drain(bus: EventBus, kind: str) -> list[dict]: + """Pull every queued event for ``kind`` off the bus and return them. + + Uses ``get_nowait`` against the private ``_subscribers`` map so + we can verify the publish happened without async machinery. + """ + events: list[dict] = [] + for queue in bus._subscribers.get(kind, []): + while not queue.empty(): + events.append(queue.get_nowait()) + return events + + +# --------------------------------------------------------------------------- +# 999 ACK publish +# --------------------------------------------------------------------------- + + +def test_add_999_ack_publishes_ack_received(bus: EventBus): + """add_999_ack must publish one ``ack_received`` event after commit.""" + with db.SessionLocal()() as s: + row = store.add_ack( + source_batch_id="999-SP25-1", + accepted_count=2, + rejected_count=0, + received_count=2, + ack_code="A", + raw_json={ + "envelope": {"control_number": "000000001"}, + "set_responses": [{"set_control_number": "PCN-1"}], + }, + event_bus=bus, + ) + + events = _drain(bus, "ack_received") + assert len(events) == 1 + payload = events[0] + # The event payload MUST match the list-endpoint shape so a + # live-tail consumer sees the same row shape as a JSON fetch. + assert payload["_kind"] == "ack_received" + assert payload["id"] == row.id + assert payload["source_batch_id"] == "999-SP25-1" + assert payload["accepted_count"] == 2 + assert payload["rejected_count"] == 0 + assert payload["received_count"] == 2 + assert payload["ack_code"] == "A" + # Patient control number surfaces through the UI serializer. + assert payload["patient_control_number"] == "PCN-1" + + +def test_add_999_ack_publish_failure_does_not_rollback(): + """Best-effort publish: a throwing subscriber MUST NOT roll back the row. + + The store wraps ``_sync_publish`` in a try/except so a misbehaving + subscriber can never break the persisted write. + """ + boom_bus = EventBus(max_queue_size=64) + + class _BoomQueue: + def put_nowait(self, _event) -> None: + raise RuntimeError("subscriber exploded") + + boom_bus._subscribers["ack_received"] = [_BoomQueue()] # type: ignore[list-item] + + with db.SessionLocal()() as s: + row = store.add_ack( + source_batch_id="999-SP25-BOOM", + accepted_count=1, + rejected_count=0, + received_count=1, + ack_code="A", + raw_json={}, + event_bus=boom_bus, + ) + + # Row still landed despite the publish failure. + assert store.get_ack(row.id) is not None + + +# --------------------------------------------------------------------------- +# TA1 ACK publish (Task 3 stub — added here so the test module collects as +# one unit; the heavy lifting lives in Task 3 / 4). +# --------------------------------------------------------------------------- + + +def test_add_ta1_ack_publishes_ta1_ack_received(bus: EventBus): + """add_ta1_ack must publish one ``ta1_ack_received`` event after commit.""" + with db.SessionLocal()() as s: + row = store.add_ta1_ack( + source_batch_id="TA1-SP25-1", + control_number="000000001", + interchange_date=date(2026, 7, 2), + interchange_time="1200", + ack_code="A", + note_code="000", + ack_generated_date=None, + sender_id="SENDER", + receiver_id="RECEIVER", + raw_json={"envelope": {"control_number": "000000001"}}, + event_bus=bus, + ) + + events = _drain(bus, "ta1_ack_received") + assert len(events) == 1 + payload = events[0] + assert payload["_kind"] == "ta1_ack_received" + assert payload["id"] == row.id + assert payload["control_number"] == "000000001" + assert payload["ack_code"] == "A" + assert payload["sender_id"] == "SENDER" + assert payload["receiver_id"] == "RECEIVER" + assert payload["interchange_date"] == "2026-07-02" + + +# --------------------------------------------------------------------------- +# 277CA ACK publish (Task 4 stub) +# --------------------------------------------------------------------------- + + +def test_add_277ca_ack_publishes_two77ca_ack_received(bus: EventBus): + """add_277ca_ack must publish one ``two77ca_ack_received`` event.""" + with db.SessionLocal()() as s: + row = store.add_277ca_ack( + source_batch_id="277CA-SP25-1", + control_number="000000001", + accepted_count=2, + rejected_count=1, + paid_count=0, + pended_count=0, + raw_json={"envelope": {"control_number": "000000001"}, "claim_statuses": []}, + event_bus=bus, + ) + + events = _drain(bus, "two77ca_ack_received") + assert len(events) == 1 + payload = events[0] + assert payload["_kind"] == "two77ca_ack_received" + assert payload["id"] == row.id + assert payload["control_number"] == "000000001" + assert payload["accepted_count"] == 2 + assert payload["rejected_count"] == 1 + + +# --------------------------------------------------------------------------- +# /api/parse-999 fires the event (Task 6 stub) +# --------------------------------------------------------------------------- + + +MINIMAL_999 = Path(__file__).parent / "fixtures" / "minimal_999.txt" + + +def test_parse_999_endpoint_fires_ack_received_event(bus: EventBus): + """The /api/parse-999 route threads the app.state event bus into add_999_ack. + + SP25: the parse endpoint no longer hand-rolls a publish. The store + owns publish-from-store; the endpoint just threads the bus. + """ + client = TestClient(app) + text = MINIMAL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999.txt", text, "text/plain")}, + ) + assert resp.status_code == 200, resp.text + + events = _drain(bus, "ack_received") + assert len(events) == 1 + assert events[0]["_kind"] == "ack_received"