// The com.fedfork.exec.iroh.connect transport for Node hosts. No deps; shells out to dumbpipe
// and ssh. A host passes it in (atlibs javascript/src/server/cli.mjs: ATLIBS_TRANSPORTS=<this repo>/js/index.js,
// or main's transports; hostCall's calls); no core path imports it.
import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
import { homedir, tmpdir } from "node:os";
import { join } from "node:path";

export const IROH_NSID = "com.fedfork.exec.iroh.connect";
const EXCHANGE_MS = 30000;
// stderr is kept only for an error message: its last STDERR_MAX bytes, however long the process runs.
const STDERR_MAX = 16384;
const tail = (text) => (text.length > STDERR_MAX ? text.slice(-STDERR_MAX) : text);
// A one-shot exchange keeps at most OUTPUT_MAX bytes of stdout (half a module's 1 MiB input, room for the
// event around it): past it the process is killed and the result says truncated.
export const OUTPUT_MAX = 1 << 19;
// dumbpipe without IROH_SECRET makes a key and prints "using secret key <hex>" to stderr, which would
// reach the module. A fresh key per process (ssh's ProxyCommand inherits the env) keeps it quiet.
// The rest of the environment stays out (ATPROTO_PASSWORD, SSH_AUTH_SOCK, ...): PATH, HOME and LANG only.
const spawnOptions = () => {
	const env = Object.fromEntries(["PATH", "HOME", "LANG"].filter((k) => process.env[k] !== undefined).map((k) => [k, process.env[k]]));
	return { stdio: ["pipe", "pipe", "pipe"], env: { ...env, IROH_SECRET: randomBytes(32).toString("hex") } };
};

// vm.onNetwork address -> dumbpipe ticket (swck requester.ValidateTicket).
export function ticket(address) {
	const t = String(address ?? "").trim().replace(/^iroh:\/\//, "");
	if (!/^[a-z0-9]{1,4096}$/.test(t)) throw new Error(`not an iroh ticket: ${address}`);
	return t;
}

// vm.hostKeys keys -> known_hosts lines. hat-7 sends publicKey as bare base64 (no type prefix),
// though the lexicon says authorized_keys form; take both, and plain authorized_keys strings.
export function knownHosts(alias, hostKeys) {
	const lines = (hostKeys ?? []).map((k) => {
		const f = (typeof k === "string" ? k : k.publicKey).trim().split(/\s+/);
		return (f.length > 1 ? f.slice(0, 2) : [k.type, f[0]]).join(" ");
	});
	if (!lines.length) throw new Error("no host keys: refusing to trust an unpinned guest");
	return lines.map((l) => `${alias} ${l}\n`).join("");
}

// ssh to root@guest through dumbpipe, the host key pinned by a known_hosts file. "--" ends option
// parsing: ssh re-parses options after the host otherwise, so a command like "-oProxyCommand=..." would run locally.
// -F /dev/null: no ~/.ssh/config or system ssh_config (ForwardAgent, RemoteForward, LocalCommand, ControlMaster)
// toward a guest the bidder controls; only these options.
export const sshArgs = (key, address, knownHostsFile, command) => [
	"-F", "/dev/null", "-i", key, "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes",
	"-o", `ProxyCommand=dumbpipe connect ${ticket(address)}`,
	"-o", "HostKeyAlias=guest", "-o", `UserKnownHostsFile=${knownHostsFile}`,
	"-o", "StrictHostKeyChecking=yes", "-o", "ConnectTimeout=60",
	"--", "root@guest", ...command,
];

// Runs cmd with input on stdin; resolves {stdout, stderr, code, truncated?} (code null: killed after ms, or
// past OUTPUT_MAX bytes of stdout, truncated). Exported for tests.
export function exchange(cmd, args, input = "", ms = EXCHANGE_MS) {
	return new Promise((resolve, reject) => {
		const child = spawn(cmd, args, spawnOptions());
		const chunks = [];
		let size = 0, stderr = "", truncated = false;
		child.stdout.on("data", (d) => {
			if (truncated) return;
			chunks.push(d);
			size += d.length;
			if (size <= OUTPUT_MAX) return;
			truncated = true;
			// our ends of the pipes too: a grandchild still holding them (ssh's ProxyCommand) never delays close
			child.stdout.destroy();
			child.stderr.destroy();
			child.kill("SIGKILL");
		});
		child.stderr.on("data", (d) => (stderr = tail(stderr + d)));
		const timer = setTimeout(() => child.kill(), ms);
		child.on("error", (e) => (clearTimeout(timer), reject(e)));
		child.on("close", (code) => {
			clearTimeout(timer);
			resolve({ stdout: Buffer.concat(chunks).subarray(0, OUTPUT_MAX).toString(), stderr, code, ...(truncated && { truncated }) });
		});
		child.stdin.on("error", () => {}); // the peer may close before reading everything
		child.stdin.end(input);
	});
}

// The com.fedfork.exec.iroh.connect host pseudo-call (lexicons/com/fedfork/exec/iroh/connect.json),
// for a Node host's call(): if (nsid === IROH_NSID) return iroh(input). key: the ssh private key
// the host holds for guests (its .pub goes in user_data). run: exchange, swappable for tests.
export function irohHost({ key, run = exchange }) {
	return async ({ ticket: address, alpn, hostKeys, command, stdin, send }) => {
		const t = ticket(address);
		if (alpn !== undefined) {
			if (typeof alpn !== "string" || !/^[\x21-\x7e]{1,64}$/.test(alpn)) throw new Error("alpn must be 1 to 64 printable ASCII bytes");
			const { stdout, stderr, code, truncated } = await run("dumbpipe", ["connect", "--custom-alpn", `utf8:${alpn}`, t], send ?? "");
			if (code !== 0 && code !== null && !stdout) throw new Error(`dumbpipe exited ${code}: ${stderr.trim()}`);
			// null: killed at the timeout (a peer that never closes) or at OUTPUT_MAX; what it sent so far, flagged (as the Rust host).
			return { output: stdout, ...(truncated ? { truncated } : code === null && { timedOut: true }) };
		}
		if (typeof command !== "string" || !command) throw new Error("command is required for ssh");
		// Secrets go over stdin, never in command: ps shows the command on the host (ssh's argv) and the guest (the shell's).
		if (stdin !== undefined && !(typeof stdin === "string" && stdin.length <= 65536)) throw new Error("stdin must be a string of at most 65536 bytes");
		const dir = await mkdtemp(join(tmpdir(), "iroh-ssh-"));
		try {
			await writeFile(join(dir, "known_hosts"), knownHosts("guest", hostKeys));
			const { stdout, stderr, code, truncated } = await run("ssh", sshArgs(key, t, join(dir, "known_hosts"), [command]), stdin ?? "", 120000);
			if (truncated) return { output: stdout, truncated };
			// 255 is ssh's own failure (host key mismatch, auth, dial); null is the timeout.
			if (code === 255 || code === null) throw new Error(`ssh ${code === null ? "timed out" : "failed"}: ${stderr.trim()}`);
			return { output: stdout, exitCode: code };
		} finally {
			await rm(dir, { recursive: true, force: true });
		}
	};
}

// The connect input -> [cmd, args, cleanup]: dumbpipe for a raw ALPN, else ssh through dumbpipe with
// the host keys pinned (command optional for a session: none is a remote shell reading stdin).
async function sessionCommand(key, { ticket: address, alpn, hostKeys, command }) {
	const t = ticket(address);
	if (alpn !== undefined) {
		if (typeof alpn !== "string" || !/^[\x21-\x7e]{1,64}$/.test(alpn)) throw new Error("alpn must be 1 to 64 printable ASCII bytes");
		return ["dumbpipe", ["connect", "--custom-alpn", `utf8:${alpn}`, t], async () => {}];
	}
	if (command !== undefined && (typeof command !== "string" || !command)) throw new Error("command must be a non-empty string");
	const dir = await mkdtemp(join(tmpdir(), "iroh-ssh-"));
	try {
		await writeFile(join(dir, "known_hosts"), knownHosts("guest", hostKeys));
	} catch (error) {
		await rm(dir, { recursive: true, force: true });
		throw error;
	}
	return ["ssh", sshArgs(key, t, join(dir, "known_hosts"), command === undefined ? [] : [command]), () => rm(dir, { recursive: true, force: true })];
}

/**
 * com.fedfork.exec.iroh.connect with subscribe: true: the same input as the
 * one-shot call, but the process stays up. Each stdout / stderr chunk is a message { stdout } / { stderr }
 * (UTF-8 text; chunk boundaries are wherever the pipe split them), the exit a last message { exitCode },
 * then the stream ends. It fails instead (closed with the error) when ssh itself fails (255) or dumbpipe
 * exits non-zero having printed nothing. send(data) writes stdin (a string as is, anything else as JSON);
 * close() and the request signal kill the process. src/lib.rs's iroh_session does the same.
 * @param {{ key: string, spawn?: typeof spawn }} options
 */
export function irohSession({ key, spawn: spawnImpl = spawn }) {
	return async (input, { signal } = {}) => {
		const [cmd, args, cleanup] = await sessionCommand(key, input ?? {});
		const child = spawnImpl(cmd, args, spawnOptions());
		const queue = [];
		let wake;
		let finished = false;
		let failure;
		let printed = false;
		let killed = false;
		let stderr = "";
		const push = (message) => (queue.push(message), wake?.());
		const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
		for (const name of ["stdout", "stderr"]) {
			child[name].on("data", (chunk) => {
				const text = decoders[name].decode(chunk, { stream: true });
				if (name === "stdout") printed = true;
				else stderr = tail(stderr + text);
				if (text) push({ [name]: text });
			});
		}
		const end = () => void cleanup().catch(() => {}).finally(() => ((finished = true), wake?.()));
		child.on("error", (error) => ((failure ??= new Error(`spawn ${cmd} ${error.message}`)), end()));
		child.on("close", (code) => {
			if (killed || failure) return end(); // failure: spawn failed (Node closes with -2 after error)
			if (cmd === "ssh" && code === 255) failure ??= new Error(`ssh failed: ${stderr.trim()}`);
			else if (cmd === "dumbpipe" && code !== 0 && code !== null && !printed) failure ??= new Error(`dumbpipe exited ${code}: ${stderr.trim()}`);
			else push({ exitCode: code });
			end();
		});
		child.stdin.on("error", () => {}); // the peer may close first
		const close = () => {
			if (finished || killed) return;
			killed = true;
			child.kill(); // SIGTERM, so ssh takes its dumbpipe ProxyCommand down; SIGKILL if it lingers
			setTimeout(() => finished || child.kill("SIGKILL"), 2000).unref();
		};
		signal?.addEventListener("abort", close, { once: true });
		if (signal?.aborted) close();
		return {
			close,
			send: (data) =>
				new Promise((resolve, reject) => {
					if (finished || killed || !child.stdin.writable) return reject(new Error(`${IROH_NSID} session is closed`));
					child.stdin.write(typeof data === "string" ? data : JSON.stringify(data), (error) => (error ? reject(error) : resolve()));
				}),
			async *[Symbol.asyncIterator]() {
				for (;;) {
					while (!queue.length && !finished) await new Promise((resolve) => (wake = resolve));
					if (queue.length) yield queue.shift();
					else if (failure) throw failure;
					else return;
				}
			},
		};
	};
}

// The iroh: scheme fetcher (vm-mcp contract v2.2 §3): iroh:///<ref>/<socket>/<path...>, ref base64url JSON
// {ticket, hostKeys}; socket "shell" (SHELL_SOCKET) or a percent-encoded absolute unix path. HTTP/1.1 over
// `ssh root@guest socialweb-guest-agent relay <socket>` (stdin/stdout <-> the guest's unix socket) through dumbpipe.
export const SHELL_SOCKET = "/run/socialweb-guest-agent/shell.sock";
// The guest's login shell parses ssh's command: the path single-quoted.
export const relayCommand = (path) => `socialweb-guest-agent relay '${path.replaceAll("'", "'\\''")}'`;
// One request's cap (a guest exec runs up to 600 s); a response's head and body caps.
const FETCH_MS = 660000;
const HEAD_MAX = 1 << 16;
export const BODY_MAX = 1 << 24;
const TOKEN = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;

/** iroh://<ticket>/<path...> (a market endpoint: HTTP/1.1 over `dumbpipe connect <ticket>`, no ssh) -> { ticket, raw, target }; else undefined. */
export function rawUrl(url) {
	const m = /^iroh:\/\/(endpoint[a-z2-7]{1,4088})(\/[^#]*)?(#.*)?$/.exec(String(url));
	if (!m) return undefined;
	const target = m[2] || "/";
	if (/[\0-\x20\x7f]/.test(target)) throw new Error(`${url}: the path has spaces or control characters`);
	return { ticket: m[1], raw: true, target };
}

/** iroh:///<ref>/<socket>/<path...> -> { ticket, hostKeys, socket, target } (target: "/" + path, query kept). */
export function irohUrl(url) {
	const m = /^iroh:\/\/\/?([A-Za-z0-9_-]+={0,2})\/([^/?#]+)([^#]*)/.exec(String(url));
	if (!m) throw new Error(`not an iroh:///<ref>/<socket>/<path> URL: ${url}`);
	let ref, socket;
	try {
		ref = JSON.parse(Buffer.from(m[1], "base64url").toString());
	} catch {
		throw new Error(`${url}: the ref is not base64url JSON`);
	}
	try {
		socket = m[2] === "shell" ? SHELL_SOCKET : decodeURIComponent(m[2]);
	} catch {
		throw new Error(`${url}: bad percent-encoding in the socket`);
	}
	if (!socket.startsWith("/") || /[\0-\x1f\x7f]/.test(socket)) throw new Error(`${url}: the socket must be "shell" or an absolute unix path`);
	const target = m[3].startsWith("/") ? m[3] : `/${m[3]}`;
	if (/[\0-\x20\x7f]/.test(target)) throw new Error(`${url}: the path has spaces or control characters`);
	return { ticket: ticket(ref?.ticket), hostKeys: ref?.hostKeys, socket, target };
}

// One relay: an HTTP/1.1 connection to the guest's socket, one request at a time (its caller holds it).
// Unref'd: an idle relay never keeps the host process up (a request's timer does).
async function relay(key, { ticket: t, hostKeys, socket, raw }, spawnImpl) {
	const [cmd, args, cleanup] = raw ? ["dumbpipe", ["connect", t], async () => {}] : await sessionCommand(key, { ticket: t, hostKeys, command: relayCommand(socket) });
	const child = spawnImpl(cmd, args, spawnOptions());
	let buf = Buffer.alloc(0), more = [], size = 0, stderr = "", ended, wake;
	const conn = { got: 0 };
	child.stdout.on("data", (d) => (more.push(d), (size += d.length), (conn.got += d.length), wake?.()));
	child.stderr.on("data", (d) => (stderr = tail(stderr + d)));
	const end = (why) => {
		if (ended) return;
		ended = why;
		void cleanup().catch(() => {});
		if (child.exitCode === null) child.kill();
		wake?.();
	};
	child.on("error", (error) => end(`spawn ${cmd} ${error.message}`));
	child.on("close", (code) => end(`exit ${code}`));
	child.stdin.on("error", () => {});
	child.unref();
	for (const pipe of [child.stdin, child.stdout, child.stderr]) pipe.unref?.();
	const failure = () => new Error(`iroh: the relay ended (${ended})${stderr.trim() ? `: ${stderr.trim()}` : ""}`);
	const wait = () => (ended ? Promise.reject(failure()) : new Promise((resolve) => (wake = resolve)));
	const merge = () => more.length && ((buf = Buffer.concat([buf, ...more])), (more = []), (size = 0));
	const line = async () => {
		for (;;) {
			merge();
			const at = buf.indexOf("\r\n");
			if (at >= 0) {
				const text = buf.subarray(0, at).toString("latin1");
				buf = buf.subarray(at + 2);
				return text;
			}
			if (buf.length > HEAD_MAX) throw new Error(`iroh: a response line over ${HEAD_MAX} bytes`);
			await wait();
		}
	};
	const take = async (n) => {
		while (buf.length + size < n) await wait();
		merge();
		const bytes = Buffer.from(buf.subarray(0, n));
		buf = buf.subarray(n);
		return bytes;
	};
	const rest = async () => {
		while (!ended) {
			if (buf.length + size > BODY_MAX) throw new Error(`iroh: a response body over ${BODY_MAX} bytes`);
			await new Promise((resolve) => (wake = resolve));
		}
		merge();
		if (buf.length > BODY_MAX) throw new Error(`iroh: a response body over ${BODY_MAX} bytes`);
		return buf;
	};
	const head = async () => {
		const status = /^HTTP\/1\.([01]) (\d{3})/.exec(await line());
		if (!status) throw new Error("iroh: not an HTTP/1.x response");
		const fields = [];
		for (let bytes = 0, l; (l = await line()); ) {
			if ((bytes += l.length) > HEAD_MAX) throw new Error(`iroh: response headers over ${HEAD_MAX} bytes`);
			const colon = l.indexOf(":");
			if (colon > 0) fields.push([l.slice(0, colon).trim().toLowerCase(), l.slice(colon + 1).trim()]);
		}
		return { minor: status[1], status: Number(status[2]), fields };
	};
	return Object.assign(conn, {
		get ended() {
			return ended;
		},
		close: (why = "closed") => end(why),
		// -> { response: {status, headers, body}, keep }: keep, the connection can take another request
		async request(method, target, headers, body) {
			conn.got = 0;
			const lines = [`${method} ${target} HTTP/1.1`, "host: guest", ...headers.map(([k, v]) => `${k}: ${v}`), `content-length: ${body.length}`];
			child.stdin.write(Buffer.concat([Buffer.from(`${lines.join("\r\n")}\r\n\r\n`), body]));
			let h;
			do h = await head();
			while (h.status >= 100 && h.status < 200);
			const field = (name) => h.fields.filter(([k]) => k === name).map(([, v]) => v).join(", ");
			let data, keep = h.minor === "1" && !/\bclose\b/i.test(field("connection"));
			if (method === "HEAD" || h.status === 204 || h.status === 304) data = Buffer.alloc(0);
			else if (/\bchunked\b/i.test(field("transfer-encoding"))) {
				const parts = [];
				for (let total = 0; ; ) {
					const size = (await line()).split(";")[0].trim();
					const n = /^[0-9a-fA-F]{1,8}$/.test(size) ? parseInt(size, 16) : NaN;
					if (!(n >= 0) || (total += n) > BODY_MAX) throw new Error(`iroh: a bad chunk size, or a response body over ${BODY_MAX} bytes`);
					if (n === 0) break;
					parts.push(await take(n));
					await line();
				}
				while (await line()); // trailers
				data = Buffer.concat(parts);
			} else if (field("content-length")) {
				const n = Number(field("content-length").split(",")[0]);
				if (!Number.isSafeInteger(n) || n < 0 || n > BODY_MAX) throw new Error(`iroh: a bad content-length, or a response body over ${BODY_MAX} bytes`);
				data = await take(n);
			} else {
				keep = false;
				data = Buffer.from(await rest());
			}
			const out = {};
			for (const [k, v] of h.fields) out[k] = k in out ? `${out[k]}, ${v}` : v;
			return { response: { status: h.status, headers: out, body: data.toString() }, keep };
		},
	});
}

/**
 * The iroh: fetcher: ({method, url, headers, body, signal}) -> {status, headers (lowercase names), body (text)}.
 * Relays per (ref, socket) are reused (HTTP keep-alive), one request in flight on each (more are opened for
 * concurrent requests); a request whose relay failed before any response byte is retried once on a fresh
 * one. fetch.close() ends the idle relays.
 * @param {{ key: string, spawn?: typeof spawn }} options
 */
export function irohFetch({ key, spawn: spawnImpl = spawn }) {
	const idle = new Map();
	const fetch = async ({ method = "GET", url, headers = {}, body, signal } = {}) => {
		if (!TOKEN.test(method)) throw new Error(`iroh: bad method ${method}`);
		const where = rawUrl(url) ?? irohUrl(url);
		const sent = [];
		for (const [k, v] of headers instanceof Headers ? headers : Object.entries(headers ?? {})) {
			if (!TOKEN.test(k) || typeof v !== "string" || /[\r\n\0]/.test(v)) throw new Error(`iroh: bad header ${k}`);
			if (!["host", "content-length", "transfer-encoding", "connection"].includes(k.toLowerCase())) sent.push([k, v]);
		}
		const bytes = body == null ? Buffer.alloc(0) : Buffer.from(body);
		if (!where.raw) await access(key).catch(() => Promise.reject(new Error(`iroh: needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at ${key}`)));
		const name = where.raw ? `raw ${where.ticket}` : `${where.ticket} ${JSON.stringify(where.hostKeys ?? [])} ${where.socket}`;
		const attempt = async (fresh) => {
			signal?.throwIfAborted();
			const pool = idle.get(name) ?? [];
			let conn = fresh ? undefined : pool.pop();
			while (conn?.ended) conn = pool.pop();
			conn ??= await relay(key, where, spawnImpl);
			const timer = setTimeout(() => conn.close(`no response in ${FETCH_MS / 1000} s`), FETCH_MS);
			const abort = () => conn.close("aborted");
			signal?.addEventListener("abort", abort, { once: true });
			try {
				const { response, keep } = await conn.request(method, where.target, sent, bytes);
				if (keep && !conn.ended) idle.set(name, [...(idle.get(name) ?? []), conn]);
				else conn.close();
				return response;
			} catch (error) {
				conn.close();
				signal?.throwIfAborted();
				throw Object.assign(error, { retry: !fresh && conn.got === 0 });
			} finally {
				clearTimeout(timer);
				signal?.removeEventListener("abort", abort);
			}
		};
		try {
			return await attempt(false);
		} catch (error) {
			if (!error.retry) throw error;
			return attempt(true);
		}
	};
	fetch.close = () => {
		for (const pool of idle.values()) for (const conn of pool) conn.close();
		idle.clear();
	};
	return fetch;
}

// The transport: com.fedfork.exec.iroh.connect with the host's ssh private key (a path; the key never
// enters wasm), for hostCall's calls. A request is irohHost, subscribe: true an irohSession (output
// streamed as messages, send writes stdin). run, spawn: swappable for tests.
export function irohCall({ key, run, spawn: spawnImpl }) {
	const iroh = irohHost({ key, ...(run && { run }) });
	const session = irohSession({ key, ...(spawnImpl && { spawn: spawnImpl }) });
	return async (request = {}) => {
		const input = request.input ?? {};
		if (input.alpn === undefined) await access(key).catch(() => Promise.reject(new Error(`${IROH_NSID} ssh needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at ${key}`)));
		return request.subscribe ? session(input, request) : iroh(input);
	};
}

// What a host loads (javascript/src/server/cli.mjs ATLIBS_TRANSPORTS, or main's transports): connect, and the
// iroh: fetcher (runModule's fetchers); the key ATLIBS_IROH_KEY, else <state>/iroh_ed25519.
const keyOf = ({ env = process.env, stateDir = env.ATLIBS_SERVER_STATE || join(homedir(), ".atlibs-server") } = {}) => env.ATLIBS_IROH_KEY || join(stateDir, "iroh_ed25519");
export const transport = { [IROH_NSID]: (host) => irohCall({ key: keyOf(host) }) };
export const fetchers = (host) => ({ "iroh:": irohFetch({ key: keyOf(host) }) });