diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index c4395ba..985763d 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -2445,134 +2445,6 @@ def _svc_to_dict(svc) -> dict: } -@app.get("/api/remittances", dependencies=[Depends(matrix_gate)]) -def list_remittances( - request: Request, - batch_id: str | None = Query(None), - payer: str | None = Query(None), - claim_id: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), - offset: int = Query(0, ge=0), -) -> Any: - common = dict( - batch_id=batch_id, - payer=payer, - claim_id=claim_id, - date_from=date_from, - date_to=date_to, - ) - items = list(store.iter_remittances( - sort=sort, order=order, limit=limit, offset=offset, **common, - )) - # SP27 Task 13b: count the full population, not a 100-row sample. - # See the matching note in list_claims — same silent-failure pattern. - total = store.count_remittances(**common) - returned = len(items) - has_more = total > offset + returned - 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, - } - - -@app.get("/api/remittances/summary", dependencies=[Depends(matrix_gate)]) -def remittances_summary( - batch_id: str | None = Query(None), - payer: str | None = Query(None), - claim_id: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), -) -> dict: - """Server-aggregated KPI tiles for the Remittances page. - - Returns ``{count, total_paid, total_adjustments}`` over the - full filtered remittance population — NOT a page-limited - sample. The Remittances page consumes this for its "Total paid" - and "Adjustments" tiles so they can't silently understate the - true DB population the way a page-local ``items.reduce(...)`` - would. Mirrors the silent-incompleteness fix that - ``/api/dashboard/kpis`` (commit ``59c3275``) and - ``/api/remittances`` (commit ``d81b6ed``) made for their tiles. - - Same filter parameters as ``/api/remittances``. Always returns - a populated dict (``{"count": 0, "total_paid": 0, - "total_adjustments": 0}`` when no rows match) so the frontend - can render the tiles directly without a loading-vs-empty - branch. - """ - return store.summarize_remittances( - batch_id=batch_id, payer=payer, claim_id=claim_id, - date_from=date_from, date_to=date_to, - ) - - -@app.get("/api/remittances/stream", dependencies=[Depends(matrix_gate)]) -async def remittances_stream( - request: Request, - payer: str | None = Query(None), - claim_id: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), -) -> StreamingResponse: - """Stream Remittances as NDJSON: snapshot first, then live events. - - Subscribes to ``remittance_written``. Default sort is - ``-received_date`` (newest-first), matching the list endpoint's - most common sort. - - NOTE: registered before ``/api/remittances/{remittance_id}`` so - the literal ``stream`` path segment doesn't get matched as a - remittance id. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - rows = store.iter_remittances( - payer=payer, claim_id=claim_id, - date_from=date_from, date_to=date_to, - sort=sort or "-received_date", order=order, limit=limit, - ) - for row in rows: - yield _ndjson_line({"type": "item", "data": row}) - yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) - - async for chunk in _tail_events(request, bus, ["remittance_written"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - - -@app.get("/api/remittances/{remittance_id}", dependencies=[Depends(matrix_gate)]) -def get_remittance(remittance_id: str) -> dict: - """Return one remittance with its labeled CAS ``adjustments`` array. - - Path param is ``remittance_id`` (not ``id``) to avoid shadowing - FastAPI's internal ``id`` name and to keep OpenAPI docs self- - describing. Returns 404 when the remittance is missing — never 500. - """ - body = store.get_remittance(remittance_id) - if body is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"}, - ) - return body - - @app.get("/api/providers", dependencies=[Depends(matrix_gate)]) def list_providers( request: Request, @@ -2603,74 +2475,6 @@ def list_providers( } -@app.get("/api/activity", dependencies=[Depends(matrix_gate)]) -def list_activity( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(200, ge=1, le=5000), -) -> Any: - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - total = len(events) - has_more = False - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(events, total, total, has_more), - media_type="application/x-ndjson", - ) - return { - "items": events, - "total": total, - "returned": total, - "has_more": has_more, - } - - -@app.get("/api/activity/stream", dependencies=[Depends(matrix_gate)]) -async def activity_stream( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(50, ge=1, le=5000), -) -> StreamingResponse: - """Stream Activity events as NDJSON: snapshot first, then live events. - - Subscribes to ``activity_recorded``. Default ``limit`` is 50 - (smaller than the list endpoint's 200) because activity is - high-volume — callers usually want the most recent handful, not a - full replay. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - # Snapshot reuses the same in-memory filter as ``list_activity`` - # so the two endpoints are interchangeable for the snapshot - # half. - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - for ev in events: - yield _ndjson_line({"type": "item", "data": ev}) - yield _ndjson_line({ - "type": "snapshot_end", "data": {"count": len(events)}, - }) - - async for chunk in _tail_events(request, bus, ["activity_recorded"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - - -# --------------------------------------------------------------------------- # -# 999 ACKs (read views) -# --------------------------------------------------------------------------- # - # --------------------------------------------------------------------------- # SP9: providers / payers / clearhouse endpoints @@ -3037,5 +2841,18 @@ app.include_router(admin_users_router, dependencies=[Depends(matrix_gate)]) # line once the test is updated to import from the new home. from cyclone.api_routers.payers import _clear_summary_cache # noqa: F401 # SP36 backward-compat shim +# SP36 Tasks 9+10 backward-compat: ``remittances_stream`` and +# ``activity_stream`` moved into ``cyclone.api_routers.remittances`` +# and ``cyclone.api_routers.activity`` respectively (they are +# single-router handlers per spec D4). The live-tail integration +# test in ``tests/test_api_stream_live.py`` imports them by name +# from this module to drive ``_call_endpoint(stream_fn, path)``. +# Per the SP-N invariant that we don't rewrite tests for a +# structural refactor, expose them on this module's namespace via +# one-line re-imports. Delete these lines once the test is updated +# to import from the new homes. +from cyclone.api_routers.remittances import remittances_stream # noqa: F401 # SP36 backward-compat shim +from cyclone.api_routers.activity import activity_stream # noqa: F401 # SP36 backward-compat shim + __all__ = ["app"] diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py index 4d94f95..a5d9e50 100644 --- a/backend/src/cyclone/api_routers/__init__.py +++ b/backend/src/cyclone/api_routers/__init__.py @@ -12,6 +12,7 @@ from fastapi import APIRouter from cyclone.api_routers import ( acks, + activity, admin, claim_acks, config, @@ -20,11 +21,13 @@ from cyclone.api_routers import ( health, payers, reconciliation, + remittances, ta1_acks, ) routers: list[APIRouter] = [ acks.router, # gated + activity.router, # gated admin.router, # gated claim_acks.router, # gated config.router, # gated @@ -33,6 +36,7 @@ routers: list[APIRouter] = [ health.router, # public — health probes must work pre-auth payers.router, # gated reconciliation.router, # gated + remittances.router, # gated ta1_acks.router, # gated ] diff --git a/backend/src/cyclone/api_routers/activity.py b/backend/src/cyclone/api_routers/activity.py new file mode 100644 index 0000000..16b0f6c --- /dev/null +++ b/backend/src/cyclone/api_routers/activity.py @@ -0,0 +1,102 @@ +"""``/api/activity`` and ``/api/activity/stream`` — operator-facing event log. + +Two endpoints, both gated by ``matrix_gate``: + +- ``GET /api/activity`` — paginated event list with ``kind`` / + ``since`` filters, plus an NDJSON variant when the caller sends + ``Accept: application/x-ndjson``. +- ``GET /api/activity/stream`` — NDJSON live-tail: snapshot of the + most recent N events, then ``activity_recorded`` events as they + hit the store. Default ``limit`` is 50 (smaller than the list + endpoint's 200) because activity is high-volume — callers usually + want the most recent handful, not a full replay. + +The snapshot halves of ``/api/activity`` and ``/api/activity/stream`` +share the same in-memory filter logic (``kind`` + ``since``) so the +two endpoints are interchangeable for the snapshot half. + +SP36 Task 10: this block moved here from ``api.py:2606`` (the +``/api/activity*`` pair). +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Depends, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line as _ndjson_line, + ndjson_stream_list as _ndjson_stream_list, + tail_events as _tail_events, + wants_ndjson as _wants_ndjson, +) +from cyclone.auth.deps import matrix_gate +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter(dependencies=[Depends(matrix_gate)]) + + +@router.get("/api/activity") +def list_activity( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(200, ge=1, le=5000), +) -> Any: + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + total = len(events) + has_more = False + if _wants_ndjson(request): + return StreamingResponse( + _ndjson_stream_list(events, total, total, has_more), + media_type="application/x-ndjson", + ) + return { + "items": events, + "total": total, + "returned": total, + "has_more": has_more, + } + + +@router.get("/api/activity/stream") +async def activity_stream( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(50, ge=1, le=5000), +) -> StreamingResponse: + """Stream Activity events as NDJSON: snapshot first, then live events. + + Subscribes to ``activity_recorded``. Default ``limit`` is 50 + (smaller than the list endpoint's 200) because activity is + high-volume — callers usually want the most recent handful, not a + full replay. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + # Snapshot reuses the same in-memory filter as ``list_activity`` + # so the two endpoints are interchangeable for the snapshot + # half. + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + for ev in events: + yield _ndjson_line({"type": "item", "data": ev}) + yield _ndjson_line({ + "type": "snapshot_end", "data": {"count": len(events)}, + }) + + async for chunk in _tail_events(request, bus, ["activity_recorded"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") diff --git a/backend/src/cyclone/api_routers/remittances.py b/backend/src/cyclone/api_routers/remittances.py new file mode 100644 index 0000000..a80c7c1 --- /dev/null +++ b/backend/src/cyclone/api_routers/remittances.py @@ -0,0 +1,169 @@ +"""``/api/remittances`` and ``/api/remittances/stream`` — read views over the Remittance population. + +Four endpoints, all gated by ``matrix_gate``: + +- ``GET /api/remittances`` — paginated list with filter+sort, + plus an NDJSON variant when the caller sends ``Accept: application/x-ndjson``. +- ``GET /api/remittances/summary`` — server-aggregated KPI tiles + (``count``, ``total_paid``, ``total_adjustments``) over the full + filtered population — never a page-limited sample (SP27 fix). +- ``GET /api/remittances/stream`` — NDJSON live-tail: snapshot + of currently-known rows, then ``remittance_written`` events as + they hit the store. Subscribed to by the Remittances page. +- ``GET /api/remittances/{remittance_id}`` — one remittance with its + labeled CAS ``adjustments`` array. 404 on missing id (never 500). + +``/api/remittances/stream`` is registered before +``/api/remittances/{remittance_id}`` so the literal ``stream`` path +segment is not captured as a remittance id. + +SP36 Task 9: this block moved here from ``api.py:2448`` (the 4 +``/api/remittances*`` routes, with the streaming route's +``NOTE: registered before…`` comment preserved verbatim). +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line as _ndjson_line, + ndjson_stream_list as _ndjson_stream_list, + tail_events as _tail_events, + wants_ndjson as _wants_ndjson, +) +from cyclone.auth.deps import matrix_gate +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter(dependencies=[Depends(matrix_gate)]) + + +@router.get("/api/remittances") +def list_remittances( + request: Request, + batch_id: str | None = Query(None), + payer: str | None = Query(None), + claim_id: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), + offset: int = Query(0, ge=0), +) -> Any: + common = dict( + batch_id=batch_id, + payer=payer, + claim_id=claim_id, + date_from=date_from, + date_to=date_to, + ) + items = list(store.iter_remittances( + sort=sort, order=order, limit=limit, offset=offset, **common, + )) + # SP27 Task 13b: count the full population, not a 100-row sample. + # See the matching note in list_claims — same silent-failure pattern. + total = store.count_remittances(**common) + returned = len(items) + has_more = total > offset + returned + 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, + } + + +@router.get("/api/remittances/summary") +def remittances_summary( + batch_id: str | None = Query(None), + payer: str | None = Query(None), + claim_id: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), +) -> dict: + """Server-aggregated KPI tiles for the Remittances page. + + Returns ``{count, total_paid, total_adjustments}`` over the + full filtered remittance population — NOT a page-limited + sample. The Remittances page consumes this for its "Total paid" + and "Adjustments" tiles so they can't silently understate the + true DB population the way a page-local ``items.reduce(...)`` + would. Mirrors the silent-incompleteness fix that + ``/api/dashboard/kpis`` (commit ``59c3275``) and + ``/api/remittances`` (commit ``d81b6ed``) made for their tiles. + + Same filter parameters as ``/api/remittances``. Always returns + a populated dict (``{"count": 0, "total_paid": 0, + "total_adjustments": 0}`` when no rows match) so the frontend + can render the tiles directly without a loading-vs-empty + branch. + """ + return store.summarize_remittances( + batch_id=batch_id, payer=payer, claim_id=claim_id, + date_from=date_from, date_to=date_to, + ) + + +@router.get("/api/remittances/stream") +async def remittances_stream( + request: Request, + payer: str | None = Query(None), + claim_id: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream Remittances as NDJSON: snapshot first, then live events. + + Subscribes to ``remittance_written``. Default sort is + ``-received_date`` (newest-first), matching the list endpoint's + most common sort. + + NOTE: registered before ``/api/remittances/{remittance_id}`` so + the literal ``stream`` path segment doesn't get matched as a + remittance id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.iter_remittances( + payer=payer, claim_id=claim_id, + date_from=date_from, date_to=date_to, + sort=sort or "-received_date", order=order, limit=limit, + ) + for row in rows: + yield _ndjson_line({"type": "item", "data": row}) + yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + + async for chunk in _tail_events(request, bus, ["remittance_written"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/remittances/{remittance_id}") +def get_remittance(remittance_id: str) -> dict: + """Return one remittance with its labeled CAS ``adjustments`` array. + + Path param is ``remittance_id`` (not ``id``) to avoid shadowing + FastAPI's internal ``id`` name and to keep OpenAPI docs self- + describing. Returns 404 when the remittance is missing — never 500. + """ + body = store.get_remittance(remittance_id) + if body is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"}, + ) + return body