// The host's git socket (#1587), `<dataRoot>/bus-host/git.sock` (0600). A
// coder's runner pushes its work to the business's working repository
// through it, and a reviewer's runner fetches a commit it was asked to
// review. The bundle is the only thing that crosses: the host never runs git
// in a session's clone and never opens a path in a session's workspace
// (packages/seat/src/working.mjs says why). The sandbox binds the socket only
// for a role that holds git.push.working or review.verdict.
//
// Every connection starts with one JSON line of at most 4096 bytes.
//
//   push   {"cap", "op": "push", "task_ref", "project", "branch", "size"}
//          The host answers {"ok": true, "send": <size>} or a refusal. On
//          "send" the runner writes exactly <size> bytes of a v2 git bundle,
//          and the host answers {"ok": true, "result": {ref, commit,
//          previous}} or a refusal.
//   fetch  {"cap", "op": "fetch", "task_ref", "request", "have": [<oid>...]}
//          The host answers {"ok": true, "result": {project, ref, commit,
//          size}} followed by <size> bytes of a bundle whose one ref is
//          refs/mosaic/review, or a refusal.
//
// A refusal is {"ok": false, "error": "<code>"}, and the connection ends.
// The host checks, in order: identity (a role of this business), the request
// (task ref, project, branch, size, at most 32 MiB: bundle-too-large), then
// the broker: git.push.working for a push, the review.requested for a fetch
// (packages/bus/src/broker.mjs). Only then does it take any bundle bytes.
// A push is recorded, as the action.allowed for git.push.working with its
// commit, under the repository's lock once the ref moved; when the broker
// refuses the record, the ref goes back.
// The host's own refusals go through the broker's refuse(), so each is an
// action.refused event on the task's trail; the broker records its own.
//
// A bundle has to arrive within bodyDeadline(size) and with no gap longer
// than BODY_IDLE_MS (body-timeout), so a slow sender can't hold one of the
// MAX_ACTIVE slots. A reviewer gone before its bundle was sent is
// fetch-aborted. git-failed and rollback-failed are the host's own faults,
// and go to the log as well.
//
// close() answers host-closing to a connection still waiting for its request
// line, since nothing of it ran, and drops every other connection.
//
// A bundle file lives in bus-host/ (0700) under a random name, and the host
// deletes it on every path.

import { randomBytes } from "node:crypto";
import { closeSync, createReadStream, existsSync, lstatSync, openSync, rmSync, statSync, writeSync, chmodSync } from "node:fs";
import { createServer } from "node:net";
import { join } from "node:path";
import { gitSocketPath } from "../../seat/src/session.mjs";
import { BUNDLE_MAX, FSCK, HAVE_MAX, push, review, validBranch, workRef, WorkingError } from "../../seat/src/working.mjs";
import { CliError } from "./errors.mjs";

const HEADER_MAX = 4096;
const HEADER_TIMEOUT_MS = 10000;
// The longest the host waits between two chunks of a bundle.
const BODY_IDLE_MS = 30000;
// The slowest rate a bundle may come at, in bytes per millisecond (256 KB/s).
// The runner sends from a file over a local socket, so a real push is far
// faster.
const BODY_MIN_RATE = 256;
// The longest a whole bundle of `size` bytes may take: 30 s plus `size` at
// BODY_MIN_RATE, so 161 s at the 32 MiB cap.
export const bodyDeadline = (size) => BODY_IDLE_MS + Math.ceil(size / BODY_MIN_RATE);
// Connections served at once; more refuse git-busy.
const MAX_ACTIVE = 8;
const CAP = /^[0-9a-f]{64}$/;
const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;

// A refusal the host decided; it goes to the broker's refuse().
class Refusal extends Error {
  constructor(code, message = code) {
    super(message);
    this.code = code;
  }
}
// A refusal the broker decided and recorded itself.
class Recorded extends Error {
  constructor(code) {
    super(code);
    this.code = code;
  }
}

// Resolves { req, rest } once a whole first line came; rest is what followed
// it. Until then `waiting` holds a function that refuses it host-closing.
function readHeader(socket, waiting) {
  return new Promise((resolve, reject) => {
    let buf = Buffer.alloc(0);
    const done = (err, value) => {
      waiting.delete(socket);
      socket.off("data", onData);
      socket.off("end", onGone);
      socket.off("close", onGone);
      socket.off("timeout", onGone);
      socket.setTimeout(0);
      socket.pause();
      err ? reject(err) : resolve(value);
    };
    const onGone = () => done(new Refusal("invalid-request", "no request line"));
    const onData = (chunk) => {
      buf = Buffer.concat([buf, chunk]);
      const nl = buf.indexOf(0x0a);
      if (nl < 0 ? buf.length > HEADER_MAX : nl > HEADER_MAX) return done(new Refusal("invalid-request", "request line too long"));
      if (nl < 0) return;
      let req;
      try {
        req = JSON.parse(buf.subarray(0, nl).toString("utf8"));
      } catch {
        return done(new Refusal("invalid-request", "request line isn't JSON"));
      }
      done(null, { req, rest: buf.subarray(nl + 1) });
    };
    waiting.set(socket, () => done(new Refusal("host-closing")));
    socket.on("data", onData);
    socket.once("end", onGone);
    socket.once("close", onGone);
    socket.setTimeout(HEADER_TIMEOUT_MS);
    socket.once("timeout", onGone);
  });
}

// Exactly `size` bytes from the socket into a new file (0600), within
// `deadlineMs` and with no gap over BODY_IDLE_MS.
function readBody(socket, file, size, deadlineMs) {
  return new Promise((resolve, reject) => {
    const fd = openSync(file, "wx", 0o600);
    let got = 0;
    let settled = false;
    const done = (err) => {
      if (settled) return;
      settled = true;
      clearTimeout(deadline);
      socket.off("data", onData);
      socket.off("end", onShort);
      socket.off("close", onShort);
      socket.off("timeout", onIdle);
      socket.setTimeout(0);
      socket.pause();
      closeSync(fd);
      err ? reject(err) : resolve();
    };
    const onShort = () => done(new Refusal("invalid-request", `the bundle ended after ${got} of ${size} bytes`));
    const onIdle = () => done(new Refusal("body-timeout", `no bundle bytes for ${BODY_IDLE_MS} ms, after ${got} of ${size}`));
    const deadline = setTimeout(() => done(new Refusal("body-timeout", `${got} of ${size} bundle bytes in ${deadlineMs} ms`)), deadlineMs);
    const onData = (chunk) => {
      if (got + chunk.length > size) return done(new Refusal("invalid-request", "more bytes than the request's size"));
      try {
        writeSync(fd, chunk);
      } catch (e) {
        return done(e);
      }
      got += chunk.length;
      if (got === size) done();
    };
    socket.on("data", onData);
    socket.once("end", onShort);
    socket.once("close", onShort);
    socket.setTimeout(BODY_IDLE_MS);
    socket.once("timeout", onIdle);
    socket.resume();
  });
}

// The file's bytes to the socket, then its end. A failed read of the host's
// own file throws as it is. The reviewer reads exactly the bundle's size and
// hangs up, which can come before the loop sees the file's end or before
// 'finish'. So a close is fetch-aborted while chunks are still to go, or when
// writing one failed (EPIPE: the reader left with bytes still queued here).
// Exported for the tests.
export async function send(socket, file) {
  const aborted = () => new Refusal("fetch-aborted", "the reviewer left before the bundle was sent");
  const served = () => !socket.errored;
  // Resolves on `event`; a close first resolves too unless a write failed.
  const until = (event) =>
    new Promise((resolve, reject) => {
      const ok = () => {
        socket.off("close", gone);
        resolve();
      };
      const gone = () => {
        socket.off(event, ok);
        served() ? resolve() : reject(aborted());
      };
      socket.once(event, ok);
      socket.once("close", gone);
    });
  for await (const chunk of createReadStream(file)) {
    if (socket.destroyed || socket.writableEnded) throw aborted();
    if (!socket.write(chunk)) await until("drain");
  }
  if (socket.destroyed) {
    if (served()) return;
    throw aborted();
  }
  const flushed = until("finish");
  socket.end();
  await flushed;
  if (!served()) throw aborted();
}

// host: startHost()'s handle; business: loadBusiness() output. bundleMax,
// bodyDeadline, maxActive and fsck are for the tests.
export function createGitSocket({ host, business: b, dataRoot, env = process.env, log = () => {}, bundleMax = BUNDLE_MAX, bodyDeadline: deadlineOf = bodyDeadline, maxActive = MAX_ACTIVE, fsck = FSCK }) {
  const path = gitSocketPath(dataRoot);
  const dir = join(dataRoot, "bus-host");
  let server = null;
  let closed = false;
  const sockets = new Set();
  const active = new Set();
  // Connections still reading their request line (readHeader).
  const waiting = new Map();

  const op = async (message) => {
    try {
      return await host.op(message);
    } catch (e) {
      throw new Recorded(e.code ?? "not-authorized");
    }
  };
  const project = (id) => (typeof id === "string" && Object.hasOwn(b.projects, id) ? b.projects[id] : null);

  async function doPush(socket, cap, req, rest, file) {
    const { task_ref, branch, size } = req;
    const p = project(req.project);
    if (!p || !validBranch(branch) || !Number.isSafeInteger(size) || size < 1) throw new Refusal("invalid-request");
    workRef(task_ref);
    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
    // The runner waits for "send"; bytes before it break the protocol.
    if (rest.length) throw new Refusal("invalid-request", "bundle bytes before send");
    await op({ op: "checkPush", cap, task_ref });
    socket.write(`${JSON.stringify({ ok: true, send: size })}\n`);
    await readBody(socket, file, size, deadlineOf(size));
    const result = await push({
      dataRoot, business: b.id, project: req.project, root: p.root, taskRef: task_ref, branch, bundle: file, env, fsck,
      record: (r) => op({ op: "recordPush", cap, push: { task_ref, project: req.project, branch, ...r } }),
    });
    socket.end(`${JSON.stringify({ ok: true, result })}\n`);
  }

  async function doFetch(socket, cap, req, file) {
    const { task_ref, request, have = [] } = req;
    if (typeof request !== "string" || !Array.isArray(have) || have.length > HAVE_MAX || have.some((h) => typeof h !== "string" || !OID.test(h))) throw new Refusal("invalid-request");
    workRef(task_ref);
    const r = await op({ op: "reviewFetch", cap, args: { task_ref, request } });
    const p = project(r.project);
    if (!p) throw new Refusal("unknown-project");
    const made = await review({ dataRoot, business: b.id, project: r.project, root: p.root, taskRef: task_ref, commit: r.commit, have, out: file, env });
    const size = statSync(file).size;
    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
    socket.write(`${JSON.stringify({ ok: true, result: { project: r.project, ref: made.ref, commit: made.commit, size } })}\n`);
    await send(socket, file);
  }

  async function serve(socket) {
    let cap = null;
    let subject = null;
    let file = null;
    const fail = (code) => {
      if (!socket.destroyed) socket.end(`${JSON.stringify({ ok: false, error: code })}\n`);
    };
    try {
      const { req, rest } = await readHeader(socket, waiting);
      if (typeof req?.cap !== "string" || !CAP.test(req.cap)) return fail("unauthenticated");
      if (closed) return fail("host-closing");
      let who;
      try {
        who = await host.op({ op: "identity", cap: req.cap });
      } catch (e) {
        return fail(e.code ?? "unauthenticated");
      }
      cap = req.cap;
      if (typeof req.task_ref === "string") subject = req.task_ref;
      if (!who.role || who.business !== b.id) throw new Refusal("not-a-role");
      file = join(dir, `${randomBytes(8).toString("hex")}.bundle`);
      if (req.op === "push") await doPush(socket, cap, req, rest, file);
      else if (req.op === "fetch") await doFetch(socket, cap, req, file);
      else throw new Refusal("invalid-request", "op is push or fetch");
    } catch (e) {
      if (e instanceof Recorded) return fail(e.code);
      const code = e instanceof Refusal || e instanceof WorkingError ? e.code : "git-failed";
      if (code === "git-failed" || code === "rollback-failed") log(`git socket: ${code}: ${e.message}`);
      if (cap) await host.op({ op: "refuse", cap, code, subject }).catch(() => {});
      fail(code);
    } finally {
      if (file) rmSync(file, { force: true });
    }
  }

  async function listen() {
    if (existsSync(path)) {
      if (!lstatSync(path).isSocket()) throw new CliError(`${path} exists and isn't a socket`, 3);
      rmSync(path);
    }
    server = createServer((socket) => {
      sockets.add(socket);
      socket.once("close", () => sockets.delete(socket));
      socket.on("error", () => {});
      if (active.size >= maxActive) return socket.end(`${JSON.stringify({ ok: false, error: "git-busy" })}\n`);
      const job = serve(socket).catch((e) => log(`git socket: ${e.message}`));
      active.add(job);
      job.finally(() => active.delete(job));
    });
    // bus-host/ is 0700, so the moment before the chmod exposes nothing.
    await new Promise((resolve, reject) => {
      server.once("error", reject);
      server.listen(path, resolve);
    });
    chmodSync(path, 0o600);
  }

  // A push whose bundle already came finishes, under its lock, while the
  // broker still runs: the launcher's close() runs before the host stops it.
  // A connection still waiting for its request line hears host-closing first.
  async function close() {
    closed = true;
    if (!server) return;
    const stopped = new Promise((r) => server.close(r));
    const told = new Set(waiting.keys());
    for (const stop of [...waiting.values()]) stop();
    for (const socket of sockets) if (!told.has(socket)) socket.destroy();
    await Promise.allSettled([...active]);
    // The refusal is a few bytes; once it is flushed, or the peer is gone,
    // nothing more is owed.
    await Promise.all([...told].map((s) => (s.destroyed || s.writableFinished ? null : new Promise((r) => (s.once("finish", r), s.once("close", r))))));
    for (const socket of sockets) socket.destroy();
    await stopped;
    rmSync(path, { force: true });
  }

  return { path, listen, close };
}