stack/packages/bus/src/proxy-lock.mjs

// Instance-wide serialization for a single, verified proxy state directory.
import fs from 'node:fs';
import path from 'node:path';
import { createHash, randomBytes } from 'node:crypto';

const fail = code => { throw Object.assign(new Error(code), { code }); };
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
const syncDir = dir => { const fd = fs.openSync(dir, 'r'); try { fs.fsyncSync(fd); } finally { fs.closeSync(fd); } };

export function createProxyLock({ stateDir, identity, waitMs = 5000 }) {
  if (typeof stateDir !== 'string' || !path.isAbsolute(stateDir) || typeof identity !== 'string' || !identity || !Number.isSafeInteger(waitMs) || waitMs < 0) fail('proxy-lock-invalid');
  try {
    const stat = fs.lstatSync(stateDir);
    if (!stat.isDirectory() || stat.isSymbolicLink() || fs.realpathSync(stateDir) !== path.resolve(stateDir) || stat.uid !== process.getuid() || (stat.mode & 0o077)) fail('proxy-lock-invalid');
  } catch { fail('proxy-lock-invalid'); }
  const lockFile = path.join(stateDir, 'mosaic-keys.lock');
  const guard = path.join(stateDir, 'mosaic-keys.reap');
  const fenceFile = path.join(stateDir, 'mosaic-keys.uncertain');
  const identityFile = path.join(stateDir, 'mosaic-proxy.identity');
  const digest = createHash('sha256').update(identity).digest('hex');
  function readSafe(file) {
    const stat = fs.lstatSync(file);
    if (!stat.isFile() || stat.isSymbolicLink() || stat.uid !== process.getuid() || (stat.mode & 0o077) || stat.size > 4096) fail('proxy-lock-invalid');
    return fs.readFileSync(file, 'utf8');
  }
  const identityTemp = path.join(stateDir, `.identity-${randomBytes(16).toString('hex')}`);
  try {
    const fd = fs.openSync(identityTemp, 'wx', 0o600);
    try { fs.writeFileSync(fd, digest + '\n'); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
    try { fs.linkSync(identityTemp, identityFile); syncDir(stateDir); }
    catch (error) { if (error.code !== 'EEXIST') throw error; }
  } catch { fail('proxy-lock-invalid'); }
  finally { try { fs.unlinkSync(identityTemp); } catch (error) { if (error.code !== 'ENOENT') fail('proxy-lock-invalid'); } }
  try { if (readSafe(identityFile) !== digest + '\n') fail('proxy-lock-invalid'); } catch { fail('proxy-lock-invalid'); }
  function assertCertain() {
    try { fs.lstatSync(fenceFile); } catch (error) { if (error.code === 'ENOENT') return; fail('proxy-uncertain'); }
    fail('proxy-uncertain');
  }
  function reap() {
    // An orphaned reaper is deliberately not guessed away. It needs operator inspection.
    try { fs.mkdirSync(guard, { mode: 0o700 }); } catch (error) { if (error.code === 'EEXIST') return false; fail('proxy-lock-uncertain'); }
    try {
      let before, owner;
      try {
        before = fs.lstatSync(lockFile);
        owner = JSON.parse(readSafe(lockFile));
        if (!Number.isSafeInteger(owner.pid) || owner.pid <= 0 || typeof owner.token !== 'string' || !/^[0-9a-f]{32}$/.test(owner.token)) fail('proxy-lock-uncertain');
      } catch (error) { if (error.code === 'ENOENT') return true; fail('proxy-lock-uncertain'); }
      try { process.kill(owner.pid, 0); return false; }
      catch (error) { if (error.code !== 'ESRCH') fail('proxy-lock-uncertain'); }
      const after = fs.lstatSync(lockFile);
      if (before.dev !== after.dev || before.ino !== after.ino) fail('proxy-lock-uncertain');
      fs.unlinkSync(lockFile); syncDir(stateDir); return true;
    } finally { fs.rmdirSync(guard); }
  }
  const generationFile = path.join(stateDir, 'keys.generation');
  const receiptFile = path.join(stateDir, 'mosaic-keys.receipts.jsonl');
  function generation() {
    let value;
    try { value = readSafe(generationFile); } catch (error) { if (error.code === 'ENOENT') return 0; throw error; }
    if (!/^(0|[1-9][0-9]*)\n$/.test(value) || !Number.isSafeInteger(Number(value))) fail('proxy-lock-invalid');
    return Number(value);
  }
  function status() {
    let value;
    try { value = readSafe(fenceFile); } catch (error) { if (error.code === 'ENOENT') return null; fail('proxy-uncertain'); }
    try { const entry = JSON.parse(value); return entry && typeof entry === 'object' && !Array.isArray(entry) ? entry : { invalid: true }; }
    catch { return { invalid: true }; }
  }
  function registration(org) {
    const file = path.join(stateDir, `org-${createHash('sha256').update(org).digest('hex')}.json`);
    try {
      const entry = JSON.parse(readSafe(file));
      if (entry.org !== org || typeof entry.sentinel !== 'string' || !/^[a-f0-9]{64}$/.test(entry.sentinel)) fail('proxy-lock-invalid');
      return entry.sentinel;
    } catch (error) { if (error.code === 'ENOENT') return null; fail('proxy-lock-invalid'); }
  }
  function register(org, sentinel) {
    const existing = registration(org), digest = createHash('sha256').update(sentinel).digest('hex');
    if (existing !== null) { if (existing !== digest) fail('proxy-open'); return; }
    const file = path.join(stateDir, `org-${createHash('sha256').update(org).digest('hex')}.json`);
    const temp = path.join(stateDir, `.org-${randomBytes(16).toString('hex')}`);
    const fd = fs.openSync(temp, 'wx', 0o600);
    try { fs.writeFileSync(fd, JSON.stringify({ org, sentinel: digest, time: new Date().toISOString() }) + '\n'); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
    try { fs.linkSync(temp, file); syncDir(stateDir); } finally { fs.unlinkSync(temp); }
  }
  function fence({ org = null, op = 'write' } = {}) {
    const fd = fs.openSync(fenceFile, 'wx', 0o600);
    try { fs.writeFileSync(fd, JSON.stringify({ org, op, time: new Date().toISOString(), pid: process.pid }) + '\n'); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
    syncDir(stateDir);
  }
  function confirm() {
    const value = generation() + 1;
    if (!Number.isSafeInteger(value)) fail('proxy-lock-invalid');
    const temp = path.join(stateDir, `.generation-${randomBytes(16).toString('hex')}`);
    const fd = fs.openSync(temp, 'wx', 0o600);
    try { fs.writeFileSync(fd, `${value}\n`); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
    fs.renameSync(temp, generationFile); syncDir(stateDir);
    fs.unlinkSync(fenceFile); syncDir(stateDir);
  }
  // `verify(previous)` (optional) re-proves a restart against the fence it would clear, inside
  // the lock, and returns the evidence the receipt records.
  async function clear({ restarted, read, verify }) {
    if (restarted !== true) fail('proxy-restart-required');
    return run(async () => {
      const previous = status();
      const evidence = verify ? await verify(previous) : undefined;
      await read();
      const fd = fs.openSync(receiptFile, 'a', 0o600);
      try { fs.writeFileSync(fd, JSON.stringify({ op: 'fence.clear', time: new Date().toISOString(), previous, ...(evidence === undefined ? {} : { evidence }) }) + '\n'); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
      syncDir(stateDir);
      if (previous) confirm();
      return { cleared: Boolean(previous), ...(evidence === undefined ? {} : { evidence }) };
    }, { allowUncertain: true });
  }
  async function run(operation, { allowUncertain = false } = {}) {
    const deadline = Date.now() + waitMs, token = randomBytes(16).toString('hex');
    let fd, stat;
    for (;;) {
      const temp = path.join(stateDir, `.lock-${randomBytes(16).toString('hex')}`);
      let occupied = false;
      try {
        fd = fs.openSync(temp, 'wx', 0o600);
        fs.writeFileSync(fd, JSON.stringify({ pid: process.pid, token }) + '\n'); fs.fsyncSync(fd);
        try { fs.linkSync(temp, lockFile); } catch (error) { if (error.code !== 'EEXIST') throw error; occupied = true; }
        fs.unlinkSync(temp);
        if (!occupied) { stat = fs.fstatSync(fd); syncDir(stateDir); break; }
        fs.closeSync(fd); fd = undefined;
      } catch {
        if (fd !== undefined) fs.closeSync(fd);
        try { fs.unlinkSync(temp); } catch {}
        fail('proxy-lock-uncertain');
      }
      if (reap()) continue;
      if (Date.now() >= deadline) fail('proxy-lock-busy');
      await pause(20);
    }
    try { if (!allowUncertain) assertCertain(); return await operation(); }
    finally {
      try {
        const current = fs.lstatSync(lockFile);
        if (current.dev !== stat.dev || current.ino !== stat.ino) fail('proxy-lock-uncertain');
        const owner = JSON.parse(readSafe(lockFile));
        if (owner.token !== token || owner.pid !== process.pid) fail('proxy-lock-uncertain');
        fs.unlinkSync(lockFile); syncDir(stateDir);
      } finally { fs.closeSync(fd); }
    }
  }
  return { run, fence, assertCertain, confirm, clear, status, generation, registration, register };
}