feat(sp25): add_*_ack publish ack_received / ta1_ack_received / two77ca_ack_received

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

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"