diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 96d04f8..03c6021 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -143,6 +143,7 @@ app.add_middleware( # working — Python prefers packages over same-named modules.) from cyclone.api_routers import ( # noqa: E402 acks, + activity, clearhouse, config, health, @@ -156,6 +157,13 @@ app.include_router(ta1_acks.router) app.include_router(providers.router) app.include_router(clearhouse.router) app.include_router(config.router) +app.include_router(activity.router) + +# Backwards-compat re-exports: test_api_stream_live.py imports these +# names directly from ``cyclone.api`` to invoke the endpoint coroutines +# in-process. Re-pointing at the router keeps the public surface stable +# while the bodies live in the routers package. +activity_stream = activity.activity_stream_endpoint def _resolve_payer(name: str) -> PayerConfig: @@ -1813,69 +1821,8 @@ def get_remittance(remittance_id: str) -> dict: return body -@app.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=500), -) -> 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") -async def activity_stream( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(50, ge=1, le=500), -) -> 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) diff --git a/backend/src/cyclone/api_routers/activity.py b/backend/src/cyclone/api_routers/activity.py new file mode 100644 index 0000000..6ca63fe --- /dev/null +++ b/backend/src/cyclone/api_routers/activity.py @@ -0,0 +1,97 @@ +"""``/api/activity`` — list & live-tail of recorded Activity events. + +The Activity log is the cross-resource audit feed: every parser write, +inbox state change, match/dismiss/resubmit, clearhouse submission, etc. +records one row keyed by ``kind``. The list endpoint returns the most +recent ``limit`` events with optional ``kind`` / ``since`` filters; the +stream endpoint emits a snapshot followed by live updates from the +``EventBus``. + +The default ``limit`` on the list endpoint is 200 (matches the prior +inline behavior); the stream endpoint defaults to 50 because activity is +high-volume — callers usually want the most recent handful, not a full +replay. +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/activity") +def list_activity_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(200, ge=1, le=500), +) -> Any: + """Return the recent Activity log, newest first. + + Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``). + Optional ``since`` is an inclusive ISO timestamp lower bound applied + to ``event["timestamp"]``. + """ + 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_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(50, ge=1, le=500), +) -> StreamingResponse: + """Stream Activity events as NDJSON: snapshot first, then live events. + + Subscribes to ``activity_recorded``. The snapshot half reuses the + same in-memory filter as ``list_activity`` so the two endpoints are + interchangeable for the snapshot half. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + 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")