"""``/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 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, AsyncIterator from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse from cyclone import db 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() log = logging.getLogger(__name__) # SP25: ``_ack_to_ui`` moved to ``cyclone.store.ui.to_ui_ack`` so the # live-tail event payload (``ack_received``) matches the list endpoint # shape byte-for-byte. @router.get("/api/acks") def list_acks_endpoint( request: Request, limit: int = Query(100, ge=1, le=5000), offset: int = Query(0, ge=0), ) -> Any: """Return the list of persisted 999 ACKs, newest first. ``limit`` caps the page size; ``offset`` lets the UI walk the full set without holding it all in memory. ``aggregates`` is summed over the *full* row set (not the page) so the KPI strip on the Acks page reflects every persisted 999, not just the visible 50. Without server-side aggregates the page would silently under-report (silent-failure mode) once the row count exceeds the page size. """ rows = store.list_acks() items = [to_ui_ack(r) for r in rows[offset : offset + limit]] total = len(rows) returned = len(items) has_more = offset + returned < total aggregates = { "accepted_count": sum(r.accepted_count or 0 for r in rows), "rejected_count": sum(r.rejected_count or 0 for r in rows), "received_count": sum(r.received_count or 0 for r in rows), } # SP28: batch-fetch linked_claim_ids per ack row in one query to # avoid N+1 — see ``find_linked_claim_ids_for_acks`` below. ack_ids = [r.id for r in rows] linked_map = _find_linked_claim_ids_for_acks(ack_ids, kind="999") for item, aid in zip(items, ack_ids[offset : offset + limit]): item["linked_claim_ids"] = linked_map.get(aid, []) if wants_ndjson(request): return StreamingResponse( ndjson_stream_list(items, total, returned, has_more), media_type="application/x-ndjson", ) return { "items": items, "total": total, "returned": returned, "has_more": has_more, "aggregates": aggregates, } def _find_linked_claim_ids_for_acks( ack_ids: list[int], *, kind: str ) -> dict[int, list[str]]: """Batch-fetch {ack_id: [claim_id, …]} for the listed ack rows. One SELECT against ``claim_acks`` keyed on the page's ack_ids — avoids an N+1 round-trip when the page renders the "🔗 N claims" badge per row. Used by the 999 / TA1 / 277CA list endpoints. Returns a ``{ack_id: [claim_id]}`` map; acks with no links map to ``[]`` (default). """ out: dict[int, list[str]] = {aid: [] for aid in ack_ids} if not ack_ids: return out with db.SessionLocal()() as s: rows = ( s.query(db.ClaimAck.ack_id, db.ClaimAck.claim_id) .filter( db.ClaimAck.ack_kind == kind, db.ClaimAck.ack_id.in_(ack_ids), db.ClaimAck.claim_id.isnot(None), ) .all() ) for ack_id, claim_id in rows: out[ack_id].append(claim_id) return out @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. Path param is ``ack_id`` (not ``id``) to avoid shadowing FastAPI's internal ``id`` name and to keep OpenAPI docs self-describing. Returns 404 when the ACK is missing — never 500. """ row = store.get_ack(ack_id) if row is None: raise HTTPException( status_code=404, detail={"error": "Not found", "detail": f"Ack {ack_id} not found"}, ) body = to_ui_ack(row) body["raw_json"] = row.raw_json # Regenerate the X12 text from raw_json so the operator can download # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry # the regenerated text to keep payloads small; detail does.) if row.raw_json: try: regenerated = ParseResult999.model_validate(row.raw_json) icn = regenerated.envelope.control_number or "000000001" body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn) except Exception as exc: # noqa: BLE001 — never 500 on a regen failure log.warning("Could not regenerate 999 for ack %s: %s", ack_id, exc) body["raw_999_text"] = None else: body["raw_999_text"] = None return body