/**
 * useControlStream: second app-level WS singleton for BooControl.
 *
 * Own React context + connection guard. Targets proxied /api/control/ws.
 * Client discards deltas with seq <= snapshot_seq per-host.
 *
 * This is NOT the same as useUserEvents — it's a separate WS connection.
 */

import { createContext, useContext, useRef, useCallback, useEffect, useState } from 'react';

export interface ControlFleetHost {
  providerId: string;
  liveness: 'connected' | 'down';
  lastSeenAt: string | null;
  seq: number;
  models: Array<{
    model: string;
    state: string;
    ts: string;
    ttlDeadline: string | null;
    inflight: number;
  }>;
}

export interface ControlRequestEntry {
  id: number;
  providerId: string;
  ts: string;
  model: string | null;
  reqPath: string | null;
  statusCode: number | null;
  durationMs: number | null;
}

export interface ControlPerfSample {
  providerId: string;
  ts: string;
  gpu: unknown;
  sys: unknown;
}

export interface ControlLogEntry {
  providerId: string;
  source: 'proxy' | 'upstream' | 'model';
  line: string;
  /** Stamped at WS-ingest time (the frame carries no ts); far better than render-time now(). */
  ts: string;
}

export type ControlConnection = 'connecting' | 'live' | 'reconnecting' | 'down';

export interface ControlJob {
  jobType: 'bench' | 'eval' | 'action';
  jobId: string;
  status: 'queued' | 'running' | 'completed' | 'failed';
  detail?: Record<string, unknown>;
  ts: string;
}

export type ControlFleetDelta = {
  type: 'control_fleet';
  seq: number;
  hosts: ControlFleetHost[];
};

export type ControlActivityFrame = {
  type: 'control_activity';
  seq: number;
  providerId: string;
  entry: ControlRequestEntry;
};

export type ControlPerfFrame = {
  type: 'control_perf';
  seq: number;
  providerId: string;
  ts: string;
  gpu: unknown;
  sys: unknown;
};

export type ControlLogFrame = {
  type: 'control_log';
  seq: number;
  providerId: string;
  source: 'proxy' | 'upstream' | 'model';
  line: string;
  ts?: string;
};

export type ControlJobFrame = {
  type: 'control_job';
  seq: number;
  jobType: 'bench' | 'eval' | 'action';
  jobId: string;
  status: 'queued' | 'running' | 'completed' | 'failed';
  detail?: Record<string, unknown>;
};

export type ControlFrame =
  | ControlFleetDelta
  | ControlActivityFrame
  | ControlPerfFrame
  | ControlLogFrame
  | ControlJobFrame;

function isValidHost(h: unknown): h is ControlFleetHost {
  if (!h || typeof h !== 'object') return false;
  const obj = h as Record<string, unknown>;
  return (
    typeof obj.providerId === 'string' &&
    ['connected', 'down'].includes(obj.liveness as string) &&
    (obj.lastSeenAt === null || typeof obj.lastSeenAt === 'string') &&
    typeof obj.seq === 'number' &&
    Array.isArray(obj.models)
  );
}

function isControlFleetDelta(data: unknown): data is ControlFleetDelta {
  if (!data || typeof data !== 'object') return false;
  const obj = data as Record<string, unknown>;
  return (
    obj.type === 'control_fleet' &&
    typeof obj.seq === 'number' &&
    Array.isArray(obj.hosts) &&
    obj.hosts.every(isValidHost)
  );
}

function isControlActivityFrame(data: unknown): data is ControlActivityFrame {
  if (!data || typeof data !== 'object') return false;
  const obj = data as Record<string, unknown>;
  return (
    obj.type === 'control_activity' &&
    typeof obj.seq === 'number' &&
    typeof obj.providerId === 'string' &&
    typeof obj.entry === 'object' &&
    obj.entry !== null
  );
}

function isControlPerfFrame(data: unknown): data is ControlPerfFrame {
  if (!data || typeof data !== 'object') return false;
  const obj = data as Record<string, unknown>;
  return (
    obj.type === 'control_perf' &&
    typeof obj.seq === 'number' &&
    typeof obj.providerId === 'string' &&
    typeof obj.ts === 'string'
  );
}

function isControlLogFrame(data: unknown): data is ControlLogFrame {
  if (!data || typeof data !== 'object') return false;
  const obj = data as Record<string, unknown>;
  return (
    obj.type === 'control_log' &&
    typeof obj.seq === 'number' &&
    typeof obj.providerId === 'string' &&
    ['proxy', 'upstream', 'model'].includes(obj.source as string) &&
    typeof obj.line === 'string'
  );
}

function isControlJobFrame(data: unknown): data is ControlJobFrame {
  if (!data || typeof data !== 'object') return false;
  const obj = data as Record<string, unknown>;
  return (
    obj.type === 'control_job' &&
    typeof obj.seq === 'number' &&
    ['bench', 'eval', 'action'].includes(obj.jobType as string) &&
    typeof obj.jobId === 'string' &&
    ['queued', 'running', 'completed', 'failed'].includes(obj.status as string)
  );
}

export interface ControlStreamState {
  hosts: ControlFleetHost[];
  requests: ControlRequestEntry[];
  perfSamples: ControlPerfSample[];
  logs: ControlLogEntry[];
  jobs: ControlJob[];
  /** Live control-WS connection state, for the cockpit status pill. */
  connection: ControlConnection;
}

const ControlContext = createContext<ControlStreamState | null>(null);

export function useControlStream(): ControlStreamState {
  const state = useContext(ControlContext);
  if (!state) throw new Error('useControlStream must be used within ControlProvider');
  return state;
}

export function ControlProvider({ children }: { children: React.ReactNode }) {
  const [state, setState] = useState<ControlStreamState>({
    hosts: [],
    requests: [],
    perfSamples: [],
    logs: [],
    jobs: [],
    connection: 'connecting',
  });
  const wsRef = useRef<WebSocket | null>(null);
  const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
  const snapshotSeqRef = useRef(0);
  const hasSnapshotRef = useRef(false);
  const backoffRef = useRef(5_000);

  const connect = useCallback(() => {
    if (wsRef.current) return;
    const ws = new WebSocket(`${window.location.protocol === 'https:' ? 'wss' : 'ws'}://${window.location.host}/api/control/ws`);
    wsRef.current = ws;

    ws.onopen = () => {
      snapshotSeqRef.current = 0;
      hasSnapshotRef.current = false;
      backoffRef.current = 5_000;
      setState((prev) => ({ ...prev, connection: 'live' }));
    };

    ws.onmessage = (event) => {
      try {
        const data: unknown = JSON.parse(event.data);
        if (typeof data !== 'object' || !data || !('type' in data)) return;
        if ((data as Record<string, unknown>).type === 'ping') return; // heartbeat

        // A3: type-guard each frame shape before applying — no 'as unknown as' casts
        if (isControlFleetDelta(data)) {
          if (!hasSnapshotRef.current) {
            // First frame after connect is the snapshot.
            hasSnapshotRef.current = true;
            snapshotSeqRef.current = data.seq;
            setState((prev) => ({ ...prev, hosts: data.hosts }));
          } else {
            // Delta: merge by providerId so a delta for one host does not wipe the others.
            if (data.seq > snapshotSeqRef.current) {
              setState((prev) => {
                const merged = [...prev.hosts];
                for (const dh of data.hosts) {
                  const idx = merged.findIndex((h) => h.providerId === dh.providerId);
                  if (idx >= 0) {
                    merged[idx] = dh;
                  } else {
                    merged.push(dh);
                  }
                }
                return { ...prev, hosts: merged };
              });
            }
          }
        } else if (isControlActivityFrame(data)) {
          setState((prev) => ({
            ...prev,
            requests: [data.entry, ...prev.requests].slice(0, 500),
          }));
        } else if (isControlPerfFrame(data)) {
          setState((prev) => ({
            ...prev,
            perfSamples: [...prev.perfSamples, { providerId: data.providerId, ts: data.ts, gpu: data.gpu, sys: data.sys }].slice(-500),
          }));
        } else if (isControlLogFrame(data)) {
          // Prefer the server-emit ts; fall back to ingest time for old frames.
          const ts = (data as ControlLogFrame).ts ?? new Date().toISOString();
          setState((prev) => ({
            ...prev,
            logs: [...prev.logs, { providerId: data.providerId, source: data.source, line: data.line, ts }].slice(-1000),
          }));
        } else if (isControlJobFrame(data)) {
          setState((prev) => {
            // Dedupe by jobId: keep one entry per job, updated to its latest status.
            // Preserve the original insertion ts so the job doesn't jump to the top
            // of the Jobs list on every status tick (ts = when first seen).
            const existing = prev.jobs.find((j) => j.jobId === data.jobId);
            const ts = existing?.ts ?? new Date().toISOString();
            const others = prev.jobs.filter((j) => j.jobId !== data.jobId);
            const job: ControlJob = { jobType: data.jobType, jobId: data.jobId, status: data.status, detail: data.detail, ts };
            return { ...prev, jobs: [...others, job].slice(-200) };
          });
        }
        // Unknown frame types are silently dropped (fail-closed)
      } catch {
        // Ignore parse errors
      }
    };

    ws.onclose = () => {
      wsRef.current = null;
      // A6 fix: exponential backoff instead of fixed 5s delay.
      const delay = backoffRef.current;
      // Once backoff has grown past the first couple of retries, call it 'down'.
      setState((prev) => ({ ...prev, connection: backoffRef.current >= 20_000 ? 'down' : 'reconnecting' }));
      backoffRef.current = Math.min(30_000, backoffRef.current * 2);
      reconnectTimerRef.current = setTimeout(connect, delay);
    };

    ws.onerror = () => {
      ws.close();
    };
  }, []);

  useEffect(() => {
    connect();
    return () => {
      if (wsRef.current) {
        wsRef.current.close();
        wsRef.current = null;
      }
      if (reconnectTimerRef.current) {
        clearTimeout(reconnectTimerRef.current);
      }
    };
  }, [connect]);

  return <ControlContext.Provider value={state}>{children}</ControlContext.Provider>;
}
