server-26/drb-frontend/lib/usePipelineHealth.ts

"use client";

/**
 * Pipeline truth for the live wall (server-26#232).
 *
 * The old '● Live · Ns ago' pill was reset by any node heartbeat, so the wall
 * read live while transcription/correlation were dead (#196). This hook keys
 * health to the PIPELINE instead: radio → STT → AI, plus the context feeds.
 *
 * Two sources, deliberately both:
 *  - Data timestamps on what the wall already subscribes to (recent calls,
 *    active incidents). These survive a c2-core restart.
 *  - GET /health/ai (main.py, auth-free) — per-tier provider state. It is
 *    in-memory per process and resets on restart, so it can only ADD a
 *    degraded verdict, never clear one the data shows.
 */
import { useEffect, useMemo, useState } from "react";
import type { CallRecord, CameraProviderFeed, IncidentRecord, NodeRecord, Ny511FeedStatus } from "@/lib/types";

const BASE = process.env.NEXT_PUBLIC_C2_URL ?? "http://localhost:8000";
const HEALTH_POLL_MS = 60_000;
const TICK_MS = 30_000;

function envMinutes(): number {
  const n = Number(process.env.NEXT_PUBLIC_WALL_STALE_MIN);
  return Number.isFinite(n) && n > 0 ? n : 10;
}
/** Minutes a stage may be dry before the wall calls itself degraded. */
export const WALL_STALE_MIN = envMinutes();
/**
 * Radio silence and "no incident written" are weaker evidence than "radio
 * heard but not transcribed" — a quiet county at 4am is not an outage, and the
 * correlator orphans in doubt — so those two stages get a longer leash.
 */
const ABSENCE_FACTOR = 3;
/**
 * Calls Whisper returned nothing for (silence, kerchunks, degenerate output)
 * get no transcript by design, so one or two blank calls after the last
 * transcript are normal. It takes this many to call STT dry.
 */
const MIN_UNSERVED = 3;

// ── /health/ai response (app/internal/ai_health.py snapshot()) ──────────────
export interface AiTierState {
  degraded: boolean;
  permanent: boolean;
  provider: string | null;
  model: string | null;
  problem: string | null;
  fix: string | null;
  first_seen: string | null;
  last_seen: string | null;
  consecutive_failures: number;
  alerted: boolean;
}
export interface AiHealthResponse {
  tiers: Record<string, AiTierState>;
}

export type StageKey = "radio" | "stt" | "ai" | "feeds";
export type StageState = "ok" | "degraded" | "unknown";

export interface PipelineStage {
  key: StageKey;
  label: string;
  state: StageState;
  /** Newest evidence this stage produced output, or null if none in the window. */
  lastAt: Date | null;
  /** One-line human explanation, shown as the chip tooltip / bar text. */
  detail: string;
}

export interface PipelineHealth {
  stages: PipelineStage[];
  /** True when a core stage (radio/STT/AI) is dry or an AI tier is degraded. */
  degraded: boolean;
  /** Bar text, e.g. "radio heard, not transcribed since 12:15 · transcription tier degraded (openai: …)". */
  message: string | null;
  degradedTiers: { tier: string; state: AiTierState }[];
  /** null until the first /health/ai answer; false if it could not be reached. */
  aiHealthReachable: boolean | null;
  /** Distinct nodes with a call started in the last WALL_STALE_MIN minutes. */
  nodesHearing: number;
  newestCallAt: Date | null;
  staleMin: number;
}

function useAiHealth(): { data: AiHealthResponse | null; reachable: boolean | null } {
  const [data, setData] = useState<AiHealthResponse | null>(null);
  const [reachable, setReachable] = useState<boolean | null>(null);
  useEffect(() => {
    let cancelled = false;
    const load = () => {
      if (document.hidden) return;
      fetch(`${BASE}/health/ai`, { cache: "no-store" })
        .then((r) => (r.ok ? r.json() : Promise.reject(new Error(`HTTP ${r.status}`))))
        .then((j: AiHealthResponse) => {
          if (cancelled) return;
          setData(j && typeof j.tiers === "object" ? j : { tiers: {} });
          setReachable(true);
        })
        .catch((e) => {
          if (cancelled) return;
          console.error("health/ai:", e);
          setReachable(false);
        });
    };
    load();
    const id = setInterval(load, HEALTH_POLL_MS);
    const onVis = () => { if (!document.hidden) load(); };
    document.addEventListener("visibilitychange", onVis);
    return () => { cancelled = true; clearInterval(id); document.removeEventListener("visibilitychange", onVis); };
  }, []);
  return { data, reachable };
}

const ms = (iso: string | null | undefined): number => {
  if (!iso) return NaN;
  const t = Date.parse(iso);
  return Number.isFinite(t) ? t : NaN;
};
const maxDate = (vals: number[]): Date | null => {
  const ok = vals.filter((v) => Number.isFinite(v));
  return ok.length ? new Date(Math.max(...ok)) : null;
};
export function clockTime(d: Date): string {
  return d.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", hour12: false });
}
function agoText(d: Date | null, now: number): string {
  if (!d) return "none in window";
  const s = Math.max(0, Math.floor((now - d.getTime()) / 1000));
  if (s < 60) return `${s}s ago`;
  if (s < 3600) return `${Math.floor(s / 60)}m ago`;
  return `${Math.floor(s / 3600)}h ago`;
}
const hasTranscript = (c: CallRecord) => !!(c.transcript || c.transcript_corrected);
const isAiTier = (t: string) => t !== "transcription";

export interface PipelineInput {
  calls: CallRecord[];
  incidents: IncidentRecord[];
  nodes: NodeRecord[];
  eventsStatus?: Ny511FeedStatus;
  cameraProviders: CameraProviderFeed[];
}

export function usePipelineHealth(input: PipelineInput): PipelineHealth {
  const { calls, incidents, nodes, eventsStatus, cameraProviders } = input;
  const { data: ai, reachable } = useAiHealth();
  // Staleness is a function of wall-clock time, not only of new data arriving.
  const [now, setNow] = useState(() => Date.now());
  useEffect(() => {
    const id = setInterval(() => setNow(Date.now()), TICK_MS);
    return () => clearInterval(id);
  }, []);
  return useMemo(
    () => derivePipelineHealth({ calls, incidents, nodes, eventsStatus, cameraProviders }, ai, reachable, now),
    [calls, incidents, nodes, eventsStatus, cameraProviders, ai, reachable, now],
  );
}

/** Pure derivation — separated from the hook so it can be exercised without React. */
export function derivePipelineHealth(
  input: PipelineInput,
  ai: AiHealthResponse | null,
  reachable: boolean | null,
  now: number,
): PipelineHealth {
  const { calls, incidents, nodes, eventsStatus, cameraProviders } = input;
  const staleMs = WALL_STALE_MIN * 60_000;
  const absenceMs = staleMs * ABSENCE_FACTOR;
  // Canonical, finished recordings only: a second node's copy of the same
  // transmission (duplicate_of) is never transcribed, and an in-flight call
  // has not been uploaded yet.
  const real = calls.filter((c) => !c.duplicate_of && c.status !== "active" && Number.isFinite(ms(c.started_at)));

  const tiers = Object.entries(ai?.tiers ?? {});
  const degradedTiers = tiers.filter(([, s]) => s?.degraded).map(([tier, state]) => ({ tier, state }));
  const tierText = (tier: string, s: AiTierState) =>
    `${tier} tier degraded${s.provider ? ` (${s.provider}${s.model ? ` ${s.model}` : ""})` : ""}${s.problem ? `: ${s.problem}` : ""}`;

  // ── Radio ────────────────────────────────────────────────────────────────
  const newestCallAt = maxDate(calls.map((c) => ms(c.started_at)));
  const anyNodeUp = nodes.some((n) => n.status !== "offline");
  // Every configured node offline is not "Listening" — nothing can be heard.
  const allNodesOffline = nodes.length > 0 && !anyNodeUp;
  const radioDry = anyNodeUp && (!newestCallAt || now - newestCallAt.getTime() > absenceMs);
  const radio: PipelineStage = {
    key: "radio", label: "Radio", lastAt: newestCallAt,
    state: nodes.length === 0 ? "unknown" : radioDry || allNodesOffline ? "degraded" : "ok",
    detail: allNodesOffline
      ? `all ${nodes.length} node${nodes.length === 1 ? "" : "s"} offline`
      : newestCallAt ? `last call heard ${agoText(newestCallAt, now)}` : "no calls in the recent window",
  };
  // Bounded to configured nodes so a call from a since-removed node can't push
  // the KPI past N/N.
  const knownNodes = new Set(nodes.map((n) => n.node_id));
  const nodesHearing = new Set(
    calls.filter((c) => now - ms(c.started_at) <= staleMs && knownNodes.has(c.node_id)).map((c) => c.node_id),
  ).size;

  // ── STT: radio heard, nothing transcribed ────────────────────────────────
  const newestTranscriptAt = maxDate(real.filter(hasTranscript).map((c) => ms(c.started_at)));
  const unserved = real
    .filter((c) => !hasTranscript(c) && (!newestTranscriptAt || ms(c.started_at) > newestTranscriptAt.getTime()))
    .map((c) => ms(c.started_at));
  const oldestUnserved = unserved.length ? Math.min(...unserved) : NaN;
  const sttDry = unserved.length >= MIN_UNSERVED && now - oldestUnserved > staleMs;
  const sttTier = degradedTiers.find((t) => t.tier === "transcription");
  const sttSince = newestTranscriptAt ?? (Number.isFinite(oldestUnserved) ? new Date(oldestUnserved) : null);
  const stt: PipelineStage = {
    key: "stt", label: "Transcripts", lastAt: newestTranscriptAt,
    state: sttDry || sttTier ? "degraded" : real.length ? "ok" : "unknown",
    detail: sttDry
      ? `radio heard, not transcribed since ${sttSince ? clockTime(sttSince) : "—"} (${unserved.length} calls waiting)`
      : newestTranscriptAt ? `last transcript ${agoText(newestTranscriptAt, now)}` : "no transcripts in the recent window",
  };

  // ── AI: transcribed, but nothing correlated ─────────────────────────────
  const newestAiAt = maxDate([
    ...incidents.flatMap((i) => [ms(i.updated_at), ms(i.last_thin_at)]),
    ...real.filter((c) => (c.incident_ids?.length ?? 0) > 0 || !!c.incident_id).map((c) => ms(c.started_at)),
  ]);
  const transcribedSinceAi = real
    .filter((c) => hasTranscript(c) && (!newestAiAt || ms(c.started_at) > newestAiAt.getTime()))
    .map((c) => ms(c.started_at));
  const oldestUncorrelated = transcribedSinceAi.length ? Math.min(...transcribedSinceAi) : NaN;
  const aiDry = transcribedSinceAi.length >= MIN_UNSERVED && now - oldestUncorrelated > absenceMs;
  const aiTiers = degradedTiers.filter((t) => isAiTier(t.tier));
  const ai_: PipelineStage = {
    key: "ai", label: "AI", lastAt: newestAiAt,
    state: aiDry || aiTiers.length ? "degraded" : reachable === false && !newestAiAt ? "unknown" : "ok",
    detail: aiDry
      ? `transcribed, no incident written since ${newestAiAt ? clockTime(newestAiAt) : clockTime(new Date(oldestUncorrelated))}`
      : `${newestAiAt ? `last incident write ${agoText(newestAiAt, now)}` : "no incident writes in the recent window"}${reachable === false ? " · /health/ai unreachable" : ""}`,
  };

  // ── Context feeds (chip only — a dead camera feed doesn't blind the pipeline) ──
  const feedProblems: string[] = [];
  if (eventsStatus && !eventsStatus.disabled) {
    if (eventsStatus.error) feedProblems.push(`511: ${eventsStatus.error}`);
    else if (eventsStatus.fetched_at && now / 1000 - eventsStatus.fetched_at > absenceMs / 1000)
      feedProblems.push(`511 data ${Math.round((now / 1000 - eventsStatus.fetched_at) / 60)}m old`);
  }
  for (const p of cameraProviders) if (p.error) feedProblems.push(`${p.name}: ${p.error}`);
  const feeds: PipelineStage = {
    key: "feeds", label: "Feeds", lastAt: null,
    state: feedProblems.length ? "degraded" : "ok",
    detail: feedProblems.length ? feedProblems.join(" · ") : "511 / camera providers OK",
  };

  // ── Verdict ─────────────────────────────────────────────────────────────
  const parts: string[] = [];
  if (allNodesOffline) parts.push(`${radio.detail}${newestCallAt ? `, last call heard ${clockTime(newestCallAt)}` : ""}`);
  else if (radioDry) parts.push(`nodes checking in, no radio heard since ${newestCallAt ? clockTime(newestCallAt) : "—"}`);
  if (sttDry) parts.push(stt.detail);
  if (aiDry) parts.push(ai_.detail);
  for (const t of degradedTiers) parts.push(tierText(t.tier, t.state));
  const degraded = parts.length > 0;

  return {
    stages: [radio, stt, ai_, feeds],
    degraded,
    message: degraded ? parts.join(" · ") : null,
    degradedTiers,
    aiHealthReachable: reachable,
    nodesHearing,
    newestCallAt,
    staleMin: WALL_STALE_MIN,
  };
}