import { createServer } from 'node:net';
import { lstatSync, chmodSync, unlinkSync } from 'node:fs';
import { BusError, TASK_VERBS, WEB_VERBS, HOST_VERBS, HOSTS_VERBS, GRANTS_VERBS, PROJECT_FILE_VERBS, LIBRARY_VERBS, PROJECT_SAVE_VERB, PROJECT_SAVE_REQUEST_MAX, CONSOLE_VERBS } from './broker.mjs';
import { inspectInterpreter } from './interpreter.mjs';
import { createPeerReader } from './peercred.mjs';
const LIMIT = 65536;
// Row 148: a document save may carry up to 1 MiB of text. Only a request that names that verb near
// its start (the proof object comes first) may pass LIMIT; the parsed verb is checked again below.
const PROJECT_SAVE_AT_START = new RegExp(`"verb":"${PROJECT_SAVE_VERB.replaceAll('.', '\\.')}"`);
const LIBRARY_SAVE_AT_START = /"verb":"library\.file\.save"/;
const SAVE_VERB_AT_START = { test: text => PROJECT_SAVE_AT_START.test(text) || LIBRARY_SAVE_AT_START.test(text) };
// Preserve every code and scoped reason, adding only static rail/next-step guidance.
const refusal = (e, other) => {
  const error = e instanceof BusError ? e : new BusError(other);
  return { ok: false, error: error.code, rail: error.rail, nextStep: error.nextStep,
    ...(typeof error.reason === 'string' ? { reason: error.reason } : {}) };
};
// 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).
// `host` (row 210) is the trusted host.run executor (host-run.mjs), or null: host.run then refuses
// host-not-configured.
// A request carries exactly one of cap, human (row S4), tester (row 110), user (FED R3) and console
// (row 252) proof.

export async function serve({ broker, path, authenticateHuman = null, authenticateTester = null, authenticateUser = null, authenticateConsole = null, peerReader = null, timeout = 5000, tasks = null, web = null, host = null }) {
  try {
    lstatSync(path);
    throw new BusError('socket-exists');
  } catch (e) {
    if (e.code !== 'ENOENT') throw e;
  }
  const peers = peerReader ?? createPeerReader({ interpreter: inspectInterpreter() });
  const sockets = new Set();
  const server = createServer({ allowHalfOpen: true }, (socket) => {
    sockets.add(socket);
    let boundCap;
    const release = () => { if (boundCap) { broker.disconnect(boundCap); boundCap = null; } };
    socket.on('close', () => { sockets.delete(socket); release(); });
    socket.on('error', () => {});
    let input = Buffer.alloc(0),
      done = false, processing = false;
    const finish = (result) => {
      if (done) return;
      done = true;
      release();
      socket.resume();
      socket.end(JSON.stringify(result) + '\n');
    };
    socket.setTimeout(timeout, () => {
      finish(refusal(new BusError('request-timeout')));
      socket.destroySoon();
    });
    socket.on('data', async (chunk) => {
      if (done) return;
      if (processing) return finish(refusal(new BusError('invalid-envelope')));
      const total = input.length + chunk.length;
      if (total > LIMIT && !(total <= PROJECT_SAVE_REQUEST_MAX &&
          SAVE_VERB_AT_START.test(Buffer.concat([input, chunk]).subarray(0, 1024).toString('latin1')))) {
        finish(refusal(new BusError('request-too-large')));
        return;
      }
      input = Buffer.concat([input, chunk]);
      const end = input.indexOf(10);
      if (end < 0) return;
      processing = true;
      let transient;
      try {
        if (end !== input.length - 1) throw new BusError('invalid-envelope');
        const r = JSON.parse(input.subarray(0, end).toString('utf8'));
        if (input.length > LIMIT && ![PROJECT_SAVE_VERB, 'library.file.save'].includes(r?.verb)) throw new BusError('request-too-large');
        if (
          !r ||
          Array.isArray(r) ||
          typeof r !== 'object' ||
          Object.keys(r).some((k) => !['cap', 'human', 'tester', 'user', 'console', 'verb', 'args'].includes(k)) ||
          typeof r.verb !== 'string' ||
          [r.cap, r.human, r.tester, r.user, r.console].filter(Boolean).length !== 1
        )
          throw new BusError('invalid-envelope');
        let cap = r.cap;
        const kind = r.human ? 'human' : r.tester ? 'tester' : r.user ? 'user' : r.console ? 'console' : null;
        // Row 252 (#1753): a Console owner proof reaches its two verbs only, refused before any proof check.
        if (kind === 'console' && !CONSOLE_VERBS.includes(r.verb)) throw new BusError('human-required');
        if (kind) {
          const authenticate = kind === 'human' ? authenticateHuman : kind === 'tester' ? authenticateTester : kind === 'user' ? authenticateUser : authenticateConsole;
          if (!authenticate) throw new BusError(kind === 'tester' ? 'tester-required' : 'human-required');
          socket.pause();
          let peer;
          try { peer = await peers.read(socket); }
          catch { throw new BusError('peercred-required'); }
          if (done || socket.destroyed) return;
          transient = cap = authenticate(r[kind], peer);
          boundCap = transient;
          socket.resume();
        }
        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(refusal(e, 'adapter-failed')),
            )
            .finally(() => {
              if (own) release();
            });
          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(refusal(e, 'web-failed')),
            )
            .finally(() => {
              if (own) release();
            });
          return;
        }
        // Row 210 (#1707, design D5): a sudo command can run for minutes. As for web, the longer wait
        // goes only to a call the broker admitted: the call's own limit plus 10 s.
        if (HOST_VERBS.includes(r.verb)) {
          const own = transient;
          transient = null;
          broker
            .requestHost(cap, { verb: r.verb, args: r.args ?? {} }, host, { onAdmit: (ms) => socket.setTimeout(ms + 10000) })
            .then(
              (result) => finish({ ok: true, result }),
              (e) => finish(refusal(e, 'host-failed')),
            )
            .finally(() => {
              if (own) release();
            });
          return;
        }
        // Async effects retain the synchronous path for every other verb.
        if (r.verb === 'processing.halt' || r.verb === 'processing.resume' ||
            r.verb === 'usage.begin' || r.verb === 'usage.end' ||
            r.verb === 'spec.read' || r.verb === 'record.write' || PROJECT_FILE_VERBS.includes(r.verb) || LIBRARY_VERBS.includes(r.verb) ||
            // Row 234 (#1740): a grant raise awaits its keyscan, and hosts.check its ssh.
            HOSTS_VERBS.includes(r.verb) ||
            // Row 236 (#1742): the grants verbs may call Gitea several times, 15 s each at most.
            GRANTS_VERBS.includes(r.verb) ||
            r.verb === 'proxy.init' ||
            (r.verb === 'decision.raise' && (Object.hasOwn(r.args ?? {}, 'spec_path') || Object.hasOwn(r.args ?? {}, 'spec_commit') ||
              Object.hasOwn(r.args ?? {}, 'grant')))) {
          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(refusal(e, 'invalid-envelope')); },
          ).finally(() => { if (own) release(); });
          return;
        }
        const result = broker.request(cap, { verb: r.verb, args: r.args ?? {} });
        // C2-06: a guarded message.send and a library business's thread.messages read Git first, so
        // the broker answers them with a promise. Every other verb stays synchronous.
        if (result instanceof Promise) {
          const own = transient;
          transient = null;
          socket.setTimeout(120_000);
          result.then(
            value => { if (!socket.destroyed) finish({ ok: true, result: value }); },
            e => { if (!socket.destroyed) finish(refusal(e, 'invalid-envelope')); },
          ).finally(() => { if (own) release(); });
          return;
        }
        finish({ ok: true, result });
      } catch (e) {
        finish(refusal(e, 'invalid-envelope'));
      } finally {
        if (transient) release();
      }
    });
    socket.on('end', () => {
      if (!done && !processing) finish(refusal(new BusError('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() {
      await peers.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;
      }
    },
  };
}