cyclone/src/hooks/useTailStream.ts

// ---------------------------------------------------------------------------
// 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 };
}