master-zapier-plan-draft/packages/verae-keep/src/keep.js

#!/usr/bin/env node
/**
 * Host keep-alive. Restarts units that should be running.
 * Honors admin pause/stop (intent files + optional fleet /api/status).
 */
import fs from 'node:fs';
import http from 'node:http';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { expandHome } from './expand.js';
import {
  loadStore,
  saveStore,
  record,
  pickIntent,
  writeUnitFile,
  fromFleetService,
} from './intent.js';
import { httpProbe, httpPost, tcpProbe } from './probe.js';
import { spawnUnit, killPid } from './spawn.js';
import { loadUnitsFile, scanRuntimeIntents } from './units.js';

const HERE = path.dirname(fileURLToPath(import.meta.url));
const ROOT = path.join(HERE, '..');
const UNITS_FILE = expandHome(process.env.KEEP_UNITS || path.join(ROOT, 'units.json'));
const STATE_DIR = expandHome(process.env.KEEP_STATE || path.join(ROOT, 'data'));
const STORE_FILE = path.join(STATE_DIR, 'intent.json');
const LOG_DIR = path.join(STATE_DIR, 'logs');
const FLEET_URL = (process.env.FLEET_URL || '').replace(/\/$/, '');

const cfg = loadUnitsFile(UNITS_FILE);
const store = loadStore(STORE_FILE);
const last = {};
const restarts = {};

function log(event, extra = {}) {
  const line = JSON.stringify({ t: new Date().toISOString(), event, ...extra });
  fs.mkdirSync(STATE_DIR, { recursive: true });
  fs.appendFileSync(path.join(STATE_DIR, 'keep.log'), `${line}\n`);
  process.stdout.write(`${line}\n`);
}

function unitById(id) {
  return cfg.units.find((u) => u.id === id);
}

function intentOf(unit, runtimeIntents, fleetMap) {
  const fromStore = store.units[unit.id];
  const fromRuntime = runtimeIntents[unit.id];
  const fromFleet = fleetMap[unit.id];
  const picked = pickIntent(fromStore, fromRuntime, fromFleet);
  if (picked) return picked;
  return { state: unit.defaultIntent || 'run', source: 'default', updatedAt: '' };
}

async function probe(unit) {
  if (unit.healthUrl) return httpProbe(unit.healthUrl);
  if (unit.tcp) return tcpProbe(unit.tcp.host || '127.0.0.1', Number(unit.tcp.port));
  return { ok: false, paused: false, error: 'no probe' };
}

function healthBase(unit) {
  if (!unit.healthUrl) return null;
  try {
    const u = new URL(unit.healthUrl);
    return `${u.protocol}//${u.host}`;
  } catch {
    return null;
  }
}

function bumpRestart(id) {
  const hour = Math.floor(Date.now() / 3600000);
  const slot = restarts[id] || { hour, n: 0 };
  if (slot.hour !== hour) {
    slot.hour = hour;
    slot.n = 0;
  }
  slot.n += 1;
  restarts[id] = slot;
  return slot.n <= 40;
}

async function apply(unit, intent) {
  const p = await probe(unit);
  last[unit.id] = { intent, probe: { ok: p.ok, paused: p.paused, error: p.error || null, pid: p.pid || null } };
  const base = healthBase(unit);
  if (!unit.managed) {
    last[unit.id].action = p.ok ? 'observe' : 'down-unmanaged';
    return last[unit.id];
  }
  if (intent.state === 'stop') {
    if (p.ok || p.paused || p.pid) {
      killPid(p.pid || last[unit.id]?.spawnedPid);
      last[unit.id].action = 'killed';
      log('stop', { id: unit.id });
    } else last[unit.id].action = 'already-stopped';
    return last[unit.id];
  }
  if (intent.state === 'pause') {
    if (p.ok && !p.paused && base) {
      await httpPost(`${base}${unit.pausePath}`);
      last[unit.id].action = 'paused';
      log('pause', { id: unit.id });
    } else if (!p.ok && !p.paused) {
      last[unit.id].action = 'leave-down';
    } else last[unit.id].action = 'already-paused';
    return last[unit.id];
  }
  // run
  if (p.ok) {
    last[unit.id].action = 'healthy';
    return last[unit.id];
  }
  if (p.paused && base) {
    await httpPost(`${base}${unit.resumePath}`);
    last[unit.id].action = 'resumed';
    log('resume', { id: unit.id });
    return last[unit.id];
  }
  if (!unit.command) {
    last[unit.id].action = 'down-no-command';
    return last[unit.id];
  }
  if (!bumpRestart(unit.id)) {
    last[unit.id].action = 'restart-capped';
    log('restart-capped', { id: unit.id });
    return last[unit.id];
  }
  const spawned = spawnUnit(unit, { logDir: LOG_DIR });
  last[unit.id].spawnedPid = spawned.pid;
  last[unit.id].action = 'spawned';
  log('spawn', { id: unit.id, pid: spawned.pid });
  return last[unit.id];
}

async function fleetMap() {
  const map = {};
  if (!FLEET_URL) return map;
  try {
    const r = await httpProbe(`${FLEET_URL}/api/status`, { timeoutMs: 2500 });
    const services = r.body?.services || {};
    for (const unit of cfg.units) {
      const sid = unit.env?.FLEET_SERVICE || unit.id.replace(/-\d+$/, '');
      const sv = services[sid];
      const via = fromFleetService(sv, unit.id);
      if (via) map[unit.id] = via;
    }
  } catch {
    /* fleet optional */
  }
  return map;
}

async function tick() {
  const runtimeIntents = cfg.runtimeDir ? scanRuntimeIntents(cfg.runtimeDir) : {};
  const fromFleet = await fleetMap();
  const actions = [];
  for (const unit of cfg.units) {
    const intent = intentOf(unit, runtimeIntents, fromFleet);
    const row = await apply(unit, intent);
    actions.push({ id: unit.id, intent: intent.state, source: intent.source, action: row.action });
  }
  return { t: new Date().toISOString(), actions };
}

function setIntent(id, state, source = 'admin') {
  const unit = unitById(id);
  if (!unit) throw Object.assign(new Error(`unknown unit ${id}`), { status: 404 });
  const row = record(store, id, state, source);
  saveStore(STORE_FILE, store);
  if (cfg.runtimeDir) {
    writeUnitFile(path.join(cfg.runtimeDir, 'data', id, 'intent.json'), id, state, source);
  }
  log('intent', { id, state, source });
  return row;
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url || '/', `http://127.0.0.1:${cfg.port}`);
  const json = (code, obj) => {
    res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' });
    res.end(JSON.stringify(obj, null, 2));
  };
  try {
    if (req.method === 'GET' && url.pathname === '/health') {
      return json(200, { ok: true, role: 'verae-keep', units: cfg.units.length, pid: process.pid });
    }
    if (req.method === 'GET' && url.pathname === '/status') {
      return json(200, {
        ok: true,
        units: cfg.units.map((u) => ({
          id: u.id,
          title: u.title,
          intent: store.units[u.id] || { state: u.defaultIntent, source: 'default' },
          last: last[u.id] || null,
        })),
      });
    }
    const m = url.pathname.match(/^\/intent\/([^/]+)$/);
    if (req.method === 'POST' && m) {
      const chunks = [];
      for await (const c of req) chunks.push(c);
      const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
      const row = setIntent(decodeURIComponent(m[1]), body.state, body.source || 'admin');
      return json(200, row);
    }
    if (req.method === 'POST' && url.pathname === '/reconcile') {
      return json(200, await tick());
    }
    json(404, { error: 'not found' });
  } catch (err) {
    json(err.status || 400, { error: err.message });
  }
});

const bind = process.env.KEEP_BIND || cfg.bind || '127.0.0.1';
const port = Number(process.env.KEEP_PORT || cfg.port || 3860);

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  server.listen(port, bind, () => {
    log('listen', { url: `http://${bind}:${port}/`, units: cfg.units.map((u) => u.id) });
  });
  const iv = setInterval(() => {
    tick().catch((err) => log('tick-error', { error: err.message }));
  }, cfg.intervalMs);
  tick().catch((err) => log('tick-error', { error: err.message }));
  const stop = () => {
    clearInterval(iv);
    server.close();
    process.exit(0);
  };
  process.on('SIGTERM', stop);
  process.on('SIGINT', stop);
}

export { tick, setIntent, cfg, store };