// The Rust run-module engine (rust/host, `atlibs bridge`) behind runModule's JS
// signatures: the same ctx (wasm.run, storage.managed_records), call stubs (signal, subscriptions,
// send, close, emit), inputs, budget and signal go over stdio to the Rust reactor, so a test can
// run one scenario through both hosts.
import { execFile, execFileSync, spawn } from "node:child_process";
import { existsSync, mkdtempSync } from "node:fs";
import { copyFile, mkdir, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { promisify } from "node:util";
import { createInterface } from "node:readline";

import { moduleOwner, withHooks } from "../../javascript/src/run-module/index.js?real";

const ROOT = new URL("../../", import.meta.url).pathname;
const messageOf = (error) => (error instanceof Error ? error.message : String(error));

/** The atlibs binary: ATLIBS_BIN, else rust/host's release or debug build (`cargo host`). */
let bin;
export function hostBin() {
	if (process.env.ATLIBS_BIN) return process.env.ATLIBS_BIN;
	if (bin) return bin;
	const triple = execFileSync("rustc", ["-vV"], { encoding: "utf8" }).match(/^host: (.+)$/m)[1];
	const found = ["release", "debug"].map((p) => `${ROOT}rust/host/target/${triple}/${p}/atlibs`).find(existsSync);
	if (!found) throw new Error("atlibs not built: cargo host");
	return (bin = found);
}

const STATE = mkdtempSync(join(tmpdir(), "atlibs-state-"));
/**
 * Runs the atlibs CLI from the repo root with a clean env: no ATPROTO_* (tests never sign in as
 * whoever runs them), a throwaway state dir unless `env` names one. Resolves { stdout, stderr }.
 */
export const atlibs = (args, { cwd = ROOT, env = {} } = {}) =>
	promisify(execFile)(hostBin(), args, { cwd, env: { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: STATE, ...env } });

const closeStream = (stream, iterator) => {
	try {
		if (typeof stream?.close === "function") stream.close();
		else Promise.resolve(iterator?.return?.()).catch(() => {});
	} catch {}
};

/**
 * runModule on the Rust engine. Extra options (not in the JS host): wasm, a .wasm path the Rust host runs
 * itself on wasmtime (else ctx.wasm.run is used over the bridge); now, a fixed `now`; didkey { jwk,
 * services } with fetch, the Rust did:key pseudo-calls in front of `call`.
 */
export async function runModuleRust(ctx, mod, entry, input, call, { signal, inputs, budget, onEvent, wasm, now, didkey, fetch, load, callFor, provides, http, fetchers, held, session } = {}) {
	// Hooks and their defaults are the JS host's (run-module withHooks); a default runs on this engine too.
	const hooked = withHooks(ctx, mod, call, { entry, load, callFor, budget, provides, http, fetchers, run: runModuleRust });
	if (hooked.error) return { entry, result: null, error: hooked.error, events: 0, calls: [] };
	({ mod, call } = hooked);
	const owner = await moduleOwner(mod);
	const child = spawn(hostBin(), ["bridge"], { stdio: ["pipe", "pipe", "inherit"] });
	const trace = process.env.ATLIBS_BRIDGE_TRACE ? (dir, m) => console.error(performance.now().toFixed(1), dir, JSON.stringify(m).slice(0, 160)) : () => {};
	const write = (message) => (trace(">", message), child.stdin.writable && child.stdin.write(`${JSON.stringify(message)}\n`));
	const controllers = new Map();
	const streams = new Map();
	const ledger = ctx.storage?.managed_records;
	write({
		module: { wasm, calls: mod.calls, anyRecords: mod.anyRecords, http: mod.http, sha256: mod.sha256, owner },
		entry,
		input,
		// JSON has no Infinity: the Rust host reads an unlimited budget value as "unlimited".
		budget: budget && Object.fromEntries(Object.entries(budget).map(([key, value]) => [key, value === Infinity ? "unlimited" : value])),
		inputs: !!inputs,
		ledger: !!ledger,
		now,
		didkey,
		onEvent: !!onEvent,
		http,
		// the schemes; each request comes back as a fetch with its scheme
		fetchers: fetchers && Object.keys(fetchers),
		held,
		// http serviceAuth mints through it: each such call comes back with session: true
		session: !!session,
	});
	if (inputs) {
		(async () => {
			try {
				for await (const data of inputs) write({ t: "input", data });
				write({ t: "inputs-end" });
			} catch (error) {
				write({ t: "inputs-end", error: messageOf(error) });
			}
		})();
	}
	const onAbort = () => write({ t: "abort" });
	signal?.addEventListener("abort", onAbort, { once: true });
	if (signal?.aborted) onAbort();

	const handlers = {
		run: async ({ entry, input, ioBytes }) => ({ ok: true, output: await ctx.wasm.run({ module: mod.module, sha256: mod.sha256, entry, input, ioBytes }) }),
		ledger: async ({ op, uri, value }) => ({ value: op === "get" ? await ledger.get(uri) : op === "put" ? await ledger.put(uri, value) : await ledger.delete(uri) }),
		// didkey's (the fetch option) and the http pseudo-call's (ctx.http.fetch), which follows no redirects
		fetch: async ({ scheme, url, method, headers, body }) => {
			if (scheme) {
				const response = await fetchers[scheme]({ method, url, headers, ...(body != null && { body }), signal: new AbortController().signal });
				return { status: response.status, headers: Object.fromEntries(new Headers(response.headers ?? {})), body: String(response.body ?? "") };
			}
			const response = await (fetch ?? ctx.http.fetch)(url, { method, headers, ...(body != null && { body }), redirect: "manual" });
			return { status: response.status, headers: Object.fromEntries(response.headers), body: await response.text() };
		},
		send: async ({ sub, data }) => {
			const stream = streams.get(sub)?.stream;
			if (typeof stream?.send !== "function") return { nosend: true };
			await stream.send(data);
			return { ok: true };
		},
		call: async ({ k, nsid, request, session: viaSession }) => {
			const controller = new AbortController();
			controllers.set(k, controller);
			// The request exactly as the JS host builds it (undefined fields present).
			const { params, input, encoding, proxy, service, endpoints, subscribe, framing, keepalive, serviceAuth } = request;
			const req = { params, input, ...(encoding && { encoding }), proxy, ...(service && { service }), ...(endpoints && { endpoints }), signal: controller.signal, ...(subscribe && { subscribe: true, framing, keepalive, ...(serviceAuth && { serviceAuth }) }), emit: (event) => write({ t: "emit", k, event }) };
			if (!request.subscribe) {
				try {
					return { ok: true, output: await (viaSession ? session : call)(nsid, req) };
				} finally {
					controllers.delete(k);
				}
			}
			const stream = await call(nsid, req);
			const iterator = (stream[Symbol.asyncIterator] ?? stream[Symbol.iterator]).call(stream);
			streams.set(k, { stream, iterator });
			write({ k, open: true });
			(async () => {
				try {
					for (let next = await iterator.next(); !next.done; next = await iterator.next()) write({ k, message: next.value });
					write({ k, end: true });
				} catch (error) {
					write({ k, end: true, error: messageOf(error) });
				}
			})();
			return undefined;
		},
	};

	return new Promise((resolve, reject) => {
		const lines = createInterface({ input: child.stdout });
		lines.on("line", (line) => {
			const message = JSON.parse(line);
			trace("<", message);
			if (message.t === "done") {
				signal?.removeEventListener("abort", onAbort);
				for (const { stream, iterator } of streams.values()) closeStream(stream, iterator);
				child.stdin.end();
				return message.error !== undefined ? reject(new Error(message.error)) : resolve(message.ran);
			}
			if (message.t === "event") return onEvent(message.event, message.out);
			if (message.t === "abort") return controllers.get(message.k)?.abort(new Error(message.reason));
			if (message.t === "close") {
				const open = streams.get(message.sub);
				streams.delete(message.sub);
				return closeStream(open?.stream, open?.iterator);
			}
			const k = message.k;
			handlers[message.t](message).then(
				(reply) => reply && write({ k, ...reply }),
				(error) => write({ k, ok: false, error: messageOf(error) }),
			);
		});
		child.on("error", reject);
		child.on("exit", (code) => code && reject(new Error(`atlibs bridge exited ${code}`)));
	});
}

// As javascript/src/run-module/index.js with runModule on the Rust engine: swap.mjs
// points every import of run-module here, so the existing JS suites assert the Rust host's behaviour.
export * from "../../javascript/src/run-module/index.js?real";
export { runModuleRust as runModule };

/** compose rfpPolicyRoutes on the Rust host (atlibs_host::compose::rfp_policy_routes): the routed manifest, or throws its error. */
export async function rfpPolicyRoutesRust(manifest, from, records) {
	const child = spawn(hostBin(), ["bridge"], { stdio: ["pipe", "pipe", "inherit"] });
	child.stdin.end(`${JSON.stringify({ rfpPolicyRoutes: { manifest, from, records } })}\n`);
	let out = "";
	for await (const chunk of child.stdout) out += chunk;
	const done = JSON.parse(out.trim().split("\n").at(-1));
	if (done.error) throw new Error(done.error);
	return done.manifest;
}

/**
 * Modules served under one sign-in, for the per-module scope tests: a temp
 * dir of records/<session>.json, modules/<key>.{wasm,json} and comp.json (instances a and b). a, b and c are the
 * compose toy, whose caller calls any NSID: a's session sa may create com.example.a records, call
 * com.example.a.ping and upload blobs; b's sb may create com.example.b records; c has no session. chat and evil
 * call com.example.chat.post through atproto-proxy did:web:chat.example#chat (caller's `proxy`): chat's session
 * allows that service, evil's only another one. h (session sh: rpc:x.y.z?aud=*) calls com.fedfork.exec.http.fetch,
 * which mints the host's http serviceAuth (lxm x.y.z) with its instance's session.
 */
export async function scopeFixture() {
	const dir = mkdtempSync(join(tmpdir(), "atlibs-scope-"));
	await mkdir(join(dir, "records"));
	await mkdir(join(dir, "modules"));
	const record = (name, scope) => writeFile(join(dir, "records", `${name}.json`), JSON.stringify({ $type: "com.fedfork.exec.oauth.clientMetadata", name, clientMetadata: { client_id: "https://tpl.example/c.json", scope }, createdAt: "2026-11-08T00:00:00.000Z" }));
	await record("sa", "atproto repo:com.example.a?action=create rpc:com.example.a.ping?aud=* blob:*/*");
	await record("sb", "atproto repo:com.example.b?action=create");
	await record("chat", "atproto rpc:com.example.chat.post?aud=did:web:chat.example%23chat");
	await record("evil", "atproto rpc:com.example.chat.post?aud=did:web:evil.example%23chat");
	await record("sh", "atproto rpc:x.y.z?aud=*");
	const calls = ["com.atproto.repo.createRecord", "com.atproto.repo.uploadBlob", "com.atproto.repo.listRecords", "com.example.a.ping"];
	const chat = { calls: ["com.example.chat.post"] };
	const modules = { a: ["compose", { calls, session: "sa" }], b: ["compose", { calls, session: "sb" }], c: ["compose", { calls }], chat: ["compose", { ...chat, session: "chat" }], evil: ["compose", { ...chat, session: "evil" }], h: ["compose", { calls: ["com.fedfork.exec.http.fetch"], http: ["*"], session: "sh" }] };
	for (const [key, [wasm, spec]] of Object.entries(modules)) {
		await copyFile(`${ROOT}target/wasm32-unknown-unknown/release/${wasm}.wasm`, join(dir, "modules", `${key}.wasm`));
		await writeFile(join(dir, "modules", `${key}.json`), JSON.stringify({ name: key, ...spec }));
	}
	await writeFile(join(dir, "comp.json"), JSON.stringify({ $type: "com.fedfork.exec.composition", name: "scopes", instances: [{ name: "a", module: "a" }, { name: "b", module: "b" }, { name: "h", module: "h" }], edges: [] }));
	return dir;
}