merge: frontend tail-stream and tail-store libs (Phase 4)

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<Uint8Array> 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<Uint8Array> | undefined;
+  const stream = new ReadableStream<Uint8Array>({
+    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<T>(iter: AsyncIterableIterator<T>): Promise<T[]> {
+  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<TailEvent[]>([
+      { 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<TailEvent[]>([
+      { 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<TailEvent[]>([
+      { 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<TailEvent[]>([
+      { 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();
+  });
+});

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<TailEvent["type"]> = 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<TailEvent> {
+  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
+    }
+  }
+}

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> = {}): 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> = {}): 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> = {}): 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);
+});

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<string, Claim>;
+  remittances: Record<string, Remittance>;
+  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<string, Claim | Remittance>,
+  batch: number,
+): { order: string[]; dict: Record<string, Claim | Remittance> } {
+  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<string, Claim | Remittance> = Object.assign({}, dict);
+  for (const id of dropped) delete nextDict[id];
+  return { order: nextOrder, dict: nextDict };
+}
+
+export const useTailStore = create<TailStore>((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<string, Claim> = 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<string, Claim>, claimOrder: order };
+      }
+      return { claims: nextClaims, claimOrder: nextOrder };
+    }),
+
+  addRemittance: (r) =>
+    set((s) => {
+      if (s.remittances[r.id]) return s;
+      const nextRemits: Record<string, Remittance> = 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<string, Remittance>,
+          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: [] };
+      }
+    }),
+}));