stack/packages/bus/src/server.mjs

import { createServer } from 'node:net';
import { lstatSync, chmodSync, unlinkSync } from 'node:fs';
import { BusError, TASK_VERBS, WEB_VERBS } from './broker.mjs';
const LIMIT = 65536;
// One request per connection. There is deliberately no reconnect/retry or SQL verb.
// `tasks` ({handle, timeout}) is the trusted S3 adapter; only the eight task verbs reach it.
// `web` is the trusted web service (web.mjs), or a function from a business id to one (or null); only
// the web verbs reach it, and without one they refuse web-not-configured (or web-not-granted).
// A request carries exactly one of cap, human (row S4) and tester (row 110) proof.
export async function serve({ broker, path, authenticateHuman = null, authenticateTester = null, timeout = 5000, tasks = null, web = null }) {
  try {
    lstatSync(path);
    throw new BusError('socket-exists');
  } catch (e) {
    if (e.code !== 'ENOENT') throw e;
  }
  const sockets = new Set();
  const server = createServer({ allowHalfOpen: true }, (socket) => {
    sockets.add(socket);
    socket.on('close', () => sockets.delete(socket));
    socket.on('error', () => {});
    let input = Buffer.alloc(0),
      done = false;
    const finish = (result) => {
      if (done) return;
      done = true;
      socket.end(JSON.stringify(result) + '\n');
    };
    socket.setTimeout(timeout, () => {
      finish({ ok: false, error: 'request-timeout' });
      socket.destroySoon();
    });
    socket.on('data', (chunk) => {
      if (done) return;
      if (input.length + chunk.length > LIMIT) {
        finish({ ok: false, error: 'request-too-large' });
        return;
      }
      input = Buffer.concat([input, chunk]);
      const end = input.indexOf(10);
      if (end < 0) return;
      let transient;
      try {
        if (end !== input.length - 1) throw new BusError('invalid-envelope');
        const r = JSON.parse(input.subarray(0, end).toString('utf8'));
        if (
          !r ||
          Array.isArray(r) ||
          typeof r !== 'object' ||
          Object.keys(r).some((k) => !['cap', 'human', 'tester', 'verb', 'args'].includes(k)) ||
          typeof r.verb !== 'string' ||
          [r.cap, r.human, r.tester].filter(Boolean).length !== 1
        )
          throw new BusError('invalid-envelope');
        let cap = r.cap;
        if (r.human) {
          if (!authenticateHuman) throw new BusError('human-required');
          transient = cap = authenticateHuman(r.human);
        }
        if (r.tester) {
          if (!authenticateTester) throw new BusError('tester-required');
          transient = cap = authenticateTester(r.tester);
        }
        if (typeof cap !== 'string') throw new BusError('unauthenticated');
        if (tasks && TASK_VERBS.includes(r.verb)) {
          const own = transient;
          transient = null;
          // A Vikunja write can outlast the idle timeout; a late reply is still outcome-unknown to the client.
          socket.setTimeout(tasks.timeout);
          broker
            .requestTask(cap, { verb: r.verb, args: r.args ?? {} }, tasks.handle)
            .then(
              (result) => finish({ ok: true, result }),
              (e) => finish({ ok: false, error: e instanceof BusError ? e.code : 'adapter-failed' }),
            )
            .finally(() => {
              if (own) broker.disconnect(own);
            });
          return;
        }
        if (WEB_VERBS.includes(r.verb)) {
          const own = transient;
          transient = null;
          const service = typeof web === 'function' ? web(broker.identity(cap).business) : web;
          // A fetch outlasts the idle timeout; the service's own deadline bounds it. The longer wait
          // is given only to a call the broker admitted (grant and arguments checked), so a refused
          // call can't hold the connection for it (vikunja:19/307).
          broker
            .requestWeb(cap, { verb: r.verb, args: r.args ?? {} }, service, { onAdmit: () => socket.setTimeout(service.timeout) })
            .then(
              (result) => finish({ ok: true, result }),
              (e) => finish({ ok: false, error: e instanceof BusError ? e.code : 'web-failed' }),
            )
            .finally(() => {
              if (own) broker.disconnect(own);
            });
          return;
        }
        // Row138: only project Git effects need an await; preserve every other sync path below.
        if (r.verb === 'spec.read' || r.verb === 'record.write' ||
            (r.verb === 'decision.raise' && (Object.hasOwn(r.args ?? {}, 'spec_path') || Object.hasOwn(r.args ?? {}, 'spec_commit')))) {
          const own = transient;
          transient = null;
          // Local Git may take 100 seconds; keep its result beyond the ordinary idle deadline.
          socket.setTimeout(120_000);
          Promise.resolve().then(() => broker.request(cap, { verb: r.verb, args: r.args ?? {} })).then(
            result => { if (!socket.destroyed) finish({ ok: true, result }); },
            e => { if (!socket.destroyed) finish({ ok: false, error: e instanceof BusError ? e.code : 'invalid-envelope' }); },
          ).finally(() => { if (own) broker.disconnect(own); });
          return;
        }
        const result = broker.request(cap, { verb: r.verb, args: r.args ?? {} });
        finish({ ok: true, result });
      } catch (e) {
        finish({ ok: false, error: e instanceof BusError ? e.code : 'invalid-envelope' });
      } finally {
        if (transient) broker.disconnect(transient);
      }
    });
    socket.on('end', () => {
      if (!done) finish({ ok: false, error: 'incomplete-request' });
    });
  });
  await new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(path, () => {
      server.removeListener('error', reject);
      resolve();
    });
  });
  chmodSync(path, 0o600);
  const owned = lstatSync(path);
  return {
    async close() {
      for (const s of sockets) s.destroy();
      await new Promise((resolve) => server.close(resolve));
      try {
        const cur = lstatSync(path);
        if (cur.ino === owned.ino && cur.dev === owned.dev) unlinkSync(path);
      } catch (e) {
        if (e.code !== 'ENOENT') throw e;
      }
    },
  };
}