diff --git a/src/lib/tail-stream.test.ts b/src/lib/tail-stream.test.ts new file mode 100644 index 0000000..654b572 --- /dev/null +++ b/src/lib/tail-stream.test.ts @@ -0,0 +1,152 @@ +// @vitest-environment node +// `streamTail` pipes a `fetch()` response body through a `TextDecoderStream` +// and a manual NDJSON-line splitter. We mock `fetch` with `vi.stubGlobal` +// and drive a fake `ReadableStream` from the test, so no DOM/happy-dom +// is needed — Node 22's globals (Response, ReadableStream, TextDecoderStream) +// are sufficient. + +import { afterEach, describe, expect, it, vi } from "vitest"; +import { streamTail, type TailEvent } from "./tail-stream"; + +/** + * Build a controllable ReadableStream and a tiny driver. The + * stream's controller is captured at construction so the test can decide + * exactly when each NDJSON line is enqueued and when to close. The body + * is wrapped in a `Response` (matching what `fetch()` would return). + */ +function makeStream() { + const encoder = new TextEncoder(); + let controllerRef: ReadableStreamDefaultController | undefined; + const stream = new ReadableStream({ + start(c) { + controllerRef = c; + }, + }); + return { + response: new Response(stream, { + status: 200, + headers: { "content-type": "application/x-ndjson" }, + }), + push: (line: string) => { + controllerRef?.enqueue(encoder.encode(line + "\n")); + }, + close: () => controllerRef?.close(), + }; +} + +/** + * Drain an async iterable into an array. Useful for asserting the full + * emission list without manually awaiting each `.next()`. + */ +async function collect(iter: AsyncIterableIterator): Promise { + const out: T[] = []; + for await (const v of iter) out.push(v); + return out; +} + +describe("streamTail", () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("test_parses_well_formed_ndjson_into_typed_events", async () => { + const driver = makeStream(); + const fetchMock = vi.fn().mockResolvedValue(driver.response); + vi.stubGlobal("fetch", fetchMock); + + const gen = streamTail("claims"); + driver.push('{"type":"item","data":{"id":"CLM-1"}}'); + driver.push('{"type":"item","data":{"id":"CLM-2"}}'); + driver.push('{"type":"item","data":{"id":"CLM-3"}}'); + driver.close(); + + const first = await gen.next(); + const second = await gen.next(); + const third = await gen.next(); + const tail = await gen.next(); + + expect(first.done).toBe(false); + expect(second.done).toBe(false); + expect(third.done).toBe(false); + expect(tail.done).toBe(true); + expect([first.value, second.value, third.value]).toEqual([ + { type: "item", data: { id: "CLM-1" } }, + { type: "item", data: { id: "CLM-2" } }, + { type: "item", data: { id: "CLM-3" } }, + ]); + + // The fetch call itself must target /api/claims/stream with the + // NDJSON Accept header. + expect(fetchMock).toHaveBeenCalledTimes(1); + const [calledUrl, calledInit] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(calledUrl).toBe("/api/claims/stream"); + expect(calledInit.headers).toMatchObject({ Accept: "application/x-ndjson" }); + }); + + it("test_yields_typed_error_on_error_line", async () => { + const driver = makeStream(); + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response)); + const gen = streamTail("remittances"); + driver.push('{"type":"error","data":{"message":"x"}}'); + driver.close(); + const events = await collect(gen); + expect(events).toEqual([ + { type: "error", data: { message: "x" } }, + ]); + }); + + it("test_handles_heartbeat_silently", async () => { + // "Silently" here means: still yields the correctly-typed heartbeat + // (no special filtering) — the consumer (useTailStream) decides what + // to do with it. The lib just has to forward it. + const driver = makeStream(); + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response)); + const gen = streamTail("activity"); + driver.push('{"type":"heartbeat","data":{"ts":"2026-06-20T12:00:00Z"}}'); + driver.close(); + const events = await collect(gen); + expect(events).toEqual([ + { type: "heartbeat", data: { ts: "2026-06-20T12:00:00Z" } }, + ]); + }); + + it("test_abort_signal_cancels_mid_stream", async () => { + const driver = makeStream(); + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response)); + const ac = new AbortController(); + const gen = streamTail("claims", { signal: ac.signal }); + + // First event arrives normally. + driver.push('{"type":"item","data":{"id":"CLM-1"}}'); + const first = await gen.next(); + expect(first.done).toBe(false); + + // Abort before any more events arrive, then close the upstream so + // any pending read completes. The generator must exit cleanly + // (no throw) and not yield further. + ac.abort(); + driver.close(); + const second = await gen.next(); + expect(second.done).toBe(true); + }); + + it("test_malformed_line_skipped_next_valid_still_emitted", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const driver = makeStream(); + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(driver.response)); + const gen = streamTail("claims"); + driver.push('{"type":"item","data":{"id":"CLM-1"}}'); + driver.push("this is not json"); + driver.push('{"type":"item","data":{"id":"CLM-2"}}'); + driver.close(); + const events = await collect(gen); + expect(events).toEqual([ + { type: "item", data: { id: "CLM-1" } }, + { type: "item", data: { id: "CLM-2" } }, + ]); + // The malformed line must produce a console.warn so operators can spot + // a buggy backend without crashing the whole tail. + expect(warnSpy).toHaveBeenCalled(); + warnSpy.mockRestore(); + }); +}); diff --git a/src/lib/tail-stream.ts b/src/lib/tail-stream.ts new file mode 100644 index 0000000..6a8b3fc --- /dev/null +++ b/src/lib/tail-stream.ts @@ -0,0 +1,189 @@ +// --------------------------------------------------------------------------- +// Live-tail NDJSON stream parser (sub-project 5, Phase 4 Task 16). +// +// Opens `GET /api/{resource}/stream` with `Accept: application/x-ndjson`, +// pipes the response body through a `TextDecoderStream` + a manual +// newline-splitter, and yields one typed `TailEvent` per line. The +// higher-level `useTailStream` hook (Phase 5) is what subscribes to this +// generator and dispatches into `tail-store`; this module is intentionally +// pure parsing — no React, no Zustand, no fetch options other than the +// ones documented in the spec. +// --------------------------------------------------------------------------- + +/** + * One event yielded by `streamTail`. The data shapes mirror the backend's + * `GET /api/{resource}/stream` NDJSON contract (see `backend/api.py`). + * + * `item.data` is intentionally `unknown` — the consumer casts to the + * page-specific shape (Claim / Remittance / Activity). The other variants + * carry the data shape the spec mandates, so consumers get a friendly + * type for the non-item events. + */ +export type TailEvent = + | { type: "item"; data: unknown } + | { type: "snapshot_end"; data: { count: number } } + | { type: "heartbeat"; data: { ts: string } } + | { type: "item_dropped"; data: { id: string } } + | { type: "error"; data: { message: string } }; + +export type TailResource = "claims" | "remittances" | "activity"; + +export interface StreamTailOptions { + /** Forwarded to `fetch`; aborting the controller cleanly exits the generator. */ + signal?: AbortSignal; + /** Override the base URL (defaults to "" — i.e. same-origin). Used in tests. */ + baseUrl?: string; +} + +const KNOWN_TYPES: ReadonlySet = new Set([ + "item", + "snapshot_end", + "heartbeat", + "item_dropped", + "error", +]); + +/** + * Open a live-tail NDJSON stream and yield one typed event per line. + * + * The generator is single-use: call `gen.return()` (or break out of a + * `for await` loop) to close the underlying fetch. When `opts.signal` + * aborts, the generator exits cleanly without throwing. + * + * @example + * for await (const ev of streamTail("claims")) { + * if (ev.type === "item") console.log(ev.data); + * } + */ +export async function* streamTail( + resource: TailResource, + opts?: StreamTailOptions, +): AsyncIterableIterator { + const base = opts?.baseUrl ?? ""; + const url = `${base}/api/${resource}/stream`; + + let res: Response; + try { + res = await fetch(url, { + headers: { Accept: "application/x-ndjson" }, + signal: opts?.signal, + }); + } catch (err) { + // Aborting the signal makes fetch reject with a DOMException; the spec + // says "if signal aborts, exit cleanly (don't throw)", so we just + // return — the consumer sees a completed iterator. + if (opts?.signal?.aborted) return; + throw err; + } + + if (!res.ok) { + throw new Error(`stream open failed: ${res.status}`); + } + if (!res.body) { + throw new Error("stream has no body"); + } + + // TextDecoderStream handles the chunked UTF-8 boundary problem for us; + // after this, we have a stream of decoded strings. We then split each + // chunk on \n and JSON.parse each non-empty line. + const stringStream = res.body.pipeThrough(new TextDecoderStream()); + const reader = stringStream.getReader(); + + let buffer = ""; + try { + // eslint-disable-next-line no-constant-condition + while (true) { + // Check the abort signal between reads so a quick abort doesn't + // require us to wait for a network read to complete. + if (opts?.signal?.aborted) return; + + let chunk: string | undefined; + let done: boolean; + try { + const result = await reader.read(); + chunk = result.value; + done = result.done; + } catch (err) { + if (opts?.signal?.aborted) return; + throw err; + } + + if (done) break; + if (chunk) buffer += chunk; + + let nl = buffer.indexOf("\n"); + while (nl !== -1) { + const line = buffer.slice(0, nl).replace(/\r$/, ""); + buffer = buffer.slice(nl + 1); + if (line.length > 0) { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch (err) { + // Malformed lines are the backend's bug, not ours. Log and + // continue so a single bad frame doesn't kill the stream. + console.warn( + "tail-stream: malformed NDJSON line, skipping", + { line, err }, + ); + nl = buffer.indexOf("\n"); + continue; + } + + if ( + typeof parsed === "object" && + parsed !== null && + "type" in parsed && + typeof (parsed as { type: unknown }).type === "string" && + KNOWN_TYPES.has((parsed as { type: string }).type as TailEvent["type"]) + ) { + yield parsed as TailEvent; + } else { + // Unknown / wrong-shape line — log and skip. A future + // backend event type shouldn't crash the page. + console.warn( + "tail-stream: unknown event shape, skipping", + { line }, + ); + } + } + nl = buffer.indexOf("\n"); + } + } + + // Flush any trailing partial line (no terminating newline). + const tail = buffer.replace(/\r$/, ""); + if (tail.length > 0) { + try { + const parsed = JSON.parse(tail) as unknown; + if ( + typeof parsed === "object" && + parsed !== null && + "type" in parsed && + typeof (parsed as { type: unknown }).type === "string" && + KNOWN_TYPES.has((parsed as { type: string }).type as TailEvent["type"]) + ) { + yield parsed as TailEvent; + } else { + console.warn( + "tail-stream: unknown trailing event shape, skipping", + { line: tail }, + ); + } + } catch (err) { + console.warn( + "tail-stream: malformed trailing NDJSON line, skipping", + { line: tail, err }, + ); + } + } + } finally { + // Release the reader lock so the underlying connection can close + // when the test / consumer drops the iterator. + try { + reader.releaseLock(); + } catch { + // already released — ignore + } + } +}