/** @file toybox end to end: a real workerd, an admin socket, a public port, a switchboard. */
import { spawn } from "node:child_process";
import { mkdir, mkdtemp, readFile, rm } from "node:fs/promises";
import { createServer as createHttpServer, request } from "node:http";
import { connect, createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { after, before, test } from "node:test";
import assert from "node:assert/strict";
import { WebSocketServer } from "ws";

const here = new URL("..", import.meta.url).pathname;
// the package binary, not the .bin shim: the shim forks it, so a kill would orphan workerd
const workerd = join(here, "node_modules/workerd/bin/workerd");
const app = (name) => readFile(join(here, "apps", name), "utf8");

const freePort = () =>
	new Promise((resolve) => {
		const s = createServer().listen(0, "127.0.0.1", () => {
			const { port } = s.address();
			s.close(() => resolve(port));
		});
	});

/** Resolves once something accepts connections at `opts` (a port or a socket path). */
const accepting = (opts) =>
	new Promise((resolve) => {
		const s = connect(opts, () => (s.end(), resolve(true)));
		s.on("error", () => resolve(false));
	});

/** @param {string} socketPath the admin socket; holding it is the authority to install */
const makeAdmin = (socketPath) => ({
	install: (vat, src, egress) =>
		new Promise((resolve, reject) => {
			const path = egress ? `/${vat}?egress=${encodeURIComponent(egress)}` : `/${vat}`;
			const req = request({ socketPath, path, method: "PUT", agent: false }, (res) => {
				let body = "";
				res.on("data", (c) => (body += c));
				res.on("end", () => resolve(body.trim()));
			});
			req.on("error", reject);
			req.end(src);
		}),
});

/** @param {string} base the public origin */
const makePublic = (base) => ({
	get: async (path) => (await fetch(`${base}/${path}`)).text(),
	put: async (vat, src) => (await fetch(`${base}/${vat}`, { method: "PUT", body: src })).status,
});

/**
 * Run workerd on `config` until `stop()`; resolves once `listening` accepts.
 *
 * @param {string} config
 * @param {string[]} args socket, directory, and external overrides
 * @param {{ port: number } | { path: string }} listening
 * @param {string[]} stale socket files a crash leaves behind; bind() would fail on them
 */
const startWorkerd = async (config, args, listening, stale = []) => {
	for (const f of stale) await rm(f, { force: true });
	const child = spawn(workerd, ["serve", "--experimental", join(here, config), ...args], {
		stdio: ["ignore", "ignore", "pipe"],
	});
	let log = "";
	child.stderr.on("data", (c) => (log += c));
	const exited = new Promise((resolve) => child.on("exit", resolve));
	for (let i = 0; i < 100; i++) {
		// SIGKILL: a crash is the harder durability test
		if (await accepting(listening)) return { stop: () => (child.kill("SIGKILL"), exited) };
		await new Promise((r) => setTimeout(r, 50));
	}
	child.kill("SIGKILL");
	throw new Error(`workerd did not start: ${log}`);
};

/** An HTTP server that answers GET with "plain ok" and echoes WebSocket messages. */
const startEcho = async () => {
	const server = createHttpServer((_req, res) => res.end("plain ok"));
	const wss = new WebSocketServer({ server });
	wss.on("connection", (ws) => ws.on("message", (m) => ws.send(String(m))));
	await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
	return {
		host: `127.0.0.1:${server.address().port}`,
		stop: () => new Promise((resolve) => (wss.close(), server.closeAllConnections(), server.close(resolve))),
	};
};

let dir, port, server, switchboard, echo, admin, pub;

const startToybox = () =>
	startWorkerd(
		"config.capnp",
		[
			"-s", `http=127.0.0.1:${port}`, "-s", `admin=unix:${join(dir, "admin.sock")}`,
			"-d", `state=${join(dir, "state")}`, "-e", `switchboard=unix:${join(dir, "switchboard.sock")}`,
		],
		{ port, host: "127.0.0.1" },
		[join(dir, "admin.sock")],
	);

before(async () => {
	dir = await mkdtemp(join(tmpdir(), "toybox-"));
	await mkdir(join(dir, "state"));
	port = await freePort();
	echo = await startEcho();
	switchboard = await startWorkerd(
		"switchboard.capnp",
		["-s", `http=unix:${join(dir, "switchboard.sock")}`],
		{ path: join(dir, "switchboard.sock") },
		[join(dir, "switchboard.sock")],
	);
	server = await startToybox();
	admin = makeAdmin(join(dir, "admin.sock"));
	pub = makePublic(`http://127.0.0.1:${port}`);
});

after(async () => {
	await server?.stop();
	await switchboard?.stop();
	await echo?.stop();
	await rm(dir, { recursive: true, force: true });
});

test("a vat with no code is 404", async () => {
	assert.equal(await pub.get("nobody"), "no code installed\n");
});

test("install, run, upgrade: the app's state carries over", async () => {
	assert.match(await admin.install("alice", await app("counter-v1.js")), /^[0-9a-f]{64}$/);
	assert.equal(await pub.get("alice"), "v1 count 1\n");
	assert.equal(await pub.get("alice"), "v1 count 2\n");
	await admin.install("alice", await app("counter-v2.js"));
	assert.equal(await pub.get("alice"), "v2 count 3\n");
});

test("vats don't share state", async () => {
	await admin.install("carol", await app("counter-v1.js"));
	assert.equal(await pub.get("carol"), "v1 count 1\n");
});

test("an app gets no fetch", async () => {
	await admin.install("bob", await app("try-fetch.js"));
	assert.match(await pub.get("bob"), /^fetch refused: This worker is not permitted to access the internet/);
});

test("the public port can't install code", async () => {
	assert.equal(await pub.put("mallory", "export class App {}"), 404);
	assert.equal(await pub.get("mallory"), "no code installed\n");
});

test("code and state survive a restart", async () => {
	await admin.install("dave", await app("counter-v1.js"));
	assert.equal(await pub.get("dave"), "v1 count 1\n");
	await server.stop();
	server = await startToybox();
	assert.equal(await pub.get("dave"), "v1 count 2\n");
});

const probe = (vat, target, ws) =>
	pub.get(`${vat}?url=${encodeURIComponent(`http://${target}/`)}${ws ? "&ws=1" : ""}`);

test("a vat granted a host reaches it through the switchboard: fetch and WebSocket", async () => {
	await admin.install("erin", await app("egress-probe.js"), echo.host);
	assert.equal(await probe("erin", echo.host), "got: plain ok\n");
	assert.equal(await probe("erin", echo.host, true), "echo: hello\n");
});

test("same code, no grant: no fetch", async () => {
	await admin.install("frank", await app("egress-probe.js"));
	assert.match(await probe("frank", echo.host), /^refused: This worker is not permitted to access the internet/);
});

test("a grant covers one host only", async () => {
	await admin.install("grace", await app("egress-probe.js"), "127.0.0.1:1");
	assert.equal(await probe("grace", echo.host), `got: egress to ${echo.host} not granted\n\n`);
	assert.match(await probe("grace", echo.host, true), /^no websocket: 403 egress to .* not granted/);
});

test("a grant survives a restart", async () => {
	await server.stop();
	server = await startToybox();
	assert.equal(await probe("erin", echo.host, true), "echo: hello\n");
});