// ATIRC_BIN=<atirc> WEBRTC_PEER=<atlibs-webrtc's webrtc-peer> deno run -A test/loopback.ts
//
// Headless Chrome (test/web/loopback.html) <-> three local `atirc mesh run` nodes over WebRTC and iroh,
// loopback only. This harness is their fake PDS (getRecord/putRecord/deleteRecord over the directory the
// nodes read through ATIRC_MESH_DIRECTORY) and fake Jetstream (a putRecord is a commit event).
//   alice.test      the owner's own node: the page's bridge
//   vm.alice.test   a VM agent: record written for it (owner's channel only), no WebRTC, no publish
//   bot.agent.test  an agent account's node with WebRTC: the page's direct peer
// and a guest: vm.alice.test's --local-listen behind webrtc-peer (the guest agent's stand-in, tag atirc).
const ROOT = new URL("..", import.meta.url).pathname;
const BIN = Deno.env.get("ATIRC_BIN") ?? "atirc";
const WEBRTC_PEER = Deno.env.get("WEBRTC_PEER") ?? "webrtc-peer";
const CHROME = Deno.env.get("CHROME") ?? (Deno.build.os === "darwin" ? "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" : "google-chrome");
const OWNER = "alice.test", AGENT = "agent.test";
const did = (handle: string) => `did:plc:${handle.replaceAll(".", "")}`;

const tmp = await Deno.makeTempDir({ prefix: "atlibs-atirc-" });
const dir = `${tmp}/directory`;
const write = async (path: string, text: string) => {
	await Deno.mkdir(path.slice(0, path.lastIndexOf("/")), { recursive: true });
	await Deno.writeTextFile(path, text);
};
const recordPath = (repo: string, collection: string, rkey: string) => {
	for (const part of [repo, collection, rkey]) if (!part || part.includes("/") || part.startsWith(".")) throw new Error("bad path");
	return `${dir}/repo/${repo}/${collection}/${rkey}.json`;
};
for (const handle of [OWNER, AGENT]) {
	await write(`${dir}/did/${did(handle)}.json`, JSON.stringify({
		id: did(handle),
		alsoKnownAs: [`at://${handle}`],
		verificationMethod: [],
		service: [{ id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "https://pds.example.com" }],
	}));
	await write(`${dir}/handle/${handle}`, did(handle));
}
await write(recordPath(did(AGENT), "com.fedfork.irc.agentOperator", "self"), JSON.stringify({ owner: did(OWNER) }));
await write(recordPath(did(OWNER), "com.fedfork.irc.agents", did(AGENT)), JSON.stringify({ agent: did(AGENT) }));

const sockets = new Set<WebSocket>();
let resolveResult: (value: unknown) => void;
const result = new Promise((resolve) => (resolveResult = resolve));
const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
const steps: Record<string, () => Promise<void>> = {};
const server = Deno.serve({ hostname: "127.0.0.1", port: 0, onListen: () => {} }, async (req) => {
	const url = new URL(req.url);
	if (url.pathname === "/subscribe") {
		const { socket, response } = Deno.upgradeWebSocket(req);
		socket.onopen = () => sockets.add(socket);
		socket.onclose = () => sockets.delete(socket);
		return response;
	}
	if (url.pathname === "/xrpc/com.atproto.repo.getRecord") {
		const [repo, collection, rkey] = ["repo", "collection", "rkey"].map((k) => url.searchParams.get(k) ?? "");
		try {
			const text = await Deno.readTextFile(recordPath(repo, collection, rkey));
			return json({ uri: `at://${repo}/${collection}/${rkey}`, cid: `local${text.length}`, value: JSON.parse(text) });
		} catch {
			return json({ error: "RecordNotFound" }, 400);
		}
	}
	if (url.pathname === "/xrpc/com.atproto.repo.putRecord" || url.pathname === "/xrpc/com.atproto.repo.deleteRecord") {
		const { repo, collection, rkey, record } = await req.json();
		const path = recordPath(repo, collection, rkey);
		const create = url.pathname.endsWith("putRecord");
		if (create) await write(path, JSON.stringify(record));
		else await Deno.remove(path).catch(() => {});
		const event = { did: repo, time_us: Date.now() * 1000, kind: "commit", commit: { operation: create ? "create" : "delete", collection, rkey, record } };
		for (const socket of sockets) socket.send(JSON.stringify(event));
		return json({});
	}
	if (url.pathname.startsWith("/step/")) {
		const step = steps[url.pathname.slice(6)];
		if (!step) return new Response("no such step", { status: 404 });
		try {
			await step();
			return json({ ok: true });
		} catch (error) {
			return new Response(String(error), { status: 500 });
		}
	}
	if (url.pathname === "/log") {
		console.error(`page: ${await req.text()}`);
		return json({});
	}
	if (url.pathname === "/result") {
		resolveResult(await req.json());
		return json({});
	}
	if (url.pathname.includes("..") || !/^\/(js|node_modules|test\/web)\//.test(url.pathname)) return new Response("not found", { status: 404 });
	try {
		const type = url.pathname.endsWith(".html") ? "text/html" : "text/javascript";
		return new Response(await Deno.readFile(`${ROOT}${url.pathname.slice(1)}`), { headers: { "content-type": type } });
	} catch {
		return new Response("not found", { status: 404 });
	}
});
const base = `http://127.0.0.1:${server.addr.port}`;

const env = (name: string, handle: string, nick?: string) => ({
	ATIRC_CONFIG_DIR: `${tmp}/${name}`,
	ATIRC_MESH_DIRECTORY: dir,
	ATIRC_IROH_LOOPBACK: "1",
	ATIRC_JETSTREAM: `ws://127.0.0.1:${server.addr.port}/subscribe`,
	ATIRC_RELAY: "off",
	ATP_USERNAME: handle,
	...(nick ? { ATIRC_NICK: nick } : {}),
	RUST_LOG: "info",
});
const children: Deno.ChildProcess[] = [];
const start = (args: string[], envs: Record<string, string>, log: string) => {
	const child = new Deno.Command(args[0], { args: args.slice(1), env: envs, stdin: "null", stdout: "piped", stderr: "piped" }).spawn();
	children.push(child);
	const file = Deno.openSync(`${tmp}/${log}`, { write: true, create: true, truncate: true });
	(async () => {
		for await (const chunk of child.stderr) file.writeSync(chunk);
	})().catch(() => {});
	return child;
};
const run = async (args: string[], envs: Record<string, string>) => {
	const out = await new Deno.Command(BIN, { args, env: envs, stdout: "piped", stderr: "piped" }).output();
	if (!out.success) throw new Error(`atirc ${args.join(" ")}: ${new TextDecoder().decode(out.stderr)}`);
	return new TextDecoder().decode(out.stdout);
};
const until = async (check: () => Promise<boolean> | boolean, what: string, ms = 30000) => {
	for (const end = Date.now() + ms; Date.now() < end; await new Promise((r) => setTimeout(r, 100))) if (await check()) return;
	throw new Error(`timed out waiting for ${what}`);
};
const exists = (path: string) => Deno.stat(path).then(() => true, () => false);
// `atirc mesh sub --cursor 0` of a node, parsed.
const subscribe = (envs: Record<string, string>, log: string) => {
	const lines: Record<string, string>[] = [];
	const sub = start([BIN, "mesh", "sub", "--cursor", "0"], envs, log);
	(async () => {
		let buffer = "";
		for await (const chunk of sub.stdout.pipeThrough(new TextDecoderStream())) {
			buffer += chunk;
			const parts = buffer.split("\n");
			buffer = parts.pop() ?? "";
			for (const part of parts) if (part.trim()) lines.push(JSON.parse(part));
		}
	})().catch(() => {});
	return (channel: string, identity: string, text: string) => lines.some((l) => l.channel === channel && l.identity === identity && l.text === text);
};

let failed = false;
try {
	const alice = env("alice", OWNER), vm = env("vm", OWNER, "vm"), bot = env("bot", AGENT, "bot");
	// Loopback tickets carry the port: the others start once alice has republished hers.
	start([BIN, "mesh", "run"], alice, "alice.log");
	await until(() => exists(`${tmp}/alice/mesh.json`), "the alice node");
	start([BIN, "mesh", "run"], bot, "bot.log");
	// What a requester writes for a VM agent: its key's record, the owner's channel only.
	await run(["peer", "publish", "-c", "#alice_test"], vm);
	const local = Deno.listen({ hostname: "127.0.0.1", port: 0 });
	const localPort = (local.addr as Deno.NetAddr).port;
	local.close();
	start([BIN, "mesh", "run", "--no-publish", "--no-webrtc", "--local-listen", `127.0.0.1:${localPort}`], vm, "vm.log");
	// The guest agent's stand-in: datachannel tag atirc -> the vm node's local listener.
	const peer = start([WEBRTC_PEER, "--tunnel", `atirc=${localPort}`, "--origin", base], {}, "webrtc-peer.log");
	const signal = await (async () => {
		let out = "";
		for await (const chunk of peer.stdout.pipeThrough(new TextDecoderStream())) {
			out += chunk;
			const line = out.split("\n").find((l) => l.startsWith("{"));
			if (line) return JSON.parse(line).signal as string;
		}
		throw new Error("webrtc-peer printed no signal");
	})();
	for (const name of ["alice", "bot", "vm"]) await until(() => exists(`${tmp}/${name}/mesh.json`), `the ${name} node`);
	await until(() => sockets.size >= 2, "the nodes' Jetstream connections");
	const vmHeard = subscribe(vm, "vm-sub.log"), botHeard = subscribe(bot, "bot-sub.log");
	await until(async () => /vm\.alice\.test/.test(await run(["mesh", "peers"], alice)), "alice linked to the vm");
	await run(["mesh", "pub", "vm online"], vm);

	steps["vm-heard"] = () => until(() => vmHeard("#alice_test", OWNER, "hi from the page"), "the page's line on the vm");
	steps["vm-own"] = async () => void (await run(["mesh", "pub", "-c", "#vm_alice_test", "vm own channel"], vm));
	steps["vm-own-heard"] = () => until(() => vmHeard("#vm_alice_test", OWNER, "page on the vm channel"), "the page's line on the vm's channel");
	steps["bot-say"] = async () => void (await run(["mesh", "pub", "bot says hi"], bot));
	steps["bot-heard"] = () => until(() => botHeard("#alice_test", "web2.alice.test", "page direct to bot"), "the direct page's line on the bot");
	steps["bot-to-guest"] = async () => void (await run(["mesh", "pub", "bot to the guest"], bot));
	steps["guest-heard"] = () => until(() => botHeard("#alice_test", "vm.alice.test", "from the guest page"), "the guest page's line on the bot, as the vm node");

	const page = `${base}/test/web/loopback.html?${new URLSearchParams({ did: did(OWNER), agent: did(AGENT), signal })}`;
	start([CHROME, "--headless=new", `--user-data-dir=${tmp}/chrome`, "--no-first-run", "--no-default-browser-check", "--disable-gpu",
		"--allow-loopback-in-peer-connection", "--disable-features=WebRtcHideLocalIpsWithMdns", page], {}, "chrome.log");
	const outcome = await Promise.race([result, new Promise((_, reject) => setTimeout(() => reject(new Error("the page reported nothing in 240 s")), 240000))]) as
		{ ok: boolean; error?: string; peersA?: string[] };
	if (!outcome.ok) throw new Error(`page: ${outcome.error}`);
	console.log(`ok atlibs-atirc loopback: bridge (alice.test -> vm.alice.test without WebRTC, both channels) and direct (bot.agent.test), both ways; guest (vm.alice.test --local-listen via webrtc-peer) both ways; A's peers ${outcome.peersA?.join(",")}`);
} catch (error) {
	failed = true;
	console.error(`FAIL atlibs-atirc loopback: ${error instanceof Error ? error.message : error}`);
	for (const log of ["alice.log", "vm.log", "bot.log", "webrtc-peer.log", "chrome.log"]) console.error(`--- ${log}\n${(await Deno.readTextFile(`${tmp}/${log}`).catch(() => "")).slice(-6000)}`);
} finally {
	for (const child of children) {
		try {
			child.kill("SIGTERM");
		} catch { /* exited */ }
	}
	await server.shutdown();
	if (Deno.env.get("KEEP")) console.error(`kept ${tmp}`);
	else await Deno.remove(tmp, { recursive: true }).catch(() => {});
}
Deno.exit(failed ? 1 : 0);