import { test } from "node:test"; import assert from "node:assert/strict"; import { atirc, channelOf, normalizeChannel, sessionRepo } from "./index.js"; const ME = "did:plc:alice"; const PEER = "computer.socialweb.peer"; const until = async (check) => { for (let i = 0; i < 200; i++, await new Promise((r) => setTimeout(r, 5))) if (check()) return; throw new Error("timed out"); }; // A repo of peer records and a WebrtcHost stand-in: each peer is a handler of the channel's lines. function world(records, handlers) { const repo = { did: ME, getRecord: async ({ repo: did, collection, rkey }) => { const value = records[`${did}/${collection}/${rkey}`]; return value ? { value, cid: "bafy" } : null; }, putRecord: async () => ({}), deleteRecord: async () => ({}), }; const host = () => ({ close() {}, async call({ params: { receipt } }) { const subhandle = receipt.uri.split("/").pop(); const handler = handlers[subhandle]; if (!handler) throw new Error(`${subhandle} does not answer`); const queue = []; let wake; let closed = false; const conn = { sent: [], push: (line) => (queue.push(`${JSON.stringify(line)}\n`), wake?.()), end: () => ((closed = true), wake?.()), }; return { close: conn.end, send: async (data) => { const line = JSON.parse(data); conn.sent.push(line); handler(line, conn); }, async *[Symbol.asyncIterator]() { for (;;) { while (!queue.length && !closed) await new Promise((r) => (wake = r)); if (queue.length) yield queue.shift(); else return; } }, }; }, }); return { repo, host }; } const record = (channels) => ({ webrtc: ME, iroh: "x", channels }); test("channel names follow atirc", () => { assert.equal(channelOf("alice.test"), "#alice_test"); assert.equal(channelOf("Bot.Agent.test"), "#bot_agent_test"); assert.equal(channelOf(""), "#atirc"); assert.equal(normalizeChannel(" Alice_Test "), "#alice_test"); }); test("through the owner's bridge: attributed lines, members, post, leave", async () => { let conn; const { repo, host } = world({ [`${ME}/${PEER}/alice.test`]: record(["#alice_test"]) }, { "alice.test": (line, c) => { conn = c; if (line.hello) c.push({ event: "hello", did: ME, subhandle: "alice.test", bridge: true, epoch: 9, since: 0 }); }, }); const chat = atirc({ repo, handle: "alice.test", host }); const messages = []; const presence = []; chat.onMessage((m) => messages.push(m)); chat.onPresence((p) => presence.push(`${p.event} ${p.identity}`)); await chat.join("#alice_test"); assert.equal(conn.sent[0].channel, "#alice_test"); conn.push({ members: [{ did: "did:plc:agent", subhandle: "bot.agent.test" }] }); conn.push({ seq: 1, text: "hi", ts: "t", did: "did:plc:agent", subhandle: "bot.agent.test" }); conn.push({ seq: 1, text: "replayed twice", ts: "t" }); await until(() => messages.length === 1); assert.deepEqual(messages[0], { channel: "#alice_test", identity: "bot.agent.test", nick: "bot", did: "did:plc:agent", text: "hi", ts: "t", seq: 1, via: "alice.test" }); const { me, channels, peers } = await chat.peers(); assert.deepEqual([me, channels, peers.map((p) => p.subhandle)], ["alice.test", ["#alice_test"], ["bot.agent.test"]]); await chat.post("alice_test", "hello agents"); assert.deepEqual([conn.sent[1].seq, conn.sent[1].text], [1, "hello agents"]); assert.deepEqual([messages.at(-1).self, messages.at(-1).identity], [true, "alice.test"], "posted as the bridge node"); chat.leave("#alice_test"); assert.deepEqual(presence, ["join alice.test", "join bot.agent.test", "leave bot.agent.test", "leave alice.test"]); assert.deepEqual((await chat.peers()).channels, []); await assert.rejects(chat.post("#alice_test", "gone"), /not in #alice_test/); }); test("without a bridge: direct peers that carry the channel speak only for themselves", async () => { const conns = {}; const reply = (subhandle) => (line, c) => { conns[subhandle] = c; if (line.hello) c.push({ event: "hello", did: "did:plc:agent", subhandle, epoch: 1, peers: [{ did: "did:plc:agent", subhandle: "two.agent.test" }] }); }; const { repo, host } = world({ [`did:plc:agent/${PEER}/one.agent.test`]: record(["#alice_test"]), [`did:plc:agent/${PEER}/two.agent.test`]: record(["#alice_test"]), [`did:plc:agent/${PEER}/elsewhere.agent.test`]: record(["#other"]), }, { "one.agent.test": reply("one.agent.test"), "two.agent.test": reply("two.agent.test") }); const chat = atirc({ repo, handle: "alice.test", host, peers: [{ did: "did:plc:agent", subhandle: "one.agent.test" }, { did: "did:plc:agent", subhandle: "elsewhere.agent.test" }], }); const messages = []; chat.onMessage((m) => messages.push(m)); await chat.join("#alice_test"); // two.agent.test came from one's peer list. await until(() => conns["two.agent.test"]); conns["one.agent.test"].push({ seq: 4, text: "mine", did: "did:plc:alice", subhandle: "alice.test" }); await until(() => messages.length === 1); assert.equal(messages[0].identity, "one.agent.test", "a direct peer cannot speak as another"); await chat.post("#alice_test", "both"); assert.equal(conns["one.agent.test"].sent.at(-1).text, "both"); assert.equal(conns["two.agent.test"].sent.at(-1).text, "both"); chat.close(); }); test("a dropped link redials with since, and nobody to dial fails the join", async () => { const hellos = []; let conn; const { repo, host } = world({ [`${ME}/${PEER}/alice.test`]: record(["#alice_test"]) }, { "alice.test": (line, c) => { conn = c; if (line.hello) hellos.push(line), c.push({ event: "hello", bridge: true, epoch: 9 }); }, }); const chat = atirc({ repo, handle: "alice.test", host, retryMs: 10 }); await chat.join("#alice_test"); conn.push({ seq: 3, text: "x" }); await new Promise((r) => setTimeout(r, 20)); conn.end(); await until(() => hellos.length === 2); assert.deepEqual([hellos[1].since, hellos[1].seen_epoch], [3, 9]); chat.close(); await assert.rejects(atirc({ repo, handle: "bob.test", host, bridge: false }).join("#nobody"), /no peer of #nobody takes WebRTC/); }); test("guests: the first VM node whose tunnel opens, as a bridge", async () => { let conn; const opened = []; const host = ({ events, signal }) => ({ close() {}, async call({ params: { receipt, tag } }) { opened.push(`${receipt.uri} ${tag} ${events.name} ${signal.name}`); if (receipt.uri.endsWith("gone")) throw new Error("the guest is gone"); return world({}, { vm: (line, c) => { conn = c; if (line.hello) c.push({ event: "hello", did: ME, subhandle: "you-bot.alice.test", bridge: true, epoch: 2 }); }, }).host().call({ params: { receipt: { uri: "at://x/y/vm" } } }); }, }); const events = async function events() {}, signal = async function signal() {}; const chat = atirc({ guests: [{ receipt: { uri: "at://b/accept/gone" }, events, signal }, { receipt: { uri: "at://b/accept/new" }, events, signal }], host }); const messages = []; chat.onMessage((m) => messages.push(m)); await chat.join("#bot_alice_test"); assert.deepEqual(opened, ["at://b/accept/gone atirc events signal", "at://b/accept/new atirc events signal"]); conn.push({ seq: 7, text: "hello", ts: "t", did: "did:plc:alice", subhandle: "bot.alice.test" }); await until(() => messages.length === 1); assert.equal(messages[0].nick, "bot"); await chat.post("#bot_alice_test", "hi bot"); assert.equal(conn.sent.at(-1).text, "hi bot"); assert.equal(messages.at(-1).identity, "you-bot.alice.test"); assert.equal((await chat.peers()).me, "you-bot.alice.test"); chat.close(); assert.throws(() => atirc({}), /needs guests, or repo/); }); test("sessionRepo: public reads at the record's PDS, writes through the session", async () => { const calls = []; const fetch = async (url) => { calls.push(String(url)); if (String(url) === "https://plc.directory/did:plc:agent") return Response.json({ service: [{ id: "#atproto_pds", serviceEndpoint: "https://pds.example" }] }); if (String(url).includes("rkey=missing")) return Response.json({ error: "RecordNotFound" }, { status: 400 }); return Response.json({ value: { webrtc: "did:plc:agent" }, cid: "c" }); }; const written = []; const session = { did: ME, fetchHandler: async (path, init) => (written.push([path, JSON.parse(init.body)]), Response.json({})) }; const repo = sessionRepo(session, { fetch }); assert.equal(repo.did, ME); assert.equal((await repo.getRecord({ repo: "did:plc:agent", collection: PEER, rkey: "bot" })).cid, "c"); assert.equal(await repo.getRecord({ repo: "did:plc:agent", collection: PEER, rkey: "missing" }), null); assert.equal(calls.filter((u) => u.startsWith("https://plc.directory")).length, 1, "the PDS is cached"); assert.ok(calls[1].startsWith("https://pds.example/xrpc/com.atproto.repo.getRecord?repo=did%3Aplc%3Aagent")); await repo.putRecord({ collection: "c", rkey: "k", record: {} }); assert.deepEqual(written[0], ["/xrpc/com.atproto.repo.putRecord", { repo: ME, collection: "c", rkey: "k", record: {} }]); });