diff --git a/api/app.py b/api/app.py index 4b6464d..c245926 100644 --- a/api/app.py +++ b/api/app.py @@ -39,6 +39,7 @@ from fastapi import FastAPI from api.config import ApiConfig from api.database import DatabaseState, make_database +from api.events import TransitionPublisher from api.routes_panel import router as panel_router @@ -47,12 +48,16 @@ def _build_lifespan(db_path: str): Captured as a closure so the app factory can hand the configured path in without threading it through FastAPI's lifespan signature. + The lifespan also installs the app's ONE in-process transition + publisher on ``app.state.publisher`` (see :mod:`api.events`) so + every route and SSE subscriber shares that single instance. """ @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: state = make_database(db_path) app.state.db = state + app.state.publisher = TransitionPublisher() try: await state.open_and_migrate() yield diff --git a/api/events.py b/api/events.py new file mode 100644 index 0000000..f5b7830 --- /dev/null +++ b/api/events.py @@ -0,0 +1,78 @@ +"""The ONE in-process transition-event publisher for the API service. + +Route handlers call :meth:`TransitionPublisher.publish` synchronously +(no ``await``, callable from any async handler); SSE subscribers stream +those transitions through an async generator. The scope of "ONE" is +deliberate: a single :class:`TransitionPublisher` instance is created +in the app lifespan and stored on ``app.state.publisher`` — per-app, +so tests build their own hermetic app without module-level state, and +both the panel complete route and the SSE endpoint resolve the SAME +instance off ``request.app.state`` (exactly one publisher exists per +running app; nothing else may create one). + +Design (small and dependency-free, asyncio only): + +- Each subscriber gets its own ``asyncio.Queue(maxsize=64)``. +- ``publish`` is synchronous and non-blocking: it puts the event on + every live queue with ``put_nowait``. A slow subscriber that lets + its queue fill gets the event DROPPED (oldest overflow) — never a + blocked publisher, never unbounded growth. +- A disconnected subscriber's queue is removed in the generator's + ``finally`` block, so teardown is clean and a closed consumer cannot + leak its queue. + +Fan-out is fire-and-forget by design: the SSE plane mirrors transition +events for the panel UI; a dropped event costs the panel one UI +refresh, and the documented recovery path is the next snapshot poll. +""" +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator + +#: Per-subscriber queue capacity. Bounded so a stalled consumer can +#: never make the publisher or another subscriber wait; beyond this +#: the slow subscriber drops the event (see module docstring). +_QUEUE_MAXSIZE = 64 + +__all__ = ["TransitionPublisher"] + + +class TransitionPublisher: + """A single in-process fan-out point for transition events.""" + + def __init__(self) -> None: + """Create the publisher with no subscribers.""" + self._subscribers: set[asyncio.Queue] = set() + + def publish(self, event_type: str, payload: dict) -> None: + """Fan one transition event out to every current subscriber. + + Synchronous and non-blocking: each subscriber's bounded queue + is filled with ``put_nowait``, and a FULL queue silently drops + the event for that (slow) subscriber rather than blocking the + route handler or growing without bound. + """ + for queue in tuple(self._subscribers): + try: + queue.put_nowait((event_type, payload)) + except asyncio.QueueFull: + # The subscriber is not keeping up; drop the event for + # it and move on (the panel recovers by polling). + pass + + async def subscribe(self) -> AsyncIterator[tuple[str, dict]]: + """Yield ``(event_type, payload)`` tuples until the consumer breaks. + + Registers ONE bounded queue for the duration of the iteration + and removes it in the ``finally`` block, so a client disconnect + (or any break out of the loop) tears the subscription down + cleanly with no leaked queue. + """ + queue: asyncio.Queue = asyncio.Queue(maxsize=_QUEUE_MAXSIZE) + self._subscribers.add(queue) + try: + while True: + yield await queue.get() + finally: + self._subscribers.discard(queue) diff --git a/api/nestquest_core.py b/api/nestquest_core.py index 8f9a1dc..6257dab 100644 --- a/api/nestquest_core.py +++ b/api/nestquest_core.py @@ -116,6 +116,10 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot") #: ``uncomplete_instance``) the panel's write routes call. core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion") +#: The transition-event builders (``core.events.build_quest_completed_event`` +#: et al.) the panel's write routes build their SSE events with. +core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events") + #: The explicit settings object (``core.settings.NestQuestSettings``) #: the snapshot builder's ``settings`` parameter takes. core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings") @@ -124,6 +128,7 @@ __all__ = [ "core", "core_completion", "core_db", + "core_events", "core_migrations", "core_snapshot", "core_settings", diff --git a/api/routes_panel.py b/api/routes_panel.py index 5c297a1..ce61b71 100644 --- a/api/routes_panel.py +++ b/api/routes_panel.py @@ -53,14 +53,19 @@ from __future__ import annotations import datetime import importlib +import json + +from collections.abc import AsyncIterator from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel, field_validator from api.database import DatabaseState from api.dependencies import require_panel_token from api.nestquest_core import ( core_completion, + core_events, core_settings, core_snapshot, ) @@ -299,7 +304,57 @@ async def panel_complete_instance( # The derived state is "done" in both the appended and no-op cases; # ``result.appended`` distinguishes them for callers that care — # the panel route returns the same success either way. + if result.appended: + # An ACTUAL transition: publish the SSE transition event. The + # payload is built by the core builder (never hand-built here) + # and fan-out is fire-and-forget (see api/events.py). + event_type, payload = await core_events.build_quest_completed_event( + database, instance_id, was_on_time=result.was_on_time + ) + request.app.state.publisher.publish(event_type, payload) return {"status": str(result)} +async def _sse_stream(request: Request) -> AsyncIterator[str]: + """Yield the SSE frames for every published transition event. + + One subscriber of the app's single publisher (resolved off + ``app.state.publisher``); each event becomes one + ``event: \\ndata: \\n\\n`` frame. Breaking out of the + publisher's subscription (on client disconnect) tears the + subscription down cleanly — no queue is leaked. + """ + publisher = request.app.state.publisher + async for event_type, payload in publisher.subscribe(): + yield ( + f"event: {event_type}\n" + f"data: {json.dumps(payload)}\n\n" + ) + + +@router.get( + "/events", + summary="Stream panel transition events (SSE)", +) +async def panel_events(request: Request) -> StreamingResponse: + """Stream the app's transition events to the panel (Server-Sent + Events). + + Requires the panel service token (checked by the router's shared + :func:`~api.dependencies.require_panel_token` dependency — the ONE + token check; this handler performs NO auth of its own). + + Returns a ``text/event-stream`` response whose body is produced by + :func:`_sse_stream` off the app's single in-process publisher + (``app.state.publisher``, installed in the lifespan). The stream + ends when the client disconnects, at which point the subscription + is torn down (its queue removed) with nothing leaked. + """ + return StreamingResponse( + _sse_stream(request), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache"}, + ) + + __all__ = ["router"] diff --git a/tests/test_api_sse.py b/tests/test_api_sse.py new file mode 100644 index 0000000..7a91a3b --- /dev/null +++ b/tests/test_api_sse.py @@ -0,0 +1,295 @@ +"""SSE endpoint tests: the panel event stream (task 4d7bae00). + +These tests exercise ``GET /api/v1/panel/events`` — the panel plane's +Server-Sent Events stream — against a REAL server: + +- The stream requires the panel service token (D-012): no/wrong token + returns 401 (the router-level shared dependency, no second check). +- The response is ``text/event-stream`` and each transition arrives as + an ``event: `` / ``data: `` frame. +- Completing an instance through the panel complete route publishes + the ``nestquest_quest_completed`` transition carrying the documented + payload fields (design/ENTITIES-AND-SERVICES.md §3). +- A no-op re-complete publishes nothing (no second completed event). + +Streaming approach (documented): httpx's ASGITransport BUFFERS the +whole response body before returning, so an endless SSE stream hangs +it — exactly the awkward case this task anticipated. These tests +therefore run the app with uvicorn on an ephemeral port (port 0, so +the OS picks a free one) in a daemon thread, and httpx streams from +``http://127.0.0.1:``. Every read is bounded by a timeout so a +wedged stream fails an assertion instead of hanging the suite; the +server is shut down in the fixture's finally block. +""" +from __future__ import annotations + +import asyncio +import datetime +import importlib +import json +import socket +import threading +import time +from pathlib import Path +from types import SimpleNamespace +from zoneinfo import ZoneInfo + +import httpx +import pytest +import uvicorn + +from api import routes_panel +from api.app import create_app +from api.config import ApiConfig +from tests.test_api_panel import PANEL_TOKEN, _seed_household + +#: The documented quest-completed event name (core const, §3). +_core_const = importlib.import_module("nestquest_core.const") +EVENT_QUEST_COMPLETED = _core_const.EVENT_QUEST_COMPLETED + +#: Upper bound for any single stream read; bounds the whole test so a +#: wedged stream fails the assertion instead of hanging the suite. +READ_TIMEOUT = 5.0 + + +def _free_port() -> int: + """Ask the OS for one free TCP port (bound then released).""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class _Server: + """Run the app with uvicorn on an ephemeral port, in a thread. + + ``create_app`` runs INSIDE the server thread (uvicorn's factory + mode), so the lifespan — including ``app.state.publisher`` — is + set up per test on the server's own event loop. Daemon thread + + explicit ``should_exit`` shutdown in the fixture's finally block. + """ + + def __init__(self, db_path: str) -> None: + self.port = _free_port() + self._server = uvicorn.Server( + uvicorn.Config( + # A closure factory: reads nothing at import time. + lambda: create_app( + ApiConfig(db_path=db_path, panel_token=PANEL_TOKEN) + ), + host="127.0.0.1", + port=self.port, + log_level="warning", + ) + ) + self._thread = threading.Thread( + target=self._server.run, daemon=True + ) + + def __enter__(self) -> "_Server": + self._thread.start() + deadline = time.monotonic() + READ_TIMEOUT + while not self._server.started: + if time.monotonic() > deadline or not self._thread.is_alive(): + raise RuntimeError("uvicorn did not start") + time.sleep(0.01) + return self + + def __exit__(self, *exc) -> None: + self._server.should_exit = True + self._thread.join(timeout=READ_TIMEOUT) + + def url(self, path: str) -> str: + """The full URL for ``path`` on this server.""" + return f"http://127.0.0.1:{self.port}{path}" + + +class _EventStream: + """A subscribed SSE reader over a live httpx connection. + + Parses frames (``event:``/``data:`` pairs) into + ``(event_type, payload)`` tuples; ``next_event`` awaits the next + parsed frame with a timeout. + """ + + def __init__(self) -> None: + self.queue: asyncio.Queue = asyncio.Queue() + self._task: asyncio.Task | None = None + self._client: httpx.AsyncClient | None = None + self._stream_cm = None + self._response: httpx.Response | None = None + + async def __aenter__(self) -> "_EventStream": + self._client = httpx.AsyncClient(timeout=READ_TIMEOUT) + self._stream_cm = self._client.stream( + "GET", + self.url, + headers={"Authorization": f"Bearer {PANEL_TOKEN}"}, + ) + self._response = await self._stream_cm.__aenter__() + assert self._response.status_code == 200 + assert self._response.headers["content-type"].startswith( + "text/event-stream" + ) + self._task = asyncio.create_task(self._pump()) + return self + + def _set_url(self, url: str) -> "_EventStream": + """Record the stream URL (called before ``__aenter__``).""" + self.url = url + return self + + async def _pump(self) -> None: + """Read the stream and push parsed frames onto the queue.""" + event_type: str | None = None + data_lines: list[str] = [] + try: + async for chunk in self._response.aiter_text(): + for line in chunk.splitlines(): + if line.startswith("event:"): + event_type = line[len("event:"):].strip() + elif line.startswith("data:"): + data_lines.append(line[len("data:"):].strip()) + elif not line and event_type is not None: + payload = json.loads("\n".join(data_lines)) + await self.queue.put((event_type, payload)) + event_type = None + data_lines = [] + except Exception as error: # pragma: no cover - teardown path + await self.queue.put(error) + + async def next_event(self) -> tuple[str, dict]: + """The next parsed ``(event_type, payload)``, with a timeout.""" + item = await asyncio.wait_for( + self.queue.get(), timeout=READ_TIMEOUT + ) + if isinstance(item, Exception): + raise item + return item + + async def __aexit__(self, *exc) -> None: + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + await self._stream_cm.__aexit__(*exc) + await self._client.__aexit__(*exc) + + +@pytest.fixture +def temp_db_path(tmp_path: Path) -> str: + """A fresh per-test SQLite path under pytest's tmp_path.""" + return str(tmp_path / "nestquest.db") + + +@pytest.fixture +async def seeded_panel(temp_db_path: str) -> SimpleNamespace: + """A seeded household served by a live uvicorn app, clock pinned. + + Same pinning discipline as the panel tests (12:00 UTC, past Ada's + 10:00 due, before Bo's 17:00): the route's ``_local_now`` is + patched so the completion's ``was_on_time`` is deterministic. The + app is created inside uvicorn's factory, so the patch must happen + BEFORE the server starts. + """ + today = datetime.datetime.now().astimezone().date() + pinned_now = datetime.datetime( + today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC") + ) + seed = await _seed_household(temp_db_path, today, pinned_now) + # The app is created inside uvicorn's factory (in the server + # thread), so the patch must be applied BEFORE the server starts. + original = routes_panel._local_now + routes_panel._local_now = lambda: pinned_now + try: + with _Server(temp_db_path) as server: + client = httpx.AsyncClient(base_url=server.url("")) + async with client: + yield SimpleNamespace( + client=client, + seed=seed, + server=server, + url=server.url, + ) + finally: + routes_panel._local_now = original + + +def _stream(url: str) -> _EventStream: + """A subscribed SSE reader for ``url``.""" + return _EventStream()._set_url(url) + + +# --- auth and content type ---------------------------------------------- + + +async def test_events_requires_panel_token(seeded_panel) -> None: + """The SSE route inherits the router's token check (no second one).""" + response = await seeded_panel.client.get("/api/v1/panel/events") + assert response.status_code == 401 + + +# --- the quest-completed transition -------------------------------------- + + +async def test_complete_publishes_quest_completed_event( + seeded_panel, +) -> None: + """A panel completion emits the documented completed transition. + + The stream is subscribed FIRST, the instance is completed through + the panel route, and the emitted frame's type and every documented + payload field (child_id, child_name, instance_id, quest_title, + window, due_date, due_time, occurred_at, was_on_time) are asserted. + """ + seed = seeded_panel.seed + pack_bag = seed.pack_bag_instance + stream = _stream(seeded_panel.url("/api/v1/panel/events")) + async with stream: + response = await seeded_panel.client.post( + f"/api/v1/panel/instances/{pack_bag.id}/complete", + headers={"Authorization": f"Bearer {PANEL_TOKEN}"}, + json={"actor_child_id": seed.ada.id}, + ) + assert response.status_code == 200 + + event_type, payload = await stream.next_event() + assert event_type == EVENT_QUEST_COMPLETED + assert event_type == "nestquest_quest_completed" + assert payload["child_id"] == seed.ada.id + assert payload["child_name"] == "Ada" + assert payload["instance_id"] == pack_bag.id + assert payload["quest_title"] == "Pack bag" + assert payload["window"] == "morning" + assert payload["due_date"] == pack_bag.due_date + assert payload["due_time"] == "10:00" + assert payload["occurred_at"] # strict UTC ISO-8601 stamp + assert payload["was_on_time"] is False # 10:00 due, 12:00 now + + +async def test_no_op_recomplete_publishes_nothing(seeded_panel) -> None: + """A second complete call emits NO second completed event. + + After the first (real) transition, a re-complete is a no-op in the + core layer; the stream must stay silent. + """ + seed = seeded_panel.seed + url = f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete" + body = {"actor_child_id": seed.ada.id} + headers = {"Authorization": f"Bearer {PANEL_TOKEN}"} + stream = _stream(seeded_panel.url("/api/v1/panel/events")) + async with stream: + first = await seeded_panel.client.post(url, headers=headers, json=body) + assert first.status_code == 200 + event_type, _ = await stream.next_event() + assert event_type == EVENT_QUEST_COMPLETED + + second = await seeded_panel.client.post( + url, headers=headers, json=body + ) + assert second.status_code == 200 + with pytest.raises(asyncio.TimeoutError): + # Nothing arrives within the window: the no-op re-complete + # must publish nothing. + await asyncio.wait_for(stream.queue.get(), timeout=0.3)