johnandersen777.bsky.social/atlibs-atirc
atirc channels in a browser page: WebRTC to atirc mesh nodes, no server, no token
0
0
// ATIRC_BIN=<atirc> WEBRTC_PEER=<atlibs-webrtc's webrtc-peer> deno run -A test/loopback.ts
//
// Headless Chrome (test/web/loopback.html) <-> three local `atirc mesh run` nodes over WebRTC and iroh,
// loopback only. This harness is their fake PDS (getRecord/putRecord/deleteRecord over the directory the
// nodes read through ATIRC_MESH_DIRECTORY) and fake Jetstream (a putRecord is a commit event).
// alice.test the owner's own node: the page's bridge
// vm.alice.test a VM agent: record written for it (owner's channel only), no WebRTC, no publish
// bot.agent.test an agent account's node with WebRTC: the page's direct peer
// and a guest: vm.alice.test's --local-listen behind webrtc-peer (the guest agent's stand-in, tag atirc).
const ROOT = new URL("..", import.meta.url).pathname;
const BIN = Deno.env.get("ATIRC_BIN") ?? "atirc";
const WEBRTC_PEER = Deno.env.get("WEBRTC_PEER") ?? "webrtc-peer";
const CHROME = Deno.env.get("CHROME") ?? (Deno.build.os === "darwin" ? "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" : "google-chrome");
const OWNER = "alice.test", AGENT = "agent.test";
const did = (handle: string) => `did:plc:${handle.replaceAll(".", "")}`;
const tmp = await Deno.makeTempDir({ prefix: "atlibs-atirc-" });
const dir = `${tmp}/directory`;
const write = async (path: string, text: string) => {
await Deno.mkdir(path.slice(0, path.lastIndexOf("/")), { recursive: true });
await Deno.writeTextFile(path, text);
};
const recordPath = (repo: string, collection: string, rkey: string) => {
for (const part of [repo, collection, rkey]) if (!part || part.includes("/") || part.startsWith(".")) throw new Error("bad path");
return `${dir}/repo/${repo}/${collection}/${rkey}.json`;
};
for (const handle of [OWNER, AGENT]) {
await write(`${dir}/did/${did(handle)}.json`, JSON.stringify({
id: did(handle),
alsoKnownAs: [`at://${handle}`],
verificationMethod: [],
service: [{ id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "https://pds.example.com" }],
}));
await write(`${dir}/handle/${handle}`, did(handle));
}
await write(recordPath(did(AGENT), "com.fedfork.irc.agentOperator", "self"), JSON.stringify({ owner: did(OWNER) }));
await write(recordPath(did(OWNER), "com.fedfork.irc.agents", did(AGENT)), JSON.stringify({ agent: did(AGENT) }));
const sockets = new Set<WebSocket>();
let resolveResult: (value: unknown) => void;
const result = new Promise((resolve) => (resolveResult = resolve));
const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
const steps: Record<string, () => Promise<void>> = {};
const server = Deno.serve({ hostname: "127.0.0.1", port: 0, onListen: () => {} }, async (req) => {
const url = new URL(req.url);
if (url.pathname === "/subscribe") {
const { socket, response } = Deno.upgradeWebSocket(req);
socket.onopen = () => sockets.add(socket);
socket.onclose = () => sockets.delete(socket);
return response;
}
if (url.pathname === "/xrpc/com.atproto.repo.getRecord") {
const [repo, collection, rkey] = ["repo", "collection", "rkey"].map((k) => url.searchParams.get(k) ?? "");
try {
const text = await Deno.readTextFile(recordPath(repo, collection, rkey));
return json({ uri: `at://${repo}/${collection}/${rkey}`, cid: `local${text.length}`, value: JSON.parse(text) });
} catch {
return json({ error: "RecordNotFound" }, 400);
}
}
if (url.pathname === "/xrpc/com.atproto.repo.putRecord" || url.pathname === "/xrpc/com.atproto.repo.deleteRecord") {
const { repo, collection, rkey, record } = await req.json();
const path = recordPath(repo, collection, rkey);
const create = url.pathname.endsWith("putRecord");
if (create) await write(path, JSON.stringify(record));
else await Deno.remove(path).catch(() => {});
const event = { did: repo, time_us: Date.now() * 1000, kind: "commit", commit: { operation: create ? "create" : "delete", collection, rkey, record } };
for (const socket of sockets) socket.send(JSON.stringify(event));
return json({});
}
if (url.pathname.startsWith("/step/")) {
const step = steps[url.pathname.slice(6)];
if (!step) return new Response("no such step", { status: 404 });
try {
await step();
return json({ ok: true });
} catch (error) {
return new Response(String(error), { status: 500 });
}
}
if (url.pathname === "/log") {
console.error(`page: ${await req.text()}`);
return json({});
}
if (url.pathname === "/result") {
resolveResult(await req.json());
return json({});
}
if (url.pathname.includes("..") || !/^\/(js|node_modules|test\/web)\//.test(url.pathname)) return new Response("not found", { status: 404 });
try {
const type = url.pathname.endsWith(".html") ? "text/html" : "text/javascript";
return new Response(await Deno.readFile(`${ROOT}${url.pathname.slice(1)}`), { headers: { "content-type": type } });
} catch {
return new Response("not found", { status: 404 });
}
});
const base = `http://127.0.0.1:${server.addr.port}`;
const env = (name: string, handle: string, nick?: string) => ({
ATIRC_CONFIG_DIR: `${tmp}/${name}`,
ATIRC_MESH_DIRECTORY: dir,
ATIRC_IROH_LOOPBACK: "1",
ATIRC_JETSTREAM: `ws://127.0.0.1:${server.addr.port}/subscribe`,
ATIRC_RELAY: "off",
ATP_USERNAME: handle,
...(nick ? { ATIRC_NICK: nick } : {}),
RUST_LOG: "info",
});
const children: Deno.ChildProcess[] = [];
const start = (args: string[], envs: Record<string, string>, log: string) => {
const child = new Deno.Command(args[0], { args: args.slice(1), env: envs, stdin: "null", stdout: "piped", stderr: "piped" }).spawn();
children.push(child);
const file = Deno.openSync(`${tmp}/${log}`, { write: true, create: true, truncate: true });
(async () => {
for await (const chunk of child.stderr) file.writeSync(chunk);
})().catch(() => {});
return child;
};
const run = async (args: string[], envs: Record<string, string>) => {
const out = await new Deno.Command(BIN, { args, env: envs, stdout: "piped", stderr: "piped" }).output();
if (!out.success) throw new Error(`atirc ${args.join(" ")}: ${new TextDecoder().decode(out.stderr)}`);
return new TextDecoder().decode(out.stdout);
};
const until = async (check: () => Promise<boolean> | boolean, what: string, ms = 30000) => {
for (const end = Date.now() + ms; Date.now() < end; await new Promise((r) => setTimeout(r, 100))) if (await check()) return;
throw new Error(`timed out waiting for ${what}`);
};
const exists = (path: string) => Deno.stat(path).then(() => true, () => false);
// `atirc mesh sub --cursor 0` of a node, parsed.
const subscribe = (envs: Record<string, string>, log: string) => {
const lines: Record<string, string>[] = [];
const sub = start([BIN, "mesh", "sub", "--cursor", "0"], envs, log);
(async () => {
let buffer = "";
for await (const chunk of sub.stdout.pipeThrough(new TextDecoderStream())) {
buffer += chunk;
const parts = buffer.split("\n");
buffer = parts.pop() ?? "";
for (const part of parts) if (part.trim()) lines.push(JSON.parse(part));
}
})().catch(() => {});
return (channel: string, identity: string, text: string) => lines.some((l) => l.channel === channel && l.identity === identity && l.text === text);
};
let failed = false;
try {
const alice = env("alice", OWNER), vm = env("vm", OWNER, "vm"), bot = env("bot", AGENT, "bot");
// Loopback tickets carry the port: the others start once alice has republished hers.
start([BIN, "mesh", "run"], alice, "alice.log");
await until(() => exists(`${tmp}/alice/mesh.json`), "the alice node");
start([BIN, "mesh", "run"], bot, "bot.log");
// What a requester writes for a VM agent: its key's record, the owner's channel only.
await run(["peer", "publish", "-c", "#alice_test"], vm);
const local = Deno.listen({ hostname: "127.0.0.1", port: 0 });
const localPort = (local.addr as Deno.NetAddr).port;
local.close();
start([BIN, "mesh", "run", "--no-publish", "--no-webrtc", "--local-listen", `127.0.0.1:${localPort}`], vm, "vm.log");
// The guest agent's stand-in: datachannel tag atirc -> the vm node's local listener.
const peer = start([WEBRTC_PEER, "--tunnel", `atirc=${localPort}`, "--origin", base], {}, "webrtc-peer.log");
const signal = await (async () => {
let out = "";
for await (const chunk of peer.stdout.pipeThrough(new TextDecoderStream())) {
out += chunk;
const line = out.split("\n").find((l) => l.startsWith("{"));
if (line) return JSON.parse(line).signal as string;
}
throw new Error("webrtc-peer printed no signal");
})();
for (const name of ["alice", "bot", "vm"]) await until(() => exists(`${tmp}/${name}/mesh.json`), `the ${name} node`);
await until(() => sockets.size >= 2, "the nodes' Jetstream connections");
const vmHeard = subscribe(vm, "vm-sub.log"), botHeard = subscribe(bot, "bot-sub.log");
await until(async () => /vm\.alice\.test/.test(await run(["mesh", "peers"], alice)), "alice linked to the vm");
await run(["mesh", "pub", "vm online"], vm);
steps["vm-heard"] = () => until(() => vmHeard("#alice_test", OWNER, "hi from the page"), "the page's line on the vm");
steps["vm-own"] = async () => void (await run(["mesh", "pub", "-c", "#vm_alice_test", "vm own channel"], vm));
steps["vm-own-heard"] = () => until(() => vmHeard("#vm_alice_test", OWNER, "page on the vm channel"), "the page's line on the vm's channel");
steps["bot-say"] = async () => void (await run(["mesh", "pub", "bot says hi"], bot));
steps["bot-heard"] = () => until(() => botHeard("#alice_test", "web2.alice.test", "page direct to bot"), "the direct page's line on the bot");
steps["bot-to-guest"] = async () => void (await run(["mesh", "pub", "bot to the guest"], bot));
steps["guest-heard"] = () => until(() => botHeard("#alice_test", "vm.alice.test", "from the guest page"), "the guest page's line on the bot, as the vm node");
const page = `${base}/test/web/loopback.html?${new URLSearchParams({ did: did(OWNER), agent: did(AGENT), signal })}`;
start([CHROME, "--headless=new", `--user-data-dir=${tmp}/chrome`, "--no-first-run", "--no-default-browser-check", "--disable-gpu",
"--allow-loopback-in-peer-connection", "--disable-features=WebRtcHideLocalIpsWithMdns", page], {}, "chrome.log");
const outcome = await Promise.race([result, new Promise((_, reject) => setTimeout(() => reject(new Error("the page reported nothing in 240 s")), 240000))]) as
{ ok: boolean; error?: string; peersA?: string[] };
if (!outcome.ok) throw new Error(`page: ${outcome.error}`);
console.log(`ok atlibs-atirc loopback: bridge (alice.test -> vm.alice.test without WebRTC, both channels) and direct (bot.agent.test), both ways; guest (vm.alice.test --local-listen via webrtc-peer) both ways; A's peers ${outcome.peersA?.join(",")}`);
} catch (error) {
failed = true;
console.error(`FAIL atlibs-atirc loopback: ${error instanceof Error ? error.message : error}`);
for (const log of ["alice.log", "vm.log", "bot.log", "webrtc-peer.log", "chrome.log"]) console.error(`--- ${log}\n${(await Deno.readTextFile(`${tmp}/${log}`).catch(() => "")).slice(-6000)}`);
} finally {
for (const child of children) {
try {
child.kill("SIGTERM");
} catch { /* exited */ }
}
await server.shutdown();
if (Deno.env.get("KEEP")) console.error(`kept ${tmp}`);
else await Deno.remove(tmp, { recursive: true }).catch(() => {});
}
Deno.exit(failed ? 1 : 0);