johnandersen777.bsky.social/atlibs

Switch repository

Loading…

// node --test rust/host/cli.test.mjs   (needs `cargo wasm` and `cargo host`; no network)
// The atlibs CLI end to end against the local fixture registry (fixture-registry.mjs) publishing
// compose and vibescrobble as alice.test. add -> lockfile + vendored copies; stage
// --target spa -> client metadata at the configured paths; run and serve the added lib.
import assert from "node:assert/strict";
import { execFileSync, spawn } from "node:child_process";
import { createHash } from "node:crypto";
import { cp, mkdir, mkdtemp, readFile, symlink, writeFile } from "node:fs/promises";
import { createServer as createHttpsServer } from "node:https";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { after, before, test } from "node:test";

import { modulePin } from "../../javascript/src/run-module/index.js";
import { contextOptions, createNodeContext } from "../../javascript/src/server/index.js";
import { atlibs, hostBin } from "./conformance.mjs";
import { DID, fixtureRegistry } from "./fixture-registry.mjs";

const ROOT = new URL("../../", import.meta.url);

let registry, records, blobs, put, publish, auth, seen, base, env, dir;
before(async () => {
	registry = await fixtureRegistry();
	({ records, blobs, put, publish, auth, seen, base } = registry);
	await publish("compose");
	await publish("vibescrobble");
	dir = await mkdtemp(join(tmpdir(), "atlibs-cli-"));
	env = { ...registry.env, ATLIBS_SERVER_STATE: join(dir, "state") };
});
after(() => registry?.close());

test("add <handle>/<lib> -> stage --target spa -> client metadata at the configured paths; run the added lib", async () => {
	await atlibs(["add", "alice.test/compose", "alice.test/vibescrobble"], { cwd: dir, env });
	const lock = JSON.parse(await readFile(join(dir, "atlibs.lock.json"), "utf8"));
	assert.equal(lock.modules.vibescrobble.uri, `at://${DID}/com.fedfork.exec.wasm.module/vibescrobble`);
	assert.equal(lock.modules.vibescrobble.ref, "alice.test/vibescrobble");
	assert.equal(lock.modules.vibescrobble.session, "vibescrobble");
	assert.equal(lock.sessions.vibescrobble.record.name, JSON.parse(await readFile(new URL("records/vibescrobble.json", ROOT), "utf8")).name);
	// the lockfile's pin covers the wasm and the record's grants: the JS modulePin of the vendored pair
	const vendored = { ...JSON.parse(await readFile(join(dir, "modules/compose.json"), "utf8")), module: await readFile(join(dir, "modules/compose.wasm")) };
	assert.equal(await modulePin(vendored), lock.modules.compose.pin);

	const out = join(dir, "site");
	const { stdout } = await atlibs(["stage", "--target", "spa", "--out", out, "--origin", "https://site.example/app/", "--oauth-path", "auth/clients"], { cwd: dir, env });
	const staged = JSON.parse(stdout);
	const doc = JSON.parse(await readFile(join(out, "auth/clients/vibescrobble/client-metadata.json"), "utf8"));
	assert.equal(doc.client_id, "https://site.example/app/auth/clients/vibescrobble/client-metadata.json");
	assert.deepEqual(doc.redirect_uris, ["https://site.example/app/"]);
	assert.equal(staged.clientId, doc.client_id, "one session: it is the page's client");
	assert.equal(JSON.parse(await readFile(join(out, "modules/vibescrobble.json"), "utf8")).session, "vibescrobble");
	assert.deepEqual(await readFile(join(out, "modules/compose.wasm")), await readFile(join(dir, "modules/compose.wasm")));

	// the lockfile name runs the vendored copy; so does the <handle>/<lib> ref, straight from the registry
	const input = JSON.stringify({ call: { input: { n: 4 } } });
	for (const ref of ["compose", "alice.test/compose"]) {
		const ran = JSON.parse((await atlibs(["run", ref, "even", input], { cwd: dir, env })).stdout);
		assert.deepEqual(ran.result, { allow: true, violations: [] }, ref);
	}
	await assert.rejects(atlibs(["run", "nope", "even"], { cwd: dir, env }), /no module nope/);
	await assert.rejects(atlibs(["add", "bob.test/compose"], { cwd: dir, env }), /resolving bob.test/);
	await assert.rejects(atlibs(["add", "alice.test/missing"], { cwd: dir, env }), /publishes no module or composition named missing/);
});

test("add refuses a blob that does not match its CID, and a pin that does not match", async () => {
	const work = await mkdtemp(join(tmpdir(), "atlibs-cli-bad-"));
	await assert.rejects(atlibs(["add", "alice.test/compose@00000000"], { cwd: work, env }), /pin 00000000 is too short/);
	await assert.rejects(atlibs(["add", `alice.test/compose@${"0".repeat(32)}`], { cwd: work, env }), /pin 0{32} not published/);
	const { blob } = await publish("compose");
	const good = blobs.get(blob);
	blobs.set(blob, Buffer.from("tampered"));
	try {
		await assert.rejects(atlibs(["add", "alice.test/compose"], { cwd: work, env }), /blob does not match its CID/);
	} finally {
		blobs.set(blob, good);
	}
});

test("serve: the index, a run over HTTP, and client metadata on the request's origin", async () => {
	const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } });
	try {
		const url = await new Promise((resolve, reject) => {
			let err = "";
			child.stderr.on("data", (chunk) => {
				err += chunk;
				const found = err.match(/atlibs serve: (http:\/\/\S+\/)/);
				if (found) resolve(found[1]);
			});
			child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`)));
		});
		const index = await (await fetch(url)).json();
		assert.ok(index.run.includes("/run/even"), JSON.stringify(index));
		const ran = await (await fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ call: { input: { n: 3 } } }) })).json();
		assert.equal(ran.result.allow, false);
		assert.equal((await fetch(new URL("run/even", url))).status, 405);
		// an entry the module does not export: 404, with the ones it does
		const nope = await fetch(new URL("run/nope", url), { method: "POST", headers: { "content-type": "application/json" }, body: "{}" });
		assert.deepEqual([nope.status, (await nope.json()).message], [404, "no entry nope (caller, even, lookup, spy)"]);
		assert.match(await (await fetch(url, { headers: { accept: "text/html" } })).text(), /<form|forms/);
	} finally {
		child.kill();
	}
	// a lib with a session serves its client metadata
	const vibescrobble = spawn(hostBin(), ["serve", "vibescrobble", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } });
	try {
		const url = await new Promise((resolve, reject) => {
			let err = "";
			vibescrobble.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1]));
			vibescrobble.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`)));
		});
		const doc = await (await fetch(new URL("oauth/vibescrobble/client-metadata.json", url))).json();
		assert.equal(doc.client_id, new URL("oauth/vibescrobble/client-metadata.json", url).href);
	} finally {
		vibescrobble.kill();
	}
});

// One module, many runs at once: concurrent POST /run on one atlibs serve (one thread, runs interleave)
// and concurrent atlibs run processes sharing a state dir. Each answers its own input.
test("concurrent runs of one module: POST /run on one atlibs serve, and parallel atlibs run processes", async () => {
	const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } });
	const even = (n) => JSON.stringify({ call: { input: { n } } });
	const expected = Array.from({ length: 8 }, (_, n) => n % 2 === 0);
	try {
		const url = await new Promise((resolve, reject) => {
			let err = "";
			child.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1]));
			child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`)));
		});
		const ran = await Promise.all(expected.map((_, n) => fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: even(n) }).then((r) => r.json())));
		assert.deepEqual(ran.map((r) => r.result?.allow), expected, JSON.stringify(ran));
	} finally {
		child.kill();
	}
	const runs = await Promise.all(expected.map((_, n) => atlibs(["run", "compose", "even", even(n)], { cwd: dir, env })));
	assert.deepEqual(runs.map((r) => JSON.parse(r.stdout).result.allow), expected);
});

// ATLIBS_WASM_CONCURRENCY=1: the runs queue for the one slot (queued time is not wall time) and all answer;
// a value that is not a positive integer stops atlibs at startup.
test("ATLIBS_WASM_CONCURRENCY: 4 overlapping POST /run with one slot all answer; a bad value is a startup error", async () => {
	const child = spawn(hostBin(), ["serve", "compose", "--port", "0"], { cwd: dir, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env, ATLIBS_WASM_CONCURRENCY: "1" } });
	const expected = [true, false, true, false];
	try {
		const url = await new Promise((resolve, reject) => {
			let err = "";
			child.stderr.on("data", (chunk) => (err += chunk).match(/atlibs serve: (http:\/\/\S+\/)/) && resolve(err.match(/atlibs serve: (http:\/\/\S+\/)/)[1]));
			child.on("exit", (code) => reject(new Error(`serve exited ${code}: ${err}`)));
		});
		const ran = await Promise.all(expected.map((_, n) => fetch(new URL("run/even", url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ call: { input: { n } } }) }).then((r) => r.json())));
		assert.deepEqual(ran.map((r) => r.result?.allow), expected, JSON.stringify(ran));
	} finally {
		child.kill();
	}
	for (const bad of ["0", "-1", "two"]) {
		const stderr = await atlibs(["serve", "compose", "--port", "0"], { cwd: dir, env: { ...env, ATLIBS_WASM_CONCURRENCY: bad } }).then(() => assert.fail(`${bad} started`), (e) => e.stderr);
		assert.equal(stderr, `ATLIBS_WASM_CONCURRENCY must be a positive integer, not "${bad}"\n`);
	}
});

test("run --handler: a composition's declared handler from a wasm module (with a callback), an http endpoint, or an opt-in command", async () => {
	const work = await mkdtemp(join(tmpdir(), "atlibs-cli-handlers-"));
	const comp = join(work, "confirm.json");
	await writeFile(comp, JSON.stringify({
		$type: "com.fedfork.exec.composition",
		name: "confirm",
		instances: [{ name: "ui", module: "compose" }],
		handlers: [{ name: "confirm", description: "someone approves" }],
		edges: [{ from: "ui", nsid: "com.example.compose.confirm", handler: "confirm", reason: "ui asks before acting" }],
	}));
	const NSID = "com.example.compose.confirm";
	const run = (...extra) => atlibs(["run", comp, "ui", "caller", JSON.stringify({ nsid: NSID, input: { n: 1 } }), ...extra], { env });
	await assert.rejects(run(), /handler confirm was not supplied/);

	// wasm: the compose module's lookup entry answers, calling the progress callback ui passed it
	const wasm = JSON.parse((await run("--handler", "confirm=wasm:compose#lookup")).stdout);
	assert.equal(wasm.error, undefined, wasm.error);
	assert.deepEqual(wasm.result, { q: { ok: true, output: { doubled: 2, ack: { seen: { step: 1 } }, caller: { instance: "ui", nsid: NSID, reason: "ui asks before acting" } } }, progress: [{ step: 1 }] });
	assert.deepEqual(wasm.graph.map((g) => [g.from, g.nsid, g.handler ?? g.cap ?? null, g.ok]).at(-1), ["ui", NSID, "confirm", true]);

	// http: the call as JSON, no credentials
	const http = JSON.parse((await run("--handler", `confirm=${base}/handler`)).stdout).result.q.output;
	assert.equal(http.approved, true);
	assert.deepEqual([http.got.handler, http.got.caller, http.got.nsid, http.got.reason, http.got.input.n, http.auth], ["confirm", "ui", NSID, "ui asks before acting", 1, null]);
	assert.match(http.got.input.progress.$cap, /^cap\.[0-9a-f]{32}\.\d+$/, "a cap stays a token");

	// exec: only with --allow-exec; the command gets the call on stdin and prints the answer
	await assert.rejects(run("--handler", "confirm=exec:cat"), /--allow-exec/);
	const exec = JSON.parse((await run("--handler", "confirm=exec:cat", "--allow-exec")).stdout).result.q.output;
	assert.deepEqual([exec.handler, exec.caller, exec.input.n], ["confirm", "ui", 1]);
	const failed = JSON.parse((await run("--handler", "confirm=exec:echo no >&2; exit 2", "--allow-exec")).stdout).result.q;
	assert.deepEqual(failed, { ok: false, error: "handler confirm: echo no >&2; exit 2 exited 2: no" });

	// names are checked against the manifest; a plain module takes no handlers
	await assert.rejects(run("--handler", "nope=exec:cat", "--allow-exec"), /declares no handler nope \(it declares: confirm\)/);
	await assert.rejects(atlibs(["run", "compose", "even", "{}", "--handler", "confirm=exec:cat", "--allow-exec"], { env }), /one module/);
	await assert.rejects(run("--handler", "confirm=http://example.com/x"), /only to localhost/);
});

// One saved session, an expired access token, and several Rust and JS processes sharing the state dir
// at once: exactly one refreshSession (under the state dir's "session" lease), no createSession.
test("concurrent atlibs and cli.mjs processes sharing a state dir refresh the saved session once", async () => {
	const state = join(await mkdtemp(join(tmpdir(), "atlibs-cli-session-")), "state");
	const seed = createNodeContext(contextOptions(state));
	await seed.settings.set("session", { identifier: "alice.test", did: DID, handle: "alice.test", accessJwt: "a1", refreshJwt: "r1", didDoc: { service: [{ id: "#atproto_pds", serviceEndpoint: base }] } });
	seed.close();
	Object.assign(auth, { refreshes: 0, creates: 0 });
	// the account is explicit (flags; the password from the variable --password-env names), never ATPROTO_* env
	const signedIn = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: state, TEST_PW: "pw", TEST_PDS: base };
	const account = ["--app-password", "alice.test", "--password-env", "TEST_PW", "--pds", base];
	const node = `import { contextOptions, createNodeContext, login, sessionStore } from ${JSON.stringify(new URL("../../javascript/src/server/index.js", import.meta.url).href)};
const ctx = createNodeContext(contextOptions(process.env.ATLIBS_SERVER_STATE));
const session = await login({ identifier: "alice.test", password: process.env.TEST_PW, service: process.env.TEST_PDS, ...sessionStore(ctx) });
await session.xrpc("com.atproto.repo.deleteRecord", { input: { repo: session.did, collection: "app.test", rkey: "x" } });`;
	const exit = (child) => {
		let err = "";
		child.stderr.on("data", (chunk) => (err += chunk));
		return new Promise((resolve) => child.on("exit", (code) => resolve(code === 0 ? 0 : err)));
	};
	const runs = await Promise.all([
		...[1, 2, 3].map(() => exit(spawn(hostBin(), ["publish", "delete", "app.test/x", ...account], { env: signedIn }))),
		...[1, 2, 3].map(() => exit(spawn(process.execPath, ["--input-type=module", "-e", node], { env: signedIn }))),
	]);
	assert.deepEqual(runs, Array(6).fill(0));
	assert.deepEqual(auth, { refreshes: 1, creates: 0 });
});

// 3a L1 / 3c M3: the input as `-` (stdin) or `@file` keeps a secret out of argv; a parse error does
// not echo the input; a run that errors exits 1 after printing its result.
test("atlibs run reads the input from - or @file, never echoes it, and exits 1 on a run error", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-cli-input-"));
	const env = { ATLIBS_SERVER_STATE: join(dir, "state") };
	const run = (args, stdin) =>
		new Promise((resolve) => {
			const child = spawn(hostBin(), ["run", "examples/rust/compose", ...args], { cwd: ROOT, env: { PATH: process.env.PATH, HOME: process.env.HOME, ...env } });
			let stdout = "", stderr = "";
			child.stdout.on("data", (d) => (stdout += d));
			child.stderr.on("data", (d) => (stderr += d));
			child.on("close", (code) => resolve({ code, stdout, stderr }));
			child.stdin.end(stdin ?? "");
		});
	const input = JSON.stringify({ call: { input: { n: 4 } }, stdin: "tok" });
	const piped = await run(["even", "-"], input);
	assert.equal(piped.code, 0, piped.stderr);
	assert.deepEqual(JSON.parse(piped.stdout).result, { allow: true, violations: [] });
	await writeFile(join(dir, "in.json"), input);
	assert.deepEqual(JSON.parse((await run(["even", `@${join(dir, "in.json")}`])).stdout).result, { allow: true, violations: [] });
	const bad = await run(["even", "-"], '{"stdin":"secret-token"');
	assert.equal(bad.code, 1);
	assert.match(bad.stderr, /input is not JSON/);
	assert.doesNotMatch(bad.stderr, /secret-token/);
	const failed = await run(["lookup", "{}"]);
	assert.equal(failed.code, 1, "a run error is the exit status");
	assert.equal(JSON.parse(failed.stdout).error, "n is required", "and the result is still printed");
	// an entry it does not export is refused before any run (review 4c): nothing printed, its entries named
	const nope = await run(["no_such_entry", "{}"]);
	assert.deepEqual([nope.code, nope.stdout, nope.stderr], [1, "", "no entry no_such_entry (caller, even, lookup, spy)\n"]);
});

// Each command's own usage on a bad invocation (not the whole help); `run <ref>` with no entry lists the entries.
test("a bad invocation prints that command's usage; run with no entry lists the module's entries", async () => {
	const env = { ATLIBS_SERVER_STATE: join(await mkdtemp(join(tmpdir(), "atlibs-cli-usage-")), "state") };
	const fails = (args) => atlibs(args, { env }).then(() => assert.fail(`${args.join(" ")} succeeded`), (e) => e.stderr);
	const listed = await fails(["run", "examples/rust/compose"]);
	assert.match(listed, /^usage: atlibs run <ref>/);
	assert.match(listed, /compose entries: caller, lookup, even/);
	assert.match(await fails(["run", "examples/rust/compose", "nope"]), /^no entry nope \(caller, even, lookup, spy\)$/m);
	for (const [args, line] of [[["stage", "--orign", "x"], /unknown flag --orign\nusage: atlibs stage /], [["serve"], /^usage: atlibs serve /], [["add"], /^usage: atlibs add /], [["verify"], /^usage: atlibs verify /], [["publish", "--nope=1"], /unknown flag --nope\nusage: atlibs publish \[<name> \.\.\.\][^\n]*\n {7}atlibs publish module [^\n]*\n {7}atlibs publish records/]]) {
		const stderr = await fails(args);
		assert.match(stderr, line, args.join(" "));
		assert.doesNotMatch(stderr, /usage: atlibs <command>/, args.join(" "));
	}
	const { stdout } = await atlibs(["help"], { env });
	for (const piece of ["publish delete", "--check-live", "--password-file <path|->", "--allow-dirty", "--out records/wasm", "@pin|@cid", "--dir .", "--record <file>", "--oauth-path oauth/] [--handler", "--allow-exec", "'<json>' | - | @file", "[--target spa|server]", "a name under examples/rust", "`bridge`: internal"]) assert.ok(stdout.includes(piece), piece);
	assert.ok(!stdout.includes("webrtc-peer"), "webrtc-peer is the atlibs-webrtc repo's own binary");
});

// Regression (verify never matched on macOS: a docker look-alike's `run` returned before the build and exited 0, and
// a host build is not the bytes a Linux rebuild makes, cargo hashing the host triple into every crate): a clone of
// this repo publishes echo to the fixture registry (built in the container) and `atlibs verify` rebuilds the same
// bytes. Needs a container engine (CONTAINER_ENGINE, or podman / docker / Apple's container answering `info`) and
// the network (the rust image, rustup, crates); skipped without an engine. Offline (stand-in engine): registry.rs.
test("atlibs publish module builds in the container and atlibs verify reproduces it", async (t) => {
	const registry = await fixtureRegistry(); // its own: the publish test counts the shared one's writes
	t.after(() => registry.close());
	const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-verify-"));
	const ws = join(tmp, "ws");
	execFileSync("git", ["clone", "-q", new URL(".", ROOT).pathname, ws]);
	await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 });
	const signedIn = { ...registry.env, ...(process.env.CONTAINER_ENGINE && { CONTAINER_ENGINE: process.env.CONTAINER_ENGINE }), XDG_CACHE_HOME: join(tmp, "cache"), ATLIBS_SERVER_STATE: join(tmp, "state") };
	const as = ["--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", registry.base];
	await atlibs(["publish", "records", "records/echo.json", ...as], { cwd: ws, env: signedIn });
	try {
		await atlibs(["publish", "module", "echo", "--out", join(tmp, "out"), ...as], { cwd: ws, env: signedIn });
	} catch (e) {
		if (/no container engine answers/.test(e.stderr)) return t.skip(e.stderr.trim());
		throw e;
	}
	const published = JSON.parse(await readFile(join(tmp, "out", "echo.manifest.json"), "utf8")).output.sha256;
	const { stdout } = await atlibs(["verify", "alice.test/echo"], { env: signedIn });
	assert.deepEqual(JSON.parse(stdout), { built: published, manifest: published, published, reproducible: true });
});

// Another repo publishes the same way: its own cargo workspace (modules/ and a path dependency under crates/),
// nothing of this repo's layout. The sources and the build come from that workspace; the dry run builds in the
// container and its pin is what publish then writes; verify reproduces it. Needs a container engine, as above.
test("atlibs publish from another repo's workspace builds in the container and atlibs verify reproduces it", async (t) => {
	const registry = await fixtureRegistry();
	t.after(() => registry.close());
	const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-other-"));
	const ws = join(tmp, "other");
	const git = (...args) => execFileSync("git", ["-c", "user.name=t", "-c", "user.email=t@t", "-C", ws, ...args]);
	await cp(new URL("rust/src/abi", ROOT), join(ws, "crates/abi"), { recursive: true });
	await cp(new URL("examples/rust/hello", ROOT), join(ws, "modules/hello"), { recursive: true });
	const root = await readFile(new URL("Cargo.toml", ROOT), "utf8");
	await writeFile(join(ws, "Cargo.toml"), root.replace(/^members = .*$/m, 'members = ["modules/*"]').replace('path = "rust/src/abi"', 'path = "crates/abi"'));
	await cp(new URL("Cargo.lock", ROOT), join(ws, "Cargo.lock"));
	execFileSync("cargo", ["metadata", "--offline", "--format-version", "1", "--manifest-path", join(ws, "Cargo.toml")], { stdio: "ignore" }); // prunes the lock to this workspace
	await writeFile(join(ws, ".gitignore"), "target\n");
	git("init", "-q");
	git("remote", "add", "origin", "https://example.test/other");
	git("add", ".");
	git("commit", "-qm", "one");
	await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 });
	const signedIn = { ...registry.env, ...(process.env.CONTAINER_ENGINE && { CONTAINER_ENGINE: process.env.CONTAINER_ENGINE }), XDG_CACHE_HOME: join(tmp, "cache"), ATLIBS_SERVER_STATE: join(tmp, "state") };
	const publish = async (...args) => {
		const { stdout } = await atlibs(["publish", "--dir", "modules", "--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", registry.base, ...args], { cwd: ws, env: signedIn });
		return Object.fromEntries(stdout.trim().split("\n").slice(1).map((line) => line.split(/\s+/)).map(([status, name, pin]) => [name, { status, pin }]));
	};
	let dry;
	try {
		dry = await publish("--dry-run");
	} catch (e) {
		if (/no container engine answers/.test(e.stderr)) return t.skip(e.stderr.trim());
		throw e;
	}
	assert.equal(dry.hello.status, "would-publish");
	const done = await publish();
	assert.deepEqual(done.hello, { status: "published", pin: dry.hello.pin });
	const manifest = JSON.parse(await readFile(join(ws, "records/wasm/hello.manifest.json"), "utf8"));
	assert.deepEqual(manifest.files.map((f) => f.path).sort(), ["Cargo.lock", "Cargo.toml", "crates/abi/Cargo.toml", ...["aturi", "ids", "lib", "out", "records", "repo"].map((m) => `crates/abi/src/${m}.rs`), "modules/hello/Cargo.toml", "modules/hello/src/lib.rs"].sort());
	assert.equal(manifest.origin.repository, "https://example.test/other");
	const { stdout } = await atlibs(["verify", "alice.test/hello"], { env: signedIn });
	assert.equal(JSON.parse(stdout).reproducible, true, stdout);
});

// Regression (`atlibs help | head` exited 101 with println!'s broken-pipe panic): a reader that closed before
// the output is written ends the process quietly, exit 0.
test("a closed stdout (atlibs help | head) exits 0 without a panic", async () => {
	const child = spawn(hostBin(), ["help"], { stdio: ["ignore", "pipe", "pipe"] });
	child.stdout.destroy();
	let stderr = "";
	child.stderr.on("data", (chunk) => (stderr += chunk));
	const code = await new Promise((resolve) => child.on("close", resolve));
	assert.doesNotMatch(stderr, /panicked/);
	assert.equal(code, 0, stderr);
});

// `atlibs publish`: a workspace (a git repo with two modules and a session record) as alice.test. The
// first run publishes all, the next none (the cache, no requests), a committed change only that module;
// --dry-run publishes nothing; dirty sources are skipped; --check-live reports drift; the cache is by DID;
// and what was published is what `atlibs add <handle>/<lib>@<pin>` fetches, signed out.
test("atlibs publish: only what changed, by a cache per DID; dry run, dirty sources, drift", async () => {
	const tmp = await mkdtemp(join(tmpdir(), "atlibs-cli-publish-"));
	const ws = join(tmp, "ws");
	const git = (...args) => execFileSync("git", ["-c", "user.name=t", "-c", "user.email=t@t", "-C", ws, ...args]);
	const release = "target/wasm32-unknown-unknown/release";
	const vibescrobble = JSON.parse(await readFile(new URL("examples/rust/vibescrobble/module.json", ROOT), "utf8"));
	await mkdir(join(ws, release), { recursive: true });
	await mkdir(join(ws, "records"));
	await cp(new URL(`${release}/compose.wasm`, ROOT), join(ws, release, "pa.wasm"));
	await cp(new URL(`${release}/vibescrobble.wasm`, ROOT), join(ws, release, "pb.wasm"));
	await cp(new URL("examples/rust/compose", ROOT), join(ws, "examples/rust/pa"), { recursive: true });
	await cp(new URL("examples/rust/vibescrobble", ROOT), join(ws, "examples/rust/pb"), { recursive: true });
	await writeFile(join(ws, "examples/rust/pb/module.json"), JSON.stringify({ ...vibescrobble, session: "pb" }));
	await cp(new URL("records/vibescrobble.json", ROOT), join(ws, "records/pb.json"));
	await symlink(new URL("lexicons", ROOT).pathname, join(ws, "lexicons"));
	await writeFile(join(ws, ".gitignore"), "target\nlexicons\n");
	git("init", "-q");
	git("remote", "add", "origin", "https://example.test/ws");
	git("add", ".");
	git("commit", "-qm", "one");
	const cache = join(tmp, "cache");
	await writeFile(join(tmp, "password"), "pw\n", { mode: 0o600 });
	const signedIn = { XDG_CACHE_HOME: cache, ATLIBS_SERVER_STATE: join(tmp, "state") };
	const publish = async (...args) => {
		const { stdout } = await atlibs(["publish", "--no-build", "--app-password", "alice.test", "--password-file", join(tmp, "password"), "--pds", base, ...args], { cwd: ws, env: signedIn });
		return Object.fromEntries(stdout.trim().split("\n").slice(1).map((line) => line.split(/\s+/)).map(([status, name, pin, cid, ref]) => [name, { status, pin, cid, ref }]));
	};
	const statuses = (rows) => Object.fromEntries(Object.entries(rows).map(([name, row]) => [name, row.status]));
	const writes = () => seen.filter((p) => /(put|create)Record|uploadBlob/.test(p)).length;

	assert.deepEqual(statuses(await publish("--dry-run")), { "records/pb": "would-publish", pa: "would-publish", pb: "would-publish" });
	assert.equal(writes(), 0, "a dry run publishes nothing");
	const first = await publish();
	assert.deepEqual(statuses(first), { "records/pb": "published", pa: "published", pb: "published" });
	assert.equal(first.pa.ref, "alice.test/pa", "by the --app-password handle, as atlibs add takes it");
	assert.equal(records.get("com.fedfork.exec.wasm.module/pb").value.session.cid, first["records/pb"].cid, "the module names its session record");
	const cached = JSON.parse(await readFile(join(cache, "atlibs", DID, "pa.json"), "utf8"));
	assert.deepEqual([cached.pin, cached.cid, cached.commit], [first.pa.pin, first.pa.cid, git("rev-parse", "HEAD").toString().trim()], "the cache is keyed by DID");

	const requests = seen.length;
	assert.deepEqual(statuses(await publish()), { "records/pb": "unchanged", pa: "unchanged", pb: "unchanged" });
	assert.equal(seen.length, requests, "unchanged by the cache: no requests");

	const pa = JSON.parse(await readFile(join(ws, "examples/rust/pa/module.json"), "utf8"));
	await writeFile(join(ws, "examples/rust/pa/module.json"), JSON.stringify({ ...pa, calls: [...(pa.calls ?? []), "com.atproto.repo.getRecord"] }));
	git("commit", "-qam", "two");
	const second = await publish();
	assert.deepEqual(statuses(second), { "records/pb": "unchanged", pa: "published", pb: "unchanged" });
	assert.notEqual(second.pa.pin, first.pa.pin, "a changed grant is a new pin");

	// signed out: add <handle>/<lib>@<pin> fetches exactly what was published
	const out = await mkdtemp(join(tmpdir(), "atlibs-cli-publish-add-"));
	await atlibs(["add", `alice.test/pa@${second.pa.pin}`], { cwd: out, env });
	assert.equal(JSON.parse(await readFile(join(out, "atlibs.lock.json"), "utf8")).modules.pa.pin, second.pa.pin);
	// ... and the first pin still the first version: a republish leaves it in place
	await atlibs(["add", `alice.test/pa@${first.pa.pin}`], { cwd: out, env });
	assert.equal(JSON.parse(await readFile(join(out, "atlibs.lock.json"), "utf8")).modules.pa.pin, first.pa.pin);
	await assert.rejects(atlibs(["add", `alice.test/pa@${"0".repeat(32)}`], { cwd: out, env }), /pin 0{32} not published/);

	// dirty sources: skipped, nothing written, exit 0
	await writeFile(join(ws, "examples/rust/pb/module.json"), JSON.stringify({ ...vibescrobble, session: "pb", description: "edited" }));
	const before = writes();
	assert.deepEqual(statuses(await publish("pb")), { "records/pb": "unchanged", pb: "skipped:dirty" });
	assert.equal(writes(), before);
	git("checkout", "--", "examples/rust/pb/module.json");

	// drift: the live record is not the one this cache published
	const live = records.get("com.fedfork.exec.wasm.module/pa").value;
	const moved = put("com.fedfork.exec.wasm.module", "pa", { ...live, createdAt: "2020-01-01T00:00:00.000Z" });
	const drift = await publish("--check-live");
	assert.deepEqual(statuses(drift), { "records/pb": "unchanged", pa: "drift", pb: "unchanged" });
	assert.equal(drift.pa.cid, moved.cid, "the cache now holds the live record");
	assert.equal(writes(), before, "same content: reported, not republished");
});

// Regression (atproto-crates linked a second rustls provider; every wss subscribe then panicked until main's
// atlibs_host::tls()): the built binary runs echo's `watch` over wss:// against a local TLS server (minimal
// WebSocket upgrade, one text frame), trusting the fixture cert through the test-only ATLIBS_EXTRA_CA.
test("atlibs run subscribes over wss:// without a rustls CryptoProvider panic", async () => {
	const fixtures = new URL("tests/fixtures/", import.meta.url);
	const tls = createHttpsServer({ cert: await readFile(new URL("localhost.pem", fixtures)), key: await readFile(new URL("localhost.key.pem", fixtures)) });
	tls.on("upgrade", (req, socket) => {
		const accept = createHash("sha1").update(`${req.headers["sec-websocket-key"]}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest("base64");
		socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
		const payload = Buffer.from(JSON.stringify({ $type: "com.atproto.sync.subscribeRepos#commit", seq: 1 }));
		socket.write(Buffer.concat([Buffer.from([0x81, payload.length]), payload])); // FIN text frame, < 126 bytes
		socket.on("data", () => {}).on("error", () => {}); // the client's close (the module cancels at its limit)
	});
	await new Promise((resolve) => tls.listen(0, "127.0.0.1", resolve));
	try {
		const service = `https://127.0.0.1:${tls.address().port}`;
		const { stdout, stderr } = await atlibs(["run", "examples/rust/echo", "watch", JSON.stringify({ did: DID, service, limit: 1, seconds: 10 })], { env: { ATLIBS_EXTRA_CA: new URL("localhost.pem", fixtures).pathname } });
		assert.doesNotMatch(stderr, /panicked|CryptoProvider/);
		const { result } = JSON.parse(stdout);
		assert.deepEqual([result.count, result.last, result.stopped], [1, { $type: "com.atproto.sync.subscribeRepos#commit", seq: 1 }, "limit"], stderr);
	} finally {
		tls.close();
		tls.closeAllConnections?.();
	}
});