From 11a4eaa480686ee058fb10c975e9519095f5a803 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 20 Jun 2026 16:02:25 -0600 Subject: [PATCH 1/2] feat(frontend): NDJSON streamTail parser for /api/{resource}/stream --- src/lib/tail-stream.test.ts | 152 +++++++++++++++++++++++++++++ src/lib/tail-stream.ts | 189 ++++++++++++++++++++++++++++++++++++ 2 files changed, 341 insertions(+) create mode 100644 src/lib/tail-stream.test.ts create mode 100644 src/lib/tail-stream.ts 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 + } + } +} From 64fb11a9cf9f29539fd44cd27ef8a2a9c2fe15b3 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 20 Jun 2026 16:05:21 -0600 Subject: [PATCH 2/2] feat(frontend): tail-store Zustand with FIFO cap 10k per slice --- src/store/tail-store.test.ts | 131 ++++++++++++++++++++++++++++ src/store/tail-store.ts | 161 +++++++++++++++++++++++++++++++++++ 2 files changed, 292 insertions(+) create mode 100644 src/store/tail-store.test.ts create mode 100644 src/store/tail-store.ts diff --git a/src/store/tail-store.test.ts b/src/store/tail-store.test.ts new file mode 100644 index 0000000..ca61566 --- /dev/null +++ b/src/store/tail-store.test.ts @@ -0,0 +1,131 @@ +// @vitest-environment node +// `useTailStore` is a pure zustand store with no DOM / React dependencies, +// so a node environment is fine. We import `getState()` and `setState` +// from the store directly (no React render) and assert on the resulting +// state shape. + +import { beforeEach, describe, expect, it } from "vitest"; +import type { Activity, Claim, Remittance } from "@/types"; +import { TAIL_CAP, useTailStore } from "./tail-store"; + +function makeClaim(id: string, overrides: Partial = {}): Claim { + return { + id, + patientName: `Patient ${id}`, + providerNpi: "1234567890", + payerName: "Medicaid", + cptCode: "99213", + billedAmount: 100, + receivedAmount: 0, + status: "submitted", + submissionDate: "2026-06-20T00:00:00Z", + ...overrides, + }; +} + +function makeRemittance(id: string, overrides: Partial = {}): Remittance { + return { + id, + claimId: `CLM-${id}`, + payerName: "Medicaid", + paidAmount: 100, + adjustmentAmount: 0, + receivedDate: "2026-06-20", + checkNumber: id, + status: "received", + ...overrides, + }; +} + +function makeActivity(id: string, overrides: Partial = {}): Activity { + return { + id, + kind: "claim_submitted", + message: `Event ${id}`, + timestamp: "2026-06-20T00:00:00Z", + ...overrides, + }; +} + +describe("useTailStore", () => { + // Each test starts from a known-empty state. The store is module-level + // (singleton), so we use `reset()` to clear between cases — that's also + // what production code calls on stream open. + beforeEach(() => { + useTailStore.getState().reset("claims"); + useTailStore.getState().reset("remittances"); + useTailStore.getState().reset("activity"); + }); + + it("test_add_claim_adds_new_claim_keyed_by_id", () => { + const { addClaim } = useTailStore.getState(); + addClaim(makeClaim("CLM-1")); + addClaim(makeClaim("CLM-2")); + const { claims } = useTailStore.getState(); + expect(Object.keys(claims)).toHaveLength(2); + expect(claims["CLM-1"]?.id).toBe("CLM-1"); + expect(claims["CLM-2"]?.id).toBe("CLM-2"); + }); + + it("test_add_claim_with_duplicate_id_is_noop", () => { + const { addClaim } = useTailStore.getState(); + addClaim(makeClaim("CLM-1", { patientName: "Original" })); + addClaim(makeClaim("CLM-1", { patientName: "Updated" })); + const { claims } = useTailStore.getState(); + expect(Object.keys(claims)).toHaveLength(1); + // First write wins; duplicates are silently dropped so the snapshot + // replay on reconnect doesn't trample the canonical row. + expect(claims["CLM-1"]?.patientName).toBe("Original"); + }); + + it("test_reset_claims_clears_only_claims_slice", () => { + const { addClaim, addRemittance, addActivity, reset } = useTailStore.getState(); + addClaim(makeClaim("CLM-1")); + addRemittance(makeRemittance("RMT-1")); + addActivity(makeActivity("ACT-1")); + + reset("claims"); + const s = useTailStore.getState(); + expect(Object.keys(s.claims)).toHaveLength(0); + expect(Object.keys(s.remittances)).toHaveLength(1); + expect(s.activity).toHaveLength(1); + }); + + it("test_add_activity_appends", () => { + const { addActivity } = useTailStore.getState(); + addActivity(makeActivity("ACT-1", { message: "first" })); + addActivity(makeActivity("ACT-2", { message: "second" })); + addActivity(makeActivity("ACT-3", { message: "third" })); + const { activity } = useTailStore.getState(); + expect(activity).toHaveLength(3); + // Activity has no stable id; it lives in an array. Insertion order is + // the only ordering signal. + expect(activity.map((a) => a.message)).toEqual(["first", "second", "third"]); + }); + + it("test_fifo_cap_evicts_oldest_when_over_10000", () => { + // Insert 5 over the cap so the eviction policy must actually run. + // The plan calls for evicting the oldest 100 (or enough to be back at + // the cap) — we just assert: size == TAIL_CAP and the very-first + // items are gone. + // + // 10 005 individual `addClaim` calls is intentionally a stress test + // of the eviction path: each call rebuilds the dict + order array + // so the wall-clock cost is O(N^2) in the number of items ever + // inserted. Bump the per-test timeout from the default 5s to 60s + // so this case can complete in CI on slow runners. + const N = TAIL_CAP + 5; + const { addClaim } = useTailStore.getState(); + for (let i = 0; i < N; i++) { + addClaim(makeClaim(`CLM-${i.toString().padStart(6, "0")}`)); + } + const { claims } = useTailStore.getState(); + expect(Object.keys(claims)).toHaveLength(TAIL_CAP); + // The five oldest (CLM-000000 .. CLM-000004) must be gone. + expect(claims["CLM-000000"]).toBeUndefined(); + expect(claims["CLM-000004"]).toBeUndefined(); + // The most recent (CLM-0010004) must be present. + const lastKey = `CLM-${(N - 1).toString().padStart(6, "0")}`; + expect(claims[lastKey]).toBeDefined(); + }, 60_000); +}); diff --git a/src/store/tail-store.ts b/src/store/tail-store.ts new file mode 100644 index 0000000..29de8f8 --- /dev/null +++ b/src/store/tail-store.ts @@ -0,0 +1,161 @@ +// --------------------------------------------------------------------------- +// Live-tail append-only store (sub-project 5, Phase 4 Task 17). +// +// Three independent slices (`claims` / `remittances` / `activity`) that +// `useTailStream` (Phase 5) writes into as `item` events arrive. Reads +// happen via `useMergedTail` (also Phase 5), which merges each slice with +// the page's JSON fetch results and applies the page's filter. +// +// Design notes: +// - `claims` and `remittances` are key-by-id maps (id is stable), so +// duplicate snapshots are dedup'd by `addClaim`/`addRemittance` (first +// write wins — the snapshot replay on reconnect must not trample the +// canonical row). +// - `activity` is an append-only array (event ids aren't guaranteed to +// be unique across snapshots; the ActivityLog renders in arrival +// order). +// - Each slice is FIFO-capped at `TAIL_CAP` (10 000) so a runaway tail +// can't grow the heap unbounded. Oldest entries are evicted in the +// add function itself — `reset` is what production calls when a new +// stream opens. +// --------------------------------------------------------------------------- + +import { create } from "zustand"; +import type { Activity, Claim, Remittance } from "@/types"; +import type { TailResource } from "@/lib/tail-stream"; + +/** Maximum number of items retained per slice before FIFO eviction kicks in. */ +export const TAIL_CAP = 10_000; + +/** + * Per-slice eviction batch. When the cap is exceeded, drop the oldest + * `EVICT_BATCH` entries in a single `set()` call. Picking a batch > 1 + * amortizes the cost of eviction when a high-volume stream pushes many + * items per frame — and 100 is small enough that a 10 005-item insert + * still lands within one `set()` call. + * + * The store does NOT require this batch to be exactly the overflow + * amount — it always drains down to the cap, so a +5 overflow evicts 5 + * even when EVICT_BATCH is 100. The batch is just an upper bound on how + * many we delete in a single pass. + */ +const EVICT_BATCH = 100; + +interface TailStore { + // --- Slices (keyed by id for the two that have stable ids) ----------- + claims: Record; + remittances: Record; + activity: Activity[]; + + // --- Insertion-order trackers (kept in sync with the dicts above) ---- + // These are private to the store; consumers only read the dicts. We + // store them as part of the state so they reactively update with the + // same `set()` call (zustand shallow-merges, so the new array ref is + // what triggers a re-render in subscribers that select `claims`). + claimOrder: string[]; + remitOrder: string[]; + + // --- Setters --------------------------------------------------------- + addClaim: (c: Claim) => void; + addRemittance: (r: RemitListItem) => void; + addActivity: (a: Activity) => void; + reset: (resource: TailResource) => void; +} + +/** + * The spec's sketch uses the placeholder name `RemitListItem` for the + * remittance shape; locally we just use the existing `Remittance` type + * from `@/types` (same value, no need to add a duplicate). + */ +type RemitListItem = Remittance; + +function evictOldest( + order: string[], + dict: Record, + batch: number, +): { order: string[]; dict: Record } { + if (order.length <= TAIL_CAP) return { order, dict }; + // Drain down to the cap; never evict more than `batch` per call so a + // 5-item overflow evicts 5, but a 1 000-item overflow evicts 100 in + // this pass and the remaining 900 in subsequent add() calls. + const toDrop = Math.min(batch, order.length - TAIL_CAP); + const dropped = order.slice(0, toDrop); + const nextOrder = order.slice(toDrop); + // Object.assign is consistently faster than `{ ...dict }` in V8 for + // large dicts; we follow up with `delete` for each evicted id. + const nextDict: Record = Object.assign({}, dict); + for (const id of dropped) delete nextDict[id]; + return { order: nextOrder, dict: nextDict }; +} + +export const useTailStore = create((set) => ({ + claims: {}, + remittances: {}, + activity: [], + claimOrder: [], + remitOrder: [], + + addClaim: (c) => + set((s) => { + // Dedup: first write wins. The snapshot replay on reconnect + // produces the same id repeatedly; we want the first occurrence + // to stick so the canonical row isn't overwritten by an older + // version that happened to be in the snapshot. + if (s.claims[c.id]) return s; + // Object.assign is faster than `{ ...s.claims, [c.id]: c }` for + // large dicts; this hot path is called once per `item` event. + const nextClaims: Record = Object.assign({}, s.claims, { + [c.id]: c, + }); + const nextOrder = s.claimOrder.concat(c.id); + if (nextOrder.length > TAIL_CAP) { + const { order, dict } = evictOldest(nextOrder, nextClaims, EVICT_BATCH); + return { claims: dict as Record, claimOrder: order }; + } + return { claims: nextClaims, claimOrder: nextOrder }; + }), + + addRemittance: (r) => + set((s) => { + if (s.remittances[r.id]) return s; + const nextRemits: Record = Object.assign( + {}, + s.remittances, + { [r.id]: r }, + ); + const nextOrder = s.remitOrder.concat(r.id); + if (nextOrder.length > TAIL_CAP) { + const { order, dict } = evictOldest(nextOrder, nextRemits, EVICT_BATCH); + return { + remittances: dict as Record, + remitOrder: order, + }; + } + return { remittances: nextRemits, remitOrder: nextOrder }; + }), + + addActivity: (a) => + set((s) => { + // Activity has no stable id, so it's a plain append. FIFO cap + // evicts the oldest with a single `slice` (the typical case is + // +1, +1, +1; an extreme burst falls back to multiple slice + // passes). + let next = [...s.activity, a]; + if (next.length > TAIL_CAP) { + next = next.slice(next.length - TAIL_CAP); + } + return { activity: next }; + }), + + reset: (resource) => + set(() => { + switch (resource) { + case "claims": + return { claims: {}, claimOrder: [] }; + case "remittances": + return { remittances: {}, remitOrder: [] }; + case "activity": + return { activity: [] }; + } + }), +}));