stack/packages/cli/src/org-bus.mjs

// Organization-scoped bus lifecycle. Registry resolution precedes every action.
import { randomBytes } from 'node:crypto';
import { processState } from '../../seat/src/process-state.mjs';
import { spawn } from 'node:child_process';
import { constants, closeSync, lstatSync, mkdirSync, openSync, readFileSync, fstatSync, unlinkSync, writeFileSync } from 'node:fs';
import { connect } from 'node:net';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { resolveOrg, withOrgLifecycle } from '../../organization/src/org-registry.mjs';
import { orgPaths } from './config.mjs';
import { cmdlineOf, hostFile, hostLock, hostStatus, readHostState, startTimeOf, stopHost } from './host.mjs';
import { CliError } from './errors.mjs';
import { readBootStatus } from './boot-status.mjs';

const CLI = fileURLToPath(new URL('./cli.mjs', import.meta.url));
const fail = message => { throw new CliError(message, 3); };
async function target(ctx, id, allowDeleting = false) {
  let org;
  try { org = await resolveOrg(ctx, id); }
  catch (error) {
    if (['ENOENT', 'EACCES', 'ENOTDIR'].includes(error.code)) fail('organization registry is inaccessible');
    throw error;
  }
  if (org.state === 'archived' && !allowDeleting) fail(`org-archived: organization ${id} is archived`);
  if (org.state === 'deleting' && !allowDeleting) fail(`organization ${id} is being deleted`);
  if (org.state !== 'active' && !(allowDeleting && ['deleting', 'archived'].includes(org.state))) fail(`organization ${id} is not active`);
  return { ...org, paths: orgPaths(org.root) };
}
function checkedState(org) {
  const s = hostStatus(org.root);
  let lock = null;
  try { lock = JSON.parse(readFileSync(hostLock(org.root), 'utf8')); }
  catch (e) { if (e.code !== 'ENOENT') fail('bus host lock is unreadable'); }
  if (lock && (!Number.isSafeInteger(lock.pid) || typeof lock.startTime !== 'string' || startTimeOf(lock.pid) !== lock.startTime)) fail('stale bus host lock');
  if (s.host && (!s.host.live || s.host.business !== org.id)) fail('stale or mismatched bus host state');
  if (s.host) {
    const args = cmdlineOf(s.host.pid) ?? [], i = args.indexOf(CLI);
    if (i < 0 || args[i + 1] !== 'bus' || args[i + 2] !== 'start' || args[i + 3] !== org.id) fail('bus host identity mismatch');
  }
  if (lock && (!s.host || lock.pid !== s.host.pid || lock.business !== org.id)) fail('bus host lock does not match host state');
  if (s.writerLock && !s.writerLock.live) fail('stale bus writer lock');
  if (!s.host && s.writerLock) fail('bus writer lock has no matching host');
  if (s.host && !lock) fail('bus host lock is missing');
  return s;
}
export function probeBusSocket(path, { signal, deadline } = {}) {
  return new Promise(resolve => {
    const remaining = deadline === undefined ? 1000 : Math.min(1000, deadline - Date.now());
    if (signal?.aborted || remaining <= 0) return resolve(false);
    let done = false, data = '';
    const socket = connect(path);
    const abort = () => finish(false);
    const finish = value => { if (done) return; done = true; signal?.removeEventListener('abort', abort); socket.destroy(); resolve(value); };
    signal?.addEventListener('abort', abort, { once: true });
    if (signal?.aborted) return abort();
    socket.setTimeout(remaining, () => finish(false));
    socket.on('error', () => finish(false));
    socket.on('connect', () => socket.write('{}\n'));
    socket.on('data', chunk => {
      data += chunk;
      if (data.length > 4096) return finish(false);
      if (data.includes('\n')) {
        try { const r = JSON.parse(data.split('\n')[0]); finish(r.ok === false && r.error === 'invalid-envelope'); }
        catch { finish(false); }
      }
    });
    socket.on('end', () => finish(false));
  });
}
export async function busStatus(ctx, id, bound) {
  const org = await target(ctx, id), state = checkedState(org);
  const healthy = Boolean(state.host?.live && state.socket && await probeBusSocket(org.paths.socket, bound));
  return { org: id, instance: ctx.instance, root: org.root, socket: org.paths.socket, healthy, pid: state.host?.pid ?? null, boot: readBootStatus(org.root) };
}
export async function busStop(ctx, id) {
  const org = await target(ctx, id, true);
  checkedState(org);
  return stopHost(org.root, { expectedBusiness: id });
}
// Deletion refuses stale or ambiguous host state and any listening socket.
export async function busStopForDelete(ctx, id) {
  const org = await target(ctx, id, true);
  for (const [part, directory] of [['bus-host', true], ['bus-host/host.json', false], ['bus-host/host.lock', false], ['bus', true], ['bus/writer.lock', false], ['bus/broker.sock', false]]) {
    try {
      const st = lstatSync(join(org.root, part));
      if (st.isSymbolicLink()) fail('symlink in bus state; inspect it before resuming deletion');
      if (directory ? !st.isDirectory() : part === 'bus/broker.sock' ? !st.isSocket() : !st.isFile()) fail('unsafe bus state type');
    } catch (e) { if (e.code !== 'ENOENT') throw e; }
  }
  checkedState(org);
  const stopped = await stopHost(org.root, { expectedBusiness: id });
  const state = checkedState(org);
  if (state.host || state.writerLock) fail('bus stop is uncertain; inspect the host before resuming deletion');
  const listening = await new Promise(resolve => {
    const socket = connect(org.paths.socket); let done = false;
    const finish = value => { if (done) return; done = true; socket.destroy(); resolve(value); };
    socket.setTimeout(1000, () => finish(true));
    socket.on('connect', () => finish(true)); socket.on('error', e => finish(!['ENOENT', 'ECONNREFUSED'].includes(e.code)));
  });
  if (listening) fail('bus socket still answers; inspect the host before resuming deletion');
  return stopped;
}
export async function busStart(ctx, id, { detach = false, devLocalOwner = false, pm = false, noLaunch = false } = {}) {
  if (pm && noLaunch) fail('bus start takes --pm or --no-launch, not both');
  // Protect parent-side directory preparation; the child takes the same fence
  // for admission and boot, so the parent must release it before spawning.
  const { org, fd } = await withOrgLifecycle(ctx, id, async () => {
    const org = await target(ctx, id);
    checkedState(org);
    if (readHostState(org.root)?.live) fail(`bus already runs for ${id}`);
    mkdirSync(org.paths.hostDir, { recursive: true, mode: 0o700 });
    const st = lstatSync(org.paths.hostDir);
    if (!st.isDirectory() || st.isSymbolicLink() || st.uid !== process.getuid() || (st.mode & 0o777) !== 0o700) fail('unsafe bus host directory');
    const fd = openSync(join(org.paths.hostDir, 'host.log'), constants.O_WRONLY | constants.O_APPEND | constants.O_CREAT | constants.O_NOFOLLOW, 0o600);
    return { org, fd };
  });
  let child;
  try {
    child = spawn(process.execPath, [CLI, 'bus', 'start', id, '--internal-host', ...(devLocalOwner ? ['--dev-local-owner'] : []), ...(pm ? ['--pm'] : []), ...(noLaunch ? ['--no-launch'] : [])], {
      env: ctx.env ?? process.env, detached: detach, stdio: ['ignore', fd, fd, 'ipc'],
    });
  } finally { closeSync(fd); }
  const done = new Promise(resolve => { child.once('error', () => resolve(1)); child.once('exit', code => resolve(code ?? 1)); });
  let exited = false, ready = false;
  child.on('message', message => { if (message?.orgReady === id) ready = true; });
  done.then(() => { exited = true; });
  const deadline = Date.now() + 30000;
  while (Date.now() < deadline && !exited) {
    // During boot the host holds its lock before its state is published.
    const state = readHostState(org.root);
    if (ready && state?.live && state.pid === child.pid && state.business === id && await probeBusSocket(org.paths.socket)) {
      child.disconnect();
      if (detach) child.unref();
      const receipt = { org: id, pid: child.pid, socket: org.paths.socket, root: org.root };
      Object.defineProperty(receipt, 'done', { value: done });
      return receipt;
    }
    await new Promise(r => setTimeout(r, 50));
  }
  if (!exited) { child.kill('SIGTERM'); await done; }
  // Name the log by its place in the organization's runtime directory, never by host path (PR #1916);
  // resolveOrg admitted `id` only after the registry's id check.
  throw new CliError(`bus failed to start; see bus-host/host.log in the runtime directory of organization ${id}`, 3);
}
export async function busRestart(ctx, id) {
  const org = await target(ctx, id);
  checkedState(org);
  const state = readHostState(org.root);
  const options = state?.options ?? {};
  await busStop(ctx, id);
  return busStart(ctx, id, { detach: true, ...options });
}

// Recover only while the org lifecycle fence excludes new starts and deletion.
export async function busRecover(ctx, id) {
  return withOrgLifecycle(ctx, id, async () => {
    const org = await target(ctx, id, true);
    const inspect = (path, socket = false) => {
      let st; try { st = lstatSync(path); } catch (e) { if (e.code === 'ENOENT') return null; throw e; }
      if (st.isSymbolicLink() || st.uid !== process.getuid() || (socket ? !st.isSocket() : !st.isFile() || (st.mode & 0o777) !== 0o600 || st.nlink !== 1)) fail('unsafe bus recovery state');
      if (socket) return { path, st };
      const fd = openSync(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
      try {
        const actual = fstatSync(fd);
        if (actual.ino !== st.ino || actual.dev !== st.dev) fail('bus recovery state changed');
        let value; try { value = JSON.parse(readFileSync(fd, 'utf8')); } catch { fail('unreadable bus recovery state'); }
        return { path, st, value };
      } finally { closeSync(fd); }
    };
    for (const dir of [org.paths.hostDir, join(org.root, 'bus')]) {
      let st; try { st = lstatSync(dir); } catch (e) { if (e.code === 'ENOENT') continue; throw e; }
      if (!st.isDirectory() || st.isSymbolicLink() || st.uid !== process.getuid() || (st.mode & 0o077)) fail('unsafe bus recovery directory');
    }
    const state = inspect(hostFile(org.root)), lock = inspect(hostLock(org.root));
    const writer = inspect(join(org.root, 'bus', 'writer.lock')), socket = inspect(org.paths.socket, true);
    if (!state) fail('bus host state is missing');
    const h = state.value;
    if (!Number.isSafeInteger(h?.pid) || h.pid <= 0 || typeof h.startTime !== 'string' || !/^\d+$/.test(h.startTime) || typeof h.startedAt !== 'string' || !Number.isFinite(Date.parse(h.startedAt)) || h.business !== id) fail('invalid or mismatched bus host state');
    let p; try { p = processState(h.pid); } catch { fail('bus host identity is uncertain'); }
    if (p.alive && p.startTime === h.startTime) {
      const argv = cmdlineOf(h.pid); if (!argv || !argv.length) fail('bus host identity is uncertain');
      const i = argv.indexOf(CLI);
      if (i >= 0 && argv[i + 1] === 'bus' && argv[i + 2] === 'start' && argv[i + 3] === id) fail(`bus host pid ${h.pid} is alive`);
    }
    if (lock && (lock.value?.pid !== h.pid || lock.value?.startTime !== h.startTime || lock.value?.business !== id || lock.value?.startedAt !== h.startedAt)) fail('bus host lock does not match host state');
    if (writer) {
      let w; try { w = processState(writer.value?.pid); } catch { fail('bus writer identity is uncertain'); }
      if (w.alive) fail(`bus writer pid ${writer.value.pid} is alive`);
    }
    if (socket) {
      const occupied = await new Promise(resolve => {
        const client = connect(org.paths.socket); let done = false;
        const finish = value => { if (!done) { done = true; client.destroy(); resolve(value); } };
        client.setTimeout(1000, () => finish(true));
        client.on('connect', () => finish(true));
        client.on('error', e => finish(!['ENOENT', 'ECONNREFUSED'].includes(e.code)));
      });
      if (occupied) fail('bus socket answers or is uncertain');
    }
    const entries = [state, lock, writer, socket].filter(Boolean);
    for (const entry of entries) {
      const current = lstatSync(entry.path);
      if (current.dev !== entry.st.dev || current.ino !== entry.st.ino) fail('bus recovery state changed');
    }
    const at = new Date().toISOString();
    const receipt = join(org.paths.hostDir, `recovery-${at.replace(/[:.]/g, '-')}-${randomBytes(4).toString('hex')}.json`);
    writeFileSync(receipt, JSON.stringify({ action: 'bus.recover', at, org: id, stopped: true, host: h, lock: lock?.value ?? null, writer: writer?.value ?? null, socket: Boolean(socket) }, null, 2) + '\n', { mode: 0o600, flag: 'wx' });
    for (const entry of entries.slice().reverse()) unlinkSync(entry.path);
    return { org: id, recovered: true, pid: h.pid, receipt };
  });
}