/**
 * BackendStatusRegistry — in-memory cache of "node" health for the
 * NodeStatusWidget (Side Info Panel, Phase B).
 *
 * A "node" is either a direct worker (the worker itself IS the node) or a
 * physical backend behind a proxy worker (e.g. a LiteLLM deployment).
 * This registry probes both shapes at a fixed cadence and exposes the
 * latest snapshot via getAll() / subscribe().
 *
 * Design notes (kept short — long-form in
 * docs/superpowers/specs/2026-05-18-multi-team-gpu-pool-and-node-status-design.md):
 *
 * - Process-local. There is exactly one registry per AAO process; cross-
 *   process sharing is out of scope (each AAO probes its own workers).
 * - Polling is timer-driven. Subscribers receive snapshots on each tick
 *   AND immediately on subscribe so the UI can paint without waiting up
 *   to a full interval.
 * - Probe target list is rebuilt from the WorkerDef list every tick, so
 *   config edits propagate without an explicit invalidate call.
 * - Failures are isolated per-node: one probe rejecting never poisons
 *   another node's status. lastProbeError carries the failure text for
 *   the widget to render a degraded badge.
 * - Phase B uses a fixed 5s tick (see Open Question #1 in the design
 *   doc — dynamic visibility-aware polling is deferred). The cadence is
 *   exposed as a constructor option mainly for tests.
 */

import type { WorkerDef } from '../config.js';
import { logger } from '../logger.js';

export interface NodeStatus {
  /** Stable identifier: workerId for direct workers, deployment id for proxy backends. */
  nodeId: string;
  /** AAO worker this node belongs to. */
  workerId: string;
  source: 'direct' | 'proxy';
  online: boolean;
  busy: boolean;
  busySlots: number;
  totalSlots: number;
  loadedModel: string | null;
  throughputTps: number | null;
  /** ISO 8601 timestamp of the latest probe that touched this node. */
  lastSeen: string;
  /** Set only when the most recent probe failed. */
  lastProbeError?: string;
}

export type NodeStatusListener = (statuses: NodeStatus[]) => void;
export type Unsubscribe = () => void;

/**
 * Optional context passed to probes so the registry can cancel an
 * in-flight cycle on shutdown. Probes may ignore it; existing probes
 * pass `signal` through to fetch so an external abort cancels the
 * pending HTTP request immediately instead of waiting out the per-probe
 * timeout.
 */
export interface ProbeContext {
  signal?: AbortSignal;
}

/**
 * Probes a direct worker (llama-server compatible). Returns a single
 * NodeStatus whose nodeId == workerId.
 */
export type DirectWorkerProbe = (worker: WorkerDef, ctx?: ProbeContext) => Promise<NodeStatus>;

/**
 * Probes a proxy worker (LiteLLM Proxy). Returns one NodeStatus per
 * deployment the proxy reports.
 */
export type ProxyWorkerProbe = (worker: WorkerDef, ctx?: ProbeContext) => Promise<NodeStatus[]>;

export interface BackendStatusRegistryDeps {
  getWorkers: () => WorkerDef[];
  probeDirect: DirectWorkerProbe;
  probeProxy: ProxyWorkerProbe;
  /**
   * Polling interval (ms) used while at least one subscriber is active —
   * either an in-process subscribe() listener or a recent
   * noteSubscriberActivity() ping from the dashboard API. Default: 5000.
   */
  pollIntervalMs?: number;
  /**
   * Polling interval (ms) used when no subscribers have been active for
   * `subscriberActiveWindowMs`. The dashboard isn't open and no one is
   * watching, so we throttle probes to spare upstream GPUs. Default:
   * 30000. Must be >= pollIntervalMs.
   */
  idlePollIntervalMs?: number;
  /**
   * How long after the last noteSubscriberActivity()/subscribe call we
   * keep treating the registry as "actively watched". Default: 30000.
   * Set just above the dashboard refetchInterval so a UI that's polling
   * every 5s never accidentally drops into idle mode between ticks.
   */
  subscriberActiveWindowMs?: number;
  /** Maximum number of probes running at once. Default: 3. */
  maxConcurrency?: number;
  /**
   * Optional clock injection for deterministic tests. Defaults to
   * `() => new Date().toISOString()`.
   */
  now?: () => string;
  /**
   * Monotonic clock for cadence decisions (ms). Injectable for tests
   * that can't use Date.now() with fake timers (vitest's fake timers
   * advance performance.now but not the wall clock unless you also use
   * setSystemTime). Defaults to `() => Date.now()`.
   */
  monotonicNowMs?: () => number;
}

export interface BackendStatusRegistry {
  start(): void;
  /**
   * Cancel any in-flight probe cycle and await its settlement. Resolves
   * once the registry is fully quiesced so shutdown handlers can chain
   * cleanly without leaving fetches dangling past process exit.
   */
  stop(): Promise<void>;
  /** Returns the latest cached snapshot. Safe to call before the first probe completes (returns []). */
  getAll(): NodeStatus[];
  getByNodeId(nodeId: string): NodeStatus | null;
  /** Subscribe to snapshot updates. The listener is invoked synchronously with the current snapshot. */
  subscribe(listener: NodeStatusListener): Unsubscribe;
  /** Force a probe cycle now (skips the polling interval). Useful for tests and the "refresh" button. */
  refresh(): Promise<void>;
  /**
   * Hint that a UI client just polled the registry. Used to bias the
   * polling cadence: active subscribers (recent GET /node-status hits)
   * keep the registry at `pollIntervalMs`; long silences drop to
   * `idlePollIntervalMs` (default 30s). Optional so legacy tests can
   * stub the interface without implementing it.
   */
  noteSubscriberActivity?(): void;
}

export function createBackendStatusRegistry(deps: BackendStatusRegistryDeps): BackendStatusRegistry {
  const pollIntervalMs = Math.max(500, deps.pollIntervalMs ?? 5000);
  const idlePollIntervalMs = Math.max(pollIntervalMs, deps.idlePollIntervalMs ?? 30_000);
  const subscriberActiveWindowMs = Math.max(pollIntervalMs, deps.subscriberActiveWindowMs ?? 30_000);
  const maxConcurrency = Math.max(1, deps.maxConcurrency ?? 3);
  const now = deps.now ?? (() => new Date().toISOString());
  const monotonicNowMs = deps.monotonicNowMs ?? (() => Date.now());

  let cache: NodeStatus[] = [];
  const listeners = new Set<NodeStatusListener>();
  // Tracks the most recent moment a subscriber signalled interest.
  // We use Number.NEGATIVE_INFINITY (not 0) so the very first scheduling
  // decision is unambiguously "no subscribers yet → idle" regardless of
  // monotonic clock origin.
  let lastSubscriberAt = Number.NEGATIVE_INFINITY;
  let timer: NodeJS.Timeout | null = null;
  let stopped = true;
  // Avoid overlapping ticks: if a probe cycle takes longer than the
  // interval (slow upstream), we skip the next tick rather than stack
  // requests on the same target.
  let inflight: Promise<void> | null = null;
  // Per-cycle AbortController, exposed via cycleAbort so stop() can
  // cancel pending probes and avoid waiting out per-probe timeouts on
  // shutdown.
  let cycleAbort: AbortController | null = null;

  function notify(snapshot: NodeStatus[]): void {
    for (const l of listeners) {
      try {
        l(snapshot);
      } catch (err) {
        logger.warn(`[backend-status-registry] listener threw: ${(err as Error).message}`);
      }
    }
  }

  async function runOnce(): Promise<void> {
    const workers = deps.getWorkers().filter(w => typeof w.id === 'string' && w.id.length > 0);
    cycleAbort = new AbortController();
    const ctx: ProbeContext = { signal: cycleAbort.signal };
    const tasks: Array<() => Promise<NodeStatus[]>> = [];
    for (const w of workers) {
      if (w.proxy === true) {
        tasks.push(() => deps.probeProxy(w, ctx).catch(err => [
          buildErrorStatus(w.id, w.id, 'proxy', err, now()),
        ]));
      } else {
        tasks.push(() => deps.probeDirect(w, ctx)
          .then(s => [s])
          .catch(err => [buildErrorStatus(w.id, w.id, 'direct', err, now())]));
      }
    }
    // Bounded parallelism: simple "next task" pool to avoid pulling in p-limit.
    const results: NodeStatus[] = [];
    let cursor = 0;
    async function worker(): Promise<void> {
      while (true) {
        const idx = cursor++;
        if (idx >= tasks.length) return;
        const t = tasks[idx]!;
        const arr = await t();
        results.push(...arr);
      }
    }
    const workersCount = Math.min(maxConcurrency, tasks.length);
    try {
      await Promise.all(Array.from({ length: workersCount }, () => worker()));
    } finally {
      cycleAbort = null;
    }

    // Don't overwrite cache or notify subscribers with a partial /
    // aborted result — leaving the previous snapshot in place is more
    // honest than synthesising "all offline" rows on shutdown.
    if (stopped) return;
    cache = results;
    notify(cache);
  }

  function activeNow(): boolean {
    // Active iff at least one in-process listener OR a recent HTTP
    // subscriber within the rolling activity window. The HTTP path
    // dominates in practice (the dashboard widget polls the REST
    // endpoint rather than wiring into subscribe() directly); the
    // listener-count check is the fast path for any in-process consumer
    // we add later.
    if (listeners.size > 0) return true;
    return monotonicNowMs() - lastSubscriberAt < subscriberActiveWindowMs;
  }

  function nextIntervalMs(): number {
    return activeNow() ? pollIntervalMs : idlePollIntervalMs;
  }

  function scheduleNext(): void {
    if (stopped) return;
    timer = setTimeout(async () => {
      if (stopped) return;
      if (inflight) {
        // Previous tick still running; reschedule and skip this one.
        scheduleNext();
        return;
      }
      inflight = runOnce().catch(err => {
        logger.warn(`[backend-status-registry] tick failed: ${(err as Error).message}`);
      }).finally(() => {
        inflight = null;
      });
      await inflight;
      scheduleNext();
    }, nextIntervalMs());
    // Don't keep the event loop alive solely for the registry timer.
    if (typeof timer.unref === 'function') timer.unref();
  }

  function rescheduleIfBandChanged(prevBandActive: boolean): void {
    // Called when a subscriber transition could flip the next-tick band.
    // We only restart the timer when (a) the registry is running, (b) no
    // probe is currently in flight (it would respect the new cadence at
    // the next scheduleNext anyway), and (c) the band actually flipped
    // from idle → active. Going active → idle doesn't need to interrupt
    // the current timer — letting the next active tick fire early is
    // harmless and avoids subtle wakeups when subscribers churn rapidly.
    if (stopped) return;
    const nowActive = activeNow();
    if (prevBandActive === nowActive) return;
    if (!nowActive) return; // active → idle: no immediate wake.
    if (timer) {
      clearTimeout(timer);
      timer = null;
    }
    scheduleNext();
  }

  return {
    start(): void {
      if (!stopped) return;
      stopped = false;
      // Kick off the first probe immediately so subscribers see data
      // within ~1 RTT, not after pollIntervalMs.
      inflight = runOnce().catch(err => {
        logger.warn(`[backend-status-registry] initial tick failed: ${(err as Error).message}`);
      }).finally(() => {
        inflight = null;
      });
      scheduleNext();
      logger.info(`[backend-status-registry] started activeIntervalMs=${pollIntervalMs} idleIntervalMs=${idlePollIntervalMs} activeWindowMs=${subscriberActiveWindowMs} concurrency=${maxConcurrency}`);
    },
    async stop(): Promise<void> {
      stopped = true;
      if (timer) {
        clearTimeout(timer);
        timer = null;
      }
      // Abort any in-flight cycle so pending fetches reject immediately
      // instead of blocking shutdown for up to (per-probe timeout) ×
      // (workers). The probe error handlers in runOnce swallow the
      // abort error, so inflight always resolves cleanly.
      if (cycleAbort) {
        try { cycleAbort.abort(); } catch { /* ignore */ }
      }
      if (inflight) {
        try { await inflight; } catch { /* swallowed by runOnce error handlers */ }
      }
      logger.info('[backend-status-registry] stopped');
    },
    getAll(): NodeStatus[] {
      return cache.slice();
    },
    getByNodeId(nodeId: string): NodeStatus | null {
      return cache.find(s => s.nodeId === nodeId) ?? null;
    },
    subscribe(listener: NodeStatusListener): Unsubscribe {
      const wasActive = activeNow();
      listeners.add(listener);
      lastSubscriberAt = monotonicNowMs();
      rescheduleIfBandChanged(wasActive);
      // Synchronous initial delivery so React subscribers can paint
      // without waiting for the first polling tick.
      try {
        listener(cache.slice());
      } catch (err) {
        logger.warn(`[backend-status-registry] initial deliver threw: ${(err as Error).message}`);
      }
      return () => {
        listeners.delete(listener);
        // Note: we don't reschedule on unsubscribe. The next scheduled
        // tick will pick the idle cadence on its own — interrupting now
        // would only matter if we wanted to lengthen the current
        // pending timer, which isn't worth the wakeup churn.
      };
    },
    noteSubscriberActivity(): void {
      const wasActive = activeNow();
      lastSubscriberAt = monotonicNowMs();
      rescheduleIfBandChanged(wasActive);
    },
    async refresh(): Promise<void> {
      if (inflight) {
        await inflight;
        return;
      }
      inflight = runOnce().finally(() => {
        inflight = null;
      });
      await inflight;
    },
  };
}

function buildErrorStatus(
  nodeId: string,
  workerId: string,
  source: 'direct' | 'proxy',
  err: unknown,
  ts: string,
): NodeStatus {
  return {
    nodeId,
    workerId,
    source,
    online: false,
    busy: false,
    busySlots: 0,
    totalSlots: 0,
    loadedModel: null,
    throughputTps: null,
    lastSeen: ts,
    lastProbeError: err instanceof Error ? err.message : String(err),
  };
}
