// 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/ in the peer's repo: {webrtc: , ...} // computer.socialweb.ingress.listen/ in the answerer's repo: {key: age1..., fingerprint} // computer.socialweb.ingress.offer/. written here: {ciphertext, subhandle, to} // computer.socialweb.ingress.answer/. 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 .). 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; } } }