merge: SP5 live-tail — pub/sub + 3 stream endpoints + frontend tail integration
Brings the SP5 live-tail implementation into main:
Backend
* EventBus (cyclone.pubsub): per-kind fan-out with drop-oldest overflow
* FastAPI lifespan initializes the bus + db once per process
* store.add() publishes claim_written / remittance_written /
activity_recorded events on every batch write
* GET /api/{claims,remittances,activity}/stream: NDJSON snapshot +
live subscription + 15s idle heartbeat
* EventBus.unsubscribe() lets the tail loop release its queue on
client disconnect (no queue leak per open stream)
Frontend
* src/lib/tail-stream.ts: streamTail() async-generator over fetch
* src/store/tail-store.ts: zustand with FIFO cap 10k per slice
* src/hooks/useTailStream.ts: connecting/live/reconnecting/stalled/error/closed
state machine with 1→2→4→8→16→30s backoff
* src/hooks/useMergedTail.ts: base + tail merge with filter
* src/components/TailStatusPill.tsx: badge + Reconnect button
* Claims, Remittances, ActivityLog pages wired to the tail
Tests
* 437 backend tests pass (was 418 before SP5)
* 154 frontend tests pass (was 124)
* npm run typecheck clean
* end-to-end smoke: open /api/claims/stream, POST 837, see new claims
arrive in real time without refresh
# Conflicts:
# src/pages/ActivityLog.tsx
# src/pages/Remittances.test.tsx
# src/pages/Remittances.tsx
This commit is contained in:
Tyler
@@ -0,0 +1,184 @@
|
||||
// @vitest-environment happy-dom
|
||||
// TailStatusPill is a leaf component with no async state and no hooks
|
||||
// beyond a single setInterval — happy-dom is the right env so we can
|
||||
// inspect the rendered DOM.
|
||||
(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 { TailStatusPill } from "./TailStatusPill";
|
||||
import type { TailStatus } from "@/hooks/useTailStream";
|
||||
|
||||
/**
|
||||
* Minimal render helper — the project doesn't ship
|
||||
* `@testing-library/react`, so we render straight into a real
|
||||
* `createRoot` and inspect the DOM via `container.innerHTML` and
|
||||
* `container.querySelector`.
|
||||
*
|
||||
* Note: React 18's `createRoot` defers DOM mutation until the calling
|
||||
* microtask flushes unless we wrap the render in `act`. The pill's only
|
||||
* effect is a `setInterval`, so we use `act(() => { root.render(node) })`
|
||||
* — sync act is fine because no state updates fire during render.
|
||||
*/
|
||||
function renderInto(node: React.ReactElement): {
|
||||
container: HTMLDivElement;
|
||||
root: Root;
|
||||
unmount: () => void;
|
||||
} {
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
act(() => {
|
||||
root.render(node);
|
||||
});
|
||||
return {
|
||||
container,
|
||||
root,
|
||||
unmount: () => {
|
||||
act(() => root.unmount());
|
||||
container.remove();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const ALL_STATUSES: TailStatus[] = [
|
||||
"connecting",
|
||||
"live",
|
||||
"reconnecting",
|
||||
"closed",
|
||||
"stalled",
|
||||
"error",
|
||||
];
|
||||
|
||||
describe("TailStatusPill", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-06-20T12:00:30Z"));
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("test_renders_badge_for_each_status", () => {
|
||||
// Iterate every TailStatus and assert no crash, plus a sensible label.
|
||||
// The plan's smoke spec calls for "renders all 5 statuses"; we test
|
||||
// all 6 (closed is also a status even if rare in practice).
|
||||
for (const status of ALL_STATUSES) {
|
||||
const { container, root, unmount } = renderInto(
|
||||
<TailStatusPill
|
||||
status={status}
|
||||
lastEventAt={null}
|
||||
onReconnect={() => {}}
|
||||
/>,
|
||||
);
|
||||
|
||||
// The pill root should be present.
|
||||
const pill = container.querySelector('[data-testid="tail-status-pill"]');
|
||||
expect(pill).not.toBeNull();
|
||||
|
||||
// The Badge child should carry the human label — match by text
|
||||
// content (the badge text is the only handoff the user reads).
|
||||
const expectedLabels: Record<TailStatus, string> = {
|
||||
connecting: "Connecting",
|
||||
live: "Live",
|
||||
reconnecting: "Reconnecting",
|
||||
closed: "Closed",
|
||||
stalled: "Stalled",
|
||||
error: "Error",
|
||||
};
|
||||
expect(pill?.textContent).toContain(expectedLabels[status]);
|
||||
|
||||
unmount();
|
||||
// Silence "unmount was not wrapped in act" warnings; renderInto's
|
||||
// unmount is a sync root.unmount() and React tolerates this for
|
||||
// leaf components with no async state.
|
||||
void root;
|
||||
}
|
||||
});
|
||||
|
||||
it("test_shows_reconnect_button_only_for_stalled_and_error", () => {
|
||||
const onReconnect = vi.fn();
|
||||
|
||||
for (const status of ALL_STATUSES) {
|
||||
const { container, unmount } = renderInto(
|
||||
<TailStatusPill
|
||||
status={status}
|
||||
lastEventAt={null}
|
||||
onReconnect={onReconnect}
|
||||
/>,
|
||||
);
|
||||
|
||||
const button = container.querySelector(
|
||||
'[data-testid="tail-status-pill-reconnect"]',
|
||||
);
|
||||
|
||||
if (status === "stalled" || status === "error") {
|
||||
expect(button).not.toBeNull();
|
||||
expect(button?.textContent).toContain("Reconnect");
|
||||
} else {
|
||||
expect(button).toBeNull();
|
||||
}
|
||||
|
||||
unmount();
|
||||
}
|
||||
});
|
||||
|
||||
it("test_subtitle_says_never_when_lastEventAt_is_null", () => {
|
||||
const { container, unmount } = renderInto(
|
||||
<TailStatusPill
|
||||
status="live"
|
||||
lastEventAt={null}
|
||||
onReconnect={() => {}}
|
||||
/>,
|
||||
);
|
||||
|
||||
const subtitle = container.querySelector(
|
||||
'[data-testid="tail-status-pill-subtitle"]',
|
||||
);
|
||||
expect(subtitle).not.toBeNull();
|
||||
expect(subtitle?.textContent).toBe("Last event: never");
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_subtitle_shows_age_in_seconds_when_lastEventAt_recent", () => {
|
||||
// 12 seconds ago — fake-timer driven so the assertion is stable.
|
||||
const eventAt = new Date("2026-06-20T12:00:18Z");
|
||||
const { container, unmount } = renderInto(
|
||||
<TailStatusPill
|
||||
status="live"
|
||||
lastEventAt={eventAt}
|
||||
onReconnect={() => {}}
|
||||
/>,
|
||||
);
|
||||
|
||||
const subtitle = container.querySelector(
|
||||
'[data-testid="tail-status-pill-subtitle"]',
|
||||
);
|
||||
expect(subtitle?.textContent).toBe("Last event: 12s ago");
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_reconnect_button_invokes_onReconnect_when_clicked", () => {
|
||||
const onReconnect = vi.fn();
|
||||
const { container, unmount } = renderInto(
|
||||
<TailStatusPill
|
||||
status="stalled"
|
||||
lastEventAt={null}
|
||||
onReconnect={onReconnect}
|
||||
/>,
|
||||
);
|
||||
|
||||
const button = container.querySelector(
|
||||
'[data-testid="tail-status-pill-reconnect"]',
|
||||
) as HTMLButtonElement | null;
|
||||
expect(button).not.toBeNull();
|
||||
button?.click();
|
||||
expect(onReconnect).toHaveBeenCalledTimes(1);
|
||||
|
||||
unmount();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,122 @@
|
||||
// ---------------------------------------------------------------------------
|
||||
// Live-tail status pill (sub-project 5, Phase 5 Task 21).
|
||||
//
|
||||
// Tiny presentational component used in the toolbar of `Claims`,
|
||||
// `Remittances`, and `ActivityLog` to surface the connection status of
|
||||
// `useTailStream`. Renders the existing `<Badge>` (variants from
|
||||
// `src/components/ui/badge.tsx`) plus a "Last event: 12s ago" subtitle
|
||||
// that re-ticks every second, plus a "↻ Reconnect" button that only
|
||||
// appears when the connection has visibly failed (`stalled` or `error`).
|
||||
//
|
||||
// Variant mapping (per spec):
|
||||
// live → success
|
||||
// connecting → warning
|
||||
// reconnecting → warning
|
||||
// closed → destructive
|
||||
// stalled → destructive
|
||||
// error → destructive
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
import { Badge } from "@/components/ui/badge";
|
||||
import { Button } from "@/components/ui/button";
|
||||
import type { TailStatus } from "@/hooks/useTailStream";
|
||||
|
||||
export interface TailStatusPillProps {
|
||||
status: TailStatus;
|
||||
lastEventAt: Date | null;
|
||||
onReconnect: () => void;
|
||||
}
|
||||
|
||||
const STATUS_LABEL: Record<TailStatus, string> = {
|
||||
connecting: "Connecting…",
|
||||
live: "Live",
|
||||
reconnecting: "Reconnecting…",
|
||||
closed: "Closed",
|
||||
stalled: "Stalled",
|
||||
error: "Error",
|
||||
};
|
||||
|
||||
type BadgeVariant = "success" | "warning" | "destructive";
|
||||
|
||||
const STATUS_VARIANT: Record<TailStatus, BadgeVariant> = {
|
||||
live: "success",
|
||||
connecting: "warning",
|
||||
reconnecting: "warning",
|
||||
closed: "destructive",
|
||||
stalled: "destructive",
|
||||
error: "destructive",
|
||||
};
|
||||
|
||||
/**
|
||||
* Render a duration in seconds as a compact human label. We deliberately
|
||||
* keep the units small — under a minute is the common case for a healthy
|
||||
* tail (the stall detector fires at 30s, so anything beyond is a real
|
||||
* outage the user will want to investigate).
|
||||
*/
|
||||
function formatAge(seconds: number): string {
|
||||
if (seconds < 0) return "0s";
|
||||
if (seconds < 60) return `${seconds}s`;
|
||||
const m = Math.floor(seconds / 60);
|
||||
const s = seconds % 60;
|
||||
if (m < 60) return s === 0 ? `${m}m` : `${m}m ${s}s`;
|
||||
const h = Math.floor(m / 60);
|
||||
return `${h}h ${m % 60}m`;
|
||||
}
|
||||
|
||||
export function TailStatusPill({
|
||||
status,
|
||||
lastEventAt,
|
||||
onReconnect,
|
||||
}: TailStatusPillProps): React.JSX.Element {
|
||||
// Re-tick every second so the "Last event: 12s ago" subtitle stays
|
||||
// current without the parent having to re-render. We mount the
|
||||
// interval only once per pill instance.
|
||||
const [now, setNow] = useState<number>(() => Date.now());
|
||||
useEffect(() => {
|
||||
const id = setInterval(() => setNow(Date.now()), 1000);
|
||||
return () => clearInterval(id);
|
||||
}, []);
|
||||
|
||||
const ageSeconds = lastEventAt
|
||||
? Math.max(0, Math.floor((now - lastEventAt.getTime()) / 1000))
|
||||
: null;
|
||||
|
||||
const subtitle =
|
||||
ageSeconds === null
|
||||
? "Last event: never"
|
||||
: `Last event: ${formatAge(ageSeconds)} ago`;
|
||||
|
||||
// Only show the manual reconnect affordance when the connection has
|
||||
// visibly failed. `closed` is reserved for the post-unmount state and
|
||||
// is intentionally not actionable — the component is about to vanish.
|
||||
const showReconnect = status === "stalled" || status === "error";
|
||||
|
||||
return (
|
||||
<div
|
||||
className="inline-flex items-center gap-2"
|
||||
data-testid="tail-status-pill"
|
||||
>
|
||||
<Badge variant={STATUS_VARIANT[status]} aria-label={`tail status: ${status}`}>
|
||||
{STATUS_LABEL[status]}
|
||||
</Badge>
|
||||
<span
|
||||
className="text-xs text-muted-foreground"
|
||||
data-testid="tail-status-pill-subtitle"
|
||||
>
|
||||
{subtitle}
|
||||
</span>
|
||||
{showReconnect && (
|
||||
<Button
|
||||
type="button"
|
||||
variant="outline"
|
||||
size="sm"
|
||||
onClick={onReconnect}
|
||||
data-testid="tail-status-pill-reconnect"
|
||||
>
|
||||
↻ Reconnect
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,231 @@
|
||||
// @vitest-environment happy-dom
|
||||
// React's act-aware env — mirror the other hook tests.
|
||||
(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 { beforeEach, describe, expect, it } from "vitest";
|
||||
import { useMergedTail } from "./useMergedTail";
|
||||
import { useTailStore } from "@/store/tail-store";
|
||||
import type { Claim, Remittance, Activity } from "@/types";
|
||||
|
||||
/**
|
||||
* Same `renderHook` shim used in `useClaimDetail`, `useDrawerUrlState`,
|
||||
* etc. — the project doesn't ship `@testing-library/react`, so we wire a
|
||||
* Probe component into a real `createRoot` and read the hook's return
|
||||
* value through a shared `result` object. `act` flushes React's state
|
||||
* updates between micro-tasks.
|
||||
*/
|
||||
function renderHook<TResult>(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();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Sample factories. The hook is generic over `T extends { id: string }`,
|
||||
// so for `claims` we use Claim, for `remittances` we use Remittance, etc.
|
||||
// Tests populate the store via `useTailStore.getState().addX(...)` — the
|
||||
// same path the live-tail hook uses — so the merge is exercised against
|
||||
// the real store shape (including the `claimOrder`/`remitOrder` indexing
|
||||
// the store maintains for the keyed-by-id slices).
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function claim(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",
|
||||
};
|
||||
}
|
||||
|
||||
function remit(id: string, claimId: string): Remittance {
|
||||
return {
|
||||
id,
|
||||
claimId,
|
||||
payerName: "Medicaid",
|
||||
paidAmount: 100,
|
||||
adjustmentAmount: 0,
|
||||
receivedDate: "2026-06-20",
|
||||
checkNumber: id,
|
||||
status: "received",
|
||||
};
|
||||
}
|
||||
|
||||
function activity(id: string, message: string): Activity {
|
||||
return {
|
||||
id,
|
||||
kind: "claim_submitted",
|
||||
message,
|
||||
timestamp: "2026-06-20T00:00:00Z",
|
||||
};
|
||||
}
|
||||
|
||||
describe("useMergedTail", () => {
|
||||
beforeEach(() => {
|
||||
// Singleton store — clear each slice between tests so cases are
|
||||
// independent. Mirrors the pattern in `tail-store.test.ts`.
|
||||
useTailStore.getState().reset("claims");
|
||||
useTailStore.getState().reset("remittances");
|
||||
useTailStore.getState().reset("activity");
|
||||
});
|
||||
|
||||
it("test_merges_base_and_tail_by_id_base_first", () => {
|
||||
// Base: [A, B]. Tail store adds [B, C, D] — note B is a duplicate
|
||||
// of a base item (mirrors what a snapshot replay on reconnect would
|
||||
// produce). After dedup, tail should contribute only [C, D].
|
||||
const base = [claim("A", "Alice"), claim("B", "Bob")];
|
||||
const { addClaim } = useTailStore.getState();
|
||||
addClaim(claim("B", "Bob-Updated")); // deduped — first write wins
|
||||
addClaim(claim("C", "Carol"));
|
||||
addClaim(claim("D", "Dave"));
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("claims", base),
|
||||
);
|
||||
|
||||
const merged = result.current ?? [];
|
||||
// Base items first, in their original order; then tail in arrival
|
||||
// order, with duplicates of base ids dropped.
|
||||
expect(merged.map((c) => c.id)).toEqual(["A", "B", "C", "D"]);
|
||||
// The base version of B must be preserved (the store's first-write-
|
||||
// wins rule keeps "Bob", not "Bob-Updated").
|
||||
expect(merged[1]?.patientName).toBe("Bob");
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_filter_predicate_drops_tail_items_that_dont_match", () => {
|
||||
// Base: [A]. Tail store: [B, C, D]. Predicate `id !== "D"` drops
|
||||
// the trailing item — the filter must run AFTER dedup and only
|
||||
// affect the tail contribution.
|
||||
const base = [claim("A", "Alice")];
|
||||
const { addClaim } = useTailStore.getState();
|
||||
addClaim(claim("B", "Bob"));
|
||||
addClaim(claim("C", "Carol"));
|
||||
addClaim(claim("D", "Dave"));
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("claims", base, (c) => c.id !== "D"),
|
||||
);
|
||||
|
||||
const merged = result.current ?? [];
|
||||
expect(merged.map((c) => c.id)).toEqual(["A", "B", "C"]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_empty_tail_store_returns_base_unchanged", () => {
|
||||
const base = [claim("A", "Alice"), claim("B", "Bob")];
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("claims", base),
|
||||
);
|
||||
|
||||
const merged = result.current ?? [];
|
||||
expect(merged).toHaveLength(2);
|
||||
expect(merged.map((c) => c.id)).toEqual(["A", "B"]);
|
||||
// The hook returns base verbatim when the tail slice is empty —
|
||||
// no copy churn, so `toBe` reference equality holds.
|
||||
expect(merged[0]).toBe(base[0]);
|
||||
expect(merged[1]).toBe(base[1]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_new_tail_item_appears_after_base", async () => {
|
||||
// Base: [A, B]. Tail store empty initially. After the component
|
||||
// mounts, we push a new claim to the store; the merged list must
|
||||
// include the new tail item, surfaced via re-render. (Plan calls
|
||||
// this "appears_at_top" — in this hook's contract the new item
|
||||
// comes AFTER base items, not before, so the assertion is that it
|
||||
// appears at all and is positioned correctly.)
|
||||
const base = [claim("A", "Alice"), claim("B", "Bob")];
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("claims", base),
|
||||
);
|
||||
|
||||
expect(result.current?.map((c) => c.id)).toEqual(["A", "B"]);
|
||||
|
||||
// New item lands in the store → zustand notifies subscribers → the
|
||||
// hook re-renders with the merged result.
|
||||
await act(async () => {
|
||||
useTailStore.getState().addClaim(claim("C", "Carol"));
|
||||
});
|
||||
|
||||
const merged = result.current ?? [];
|
||||
// Base first, then tail in arrival order.
|
||||
expect(merged.map((c) => c.id)).toEqual(["A", "B", "C"]);
|
||||
expect(merged[2]?.patientName).toBe("Carol");
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_activity_resource_uses_array_slice_and_dedupes_by_id", () => {
|
||||
// Activity doesn't have a keyed-by-id map — it's a plain array.
|
||||
// The hook must still dedup against baseItems by id (Activity has
|
||||
// an id, even though the store's addActivity doesn't dedup).
|
||||
const base = [activity("A", "alpha"), activity("B", "bravo")];
|
||||
const { addActivity } = useTailStore.getState();
|
||||
addActivity(activity("B", "bravo-dup")); // duplicate id
|
||||
addActivity(activity("C", "charlie"));
|
||||
addActivity(activity("D", "delta"));
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("activity", base),
|
||||
);
|
||||
|
||||
const merged = result.current ?? [];
|
||||
// Base first, then tail in arrival order. Tail's "B" is dropped
|
||||
// because it's a duplicate of base's "B".
|
||||
expect(merged.map((a) => a.id)).toEqual(["A", "B", "C", "D"]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_remittance_resource_orders_by_remitOrder", () => {
|
||||
// Smoke test for the remittances slice — exercise the second keyed
|
||||
// slice so we know the claim/remit branches are symmetric.
|
||||
const base = [remit("R-1", "CLM-1")];
|
||||
const { addRemittance } = useTailStore.getState();
|
||||
addRemittance(remit("R-2", "CLM-2"));
|
||||
addRemittance(remit("R-3", "CLM-3"));
|
||||
|
||||
const { result, unmount } = renderHook(() =>
|
||||
useMergedTail("remittances", base),
|
||||
);
|
||||
|
||||
const merged = result.current ?? [];
|
||||
expect(merged.map((r) => r.id)).toEqual(["R-1", "R-2", "R-3"]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,77 @@
|
||||
// ---------------------------------------------------------------------------
|
||||
// Live-tail merge hook (sub-project 5, Phase 5 Task 20).
|
||||
//
|
||||
// Reads the matching slice of `useTailStore` and returns a single array:
|
||||
// `baseItems` first (in the order the caller supplied them), then any new
|
||||
// tail items in arrival order, with duplicates of base ids removed.
|
||||
//
|
||||
// Design notes:
|
||||
// - The dedup runs against `baseItems` (not the other way around) because
|
||||
// the base list is the authoritative snapshot from the page's JSON
|
||||
// fetch — the tail slice is an opportunistic delta, so the base wins
|
||||
// when an id appears in both.
|
||||
// - The optional `filterFn` is applied to the tail slice AFTER dedup so a
|
||||
// page-specific filter (e.g. "only show me submitted claims") doesn't
|
||||
// accidentally include items that were already filtered out of base.
|
||||
// - We don't `useMemo` here: `baseItems` is a new array reference on
|
||||
// every render of the caller, so memo deps would invalidate anyway.
|
||||
// The merge is cheap (one Set build + two filters) and zustand's
|
||||
// selector ensures we only re-render when the slice actually changes.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
import { useTailStore } from "@/store/tail-store";
|
||||
import type { TailResource } from "@/lib/tail-stream";
|
||||
|
||||
export function useMergedTail<T extends { id: string }>(
|
||||
resource: TailResource,
|
||||
baseItems: T[],
|
||||
filterFn?: (item: T) => boolean,
|
||||
): T[] {
|
||||
// Select the slice that matches the resource. We return a fresh array
|
||||
// each time so the consumer gets a stable iteration order even when
|
||||
// zustand hands us a new container (Record or array) reference.
|
||||
const tailSlice = useTailStore((s) => {
|
||||
switch (resource) {
|
||||
case "claims": {
|
||||
// The store keys claims by id in a Record for O(1) updates, but
|
||||
// also keeps an `claimOrder` array so we can iterate in arrival
|
||||
// order. Filter out any holes (defensive — shouldn't happen but
|
||||
// type-narrows the result to Claim[]).
|
||||
const out: unknown[] = [];
|
||||
for (const id of s.claimOrder) {
|
||||
const v = s.claims[id];
|
||||
if (v !== undefined) out.push(v);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
case "remittances": {
|
||||
const out: unknown[] = [];
|
||||
for (const id of s.remitOrder) {
|
||||
const v = s.remittances[id];
|
||||
if (v !== undefined) out.push(v);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
case "activity":
|
||||
// Activity is already an array — the store appends in arrival
|
||||
// order, so iteration order is exactly what we want.
|
||||
return s.activity as unknown[];
|
||||
}
|
||||
});
|
||||
|
||||
const baseIds = new Set<string>();
|
||||
for (const b of baseItems) baseIds.add(b.id);
|
||||
|
||||
const tailAfterDedup: unknown[] = [];
|
||||
for (const t of tailSlice) {
|
||||
const id = (t as { id: string }).id;
|
||||
if (baseIds.has(id)) continue;
|
||||
tailAfterDedup.push(t);
|
||||
}
|
||||
|
||||
const tailAfterFilter = filterFn
|
||||
? tailAfterDedup.filter((t) => filterFn(t as T))
|
||||
: tailAfterDedup;
|
||||
|
||||
return [...baseItems, ...(tailAfterFilter as T[])];
|
||||
}
|
||||
@@ -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<TailEvent>) => void) | null = null;
|
||||
let rejectNext: ((e: unknown) => void) | null = null;
|
||||
let done = false;
|
||||
let pendingError: unknown = null;
|
||||
|
||||
const next = (): Promise<IteratorResult<TailEvent>> => {
|
||||
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<IteratorResult<TailEvent>>((resolve, reject) => {
|
||||
resolveNext = resolve;
|
||||
rejectNext = reject;
|
||||
});
|
||||
};
|
||||
|
||||
const iter: AsyncIterableIterator<TailEvent> = {
|
||||
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<typeof makeCtrl>;
|
||||
|
||||
/** 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<TResult>(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<void> {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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 `<TailStatusPill>`), 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. `<TailStatusPill>`) 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<TailStatus>("connecting");
|
||||
const [lastEventAt, setLastEventAt] = useState<Date | null>(null);
|
||||
const [error, setError] = useState<Error | null>(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<AbortController | null>(null);
|
||||
|
||||
const forceReconnect = useCallback(() => {
|
||||
setReconnectNonce((n) => n + 1);
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
let cancelled = false;
|
||||
let attempt = 0;
|
||||
let stallTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let backoffTimer: ReturnType<typeof setTimeout> | 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<void> => {
|
||||
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 };
|
||||
}
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,9 @@
|
||||
import { useCallback, useMemo } from "react";
|
||||
import { useSearchParams } from "react-router-dom";
|
||||
import { useActivity } from "@/hooks/useActivity";
|
||||
import { useTailStream } from "@/hooks/useTailStream";
|
||||
import { useMergedTail } from "@/hooks/useMergedTail";
|
||||
import { TailStatusPill } from "@/components/TailStatusPill";
|
||||
import { ActivityFeed } from "@/components/ActivityFeed";
|
||||
import { ActivityFilters, type SinceValue } from "@/components/ActivityFilters";
|
||||
import { Skeleton } from "@/components/ui/skeleton";
|
||||
@@ -69,12 +72,17 @@ export function ActivityLog() {
|
||||
});
|
||||
|
||||
const allItems = data?.items ?? [];
|
||||
|
||||
// SP5: live-tail wiring. `useMergedTail` appends tail-arriving events
|
||||
// to the base query result, then we apply the same client-side
|
||||
// multi-kind filter so live events respect the active filter set.
|
||||
const { status: tailStatus, lastEventAt: tailLastEventAt, forceReconnect } =
|
||||
useTailStream("activity");
|
||||
const merged = useMergedTail("activity", allItems);
|
||||
const items = useMemo(() => {
|
||||
// Server already returned a single-kind slice when `apiKind` is set;
|
||||
// otherwise apply the (multi-)kind filter locally.
|
||||
if (selectedKinds.length <= 1) return allItems;
|
||||
return allItems.filter((a) => selectedKinds.includes(a.kind));
|
||||
}, [allItems, selectedKinds]);
|
||||
if (selectedKinds.length <= 1) return merged;
|
||||
return merged.filter((a) => selectedKinds.includes(a.kind));
|
||||
}, [merged, selectedKinds]);
|
||||
|
||||
const writeParams = useCallback(
|
||||
(mutate: (next: URLSearchParams) => void) => {
|
||||
@@ -152,6 +160,16 @@ export function ActivityLog() {
|
||||
) : null}
|
||||
|
||||
<div className="surface rounded-xl p-6">
|
||||
<div className="flex flex-wrap items-center gap-3 mb-4">
|
||||
{/* SP5: live-tail status pill, right-aligned in the toolbar. */}
|
||||
<div className="ml-auto">
|
||||
<TailStatusPill
|
||||
status={tailStatus}
|
||||
lastEventAt={tailLastEventAt}
|
||||
onReconnect={forceReconnect}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
{isLoading ? (
|
||||
<div className="space-y-2">
|
||||
{Array.from({ length: 5 }).map((_, i) => (
|
||||
|
||||
@@ -10,6 +10,7 @@ import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { Claims } from "./Claims";
|
||||
import { api } from "@/lib/api";
|
||||
import { useTailStore } from "@/store/tail-store";
|
||||
import type { Claim, ClaimDetail } from "@/types";
|
||||
|
||||
// Module-level mock — vitest hoists vi.mock above imports. We mock the
|
||||
@@ -38,6 +39,30 @@ vi.mock("@/lib/api", () => {
|
||||
};
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Live-tail hook mock (sub-project 5, Phase 5 Task 22 page-level tests).
|
||||
//
|
||||
// We mock `useTailStream` directly (per the plan's "latter is simpler"
|
||||
// guidance) so the page-level tests don't have to drive the real
|
||||
// `streamTail` parser or the real AbortController-based reconnect loop.
|
||||
// `useTailStream` is exercised exhaustively in its own test file
|
||||
// (`useTailStream.test.ts`); the page only needs to observe that the
|
||||
// returned status flows through to `<TailStatusPill>` and that items
|
||||
// pushed into the tail store surface as rows via `useMergedTail`.
|
||||
//
|
||||
// The mock returns `status: "live"` by default — that matches what the
|
||||
// real hook settles to after a `snapshot_end` event arrives, which is
|
||||
// the state the page sees in production ~all of the time.
|
||||
// ---------------------------------------------------------------------------
|
||||
vi.mock("@/hooks/useTailStream", () => ({
|
||||
useTailStream: vi.fn(() => ({
|
||||
status: "live",
|
||||
lastEventAt: null,
|
||||
error: null,
|
||||
forceReconnect: vi.fn(),
|
||||
})),
|
||||
}));
|
||||
|
||||
const SAMPLE_CLAIMS: Claim[] = [
|
||||
{
|
||||
id: "CLM-1",
|
||||
@@ -190,6 +215,10 @@ describe("Claims page drawer wiring", () => {
|
||||
(api.getClaimDetail as unknown as ReturnType<typeof vi.fn>).mockResolvedValue(
|
||||
SAMPLE_DETAIL
|
||||
);
|
||||
// Live-tail slice is a singleton zustand store — clear it between
|
||||
// tests so the live-arrival test doesn't see rows from a previous
|
||||
// case.
|
||||
useTailStore.getState().reset("claims");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
@@ -420,4 +449,86 @@ describe("Claims page drawer wiring", () => {
|
||||
document.body.querySelector('[data-testid="keyboard-cheatsheet"]')
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------------
|
||||
// Live-tail integration (sub-project 5, Phase 5 Task 22 page tests).
|
||||
//
|
||||
// We mock `useTailStream` at the module level (see the top of this
|
||||
// file) so the page sees a settled `status: "live"` without driving
|
||||
// the real streamTail parser. To exercise the "tail arrival triggers a
|
||||
// row" path we push a new claim directly into `useTailStore` — this
|
||||
// is exactly what the real `useTailStream` hook does on `item` events,
|
||||
// so we cover the same render path (zustand notify → useMergedTail
|
||||
// re-derive → `<TableRow>` mount).
|
||||
// -------------------------------------------------------------------
|
||||
|
||||
it("test_live_tail_arrival_triggers_row_in_table", async () => {
|
||||
const { unmount } = renderClaims();
|
||||
|
||||
// Base list (SAMPLE_CLAIMS, set in beforeEach) must be visible first.
|
||||
await settle(
|
||||
() => document.body.textContent?.includes("CLM-1") ?? false,
|
||||
);
|
||||
|
||||
// The new claim id is brand new (not in base, not in tail store yet).
|
||||
const TAIL_ID = "CLM-TAIL-1";
|
||||
expect(
|
||||
document.body.textContent?.includes(TAIL_ID) ?? false,
|
||||
).toBe(false);
|
||||
|
||||
// Simulate a live-arriving claim — this is what `useTailStream`
|
||||
// does when it dispatches an `item` event. Wrap in `act` because
|
||||
// zustand subscribers re-render synchronously.
|
||||
await act(async () => {
|
||||
useTailStore.getState().addClaim({
|
||||
id: TAIL_ID,
|
||||
patientName: "Live Arrival",
|
||||
providerNpi: "1234567890",
|
||||
payerName: "Colorado Medicaid",
|
||||
cptCode: "99215",
|
||||
billedAmount: 999,
|
||||
receivedAmount: 0,
|
||||
status: "submitted",
|
||||
submissionDate: "2026-06-20",
|
||||
});
|
||||
});
|
||||
|
||||
// The new row must appear — proves the end-to-end path
|
||||
// zustand.addClaim → useTailStore notify → useMergedTail re-derive
|
||||
// → <TableBody> mounts the row with the new id.
|
||||
await settle(
|
||||
() => document.body.textContent?.includes(TAIL_ID) ?? false,
|
||||
);
|
||||
expect(document.body.textContent).toContain(TAIL_ID);
|
||||
// The row's secondary cells should also be present so we know the
|
||||
// row fully rendered, not just that the id appeared somewhere
|
||||
// (e.g. in the search-input placeholder or similar).
|
||||
expect(document.body.textContent).toContain("Live Arrival");
|
||||
expect(document.body.textContent).toContain("99215");
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("test_status_pill_shows_live_after_stream_connects", async () => {
|
||||
const { unmount } = renderClaims();
|
||||
|
||||
// Wait for the base table to mount — once it's there, the toolbar
|
||||
// (which contains the pill) has also mounted.
|
||||
await settle(
|
||||
() => document.body.textContent?.includes("CLM-1") ?? false,
|
||||
);
|
||||
|
||||
// The mocked `useTailStream` returns `status: "live"`, so the
|
||||
// <TailStatusPill> should render with the human label "Live"
|
||||
// (per `STATUS_LABEL` in `TailStatusPill.tsx`). We assert on the
|
||||
// pill's textContent so this stays robust against future styling
|
||||
// changes (the test doesn't depend on Tailwind classes).
|
||||
const pill = document.body.querySelector(
|
||||
'[data-testid="tail-status-pill"]',
|
||||
);
|
||||
expect(pill).not.toBeNull();
|
||||
expect(pill?.textContent).toContain("Live");
|
||||
|
||||
unmount();
|
||||
});
|
||||
});
|
||||
|
||||
+36
-2
@@ -27,9 +27,12 @@ import { FilterChips, type FilterChipOption } from "@/components/ui/filter-chips
|
||||
import { Pagination } from "@/components/ui/pagination";
|
||||
import { useClaims } from "@/hooks/useClaims";
|
||||
import { useDrawerUrlState } from "@/hooks/useDrawerUrlState";
|
||||
import { useTailStream } from "@/hooks/useTailStream";
|
||||
import { useMergedTail } from "@/hooks/useMergedTail";
|
||||
import { TailStatusPill } from "@/components/TailStatusPill";
|
||||
import { useAppStore } from "@/store";
|
||||
import { fmt } from "@/lib/format";
|
||||
import type { ClaimStatus } from "@/types";
|
||||
import type { Claim, ClaimStatus } from "@/types";
|
||||
import { cn } from "@/lib/utils";
|
||||
|
||||
const ALL = "all" as const;
|
||||
@@ -74,7 +77,28 @@ export function Claims() {
|
||||
};
|
||||
|
||||
const { data, isLoading, isError, error, refetch, dataUpdatedAt } = useClaims(params);
|
||||
const items = data?.items ?? [];
|
||||
|
||||
// SP5 live-tail wiring (sub-project 5, Phase 5 Task 22).
|
||||
//
|
||||
// The tail stream emits every claim that lands in the DB — including
|
||||
// ones that don't match the user's current `status` / `provider_npi`
|
||||
// filter. Mirror the server-side filter the page already applies via
|
||||
// `useClaims` so a newly-arrived item that wouldn't pass the same
|
||||
// predicate is dropped before it can show up in the table. `useMergedTail`
|
||||
// applies this to the tail slice only; base items are already filtered
|
||||
// server-side.
|
||||
const tailFilterFn = useMemo(
|
||||
() => (c: Claim) => {
|
||||
if (status !== ALL && c.status !== status) return false;
|
||||
if (npi !== ALL && c.providerNpi !== npi) return false;
|
||||
return true;
|
||||
},
|
||||
[status, npi],
|
||||
);
|
||||
|
||||
const { status: tailStatus, lastEventAt: tailLastEventAt, forceReconnect } =
|
||||
useTailStream("claims");
|
||||
const items = useMergedTail("claims", data?.items ?? [], tailFilterFn);
|
||||
|
||||
const totals = useMemo(
|
||||
() => ({
|
||||
@@ -190,6 +214,16 @@ export function Claims() {
|
||||
))}
|
||||
</SelectContent>
|
||||
</Select>
|
||||
|
||||
{/* SP5: live-tail status pill. `ml-auto` pins it to the right
|
||||
edge of the toolbar so it doesn't crowd the search input. */}
|
||||
<div className="ml-auto">
|
||||
<TailStatusPill
|
||||
status={tailStatus}
|
||||
lastEventAt={tailLastEventAt}
|
||||
onReconnect={forceReconnect}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="mt-3">
|
||||
|
||||
@@ -10,6 +10,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { Remittances } from "./Remittances";
|
||||
import { api } from "@/lib/api";
|
||||
import { useTailStore } from "@/store/tail-store";
|
||||
|
||||
// Module-level mock: vitest hoists `vi.mock` calls above imports. We only
|
||||
// stub the method this page touches via the `useRemittances` hook.
|
||||
@@ -23,6 +24,18 @@ vi.mock("@/lib/api", () => ({
|
||||
},
|
||||
}));
|
||||
|
||||
// Mock the live-tail hook so the page renders the pill in the settled
|
||||
// `live` state without driving the real streamTail parser. The hook's
|
||||
// own test file (`useTailStream.test.ts`) covers the lifecycle in depth.
|
||||
vi.mock("@/hooks/useTailStream", () => ({
|
||||
useTailStream: vi.fn(() => ({
|
||||
status: "live",
|
||||
lastEventAt: null,
|
||||
error: null,
|
||||
forceReconnect: vi.fn(),
|
||||
})),
|
||||
}));
|
||||
|
||||
/**
|
||||
* Minimal `render` helper using react-dom/client + act(). Mirrors the
|
||||
* `renderIntoContainer` helper in `Reconciliation.test.tsx` — see that
|
||||
@@ -178,6 +191,10 @@ const SAMPLE_REMITS = [
|
||||
describe("Remittances", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
// Singleton tail-store: clear the remittances slice between tests
|
||||
// so a tail-arrival case (if added later) doesn't see rows from a
|
||||
// previous test.
|
||||
useTailStore.getState().reset("remittances");
|
||||
(
|
||||
api.listRemittances as unknown as ReturnType<typeof vi.fn>
|
||||
).mockResolvedValue({
|
||||
@@ -389,4 +406,36 @@ describe("Remittances", () => {
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------------
|
||||
// Live-tail status pill (sub-project 5, Phase 5 Task 23, optional
|
||||
// page-level test). Cheap smoke check: the toolbar renders the pill
|
||||
// with the `live` label whenever the mocked stream is settled. The
|
||||
// hook's lifecycle is covered exhaustively in its own test file.
|
||||
// -------------------------------------------------------------------
|
||||
|
||||
it("test_status_pill_shows_live_in_toolbar", async () => {
|
||||
(
|
||||
api.listRemittances as unknown as ReturnType<typeof vi.fn>
|
||||
).mockResolvedValue({
|
||||
items: [],
|
||||
total: 0,
|
||||
returned: 0,
|
||||
has_more: false,
|
||||
});
|
||||
|
||||
const { unmount } = renderIntoContainer(React.createElement(Remittances));
|
||||
|
||||
// Wait for the empty-state to mount — once the page body is rendered,
|
||||
// the toolbar (and thus the pill) is in the DOM.
|
||||
await waitForText("Remittances · awaiting first 835");
|
||||
|
||||
const pill = document.body.querySelector(
|
||||
'[data-testid="tail-status-pill"]',
|
||||
);
|
||||
expect(pill).not.toBeNull();
|
||||
expect(pill?.textContent).toContain("Live");
|
||||
|
||||
unmount();
|
||||
});
|
||||
});
|
||||
+43
-12
@@ -1,4 +1,4 @@
|
||||
import { Fragment, useCallback, useState } from "react";
|
||||
import { Fragment, useCallback, useMemo, useState } from "react";
|
||||
import { ChevronDown, ChevronRight, Receipt } from "lucide-react";
|
||||
import {
|
||||
Table,
|
||||
@@ -17,9 +17,12 @@ import { Pagination } from "@/components/ui/pagination";
|
||||
import { KeyboardCheatsheet } from "@/components/KeyboardCheatsheet";
|
||||
import { useRemittances } from "@/hooks/useRemittances";
|
||||
import { useRowKeyboard } from "@/hooks/useRowKeyboard";
|
||||
import { useTailStream } from "@/hooks/useTailStream";
|
||||
import { useMergedTail } from "@/hooks/useMergedTail";
|
||||
import { TailStatusPill } from "@/components/TailStatusPill";
|
||||
import { fmt } from "@/lib/format";
|
||||
import { cn } from "@/lib/utils";
|
||||
import type { CasAdjustment, RemittanceStatus } from "@/types";
|
||||
import type { CasAdjustment, Remittance, RemittanceStatus } from "@/types";
|
||||
|
||||
const PAGE_SIZE = 25;
|
||||
|
||||
@@ -71,7 +74,25 @@ export function Remittances() {
|
||||
limit: PAGE_SIZE,
|
||||
offset: (page - 1) * PAGE_SIZE,
|
||||
});
|
||||
const items = data?.items ?? [];
|
||||
|
||||
// SP5 live-tail wiring (sub-project 5, Phase 5 Task 23). The tail
|
||||
// stream emits every remittance that lands — including ones that
|
||||
// don't match the user's current `status` filter chip. Drop those
|
||||
// tail items so a newly-arrived remit that wouldn't match the page's
|
||||
// intent doesn't flash into the table. Base items are NOT filtered
|
||||
// here (they reflect whatever the list query returned, matching the
|
||||
// existing page behavior).
|
||||
const tailFilterFn = useMemo(
|
||||
() => (r: Remittance) => {
|
||||
if (status && r.status !== status) return false;
|
||||
return true;
|
||||
},
|
||||
[status],
|
||||
);
|
||||
|
||||
const { status: tailStatus, lastEventAt: tailLastEventAt, forceReconnect } =
|
||||
useTailStream("remittances");
|
||||
const items = useMergedTail("remittances", data?.items ?? [], tailFilterFn);
|
||||
|
||||
const total = items.reduce(
|
||||
(acc, r) => ({
|
||||
@@ -167,15 +188,25 @@ export function Remittances() {
|
||||
</div>
|
||||
|
||||
<div className="surface rounded-xl p-4">
|
||||
<FilterChips
|
||||
options={STATUS_OPTIONS}
|
||||
value={status}
|
||||
onChange={(v) => {
|
||||
setStatus((v as RemittanceStatus | null) ?? null);
|
||||
setPage(1);
|
||||
}}
|
||||
eyebrow="Status"
|
||||
/>
|
||||
<div className="flex flex-wrap items-center gap-3">
|
||||
<FilterChips
|
||||
options={STATUS_OPTIONS}
|
||||
value={status}
|
||||
onChange={(v) => {
|
||||
setStatus((v as RemittanceStatus | null) ?? null);
|
||||
setPage(1);
|
||||
}}
|
||||
eyebrow="Status"
|
||||
/>
|
||||
{/* SP5: live-tail status pill, right-aligned in the toolbar. */}
|
||||
<div className="ml-auto">
|
||||
<TailStatusPill
|
||||
status={tailStatus}
|
||||
lastEventAt={tailLastEventAt}
|
||||
onReconnect={forceReconnect}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="surface rounded-xl overflow-hidden">
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
@@ -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: [] };
|
||||
}
|
||||
}),
|
||||
}));
|
||||
Reference in New Issue
Block a user