/**
 * Fleet connector: SSE client consuming llama-swap /api/events per enabled host.
 *
 * Ports the opencode-sse.ts reconnectDecision pattern (exponential backoff +
 * circuit-breaker) with one critical addition: **jitter**. The source pattern
 * has NO jitter, which causes thundering-herd reconnections across N hosts.
 *
 * Jitter: random 0-50% of computed delay. Pure function for testability.
 *
 * Event parsing is NEW code — llama-swap's SSE envelope (modelStatus | logData |
 * metrics | inflight) differs from the opencode SDK's Event type.
 */

import type { FastifyBaseLogger } from 'fastify';
import type { Sql } from '../db.js';

/** Add random 0-50% jitter to a delay value. */
export function addJitter(delayMs: number): number {
  const jitter = delayMs * Math.random() * 0.5;
  return delayMs + jitter;
}

export interface ReconnectPolicy {
  baseMs: number;
  maxMs: number;
  maxAttempts: number;
}

export const DEFAULT_RECONNECT_POLICY: ReconnectPolicy = {
  baseMs: 1_000,
  maxMs: 30_000,
  maxAttempts: 6,
};

export type ReconnectDecision =
  | { action: 'reconnect'; delayMs: number }
  | { action: 'give-up' };

export function reconnectDecision(
  failures: number,
  policy: ReconnectPolicy = DEFAULT_RECONNECT_POLICY,
): ReconnectDecision {
  if (failures > policy.maxAttempts) return { action: 'give-up' };
  const exp = policy.baseMs * 2 ** (failures - 1);
  const capped = Math.min(policy.maxMs, exp);
  return { action: 'reconnect', delayMs: addJitter(capped) };
}

// Per-type payload shapes, verified against the fork source
// (/opt/forks/llama-swap/internal/server/apigroup.go sendModels/sendLogData/
// sendMetrics/sendInFlight, apiModel struct at :20):
//   modelStatus -> []apiModel        (FULL-FLEET snapshot array, not a single transition)
//   logData     -> {source, data}    (field is 'data', not 'line')
//   metrics     -> []ActivityLogEntry (BARE array, tokens nested)
//   inflight    -> {total}           (host-level total, NOT per-model)
export type LlamaSweepSSEEvent =
  | { type: 'modelStatus'; data: ModelStatusEntry[] }
  | { type: 'logData'; data: LogData }
  | { type: 'metrics'; data: MetricsEntry[] }
  | { type: 'inflight'; data: InflightData };

/** One entry of the modelStatus full-fleet array (fork apiModel struct). */
export interface ModelStatusEntry {
  id: string;
  name: string;
  description: string;
  state: string;
  unlisted: boolean;
  peerID: string;
  aliases?: string[];
}

export interface LogData {
  source: string;
  data: string;
}

// Real /api/metrics shape: bare JSON array of entries with NESTED tokens.
// {id, timestamp, model, req_path, resp_status_code, tokens:{...}, duration_ms, has_capture}
// NOTE: ActivityLogEntry does NOT carry request headers or source field.
// Headers exist only in ReqRespCapture (fetched on-demand via /api/captures/:id).
// See design §7 "Implementation notes" for the discrepancy.
export interface MetricsEntry {
  id: number;
  timestamp: string;
  model: string;
  req_path: string;
  resp_status_code: number;
  tokens: {
    cache_tokens: number;
    input_tokens: number;
    output_tokens: number;
    prompt_per_second: number;
    tokens_per_second: number;
  };
  duration_ms: number;
  has_capture: boolean;
  capture?: string;
}

export interface InflightData {
  total: number;
}

export interface FleetConnectorDeps {
  isUp: () => boolean;
  sql: Sql;
  log: FastifyBaseLogger;
  onEvent: (providerId: string, event: LlamaSweepSSEEvent) => void | Promise<void>;
  onReconnectGiveUp: (providerId: string) => Promise<void>;
  sleep?: (ms: number) => Promise<void>;
  policy?: ReconnectPolicy;
}

function defaultSleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

/**
 * Parse llama-swap SSE lines.
 *
 * Real wire shape (apigroup.go):
 *   event:message
 *   data:{"type":"modelStatus","data":"<ESCAPED JSON STRING>"}
 *
 * The SSE event name is always 'message'. The discriminator is the outer JSON's
 * .type field. The payload is DOUBLE-ENCODED: JSON.parse(data) gives {type, data:string},
 * then JSON.parse(that.data) gives the actual payload.
 *
 * Returns the fully-decoded event, or null for non-data lines.
 */
export function parseSseLine(line: string): LlamaSweepSSEEvent | null {
  const trimmed = line.trim();
  if (!trimmed) return null;

  // The SSE event name is always 'event:message' -- we ignore it.
  if (trimmed.startsWith('event:')) {
    return null;
  }

  // "data: <json>" -- the only line that carries payload.
  if (trimmed.startsWith('data:')) {
    const dataStr = trimmed.slice(5).trimStart();
    if (!dataStr) return null;

    // First JSON parse: { type: "modelStatus", data: "<escaped json>" }
    let outer: { type: string; data: string };
    try {
      outer = JSON.parse(dataStr) as { type: string; data: string };
    } catch {
      return null;
    }

    if (!outer.type || typeof outer.data !== 'string' || !outer.data) {
      return null;
    }

    // Second JSON parse: the actual payload (double-encoded string).
    let inner: unknown;
    try {
      inner = JSON.parse(outer.data);
    } catch {
      return null;
    }

    return { type: outer.type, data: inner } as LlamaSweepSSEEvent;
  }

  return null;
}

export function startFleetConnector(providerId: string, baseUrl: string, deps: FleetConnectorDeps): AbortController {
  const abort = new AbortController();
  void runFleetConnector(providerId, baseUrl, abort, deps).finally(() => {
    if (abort.signal.aborted) {
      // connection dropped — cleanup handled by caller
    }
  });
  return abort;
}

export async function runFleetConnector(
  providerId: string,
  baseUrl: string,
  abort: AbortController,
  deps: FleetConnectorDeps,
): Promise<void> {
  const signal = abort.signal;
  const sleep = deps.sleep ?? defaultSleep;
  const policy = deps.policy ?? DEFAULT_RECONNECT_POLICY;
  let failures = 0;

  while (deps.isUp() && !signal.aborted) {
    const url = `${baseUrl}/api/events`;
    try {
      const res = await fetch(url, { signal });
      if (!res.ok) {
        throw new Error(`SSE connect failed: ${res.status} ${res.statusText}`);
      }

      const reader = res.body?.getReader();
      if (!reader) throw new Error('no response body');

      const decoder = new TextDecoder();
      let buffer = '';

      while (!signal.aborted) {
        const { done, value } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });

        const lines = buffer.split('\n');
        buffer = lines.pop() ?? '';

        for (const line of lines) {
          if (signal.aborted) break;
          const event = parseSseLine(line);
          if (!event) continue;

          try {
            await Promise.resolve(deps.onEvent(providerId, event));
          } catch (err) {
            deps.log.error({ providerId, err: (err as Error).message }, 'fleet: onEvent failed');
          }
        }
      }

      // Clean stream end — healthy reconnect at base delay (pre-hardening).
      failures = 0;
      if (deps.isUp() && !signal.aborted) {
        await sleep(policy.baseMs);
      }
    } catch (err) {
      if (!deps.isUp() || signal.aborted) break;
      failures += 1;
      const decision = reconnectDecision(failures, policy);
      deps.log.warn(
        { providerId, failures, action: decision.action, err: (err as Error).message },
        'fleet: SSE error; reconnecting',
      );
      if (decision.action === 'give-up') {
        deps.log.warn({ providerId, failures }, 'fleet: SSE reconnect gave up (circuit breaker)');
        await deps.onReconnectGiveUp(providerId);
        break;
      }
      await sleep(decision.delayMs);
    }
  }
}
