// node --test javascript/src/compose/compose.test.mjs   (needs `cargo wasm` and `cargo host`; no network)
import assert from "node:assert/strict";
import { mkdtemp, readFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { test } from "node:test";
import { pathToFileURL } from "node:url";

import { atlibs } from "../../../rust/host/conformance.mjs";
import { loadComposition, runCompose, staged, validateComposition } from "./index.js";

const wasm = new Uint8Array(await readFile(new URL("../../../target/wasm32-unknown-unknown/release/compose.wasm", import.meta.url)));
const spec = JSON.parse(await readFile(new URL("../../../examples/rust/compose/module.json", import.meta.url), "utf8"));
const LOOKUP = "com.example.compose.lookup";

// ctx.wasm.run for any module bytes, in process.
const instances = new Map();
const ctx = {
	wasm: {
		run: async ({ module, entry, input }) => {
			if (!instances.has(module)) instances.set(module, new WebAssembly.Instance(new WebAssembly.Module(module), {}).exports);
			const exports = instances.get(module);
			const data = new TextEncoder().encode(JSON.stringify(input));
			const ptr = exports.alloc(data.length);
			new Uint8Array(exports.memory.buffer, ptr, data.length).set(data);
			const packed = BigInt.asUintN(64, exports[entry](ptr, data.length));
			return JSON.parse(new TextDecoder().decode(new Uint8Array(exports.memory.buffer, Number(packed >> 32n), Number(packed & 0xffffffffn))));
		},
	},
};
// The depth and cycle tests route x.y.a, x.y.b, x.y.c, x.y.never between instances: hooks of the module.
const hooks = [...spec.hooks, ...["x.y.a", "x.y.b", "x.y.c", "x.y.never"].map((nsid) => ({ nsid, reason: "a test route" }))];
// ...and serve them at caller
const implementsAll = [...spec.implements, ...["x.y.a", "x.y.b", "x.y.c", "x.y.never"].map((nsid) => ({ nsid, entry: "caller" }))];
const load = async (key) => ({ ...spec, hooks, implements: implementsAll, module: wasm, ...(key === "compose-net" && { calls: ["com.example.compose.net"] }), ...(key === "compose-http" && { calls: ["com.fedfork.exec.http.fetch"], http: ["https://svc.example"] }) });
const compose = (manifest) => loadComposition(ctx, { name: "test", ...manifest }, { load });
const toy = (name, extra) => ({ name, module: "compose", ...extra });
const edge = (from, nsid, to, extra) => ({ from, nsid, to, reason: `${from} asks ${to}`, ...extra });

test("A calls B over an edge and B calls A's callback; the answer comes back", async () => {
	const comp = await compose({ instances: [toy("a"), toy("b")], edges: [edge("a", LOOKUP, "b")] });
	const ran = await runCompose(ctx, comp, "a", "caller", { nsid: LOOKUP, input: { n: 21 } });
	assert.equal(ran.error, undefined);
	assert.deepEqual(ran.result, {
		q: { ok: true, output: { doubled: 42, ack: { seen: { step: 1 } }, caller: { instance: "a", nsid: LOOKUP, reason: "a asks b" } } },
		progress: [{ step: 1 }],
	});
	assert.deepEqual(ran.graph.map(({ from, nsid, to, ok }) => [from, nsid, to, ok]), [
		["a", "cap", undefined, true], // the reply
		["b", "cap", undefined, true], // the progress callback
		["a", LOOKUP, "b", true],
	]);
});

test("a caller handler gets the call, and the module's callback as a function", async () => {
	let late;
	const comp = await compose({
		instances: [toy("a")],
		handlers: [{ name: "confirm" }],
		edges: [{ from: "a", nsid: "com.example.compose.confirm", handler: "confirm", reason: "the user approves" }],
	});
	const seen = [];
	const ran = await runCompose(ctx, comp, "a", "caller", { nsid: "com.example.compose.confirm", input: { n: 1 } }, {
		handlers: {
			confirm: async ({ n, progress }, { caller, reason }) => {
				seen.push([n, caller, reason]);
				late = progress;
				return { approved: true, ack: await progress({ asked: n }) };
			},
		},
	});
	assert.deepEqual(seen, [[1, "a", "the user approves"]]);
	assert.deepEqual(ran.result, { q: { ok: true, output: { approved: true, ack: { seen: { asked: 1 } } } }, progress: [{ asked: 1 }] });
	await assert.rejects(late({}), /capability revoked/);
});

test("JS functions in the root input are callbacks the root may call; forged tokens are refused", async () => {
	const comp = await compose({ instances: [toy("b")] });
	const ran = await runCompose(ctx, comp, "b", "lookup", { n: 2, progress: async (data) => ({ host: data }) });
	assert.deepEqual(ran.result, { doubled: 4, ack: { host: { step: 1 } }, caller: null });
	const forged = await runCompose(ctx, comp, "b", "lookup", { n: 2, progress: { $cap: `cap.${"0".repeat(32)}.1` } });
	assert.match(forged.result.ack.error, /is not a capability this run holds/);
});

test("a policy module allows or denies the call before it is made", async () => {
	const comp = await compose({
		instances: [toy("a"), toy("b"), toy("even")],
		edges: [edge("a", LOOKUP, "b", { policies: [{ instance: "even", entry: "even" }] })],
	});
	const denied = await runCompose(ctx, comp, "a", "caller", { nsid: LOOKUP, input: { n: 21 } });
	assert.deepEqual(denied.result.q, { ok: false, error: "policy even denied: n 21 is not even" });
	assert.ok(!denied.graph.some((g) => g.to === "b" && g.ok));
	const allowed = await runCompose(ctx, comp, "a", "caller", { nsid: LOOKUP, input: { n: 4 } });
	assert.equal(allowed.result.q.output.doubled, 8);
	assert.deepEqual(allowed.graph.find((g) => g.policy), { from: "a", nsid: LOOKUP, policy: "even", ok: true });
});

test("a policy run does not hold the caller's capabilities", async () => {
	const comp = await compose({
		instances: [toy("a"), toy("b"), toy("spy")],
		edges: [edge("a", LOOKUP, "b", { policies: [{ instance: "spy", entry: "spy" }] })],
	});
	const invoked = [];
	const ran = await runCompose(ctx, comp, "a", "caller", { nsid: LOOKUP, input: { n: 4, host: async (data) => (invoked.push(data), "secret") } });
	assert.deepEqual(invoked, [], "the policy could not invoke the root's host function");
	assert.equal(ran.result.q.output.doubled, 8);
	assert.deepEqual(ran.graph.find((g) => g.policy), { from: "a", nsid: LOOKUP, policy: "spy", ok: true });
	assert.ok(ran.graph.some((g) => g.from === "spy" && g.nsid === "cap" && g.error === "not held"));
});

test("the host's budget and limits cap the manifest's", async () => {
	// limits: the smaller of the host's and the manifest's, whichever sets it lower
	const deep = await compose({
		instances: [toy("a"), toy("b"), toy("c")],
		edges: [edge("a", "x.y.b", "b"), edge("b", "x.y.c", "c")],
		limits: { depth: 50 },
	});
	const capped = await runCompose(ctx, deep, "a", "caller", { nsid: "x.y.b", input: { nsid: "x.y.c", input: { nsid: "x.y.d" } } }, { limits: { depth: 2 } });
	assert.deepEqual(capped.result.q.output.q, { ok: false, error: "composition exceeded depth 2" });
	const lowered = await runCompose(ctx, { ...deep, manifest: { ...deep.manifest, limits: { depth: 2 } } }, "a", "caller", { nsid: "x.y.b", input: { nsid: "x.y.c", input: { nsid: "x.y.d" } } }, { limits: { depth: 50 } });
	assert.deepEqual(lowered.result.q.output.q, { ok: false, error: "composition exceeded depth 2" });
	const slow = await compose({
		instances: [toy("w", { budget: { wallMs: 1e9 } })],
		handlers: [{ name: "hang" }],
		edges: [{ from: "w", nsid: "x.y.never", handler: "hang", reason: "never answers" }],
	});
	const timed = await runCompose(ctx, slow, "w", "caller", { nsid: "x.y.never" }, { budget: { wallMs: 50 }, handlers: { hang: () => new Promise(() => {}) } });
	assert.match(String(timed.error), /exceeded 50 ms/, "a manifest budget cannot raise the host's wallMs");
	// only positive finite numbers count: a host value that is not one leaves the manifest's (or the default)
	const own = await compose({ ...slow.manifest, instances: [toy("w", { budget: { wallMs: 50 } })] });
	const ownTimed = await runCompose(ctx, own, "w", "caller", { nsid: "x.y.never" }, { budget: { wallMs: "lots", calls: -1 }, handlers: { hang: () => new Promise(() => {}) } });
	assert.match(String(ownTimed.error), /exceeded 50 ms/, "with no usable host value the manifest's own applies");
	for (const budget of [{ calls: "1" }, { calls: 0 }, { wallMs: Infinity }, { wallMs: "unlimited" }, { nope: 1 }, [1]]) {
		assert.throws(() => validateComposition({ instances: [toy("a", { budget })] }), /budget must map/, JSON.stringify(budget));
	}
	// unlimited is the host's alone: a manifest cannot say it nor raise past a finite host limit, and under an
	// unlimited host its own lower limit still holds
	for (const limits of [{ calls: Infinity }, { calls: "unlimited" }]) {
		assert.throws(() => validateComposition({ instances: [toy("a")], limits }), /limits must map/, String(limits.calls));
	}
	const chain = { nsid: "x.y.b", input: { nsid: "x.y.c", input: { nsid: "x.y.d" } } };
	const raised = await runCompose(ctx, { ...deep, manifest: { ...deep.manifest, limits: { calls: 50 } } }, "a", "caller", chain, { limits: { calls: 1 } });
	assert.match(raised.result.q.output.q.error, /composition exceeded 1 calls/);
	const under = await runCompose(ctx, { ...deep, manifest: { ...deep.manifest, limits: { calls: 1 } } }, "a", "caller", chain, { limits: { calls: Infinity } });
	assert.match(under.result.q.output.q.error, /composition exceeded 1 calls/);
	const free = await runCompose(ctx, deep, "a", "caller", chain, { limits: { calls: Infinity }, budget: { calls: Infinity, wallMs: Infinity } });
	assert.equal(free.error, undefined);
	assert.doesNotMatch(JSON.stringify(free.result), /exceeded/);
	assert.equal(free.result.q.output.q.output.q.ok, false, "c's call has no edge");
});

test("calls with no edge are refused unless allowlisted; allowlisted ones use the instance's session and policies", async () => {
	const comp = await compose({
		instances: [toy("a"), toy("n", { module: "compose-net", session: "net", policies: [{ instance: "even", entry: "even" }] }), toy("even")],
	});
	const refused = await runCompose(ctx, comp, "a", "caller", { nsid: "com.example.compose.net", input: { n: 2 } });
	assert.match(refused.result.q.error, /a may not call com.example.compose.net: no edge/);
	const sessions = [];
	const callFor = (inst) => async (nsid, request) => (sessions.push([inst.session, nsid, request.input.n]), { via: inst.session });
	const made = await runCompose(ctx, comp, "n", "caller", { nsid: "com.example.compose.net", input: { n: 2 } }, { callFor });
	assert.deepEqual(made.result.q, { ok: true, output: { via: "net" } });
	assert.deepEqual(sessions, [["net", "com.example.compose.net", 2]]);
	const odd = await runCompose(ctx, comp, "n", "caller", { nsid: "com.example.compose.net", input: { n: 3 } }, { callFor });
	assert.match(odd.result.q.error, /policy even denied/);
	assert.equal(sessions.length, 1);
});

test("the host's http serviceAuth is minted with the instance's session (callFor), not a call the module makes", async () => {
	const comp = await compose({ instances: [toy("h", { module: "compose-http", session: "s" })] });
	const sent = [];
	const httpCtx = { ...ctx, http: { fetch: async (url, init) => (sent.push([url, init.headers.authorization]), new Response("ok")) } };
	const http = { "https://svc.example": { serviceAuth: { aud: "did:web:svc.example", lxm: "x.y.z" } } };
	const fetch = { nsid: "com.fedfork.exec.http.fetch", input: { url: "https://svc.example/x" } };
	const minted = [];
	const callFor = (inst) => async (nsid, request) => (minted.push([inst.session, nsid, request.params]), { token: "T" });
	const ran = await runCompose(httpCtx, comp, "h", "caller", fetch, { callFor, http });
	assert.equal(ran.result.q.ok, true, JSON.stringify(ran.result));
	assert.equal(ran.result.q.output.status, 200);
	// the module lists no getServiceAuth and no edge routes it: the host mints it, as runModule does on its own
	assert.deepEqual(minted, [["s", "com.atproto.server.getServiceAuth", { aud: "did:web:svc.example", lxm: "x.y.z" }]]);
	assert.deepEqual(sent, [["https://svc.example/x", "Bearer T"]]);
	// no session: no token, and nothing sent
	const none = await runCompose(httpCtx, comp, "h", "caller", fetch, { http });
	assert.match(none.result.q.error, /needs a service-auth token \(aud did:web:svc.example, lxm x.y.z\)/);
	assert.equal(sent.length, 1);
});

test("cycles and depth past the limit are refused", async () => {
	const comp = await compose({
		instances: [toy("a"), toy("b"), toy("c")],
		edges: [edge("a", "x.y.b", "b"), edge("b", "x.y.a", "a"), edge("b", "x.y.c", "c")],
	});
	const cycle = await runCompose(ctx, comp, "a", "caller", { nsid: "x.y.b", input: { nsid: "x.y.a", input: { n: 1 } } });
	assert.deepEqual(cycle.result.q.output.q, { ok: false, error: "cycle: a -> b -> a" });
	const deep = await runCompose(ctx, comp, "a", "caller", { nsid: "x.y.b", input: { nsid: "x.y.c", input: { nsid: "x.y.d" } } }, { limits: { depth: 2 } });
	assert.deepEqual(deep.result.q.output.q, { ok: false, error: "composition exceeded depth 2" });
	const few = await runCompose(ctx, comp, "a", "caller", { nsid: "x.y.b", input: { nsid: "x.y.c", input: { nsid: "x.y.d" } } }, { limits: { calls: 1 } });
	assert.match(few.result.q.output.q.error, /composition exceeded 1 calls/);
});

test("manifests are checked: reasons, declared handlers, routes, pins", async () => {
	assert.throws(() => validateComposition({ instances: [toy("a")], edges: [{ from: "a", nsid: LOOKUP, to: "a" }] }), /reason is required/);
	assert.throws(() => validateComposition({ instances: [toy("a")], edges: [{ ...edge("a", LOOKUP, "a"), entry: "lookup" }] }), /edge a com.example.compose.lookup: an edge has no entry: it reaches the entry its target's module implements/);
	assert.throws(() => validateComposition({ instances: [toy("a")], edges: [edge("a", LOOKUP, "zz")] }), /to is not an instance/);
	assert.throws(() => validateComposition({ instances: [toy("a")], edges: [{ from: "a", nsid: LOOKUP, handler: "h", reason: "r" }] }), /not declared/);
	await assert.rejects(compose({ instances: [toy("a", { pin: "0".repeat(64) })] }), /composition: instance a module does not match its pin/);
	assert.throws(() => validateComposition({ instances: [toy("a", { pin: "deadbeef" })] }), /composition: instance a pin must be 64 lowercase hex characters/);
	const comp = await compose({ instances: [toy("a")], handlers: [{ name: "confirm" }] });
	await assert.rejects(runCompose(ctx, comp, "a", "caller", {}), /handler confirm was not supplied/);
});

test("an edge reaches the entry its target implements; a target that does not implement the NSID is refused", async () => {
	const comp = await compose({ instances: [toy("a"), toy("b")], edges: [edge("a", LOOKUP, "b")] });
	assert.deepEqual((await runCompose(ctx, comp, "a", "caller", { nsid: LOOKUP, input: { n: 2 } })).graph.at(-1), { from: "a", nsid: LOOKUP, to: "b", entry: "lookup", reason: "a asks b", ok: true });
	const bare = (key) => ({ ...spec, implements: undefined, module: wasm, id: key });
	await assert.rejects(loadComposition(ctx, { name: "t", instances: [toy("a"), toy("b")], edges: [edge("a", LOOKUP, "b")] }, { load: bare }),
		/composition: edge a com.example.compose.lookup: b \(compose\) does not implement com.example.compose.lookup/);
});

test("an instance's input is its module's config: keys, types, parameter types, required fields", async () => {
	const ok = await compose({ parameters: [{ key: "n", type: "integer" }], instances: [toy("a", { input: { n: { $param: "n" } } }), toy("b", { input: { n: 3 } })] });
	assert.equal(ok.manifest.instances.length, 2);
	const refused = async (manifest, re) => assert.rejects(compose(manifest), re);
	await refused({ instances: [toy("a", { input: { m: 1 } })] }, /composition: instance a: input m is not in compose's config \(n\)/);
	await refused({ instances: [toy("a", { input: { n: "3" } })] }, /composition: instance a: input n must be of type integer/);
	await refused({ parameters: [{ key: "n" }], instances: [toy("a", { input: { n: { $param: "n" } } })] }, /composition: instance a: input n is integer, but parameter n is string/);
	await refused({ instances: [toy("a", { input: [1] })] }, /composition: instance a: input must be an object/);
	const config = (config) => async (key) => ({ ...(await load(key)), config });
	await assert.rejects(loadComposition(ctx, { name: "t", instances: [toy("a", { input: { n: 1 } })] }, { load: config(undefined) }), /composition: instance a: input n is not in compose's config$/);
	await assert.rejects(loadComposition(ctx, { name: "t", instances: [toy("a")] }, { load: config([{ key: "n", label: "n", required: true }]) }), /composition: instance a: input n is required by compose's config/);
	await assert.rejects(loadComposition(ctx, { name: "t", instances: [toy("a")] }, { load: config([{ key: "n", type: "date" }]) }), /module config must be a list of \{ key, label, type\?/);
});

test("atlibs stage stages a composition; a host loads it back from the staged directory", async () => {
	const out = await mkdtemp(join(tmpdir(), "atlibs-compose-"));
	const manifest = new URL("../../../examples/rust/compose/composition.json", import.meta.url).pathname;
	await atlibs(["stage", "--target", "server", "--out", out, manifest]);
	const written = JSON.parse(await readFile(join(out, "composition.json"), "utf8"));
	assert.deepEqual(written.instances.map((i) => [i.module, i.pin.length]), [["compose", 64], ["compose", 64], ["compose", 64]]);
	assert.equal(written.target, "server");
	assert.equal(JSON.parse(await readFile(join(out, "stage.json"), "utf8")).modules.compose.pin, written.instances[0].pin);
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const comp = await loadComposition(ctx, written, { load: staged(pathToFileURL(`${out}/`), fileFetch) });
	const ran = await runCompose(ctx, comp, "ui", "caller", { nsid: "com.example.compose.lookup", input: { n: 6 } }, { handlers: { confirm: () => true } });
	assert.equal(ran.result.q.output.doubled, 12);
	await assert.rejects(atlibs(["stage", "--target", "edge", "--out", out, manifest]), /--target must be spa or server/);
});

test("staged() takes only module record fields: a staged json cannot pick its id or ledger owner", async () => {
	const files = { "modules/m.json": JSON.stringify({ ...spec, id: "/etc/passwd", owner: "someone-else", uri: "at://did:plc:other/x/y", sha256: "00" }), "modules/m.wasm": wasm };
	const base = "https://site.example/app/";
	const mod = await staged(base, async (url) => new Response(files[url.href.slice(base.length)]))("m");
	assert.equal(mod.id, "https://site.example/app/modules/m.wasm");
	assert.equal(mod.owner, undefined);
	assert.equal(mod.uri, undefined);
	assert.equal(mod.sha256, undefined);
	assert.deepEqual(mod.calls, spec.calls);
});

// Review 3b #3, #6 and the policy-stdin note, with a stub engine (ctx.wasm.run by entry).
const stub = (entries) => ({ wasm: { run: async ({ entry, input }) => entries[entry](input) } });
const ASK = "com.example.ask.x";

test("limits: depth and calls must be positive numbers; a manifest's bad values never disable the cap", async () => {
	for (const limits of [{ calls: "x" }, { calls: null }, { depth: 0 }, { depth: Infinity }, { other: 1 }, [], "8"]) {
		assert.throws(() => validateComposition({ instances: [{ name: "a", module: "m" }], limits }), /composition: limits must map depth, calls to positive numbers/, JSON.stringify(limits));
	}
	validateComposition({ instances: [{ name: "a", module: "m" }], limits: { depth: 3, calls: 10 } });
	let handled = 0;
	const go = ({ event }) => (event.type === "start" ? { calls: Array.from({ length: 150 }, (_, i) => ({ id: `c${i}`, nsid: ASK })) } : {});
	const manifest = { instances: [{ name: "a", module: "m", budget: { calls: 500 } }], handlers: [{ name: "h" }], edges: [{ from: "a", nsid: ASK, handler: "h", reason: "r" }], limits: { calls: "x" } };
	const ran = await runCompose(stub({ go }), { manifest, modules: { a: { module: new Uint8Array([1]) } } }, "a", "go", {}, { handlers: { h: async () => (handled++, 1) } });
	assert.equal(handled, 100, "an unvalidated manifest's limits.calls 'x' leaves the default 100");
	assert.match(ran.calls.find((c) => !c.ok).error, /composition exceeded 100 calls/);
});

test("subscribing to a held capability is refused before it runs", async () => {
	let invoked = 0;
	const closed = [];
	const go = ({ event, ping }) => {
		if (event.type === "start") return { calls: [{ id: "c", nsid: ping.$cap, subscribe: true, params: { hi: 1 } }] };
		if (event.type === "closed" || event.type === "result") closed.push(event.error);
		return {};
	};
	const ran = await runCompose(stub({ go }), { manifest: { instances: [{ name: "a", module: "m" }] }, modules: { a: { module: new Uint8Array([1]) } } }, "a", "go", { ping: async () => (invoked++, {}) });
	assert.equal(invoked, 0);
	assert.match(closed[0], /cap\.[0-9a-f]+\.\d+ cannot be subscribed to/);
	assert.deepEqual(ran.graph, []);
});

test("a policy sees the call without its stdin; the handler still gets it", async () => {
	const seen = {};
	const entries = {
		go: ({ event }) => (event.type === "start" ? { calls: [{ id: "x", nsid: ASK, input: { stdin: "tok", a: 1 } }] } : { result: event, done: true }),
		evaluate: ({ call }) => ((seen.policy = call.input), { result: { allow: true }, done: true }),
	};
	const manifest = { instances: [{ name: "a", module: "m" }, { name: "p", module: "m" }], handlers: [{ name: "h" }], edges: [{ from: "a", nsid: ASK, handler: "h", reason: "r", policies: [{ instance: "p" }] }] };
	const modules = { a: { module: new Uint8Array([1]) }, p: { module: new Uint8Array([1]) } };
	const ran = await runCompose(stub(entries), { manifest, modules }, "a", "go", {}, { handlers: { h: async (input) => ((seen.handler = input), "ok") } });
	assert.equal(ran.result.ok, true, ran.result.error);
	assert.deepEqual(seen, { policy: { a: 1 }, handler: { stdin: "tok", a: 1 } });
});

test("the host's held values start the root's run", async () => {
	const stub = { wasm: { run: async () => ({ result: { t: { $held: "t" } }, done: true }) } };
	const comp = await loadComposition(stub, { name: "test", instances: [toy("a")] }, { load });
	const ran = await runCompose(stub, comp, "a", "caller", {}, { held: { t: [1] } });
	assert.equal(ran.error, undefined);
	assert.deepEqual(ran.result, { t: [1] });
});