johnandersen777.bsky.social/atlibs-webrtc

Switch repository

Loading…

// WebrtcHost's events/signal over exact-key atproto records, for mesh peers (atirc's daemon, and any peer
// whose computer.socialweb.peer record names a `webrtc` DID). No listRecords: every read is one getRecord.
//
//   computer.socialweb.peer/<subhandle>           in the peer's repo: {webrtc: <answerer DID>, ...}
//   computer.socialweb.ingress.listen/<subhandle> in the answerer's repo: {key: age1..., fingerprint}
//   computer.socialweb.ingress.offer/<answerer DID>.<tid>  written here: {ciphertext, subhandle, to}
//   computer.socialweb.ingress.answer/<our DID>.<tid>      the answerer's: {ciphertext}
//
// The receipt WebrtcHost carries is the peer record's strong ref {uri, cid}. `repo` is the page's PDS access:
//   { did, getRecord({repo, collection, rkey}) -> {value, cid} | null, putRecord({collection, rkey, record}),
//     deleteRecord({collection, rkey}) }
// `subhandle` is the page's own identity (the account handle or <nick>.<handle>).

export const PEER = "computer.socialweb.peer";
export const LISTEN = "computer.socialweb.ingress.listen";
export const OFFER = "computer.socialweb.ingress.offer";
export const ANSWER = "computer.socialweb.ingress.answer";
export const MESH_LABEL = "atirc/mesh/1";

const B32 = "234567abcdefghijklmnopqrstuvwxyz";
// An atproto TID: microseconds since the epoch and a random clock id, base32-sortable.
export function tid(now = Date.now()) {
	let n = (BigInt(now) * 1000n) << 10n | BigInt(Math.floor(Math.random() * 1024));
	let out = "";
	for (let i = 0; i < 13; i++, n >>= 5n) out = B32[Number(n & 31n)] + out;
	return out;
}

const parsePeer = (uri) => {
	const m = /^at:\/\/(did:[^/]+)\/computer\.socialweb\.peer\/([^/]+)$/.exec(uri ?? "");
	if (!m) throw new Error(`not a ${PEER} record uri: ${uri}`);
	return { did: m[1], subhandle: m[2] };
};

// The strong ref of a peer's binding, for WebrtcHost's receipt.
export async function peerReceipt(repo, did, subhandle) {
	const found = await repo.getRecord({ repo: did, collection: PEER, rkey: subhandle });
	if (!found) throw new Error(`${subhandle} has no ${PEER} record`);
	return { uri: `at://${did}/${PEER}/${subhandle}`, cid: found.cid };
}

export function recordSignals({ repo, subhandle }) {
	const sessions = new Map(); // peer uri -> { answerer, tid, offered }
	const answererOf = async (uri) => {
		// a market endpoint's listen record (kcp bidders): that DID answers, the rkey is its listen name
		const listen = /^at:\/\/(did:[^/]+)\/computer\.socialweb\.ingress\.listen\/([^/]+)$/.exec(uri ?? "");
		if (listen) return { answerer: listen[1], peer: listen[2] };
		const { did, subhandle: peer } = parsePeer(uri);
		const record = await repo.getRecord({ repo: did, collection: PEER, rkey: peer });
		const answerer = record?.value?.webrtc;
		if (typeof answerer !== "string" || !answerer.startsWith("did:")) throw new Error(`${peer} does not take WebRTC offers`);
		return { answerer, peer };
	};
	return {
		async events({ receipt, tag }) {
			const { answerer, peer } = await answererOf(receipt.uri);
			const listen = await repo.getRecord({ repo: answerer, collection: LISTEN, rkey: peer });
			const events = [];
			if (listen?.value?.key && listen?.value?.fingerprint) {
				events.push({ kind: "guest.publicKey", value: { kind: "age-x25519", tag, key: listen.value.key } });
				events.push({ kind: "vm.onNetwork", value: { kind: "webrtc", tag, address: listen.value.fingerprint } });
			}
			const session = sessions.get(receipt.uri);
			if (session) {
				const answer = await repo.getRecord({ repo: answerer, collection: ANSWER, rkey: `${repo.did}.${session.tid}` });
				if (answer?.value?.ciphertext) {
					events.push({ kind: "guest.signalWebRTC", value: { kind: "answer", tag, ciphertext: answer.value.ciphertext } });
					// Answered: the offer has done its job.
					repo.deleteRecord({ collection: OFFER, rkey: `${answerer}.${session.tid}` }).catch(() => undefined);
				}
			}
			return events;
		},
		async signal({ receipt, offer }) {
			const { answerer, peer } = await answererOf(receipt.uri);
			const session = { answerer, tid: tid() };
			sessions.set(receipt.uri, session);
			await repo.putRecord({
				collection: OFFER,
				rkey: `${answerer}.${session.tid}`,
				record: { $type: OFFER, ciphertext: offer, subhandle, to: peer, createdAt: new Date().toISOString() },
			});
		},
	};
}

// A mesh channel over WebrtcHost's datachannel (tag MESH_LABEL): the JSON lines atirc speaks over iroh.
// Returns { reply, send(text), messages (async iterator of {seq, text, ts}), close }.
export async function meshChannel(host, receipt, { channel, since = 0, seenEpoch = 0, epoch = Date.now() } = {}) {
	const ch = await host.call({ subscribe: true, params: { receipt, tag: MESH_LABEL } });
	const iterator = lines(ch);
	await ch.send(`${JSON.stringify({ hello: 1, kind: "channel", channel, since, seen_epoch: seenEpoch, epoch })}\n`);
	const first = await iterator.next();
	if (first.done) throw new Error("the peer closed the channel during hello");
	const reply = JSON.parse(first.value);
	if (reply.event !== "hello") throw new Error(`the peer refused: ${reply.message ?? first.value}`);
	let seq = 0;
	return {
		reply,
		send: (text) => ch.send(`${JSON.stringify({ seq: ++seq, text, ts: new Date().toISOString() })}\n`),
		messages: (async function* () {
			for await (const line of iterator) yield JSON.parse(line);
		})(),
		close: () => ch.close(),
	};
}

async function* lines(ch) {
	let buffer = "";
	for await (const chunk of ch) {
		buffer += chunk;
		let at;
		while ((at = buffer.indexOf("\n")) >= 0) {
			const line = buffer.slice(0, at);
			buffer = buffer.slice(at + 1);
			if (line.trim()) yield line;
		}
	}
}