johnandersen777.bsky.social/atlibs

Switch repository

Loading…

// node --test javascript/src/server/server.test.mjs   (needs `cargo wasm` for hello.wasm and echo.wasm; no network)
import assert from "node:assert/strict";
import { createPublicKey, verify } from "node:crypto";
import { spawn } from "node:child_process";
import { createServer } from "node:http";
import { copyFileSync, mkdtempSync, readFileSync, realpathSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { DatabaseSync } from "node:sqlite";
import { test } from "node:test";

import { attestHost, attestationCid } from "../attest/index.js";
import { localModule, runModule } from "../run-module/index.js";
import { main, parse } from "./cli.mjs";
import { didkeyCall } from "../didkey/index.js";
import { contextOptions, createNodeContext, didkeyOf, fileFetch, hostCall, login, secp256k1, sessionStore } from "./index.js";

const HELLO = new URL("../../../target/wasm32-unknown-unknown/release/hello.wasm", import.meta.url).pathname;
const MANIFEST = new URL("../../../examples/rust/hello/module.json", import.meta.url).pathname;
const ECHO = new URL("../../../target/wasm32-unknown-unknown/release/echo.wasm", import.meta.url).pathname;
const ECHO_MANIFEST = new URL("../../../examples/rust/echo/module.json", import.meta.url).pathname;
const VIBES = new URL("../../../target/wasm32-unknown-unknown/release/vibescrobble.wasm", import.meta.url).pathname;
const VIBES_MANIFEST = new URL("../../../examples/rust/vibescrobble/module.json", import.meta.url).pathname;
const IROH = "com.fedfork.exec.iroh.connect";
const T = "endpointabex5k47heqqkdykv2b27ydg75jjaxeucm2ej74nt34a7up6wqlkyaiaenuhi5dqom5c6l3von3tcljrfzzgk3dbpexg4mbonfzg62bonruw42zof4";
// A stand-in transport (the real ones, e.g. the atlibs-iroh repo's, test themselves): one-shot connect echoes `send`.
const STUB_TRANSPORT = `export const transport = { "${IROH}": () => async ({ input }) => ({ output: \`got \${input.send}\` }) };\n`;
const DID = "did:plc:lpfuqerea3deuoyrn7ojser4";
const HAT7 = "did:plc:7q5lnitxv4meizdon5fmhzrq";
const stateDir = () => mkdtempSync(join(tmpdir(), "atlibs-server-"));
const context = (dir, extra) => createNodeContext({ id: "t", stateDir: dir, storage: { items: {} }, settingsSchema: { token: { type: "secret" } }, ...extra });

// A fetch that answers from a table of "METHOD url-prefix" -> body, recording each request.
function fakeFetch(routes) {
	const requests = [];
	const fetch = async (url, init = {}) => {
		requests.push({ url: String(url), method: init.method ?? "GET", headers: init.headers ?? {}, body: init.body && JSON.parse(init.body) });
		const key = Object.keys(routes).find((prefix) => `${init.method ?? "GET"} ${url}`.startsWith(prefix));
		if (!key) return Response.json({ error: "NotFound", message: String(url) }, { status: 404 });
		const out = typeof routes[key] === "function" ? routes[key](requests.at(-1)) : routes[key];
		return out instanceof Response ? out : Response.json(out);
	};
	return { fetch, requests };
}

test("storage: the browser collection surface, persisted in store.db", async () => {
	const dir = stateDir();
	const ctx = context(dir);
	const items = ctx.storage.items;
	await items.put("a", { n: 1, tag: "x" });
	await items.putMany([{ id: "b", data: { n: 2, tag: "y" } }, { id: "c", data: { n: 3, tag: "x" } }]);
	assert.deepEqual(await items.get("a"), { n: 1, tag: "x" });
	assert.equal(await items.get("zz"), null);
	assert.equal(await items.exists("b"), true);
	const { revision } = await items.getVersioned("a");
	assert.deepEqual(await items.compareAndSet("a", "stale", { n: 9 }), { applied: false });
	const set = await items.compareAndSet("a", revision, { n: 10, tag: "x" });
	assert.equal(set.applied, true);
	assert.deepEqual(await items.compareAndSet("new", null, { n: 0 }), { applied: true, revision: (await items.getVersioned("new")).revision });
	assert.deepEqual(await items.compareAndSet("new", null, { n: 1 }), { applied: false });
	assert.deepEqual(await items.compareAndDelete("new", revision), { applied: false });
	assert.deepEqual(await items.compareAndDelete("new", (await items.getVersioned("new")).revision), { applied: true });

	const page = await items.query({ where: { tag: "x" }, orderBy: { n: "desc" }, limit: 1 });
	assert.deepEqual(page, { items: [{ id: "a", data: { n: 10, tag: "x" } }], hasMore: true, cursor: "1" });
	assert.deepEqual((await items.query({ where: { tag: "x" }, orderBy: { n: "desc" }, cursor: page.cursor })).items.map((i) => i.id), ["c"]);
	assert.equal(await items.count({ n: { gte: 3 } }), 2);
	assert.deepEqual([...(await items.getMany(["a", "zz", "b"])).keys()], ["a", "b"]);

	assert.deepEqual(await items.updateIf("b", { where: { tag: "y" }, set: { tag: "z" }, delta: { n: { inc: 5 } } }), { applied: true, data: { n: 7, tag: "z" } });
	assert.deepEqual(await items.updateIf("b", { where: { tag: "y" }, set: { tag: "q" } }), { applied: false });
	assert.deepEqual(await items.updateIf("missing", { set: { tag: "q" } }), { applied: false });
	await assert.rejects(async () => items.updateIf("b", { set: { n: 1 }, delta: { n: { inc: 1 } } }), /both set and delta/);

	await ctx.kv.set("p:1", "one");
	await ctx.kv.set("q:1", "other");
	assert.deepEqual(await ctx.kv.list("p:"), [{ key: "p:1", value: "one" }]);
	assert.equal(await items.deleteMany(["b", "c", "zz"]), 2);
	assert.equal(await items.delete("a"), true);
	assert.equal(await items.delete("a"), false);
	await items.put("kept", { ok: true });
	ctx.close();

	const again = context(dir);
	assert.deepEqual(await again.storage.items.get("kept"), { ok: true });
	assert.equal(await again.kv.get("p:1"), "one");
	again.close();
});

test("settings: secret settings are sealed at rest with secret.key, plain ones are not", async () => {
	const dir = stateDir();
	const ctx = context(dir);
	await ctx.settings.set("token", { refresh: "hunter2" });
	await ctx.settings.set("name", "plain");
	assert.deepEqual(await ctx.settings.get("token"), { refresh: "hunter2" });
	ctx.close();

	const raw = new DatabaseSync(join(dir, "store.db")).prepare("SELECT id, value FROM rows WHERE ns = 't:settings'").all();
	const stored = Object.fromEntries(raw.map(({ id, value }) => [id, value]));
	assert.ok(JSON.parse(stored.token).$secret);
	assert.doesNotMatch(stored.token, /hunter2/);
	assert.equal(JSON.parse(stored.name), "plain");

	const again = context(dir);
	assert.deepEqual(await again.settings.get("token"), { refresh: "hunter2" });
	assert.deepEqual((await again.settings.list("")).map((s) => s.key).sort(), ["name", "token"]);
	again.close();
	// store.db without its secret.key: a fresh key is made and the sealed value does not open.
	const moved = stateDir();
	copyFileSync(join(dir, "store.db"), join(moved, "store.db"));
	const other = context(moved);
	assert.equal(await other.settings.get("name"), "plain");
	await assert.rejects(other.settings.get("token"));
	other.close();
});

// Hand-assembled: exports memory, alloc (returns 0) and spin (loops forever).
const SPIN = Uint8Array.from([
	0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00,
	0x01, 0x0c, 0x02, 0x60, 0x01, 0x7f, 0x01, 0x7f, 0x60, 0x02, 0x7f, 0x7f, 0x01, 0x7e,
	0x03, 0x03, 0x02, 0x00, 0x01,
	0x05, 0x03, 0x01, 0x00, 0x01,
	0x07, 0x19, 0x03, 0x06, ...new TextEncoder().encode("memory"), 0x02, 0x00, 0x05, ...new TextEncoder().encode("alloc"), 0x00, 0x00, 0x04, ...new TextEncoder().encode("spin"), 0x00, 0x01,
	0x0a, 0x10, 0x02, 0x04, 0x00, 0x41, 0x00, 0x0b, 0x09, 0x00, 0x03, 0x40, 0x0c, 0x00, 0x0b, 0x42, 0x00, 0x0b,
]);

test("wasm: runs in a worker_thread, checks sha256, stops a module that never returns", async () => {
	const ctx = context(stateDir(), { wasmTimeoutMs: 300 });
	const { readFile } = await import("node:fs/promises");
	const module = new Uint8Array(await readFile(HELLO));
	const sha256 = Buffer.from(await crypto.subtle.digest("SHA-256", module)).toString("hex");
	const out = await ctx.wasm.run({ module, sha256, entry: "count", input: { did: HAT7, collection: "c", now: "2026-11-08T00:00:00Z", state: null, event: { type: "start" } } });
	assert.equal(out.calls[0].service, HAT7);
	await assert.rejects(ctx.wasm.run({ module, sha256: "00".repeat(32), entry: "count", input: {} }), { code: "WASM_DIGEST_MISMATCH" });
	await assert.rejects(ctx.wasm.run({ module, entry: "no-such entry", input: {} }), { code: "WASM_ABI_INVALID" });
	const started = Date.now();
	await assert.rejects(ctx.wasm.run({ module: SPIN, entry: "spin", input: {} }), { code: "WASM_TIMEOUT" });
	assert.ok(Date.now() - started < 2000);
	ctx.close();
});

// The wasm queue is per process, sized at import: each case is its own node process (a file, not -e: the
// worker shim is eval'd CommonJS, and --input-type=module would carry over to it).
const node = (script, args, env) =>
	new Promise((resolve) => {
		const file = join(stateDir(), "main.mjs");
		writeFileSync(file, script);
		const child = spawn(process.execPath, [file, ...args], { env: { PATH: process.env.PATH, ...env } });
		let [stdout, stderr] = ["", ""];
		child.stdout.on("data", (chunk) => (stdout += chunk));
		child.stderr.on("data", (chunk) => (stderr += chunk));
		child.on("exit", (code) => resolve({ code, stdout, stderr }));
	});

// Each spin holds its slot for its full wall time T, so with 2 slots the (2k+1)th and (2k+2)th end no earlier
// than (k+1)·T (never more than 2 at once) and the first two end together; the run queued behind them waits
// ~3T, longer than T, and still succeeds (its timeout starts when it gets a slot).
test("wasm: ATLIBS_WASM_CONCURRENCY runs at once per process, the rest queue; queued time is not wall time", async () => {
	const T = 300;
	const count = { did: HAT7, collection: "c", now: "2026-11-08T00:00:00Z", state: null, event: { type: "start" } };
	const script = `import { readFileSync } from "node:fs";
import { createNodeContext } from ${JSON.stringify(new URL("./index.js", import.meta.url).href)};
const ctx = createNodeContext({ id: "t", stateDir: process.argv[2], wasmTimeoutMs: ${T} });
const started = Date.now();
const timed = (p) => p.then((out) => [out.calls[0].service, Date.now() - started], (e) => [e.code, Date.now() - started]);
const spins = Array.from({ length: 6 }, () => timed(ctx.wasm.run({ module: Uint8Array.from(${JSON.stringify([...SPIN])}), entry: "spin", input: {} })));
const queued = timed(ctx.wasm.run({ module: new Uint8Array(readFileSync(process.argv[3])), entry: "count", input: ${JSON.stringify(count)} }));
console.log(JSON.stringify({ spins: await Promise.all(spins), queued: await queued }));
ctx.close();`;
	const ran = await node(script, [stateDir(), HELLO], { ATLIBS_WASM_CONCURRENCY: "2" });
	assert.equal(ran.code, 0, ran.stderr);
	const { spins, queued } = JSON.parse(ran.stdout);
	assert.deepEqual(new Set(spins.map(([code]) => code)), new Set(["WASM_TIMEOUT"]));
	const ends = spins.map(([, end]) => end).sort((a, b) => a - b);
	ends.forEach((end, i) => assert.ok(end >= (Math.floor(i / 2) + 1) * (T - 10), `run ${i}: ${ends}`));
	assert.ok(ends[1] < 2 * T - 50, `two at once: ${ends}`);
	assert.equal(queued[0], HAT7);
	assert.ok(queued[1] >= 3 * (T - 10), `queued behind the spins: ${queued[1]}`);
	for (const bad of ["0", "-1", "two"]) {
		const failed = await node(`import ${JSON.stringify(new URL("./index.js", import.meta.url).href)};`, [], { ATLIBS_WASM_CONCURRENCY: bad });
		assert.equal(failed.code, 1, bad);
		assert.match(failed.stderr, new RegExp(`Error: ATLIBS_WASM_CONCURRENCY must be a positive integer, not "${bad}"`));
	}
});

test("login: createSession, XRPC to the DID document's PDS, refresh on ExpiredToken, saved sessions", async () => {
	const valid = new Set(["a2", "a3"]); // createSession's first token "a1" is already expired
	let created = 0;
	const didDoc = { service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example" }] };
	const { fetch, requests } = fakeFetch({
		"POST https://entry.example/xrpc/com.atproto.server.createSession": () => ({ did: DID, handle: "alice.test", accessJwt: created++ ? "a3" : "a1", refreshJwt: "r1", didDoc }),
		"POST https://pds.example/xrpc/com.atproto.server.refreshSession": (req) =>
			req.headers.authorization === "Bearer r1" ? { did: DID, accessJwt: "a2", refreshJwt: "r2", didDoc } : Response.json({ error: "ExpiredToken" }, { status: 400 }),
		"GET https://pds.example/xrpc/app.test.get": (req) => (valid.has(req.headers.authorization.slice(7)) ? { ok: 1 } : Response.json({ error: "ExpiredToken" }, { status: 400 })),
		"GET https://pds.example/xrpc/app.test.bad": Response.json({ error: "InvalidRequest", message: "nope" }, { status: 400 }),
		"POST https://limited.example/": Response.json({ error: "RateLimitExceeded", message: "Rate Limit Exceeded" }, { status: 429, headers: { "ratelimit-reset": "1794182400", "ratelimit-policy": "10;w=86400" } }),
	});
	await assert.rejects(login({ identifier: "x", password: "y", service: "https://limited.example", fetch }), /RateLimitExceeded: Rate Limit Exceeded \(10;w=86400; resets 2026-11-09T00:00:00.000Z\)/);
	requests.length = 0;
	const sessions = [];
	const session = await login({ identifier: "alice.test", password: "pw", service: "https://entry.example", fetch, onSession: (s) => sessions.push(s) });
	assert.equal(session.did, DID);
	assert.equal(session.pds, "https://pds.example");
	assert.deepEqual(requests[0].body, { identifier: "alice.test", password: "pw" });
	assert.deepEqual(await session.xrpc("app.test.get", { params: { a: "b" } }), { ok: 1 });
	assert.deepEqual(requests.slice(1).map((r) => r.url), ["https://pds.example/xrpc/app.test.get?a=b", "https://pds.example/xrpc/com.atproto.server.refreshSession", "https://pds.example/xrpc/app.test.get?a=b"]);
	assert.deepEqual(sessions.map((s) => [s.identifier, s.accessJwt]), [["alice.test", "a1"], ["alice.test", "a2"]]);
	await assert.rejects(session.xrpc("app.test.bad"), /app.test.bad failed with InvalidRequest: nope/);
	await assert.rejects(login({ identifier: "x", password: "y", service: "https://nowhere.example", fetch }), /createSession failed with NotFound/);

	// A saved session skips createSession; a refused refresh logs in again.
	const path = (from) => requests.slice(from).map((r) => r.url.split("/xrpc/")[1]);
	let from = requests.length;
	const reuse = await login({ identifier: "alice.test", password: "pw", service: "https://entry.example", fetch, load: async () => ({ ...sessions[1], accessJwt: "old", refreshJwt: "revoked" }) });
	assert.deepEqual(path(from), []);
	assert.deepEqual(await reuse.xrpc("app.test.get"), { ok: 1 });
	assert.deepEqual(path(from), ["app.test.get", "com.atproto.server.refreshSession", "com.atproto.server.createSession", "app.test.get"]);
	// A session saved for another identifier is ignored.
	from = requests.length;
	await login({ identifier: "bob.test", password: "pw", service: "https://entry.example", fetch, load: async () => sessions[1] });
	assert.deepEqual(path(from), ["com.atproto.server.createSession"]);
});

// A PDS (node:http, so other processes can use it) that takes each refresh token once (rotation: a
// second refresh with r1 fails and falls back to createSession), with a1 expired, and refused as
// invalid once rotated. Responses take 10 ms so concurrent runs overlap.
async function rotatingPds() {
	const counts = { refreshes: 0, creates: 0 };
	const server = createServer((req, res) => {
		const nsid = req.url.split("/xrpc/")[1]?.split("?")[0];
		const bearer = req.headers.authorization ?? "";
		const json = (status, body) => setTimeout(() => res.writeHead(status, { "content-type": "application/json" }).end(JSON.stringify(body)), 10);
		const didDoc = { service: [{ id: "#atproto_pds", serviceEndpoint: base }] };
		if (nsid === "com.atproto.server.createSession") return counts.creates++, json(200, { did: DID, handle: "alice.test", accessJwt: "a9", refreshJwt: "r9", didDoc });
		if (nsid === "com.atproto.server.refreshSession")
			return ++counts.refreshes === 1 && bearer === "Bearer r1" ? json(200, { did: DID, handle: "alice.test", accessJwt: "a2", refreshJwt: "r2", didDoc }) : json(400, { error: "ExpiredToken", message: "refresh token already used" });
		if (bearer === "Bearer a1") return counts.refreshes ? json(401, { error: "InvalidToken" }) : json(400, { error: "ExpiredToken" });
		json(200, { ok: 1 });
	});
	await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
	const base = `http://127.0.0.1:${server.address().port}`;
	const seed = (dir) => {
		Object.assign(counts, { refreshes: 0, creates: 0 });
		const ctx = createNodeContext(contextOptions(dir));
		return ctx.settings.set("session", { identifier: "alice.test", did: DID, handle: "alice.test", accessJwt: "a1", refreshJwt: "r1", didDoc: { service: [{ id: "#atproto_pds", serviceEndpoint: base }] } }).finally(() => ctx.close());
	};
	return { base, counts, seed, close: () => server.close() };
}

test("login: concurrent runs share one refresh, in this process and across processes sharing the state dir", async () => {
	const pds = await rotatingPds();
	const dir = stateDir();
	const contexts = [];
	const open = () => contexts.push(createNodeContext(contextOptions(dir))) && contexts.at(-1); // its own sqlite connection
	const signIn = (ctx) => login({ identifier: "alice.test", password: "pw", service: pds.base, ...sessionStore(ctx) });
	try {
		// 8 runs on one session: one refresh, everyone retries with it
		await pds.seed(dir);
		const one = await signIn(open());
		assert.deepEqual(await Promise.all(Array.from({ length: 8 }, () => one.xrpc("app.test.get"))), Array(8).fill({ ok: 1 }));
		assert.deepEqual(pds.counts, { refreshes: 1, creates: 0 });

		// two connections (as two processes have) with the same saved, expired session: one refresh
		await pds.seed(dir);
		const [a, b] = [await signIn(open()), await signIn(open())];
		assert.deepEqual(await Promise.all(Array.from({ length: 8 }, (_, i) => (i % 2 ? a : b).xrpc("app.test.get"))), Array(8).fill({ ok: 1 }));
		assert.deepEqual(pds.counts, { refreshes: 1, creates: 0 });

		// a run still holding the rotated-away token (401 InvalidToken) re-reads the saved session: no refresh
		await pds.seed(dir);
		const [first, stale] = [await signIn(open()), await signIn(open())];
		assert.deepEqual(await first.xrpc("app.test.get"), { ok: 1 });
		assert.deepEqual(await stale.xrpc("app.test.get"), { ok: 1 });
		assert.deepEqual(pds.counts, { refreshes: 1, creates: 0 });
		assert.equal((await open().settings.get("session")).refreshJwt, "r2");

		// an empty state dir: concurrent first logins create one session
		const empty = stateDir();
		Object.assign(pds.counts, { refreshes: 0, creates: 0 });
		const fresh = () => contexts.push(createNodeContext(contextOptions(empty))) && contexts.at(-1);
		await Promise.all([signIn(fresh()), signIn(fresh())]);
		assert.deepEqual(pds.counts, { refreshes: 0, creates: 1 });

		// 4 real processes sharing the state dir
		await pds.seed(dir);
		const script = `import { contextOptions, createNodeContext, login, sessionStore } from ${JSON.stringify(new URL("./index.js", import.meta.url).href)};
const ctx = createNodeContext(contextOptions(process.env.ATLIBS_SERVER_STATE));
const session = await login({ identifier: "alice.test", password: "pw", service: process.env.TEST_PDS, ...sessionStore(ctx) });
console.log(JSON.stringify(await session.xrpc("app.test.get")));`;
		const runs = await Promise.all(
			Array.from({ length: 4 }, () => {
				const child = spawn(process.execPath, ["--input-type=module", "-e", script], { env: { PATH: process.env.PATH, ATLIBS_SERVER_STATE: dir, TEST_PDS: pds.base } });
				let out = "";
				child.stdout.on("data", (chunk) => (out += chunk));
				child.stderr.on("data", (chunk) => (out += chunk));
				return new Promise((resolve) => child.on("exit", (code) => resolve({ code, out: out.trim() })));
			}),
		);
		assert.deepEqual(runs, Array(4).fill({ code: 0, out: '{"ok":1}' }));
		assert.deepEqual(pds.counts, { refreshes: 1, creates: 0 });
	} finally {
		contexts.forEach((ctx) => ctx.close());
		pds.close();
	}
});

test("ctx.lock: one holder at a time across connections; a dead holder's lease runs out", async () => {
	const dir = stateDir();
	const [a, b] = [context(dir), context(dir)];
	const order = [];
	const hold = (ctx, tag) => ctx.lock("x", async () => (order.push(`${tag}+`), await new Promise((r) => setTimeout(r, 30)), order.push(`${tag}-`)));
	await Promise.all([hold(a, "a"), hold(b, "b"), hold(a, "c")]);
	assert.equal(order.length, 6);
	for (let i = 0; i < order.length; i += 2) assert.deepEqual([order[i], order[i + 1]], [`${order[i][0]}+`, `${order[i][0]}-`], order.join(" ")); // never interleaved
	// a lease left behind by a holder that died (until in the past) is taken over
	const rows = new DatabaseSync(join(dir, "store.db"));
	rows.prepare("INSERT INTO rows (ns, id, value, revision) VALUES (?, ?, ?, ?)").run("t:locks", "y", JSON.stringify({ owner: "dead", until: Date.now() - 1 }), "r");
	rows.close();
	assert.equal(await b.lock("y", async () => "taken"), "taken");
	a.close();
	b.close();
});

test("runModule: concurrent runs of one module against one context stay apart (a fresh instance per event)", async () => {
	const ctx = createNodeContext(contextOptions(stateDir()));
	try {
		const wasm = new URL("../../../target/wasm32-unknown-unknown/release/compose.wasm", import.meta.url);
		const mod = await localModule(wasm, new URL("../../../examples/rust/compose/module.json", import.meta.url), fileFetch);
		const ran = await Promise.all(Array.from({ length: 8 }, (_, n) => runModule(ctx, mod, "even", { call: { input: { n } } }, async () => ({}))));
		assert.deepEqual(ran.map((r) => [r.error, r.result?.allow]), Array.from({ length: 8 }, (_, n) => [undefined, n % 2 === 0]));
	} finally {
		ctx.close();
	}
});

// Two connections on one state dir stand in for two processes (each has its own sqlite connection).
test("did:keys across processes: first use of different modules at once keeps both; one serving instance per did:key", async () => {
	const dir = stateDir();
	const [a, b] = [createNodeContext(contextOptions(dir)), createNodeContext(contextOptions(dir))];
	try {
		const made = await Promise.all(Array.from({ length: 6 }, (_, i) => didkeyOf(i % 2 ? a : b, `module-${i % 3}`)));
		const keys = await a.settings.get("didkeys");
		assert.deepEqual(Object.keys(keys).sort(), ["module-0", "module-1", "module-2"], "no key lost to a concurrent write");
		for (const key of made) assert.equal(key.did, (await didkeyOf(a, `module-${made.indexOf(key) % 3}`)).did, "every run of one module got the saved key");

		const key = made[0];
		const sub = () => ({ async *[Symbol.asyncIterator]() {}, close() {} });
		const serve = (ctx) => didkeyCall(key, async () => sub(), { services: ["https://relay.example"], lease: (name) => ctx.lease(name) })("com.fedproxy.temp.xrpc.subscribe", { subscribe: true, service: "https://relay.example", serviceAuth: {} });
		const serving = await serve(a);
		await assert.rejects(serve(b), /already serving in another run/);
		serving.close();
		await new Promise((resolve) => setTimeout(resolve, 10)); // the release is a write
		(await serve(b)).close();
	} finally {
		a.close();
		b.close();
	}
});

test("hostCall: own PDS, proxy header, https service with service auth, DID service read, attest", async () => {
	const { fetch, requests } = fakeFetch({
		[`GET https://plc.directory/${HAT7}`]: { service: [{ id: "#atproto_pds", serviceEndpoint: "https://hat7.example" }] },
		"GET https://hat7.example/xrpc/com.atproto.repo.listRecords": { records: [] },
		"POST https://bidder.example/xrpc/com.publicdomainrelay.temp.market.submitRfp": { ok: true },
	});
	const xrpcs = [];
	const xrpc = async (nsid, init) => (xrpcs.push({ nsid, ...init }), nsid === "com.atproto.server.getServiceAuth" ? { token: "svc-token" } : { nsid });
	const call = hostCall({ ctx: { http: { fetch } }, xrpc, attest: async (input) => ({ attested: input }) });

	assert.deepEqual(await call("com.atproto.repo.createRecord", { input: { a: 1 } }), { nsid: "com.atproto.repo.createRecord" });
	assert.deepEqual(xrpcs.at(-1), { nsid: "com.atproto.repo.createRecord", params: undefined, input: { a: 1 }, headers: undefined });
	await call("app.test.associateConfirm", { input: {}, proxy: `${HAT7}#bidder_associate` });
	assert.deepEqual(xrpcs.at(-1).headers, { "atproto-proxy": `${HAT7}#bidder_associate` });

	assert.deepEqual(await call("com.atproto.repo.listRecords", { params: { repo: HAT7, collection: "c" }, service: HAT7 }), { records: [] });
	assert.equal(requests.at(-1).url, `https://hat7.example/xrpc/com.atproto.repo.listRecords?${new URLSearchParams({ repo: HAT7, collection: "c" })}`);
	assert.equal(requests.at(-1).headers.authorization, undefined);

	assert.deepEqual(await call("com.publicdomainrelay.temp.market.submitRfp", { input: { rfpUri: "at://x" }, service: "https://bidder.example" }), { ok: true });
	assert.deepEqual(xrpcs.at(-1), { nsid: "com.atproto.server.getServiceAuth", params: { aud: "did:web:bidder.example", lxm: "com.publicdomainrelay.temp.market.submitRfp" } });
	assert.deepEqual(requests.at(-1), { url: "https://bidder.example/xrpc/com.publicdomainrelay.temp.market.submitRfp", method: "POST", headers: { authorization: "Bearer svc-token", "content-type": "application/json" }, body: { rfpUri: "at://x" } });

	assert.deepEqual(await call("com.fedfork.exec.attest", { input: { repository: DID } }), { attested: { repository: DID } });
	await assert.rejects(hostCall({ ctx: { http: { fetch } }, xrpc })("com.fedfork.exec.attest", { input: {} }), /attest needs a session/);
	await assert.rejects(hostCall({ ctx: { http: { fetch } } })("com.atproto.repo.createRecord", { input: {} }), /needs a session/);
	assert.deepEqual(await hostCall({ ctx: { http: { fetch } } })("com.atproto.repo.listRecords", { params: { repo: HAT7 }, service: HAT7 }), { records: [] });
});

test("attest: node:crypto secp256k1 under attestHost, key sealed in settings, signature verifies low-S", async () => {
	const ctx = createNodeContext({ id: "atlibs-server", stateDir: stateDir(), settingsSchema: { attestKeys: { type: "secret" } } });
	const attest = attestHost(ctx, secp256k1, DID);
	const { key } = await attest({ repository: DID });
	const record = { $type: "com.publicdomainrelay.temp.market.rfp", domain: "compute", createdAt: "2026-11-08T00:00:00.000Z" };
	const { record: signed } = await attest({ repository: DID, record });
	const entry = signed.signatures[0];
	assert.equal(entry.key, key);
	const { bytes, cid } = await attestationCid(signed, entry, DID);
	assert.equal(entry.cid, cid);
	const rs = Buffer.from(entry.signature.$bytes, "base64");
	assert.ok(BigInt(`0x${rs.subarray(32).toString("hex")}`) <= 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141n / 2n);
	const secret = Buffer.from((await ctx.settings.get("attestKeys"))[DID], "hex");
	const point = secp256k1.getPublicKey(secret, false);
	const publicKey = createPublicKey({ key: { kty: "EC", crv: "secp256k1", x: Buffer.from(point.subarray(1, 33)).toString("base64url"), y: Buffer.from(point.subarray(33)).toString("base64url") }, format: "jwk" });
	assert.ok(verify("sha256", bytes, { key: publicKey, dsaEncoding: "ieee-p1363" }, rs));
	await assert.rejects(attest({ repository: "did:plc:other", record }), /session DID/);
	assert.ok((await ctx.settings.getVersioned("attestKeys")).value[DID]);
	ctx.close();
});

test("runModule + hostCall: hello counts another account's records end to end, signed out, from a fake network", async () => {
	const ctx = context(stateDir(), {
		fetch: fakeFetch({
			[`GET https://plc.directory/${HAT7}`]: { service: [{ id: "#atproto_pds", serviceEndpoint: "https://hat7.example" }] },
			"GET https://hat7.example/xrpc/com.atproto.repo.listRecords": { records: [{ uri: `at://${HAT7}/x/1`, value: { endpointUrl: "https://bidder.example", appliesTo: ["com.publicdomainrelay.temp.compute.vm"] } }] },
		}).fetch,
	});
	const { readFile } = await import("node:fs/promises");
	const module = { module: new Uint8Array(await readFile(HELLO)), calls: JSON.parse(await readFile(MANIFEST, "utf8")).calls };
	const { result } = await runModule(ctx, module, "count", { did: HAT7, collection: "c" }, hostCall({ ctx }));
	assert.deepEqual(result, { collection: "c", records: 1 });
	ctx.close();
});

test("cli: arguments", async () => {
	assert.deepEqual(parse(["run", "at://did:plc:x/com.fedfork.exec.wasm.module/requester", "offerings", '{"bidders":[]}']), {
		uri: "at://did:plc:x/com.fedfork.exec.wasm.module/requester", entry: "offerings", input: { bidders: [] },
	});
	assert.deepEqual(parse(["run", "m.wasm", "module.json", "bind_key"]), { wasm: "m.wasm", manifest: "module.json", entry: "bind_key", input: {} });
	for (const ref of ["bob.example/requester", "bob.example/requester@bafyrecord", "did:plc:x/requester"]) assert.deepEqual(parse(["run", ref, "offerings"]), { uri: ref, entry: "offerings", input: {} }, ref);
	assert.deepEqual(parse(["run", "build/m.wasm", "module.json", "e"]).wasm, "build/m.wasm");
	assert.throws(() => parse([]), /usage/);
	assert.throws(() => parse(["exec", "m.wasm", "module.json", "e"]), /usage/);
	assert.throws(() => parse(["run", "m.wasm", "module.json"]), /usage/);
	assert.throws(() => parse(["run", "at://did:plc:x/c/r"]), /usage/);
	assert.throws(() => parse(["run", "at://did:plc:x/c/r", "e", "{}", "extra"]), /usage/);
	assert.throws(() => parse(["run", "m.wasm", "module.json", "e", "{nope"]), /not JSON/);
	assert.throws(() => parse(["run", "m.wasm", "module.json", "e", "[1]"]), /JSON object/);
	// The input from stdin (-) or a file (@path), so a secret stays out of argv; a parse error never echoes it.
	const sources = [];
	const read = (source) => (sources.push(source), '{"stdin":"tok"}');
	assert.deepEqual(parse(["run", "m.wasm", "module.json", "e", "-"], read).input, { stdin: "tok" });
	assert.deepEqual(parse(["run", "m.wasm", "module.json", "e", "@in.json"], read).input, { stdin: "tok" });
	assert.deepEqual(sources, ["-", "@in.json"]);
	assert.throws(() => parse(["run", "m.wasm", "module.json", "e", '{"stdin":"secret'], read), (e) => e.message === "input is not JSON");
	// Signed out (no --app-password): hello needs a did, and a record write needs a session.
	const env = { ATLIBS_SERVER_STATE: stateDir() };
	// Module and call errors come back as data, not throws.
	const failure = (argv) => main(argv, env).then((ran) => ran.error ?? JSON.stringify(ran.calls), (error) => error.message);
	assert.match(await failure(["run", HELLO, MANIFEST, "count", '{"collection":"c"}']), /missing field `did`/);
	assert.match(await failure(["run", VIBES, VIBES_MANIFEST, "scrobble", `{"did":"${DID}","text":"🚀 shipped"}`]), /needs a session/);
});

test("hostCall: subscribe opens wss at the service, the DID's PDS, or the session PDS; no proxy", async () => {
	const urls = [];
	class FakeSocket {
		constructor(url) {
			urls.push(url);
			this.readyState = 1;
		}
		close() {}
	}
	const ctx = { http: { fetch: async () => new Response(JSON.stringify({ service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.bob.test" }] })) } };
	const call = hostCall({ ctx, pds: "https://pds.me.test", WebSocket: FakeSocket });
	const SUB = "com.atproto.sync.subscribeRepos";
	await call(SUB, { subscribe: true, params: { cursor: "1" } });
	await call(SUB, { subscribe: true, service: "did:plc:bob" });
	const stream = await call("com.fedproxy.temp.xrpc.subscribe", { subscribe: true, service: "https://xrpc.fedproxy.com", params: { did: "did:key:z" } });
	assert.equal(typeof stream.send, "function");
	assert.deepEqual(urls, [`wss://pds.me.test/xrpc/${SUB}?cursor=1`, `wss://pds.bob.test/xrpc/${SUB}`, "wss://xrpc.fedproxy.com/xrpc/com.fedproxy.temp.xrpc.subscribe?did=did%3Akey%3Az"]);
	await assert.rejects(call(SUB, { subscribe: true, proxy: "did:web:x#y" }), /atproto-proxy/);
	await assert.rejects(hostCall({ ctx })(SUB, { subscribe: true }), /needs a service or a session/);
});

test("hostCall: a subscribe with serviceAuth gets the session's service-auth token as a header or query param", async () => {
	const sockets = [];
	class FakeSocket {
		constructor(url, options) {
			sockets.push({ url, headers: options?.headers });
			this.readyState = 1;
		}
		close() {}
	}
	const minted = [];
	const xrpc = async (nsid, { params }) => (minted.push([nsid, params]), { token: "jwt1" });
	const call = hostCall({ ctx: { http: { fetch: async () => new Response("{}") } }, xrpc, pds: "https://pds.me.test", WebSocket: FakeSocket });
	const NSID = "com.fedproxy.temp.xrpc.subscribe";
	await call(NSID, { subscribe: true, service: "https://xrpc.fedproxy.com", serviceAuth: {} });
	await call(NSID, { subscribe: true, service: "https://xrpc.fedproxy.com", params: { did: "d" }, serviceAuth: { param: "service_auth" } });
	assert.deepEqual(minted, [["com.atproto.server.getServiceAuth", { aud: "did:web:xrpc.fedproxy.com", lxm: NSID }], ["com.atproto.server.getServiceAuth", { aud: "did:web:xrpc.fedproxy.com", lxm: NSID }]]);
	assert.deepEqual(sockets, [
		{ url: `wss://xrpc.fedproxy.com/xrpc/${NSID}`, headers: { authorization: "Bearer jwt1" } },
		{ url: `wss://xrpc.fedproxy.com/xrpc/${NSID}?did=d&service_auth=jwt1`, headers: undefined },
	]);
	await assert.rejects(hostCall({ ctx: {}, WebSocket: FakeSocket })(NSID, { subscribe: true, service: "https://xrpc.fedproxy.com", serviceAuth: {} }), /serviceAuth needs a session/);
});

test("cli: input on stdin (-); the process exits 1 when the run failed; ATLIBS_TRANSPORTS loads transports", async () => {
	const { spawnSync } = await import("node:child_process");
	const cli = new URL("./cli.mjs", import.meta.url).pathname;
	const stub = join(stateDir(), "stub-transport.mjs");
	writeFileSync(stub, STUB_TRANSPORT);
	const run = (wasm, manifest, entry, input, transports) =>
		spawnSync(process.execPath, [cli, "run", wasm, manifest, entry, "-"], { input, encoding: "utf8", env: { PATH: process.env.PATH, ATLIBS_SERVER_STATE: stateDir(), ...(transports && { ATLIBS_TRANSPORTS: transports }) } });
	const failed = run(HELLO, MANIFEST, "count", '{"collection":"c"}');
	assert.equal(failed.status, 1);
	assert.match(JSON.parse(failed.stdout).error, /missing field `did`/);
	// echo `connect` over a stub transport: the stdin-fed input reaches iroh.connect (send comes back).
	const input = JSON.stringify({ ticket: T, alpn: "echo/1", send: "tok\n" });
	const ok = run(ECHO, ECHO_MANIFEST, "connect", input, stub);
	assert.equal(ok.status, 0, ok.stderr);
	assert.deepEqual(JSON.parse(ok.stdout).result, { output: "got tok\n" });
	// none given: refused as a host built without it; a module that exports no transport is an error
	assert.match(JSON.parse(run(ECHO, ECHO_MANIFEST, "connect", input).stdout).error, /built without iroh/);
	const none = run(ECHO, ECHO_MANIFEST, "connect", input, cli);
	assert.equal(none.status, 1);
	assert.match(none.stderr, /ATLIBS_TRANSPORTS: .*cli\.mjs exports no transport/);
});

test("cli: the managed_records ledger persists in the state dir across runs", async () => {
	const dir = stateDir();
	const uri = `at://${DID}/com.publicdomainrelay.temp.compute.vm/3abc`;
	const first = createNodeContext(contextOptions(dir));
	await first.storage.managed_records.put(uri, { createdAt: "2026-11-09T00:00:00.000Z" });
	first.close();
	const next = createNodeContext(contextOptions(dir));
	assert.ok(await next.storage.managed_records.get(uri));
	next.close();
});

test("cli: a did:key is sealed per module identity (its absolute path), never the name it gives itself", async () => {
	const env = { ATLIBS_SERVER_STATE: stateDir() };
	const spec = JSON.parse(readFileSync(MANIFEST, "utf8"));
	const manifest = join(stateDir(), "module.json");
	writeFileSync(manifest, JSON.stringify({ ...spec, name: "ingress", didkey: { services: ["https://xrpc.fedproxy.com"] } }));
	const copies = [stateDir(), stateDir()].map((dir) => (copyFileSync(HELLO, join(dir, "m.wasm")), join(dir, "m.wasm")));
	for (const wasm of copies) await main(["run", wasm, manifest, "count", '{"collection":"c"}'], env);
	const ctx = createNodeContext(contextOptions(env.ATLIBS_SERVER_STATE));
	const keys = await ctx.settings.get("didkeys");
	ctx.close();
	assert.deepEqual(Object.keys(keys).sort(), copies.map((c) => realpathSync(c)).sort());
	assert.notDeepEqual(keys[realpathSync(copies[0])], keys[realpathSync(copies[1])]);
	// calls must be a list: a string would allow every NSID it contains
	writeFileSync(manifest, JSON.stringify({ ...spec, calls: "com.atproto.repo.createRecord" }));
	await assert.rejects(main(["run", copies[0], manifest, "count", "{}"], env), /calls must be a list of strings/);
});

test("cli: main() given no transport refuses iroh.connect; given one it serves it", async () => {
	const args = ["run", ECHO, ECHO_MANIFEST, "connect", JSON.stringify({ ticket: T, alpn: "echo/1", send: "hi\n" })];
	const env = { ATLIBS_SERVER_STATE: stateDir() };
	const without = await main(args, env).catch((error) => ({ error: error.message }));
	assert.match(without.error, /com\.fedfork\.exec\.iroh\.connect.*built without iroh/);
	const made = [];
	const stub = { [IROH]: (given) => (made.push(Object.keys(given).sort()), async ({ input }) => ({ output: `got ${input.send}` })) };
	const ran = await main(args, env, undefined, stub);
	assert.equal(ran.error, undefined, ran.error);
	assert.deepEqual(ran.result, { output: "got hi\n" });
	assert.deepEqual(made, [["env", "stateDir"]], "made per run with the env and state dir");
});

test("sessions are explicit: sessionFlags and openSession never read ATPROTO_* (no config, no session)", async () => {
	const { openSession, sessionFlags } = await import("./index.js");
	const env = { ATPROTO_HANDLE: "alice.test", ATPROTO_PASSWORD: "pw", PW: "secret" };
	assert.deepEqual(sessionFlags(["run", "x"], { env }), { rest: ["run", "x"] });
	assert.equal(await openSession(undefined, undefined), undefined);
	assert.deepEqual(sessionFlags(["run", "--app-password", "bob.test", "--password-env", "PW", "--pds=http://pds"], { env }).config, { appPassword: { handle: "bob.test", password: "secret", service: "http://pds" } });
	assert.deepEqual(sessionFlags(["--app-password", "bob.test", "--password-file", "-"], { env, read: () => "filepw\nrest" }).config.appPassword.password, "filepw");
	for (const bad of [["--password-env", "PW"], ["--app-password", "b", "--password-env", "NOPE"], ["--app-password", "b", "--password-env", "PW", "--password-file", "f"], ["--app-password"], ["--pds", "x"]]) assert.throws(() => sessionFlags(bad, { env, read: () => "x" }), undefined, bad.join(" "));
	await assert.rejects(openSession(undefined, { oauth: { did: "did:plc:x" } }), /oauthServer/);
});