Task e57facc2: Undo and live SSE updates in the admin PWA
- api/events.ts: fetch-based reader for GET /api/v1/admin/events (admin JWT
in the Authorization header via apiFetch, never in the URL), an incremental
SSE frame parser that buffers frames split across chunks and drops malformed
frames, and bounded-backoff reconnect with a clean abort on stop.
- api/useTransitions.tsx: one stream per signed-in shell (TransitionsProvider
in App) fanned out to screens; no provider means no stream.
- Today: Undo posts /instances/{id}/uncomplete, disables every Undo while in
flight, shows a dismissible message on failure, and refetches the snapshot;
any transition refetches in place. Mark done stays inert.
- History: quest transitions refetch the active filter in place; rows stay
read-only.
- Tests for parsing, stream auth/teardown/reconnect, Undo, and live refetch.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { waitFor } from "@testing-library/react";
|
||||
import { EVENTS_PATH, SseParser, openTransitionStream, type TransitionEvent } from "./events";
|
||||
import { frame, testStream, type TestStream } from "./testStream";
|
||||
import { storeTokens } from "../auth/oidc";
|
||||
|
||||
const COMPLETED = {
|
||||
child_id: 1,
|
||||
child_name: "Declan",
|
||||
instance_id: 12,
|
||||
quest_title: "Make bed",
|
||||
window: "morning",
|
||||
due_date: "2026-09-23",
|
||||
due_time: null,
|
||||
occurred_at: "2026-09-23T11:04:00Z",
|
||||
};
|
||||
const DAY_COMPLETE = {
|
||||
child_id: 2,
|
||||
child_name: "Maeve",
|
||||
quests_due: 4,
|
||||
quests_completed: 4,
|
||||
occurred_at: "2026-09-23T12:00:00Z",
|
||||
};
|
||||
|
||||
describe("SseParser", () => {
|
||||
it("yields every frame from one chunk", () => {
|
||||
const parser = new SseParser();
|
||||
const events = parser.push(
|
||||
frame("nestquest_quest_completed", COMPLETED) +
|
||||
frame("nestquest_child_day_complete", DAY_COMPLETE),
|
||||
);
|
||||
expect(events).toEqual([
|
||||
{ type: "nestquest_quest_completed", payload: COMPLETED },
|
||||
{ type: "nestquest_child_day_complete", payload: DAY_COMPLETE },
|
||||
]);
|
||||
});
|
||||
|
||||
it("buffers a frame split across two chunks", () => {
|
||||
const parser = new SseParser();
|
||||
const text = frame("nestquest_quest_uncompleted", COMPLETED);
|
||||
const cut = text.indexOf("Make");
|
||||
expect(parser.push(text.slice(0, cut))).toEqual([]);
|
||||
expect(parser.push(text.slice(cut))).toEqual([
|
||||
{ type: "nestquest_quest_uncompleted", payload: COMPLETED },
|
||||
]);
|
||||
});
|
||||
|
||||
it("buffers a split inside the event name and the blank-line terminator", () => {
|
||||
const parser = new SseParser();
|
||||
const text = frame("nestquest_quest_missed", COMPLETED);
|
||||
expect(parser.push(text.slice(0, 12))).toEqual([]);
|
||||
expect(parser.push(text.slice(12, -1))).toEqual([]);
|
||||
expect(parser.push(text.slice(-1))).toEqual([
|
||||
{ type: "nestquest_quest_missed", payload: COMPLETED },
|
||||
]);
|
||||
});
|
||||
|
||||
it("treats a CRLF split across chunks as one line break", () => {
|
||||
const parser = new SseParser();
|
||||
const text = `event: nestquest_quest_completed\r\ndata: ${JSON.stringify(COMPLETED)}\r\n\r\n`;
|
||||
const cut = text.indexOf("\n"); // chunk ends on "\r"
|
||||
expect(parser.push(text.slice(0, cut))).toEqual([]);
|
||||
expect(parser.push(text.slice(cut))).toEqual([
|
||||
{ type: "nestquest_quest_completed", payload: COMPLETED },
|
||||
]);
|
||||
});
|
||||
|
||||
it("drops malformed, unknown and comment frames without throwing", () => {
|
||||
const parser = new SseParser();
|
||||
const events = parser.push(
|
||||
"event: nestquest_quest_completed\ndata: not-json\n\n" +
|
||||
"event: nestquest_quest_completed\ndata: [1,2]\n\n" +
|
||||
"event: something_else\ndata: {}\n\n" +
|
||||
": keep-alive\n\n" +
|
||||
"data: {}\n\n" +
|
||||
frame("nestquest_quest_completed", COMPLETED),
|
||||
);
|
||||
expect(events).toEqual([{ type: "nestquest_quest_completed", payload: COMPLETED }]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("openTransitionStream", () => {
|
||||
let fetchMock: ReturnType<typeof vi.fn>;
|
||||
let streams: TestStream[];
|
||||
let stop: (() => void) | null;
|
||||
|
||||
beforeEach(() => {
|
||||
sessionStorage.clear();
|
||||
storeTokens({ access_token: "at-123", expires_in: 600 });
|
||||
streams = [];
|
||||
stop = null;
|
||||
fetchMock = vi.fn(() => {
|
||||
const stream = testStream();
|
||||
streams.push(stream);
|
||||
return Promise.resolve(stream.response);
|
||||
});
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
stop?.();
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
it("sends the token in the Authorization header, never in the URL", async () => {
|
||||
stop = openTransitionStream({ onTransition: () => {} });
|
||||
await waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
|
||||
const [url, init] = fetchMock.mock.calls[0];
|
||||
expect(String(url)).toMatch(new RegExp(`${EVENTS_PATH}$`));
|
||||
expect(String(url)).not.toContain("at-123");
|
||||
expect(init.headers.Authorization).toBe("Bearer at-123");
|
||||
expect(init.headers.Accept).toBe("text/event-stream");
|
||||
});
|
||||
|
||||
it("delivers transitions from frames split across chunks", async () => {
|
||||
const seen: TransitionEvent[] = [];
|
||||
stop = openTransitionStream({ onTransition: (event) => seen.push(event) });
|
||||
await waitFor(() => expect(streams).toHaveLength(1));
|
||||
const text = frame("nestquest_quest_completed", COMPLETED);
|
||||
streams[0].push(text.slice(0, 20));
|
||||
streams[0].push(text.slice(20));
|
||||
await waitFor(() =>
|
||||
expect(seen).toEqual([{ type: "nestquest_quest_completed", payload: COMPLETED }]),
|
||||
);
|
||||
});
|
||||
|
||||
it("stop aborts the request and cancels the body", async () => {
|
||||
stop = openTransitionStream({ onTransition: () => {} });
|
||||
await waitFor(() => expect(streams).toHaveLength(1));
|
||||
const signal = fetchMock.mock.calls[0][1].signal as AbortSignal;
|
||||
stop();
|
||||
expect(signal.aborted).toBe(true);
|
||||
await waitFor(() => expect(streams[0].cancelled()).toBe(true));
|
||||
});
|
||||
|
||||
it("reconnects after the stream drops", async () => {
|
||||
const seen: TransitionEvent[] = [];
|
||||
stop = openTransitionStream({ onTransition: (event) => seen.push(event), minDelayMs: 5 });
|
||||
await waitFor(() => expect(streams).toHaveLength(1));
|
||||
streams[0].close();
|
||||
await waitFor(() => expect(streams).toHaveLength(2));
|
||||
streams[1].push(frame("nestquest_quest_missed", COMPLETED));
|
||||
await waitFor(() => expect(seen).toHaveLength(1));
|
||||
});
|
||||
|
||||
it("does not reconnect after stop, even with a reconnect pending", async () => {
|
||||
stop = openTransitionStream({ onTransition: () => {}, minDelayMs: 20 });
|
||||
await waitFor(() => expect(streams).toHaveLength(1));
|
||||
streams[0].close();
|
||||
stop();
|
||||
await new Promise((resolve) => setTimeout(resolve, 60));
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("gives up on a 403 refusal instead of retrying", async () => {
|
||||
fetchMock.mockImplementation(() => Promise.resolve(new Response("", { status: 403 })));
|
||||
stop = openTransitionStream({ onTransition: () => {}, minDelayMs: 5 });
|
||||
await waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
|
||||
await new Promise((resolve) => setTimeout(resolve, 40));
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,196 @@
|
||||
/**
|
||||
* Live transition stream: GET /api/v1/admin/events (text/event-stream).
|
||||
*
|
||||
* The admin JWT travels in the Authorization header (via apiFetch), so the
|
||||
* stream is read with fetch + a ReadableStream reader — EventSource cannot
|
||||
* set headers, and the token must never go into a URL.
|
||||
*/
|
||||
import { ApiForbiddenError, ApiUnauthorizedError, apiFetch } from "./client";
|
||||
|
||||
export const EVENTS_PATH = "/api/v1/admin/events";
|
||||
|
||||
/** The four documented transitions, as the API names them on the wire. */
|
||||
export const TRANSITION_TYPES = [
|
||||
"nestquest_quest_completed",
|
||||
"nestquest_quest_uncompleted",
|
||||
"nestquest_quest_missed",
|
||||
"nestquest_child_day_complete",
|
||||
] as const;
|
||||
|
||||
export type TransitionType = (typeof TRANSITION_TYPES)[number];
|
||||
|
||||
/** Quest transitions carry the instance; a day-complete carries the child's counts. */
|
||||
export interface QuestTransitionPayload {
|
||||
child_id: number;
|
||||
child_name: string | null;
|
||||
instance_id: number;
|
||||
quest_title: string | null;
|
||||
window: string;
|
||||
due_date: string;
|
||||
due_time: string | null;
|
||||
occurred_at: string;
|
||||
}
|
||||
|
||||
export interface ChildDayCompletePayload {
|
||||
child_id: number;
|
||||
child_name: string | null;
|
||||
quests_due: number;
|
||||
quests_completed: number;
|
||||
occurred_at: string;
|
||||
}
|
||||
|
||||
export type TransitionEvent =
|
||||
| {
|
||||
type:
|
||||
| "nestquest_quest_completed"
|
||||
| "nestquest_quest_uncompleted"
|
||||
| "nestquest_quest_missed";
|
||||
payload: QuestTransitionPayload;
|
||||
}
|
||||
| { type: "nestquest_child_day_complete"; payload: ChildDayCompletePayload };
|
||||
|
||||
function isTransitionType(value: string): value is TransitionType {
|
||||
return (TRANSITION_TYPES as readonly string[]).includes(value);
|
||||
}
|
||||
|
||||
/**
|
||||
* Incremental SSE frame parser. Feed it decoded text in whatever chunks the
|
||||
* network delivers; it buffers a partial line or frame until it completes.
|
||||
* Comments (keep-alives), unknown event types and frames whose data is not a
|
||||
* JSON object are dropped, never thrown.
|
||||
*/
|
||||
export class SseParser {
|
||||
private buffer = "";
|
||||
private eventType = "";
|
||||
private data: string[] = [];
|
||||
/** A chunk ended on "\r": a leading "\n" in the next chunk is the same line break. */
|
||||
private pendingCr = false;
|
||||
|
||||
push(text: string): TransitionEvent[] {
|
||||
if (this.pendingCr && text.startsWith("\n")) text = text.slice(1);
|
||||
this.pendingCr = text.endsWith("\r");
|
||||
this.buffer += text;
|
||||
const lines = this.buffer.split(/\r\n|\r|\n/);
|
||||
this.buffer = lines.pop() ?? "";
|
||||
const events: TransitionEvent[] = [];
|
||||
for (const line of lines) {
|
||||
const event = this.line(line);
|
||||
if (event) events.push(event);
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
private line(line: string): TransitionEvent | null {
|
||||
if (line === "") return this.dispatch();
|
||||
if (line.startsWith(":")) return null;
|
||||
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);
|
||||
if (field === "event") this.eventType = value;
|
||||
else if (field === "data") this.data.push(value);
|
||||
return null;
|
||||
}
|
||||
|
||||
private dispatch(): TransitionEvent | null {
|
||||
const type = this.eventType;
|
||||
const data = this.data.join("\n");
|
||||
this.eventType = "";
|
||||
this.data = [];
|
||||
if (!isTransitionType(type)) return null;
|
||||
let payload: unknown;
|
||||
try {
|
||||
payload = JSON.parse(data);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null;
|
||||
return { type, payload } as TransitionEvent;
|
||||
}
|
||||
}
|
||||
|
||||
export interface TransitionStreamOptions {
|
||||
onTransition: (event: TransitionEvent) => void;
|
||||
/** Reconnect delay after a drop: starts at `minDelayMs`, doubles up to `maxDelayMs`. */
|
||||
minDelayMs?: number;
|
||||
maxDelayMs?: number;
|
||||
}
|
||||
|
||||
function wait(ms: number, signal: AbortSignal): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(done, ms);
|
||||
signal.addEventListener("abort", done, { once: true });
|
||||
function done() {
|
||||
clearTimeout(timer);
|
||||
signal.removeEventListener("abort", done);
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Read one connection until the body ends (or the signal aborts). */
|
||||
async function readOnce(
|
||||
signal: AbortSignal,
|
||||
onTransition: (event: TransitionEvent) => void,
|
||||
): Promise<void> {
|
||||
const response = await apiFetch(EVENTS_PATH, {
|
||||
signal,
|
||||
headers: { Accept: "text/event-stream" },
|
||||
});
|
||||
if (!response.ok || !response.body) {
|
||||
throw new Error(`Event stream request failed (HTTP ${response.status}).`);
|
||||
}
|
||||
const reader = response.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
const parser = new SseParser();
|
||||
const cancel = () => void reader.cancel().catch(() => {});
|
||||
signal.addEventListener("abort", cancel, { once: true });
|
||||
try {
|
||||
for (;;) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done || signal.aborted) break;
|
||||
for (const event of parser.push(decoder.decode(value, { stream: true }))) {
|
||||
if (signal.aborted) break;
|
||||
onTransition(event);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
signal.removeEventListener("abort", cancel);
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to the admin transition stream, reconnecting with a bounded
|
||||
* backoff after a drop. Returns a stop function that aborts the open request
|
||||
* and any pending reconnect. A 401/403 stops the stream for good: apiFetch
|
||||
* has already started a login (401) or the account is refused (403).
|
||||
*/
|
||||
export function openTransitionStream({
|
||||
onTransition,
|
||||
minDelayMs = 1_000,
|
||||
maxDelayMs = 30_000,
|
||||
}: TransitionStreamOptions): () => void {
|
||||
const controller = new AbortController();
|
||||
const { signal } = controller;
|
||||
|
||||
void (async () => {
|
||||
let delay = minDelayMs;
|
||||
while (!signal.aborted) {
|
||||
try {
|
||||
await readOnce(signal, onTransition);
|
||||
// A connection that was accepted and then ended starts the backoff over.
|
||||
delay = minDelayMs;
|
||||
} catch (error: unknown) {
|
||||
if (signal.aborted) return;
|
||||
if (error instanceof ApiUnauthorizedError || error instanceof ApiForbiddenError) return;
|
||||
// Dropped or refused connection: fall through to the backoff.
|
||||
}
|
||||
if (signal.aborted) return;
|
||||
await wait(delay, signal);
|
||||
delay = Math.min(delay * 2, maxDelayMs);
|
||||
}
|
||||
})();
|
||||
|
||||
return () => controller.abort();
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* Typed client for the admin instance mutation route
|
||||
* (POST /api/v1/admin/instances/{id}/uncomplete).
|
||||
*/
|
||||
import { apiFetch } from "./client";
|
||||
import { readJson } from "./definitions";
|
||||
|
||||
export interface UncompleteResult {
|
||||
instance_id: number;
|
||||
/** The derived state after the reversal: "open", or "missed" when past due. */
|
||||
state: string;
|
||||
/** False when the instance was already open (a no-op). */
|
||||
appended: boolean;
|
||||
}
|
||||
|
||||
export function uncompletePath(instanceId: number): string {
|
||||
return `/api/v1/admin/instances/${instanceId}/uncomplete`;
|
||||
}
|
||||
|
||||
/** Reverse one completion; raises ApiRequestError with the API's detail on failure. */
|
||||
export async function uncompleteInstance(instanceId: number): Promise<UncompleteResult> {
|
||||
return readJson<UncompleteResult>(await apiFetch(uncompletePath(instanceId), { method: "POST" }));
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
/**
|
||||
* Test helper: a controllable text/event-stream Response whose body the test
|
||||
* feeds chunk by chunk (so a frame can be split across chunk boundaries).
|
||||
*/
|
||||
export interface TestStream {
|
||||
response: Response;
|
||||
push(text: string): void;
|
||||
close(): void;
|
||||
/** True once the reader cancelled the body (the stream was torn down). */
|
||||
cancelled(): boolean;
|
||||
}
|
||||
|
||||
export function testStream(): TestStream {
|
||||
const encoder = new TextEncoder();
|
||||
let controller!: ReadableStreamDefaultController<Uint8Array>;
|
||||
let cancelled = false;
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
start(c) {
|
||||
controller = c;
|
||||
},
|
||||
cancel() {
|
||||
cancelled = true;
|
||||
},
|
||||
});
|
||||
return {
|
||||
response: new Response(body, {
|
||||
status: 200,
|
||||
headers: { "Content-Type": "text/event-stream" },
|
||||
}),
|
||||
push: (text) => controller.enqueue(encoder.encode(text)),
|
||||
close: () => controller.close(),
|
||||
cancelled: () => cancelled,
|
||||
};
|
||||
}
|
||||
|
||||
/** One SSE frame exactly as api/sse.py writes it. */
|
||||
export function frame(type: string, payload: unknown): string {
|
||||
return `event: ${type}\ndata: ${JSON.stringify(payload)}\n\n`;
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
/**
|
||||
* One live transition stream per signed-in shell, fanned out to the screens.
|
||||
*
|
||||
* <TransitionsProvider> opens the stream on mount and aborts it on unmount;
|
||||
* useTransitions(handler) registers a screen's handler for as long as the
|
||||
* screen is mounted. Without a provider the hook does nothing, so a screen
|
||||
* rendered on its own (e.g. in a test) never opens a stream.
|
||||
*/
|
||||
import { createContext, useContext, useEffect, useRef, useState, type ReactNode } from "react";
|
||||
import { openTransitionStream, type TransitionEvent } from "./events";
|
||||
|
||||
type Listener = (event: TransitionEvent) => void;
|
||||
|
||||
interface TransitionHub {
|
||||
subscribe(listener: Listener): () => void;
|
||||
}
|
||||
|
||||
const TransitionsContext = createContext<TransitionHub | null>(null);
|
||||
|
||||
export function TransitionsProvider({ children }: { children: ReactNode }) {
|
||||
const [hub] = useState(() => {
|
||||
const listeners = new Set<Listener>();
|
||||
return {
|
||||
listeners,
|
||||
subscribe(listener: Listener) {
|
||||
listeners.add(listener);
|
||||
return () => {
|
||||
listeners.delete(listener);
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
useEffect(
|
||||
() =>
|
||||
openTransitionStream({
|
||||
onTransition: (event) => {
|
||||
for (const listener of [...hub.listeners]) {
|
||||
try {
|
||||
listener(event);
|
||||
} catch {
|
||||
// One screen's handler must not tear down the shared stream.
|
||||
}
|
||||
}
|
||||
},
|
||||
}),
|
||||
[hub],
|
||||
);
|
||||
|
||||
return <TransitionsContext.Provider value={hub}>{children}</TransitionsContext.Provider>;
|
||||
}
|
||||
|
||||
/** Call `handler` for every transition while the calling component is mounted. */
|
||||
export function useTransitions(handler: Listener): void {
|
||||
const hub = useContext(TransitionsContext);
|
||||
const handlerRef = useRef(handler);
|
||||
useEffect(() => {
|
||||
handlerRef.current = handler;
|
||||
});
|
||||
useEffect(() => hub?.subscribe((event) => handlerRef.current(event)), [hub]);
|
||||
}
|
||||
Reference in New Issue
Block a user