import { spawnSync } from "node:child_process";
import { homedir } from "node:os";
import { join } from "node:path";
import { z } from "zod";
import { isValidHostname } from "./capability";
import type {
  HostMachineMetrics,
  HostMetricsSample,
  HostMetricsSampler,
  TelemetryGap,
} from "./metrics";

// Measured cost of one telemetry read over tailscale is ~3s; 5s left no margin for a cold
// path and silently produced a host with no numbers.
export const TELEMETRY_SSH_TIMEOUT_MS = 8_000;

type HostSample =
  | { host: string; metrics: HostMachineMetrics; reason: null }
  | { host: string; metrics: null; reason: string };

const RemoteHostMetricsSchema = z.object({
  sampledAtEpochMs: z.number().int().nonnegative(),
  sessions: z.number().int().nonnegative().nullable(),
  ciJobsRunning: z.number().int().nonnegative().nullable().optional(),
  // Optional so a box on the previous telemetry build still parses: an absent field is a
  // coverage gap on one counter, never a reason to discard the whole sample.
  remoteWork: z.object({
    agentSeats: z.number().int().nonnegative(),
    remoteBuildJobs: z.number().int().nonnegative(),
    offloadShells: z.number().int().nonnegative(),
  }).strict().nullable().optional(),
  load: z.number().finite(),
  cores: z.number().int().nonnegative(),
  coreLoadPercent: z.array(z.number().finite()),
  memUsedBytes: z.number().int().nonnegative(),
  memTotalBytes: z.number().int().nonnegative(),
  swapUsedBytes: z.number().int().nonnegative(),
  swapTotalBytes: z.number().int().nonnegative(),
  netRxBytesPerSecond: z.number().finite().nonnegative(),
  netTxBytesPerSecond: z.number().finite().nonnegative(),
  disks: z.array(z.object({
    mountpoint: z.string().min(1),
    freeBytes: z.number().int().nonnegative(),
    sizeBytes: z.number().int().nonnegative(),
  }).strict()),
  temperatures: z.object({
    pkg: z.number().finite(),
    max: z.number().finite(),
    crit: z.number().finite(),
  }).strict(),
  guard: z.object({
    memoryStallSome60: z.number().finite().nonnegative(),
    memoryStallFull60: z.number().finite().nonnegative(),
    memoryStallFull300: z.number().finite().nonnegative(),
    tmpUsedBytes: z.number().int().nonnegative(),
    tmpSizeBytes: z.number().int().nonnegative(),
    workSlices: z.array(z.object({
      slice: z.string().min(1),
      memoryBytes: z.number().int().nonnegative(),
      oomKillTotal: z.number().int().nonnegative(),
      pidsCurrent: z.number().int().nonnegative(),
    }).strict()),
  }).strict(),
}).strict();

export type SyncSpawnExec = (
  cmd: string[],
  options: { shell: false; timeoutMs?: number },
) => { exitCode: number | null; stdout: string; stderr: string };

const REMOTE_METRICS_PATH = ".local/state/overdeck/buildbox-telemetry.json";

/**
 * A summary older than this is refused, in either direction. The box republishes every 15s,
 * so anything past this bound means its sampler stopped or its clock disagrees; both make the
 * numbers unusable for placement and for alerting on what a box is doing now.
 */
const MAX_SAMPLE_AGE_MS = 90_000;

function defaultSyncSpawn(
  cmd: string[],
  options: { shell: false; timeoutMs?: number },
): { exitCode: number | null; stdout: string; stderr: string } {
  const [command, ...args] = cmd;
  if (!command) throw new Error("spawn command must not be empty");
  const result = spawnSync(command, args, {
    encoding: "utf8",
    timeout: options.timeoutMs,
  });
  return {
    exitCode: result.status,
    stdout: result.stdout ?? "",
    stderr: result.stderr ?? "",
  };
}

function sshArgv(host: string, home: string): string[] {
  return [
    "ssh",
    "-p",
    "2222",
    "-i",
    join(home, ".ssh/id_ed25519_buildbox"),
    host,
    "cat",
    REMOTE_METRICS_PATH,
  ];
}

function sampleHost(
  host: string,
  exec: SyncSpawnExec,
  home: string,
  timeoutMs: number,
): HostSample {
  if (!isValidHostname(host)) {
    return { host, metrics: null, reason: `not a valid hostname: ${host}` };
  }

  let result: { exitCode: number | null; stdout: string; stderr: string };
  try {
    result = exec(sshArgv(host, home), { shell: false, timeoutMs });
  } catch (err) {
    return { host, metrics: null, reason: `cannot spawn ssh: ${(err as Error).message}` };
  }

  if (result.exitCode !== 0) {
    const detail = result.stderr.trim().split("\n").at(-1);
    return {
      host,
      metrics: null,
      reason: result.exitCode === null
        ? `ssh read of ${REMOTE_METRICS_PATH} timed out after ${timeoutMs}ms`
        : `ssh read of ${REMOTE_METRICS_PATH} exited ${result.exitCode}${detail ? `: ${detail}` : ""}`,
    };
  }

  const line = result.stdout.trim().split("\n").at(-1) ?? "";
  if (!line) {
    return { host, metrics: null, reason: `${REMOTE_METRICS_PATH} is empty on ${host}` };
  }

  let json: unknown;
  try {
    json = JSON.parse(line);
  } catch (err) {
    return {
      host,
      metrics: null,
      reason: `${REMOTE_METRICS_PATH} is not valid JSON: ${(err as Error).message}`,
    };
  }

  const parsed = RemoteHostMetricsSchema.safeParse(json);
  if (!parsed.success) {
    const issue = parsed.error.issues[0];
    return {
      host,
      metrics: null,
      reason: `${REMOTE_METRICS_PATH} does not match the telemetry schema${
        issue ? `: ${issue.path.join(".") || "(root)"} ${issue.message}` : ""
      }`,
    };
  }

  const { sampledAtEpochMs, remoteWork, ...machine } = parsed.data;
  const ageMs = Date.now() - sampledAtEpochMs;
  if (Math.abs(ageMs) > MAX_SAMPLE_AGE_MS) {
    return {
      host,
      metrics: null,
      reason: ageMs > 0
        ? `telemetry is ${Math.round(ageMs / 1000)}s old (max ${MAX_SAMPLE_AGE_MS / 1000}s)`
        : `telemetry is ${Math.round(-ageMs / 1000)}s in the future — clock disagreement`,
    };
  }

  return { host, metrics: { host, ...machine, remoteWork: remoteWork ?? null }, reason: null };
}

/** Reads the summary each buildbox publishes for itself. */
export function createSshHostMetricsSampler(
  exec: SyncSpawnExec = defaultSyncSpawn,
  home: string = homedir(),
  timeoutMs: number = TELEMETRY_SSH_TIMEOUT_MS,
): HostMetricsSampler {
  return {
    sample(hosts) {
      const metrics: HostMachineMetrics[] = [];
      const gaps: TelemetryGap[] = [];
      for (const host of hosts) {
        const sample = sampleHost(host, exec, home, timeoutMs);
        if (sample.metrics) metrics.push(sample.metrics);
        else gaps.push({ host, reason: sample.reason, at: new Date().toISOString() });
      }
      return { metrics, gaps };
    },
  };
}

/** Refreshes host metrics on an interval so /api/v1/query never blocks on SSH. */
export function createBackgroundHostMetricsSampler(
  inner: HostMetricsSampler,
  hostSource: () => readonly string[],
  intervalMs = 30_000,
): HostMetricsSampler & { stop(): void } {
  let cache: HostMetricsSample = { metrics: [], gaps: [] };
  const refresh = () => {
    const hosts = hostSource();
    if (hosts.length === 0) {
      cache = { metrics: [], gaps: [] };
      return;
    }
    cache = inner.sample(hosts);
  };
  const timer = setInterval(refresh, intervalMs);
  void Promise.resolve().then(refresh);
  return {
    sample(hosts) {
      const wanted = new Set(hosts);
      return {
        metrics: cache.metrics.filter((metric) => wanted.has(metric.host)),
        // A host with neither metrics nor a recorded gap has never been sampled; say so
        // rather than letting it read as healthy-with-no-numbers.
        gaps: [...hosts]
          .filter((host) => !cache.metrics.some((metric) => metric.host === host))
          .map((host) =>
            cache.gaps.find((gap) => gap.host === host)
              ?? { host, reason: "not sampled yet", at: null }
          ),
      };
    },
    stop() {
      clearInterval(timer);
    },
  };
}