From 0e766ce654151b1c010a38cc2412a4869a256a5c Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 20 Jun 2026 17:03:36 -0600 Subject: [PATCH] feat(frontend): useTailStream lifecycle hook with backoff + stall detection --- src/hooks/useTailStream.test.ts | 403 ++++++++++++++++++++++++++++++++ src/hooks/useTailStream.ts | 237 +++++++++++++++++++ 2 files changed, 640 insertions(+) create mode 100644 src/hooks/useTailStream.test.ts create mode 100644 src/hooks/useTailStream.ts diff --git a/src/hooks/useTailStream.test.ts b/src/hooks/useTailStream.test.ts new file mode 100644 index 0000000..5247e4b --- /dev/null +++ b/src/hooks/useTailStream.test.ts @@ -0,0 +1,403 @@ +// @vitest-environment happy-dom +// React's `act` warnings need an act-aware environment; mirror the other +// hook tests in this repo (`useClaimDetail`, `useReconciliation`, ...). +(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + +import React, { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { + afterEach, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; +import { useTailStream } from "./useTailStream"; +import { useTailStore } from "@/store/tail-store"; +import type { Claim } from "@/types"; +import type { TailEvent } from "@/lib/tail-stream"; + +// --------------------------------------------------------------------------- +// Mock `streamTail` so we can drive the async iterator from the test. The +// hook calls `streamTail(resource, { signal })` once per connection attempt; +// each call must return an async iterator we control. We replace it with a +// factory that records every generator we hand out so the test can push +// events / errors / close on demand. +// --------------------------------------------------------------------------- + +vi.mock("@/lib/tail-stream", () => ({ + streamTail: vi.fn(), +})); + +import { streamTail } from "@/lib/tail-stream"; +const mockStreamTail = vi.mocked(streamTail); + +/** + * One controllable async iterator. `push(ev)` resolves the next pending + * `next()` with `ev`. `failWith(err)` rejects the next pending `next()`. + * `close()` resolves any pending `next()` with `done: true` so the + * `for await` loop exits cleanly (this is how a server-side EOF looks to + * the hook — the spec says that triggers a reconnect). + * + * If `next()` is called when no event is queued and the iterator isn't + * closed, we suspend on a promise — exactly mirroring `ReadableStream`'s + * behaviour and letting the test decide when each event arrives. + */ +function makeCtrl() { + const queue: TailEvent[] = []; + let resolveNext: ((v: IteratorResult) => void) | null = null; + let rejectNext: ((e: unknown) => void) | null = null; + let done = false; + let pendingError: unknown = null; + + const next = (): Promise> => { + if (pendingError !== null) { + const e = pendingError; + pendingError = null; + return Promise.reject(e); + } + if (queue.length > 0) { + return Promise.resolve({ value: queue.shift() as TailEvent, done: false }); + } + if (done) { + return Promise.resolve({ value: undefined as unknown as TailEvent, done: true }); + } + return new Promise>((resolve, reject) => { + resolveNext = resolve; + rejectNext = reject; + }); + }; + + const iter: AsyncIterableIterator = { + next, + return: () => { + done = true; + if (resolveNext) { + const r = resolveNext; + resolveNext = null; + rejectNext = null; + r({ value: undefined as unknown as TailEvent, done: true }); + } + return Promise.resolve({ value: undefined as unknown as TailEvent, done: true }); + }, + throw: (err: unknown) => { + done = true; + if (rejectNext) { + const r = rejectNext; + resolveNext = null; + rejectNext = null; + r(err); + } + return Promise.reject(err); + }, + [Symbol.asyncIterator]() { + return this; + }, + }; + + return { + iter, + push(ev: TailEvent): void { + if (resolveNext) { + const r = resolveNext; + resolveNext = null; + rejectNext = null; + r({ value: ev, done: false }); + } else { + queue.push(ev); + } + }, + close(): void { + done = true; + if (resolveNext) { + const r = resolveNext; + resolveNext = null; + rejectNext = null; + r({ value: undefined as unknown as TailEvent, done: true }); + } + }, + failWith(err: unknown): void { + if (rejectNext) { + const r = rejectNext; + resolveNext = null; + rejectNext = null; + r(err); + } else { + pendingError = err; + } + }, + }; +} + +type Ctrl = ReturnType; + +/** Configure `mockStreamTail` so every call returns a fresh controllable iterator. */ +function trackCalls(): { ctrls: Ctrl[] } { + const ctrls: Ctrl[] = []; + mockStreamTail.mockImplementation(() => { + const c = makeCtrl(); + ctrls.push(c); + return c.iter; + }); + return { ctrls }; +} + +/** Same `renderHook` shim used by `useClaimDetail`, `useReconciliation`, etc. */ +function renderHook(setup: () => TResult): { + result: { current: TResult | undefined }; + unmount: () => void; +} { + const result: { current: TResult | undefined } = { current: undefined }; + const container = document.createElement("div"); + document.body.appendChild(container); + + function Probe() { + result.current = setup(); + return null; + } + + const root: Root = createRoot(container); + act(() => { + root.render(React.createElement(Probe)); + }); + + return { + result, + unmount: () => { + act(() => root.unmount()); + container.remove(); + }, + }; +} + +/** Flush microtasks + React state until `predicate()` holds (or we time out). */ +async function waitFor( + predicate: () => boolean, + timeoutMs = 1000, +): Promise { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > timeoutMs) { + throw new Error( + `waitFor: predicate did not hold within ${timeoutMs}ms`, + ); + } + await act(async () => { + await Promise.resolve(); + }); + } +} + +/** Build a valid Claim shape for the dispatch test. */ +function makeClaim(id: string, patientName: string): Claim { + return { + id, + patientName, + providerNpi: "1234567890", + payerName: "Medicaid", + cptCode: "99213", + billedAmount: 100, + receivedAmount: 0, + status: "submitted", + submissionDate: "2026-06-20T00:00:00Z", + }; +} + +describe("useTailStream", () => { + beforeEach(() => { + mockStreamTail.mockReset(); + // Reset the singleton tail-store between tests so each case sees a + // clean claims/remittances/activity slate. + useTailStore.getState().reset("claims"); + useTailStore.getState().reset("remittances"); + useTailStore.getState().reset("activity"); + }); + + afterEach(() => { + // Restore real timers so a fake-timers-using test doesn't leak into + // the next case. + vi.useRealTimers(); + }); + + it("test_on_mount_status_connecting_then_live_after_snapshot_end", async () => { + const { ctrls } = trackCalls(); + const { result, unmount } = renderHook(() => useTailStream("claims")); + + // Wait for the effect to call streamTail once. + await waitFor(() => ctrls.length === 1); + // Initial status: connecting. + expect(result.current?.status).toBe("connecting"); + + // Server finishes replaying the snapshot — status flips to live. + ctrls[0].push({ type: "snapshot_end", data: { count: 0 } }); + await waitFor(() => result.current?.status === "live"); + expect(result.current?.status).toBe("live"); + + unmount(); + }); + + it("test_on_stream_error_status_error", async () => { + const { ctrls } = trackCalls(); + const { result, unmount } = renderHook(() => useTailStream("claims")); + + await waitFor(() => ctrls.length === 1); + + // Simulate a stream-level error event (yielded, not thrown). The hook + // should turn this into an Error and surface `status === "error"`. + ctrls[0].failWith(new Error("boom")); + + await waitFor(() => result.current?.status === "error"); + expect(result.current?.status).toBe("error"); + expect(result.current?.error).toBeInstanceOf(Error); + expect((result.current?.error as Error).message).toBe("boom"); + + unmount(); + }); + + it("test_on_abort_status_closed_no_reconnect", async () => { + const { ctrls } = trackCalls(); + const { unmount } = renderHook(() => useTailStream("claims")); + + await waitFor(() => ctrls.length === 1); + + // Capture the AbortSignal the hook passed to streamTail — the contract + // is that the hook aborts it on unmount. + const call = mockStreamTail.mock.calls[0]; + expect(call).toBeDefined(); + const opts = call[1] as { signal?: AbortSignal } | undefined; + expect(opts?.signal).toBeDefined(); + expect(opts?.signal?.aborted).toBe(false); + + unmount(); + + // After unmount the signal MUST be aborted — that's how `streamTail` + // tears down its fetch loop, and it's the externally-observable + // evidence that we cleaned up. (`status === "closed"` is set on the + // hook's state, but a renderHook shim can't observe post-unmount + // re-renders, so we assert on the abort instead.) + expect(opts?.signal?.aborted).toBe(true); + + // No reconnect must be scheduled. Advance past the entire backoff + // window and confirm `streamTail` is still at exactly one call. + vi.useFakeTimers(); + try { + await act(async () => { + vi.advanceTimersByTime(60_000); + }); + expect(mockStreamTail).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it("test_on_event_dispatches_to_tail_store", async () => { + const { ctrls } = trackCalls(); + const { unmount } = renderHook(() => useTailStream("claims")); + + await waitFor(() => ctrls.length === 1); + + // Bring the stream to "live" so the dispatcher is in the happy path. + ctrls[0].push({ type: "snapshot_end", data: { count: 0 } }); + await waitFor(() => useTailStore.getState().claims !== undefined); + + // Push an item — it must end up in the claims slice. + ctrls[0].push({ + type: "item", + data: makeClaim("CLM-1", "Patient One"), + }); + await waitFor( + () => useTailStore.getState().claims["CLM-1"] !== undefined, + ); + + const stored = useTailStore.getState().claims["CLM-1"]; + expect(stored).toBeDefined(); + expect(stored?.patientName).toBe("Patient One"); + + unmount(); + }); + + it("test_reconnect_status_cycles_reconnecting_to_connecting_to_live", async () => { + const { ctrls } = trackCalls(); + + // Compress the backoff schedule so the test doesn't sit on a real + // 1-second timer. We override the global setTimeout / clearTimeout to + // schedule timers via `setImmediate`-style micro-tasks instead of + // wall-clock — but the simpler path is just `vi.useFakeTimers` and + // manually advance. We go with the simpler path. + vi.useFakeTimers(); + try { + const { result, unmount } = renderHook(() => useTailStream("claims")); + await act(async () => { + // Yield so the initial useEffect runs and calls streamTail. + await Promise.resolve(); + }); + expect(ctrls.length).toBe(1); + + // First attempt fails — hook sets status=error and schedules a + // reconnect after the first backoff step (1000ms in the real + // schedule, but we advance just past it). + ctrls[0].failWith(new Error("first failed")); + await act(async () => { + await Promise.resolve(); + }); + expect(result.current?.status).toBe("error"); + + // Advance past the first backoff step. The hook should reopen — + // status becomes "connecting" (attempt counter resets on success + // but only after snapshot_end; while retrying it's still + // "reconnecting" until the new attempt actually opens). + await act(async () => { + vi.advanceTimersByTime(1500); + await Promise.resolve(); + }); + expect(ctrls.length).toBe(2); + + // Second attempt completes the snapshot — status flips to live. + ctrls[1].push({ type: "snapshot_end", data: { count: 0 } }); + await waitFor(() => result.current?.status === "live"); + expect(result.current?.status).toBe("live"); + + unmount(); + } finally { + vi.useRealTimers(); + } + }); + + it("test_reconnect_dedup_duplicate_items_appear_once_in_store", async () => { + const { ctrls } = trackCalls(); + const { unmount } = renderHook(() => useTailStream("claims")); + + await waitFor(() => ctrls.length === 1); + ctrls[0].push({ type: "snapshot_end", data: { count: 0 } }); + await waitFor(() => useTailStore.getState().claims !== undefined); + + // Push the same claim twice (mimicking a snapshot replay on + // reconnect). The store must keep exactly one copy. + ctrls[0].push({ + type: "item", + data: makeClaim("CLM-DUP", "Original"), + }); + ctrls[0].push({ + type: "item", + data: makeClaim("CLM-DUP", "Updated"), + }); + + await waitFor( + () => useTailStore.getState().claims["CLM-DUP"] !== undefined, + ); + // Give the second push time to be processed — we want to assert it + // was rejected by the store's first-write-wins rule. + await act(async () => { + await Promise.resolve(); + await Promise.resolve(); + }); + + expect(Object.keys(useTailStore.getState().claims)).toHaveLength(1); + expect(useTailStore.getState().claims["CLM-DUP"]?.patientName).toBe( + "Original", + ); + + unmount(); + }); +}); diff --git a/src/hooks/useTailStream.ts b/src/hooks/useTailStream.ts new file mode 100644 index 0000000..5394c22 --- /dev/null +++ b/src/hooks/useTailStream.ts @@ -0,0 +1,237 @@ +// --------------------------------------------------------------------------- +// Live-tail connection lifecycle hook (sub-project 5, Phase 5 Task 19). +// +// Opens a `streamTail(resource)` connection, dispatches `item` events into +// the matching `useTailStore` slice, exposes the connection status to the +// page (for ``), and survives transient backend failures +// with an exponential-backoff retry (1s → 2s → 4s → 8s → 16s, capped at +// 30s — spec §3.4). A stall detector flips status to `stalled` after 30s +// of silence (no event, including heartbeats) so the UI can show a stale +// connection without polling. +// +// Design notes: +// - One effect per `(resource, reconnectNonce)` pair. Bumping +// `reconnectNonce` from `forceReconnect()` re-runs the effect, which +// aborts the in-flight stream (cleanup) and opens a fresh one. +// - The AbortController is stored in a ref so `forceReconnect` can abort +// even from outside React's render cycle. +// - Backoff is indexed by an `attempt` counter that resets to 0 once the +// server completes a snapshot (i.e. we're "live"). A clean server-side +// EOF is treated like an error — we reconnect with backoff. +// - `setStatus("closed")` fires from the effect cleanup; the consumer +// (e.g. ``) typically unmounts at the same time, so +// this update is rarely observed — but it's there for any parent that +// keeps the consumer rendered after the hook is detached. +// --------------------------------------------------------------------------- + +import { useCallback, useEffect, useRef, useState } from "react"; +import { streamTail, type TailResource } from "@/lib/tail-stream"; +import { useTailStore } from "@/store/tail-store"; +import type { Activity, Claim, Remittance } from "@/types"; + +export type TailStatus = + | "connecting" + | "live" + | "reconnecting" + | "closed" + | "stalled" + | "error"; + +export interface UseTailStreamResult { + status: TailStatus; + lastEventAt: Date | null; + error: Error | null; + forceReconnect: () => void; +} + +/** Backoff schedule per spec §3.4: 1s, 2s, 4s, 8s, 16s, then cap at 30s. */ +const BACKOFF_STEPS_MS: readonly number[] = [ + 1_000, 2_000, 4_000, 8_000, 16_000, 30_000, +]; + +/** No event (including heartbeat) for this long → flip to `stalled`. */ +const STALL_TIMEOUT_MS = 30_000; + +function backoffDelayMs(attempt: number): number { + const i = Math.min(Math.max(attempt, 0), BACKOFF_STEPS_MS.length - 1); + return BACKOFF_STEPS_MS[i] as number; +} + +/** + * Dispatch a single item event into the matching slice of the tail store. + * The store slices are typed with the canonical shapes (`Claim`, + * `Remittance`, `Activity`); the stream yields `unknown` so we cast here. + */ +function dispatch(resource: TailResource, data: unknown): void { + const store = useTailStore.getState(); + switch (resource) { + case "claims": + store.addClaim(data as Claim); + break; + case "remittances": + store.addRemittance(data as Remittance); + break; + case "activity": + store.addActivity(data as Activity); + break; + } +} + +export function useTailStream(resource: TailResource): UseTailStreamResult { + const [status, setStatus] = useState("connecting"); + const [lastEventAt, setLastEventAt] = useState(null); + const [error, setError] = useState(null); + + /** + * Bumping this state causes the effect to re-run, aborting the current + * stream and starting a fresh one. `forceReconnect` is the only place + * we mutate it from outside the effect itself. + */ + const [reconnectNonce, setReconnectNonce] = useState(0); + + /** + * Hold the in-flight AbortController so `forceReconnect` can tear it + * down even when called from a stale render. The effect cleanup also + * aborts via this ref. + */ + const abortRef = useRef(null); + + const forceReconnect = useCallback(() => { + setReconnectNonce((n) => n + 1); + }, []); + + useEffect(() => { + let cancelled = false; + let attempt = 0; + let stallTimer: ReturnType | null = null; + let backoffTimer: ReturnType | null = null; + + const clearStall = (): void => { + if (stallTimer) { + clearTimeout(stallTimer); + stallTimer = null; + } + }; + + /** + * Re-arm the stall detector. Called on every event (including + * heartbeats and `item_dropped` notices) so a quiet backend that + * still pings every <30s doesn't get flagged as stalled. The stall + * timer only flips status if we were already in a "still trying" + * state — once we're `closed` or `error`, a missed heartbeat + * shouldn't override that. + */ + const armStall = (): void => { + clearStall(); + stallTimer = setTimeout(() => { + if (cancelled) return; + setStatus((prev) => { + if ( + prev === "closed" || + prev === "reconnecting" || + prev === "error" || + prev === "stalled" + ) { + return prev; + } + return "stalled"; + }); + }, STALL_TIMEOUT_MS); + }; + + const scheduleReconnect = (): void => { + if (cancelled) return; + if (backoffTimer) { + clearTimeout(backoffTimer); + backoffTimer = null; + } + const delay = backoffDelayMs(attempt); + backoffTimer = setTimeout(() => { + if (cancelled) return; + backoffTimer = null; + attempt += 1; + void openOnce(); + }, delay); + }; + + const openOnce = async (): Promise => { + if (cancelled) return; + const controller = new AbortController(); + abortRef.current = controller; + // First attempt → "connecting". Subsequent retries → "reconnecting" + // so the user can see that we've lost the prior connection. + setStatus(attempt === 0 ? "connecting" : "reconnecting"); + + try { + const iter = streamTail(resource, { signal: controller.signal }); + // eslint-disable-next-line @typescript-eslint/no-unused-vars + for await (const ev of iter) { + if (cancelled) return; + setLastEventAt(new Date()); + armStall(); + + switch (ev.type) { + case "snapshot_end": + // Server finished replaying; we are now live. Reset the + // backoff counter so a transient blip doesn't penalize the + // next outage. + attempt = 0; + setStatus("live"); + break; + case "item": + dispatch(resource, ev.data); + break; + case "heartbeat": + case "item_dropped": + // No state change — the stall timer has already been + // re-armed. These exist purely so the hook knows the + // connection is alive. + break; + case "error": + // Yielded (not thrown) error event. Promote to a thrown + // Error so the catch block below runs the reconnect + // machinery. + throw new Error(ev.data.message); + } + } + // Stream finished cleanly (server-side EOF). Treat as a + // reconnect-trigger: if we're still mounted, schedule a retry. + if (!cancelled) { + setStatus("reconnecting"); + scheduleReconnect(); + } + } catch (err) { + if (cancelled) return; + // An abort during teardown isn't really an error — the stream + // is being torn down on purpose. Skip the reconnect. + if (controller.signal.aborted) return; + const e = err instanceof Error ? err : new Error(String(err)); + setError(e); + setStatus("error"); + scheduleReconnect(); + } + }; + + void openOnce(); + armStall(); + + return () => { + cancelled = true; + if (abortRef.current) { + abortRef.current.abort(); + abortRef.current = null; + } + if (backoffTimer) { + clearTimeout(backoffTimer); + backoffTimer = null; + } + clearStall(); + // Per spec: on unmount, status = closed. This is the only way a + // parent that keeps the consumer mounted (e.g. for a transition) + // can observe the closed state. + setStatus("closed"); + }; + }, [resource, reconnectNonce]); + + return { status, lastEventAt, error, forceReconnect }; +}