Compare commits

...

2 commits

Author SHA1 Message Date
948b49bb49 Fix scrollbars and enter to submit
All checks were successful
CI / Frontend lint (push) Successful in 24s
CI / Frontend unit tests (push) Successful in 24s
CI / Backend tests (push) Successful in 25s
CI / Frontend build (push) Successful in 24s
Docker / build-and-push (push) Successful in 1m6s
CI / Playwright e2e (push) Successful in 51s
2026-06-05 11:37:54 +01:00
688bc0cfe9 Add SSE updates 2026-06-04 22:12:10 +01:00
17 changed files with 1101 additions and 50 deletions

View file

@ -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": <current> }`. 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

View file

@ -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",
},
)

View file

@ -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, ()))

View file

@ -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;

View file

@ -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):

View file

@ -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

View file

@ -0,0 +1,109 @@
import { test, expect } from '@playwright/test';
import { randomUUID } from 'node:crypto';
/**
* Regression guard for the tasks accordion ("todos") double-scrollbar bug.
*
* When the pending-task list is tall it must show EXACTLY ONE scrollbar (inside
* the white card), not two. The cause was two nested scroll containers both
* firing: the `lt-tasks` host (overflow:auto) AND the inner `.container` card
* (overflow-y:auto). The fix makes the host a height-bounding flex column that
* only CLIPS, leaving the inner card as the sole scroller.
*
* Seeds many pending tasks via a direct PUT (far more robust than driving the
* carousel ~18 times), then reloads so the store renders the seeded tree. All
* ids are generated here in Node `crypto.randomUUID()` throws in the page
* because the dev origin is plain HTTP (not a secure context).
*/
test('tasks accordion with many tasks shows a single scrollbar', async ({ page }) => {
await page.goto('/');
// init() mints + registers a token on load; wait for it to land.
await page.waitForFunction(() => !!localStorage.getItem('life-towers.token.v4'), null, {
timeout: 15000,
});
// Build a tree: one page, one tower, many PENDING (is_done:false) tasks.
const now = Math.floor(Date.now() / 1000);
const tree = {
pages: [
{
id: randomUUID(),
name: 'Hobbies',
hide_create_tower_button: false,
keep_tasks_open: false,
default_date_from: null,
default_date_to: null,
towers: [
{
id: randomUUID(),
name: 'Reading',
base_color: { h: 0.92, s: 0.7, l: 0.55 },
blocks: Array.from({ length: 18 }, (_, i) => ({
id: randomUUID(),
tag: 'novel',
description: `Pending task ${i + 1} — read another chapter tonight`,
is_done: false,
difficulty: 1,
created_at: now - i * 3600,
})),
},
],
},
],
};
// PUT the tree (unguarded — no If-Match), then reload to render it.
const status = await page.evaluate(async (body) => {
const res = await fetch('api/v1/data', {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${localStorage.getItem('life-towers.token.v4')}`,
},
body: JSON.stringify(body),
});
return res.status;
}, tree);
expect(status).toBe(200);
await page.reload();
await page.waitForSelector('lt-tower', { timeout: 15000 });
// Open the accordion (collapsed by default since keep_tasks_open is false).
await page.locator('lt-tasks .header').first().click();
await page.waitForTimeout(400); // expand animation (200ms) + buffer
// Measure the scroll topology in the accordion subtree.
const m = await page.locator('lt-tasks').first().evaluate((host) => {
const card = host.querySelector('.container') as HTMLElement;
const cs = (el: Element) => getComputedStyle(el);
const overflows = (el: HTMLElement) => el.scrollHeight > el.clientHeight + 1;
return {
hostOverflowY: cs(host).overflowY,
cardOverflowY: cs(card).overflowY,
cardOverflows: overflows(card),
hostHasOwnOverflow: host.scrollHeight > host.clientHeight + 1,
hostHeight: host.getBoundingClientRect().height,
cardHeight: card.getBoundingClientRect().height,
viewport30vh: window.innerHeight * 0.3,
};
});
// The list must actually overflow, otherwise the test proves nothing.
expect(m.cardOverflows).toBe(true);
// Exactly one scroller: the inner card. The host only clips.
expect(m.hostOverflowY).toBe('hidden');
expect(m.cardOverflowY).toBe('auto');
// The host has no overflowing content of its own → no second scrollbar.
expect(m.hostHasOwnOverflow).toBe(false);
// The card shrank to fit within the host's bound (not clipped past it).
expect(m.cardHeight).toBeLessThanOrEqual(m.hostHeight + 1);
// Host height is capped by min(30vh, 45%) → at most ~30vh.
expect(m.hostHeight).toBeLessThanOrEqual(m.viewport30vh + 2);
await page.locator('lt-tasks .container').first().screenshot({
path: 'visuals/04d-tasks-accordion-overflow-single-scrollbar.png',
});
});

View file

@ -187,6 +187,7 @@ export function createDoneValue(defaultDone: boolean, currentDone: boolean, edit
maxlength="10000"
[value]="newValue().description"
(input)="updateNewDescription($any($event.target).value)"
(keydown.enter)="onNewDescriptionEnter($event)"
></textarea>
<label class="done-checkbox">
@ -414,6 +415,13 @@ export function createDoneValue(defaultDone: boolean, currentDone: boolean, edit
}
}
textarea {
// The global reset (styles.scss) zeroes padding, so the focus outline
// hugs the text. Re-pad so the outline clears the description text.
// box-sizing: border-box (forms.scss) keeps the outer size unchanged.
padding: 6px 8px;
}
.done-checkbox {
@include medium-text();
display: flex;
@ -779,6 +787,18 @@ export class BlockEditComponent implements AfterViewInit {
this.newValue.update((v) => ({ ...v, difficulty: clampDifficulty(v.difficulty + delta) }));
}
/**
* Bare Enter in the create-card description submits the new task and exits.
* Angular's `keydown.enter` pseudo-event matches *only* unmodified Enter, so
* Ctrl+Enter / Shift+Enter never reach here they fall through to the
* textarea's default behaviour and insert a newline.
*/
onNewDescriptionEnter(event: Event): void {
event.preventDefault();
event.stopPropagation();
this.submitNew();
}
submitNew(): void {
const v = this.newValue();
if (!v.tag) return;

View file

@ -105,7 +105,12 @@ export function taskListMaxHeight(expanded: boolean): string {
padding: calc(var(--small-padding) / 2);
margin: calc(var(--small-padding) / 2);
max-height: 30vh;
// Height is bounded by the host (lt-tasks) flex column, which clips but
// does not scroll. As the sole scroller, this card shrinks to that
// bound (min-height: 0) and scrolls a tall list inside itself — one
// scrollbar, sitting within the white card.
flex: 0 1 auto;
min-height: 0;
overflow-y: auto;
.header {

View file

@ -339,19 +339,20 @@ export function selectVisibleStyledBlocks(
flex: 0 1 auto;
min-height: 56px;
max-height: min(30vh, 45%);
overflow: auto;
display: block;
// The host only bounds the accordion's height and CLIPS — it must
// not scroll. Scrolling lives solely on the inner card
// (tasks.component .container), so a tall task list shows ONE
// scrollbar (inside the card), not two. Flex column + the card's
// min-height: 0 lets the card shrink to this bound and scroll.
overflow: hidden;
display: flex;
flex-direction: column;
width: 100%;
@media (max-width: $mobile-width) {
min-height: 44px;
max-height: min(25vh, 45%);
}
.container {
max-height: 100%;
overflow-y: auto;
}
}
.stack-zone {

View file

@ -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'

View file

@ -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<TreeDto> {
getData(token: string): Promise<DataResponse> {
return firstValueFrom(
this.http.get<TreeDto>('api/v1/data', { headers: this.authHeaders(token) }),
this.http.get<DataResponse>('api/v1/data', { headers: this.authHeaders(token) }),
);
}
async putData(token: string, tree: TreeDto): Promise<void> {
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<number> {
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<void> {
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 {

View file

@ -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);
});
});

View file

@ -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<typeof setTimeout> | 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<void> {
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<void> {
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);
}

View file

@ -51,17 +51,35 @@ interface MockApi {
getData: ReturnType<typeof vi.fn>;
putData: ReturnType<typeof vi.fn>;
health: ReturnType<typeof vi.fn>;
openEventStream: ReturnType<typeof vi.fn>;
}
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<void> {
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,
);
});
});

View file

@ -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.
}
};
}

View file

@ -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}' }]);
});
});