From 688bc0cfe903f966bbcee4f7dda06e4ba2290149 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Thu, 4 Jun 2026 22:12:10 +0100 Subject: [PATCH] Add SSE updates --- CLAUDE.md | 12 +- backend/src/life_towers/api.py | 136 +++++++++++- backend/src/life_towers/events.py | 77 +++++++ .../life_towers/migrations/002_revision.sql | 7 + backend/src/life_towers/models.py | 8 + backend/tests/test_api.py | 203 +++++++++++++++++- frontend/src/app/models/index.ts | 6 + frontend/src/app/services/api.service.ts | 79 ++++++- .../src/app/services/api.service.vitest.ts | 19 +- frontend/src/app/services/store.service.ts | 176 +++++++++++++-- .../src/app/services/store.service.vitest.ts | 173 ++++++++++++++- frontend/src/app/utils/sse.ts | 54 +++++ frontend/src/app/utils/sse.vitest.ts | 50 +++++ 13 files changed, 958 insertions(+), 42 deletions(-) create mode 100644 backend/src/life_towers/events.py create mode 100644 backend/src/life_towers/migrations/002_revision.sql create mode 100644 frontend/src/app/utils/sse.ts create mode 100644 frontend/src/app/utils/sse.vitest.ts diff --git a/CLAUDE.md b/CLAUDE.md index 9c696f9..0fddce3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -78,6 +78,7 @@ The app is deployed under a path: `https://schmelczer.dev/towers/`. The mechanis proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } ``` +- **SSE (`/api/v1/events`) through nginx**: the endpoint sends `X-Accel-Buffering: no`, which nginx honors to disable response buffering for that stream — so the block above works without a dedicated location. The 20s server-side keepalive comment stays under nginx's default 60s `proxy_read_timeout`, so idle streams don't get culled. If you ever lower `proxy_read_timeout` below ~20s, add `proxy_buffering off;` + a larger read timeout on a `location /towers/api/v1/events`. - **Dev/e2e stay at the root**: `docker-compose.dev.yml` builds with the default `BASE_HREF=/`, and Playwright hits `http://life-towers:8000/`. The relative API paths work identically there, so e2e exercises the same code. ## Build / dev / test / deploy @@ -174,13 +175,22 @@ Font sizes (`library/text.scss`): `--larger/large/medium/small-font-size` = `22/ - **No granular endpoints** for individual entity CRUD. Keep it tree-replace. - Spec is in `docs/api-spec.md`. Backend pydantic models match it. Frontend `models/index.ts` matches it. Field names are **snake_case** on both sides. +### Multi-client sync (revision + CAS + SSE) + +Multiple clients can share one token (same user on phone + laptop). Three pieces keep them in step — see `backend/src/life_towers/events.py`, `api.py`, and `frontend/src/app/{services,utils}`: + +- **Per-user `revision`** (`users.revision`, migration `002_revision.sql`): a monotonic counter bumped inside the same `BEGIN IMMEDIATE` as every PUT. `GET /data` returns it; `PUT /data` returns the new one (`{ "revision": N }`, status 200 — **not** 204 anymore). +- **Compare-and-swap**: the client sends its base revision as the `If-Match` header on PUT. If the stored revision moved underneath it, the server rolls back and returns **409** with `{ "error": "conflict", "revision": }`. Absent `If-Match` ⇒ unguarded write (keeps older cached clients working across a deploy). On 409 the store resolves **server-wins**: it refetches the current server tree and adopts it, discarding this device's un-pushed edit (`store.service.adoptServerData`). The CAS still prevents a stale write from clobbering the other device's data; there is deliberately **no client-side merge** — a conflicting in-flight edit on the losing device is dropped, which is acceptable for a single user across a few devices. (An earlier `utils/sync-merge.ts` 3-way merge was removed in favour of this simpler policy.) +- **SSE push** (`GET /api/v1/events`, `text/event-stream`): an in-process asyncio pub/sub (`events.py`, single worker — see Dockerfile) pushes the new revision to every live connection right after a PUT commits. The client consumes it with **fetch()+ReadableStream**, not `EventSource`, so the Bearer token rides in a header instead of the URL (`api.service.openEventStream` + `utils/sse.ts` parser). On a newer-revision event the store refetches **only when clean**; if it has pending edits it lets the next PUT's CAS resolve it (server-wins). The stream auto-reconnects with backoff; `switchToken`/`ngOnDestroy` tear it down. +- **Gotcha**: Starlette's `BaseHTTPMiddleware` is fine with streaming responses, but **httpx's in-memory `ASGITransport` is not** — it can't open a long-lived stream with concurrent requests, so SSE wire behaviour is verified against real uvicorn, not in pytest. Pytest covers the pub/sub bus + that a PUT publishes onto it; `tests/test_api.py` has the pattern. + ### Backend - All endpoints are inside `APIRouter(prefix="/api/v1")`. Spec drives behavior — if you change a limit, update both spec and code. - Migrations: package data under `src/life_towers/migrations/`, loaded via `importlib.resources.files("life_towers").joinpath("migrations")`. The runner tracks applied state in a `schema_migrations(filename TEXT PRIMARY KEY, applied_at INTEGER)` table. - All sqlite connections must do `PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000`. The `get_connection()` factory does this. - Errors → JSON `{"error": code, "detail": str}` via a single `HTTPException` handler in `main.py`. Stack traces never leak (logged server-side). -- Rate limits via slowapi: `/register` 30/hour/IP, `GET /data` 60/min/token, `PUT /data` 30/min/token. +- Rate limits via slowapi: `/register` 30/hour/IP, `GET /data` 60/min/token, `PUT /data` 30/min/token. `GET /events` (SSE) is intentionally **un**-limited — it's one long-lived connection per client. ## Visual + interaction details that bit me diff --git a/backend/src/life_towers/api.py b/backend/src/life_towers/api.py index 8fbcc7f..6cc7e04 100644 --- a/backend/src/life_towers/api.py +++ b/backend/src/life_towers/api.py @@ -1,12 +1,16 @@ from __future__ import annotations +import asyncio +import json import sqlite3 import time from typing import Annotated import structlog -from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi import APIRouter, Depends, Header, HTTPException, Request +from fastapi.responses import StreamingResponse +from . import events from .auth import get_current_user from .db import db_connection from .limits import limiter @@ -18,6 +22,7 @@ from .models import ( HealthResponse, HslColor, PageOut, + PutDataResponse, RegisterRequest, RegisterResponse, TowerOut, @@ -26,6 +31,34 @@ from .models import ( router = APIRouter(prefix="/api/v1") logger = structlog.get_logger(__name__) +# How long the SSE loop waits for a new revision before emitting a keepalive +# comment. Keeps the connection warm through proxies and bounds disconnect +# detection latency. +SSE_KEEPALIVE_SECONDS = 20 + + +def _read_revision(conn: sqlite3.Connection, user_id: str) -> int: + row = conn.execute( + "SELECT revision FROM users WHERE id = ?", (user_id,) + ).fetchone() + return int(row["revision"]) if row is not None else 0 + + +def _parse_if_match(value: str | None) -> int | None: + """Parse the client's CAS base revision from the If-Match header. + + Tolerates an ETag-style quoted form (``"5"``). Returns None when the header + is absent or unparseable, in which case the write proceeds unguarded + (last-writer-wins) — this keeps older cached clients working across a deploy. + """ + if value is None: + return None + cleaned = value.strip().strip('"') + try: + return int(cleaned) + except ValueError: + return None + @router.get("/health", response_model=HealthResponse) async def health() -> HealthResponse: @@ -63,6 +96,8 @@ async def get_data( user_id: Annotated[str, Depends(get_current_user)], ) -> DataOut: with db_connection() as conn: + revision = _read_revision(conn, user_id) + pages_rows = conn.execute( """ SELECT id, name, hide_create_tower_button, keep_tasks_open, default_date_from, default_date_to @@ -140,24 +175,44 @@ async def get_data( ) ) - return DataOut(pages=pages_out) + return DataOut(pages=pages_out, revision=revision) -@router.put("/data", status_code=204) +@router.put("/data", response_model=PutDataResponse) @limiter.limit("30/minute") async def put_data( request: Request, body: DataIn, user_id: Annotated[str, Depends(get_current_user)], -) -> None: + if_match: Annotated[str | None, Header(alias="If-Match")] = None, +) -> PutDataResponse: # Tree-replace semantics mean the request size IS the user's total storage, # so the 2 MiB request cap (enforced by payload_size_middleware) is the only # quota we need. now = int(time.time()) + base_revision = _parse_if_match(if_match) with db_connection() as conn: conn.execute("BEGIN IMMEDIATE") try: + current_revision = _read_revision(conn, user_id) + + # Compare-and-swap: reject if another client advanced the revision + # under us. The client refetches, merges, and retries with the + # fresh base. An absent If-Match opts out of the guard. + if base_revision is not None and base_revision != current_revision: + conn.rollback() + raise HTTPException( + status_code=409, + detail={ + "error": "conflict", + "detail": "Revision is stale; refetch and retry", + "revision": current_revision, + }, + ) + + new_revision = current_revision + 1 + # Delete existing data for this user (cascades to towers + blocks) conn.execute("DELETE FROM pages WHERE user_id = ?", (user_id,)) @@ -231,13 +286,16 @@ async def put_data( ), ) - # Update last_seen_at + # Bump the revision and refresh last_seen_at in the same transaction. conn.execute( - "UPDATE users SET last_seen_at = ? WHERE id = ?", - (now, user_id), + "UPDATE users SET last_seen_at = ?, revision = ? WHERE id = ?", + (now, new_revision, user_id), ) conn.commit() + except HTTPException: + # Already rolled back (e.g. the CAS 409); just propagate. + raise except sqlite3.IntegrityError as exc: conn.rollback() raise HTTPException( @@ -248,4 +306,66 @@ async def put_data( conn.rollback() raise - logger.info("data_replaced", user_id=token_log_id(user_id), pages=len(body.pages)) + # Notify other live clients for this user to refetch. Done after commit so + # subscribers never observe a revision that isn't durably stored yet. + events.publish(user_id, new_revision) + + logger.info( + "data_replaced", + user_id=token_log_id(user_id), + pages=len(body.pages), + revision=new_revision, + ) + return PutDataResponse(revision=new_revision) + + +def _format_revision_event(revision: int) -> str: + return f"event: revision\ndata: {json.dumps({'revision': revision})}\n\n" + + +@router.get("/events") +async def events_stream( + request: Request, + user_id: Annotated[str, Depends(get_current_user)], +) -> StreamingResponse: + """Server-Sent Events stream that notifies a client to refetch. + + Authenticated with the same Bearer token as the rest of the API — clients + consume it via fetch()+ReadableStream (EventSource cannot set headers), so + the token never leaks into a URL. Each event carries the current revision; + the client refetches GET /data when it sees a revision newer than its own. + """ + queue = events.subscribe(user_id) + + async def event_generator(): + try: + # Emit the current revision immediately so a (re)connecting client + # can catch up on anything it missed while disconnected. + with db_connection() as conn: + yield _format_revision_event(_read_revision(conn, user_id)) + + while True: + try: + revision = await asyncio.wait_for( + queue.get(), timeout=SSE_KEEPALIVE_SECONDS + ) + yield _format_revision_event(revision) + except asyncio.TimeoutError: + if await request.is_disconnected(): + break + # Comment line: keeps proxies from idling the connection out. + yield ": keepalive\n\n" + finally: + events.unsubscribe(user_id, queue) + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + # Disable nginx response buffering for this stream even if the + # location block forgets `proxy_buffering off`. + "X-Accel-Buffering": "no", + }, + ) diff --git a/backend/src/life_towers/events.py b/backend/src/life_towers/events.py new file mode 100644 index 0000000..d6aa06a --- /dev/null +++ b/backend/src/life_towers/events.py @@ -0,0 +1,77 @@ +"""In-process pub/sub for Server-Sent Events. + +Single-worker deployment (see Dockerfile: uvicorn with no ``--workers``), so a +plain in-memory registry is enough — there is one event loop and every +subscriber queue lives on it. If this ever grows to multiple workers, this +module is the seam to swap for a cross-process bus (Redis pub/sub, LISTEN/NOTIFY, +or polling the ``users.revision`` column). + +Each subscriber gets a 1-slot queue that always holds the *latest* revision to +notify. Coalescing is intentional: the SSE payload is just "something changed, +current revision is N", so a burst of writes collapses to a single refetch +signal rather than a backlog. +""" + +from __future__ import annotations + +import asyncio + +import structlog + +logger = structlog.get_logger(__name__) + +# user_id -> set of per-connection queues. Each queue carries the most recent +# revision the connection still needs to flush to its client. +_subscribers: dict[str, set[asyncio.Queue[int]]] = {} + + +def subscribe(user_id: str) -> asyncio.Queue[int]: + """Register a new SSE connection for ``user_id`` and return its queue. + + Must be called from the event loop (the SSE route handler is async), so the + queue binds to the running loop. + """ + queue: asyncio.Queue[int] = asyncio.Queue(maxsize=1) + _subscribers.setdefault(user_id, set()).add(queue) + return queue + + +def unsubscribe(user_id: str, queue: asyncio.Queue[int]) -> None: + """Drop a connection's queue; prune the user entry when the last one goes.""" + subs = _subscribers.get(user_id) + if subs is None: + return + subs.discard(queue) + if not subs: + _subscribers.pop(user_id, None) + + +def publish(user_id: str, revision: int) -> None: + """Notify every live connection for ``user_id`` of the new revision. + + Called from the (single) event loop right after a PUT commits, so the + ``put_nowait`` calls are loop-safe. Coalesces into each 1-slot queue: if a + connection has not yet drained its previous signal, replace it with the + newer revision rather than block or grow unbounded. + """ + subs = _subscribers.get(user_id) + if not subs: + return + for queue in subs: + try: + queue.put_nowait(revision) + except asyncio.QueueFull: + # Connection is behind — drop the stale value and keep the newest. + try: + queue.get_nowait() + except asyncio.QueueEmpty: + pass + try: + queue.put_nowait(revision) + except asyncio.QueueFull: + pass + + +def connection_count(user_id: str) -> int: + """Number of live SSE connections for a user (used in tests).""" + return len(_subscribers.get(user_id, ())) diff --git a/backend/src/life_towers/migrations/002_revision.sql b/backend/src/life_towers/migrations/002_revision.sql new file mode 100644 index 0000000..aaba6e9 --- /dev/null +++ b/backend/src/life_towers/migrations/002_revision.sql @@ -0,0 +1,7 @@ +-- Per-user revision counter for multi-client sync. +-- Bumped inside the same transaction as every PUT /data so clients can: +-- * detect another client's write (cheap compare, no full-tree diff), and +-- * guard their own writes with compare-and-swap (stale If-Match -> 409). +-- Existing users start at 0; their next confirmed GET seeds the client base. + +ALTER TABLE users ADD COLUMN revision INTEGER NOT NULL DEFAULT 0; diff --git a/backend/src/life_towers/models.py b/backend/src/life_towers/models.py index b40d5e3..1d78a8b 100644 --- a/backend/src/life_towers/models.py +++ b/backend/src/life_towers/models.py @@ -127,6 +127,14 @@ class DataIn(BaseModel): class DataOut(BaseModel): pages: list[PageOut] + # Monotonic per-user version, bumped on every successful PUT. Clients keep + # it as their compare-and-swap base and to detect another client's writes. + revision: int + + +class PutDataResponse(BaseModel): + # The new revision after this write; the client adopts it as its CAS base. + revision: int class RegisterRequest(BaseModel): diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index a4f8538..4080f5c 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import uuid from pathlib import Path from typing import AsyncGenerator @@ -198,7 +199,7 @@ async def test_get_data_empty_user(client: AsyncClient) -> None: headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 - assert resp.json() == {"pages": []} + assert resp.json() == {"pages": [], "revision": 0} # --------------------------------------------------------------------------- @@ -253,12 +254,14 @@ async def test_round_trip(client: AsyncClient) -> None: tree = _make_tree() put_resp = await client.put("/api/v1/data", json=tree, headers=headers) - assert put_resp.status_code == 204 + assert put_resp.status_code == 200 + assert put_resp.json() == {"revision": 1} get_resp = await client.get("/api/v1/data", headers=headers) assert get_resp.status_code == 200 data = get_resp.json() + assert data["revision"] == 1 assert len(data["pages"]) == 2 for pi, page in enumerate(data["pages"]): assert page["id"] == tree["pages"][pi]["id"] @@ -289,7 +292,7 @@ async def test_difficulty_defaults_to_one_when_omitted(client: AsyncClient) -> N del tree["pages"][0]["towers"][0]["blocks"][0]["difficulty"] put_resp = await client.put("/api/v1/data", json=tree, headers=headers) - assert put_resp.status_code == 204 + assert put_resp.status_code == 200 data = (await client.get("/api/v1/data", headers=headers)).json() assert data["pages"][0]["towers"][0]["blocks"][0]["difficulty"] == 1 @@ -342,7 +345,7 @@ async def test_put_cross_user_id_conflict_returns_409(client: AsyncClient) -> No json=tree, headers={"Authorization": f"Bearer {first_token}"}, ) - assert first_resp.status_code == 204 + assert first_resp.status_code == 200 second_resp = await client.put( "/api/v1/data", @@ -455,3 +458,195 @@ async def test_register_rate_limit(client: AsyncClient) -> None: resp = await client.post("/api/v1/register", json={"token": make_uuidv4()}) responses.append(resp.status_code) assert responses[-1] == 429, f"Expected 429 on 31st request, got: {responses[-3:]}" + + +# --------------------------------------------------------------------------- +# Revision counter + compare-and-swap (multi-client sync) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_revision_increments_on_each_put(client: AsyncClient) -> None: + token = make_uuidv4() + await client.post("/api/v1/register", json={"token": token}) + headers = {"Authorization": f"Bearer {token}"} + + # Fresh user starts at revision 0. + assert (await client.get("/api/v1/data", headers=headers)).json()["revision"] == 0 + + r1 = await client.put("/api/v1/data", json=_make_tree(), headers=headers) + assert r1.json() == {"revision": 1} + r2 = await client.put("/api/v1/data", json=_make_tree(), headers=headers) + assert r2.json() == {"revision": 2} + + assert (await client.get("/api/v1/data", headers=headers)).json()["revision"] == 2 + + +@pytest.mark.asyncio +async def test_stale_if_match_returns_409_with_current_revision( + client: AsyncClient, +) -> None: + token = make_uuidv4() + await client.post("/api/v1/register", json={"token": token}) + headers = {"Authorization": f"Bearer {token}"} + + # Base 0 matches a fresh user -> succeeds, revision becomes 1. + ok = await client.put( + "/api/v1/data", json=_make_tree(), headers={**headers, "If-Match": "0"} + ) + assert ok.status_code == 200 + assert ok.json() == {"revision": 1} + + # Re-using the now-stale base 0 is rejected; the body carries the truth. + stale = await client.put( + "/api/v1/data", json=_make_tree(), headers={**headers, "If-Match": "0"} + ) + assert stale.status_code == 409 + body = stale.json() + assert body["error"] == "conflict" + assert body["revision"] == 1 + + # The conflicting write must NOT have advanced the revision. + assert (await client.get("/api/v1/data", headers=headers)).json()["revision"] == 1 + + # Retrying with the fresh base succeeds. + retry = await client.put( + "/api/v1/data", json=_make_tree(), headers={**headers, "If-Match": "1"} + ) + assert retry.status_code == 200 + assert retry.json() == {"revision": 2} + + +@pytest.mark.asyncio +async def test_put_without_if_match_skips_the_guard(client: AsyncClient) -> None: + """Absent If-Match keeps older cached clients writing (last-writer-wins).""" + token = make_uuidv4() + await client.post("/api/v1/register", json={"token": token}) + headers = {"Authorization": f"Bearer {token}"} + + # Advance to revision 2 first. + await client.put("/api/v1/data", json=_make_tree(), headers=headers) + await client.put("/api/v1/data", json=_make_tree(), headers=headers) + + # No If-Match -> write goes through regardless of current revision. + resp = await client.put("/api/v1/data", json=_make_tree(), headers=headers) + assert resp.status_code == 200 + assert resp.json() == {"revision": 3} + + +# --------------------------------------------------------------------------- +# Server-Sent Events stream (notify-to-refetch) +# +# The full SSE wire behaviour (incremental flush + live push) is exercised +# end-to-end against real uvicorn — httpx's in-memory ASGITransport can't model +# a long-lived streaming response with concurrent requests, so here we cover the +# two seams it CAN reach reliably: the in-process pub/sub bus, and the fact that +# a PUT publishes the new revision onto that bus (which the stream then drains). +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_event_bus_subscribe_publish_unsubscribe() -> None: + from life_towers import events + + user = make_uuidv4() + queue = events.subscribe(user) + assert events.connection_count(user) == 1 + + events.publish(user, 5) + assert await asyncio.wait_for(queue.get(), 1) == 5 + + events.unsubscribe(user, queue) + assert events.connection_count(user) == 0 + + +@pytest.mark.asyncio +async def test_event_bus_coalesces_to_latest_revision() -> None: + """A backed-up connection should see only the newest revision, not a queue.""" + from life_towers import events + + user = make_uuidv4() + queue = events.subscribe(user) + events.publish(user, 1) + events.publish(user, 2) + events.publish(user, 3) + + assert await asyncio.wait_for(queue.get(), 1) == 3 + assert queue.empty() + events.unsubscribe(user, queue) + + +@pytest.mark.asyncio +async def test_event_bus_fans_out_to_all_connections() -> None: + from life_towers import events + + user = make_uuidv4() + q1 = events.subscribe(user) + q2 = events.subscribe(user) + assert events.connection_count(user) == 2 + + events.publish(user, 7) + assert await asyncio.wait_for(q1.get(), 1) == 7 + assert await asyncio.wait_for(q2.get(), 1) == 7 + + events.unsubscribe(user, q1) + events.unsubscribe(user, q2) + assert events.connection_count(user) == 0 + + +@pytest.mark.asyncio +async def test_event_bus_publish_without_subscribers_is_noop() -> None: + from life_towers import events + + events.publish(make_uuidv4(), 99) # must not raise + + +@pytest.mark.asyncio +async def test_put_publishes_new_revision_to_subscribers(client: AsyncClient) -> None: + """The integration seam: a real PUT must notify a live SSE subscriber.""" + from life_towers import events + + token = make_uuidv4() + await client.post("/api/v1/register", json={"token": token}) + headers = {"Authorization": f"Bearer {token}"} + + queue = events.subscribe(token) + try: + await client.put("/api/v1/data", json=_make_tree(), headers=headers) + assert await asyncio.wait_for(queue.get(), 2) == 1 + + await client.put("/api/v1/data", json=_make_tree(), headers=headers) + assert await asyncio.wait_for(queue.get(), 2) == 2 + finally: + events.unsubscribe(token, queue) + + +@pytest.mark.asyncio +async def test_rejected_put_does_not_publish(client: AsyncClient) -> None: + """A CAS-rejected (409) write must not emit a spurious refetch signal.""" + from life_towers import events + + token = make_uuidv4() + await client.post("/api/v1/register", json={"token": token}) + headers = {"Authorization": f"Bearer {token}"} + + # Advance to revision 1. + await client.put("/api/v1/data", json=_make_tree(), headers=headers) + + queue = events.subscribe(token) + try: + # Stale base -> 409, must not publish. + stale = await client.put( + "/api/v1/data", json=_make_tree(), headers={**headers, "If-Match": "0"} + ) + assert stale.status_code == 409 + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(queue.get(), 0.2) + finally: + events.unsubscribe(token, queue) + + +@pytest.mark.asyncio +async def test_sse_requires_auth(client: AsyncClient) -> None: + # No Bearer token: auth fails before any streaming starts, so a plain GET + # returns 401 without holding the connection open. + resp = await client.get("/api/v1/events") + assert resp.status_code == 401 diff --git a/frontend/src/app/models/index.ts b/frontend/src/app/models/index.ts index 083f09d..5979974 100644 --- a/frontend/src/app/models/index.ts +++ b/frontend/src/app/models/index.ts @@ -35,6 +35,12 @@ export interface TreeDto { pages: Page[]; } +/** Response of GET /data: the tree plus the user's current sync revision. */ +export interface DataResponse { + pages: Page[]; + revision: number; +} + export type SaveStatus = | 'idle' | 'saving' diff --git a/frontend/src/app/services/api.service.ts b/frontend/src/app/services/api.service.ts index 3572df0..7d54525 100644 --- a/frontend/src/app/services/api.service.ts +++ b/frontend/src/app/services/api.service.ts @@ -1,7 +1,17 @@ import { Injectable, inject } from '@angular/core'; import { HttpClient, HttpHeaders } from '@angular/common/http'; import { firstValueFrom } from 'rxjs'; -import { TreeDto } from '../models'; +import { DataResponse, TreeDto } from '../models'; +import { createSseParser } from '../utils/sse'; + +/** Callbacks for a live event stream. */ +export interface EventStreamHandlers { + /** A revision the server says is current; the store decides whether to refetch. */ + onRevision: (revision: number) => void; + /** The stream ended on its own (server closed or network error) — not an + * intentional close. The caller may reconnect. */ + onClosed: () => void; +} @Injectable({ providedIn: 'root' }) export class ApiService { @@ -17,16 +27,73 @@ export class ApiService { ); } - getData(token: string): Promise { + getData(token: string): Promise { return firstValueFrom( - this.http.get('api/v1/data', { headers: this.authHeaders(token) }), + this.http.get('api/v1/data', { headers: this.authHeaders(token) }), ); } - async putData(token: string, tree: TreeDto): Promise { - await firstValueFrom( - this.http.put('api/v1/data', tree, { headers: this.authHeaders(token) }), + /** + * Replace the user's tree. `baseRevision` is the client's compare-and-swap + * base, sent as If-Match; the server rejects with 409 if it has moved on. + * Resolves to the new revision the write produced. + */ + async putData(token: string, tree: TreeDto, baseRevision: number): Promise { + const headers = this.authHeaders(token).set('If-Match', String(baseRevision)); + const res = await firstValueFrom( + this.http.put<{ revision: number }>('api/v1/data', tree, { headers }), ); + return res.revision; + } + + /** + * Open the SSE stream for a token via fetch()+ReadableStream so the Bearer + * token travels in a header (EventSource can't do that). Returns a function + * that closes the stream; closing it does NOT invoke `onClosed`. + */ + openEventStream(token: string, handlers: EventStreamHandlers): () => void { + const controller = new AbortController(); + void this.consumeEventStream(token, handlers, controller.signal); + return () => controller.abort(); + } + + private async consumeEventStream( + token: string, + handlers: EventStreamHandlers, + signal: AbortSignal, + ): Promise { + try { + const response = await fetch('api/v1/events', { + headers: { Authorization: `Bearer ${token}`, Accept: 'text/event-stream' }, + signal, + cache: 'no-store', + }); + if (!response.ok || !response.body) return; + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const feed = createSseParser((message) => { + if (message.event !== null && message.event !== 'revision') return; + try { + const parsed = JSON.parse(message.data) as { revision?: unknown }; + if (typeof parsed.revision === 'number') handlers.onRevision(parsed.revision); + } catch { + /* ignore malformed frames */ + } + }); + + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + feed(decoder.decode(value, { stream: true })); + } + } catch { + /* aborted or network error — handled below */ + } finally { + // An intentional close (controller.abort()) should not trigger a + // reconnect; only an unexpected end does. + if (!signal.aborted) handlers.onClosed(); + } } private authHeaders(token: string): HttpHeaders { diff --git a/frontend/src/app/services/api.service.vitest.ts b/frontend/src/app/services/api.service.vitest.ts index 282fdbe..1dfe41a 100644 --- a/frontend/src/app/services/api.service.vitest.ts +++ b/frontend/src/app/services/api.service.vitest.ts @@ -3,7 +3,7 @@ import { TestBed } from '@angular/core/testing'; import { provideHttpClient } from '@angular/common/http'; import { HttpTestingController, provideHttpClientTesting } from '@angular/common/http/testing'; import { ApiService } from './api.service'; -import type { TreeDto } from '../models'; +import type { DataResponse, TreeDto } from '../models'; describe('ApiService', () => { let service: ApiService; @@ -21,24 +21,25 @@ describe('ApiService', () => { http.verify(); }); - it('gets data with a bearer token', async () => { - const tree: TreeDto = { pages: [] }; + it('gets data with a bearer token and returns the revision', async () => { + const body: DataResponse = { pages: [], revision: 7 }; const promise = service.getData('token-1'); const req = http.expectOne('api/v1/data'); expect(req.request.method).toBe('GET'); expect(req.request.headers.get('Authorization')).toBe('Bearer token-1'); - req.flush(tree); - await expect(promise).resolves.toEqual(tree); + req.flush(body); + await expect(promise).resolves.toEqual(body); }); - it('puts data with a bearer token', async () => { + it('puts data with a bearer token + If-Match base revision and returns the new revision', async () => { const tree: TreeDto = { pages: [] }; - const promise = service.putData('token-1', tree); + const promise = service.putData('token-1', tree, 4); const req = http.expectOne('api/v1/data'); expect(req.request.method).toBe('PUT'); expect(req.request.headers.get('Authorization')).toBe('Bearer token-1'); + expect(req.request.headers.get('If-Match')).toBe('4'); expect(req.request.body).toBe(tree); - req.flush(null); - await expect(promise).resolves.toBeUndefined(); + req.flush({ revision: 5 }); + await expect(promise).resolves.toBe(5); }); }); diff --git a/frontend/src/app/services/store.service.ts b/frontend/src/app/services/store.service.ts index 2431c91..62d8a4d 100644 --- a/frontend/src/app/services/store.service.ts +++ b/frontend/src/app/services/store.service.ts @@ -1,13 +1,16 @@ import { Injectable, inject, signal, OnDestroy } from '@angular/core'; import { ApiService } from './api.service'; import { AnalyticsService } from './analytics.service'; -import { Page, Tower, Block, TreeDto, SaveStatus, HslColor } from '../models'; +import { Page, Tower, Block, TreeDto, DataResponse, SaveStatus, HslColor } from '../models'; const TOKEN_KEY = 'life-towers.token.v4'; const CACHE_KEY_PREFIX = 'life-towers.cache.v4'; const PENDING_CACHE_KEY_PREFIX = 'life-towers.cache-pending.v4'; const DEBOUNCE_MS = 750; const MAX_RETRIES = 5; +// SSE reconnect backoff after the stream drops (network blip, server restart). +const SSE_RECONNECT_BASE_MS = 1000; +const SSE_RECONNECT_MAX_MS = 30_000; // RFC 4122 v4 UUID. Prefers crypto.randomUUID (secure contexts only) and // falls back to crypto.getRandomValues — which works on plain http origins @@ -115,6 +118,17 @@ export class StoreService implements OnDestroy { // clearing a newer pending cache entry when it completes. private localMutationRevision = 0; + // ── Server revision (compare-and-swap base) ───────────────────────────────── + // The revision the server last confirmed for us; sent as the If-Match base on + // every PUT and compared against SSE notifications to decide whether to refetch. + private serverRevision = 0; + + // ── Live sync (SSE) ───────────────────────────────────────────────────────── + private closeEventStream: (() => void) | null = null; + private eventStreamToken = ''; + private sseReconnectTimer: ReturnType | null = null; + private sseReconnectAttempts = 0; + // ── Cross-tab sync ───────────────────────────────────────────────────────── private readonly storageListener = (e: StorageEvent) => { if (e.key === TOKEN_KEY && e.newValue && e.newValue !== this._token()) { @@ -196,6 +210,12 @@ export class StoreService implements OnDestroy { if (this.initGeneration === generation) { this._loading.set(false); } + // Subscribe to live updates for this token. Started even if the data load + // failed (we'll have fallen back to cache) — the stream self-heals when + // connectivity returns and its first event triggers a refetch. + if (this.isCurrentInit(generation, token)) { + this.startEventStream(token); + } } } @@ -207,8 +227,13 @@ export class StoreService implements OnDestroy { * Apply a freshly-fetched server tree. If the server is empty but our local * cache holds data, the cache wins and we schedule a push — otherwise the * "server forgot me" recovery would silently wipe offline edits. + * + * The server's revision becomes our CAS base regardless of which view we + * display: even when the cache wins, the next PUT is guarded against it. */ - private adoptServerTree(tree: TreeDto, token: string): void { + private adoptServerTree(data: DataResponse, token: string): void { + this.setServerRevision(data, token); + if (safeGet(pendingCacheKeyForToken(token))) { const cachedTree = this.readCachedTree(token); if (cachedTree?.pages && cachedTree.pages.length > 0) { @@ -218,7 +243,7 @@ export class StoreService implements OnDestroy { } } - if (tree.pages.length === 0) { + if (data.pages.length === 0) { const cachedTree = this.readCachedTree(token); if (cachedTree?.pages && cachedTree.pages.length > 0) { this._pages.set(cachedTree.pages); @@ -226,8 +251,14 @@ export class StoreService implements OnDestroy { return; } } - this._pages.set(tree.pages); - this.updateCache(token, tree); + this._pages.set(data.pages); + this.updateCache(token, { pages: data.pages }); + } + + /** Record the server's revision as our compare-and-swap base. */ + private setServerRevision(data: DataResponse, token: string): void { + if (this._token() !== token) return; + this.serverRevision = data.revision ?? 0; } private loadFromCache(token: string): void { @@ -481,6 +512,9 @@ export class StoreService implements OnDestroy { const token = newToken.toLowerCase(); if (!isUuidV4(token)) return; this.cancelPendingWrites(); + // Tear down the old account's live stream before init() opens a new one. + this.stopEventStream(); + this.sseReconnectAttempts = 0; this.initGeneration++; this.initPromise = null; safeSet(TOKEN_KEY, token); @@ -489,6 +523,7 @@ export class StoreService implements OnDestroy { this._loading.set(true); this._saveStatus.set('idle'); this.localMutationRevision = 0; + this.serverRevision = 0; void this.init(); } @@ -549,14 +584,17 @@ export class StoreService implements OnDestroy { private async attempt(put: PendingPut, attempt: number): Promise { this._saveStatus.set(attempt === 0 ? 'saving' : 'retrying'); try { - await this.api.putData(put.token, put.tree); + const newRevision = await this.api.putData(put.token, put.tree, this.serverRevision); this._saveStatus.set('saved'); - if ( - this._token() === put.token && - put.revision === this.localMutationRevision && - !this.dirtyDuringFlush - ) { - this.updateCache(put.token, put.tree); + if (this._token() === put.token) { + // A successful write advances the revision by exactly one, so fall back + // to that if the response didn't carry a number. + this.serverRevision = Number.isFinite(newRevision) + ? newRevision + : this.serverRevision + 1; + if (put.revision === this.localMutationRevision && !this.dirtyDuringFlush) { + this.updateCache(put.token, put.tree); + } } return; } catch (err: unknown) { @@ -573,6 +611,24 @@ export class StoreService implements OnDestroy { return; } + // 409: another client wrote since our base revision. Resolve server-wins — + // refetch the current server tree and adopt it, discarding this device's + // un-pushed edit. The CAS still prevents a stale write from clobbering the + // other device's data; we just don't merge the two views. + if (status === 409) { + this._saveStatus.set('retrying'); + try { + const remote = await this.api.getData(put.token); + if (this._token() !== put.token) return; + this.adoptServerData(remote, put.token); + this._saveStatus.set('saved'); + return; + } catch { + // Couldn't refetch (network); fall through to backoff and retry the + // PUT, which will 409 again and re-attempt the refetch. + } + } + // 401 mid-PUT: server forgot us. Re-register (idempotent) and retry. if (status === 401) { try { @@ -619,6 +675,101 @@ export class StoreService implements OnDestroy { } } + // ── Live sync (SSE) ───────────────────────────────────────────────────────── + + private startEventStream(token: string): void { + if (typeof window === 'undefined') return; + if (this.eventStreamToken === token && this.closeEventStream) return; + this.stopEventStream(); + this.eventStreamToken = token; + this.closeEventStream = this.api.openEventStream(token, { + onRevision: (revision) => this.onRemoteRevision(token, revision), + onClosed: () => this.onEventStreamClosed(token), + }); + } + + private onRemoteRevision(token: string, revision: number): void { + if (this._token() !== token) return; + // A delivered event proves the stream works — reset the reconnect backoff. + this.sseReconnectAttempts = 0; + // Our own echo, or an out-of-order/stale frame: nothing new to pull. + if (revision <= this.serverRevision) return; + // If a save is pending/in-flight its compare-and-swap will reconcile via a + // 409; refetching now would race it. Only adopt when we're clean. + if (this.hasPendingWork()) return; + void this.pullFromRemote(token); + } + + private onEventStreamClosed(token: string): void { + this.closeEventStream = null; + this.eventStreamToken = ''; + if (this._token() !== token) return; + if (this.sseReconnectTimer !== null) clearTimeout(this.sseReconnectTimer); + const delay = Math.min( + SSE_RECONNECT_BASE_MS * 2 ** this.sseReconnectAttempts, + SSE_RECONNECT_MAX_MS, + ); + this.sseReconnectAttempts += 1; + this.sseReconnectTimer = setTimeout(() => { + this.sseReconnectTimer = null; + if (this._token() === token) this.startEventStream(token); + }, delay); + } + + private stopEventStream(): void { + if (this.sseReconnectTimer !== null) { + clearTimeout(this.sseReconnectTimer); + this.sseReconnectTimer = null; + } + if (this.closeEventStream) { + this.closeEventStream(); + this.closeEventStream = null; + } + this.eventStreamToken = ''; + } + + /** + * Pull the server tree after a remote-change notification. Reached only when + * we're clean, so the server is authoritative and we adopt it wholesale — + * unless the user starts editing during the fetch, in which case we back off + * and let the next save's compare-and-swap reconcile (so the edit survives). + */ + private async pullFromRemote(token: string): Promise { + if (this.hasPendingWork()) return; + let remote: DataResponse; + try { + remote = await this.api.getData(token); + } catch { + return; // transient; a later event or reconnect retries + } + if (this._token() !== token) return; + if (this.hasPendingWork()) return; // edited mid-fetch → defer to CAS + this.adoptServerData(remote, token); + } + + /** + * Adopt a server tree as the new truth. Used both when a clean client pulls a + * remote change (nothing local to lose) and on the 409 server-wins path (any + * un-pushed local edit on this device is intentionally discarded). + */ + private adoptServerData(data: DataResponse, token: string): void { + if (this._token() !== token) return; + this.setServerRevision(data, token); + this._pages.set(data.pages); + this.updateCache(token, { pages: data.pages }); + } + + /** True while any local change is unsaved, being saved, or awaiting retry. */ + private hasPendingWork(): boolean { + return ( + this.debounceTimer !== null || + this.retryTimer !== null || + this.flushInFlight || + this.dirtyDuringFlush || + !!safeGet(pendingCacheKeyForToken(this._token())) + ); + } + // ── Example data ────────────────────────────────────────────────────────── loadExample(): string { @@ -745,6 +896,7 @@ export class StoreService implements OnDestroy { ngOnDestroy(): void { this.cancelPendingWrites(); + this.stopEventStream(); if (typeof window !== 'undefined') { window.removeEventListener('storage', this.storageListener); } diff --git a/frontend/src/app/services/store.service.vitest.ts b/frontend/src/app/services/store.service.vitest.ts index 2ecbe76..18113dd 100644 --- a/frontend/src/app/services/store.service.vitest.ts +++ b/frontend/src/app/services/store.service.vitest.ts @@ -51,17 +51,35 @@ interface MockApi { getData: ReturnType; putData: ReturnType; health: ReturnType; + openEventStream: ReturnType; } function makeMockApi(): MockApi { return { health: vi.fn().mockResolvedValue({ status: 'ok' }), register: vi.fn().mockResolvedValue({ user_id: 'u' }), - getData: vi.fn().mockResolvedValue({ pages: [] } satisfies TreeDto), - putData: vi.fn().mockResolvedValue(undefined), + getData: vi.fn().mockResolvedValue({ pages: [], revision: 0 }), + putData: vi.fn().mockResolvedValue(1), + // Returns the stream's close handle; tests override to capture callbacks. + openEventStream: vi.fn().mockReturnValue(() => {}), }; } +// Grab the handlers the store last passed to openEventStream so a test can +// simulate a server push. +function lastStreamHandlers(api: MockApi): { + onRevision: (revision: number) => void; + onClosed: () => void; +} { + const calls = api.openEventStream.mock.calls; + return calls[calls.length - 1][1]; +} + +// Flush awaited promise chains that contain no timers. +async function flush(): Promise { + await vi.advanceTimersByTimeAsync(0); +} + const FIXED_UUID = '11111111-2222-4333-8444-555555555555'; const TOKEN_KEY = 'life-towers.token.v4'; const CACHE_KEY = `life-towers.cache.v4.${FIXED_UUID}`; @@ -673,4 +691,155 @@ describe('StoreService', () => { expect(store.pages()).toHaveLength(1); expect(store.pages()[0].name).toBe('from-other-tab'); }); + + // ── Multi-client sync: revision + compare-and-swap + SSE ──────────────────── + + it('sends the server revision as the PUT base and adopts the returned one', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + api.getData.mockResolvedValue({ pages: [], revision: 3 }); + api.putData.mockResolvedValue(4); + const store = configure(api); + await store.init(); + + store.addPage('x'); + await vi.advanceTimersByTimeAsync(750); + expect(api.putData).toHaveBeenLastCalledWith(FIXED_UUID, expect.anything(), 3); + + // The 4 returned by the first PUT becomes the base of the next one. + api.putData.mockResolvedValue(5); + store.addPage('y'); + await vi.advanceTimersByTimeAsync(750); + expect(api.putData).toHaveBeenLastCalledWith(FIXED_UUID, expect.anything(), 4); + }); + + it('opens an event stream for the token on init', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + const store = configure(api); + await store.init(); + + expect(api.openEventStream).toHaveBeenCalledTimes(1); + expect(api.openEventStream.mock.calls[0][0]).toBe(FIXED_UUID); + }); + + it('refetches and adopts the server tree on a newer-revision SSE event when clean', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + api.getData.mockResolvedValueOnce({ pages: [], revision: 1 }); + const store = configure(api); + await store.init(); + + api.getData.mockResolvedValueOnce({ + pages: [mkPage('from-other-device')], + revision: 5, + }); + lastStreamHandlers(api).onRevision(5); + await flush(); + + expect(api.getData).toHaveBeenCalledTimes(2); + expect(store.pages()).toHaveLength(1); + expect(store.pages()[0].name).toBe('from-other-device'); + }); + + it('ignores an SSE event that is not newer than our revision (our own echo)', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + api.getData.mockResolvedValue({ pages: [], revision: 3 }); + const store = configure(api); + await store.init(); + + api.getData.mockClear(); + lastStreamHandlers(api).onRevision(3); + await flush(); + + expect(api.getData).not.toHaveBeenCalled(); + }); + + it('defers an SSE refetch while there are pending local edits (CAS handles it)', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + api.getData.mockResolvedValue({ pages: [], revision: 1 }); + const store = configure(api); + await store.init(); + + store.addPage('local'); // now dirty: debounce pending + pending cache + api.getData.mockClear(); + lastStreamHandlers(api).onRevision(9); + await flush(); + + expect(api.getData).not.toHaveBeenCalled(); + }); + + it('on 409 adopts the server tree (server wins) and discards the local edit', async () => { + const PAGE_A = 'aaaaaaaa-1111-4111-8111-111111111111'; + const PAGE_B = 'bbbbbbbb-2222-4222-8222-222222222222'; + const pageWith = (id: string, name: string): TreeDto['pages'][number] => ({ + id, + name, + hide_create_tower_button: false, + keep_tasks_open: false, + default_date_from: null, + default_date_to: null, + towers: [], + }); + + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + api.getData.mockResolvedValueOnce({ pages: [pageWith(PAGE_A, 'A')], revision: 1 }); + // The PUT is rejected as stale; under server-wins we do NOT retry it. + api.putData.mockRejectedValueOnce(httpError(409)); + // The 409 refetch returns a tree where another device added page B. + api.getData.mockResolvedValueOnce({ + pages: [pageWith(PAGE_A, 'A'), pageWith(PAGE_B, 'from-other-device')], + revision: 5, + }); + const store = configure(api); + await store.init(); + + // Local edit to page A, then save. + store.updatePage(PAGE_A, { name: 'A-edited-locally' }); + await vi.advanceTimersByTimeAsync(750); + await vi.runAllTimersAsync(); + + const byId = new Map(store.pages().map((p) => [p.id, p.name])); + // Server wins: the local edit to A is discarded and the remote tree adopted. + expect(byId.get(PAGE_A)).toBe('A'); + expect(byId.get(PAGE_B)).toBe('from-other-device'); + + // The stale PUT fired once and was not retried; the refetched revision (5) + // is now our CAS base. + expect(api.putData).toHaveBeenCalledTimes(1); + expect(store.saveStatus()).toBe('saved'); + }); + + it('closes the event stream on destroy', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + const close = vi.fn(); + api.openEventStream.mockReturnValue(close); + const store = configure(api); + await store.init(); + + store.ngOnDestroy(); + expect(close).toHaveBeenCalledTimes(1); + }); + + it('closes the old stream and opens a new one on switchToken', async () => { + storage[TOKEN_KEY] = FIXED_UUID; + const api = makeMockApi(); + const closeOld = vi.fn(); + api.openEventStream.mockReturnValueOnce(closeOld).mockReturnValue(() => {}); + const store = configure(api); + await store.init(); + + store.switchToken(OTHER_TOKEN); + await vi.advanceTimersByTimeAsync(0); + + expect(closeOld).toHaveBeenCalledTimes(1); + expect(api.openEventStream.mock.calls.length).toBeGreaterThanOrEqual(2); + expect(api.openEventStream.mock.calls[api.openEventStream.mock.calls.length - 1][0]).toBe( + OTHER_TOKEN, + ); + }); }); diff --git a/frontend/src/app/utils/sse.ts b/frontend/src/app/utils/sse.ts new file mode 100644 index 0000000..ed1c3ce --- /dev/null +++ b/frontend/src/app/utils/sse.ts @@ -0,0 +1,54 @@ +/** + * Minimal incremental parser for the Server-Sent Events wire format. + * + * We consume the event stream with fetch()+ReadableStream rather than the + * EventSource API (which can't send an Authorization header), so we parse the + * frames ourselves. Returns a function you feed decoded text chunks; it buffers + * across chunk boundaries and invokes `onMessage` once per complete event + * (events are terminated by a blank line). Comment lines (": ...", used for + * keepalives) are ignored. + */ +export interface SseMessage { + event: string | null; + data: string; +} + +export function createSseParser( + onMessage: (message: SseMessage) => void, +): (chunk: string) => void { + let buffer = ''; + let dataLines: string[] = []; + let eventName: string | null = null; + + const dispatch = (): void => { + if (dataLines.length === 0 && eventName === null) return; // stray blank line + onMessage({ event: eventName, data: dataLines.join('\n') }); + dataLines = []; + eventName = null; + }; + + return (chunk: string): void => { + buffer += chunk; + let newlineIndex: number; + while ((newlineIndex = buffer.indexOf('\n')) !== -1) { + let line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + if (line.endsWith('\r')) line = line.slice(0, -1); // tolerate CRLF + + if (line === '') { + dispatch(); + continue; + } + if (line.startsWith(':')) continue; // comment / keepalive + + const colon = line.indexOf(':'); + const field = colon === -1 ? line : line.slice(0, colon); + let value = colon === -1 ? '' : line.slice(colon + 1); + if (value.startsWith(' ')) value = value.slice(1); // SSE strips one leading space + + if (field === 'event') eventName = value; + else if (field === 'data') dataLines.push(value); + // 'id' / 'retry' are unused by this app. + } + }; +} diff --git a/frontend/src/app/utils/sse.vitest.ts b/frontend/src/app/utils/sse.vitest.ts new file mode 100644 index 0000000..4d6bf12 --- /dev/null +++ b/frontend/src/app/utils/sse.vitest.ts @@ -0,0 +1,50 @@ +import { describe, it, expect } from 'vitest'; +import { createSseParser, SseMessage } from './sse'; + +function collect(): { feed: (c: string) => void; messages: SseMessage[] } { + const messages: SseMessage[] = []; + return { feed: createSseParser((m) => messages.push(m)), messages }; +} + +describe('createSseParser', () => { + it('parses a complete event/data frame', () => { + const { feed, messages } = collect(); + feed('event: revision\ndata: {"revision": 3}\n\n'); + expect(messages).toEqual([{ event: 'revision', data: '{"revision": 3}' }]); + }); + + it('buffers across chunk boundaries that split a frame', () => { + const { feed, messages } = collect(); + feed('event: revis'); + feed('ion\ndata: {"revisi'); + feed('on": 9}\n\n'); + expect(messages).toEqual([{ event: 'revision', data: '{"revision": 9}' }]); + }); + + it('emits one message per frame for back-to-back events', () => { + const { feed, messages } = collect(); + feed('event: revision\ndata: {"revision": 1}\n\nevent: revision\ndata: {"revision": 2}\n\n'); + expect(messages.map((m) => m.data)).toEqual(['{"revision": 1}', '{"revision": 2}']); + }); + + it('ignores keepalive comment lines', () => { + const { feed, messages } = collect(); + feed(': keepalive\n\n'); + feed('data: {"revision": 4}\n\n'); + expect(messages).toEqual([{ event: null, data: '{"revision": 4}' }]); + }); + + it('tolerates CRLF line endings', () => { + const { feed, messages } = collect(); + feed('event: revision\r\ndata: {"revision": 5}\r\n\r\n'); + expect(messages).toEqual([{ event: 'revision', data: '{"revision": 5}' }]); + }); + + it('does not emit until a frame is terminated by a blank line', () => { + const { feed, messages } = collect(); + feed('data: {"revision": 6}\n'); + expect(messages).toEqual([]); + feed('\n'); + expect(messages).toEqual([{ event: null, data: '{"revision": 6}' }]); + }); +});