johnandersen777.bsky.social/atlibs-atirc

Switch repository

Loading…

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: {} }]);
});