diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py index 19a6636..c1d5c04 100644 --- a/backend/src/cyclone/api_routers/acks.py +++ b/backend/src/cyclone/api_routers/acks.py @@ -1,4 +1,4 @@ -"""``/api/acks`` — list & detail endpoints for the 999 ACK inbox. +"""``/api/acks`` — list, detail, and live-tail stream for 999 ACKs. These are the persisted acknowledgment rows produced by ``POST /api/parse-999``. The frontend ``useAcks`` hook re-shapes the @@ -7,18 +7,29 @@ list payload to its ``Ack`` interface in ``src/types/index.ts``. The detail endpoint returns the full ``raw_json`` payload plus the regenerated ``raw_999_text`` so the UI can show "view source" without a second round-trip. + +SP25: ``/api/acks/stream`` joins the live-tail triplet — the Acks +page mounts ``useTailStream("acks")`` and ``useMergedTail("acks", …)`` +to see new 999 acks the moment they land (whether from the SFTP +poller or a manual upload). """ from __future__ import annotations import logging -from typing import Any +from typing import Any, AsyncIterator from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse -from cyclone.api_helpers import ndjson_stream_list, wants_ndjson +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) from cyclone.parsers.models_999 import ParseResult999 from cyclone.parsers.serialize_999 import serialize_999 +from cyclone.pubsub import EventBus from cyclone.store import store, to_ui_ack router = APIRouter() @@ -71,6 +82,35 @@ def list_acks_endpoint( } +@router.get("/api/acks/stream") +async def acks_stream( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream 999 ACKs as NDJSON: snapshot first, then live events. + + SP25: this endpoint joins the live-tail triplet — subscribes to + ``ack_received`` and emits one ``item`` per snapshot row plus a + single ``snapshot_end`` line, then forwards live events from + the bus. Matches the wire format used by ``/api/claims/stream``, + ``/api/remittances/stream``, and ``/api/activity/stream``. + + NOTE: registered BEFORE ``/api/acks/{ack_id}`` so the literal + ``stream`` path segment doesn't get matched as an ack id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.list_acks()[:limit] + for row in rows: + yield ndjson_line({"type": "item", "data": to_ui_ack(row)}) + yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + async for chunk in tail_events(request, bus, ["ack_received"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + @router.get("/api/acks/{ack_id}") def get_ack_endpoint(ack_id: int) -> dict: """Return one persisted ACK row with its parsed detail. diff --git a/backend/src/cyclone/api_routers/ta1_acks.py b/backend/src/cyclone/api_routers/ta1_acks.py index fb81b53..b95bc20 100644 --- a/backend/src/cyclone/api_routers/ta1_acks.py +++ b/backend/src/cyclone/api_routers/ta1_acks.py @@ -1,4 +1,4 @@ -"""``/api/ta1-acks`` — list & detail endpoints for persisted TA1 envelopes. +"""``/api/ta1-acks`` — list, detail, and live-tail stream for TA1 envelopes. TA1 is the interchange-control ACK (ISA/IEA acknowledgement). It's a single segment, no functional group, no transaction set. Cyclone @@ -7,14 +7,21 @@ row can sit alongside the 999 / 277CA ack rows without special-casing. The detail endpoint also reconstructs the TA1 segment string (``TA1*...~``) so the operator can copy it into a downstream tool. + +SP25: ``/api/ta1-acks/stream`` joins the live-tail triplet — the +Acks page mounts ``useTailStream("ta1_acks")`` so the TA1 envelope +ack section sees new rows the moment they land. """ from __future__ import annotations -from typing import Any +from typing import Any, AsyncIterator -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import StreamingResponse +from cyclone.api_helpers import ndjson_line, tail_events from cyclone import db +from cyclone.pubsub import EventBus from cyclone.store import store, to_ui_ta1_ack router = APIRouter() @@ -52,6 +59,30 @@ def list_ta1_acks_endpoint( } +@router.get("/api/ta1-acks/stream") +async def ta1_acks_stream( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream TA1 envelope acks as NDJSON. + + Subscribes to ``ta1_ack_received`` and emits the same wire format + as the other live-tail endpoints. Registered BEFORE + ``/api/ta1-acks/{ack_id}`` so ``stream`` isn't matched as an id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.list_ta1_acks()[:limit] + for row in rows: + yield ndjson_line({"type": "item", "data": to_ui_ta1_ack(row)}) + yield ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + async for chunk in tail_events(request, bus, ["ta1_ack_received"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + @router.get("/api/ta1-acks/{ack_id}") def get_ta1_ack_endpoint(ack_id: int) -> dict: """Return one persisted TA1 ACK row with its parsed detail.""" diff --git a/backend/tests/test_api_ack_stream.py b/backend/tests/test_api_ack_stream.py new file mode 100644 index 0000000..68bf84d --- /dev/null +++ b/backend/tests/test_api_ack_stream.py @@ -0,0 +1,303 @@ +"""Integration tests for /api/acks/stream and /api/ta1-acks/stream. + +SP25: both endpoints follow the live-tail wire format used by +/api/claims/stream, /api/remittances/stream, /api/activity/stream: + + * Snapshot: one ``item`` line per existing row (newest first). + * Closing: one ``snapshot_end`` line. + * Live: forwarded ``item`` lines for each ``ack_received`` / + ``ta1_ack_received`` event published on the bus. + +These tests follow the same direct-coroutine pattern as +``test_api_stream_live.py`` because ``httpx.ASGITransport`` buffers +the response and never delivers a disconnect message — calling the +endpoint coroutine with a synthetic ``Request`` lets us iterate the +body iterator byte-by-byte and use ``body_iterator.aclose()`` to +simulate a client disconnect. +""" +from __future__ import annotations + +import asyncio +import json +from datetime import date + +import pytest +from starlette.requests import Request + +from cyclone import db +from cyclone.api import app +from cyclone.api_routers.acks import acks_stream +from cyclone.api_routers.ta1_acks import ta1_acks_stream +from cyclone.pubsub import EventBus +from cyclone.store import store + + +# --------------------------------------------------------------------------- +# Shared helpers — mirror the patterns in test_api_stream_live.py +# --------------------------------------------------------------------------- + + +def _make_request(path: str = "/api/acks/stream") -> Request: + """Synthetic Starlette ``Request`` with a no-op receive channel.""" + + async def receive(): + # Block forever; ``is_disconnected``'s cancel scope will cancel + # this await and we'll return None (treated as "no message"). + await asyncio.Event().wait() + + scope = { + "type": "http", + "method": "GET", + "headers": [], + "query_string": b"", + "path": path, + "app": app, + "scheme": "http", + "server": ("testserver", 80), + "client": ("testclient", 50000), + } + return Request(scope, receive=receive) + + +def _resolve_query_defaults(endpoint) -> dict: + """Resolve FastAPI Query(...) defaults to their inner ``.default``.""" + import inspect + sig = inspect.signature(endpoint) + resolved: dict = {} + for name, param in sig.parameters.items(): + default = param.default + if hasattr(default, "default"): + resolved[name] = default.default + else: + resolved[name] = default + return resolved + + +async def _call_endpoint(endpoint, path: str): + request = _make_request(path) + defaults = _resolve_query_defaults(endpoint) + defaults.pop("request", None) + return await endpoint(request, **defaults) + + +async def _read_lines(body_iter, n: int, timeout: float = 5.0) -> list[str]: + buffer = b"" + lines: list[str] = [] + async with asyncio.timeout(timeout): + while len(lines) < n: + try: + chunk = await body_iter.__anext__() + except StopAsyncIteration: + break + buffer += chunk + while b"\n" in buffer and len(lines) < n: + raw, buffer = buffer.split(b"\n", 1) + if raw: + lines.append(raw.decode("utf-8")) + return lines + + +async def _read_one_line(body_iter, timeout: float = 5.0) -> str | None: + buffer = b"" + async with asyncio.timeout(timeout): + while True: + try: + chunk = await body_iter.__anext__() + except StopAsyncIteration: + return None + buffer += chunk + if b"\n" in buffer: + raw, _ = buffer.split(b"\n", 1) + if raw: + return raw.decode("utf-8") + + +async def _read_until_type( + body_iter, target: str, *, timeout: float = 5.0, max_lines: int = 50 +) -> dict | None: + buffer = b"" + seen = 0 + async with asyncio.timeout(timeout): + while seen < max_lines: + try: + chunk = await body_iter.__anext__() + except StopAsyncIteration: + return None + buffer += chunk + while b"\n" in buffer: + raw, buffer = buffer.split(b"\n", 1) + if not raw: + continue + seen += 1 + obj = json.loads(raw.decode("utf-8")) + if obj.get("type") == target: + return obj + return None + + +async def _drain_until_disconnect(body_iter, max_chunks: int = 20) -> None: + try: + await body_iter.aclose() + except (asyncio.CancelledError, GeneratorExit): + pass + + +# --------------------------------------------------------------------------- +# Per-test heartbeat patch — 0.2s so idle generators exit promptly. +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _short_heartbeat(monkeypatch): + monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2") + + +@pytest.fixture +def bus() -> EventBus: + """Fresh EventBus attached to app.state. Tests publish via ``bus.publish``. + + ``add_ack`` / ``add_ta1_ack`` already publish to whichever bus is + passed in, but the live-event tests want to publish synchronously + against ``app.state.event_bus`` so the existing-tail subscription + picks them up. + """ + new_bus = EventBus(max_queue_size=64) + app.state.event_bus = new_bus + return new_bus + + +# --------------------------------------------------------------------------- +# /api/acks/stream +# --------------------------------------------------------------------------- + + +async def test_acks_stream_snapshots_existing_rows(bus: EventBus): + """An existing 999 row surfaces as the first ``item`` line.""" + with db.SessionLocal()() as s: + store.add_ack( + source_batch_id="999-SEED-1", + accepted_count=1, + rejected_count=0, + received_count=1, + ack_code="A", + raw_json={ + "envelope": {"control_number": "000000001"}, + "set_responses": [{"set_control_number": "PCN-1"}], + }, + event_bus=bus, + ) + + response = await _call_endpoint(acks_stream, "/api/acks/stream") + assert response.media_type.startswith("application/x-ndjson") + + lines = await _read_lines(response.body_iterator, n=2) + items = [json.loads(line) for line in lines] + assert items[0]["type"] == "item" + assert items[0]["data"]["source_batch_id"] == "999-SEED-1" + assert items[0]["data"]["patient_control_number"] == "PCN-1" + assert items[1] == {"type": "snapshot_end", "data": {"count": 1}} + await _drain_until_disconnect(response.body_iterator) + + +async def test_acks_stream_emits_live_event(bus: EventBus): + """A 999 written while the stream is open shows up as a live item.""" + response = await _call_endpoint(acks_stream, "/api/acks/stream") + + # Drain snapshot (1 snapshot_end line, no prior rows). + snap = await _read_until_type( + response.body_iterator, "snapshot_end", timeout=2.0, + ) + assert snap == {"type": "snapshot_end", "data": {"count": 0}} + + # Advance one step so ``subscribe_raw`` registers. Next line may be + # a heartbeat — that's fine, we just need the subscription live. + _ = await _read_one_line(response.body_iterator, timeout=2.0) + + with db.SessionLocal()() as s: + store.add_ack( + source_batch_id="999-LIVE-1", + accepted_count=1, + rejected_count=0, + received_count=1, + ack_code="A", + raw_json={ + "envelope": {"control_number": "000000002"}, + "set_responses": [], + }, + event_bus=bus, + ) + + obj = await _read_until_type( + response.body_iterator, "item", timeout=2.0, + ) + assert obj is not None, "no item line arrived after publish" + assert obj["data"]["source_batch_id"] == "999-LIVE-1" + await _drain_until_disconnect(response.body_iterator) + + +# --------------------------------------------------------------------------- +# /api/ta1-acks/stream +# --------------------------------------------------------------------------- + + +async def test_ta1_acks_stream_snapshots_existing_rows(bus: EventBus): + """An existing TA1 row surfaces as the first ``item`` line.""" + with db.SessionLocal()() as s: + store.add_ta1_ack( + source_batch_id="TA1-SEED-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="S", + receiver_id="R", + raw_json={"envelope": {"control_number": "000000001"}}, + event_bus=bus, + ) + + response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream") + assert response.media_type.startswith("application/x-ndjson") + + lines = await _read_lines(response.body_iterator, n=2) + items = [json.loads(line) for line in lines] + assert items[0]["type"] == "item" + assert items[0]["data"]["control_number"] == "000000001" + assert items[0]["data"]["interchange_date"] == "2026-07-02" + assert items[1] == {"type": "snapshot_end", "data": {"count": 1}} + await _drain_until_disconnect(response.body_iterator) + + +async def test_ta1_acks_stream_emits_live_event(bus: EventBus): + """A TA1 written while the stream is open shows up as a live item.""" + response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream") + + snap = await _read_until_type( + response.body_iterator, "snapshot_end", timeout=2.0, + ) + assert snap == {"type": "snapshot_end", "data": {"count": 0}} + _ = await _read_one_line(response.body_iterator, timeout=2.0) + + with db.SessionLocal()() as s: + store.add_ta1_ack( + source_batch_id="TA1-LIVE-1", + control_number="000000099", + interchange_date=date(2026, 7, 2), + interchange_time="1330", + ack_code="A", + note_code="000", + ack_generated_date=None, + sender_id="SX", + receiver_id="RX", + raw_json={"envelope": {"control_number": "000000099"}}, + event_bus=bus, + ) + + obj = await _read_until_type( + response.body_iterator, "item", timeout=2.0, + ) + assert obj is not None, "no item line arrived after publish" + assert obj["data"]["control_number"] == "000000099" + assert obj["data"]["sender_id"] == "SX" + await _drain_until_disconnect(response.body_iterator) \ No newline at end of file