import { spawn } from "node:child_process";
import { closeSync, createWriteStream, fsyncSync, mkdirSync, openSync, readFileSync, readdirSync, renameSync, rmSync, unlinkSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { spawnSync } from "node:child_process";

export const SUPERVISOR_REVISION = "command-supervisor@1";
const DEFAULT_LEASE_MS = 60_000;
const DEFAULT_TERMINAL_GRACE_MS = 5_000;
const POLL_MS = 25;
const CLEANUP_TERM_MS = 100;
const CLEANUP_KILL_MS = 100;
const MAX_EVENT_BYTES = 64 * 1024;
const EVENT_TYPES = new Set(["heartbeat", "progress", "terminal"]);

export function monotonicMs() {
  return Number(process.hrtime.bigint() / 1_000_000n);
}

function sleepMs(ms) {
  return new Promise((resolve) => setTimeout(resolve, Math.max(0, ms)));
}

function atomicWrite(path, value) {
  const dir = dirname(path);
  mkdirSync(dir, { recursive: true });
  const tmp = `${path}.tmp-${process.pid}-${process.hrtime.bigint()}`;
  let fd;
  try {
    fd = openSync(tmp, "wx", 0o600);
    writeFileSync(fd, value);
    fsyncSync(fd);
    closeSync(fd);
    fd = undefined;
    renameSync(tmp, path);
  } finally {
    if (fd !== undefined) closeSync(fd);
    try { unlinkSync(tmp); } catch {}
  }
}

function normalizeInteger(value, name, { min = 0 } = {}) {
  if (!Number.isSafeInteger(value) || value < min) throw new Error(`${name} must be an integer >= ${min}`);
  return value;
}

export function validateDeclaredEvent(event, expectedSeq) {
  if (!event || typeof event !== "object" || Array.isArray(event)) throw new Error("event must be a JSON object");
  const seq = normalizeInteger(event.seq, "event.seq", { min: 1 });
  if (seq !== expectedSeq) throw new Error(`event sequence gap: expected ${expectedSeq}, received ${seq}`);
  if (typeof event.type !== "string" || !EVENT_TYPES.has(event.type)) throw new Error("event.type is invalid");
  normalizeInteger(event.lease_ms, "event.lease_ms", { min: 1 });
  if (event.type === "terminal") normalizeInteger(event.rc, "event.rc", { min: 0 });
  return event;
}

// kill(-pgid, 0) succeeds while the group still holds unreaped zombies.
export function groupHasRunnableMember(pid, { listPids = defaultListPids, readStat = defaultReadStat } = {}) {
  let entries;
  try { entries = listPids(); } catch { return true; }
  for (const entry of entries) {
    if (!/^\d+$/.test(entry)) continue;
    let stat;
    try { stat = readStat(entry); } catch { continue; }
    if (typeof stat !== "string") continue;
    const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
    if (fields[0] === "Z") continue;
    if (Number(fields[2]) === pid) return true;
  }
  return false;
}

function defaultListPids() {
  return readdirSync("/proc");
}

function defaultReadStat(entry) {
  return readFileSync(`/proc/${entry}/stat`, "utf8");
}

function processGroupAlive(pid) {
  try {
    process.kill(-pid, 0);
  } catch (error) {
    return error?.code === "EPERM";
  }
  return groupHasRunnableMember(pid);
}

export function spawnErrorStatus(error) {
  switch (error?.code) {
    case "ENOENT": return 127;
    case "EACCES":
    case "EPERM":
    case "ENOEXEC": return 126;
    default: return 125;
  }
}

function processTree(pid) {
  if (!Number.isInteger(pid) || pid <= 0) return "process-tree-unavailable: command never started (no pid)";
  try {
    const result = spawnSync("ps", ["-o", "pid=,ppid=,pgid=,stat=,cmd=", "--forest", "-g", String(pid)], { encoding: "utf8" });
    return String(result.stdout || result.stderr || "").trim();
  } catch (error) {
    return `process-tree-unavailable: ${error.message}`;
  }
}

function killGroup(pid, signal) {
  if (!pid) return;
  try { process.kill(-pid, signal); } catch (error) { if (!['ESRCH', 'EPERM'].includes(error?.code)) throw error; }
  try { process.kill(pid, signal); } catch (error) { if (!['ESRCH', 'EPERM'].includes(error?.code)) throw error; }
}

function createOwnedCgroup(pid) {
  const parent = process.env.SUPERVISOR_CGROUP_PARENT;
  const explicit = process.env.SUPERVISOR_CGROUP_PATH;
  if (explicit) return { path: explicit, owned: false };
  if (!parent) return null;
  const path = join(parent, `command-supervisor-${pid}`);
  try {
    mkdirSync(path, { recursive: false });
    writeFileSync(join(path, "cgroup.procs"), `${pid}\n`);
    return { path, owned: true };
  } catch {
    try { rmSync(path, { recursive: false, force: true }); } catch {}
    return null;
  }
}

function killCgroup(cgroup) {
  if (!cgroup?.path) return;
  try { writeFileSync(join(cgroup.path, "cgroup.kill"), "1\n"); } catch {}
}

function removeOwnedCgroup(cgroup) {
  if (!cgroup?.owned) return;
  try { rmSync(cgroup.path, { recursive: false, force: true }); } catch {}
}

async function terminateChild(child, cgroup, sleep = sleepMs) {
  killCgroup(cgroup);
  killGroup(child?.pid, "SIGTERM");
  await sleep(CLEANUP_TERM_MS);
  killCgroup(cgroup);
  killGroup(child?.pid, "SIGKILL");
  await sleep(CLEANUP_KILL_MS);
}

function childExitPromise(child) {
  return new Promise((resolve) => {
    let settled = false;
    const done = (result) => {
      if (settled) return;
      settled = true;
      resolve(result);
    };
    child.once("error", (error) => done({ error, code: null, signal: null }));
    child.once("exit", (code, signal) => done({ error: null, code, signal }));
  });
}

function eventQueue(readable, push) {
  let buffer = "";
  const emit = (line) => push({ kind: "event-line", line });
  readable.setEncoding("utf8");
  readable.on("data", (chunk) => {
    buffer += chunk;
    if (Buffer.byteLength(buffer) > MAX_EVENT_BYTES) {
      push({ kind: "event-error", error: new Error("event line exceeds maximum size") });
      buffer = "";
      return;
    }
    for (;;) {
      const newline = buffer.indexOf("\n");
      if (newline < 0) break;
      emit(buffer.slice(0, newline));
      buffer = buffer.slice(newline + 1);
    }
  });
  readable.on("error", (error) => push({ kind: "event-error", error }));
  readable.on("end", () => {
    if (buffer.length) emit(buffer);
    push({ kind: "event-eof" });
  });
}

function makeClock(clock) {
  return {
    now: typeof clock?.now === "function" ? clock.now : monotonicMs,
    sleep: typeof clock?.sleep === "function" ? clock.sleep : sleepMs,
  };
}

function makeSignalQueue() {
  const queue = [];
  const waiters = [];
  const push = (value) => {
    const waiter = waiters.shift();
    if (waiter) waiter(value); else queue.push(value);
  };
  const next = () => queue.length ? Promise.resolve(queue.shift()) : new Promise((resolve) => waiters.push(resolve));
  return { push, next };
}

function parseEventLine(line) {
  if (!String(line).trim()) throw new Error("empty event line");
  try { return JSON.parse(line); } catch (error) { throw new Error(`malformed event JSON: ${error.message}`); }
}

export async function supervise({
  command,
  args = [],
  cwd,
  env = process.env,
  mode = "opaque",
  jobDir = cwd || process.cwd(),
  statePath = join(jobDir, "state.json"),
  rcPath = join(jobDir, "rc"),
  diagnosticsPath = join(jobDir, "diagnostics.json"),
  leaseMs = DEFAULT_LEASE_MS,
  terminalGraceMs = DEFAULT_TERMINAL_GRACE_MS,
  clock,
  spawnImpl = spawn,
  groupAlive = processGroupAlive,
  processTreeImpl = processTree,
  installSignals = false,
} = {}) {
  if (!Array.isArray(command) || command.length === 0) throw new Error("command is required");
  if (!['opaque', 'declared'].includes(mode)) throw new Error(`invalid supervisor mode: ${mode}`);
  normalizeInteger(leaseMs, "lease_ms", { min: 1 });
  normalizeInteger(terminalGraceMs, "terminal_grace_ms", { min: 1 });
  mkdirSync(jobDir, { recursive: true });
  const time = makeClock(clock);
  const signals = makeSignalQueue();
  const listeners = [];
  const childEnv = { ...env, SUPERVISOR_EVENT_FD: "3", SUPERVISOR_REVISION };
  const child = spawnImpl(command[0], [...command.slice(1), ...args], {
    cwd,
    env: childEnv,
    detached: true,
    stdio: ["ignore", "inherit", "inherit", mode === "declared" ? "pipe" : "ignore"],
  });
  const cgroup = createOwnedCgroup(child.pid);
  const childExit = childExitPromise(child);
  const pending = makeSignalQueue();
  const state = {
    revision: SUPERVISOR_REVISION,
    mode,
    phase: "starting",
    seq: 0,
    pid: child.pid,
    started_monotonic_ms: time.now(),
    last_event: null,
  };
  let lastEvent = null;
  let expectedSeq = 1;
  let leaseDeadline = state.started_monotonic_ms + leaseMs;
  let terminal = null;
  let exit = null;
  let rootExitDeadline = null;
  let settled = false;

  const updateState = (patch) => {
    Object.assign(state, patch);
    atomicWrite(statePath, `${JSON.stringify(state)}\n`);
  };
  const reportFailure = (classification, detail) => {
    const tree = processTreeImpl(child.pid);
    const diagnostic = {
      revision: SUPERVISOR_REVISION,
      classification,
      detail: String(detail || ""),
      pid: child.pid,
      last_event: lastEvent,
      process_tree: tree,
      observed_monotonic_ms: time.now(),
    };
    try { atomicWrite(diagnosticsPath, `${JSON.stringify(diagnostic)}\n`); } catch (error) { console.error(`command-supervisor diagnostics write failed: ${error.message}`); }
    console.error(`command-supervisor classification=${classification}`);
    if (diagnostic.detail) console.error(`command-supervisor detail=${diagnostic.detail}`);
    console.error(`command-supervisor last-event=${lastEvent ? JSON.stringify(lastEvent) : "none"}`);
    console.error(`command-supervisor process-tree=${tree || "empty"}`);
  };
  const finish = async (status, classification, detail, failed = status !== 0) => {
    if (settled) return { status, classification };
    settled = true;
    let cleanupError = null;
    if (failed) {
      try { reportFailure(classification, detail); } catch (error) { cleanupError = error; }
      try { await terminateChild(child, cgroup, time.sleep); } catch (error) { cleanupError ||= error; }
    }
    const phase = failed ? "failed" : "completed";
    try { updateState({ phase, classification, rc: status, finished_monotonic_ms: time.now() }); } catch (error) { console.error(`command-supervisor state write failed: ${error.message}`); }
    try { atomicWrite(rcPath, `${status}\n`); } catch (error) { console.error(`command-supervisor rc write failed: ${error.message}`); }
    if (cleanupError) console.error(`command-supervisor cleanup failed: ${cleanupError.message}`);
    removeOwnedCgroup(cgroup);
    for (const [signal, listener] of listeners) process.removeListener(signal, listener);
    return { status, classification };
  };

  if (installSignals) {
    for (const signal of ["SIGHUP", "SIGINT", "SIGQUIT", "SIGTERM"]) {
      const listener = () => pending.push({ kind: "signal", signal });
      process.on(signal, listener);
      listeners.push([signal, listener]);
    }
  }

  try {
    updateState({ phase: "running" });
    if (mode === "declared") eventQueue(child.stdio[3], (item) => pending.push(item));
    else if (child.stdio[3]) child.stdio[3].resume();
    for (;;) {
      if (exit === null) {
        const next = await Promise.race([
          childExit.then((value) => ({ kind: "child-exit", value })),
          pending.next(),
          time.sleep(mode === "declared" && terminal === null ? Math.min(POLL_MS, Math.max(1, leaseDeadline - time.now())) : POLL_MS).then(() => ({ kind: "tick" })),
        ]);
        if (next.kind === "child-exit") exit = next.value;
        else if (next.kind === "signal") return finish(128 + ({ SIGHUP: 1, SIGINT: 2, SIGQUIT: 3, SIGTERM: 15 }[next.signal] || 15), `signal-${next.signal}`, "supervisor received signal");
        else if (next.kind === "event-line") {
          if (terminal) return finish(254, "event-after-terminal", "event received after terminal event");
          let event;
          try { event = parseEventLine(next.line); } catch (error) { return finish(254, "malformed-event", error.message); }
          try { validateDeclaredEvent(event, expectedSeq); } catch (error) { return finish(254, "invalid-event", error.message); }
          expectedSeq += 1;
          lastEvent = event;
          leaseDeadline = time.now() + event.lease_ms;
          if (event.type === "terminal") {
            terminal = { rc: event.rc, deadline: time.now() + terminalGraceMs };
            updateState({ phase: "terminal", seq: event.seq, last_event: event, terminal_rc: event.rc, terminal_deadline_monotonic_ms: terminal.deadline });
          } else {
            updateState({ phase: "running", seq: event.seq, last_event: event, lease_deadline_monotonic_ms: leaseDeadline });
          }
        } else if (next.kind === "event-error") return finish(254, "event-stream-error", next.error.message);
        else if (next.kind === "event-eof" && !terminal) updateState({ event_stream: "eof" });
        if (mode === "declared" && terminal && exit === null && time.now() >= terminal.deadline && groupAlive(child.pid)) {
          return finish(254, "terminal-with-live-descendants", "terminal command left live descendants");
        }
        if (mode === "declared" && terminal === null && time.now() >= leaseDeadline) return finish(254, "lease-expired", "declared command did not renew its lease");
      }
      if (exit !== null) {
        const alive = groupAlive(child.pid);
        if (mode === "opaque") {
          if (alive) {
            if (rootExitDeadline === null) rootExitDeadline = time.now() + terminalGraceMs;
            if (time.now() < rootExitDeadline) { await time.sleep(POLL_MS); continue; }
            return finish(254, "root-exit-with-descendants", `root exited rc=${exit.code ?? 128}`);
          }
          const status = exit.error ? spawnErrorStatus(exit.error) : (exit.signal ? 128 + (exit.signal === "SIGTERM" ? 15 : exit.signal === "SIGINT" ? 2 : 1) : (exit.code ?? 254));
          return finish(status, exit.error ? "spawn-error" : "opaque-exit", exit.error?.message || "", status !== 0);
        }
        // A command that never started has no terminal event by construction; classify
        // the spawn failure before the protocol rules mistake it for a broken contract.
        if (exit.error && !alive) return finish(spawnErrorStatus(exit.error), "spawn-error", exit.error.message);
        if (!terminal) return finish(254, alive ? "root-exit-with-descendants" : "root-exit-without-terminal", "declared command exited without a terminal event");
        if (!alive) {
          const status = exit.signal ? 128 + (exit.signal === "SIGTERM" ? 15 : exit.signal === "SIGINT" ? 2 : 1) : (exit.code ?? 254);
          if (status !== terminal.rc) return finish(254, "terminal-rc-mismatch", `terminal rc=${terminal.rc}, process rc=${status}`);
          return finish(status, "terminal", "", status !== 0);
        }
        if (time.now() >= terminal.deadline) return finish(254, "terminal-with-live-descendants", "terminal command left live descendants");
      }
    }
  } catch (error) {
    return finish(254, "supervisor-error", error.message);
  }
}

function parseArgs(argv) {
  const out = { mode: "opaque", jobDir: process.cwd(), statePath: null, rcPath: null, diagnosticsPath: null, leaseMs: DEFAULT_LEASE_MS, terminalGraceMs: DEFAULT_TERMINAL_GRACE_MS, command: [] };
  for (let index = 0; index < argv.length; index += 1) {
    const arg = argv[index];
    if (arg === "--") { out.command = argv.slice(index + 1); break; }
    if (arg === "--mode") out.mode = argv[++index];
    else if (arg === "--job-dir") out.jobDir = argv[++index];
    else if (arg === "--state") out.statePath = argv[++index];
    else if (arg === "--rc") out.rcPath = argv[++index];
    else if (arg === "--diagnostics") out.diagnosticsPath = argv[++index];
    else if (arg === "--lease-ms") out.leaseMs = Number(argv[++index]);
    else if (arg === "--terminal-grace-ms") out.terminalGraceMs = Number(argv[++index]);
    else throw new Error(`unknown arg: ${arg}`);
  }
  if (!out.command.length) throw new Error("missing command after --");
  out.statePath ||= join(out.jobDir, "state.json");
  out.rcPath ||= join(out.jobDir, "rc");
  out.diagnosticsPath ||= join(out.jobDir, "diagnostics.json");
  return out;
}

const isMain = process.argv[1] && process.argv[1].endsWith("command-supervisor.mjs");
if (isMain) {
  if (process.argv[2] === "--version") {
    process.stdout.write(`${SUPERVISOR_REVISION}\n`);
  } else {
    try {
      const options = parseArgs(process.argv.slice(2));
      const result = await supervise({ ...options, installSignals: true });
      process.exitCode = result.status;
    } catch (error) {
      process.stderr.write(`command-supervisor: ${error.message}\n`);
      process.exitCode = 254;
    }
  }
}
