// node --test tests/loopback.test.mjs   (needs `cargo build --release` and ../atlibs; no VM, no market, no account)
// The WebRTC check: `webrtc-peer` (this crate's binary: str0m, the guest
// side: age-sealed offer -> answer, datachannel label -> TCP port) in front of a throwaway sshd (ecdsa host
// key: the page's ssh library has no ed25519), and web/loopback.html in headless Chrome: the
// page runs com.fedfork.exec.webrtc.connect {tag: ssh, command, stdin} through the local signal
// endpoint (same sealed messages as the records), then ssh exec over the datachannel; and the webrtc:
// fetcher to a stub MCP server (Streamable HTTP) behind tag shell. Skips (with the reason)
// when Chrome, sshd or ssh-keygen is missing, or there is no `cargo build --release`.
import assert from "node:assert/strict";
import { access, mkdtemp, readFile, writeFile } from "node:fs/promises";
import { createServer as createHttpServer } from "node:http";
import { tmpdir, userInfo } from "node:os";
import { extname, join, normalize } from "node:path";
import { after, before, describe, test } from "node:test";

// The core's loopback helpers, from the sibling checkout (../atlibs).
import { cleanup, exec, missing, servers, spawnUntil, sshd, temps, which } from "../../atlibs/rust/host/loopback.mjs";

const ROOT = new URL("../", import.meta.url).pathname;

after(cleanup);

// Headless Chrome over the DevTools protocol (one browser WebSocket, flattened target sessions).
const CHROMES = ["/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "google-chrome", "google-chrome-stable", "chromium", "chromium-browser"];
async function chrome() {
	for (const c of [process.env.CHROME, ...CHROMES].filter(Boolean)) {
		if (c.startsWith("/") ? await access(c).then(() => true, () => false) : await which(c)) return c;
	}
	return null;
}

async function devtools(bin, profile) {
	const args = ["--headless=new", "--remote-debugging-port=0", `--user-data-dir=${profile}`, "--no-first-run", "--no-default-browser-check", "--disable-gpu",
		// loopback ICE candidates, and real addresses instead of mDNS names (the peer cannot resolve .local)
		"--allow-loopback-in-peer-connection", "--disable-features=WebRtcHideLocalIpsWithMdns", "about:blank"];
	const { match } = await spawnUntil(bin, args, /DevTools listening on (ws:\/\/\S+)/, {}, 30000);
	const ws = new WebSocket(match[1]);
	await new Promise((resolve, reject) => ((ws.onopen = resolve), (ws.onerror = reject)));
	let id = 0;
	const pending = new Map();
	ws.onmessage = ({ data }) => {
		const m = JSON.parse(data);
		const p = pending.get(m.id);
		if (!p) return;
		pending.delete(m.id);
		m.error ? p.reject(new Error(`${m.error.message} ${m.error.data ?? ""}`)) : p.resolve(m.result);
	};
	const send = (method, params = {}, sessionId) =>
		new Promise((resolve, reject) => {
			pending.set(++id, { resolve, reject });
			ws.send(JSON.stringify({ id, method, params, sessionId }));
		});
	return {
		close: () => ws.close(),
		async open(url) {
			const { targetId } = await send("Target.createTarget", { url });
			const { sessionId } = await send("Target.attachToTarget", { targetId, flatten: true });
			// An expression in the page, awaited; its value or the page's exception.
			return async (expression) => {
				const { result, exceptionDetails } = await send("Runtime.evaluate", { expression, awaitPromise: true, returnByValue: true }, sessionId);
				if (exceptionDetails) throw new Error(exceptionDetails.exception?.description ?? exceptionDetails.text);
				return result.value;
			};
		},
	};
}

// web/ as files (symlinks: atlibs/ is ../atlibs/javascript/src, webrtc/ is js/), as the page is served in production.
function staticServer(dir) {
	const types = { ".html": "text/html", ".js": "text/javascript", ".json": "application/json", ".wasm": "application/wasm" };
	return createHttpServer(async (req, res) => {
		const path = normalize(decodeURIComponent(new URL(req.url, "http://x").pathname));
		if (path.includes("..")) return res.writeHead(400).end();
		try {
			const body = await readFile(join(dir, path));
			res.writeHead(200, { "content-type": types[extname(path)] ?? "application/octet-stream" }).end(body);
		} catch {
			res.writeHead(404).end();
		}
	}).listen(0, "127.0.0.1");
}

// A stub MCP server (Streamable HTTP, JSON responses): what the guest agent serves on tag shell. tools/call
// echoes arguments.command after arguments.waitMs. seen: TCP connections (one per datachannel) and the
// Mcp-Session-Id of each request.
function mcpServer() {
	const seen = { connections: 0, sessions: [] };
	const server = createHttpServer((req, res) => {
		let body = "";
		req.on("data", (d) => (body += d)).on("end", () => {
			const m = JSON.parse(body);
			seen.sessions.push(req.headers["mcp-session-id"] ?? null);
			if (req.method !== "POST" || req.url !== "/mcp") return res.writeHead(404).end();
			if (m.id === undefined) return res.writeHead(202).end();
			const reply = (result) => res.writeHead(200, { "content-type": "application/json", ...(m.method === "initialize" && { "mcp-session-id": "s1" }) }).end(JSON.stringify({ jsonrpc: "2.0", id: m.id, result }));
			if (m.method === "initialize") return reply({ protocolVersion: "2025-11-25", capabilities: { tools: {} }, serverInfo: { name: "stub" } });
			if (m.method === "tools/list") return reply({ tools: [{ name: "exec", inputSchema: { type: "object" } }] });
			const { command, waitMs = 0 } = m.params.arguments;
			setTimeout(() => reply({ content: [{ type: "text", text: command }] }), waitMs);
		});
	}).on("connection", () => seen.connections++);
	return { server: server.listen(0, "127.0.0.1"), seen };
}

describe("webrtc: headless Chrome <-> webrtc-peer + a throwaway sshd", async () => {
	const absent = await missing("sshd", "ssh-keygen");
	const bin = await chrome();
	const peerBin = process.env.WEBRTC_PEER ?? join(ROOT, "target/release/webrtc-peer");
	const built = await access(peerBin).then(() => true, () => false);
	const skip = !bin ? "no Chrome (set CHROME to its binary)" : absent.length ? `missing ${absent.join(", ")} (OpenSSH)` : !built ? `no ${peerBin}: cargo build --release` : false;
	let dir, peer, evaluate, browser, hostFingerprint, page, mcp;
	const RECEIPT = { uri: "at://did:plc:loopback/com.publicdomainrelay.temp.market.event/receipt", cid: "bafyreiloopback" };

	before(async () => {
		if (skip) return;
		dir = await mkdtemp(join(tmpdir(), "atlibs-loopback-webrtc-"));
		temps.push(dir);
		await writeFile(join(dir, "authorized_keys"), "");
		const { port } = await sshd(dir, "ecdsa");
		hostFingerprint = (await exec("ssh-keygen", ["-lf", join(dir, "host_key.pub")])).stdout.split(" ")[1];
		const files = staticServer(join(ROOT, "web"));
		servers.push(files);
		await new Promise((r) => files.once("listening", r));
		page = `http://127.0.0.1:${files.address().port}`;
		// --origin: the page is the one origin CORS lets read the signal endpoint (and POST to it)
		mcp = mcpServer();
		servers.push(mcp.server);
		await new Promise((r) => mcp.server.once("listening", r));
		const started = await spawnUntil(peerBin, ["--tunnel", `ssh=${port}`, "--tunnel", `shell=${mcp.server.address().port}`, "--origin", page], /^\{.*\}$/m, { env: { PATH: process.env.PATH, HOME: process.env.HOME } });
		peer = JSON.parse(started.match[0]);
		// A kept profile under target/: the page's esm.sh modules (pinned versions) come from its cache after the first run.
		browser = await devtools(bin, join(ROOT, "target/loopback-chrome"));
		evaluate = await browser.open(`${page}/loopback.html?signal=${encodeURIComponent(peer.signal)}&user=${userInfo().username}`);
		for (const end = Date.now() + 30000; (await evaluate("document.title")) !== "ready"; await new Promise((r) => setTimeout(r, 100))) {
			if (Date.now() > end) throw new Error("loopback.html did not load (its modules come from esm.sh: offline with an empty cache?)");
		}
	});
	after(() => browser?.close());

	test("the local signal endpoint has the guest's age key and DTLS fingerprint for the tunnel's tag", { skip }, async () => {
		const { events } = await (await fetch(new URL("events", peer.signal))).json();
		assert.deepEqual(events.map((e) => [e.kind, e.value.tag]).sort(), [["guest.publicKey", "shell"], ["guest.publicKey", "ssh"], ["vm.onNetwork", "shell"], ["vm.onNetwork", "ssh"]]);
		assert.match(events.find((e) => e.kind === "guest.publicKey").value.key, /^age1[a-z0-9]{58}$/);
		assert.match(events.find((e) => e.kind === "vm.onNetwork").value.address, /^sha-256 ([0-9A-F]{2}:){31}[0-9A-F]{2}$/);
	});

	// review 4a L5: only the page's origin and this listener's Host get in (another site's page, DNS rebinding)
	test("the signal endpoint answers CORS for the page only and refuses other Origins and Hosts", { skip }, async () => {
		const ours = await fetch(new URL("events", peer.signal), { headers: { origin: page } });
		assert.deepEqual([ours.status, ours.headers.get("access-control-allow-origin")], [200, page]);
		const evil = await fetch(new URL("signal", peer.signal), { method: "POST", headers: { origin: "https://evil.example" }, body: "{}" });
		assert.equal(evil.status, 403);
		const { port } = new URL(peer.signal);
		const { stdout } = await exec("curl", ["-s", "-o", "/dev/null", "-w", "%{http_code}", "-H", `Host: evil.example:${port}`, new URL("events", peer.signal).href]);
		assert.equal(stdout, "403", "a rebound Host");
	});

	test("webrtc.connect from the page: sealed offer, answer, datachannel, ssh exec with stdin", { skip }, async () => {
		await writeFile(join(dir, "authorized_keys"), `${await evaluate("webrtcCubit.sshKey()")}\n`);
		const started = performance.now();
		const result = await evaluate(`webrtcCubit.call("com.fedfork.exec.webrtc.connect", { input: { receipt: ${JSON.stringify(RECEIPT)}, tag: "ssh", command: "echo over-webrtc; cat", stdin: "from-stdin" } })`);
		const ms = performance.now() - started;
		assert.deepEqual(result, { tag: "ssh", fingerprint: peer.fingerprint, output: "over-webrtc\nfrom-stdin", hostKey: { type: "ecdsa-sha2-nistp256", fingerprint: hostFingerprint, pinned: false } });
		assert.equal(await evaluate(`webrtcCubit.sshExec("echo second; exit 3")`), "second\n", "a second exec on the same ssh session");
		const { events } = await (await fetch(new URL("events", peer.signal))).json();
		assert.equal(events.filter((e) => e.kind === "guest.signalWebRTC" && e.value.kind === "answer").length, 1);
		console.log(`webrtc.connect + ssh exec: ${ms.toFixed(0)} ms`);
	});

	test("the webrtc: fetcher from the page: MCP over HTTP, a notification, two concurrent requests on two datachannels", { skip }, async () => {
		let session;
		const post = async (message) => {
			const headers = { "content-type": "application/json", accept: "application/json, text/event-stream", ...(session && { "mcp-session-id": session }) };
			const response = await evaluate(`webrtcCubit.fetch(${JSON.stringify({ receipt: RECEIPT, tag: "shell", path: "mcp", method: "POST", headers, body: JSON.stringify(message) })})`);
			session ??= response.headers["mcp-session-id"];
			return response;
		};
		const started = performance.now();
		const init = await post({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-11-25", capabilities: {}, clientInfo: { name: "loopback" } } });
		console.log(`webrtc: fetch initialize (connect + first request): ${(performance.now() - started).toFixed(0)} ms`);
		assert.deepEqual([init.status, JSON.parse(init.body).result.serverInfo.name, session], [200, "stub", "s1"]);
		assert.equal((await post({ jsonrpc: "2.0", method: "notifications/initialized" })).status, 202);
		assert.equal(JSON.parse((await post({ jsonrpc: "2.0", id: 2, method: "tools/list" })).body).result.tools[0].name, "exec");
		assert.equal(mcp.seen.connections, 1, "keep-alive: one datachannel so far");
		const call = async (id, command, waitMs) => JSON.parse((await post({ jsonrpc: "2.0", id, method: "tools/call", params: { name: "exec", arguments: { command, waitMs } } })).body);
		const [slow, fast] = await Promise.all([call(3, "slow", 1500), call(4, "fast", 0)]);
		assert.deepEqual([slow.id, slow.result.content[0].text, fast.id, fast.result.content[0].text], [3, "slow", 4, "fast"]);
		assert.equal(mcp.seen.connections, 2, "a second datachannel for the concurrent request");
		assert.deepEqual(mcp.seen.sessions, [null, "s1", "s1", "s1", "s1"], "the client's Mcp-Session-Id rides through");
		const { events } = await (await fetch(new URL("events", peer.signal))).json();
		assert.equal(events.filter((e) => e.kind === "guest.signalWebRTC" && e.value.tag === "shell").length, 1, "one peer connection for (receipt, shell)");
	});
});