cyclone/src/hooks/useTailStream.test.ts

// @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();
  });
});