/**
 * Spawn, pause, resume, stop, restart replicas. Keep replica floor (tree-node min).
 * @module supervisor
 */

import { spawn } from 'node:child_process';
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { loadFleet } from './load.js';
import { httpProbe, postControl } from './health.js';
import { classifyService } from './classify.js';
import { pickMachine, upsertMachine, saveOverlay } from './machines.js';
import { summarizeRtt } from './rtt.js';
import { sshSpawnWorker, sshKillWorker, sshHttp, sshCheck } from './ssh.js';

const WORKER = path.join(path.dirname(fileURLToPath(import.meta.url)), 'worker.js');

/**
 * Real HTTP apps (zappier-edge, middleware) set `spawn` in services/*.json.
 * Everything else is the fleet health worker.
 */
export function spawnLocal(root, spec, env, healthPort) {
  const merged = { ...env, ...(spec.env || {}), PORT: String(healthPort) };
  if (spec.spawn?.command) {
    const cwd = path.resolve(root, spec.spawn.cwd || '.');
    return spawn(spec.spawn.command, spec.spawn.args || [], {
      env: merged,
      cwd,
      stdio: ['ignore', 'pipe', 'pipe'],
    });
  }
  return spawn(process.execPath, [WORKER], { env: merged, stdio: ['ignore', 'pipe', 'pipe'] });
}

export class Supervisor {
  /**
   * @param {object} [opts]
   * @param {ReturnType<typeof loadFleet>} [opts.loaded]
   * @param {string} [opts.stateRoot]
   */
  constructor(opts = {}) {
    this.loaded = opts.loaded || loadFleet();
    this.stateRoot = opts.stateRoot || path.join(this.loaded.root, 'data', 'runtime');
    fs.mkdirSync(this.stateRoot, { recursive: true });
    /** @type {Map<string, object>} */
    this.instances = new Map();
    this.events = [];
    this._seq = 0;
    if (!Array.isArray(this.loaded.machines) || !this.loaded.machines.length) {
      this.loaded.machines = [
        { id: 'local', title: 'local', kind: 'local', host: '127.0.0.1', enabled: true, capacity: 32, roles: ['*'] },
      ];
    }
  }

  machines() {
    return this.loaded.machines;
  }

  addMachine(spec) {
    const m = upsertMachine(this.loaded.machines, spec);
    if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines);
    this.log('machine-add', { id: m.id, host: m.host, kind: m.kind });
    return m;
  }

  setMachineEnabled(id, enabled) {
    const m = this.loaded.machines.find((x) => x.id === id);
    if (!m) throw new Error(`unknown machine ${id}`);
    m.enabled = Boolean(enabled);
    if (this.loaded.machinesOverlay) saveOverlay(this.loaded.machinesOverlay, this.loaded.machines);
    return m;
  }

  async checkMachine(id) {
    const m = this.machines().find((x) => x.id === id);
    if (!m) throw new Error(`unknown machine ${id}`);
    if (m.kind === 'ssh') return sshCheck(m);
    if (m.kind === 'agent') {
      const r = await httpProbe({ host: m.host, port: m.agentPort, path: '/health', timeoutMs: 2000 });
      return { ok: r.ok, kind: 'agent', body: r.body };
    }
    return { ok: true, kind: 'local', host: m.host };
  }

  machineOf(rec) {
    return this.machines().find((x) => x.id === rec.machine);
  }

  async controlInstance(rec, pathname, body) {
    if (rec.kind === 'ssh') {
      const m = this.machineOf(rec);
      if (!m) throw new Error(`machine ${rec.machine} missing`);
      return sshHttp(m, 'POST', pathname, { port: rec.healthPort, body });
    }
    return postControl({
      host: rec.host || '127.0.0.1',
      port: rec.healthPort,
      path: pathname,
      body,
    });
  }

  log(type, detail) {
    const ev = { t: new Date().toISOString(), type, ...detail };
    this.events.push(ev);
    if (this.events.length > 500) this.events.splice(0, this.events.length - 400);
    return ev;
  }

  spec(serviceId) {
    const s = this.loaded.services[serviceId];
    if (!s) throw new Error(`unknown service ${serviceId}`);
    return s;
  }

  nextIndex(serviceId) {
    const used = new Set(
      [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.index),
    );
    const max = this.spec(serviceId).max ?? 8;
    for (let i = 0; i < max; i += 1) if (!used.has(i)) return i;
    return null;
  }

  /**
   * @param {string} serviceId
   * @param {{ index?: number }} [opts]
   */
  async startOne(serviceId, opts = {}) {
    const spec = this.spec(serviceId);
    if (!spec.managed) {
      this.log('skip', { service: serviceId, reason: 'unmanaged' });
      return null;
    }
    if (!spec.enabled) {
      this.log('skip', { service: serviceId, reason: 'disabled' });
      return null;
    }
    const index = opts.index ?? this.nextIndex(serviceId);
    if (index == null) {
      this.log('skip', { service: serviceId, reason: 'at-max' });
      return null;
    }
    const instance = `${serviceId}-${index}`;
    if (this.instances.has(instance)) return this.instances.get(instance);
    const machine =
      opts.machine ||
      pickMachine(
        this.machines(),
        [...this.instances.values()],
        serviceId,
        spec.role || serviceId,
        opts.excludeMachines || [],
      );
    if (!machine) {
      this.log('skip', { service: serviceId, reason: 'no-machine-capacity' });
      return null;
    }
    const healthPort = (spec.ports?.healthBase || 19000) + index;
    const stateDir = path.join(this.stateRoot, instance);
    fs.mkdirSync(stateDir, { recursive: true });
    const env = {
      ...process.env,
      FLEET_ROLE: spec.role || serviceId,
      FLEET_SERVICE: serviceId,
      FLEET_INSTANCE: instance,
      FLEET_MACHINE: machine.id,
      FLEET_HEALTH_PORT: String(healthPort),
      FLEET_HEALTH_BIND: machine.kind === 'local' ? '127.0.0.1' : '127.0.0.1',
      FLEET_STATE_DIR: machine.kind === 'ssh' ? `${machine.remoteDir || '~/verae-fleet-runtime'}/data/${instance}` : stateDir,
      FLEET_MESSAGE_HZ: process.env.FLEET_MESSAGE_HZ || '8',
    };
    let child = null;
    let pid = null;
    if (machine.kind === 'ssh') {
      try {
        const spawned = await sshSpawnWorker(machine, { instance, env });
        pid = spawned.pid;
      } catch (err) {
        this.log('skip', { service: serviceId, reason: 'ssh-spawn-failed', machine: machine.id, error: err.message });
        const exclude = [...(opts.excludeMachines || []), machine.id];
        if (exclude.length < this.machines().length) {
          return this.startOne(serviceId, { ...opts, excludeMachines: exclude });
        }
        return null;
      }
    } else if (machine.kind === 'agent') {
      try {
        const spawned = await postControl({
          host: machine.host,
          port: machine.agentPort,
          path: '/spawn',
          body: { env, instance },
          timeoutMs: 4000,
        });
        pid = spawned.pid || null;
      } catch (err) {
        this.log('skip', { service: serviceId, reason: 'agent-spawn-failed', machine: machine.id, error: err.message });
        return null;
      }
    } else {
      child = spawnLocal(this.loaded.root, spec, env, healthPort);
      pid = child.pid;
    }
    const rec = {
      id: instance,
      service: serviceId,
      index,
      machine: machine.id,
      host: machine.host,
      pid,
      healthPort,
      stateDir,
      child,
      kind: machine.kind,
      _machine: machine,
      paused: false,
      startedAt: Date.now(),
      restarts: 0,
      lastError: null,
      rttSamples: [],
    };
    if (child) {
      child.stderr?.on('data', () => {});
      child.on('exit', (code, signal) => {
        rec.exitCode = code;
        rec.signal = signal;
        rec.pid = null;
        this.log('exit', { instance, service: serviceId, code, signal });
      });
    }
    this.instances.set(instance, rec);
    this.log('start', { instance, service: serviceId, machine: machine.id, pid: rec.pid, healthPort });
    const h = await waitForHealth(rec, 8000);
    rec.healthy = h.ok;
    rec.lastProbe = h;
    return rec;
  }

  async stopInstance(instanceId, { replace = true } = {}) {
    const rec = this.instances.get(instanceId);
    if (!rec) return null;
    rec.stopping = true;
    if (rec.kind === 'ssh') {
      const m = this.machines().find((x) => x.id === rec.machine);
      if (m) {
        try {
          await sshKillWorker(m, rec.pid, instanceId);
        } catch {
          /* ignore */
        }
      }
    } else if (rec.kind === 'agent') {
      const m = this.machines().find((x) => x.id === rec.machine);
      if (m) {
        try {
          await postControl({ host: m.host, port: m.agentPort, path: '/kill', body: { instance: instanceId } });
        } catch {
          /* ignore */
        }
      }
    } else if (rec.child && rec.pid) {
      rec.child.kill('SIGTERM');
      await waitExit(rec.child, 2000);
    }
    this.instances.delete(instanceId);
    this.log('stop', { instance: instanceId, service: rec.service });
    if (replace) await this.reconcile(rec.service);
    return rec;
  }

  async pauseInstance(instanceId, { replace = true } = {}) {
    const rec = this.instances.get(instanceId);
    if (!rec) throw new Error(`no instance ${instanceId}`);
    await this.controlInstance(rec, '/pause');
    rec.paused = true;
    this.log('pause', { instance: instanceId, service: rec.service });
    if (replace) await this.reconcile(rec.service);
    return rec;
  }

  async resumeInstance(instanceId) {
    const rec = this.instances.get(instanceId);
    if (!rec) throw new Error(`no instance ${instanceId}`);
    await this.controlInstance(rec, '/resume');
    rec.paused = false;
    this.log('resume', { instance: instanceId, service: rec.service });
    return rec;
  }

  async restartInstance(instanceId) {
    const rec = this.instances.get(instanceId);
    if (!rec) throw new Error(`no instance ${instanceId}`);
    const { service, index, machine } = rec;
    await this.stopInstance(instanceId, { replace: false });
    const next = await this.startOne(service, { index, machine: this.machines().find((m) => m.id === machine) });
    if (next) next.restarts = (rec.restarts || 0) + 1;
    this.log('restart', { instance: instanceId, service });
    return next;
  }

  async markUnhealthy(instanceId) {
    const rec = this.instances.get(instanceId);
    if (!rec) throw new Error(`no instance ${instanceId}`);
    await this.controlInstance(rec, '/unhealthy');
    return rec;
  }

  async probe(rec) {
    const spec = this.spec(rec.service);
    const pathName = spec.health?.path || '/health';
    let r;
    if (rec.kind === 'ssh') {
      const m = this.machineOf(rec);
      try {
        const body = await sshHttp(m, 'GET', pathName, { port: rec.healthPort });
        r = { ok: body.ok !== false, paused: Boolean(body.paused), body, statusCode: body.ok === false ? 503 : 200 };
      } catch (err) {
        r = { ok: false, paused: false, error: err.message };
      }
    } else {
      r = await httpProbe({
        host: rec.host || '127.0.0.1',
        port: rec.healthPort,
        path: pathName,
        timeoutMs: spec.health?.timeoutMs || 800,
      });
    }
    rec.paused = Boolean(r.paused);
    rec.lastProbe = r;
    rec.healthy = r.ok;
    if (r.ok && !rec.paused) {
      try {
        const msg = await this.controlInstance(rec, '/message');
        if (Number.isFinite(msg.rttMs)) {
          rec.rttSamples = rec.rttSamples || [];
          rec.rttSamples.push(msg.rttMs);
          if (rec.rttSamples.length > 200) rec.rttSamples.splice(0, rec.rttSamples.length - 200);
        }
        rec.rtt = msg.rtt || r.body?.rtt || summarizeRtt(rec.rttSamples);
      } catch {
        rec.rtt = r.body?.rtt || summarizeRtt(rec.rttSamples || []);
      }
    }
    return r;
  }

  available(serviceId) {
    return [...this.instances.values()].filter(
      (i) => i.service === serviceId && i.pid && !i.paused && i.healthy !== false,
    );
  }

  /**
   * Ensure running healthy unpaused copies >= min when keepFloor.
   * @param {string} [only]
   */
  async reconcile(only) {
    const actions = [];
    const ids = only ? [only] : Object.keys(this.loaded.services);
    for (const serviceId of ids) {
      const spec = this.loaded.services[serviceId];
      if (!spec?.managed || !spec.enabled) continue;
      for (const rec of [...this.instances.values()].filter((i) => i.service === serviceId)) {
        if (!rec.pid && !rec.stopping) {
          this.instances.delete(rec.id);
          continue;
        }
        if (rec.pid) await this.probe(rec);
        if (rec.pid && rec.healthy === false && !rec.paused) {
          this.log('unhealthy', { instance: rec.id, service: serviceId });
          await this.restartInstance(rec.id);
          actions.push({ op: 'restart', instance: rec.id });
        }
      }
      const avail = this.available(serviceId).length;
      const want = spec.keepFloor ? spec.min : Math.min(spec.min, spec.max);
      if (avail < want) {
        const need = want - avail;
        for (let n = 0; n < need; n += 1) {
          const started = await this.startOne(serviceId);
          if (started) actions.push({ op: 'start', instance: started.id });
        }
      }
    }
    return actions;
  }

  async startService(serviceId) {
    const spec = this.spec(serviceId);
    spec.enabled = true;
    const have = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).length;
    const want = spec.min || 1;
    const started = [];
    for (let i = have; i < want; i += 1) {
      const rec = await this.startOne(serviceId);
      if (rec) started.push(rec.id);
    }
    return started;
  }

  async stopService(serviceId, { holdFloor = false } = {}) {
    const spec = this.spec(serviceId);
    if (!holdFloor) spec.enabled = false;
    const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id);
    for (const id of ids) await this.stopInstance(id, { replace: false });
    if (holdFloor) await this.reconcile(serviceId);
    return ids;
  }

  async pauseService(serviceId) {
    const ids = [...this.instances.values()].filter((i) => i.service === serviceId && i.pid).map((i) => i.id);
    for (const id of ids) await this.pauseInstance(id, { replace: false });
    await this.reconcile(serviceId);
    return ids;
  }

  async resumeService(serviceId) {
    const ids = [...this.instances.values()].filter((i) => i.service === serviceId).map((i) => i.id);
    for (const id of ids) await this.resumeInstance(id);
    return ids;
  }

  async stopAll() {
    for (const id of [...this.instances.keys()]) {
      await this.stopInstance(id, { replace: false });
    }
  }

  status() {
    const services = {};
    for (const spec of Object.values(this.loaded.services)) {
      const inst = [...this.instances.values()].filter((i) => i.service === spec.id);
      const available = inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length;
      services[spec.id] = {
        id: spec.id,
        title: spec.title,
        configPath: spec.configPath,
        managed: spec.managed,
        enabled: spec.enabled,
        min: spec.min,
        max: spec.max,
        keepFloor: spec.keepFloor,
        running: inst.filter((i) => i.pid).length,
        paused: inst.filter((i) => i.paused).length,
        available,
        belowFloor: spec.keepFloor && spec.enabled && spec.managed && available < spec.min,
        instances: inst.map((i) => ({
          id: i.id,
          pid: i.pid,
          machine: i.machine || 'local',
          host: i.host || '127.0.0.1',
          healthPort: i.healthPort,
          paused: i.paused,
          healthy: i.healthy,
          restarts: i.restarts,
          rtt: i.rtt || summarizeRtt(i.rttSamples || []),
        })),
      };
      const allSamples = inst.flatMap((i) => i.rttSamples || []);
      services[spec.id].rtt = summarizeRtt(allSamples);
      services[spec.id].health = classifyService(services[spec.id]);
    }
    const machines = this.machines().map((m) => {
      const inst = [...this.instances.values()].filter((i) => i.machine === m.id);
      const samples = inst.flatMap((i) => i.rttSamples || []);
      return {
        ...m,
        running: inst.filter((i) => i.pid).length,
        available: inst.filter((i) => i.pid && !i.paused && i.healthy !== false).length,
        rtt: summarizeRtt(samples),
      };
    });
    return {
      control: this.loaded.control,
      nats: this.loaded.nats,
      machines,
      services,
      events: this.events.slice(-50),
    };
  }
}

function waitExit(child, ms) {
  return new Promise((resolve) => {
    if (!child || child.exitCode != null) return resolve();
    const t = setTimeout(() => {
      try {
        child.kill('SIGKILL');
      } catch {
        /* ignore */
      }
      resolve();
    }, ms);
    child.once('exit', () => {
      clearTimeout(t);
      resolve();
    });
  });
}

async function waitForHealth(rec, ms) {
  const deadline = Date.now() + ms;
  while (Date.now() < deadline) {
    let r;
    if (rec.kind === 'ssh' && rec._machine) {
      try {
        const body = await sshHttp(rec._machine, 'GET', '/health', { port: rec.healthPort });
        r = { ok: body.ok !== false, statusCode: 200, paused: Boolean(body.paused), body };
      } catch {
        r = { ok: false };
      }
    } else {
      r = await httpProbe({
        host: rec.host || '127.0.0.1',
        port: rec.healthPort,
        path: '/health',
        timeoutMs: 400,
      });
    }
    if (r.ok || r.statusCode === 503) return r;
    await new Promise((res) => setTimeout(res, 80));
  }
  return { ok: false, error: 'start-timeout' };
}