johnandersen777.bsky.social/atlibs-socialweb-computer-app
app.socialweb.computer: socialweb.computer in one page, one atlibs composition
0
0
// atirc channels from a browser (or Node with an RTCPeerConnection), over WebRTC datachannels speaking atirc's
// `atirc/mesh/1` channel lines to `atirc mesh run` nodes. No atirc server, no token. Two ways in:
//
// - guests: your own node on a VM you rent (`atirc mesh run --local-listen 127.0.0.1:P`), through the contract's
// WebRTC tunnel {tag: "atirc", port: P}: signalled by the guest agent (contract events), so only you reach it.
// - mesh: your account's nodes by their computer.socialweb.peer records (exact-key record signalling). The
// account's own node (identity = its handle) is a bridge; without it, each WebRTC peer of the channel.
//
// A bridge (either way) carries the whole channel: every line attributed, member lists, channels joined on
// demand, your lines posted as that node.
import { MESH_LABEL, PEER, WebrtcHost, peerReceipt, recordSignals } from "@publicdomainrelay/atlibs-webrtc";
// atirc's channel for a handle or identity: `#` + lowercase, `.` `,` NUL and whitespace as `_`, 63 chars.
export function channelOf(name) {
const channel = `#${[...String(name)].slice(0, 63).map((c) => (/[.,\0\s]/.test(c) ? "_" : c)).join("")}`.toLowerCase();
return channel === "#" ? "#atirc" : channel;
}
export function normalizeChannel(channel) {
const c = String(channel).trim().toLowerCase();
return c.startsWith("#") ? c : `#${c}`;
}
export const GUEST_TAG = "atirc";
const keyOf = ({ did, subhandle }) => `${did} ${subhandle}`;
const nickOf = (identity) => String(identity ?? "").split(".")[0];
const RETRY_MS = 5000;
const MAX_RETRY_MS = 60000;
const BRIDGE_WAIT_MS = 30000;
// One channel stream over a datachannel subscription: hello, the node's reply, then JSON lines.
export async function channelOver(ch, { channel, since = 0, seenEpoch = 0, epoch = Date.now() }) {
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 node closed the channel during hello");
const reply = JSON.parse(first.value);
if (reply.event !== "hello") throw new Error(`the node 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;
}
}
}
// guests: [{ receipt: {uri, cid}, events, signal, tag? }]: your VMs' nodes, newest first (events/signal: the
// requester's watch and signal for that receipt, as atlibs-webrtc's WebrtcHost takes them)
// repo: { did, getRecord({repo, collection, rkey}) -> {value, cid} | null, putRecord, deleteRecord } for mesh
// peers (sessionRepo makes one from an OAuth session); handle: the signed-in account's handle
// subhandle: this page's identity on mesh peers (default web.<handle>); owner: { did, handle } of the mesh owner
// bridge: { did, subhandle } or false; peers: [{ did, subhandle }] to dial directly
// age: the age-encryption module; makePeer: config -> RTCPeerConnection
export function atirc({
guests = [],
repo,
handle,
subhandle = handle && `web.${handle}`,
owner = repo && { did: repo.did, handle },
bridge = owner && { did: owner.did, subhandle: owner.handle },
peers = [],
age,
makePeer = (config) => new RTCPeerConnection(config),
host = (options) => new WebrtcHost(options),
retryMs = RETRY_MS,
bridgeWaitMs = BRIDGE_WAIT_MS,
} = {}) {
if (!guests.length && (!repo?.did || !handle)) throw new Error("atirc needs guests, or repo (with did) and handle");
const channels = new Map();
const messageListeners = new Set();
const presenceListeners = new Set();
const emit = (listeners, event) => listeners.forEach((cb) => cb(event));
const known = new Map(peers.map((p) => [keyOf(p), p]));
const presence = (name, event, peer) => emit(presenceListeners, { channel: name, event, did: peer.did, identity: peer.subhandle });
// Where a link goes: a guest's tunnel, or a mesh peer's record-signalled datachannel.
const guestTarget = (g) => ({
key: `guest ${g.receipt.uri}`,
async connect(name, seen) {
const webrtc = host({ events: g.events, signal: g.signal, makePeer, age });
const ch = await webrtc.call({ subscribe: true, params: { receipt: g.receipt, tag: g.tag ?? GUEST_TAG } }).catch((e) => (webrtc.close?.(), Promise.reject(e)));
return { webrtc, mesh: await channelOver(ch, { channel: name, ...seen }).catch((e) => (webrtc.close?.(), Promise.reject(e))) };
},
});
const meshTarget = (peer) => ({
key: keyOf(peer),
peer,
async connect(name, seen) {
const signals = recordSignals({ repo, subhandle });
const webrtc = host({ events: signals.events, signal: signals.signal, makePeer, age });
try {
const receipt = await peerReceipt(repo, peer.did, peer.subhandle);
const ch = await webrtc.call({ subscribe: true, params: { receipt, tag: MESH_LABEL } });
return { webrtc, mesh: await channelOver(ch, { channel: name, ...seen }) };
} catch (error) {
webrtc.close?.();
throw error;
}
},
});
// A mesh peer we may dial for a channel: its record names a WebRTC answerer and carries the channel.
async function dialable(peer, name) {
const found = await repo.getRecord({ repo: peer.did, collection: PEER, rkey: peer.subhandle }).catch(() => null);
const value = found?.value;
return typeof value?.webrtc === "string" && Array.isArray(value.channels) && value.channels.includes(name);
}
async function open(name, ch, target) {
const { key } = target;
if (ch.links.has(key) || ch.dialing.has(key)) return ch.links.get(key);
ch.dialing.add(key);
try {
const seen = ch.seen.get(key) ?? { since: 0, epoch: 0 };
const { webrtc, mesh } = await target.connect(name, { since: seen.since, seenEpoch: seen.epoch });
const peer = { did: mesh.reply.did ?? target.peer?.did, subhandle: mesh.reply.subhandle ?? target.peer?.subhandle };
const link = { target, peer, bridge: mesh.reply.bridge === true, epoch: mesh.reply.epoch ?? 0, mesh, webrtc, members: new Map() };
if (!ch.joined || (ch.bridge && !link.bridge)) {
mesh.close();
webrtc.close?.();
return null;
}
ch.links.set(key, link);
ch.retry = retryMs;
presence(name, "join", peer);
if (link.bridge) {
ch.bridge = link;
for (const [other, direct] of ch.links) if (other !== key) drop(name, ch, other, direct);
} else {
for (const p of mesh.reply.peers ?? []) if (p?.did && p?.subhandle) known.set(keyOf(p), { did: p.did, subhandle: p.subhandle });
if (!ch.bridge && !guests.length) dialKnown(name, ch);
}
pump(name, ch, key, link);
return link;
} finally {
ch.dialing.delete(key);
}
}
function drop(name, ch, key, link) {
ch.links.delete(key);
link.mesh.close();
link.webrtc.close?.();
for (const member of link.members.values()) presence(name, "leave", member);
presence(name, "leave", link.peer);
}
async function pump(name, ch, key, link) {
try {
for await (const line of link.mesh.messages) receive(name, ch, key, link, line);
} catch {
// a closed or broken channel ends like a clean close
}
if (ch.links.get(key) !== link) return;
drop(name, ch, key, link);
if (ch.bridge === link) ch.bridge = null;
retry(name, ch, link.target);
}
// Redial with backoff while joined: the same target while others stay linked, else the whole plan.
function retry(name, ch, target) {
if (!ch.joined) return;
const wait = ch.retry;
ch.retry = Math.min(ch.retry * 2, MAX_RETRY_MS);
setTimeout(async () => {
if (!ch.joined) return;
const again = ch.links.size ? open(name, ch, target) : plan(name, ch);
if (!(await again.then(() => true, () => false)) && !ch.links.size) retry(name, ch, target);
}, wait);
}
function receive(name, ch, key, link, line) {
if (link.bridge && Array.isArray(line.members)) {
const now = new Map(line.members.filter((m) => m?.did && m?.subhandle).map((m) => [keyOf(m), { did: m.did, subhandle: m.subhandle }]));
for (const [k, m] of link.members) if (!now.has(k)) presence(name, "leave", m);
for (const [k, m] of now) if (!link.members.has(k)) presence(name, "join", m);
link.members = now;
return;
}
if (typeof line.seq !== "number" || typeof line.text !== "string") return;
const seen = ch.seen.get(key);
if (seen?.epoch === link.epoch && line.seq <= seen.since) return;
ch.seen.set(key, { since: line.seq, epoch: link.epoch });
// Only a bridge speaks for others; a direct peer's lines are its own.
const author = link.bridge && line.did && line.subhandle ? { did: line.did, subhandle: line.subhandle } : link.peer;
emit(messageListeners, { channel: name, identity: author.subhandle, nick: nickOf(author.subhandle), did: author.did, text: line.text, ts: line.ts, seq: line.seq, via: link.peer.subhandle });
}
function dialKnown(name, ch) {
for (const peer of known.values()) {
if (peer.subhandle === subhandle && peer.did === repo.did) continue;
if (ch.links.has(keyOf(peer)) || ch.dialing.has(keyOf(peer))) continue;
dialable(peer, name).then((ok) => ok && !ch.bridge && open(name, ch, meshTarget(peer))).catch(() => {});
}
}
const opened = (target, name, ch) => open(name, ch, target).then((l) => l ?? Promise.reject(new Error("closed")));
// Guests in order (the first that opens); else the mesh bridge first, then every WebRTC peer that carries
// the channel once the bridge fails or takes over bridgeWaitMs (a bridge that opens later takes over).
async function plan(name, ch) {
if (guests.length) {
const errors = [];
for (const g of guests) {
try {
return void (await opened(guestTarget(g), name, ch));
} catch (error) {
errors.push(error.message);
}
}
throw new Error(`no guest node opened ${name}: ${errors.join("; ")}`);
}
const attempts = [];
if (bridge) {
const viaBridge = opened(meshTarget(bridge), name, ch);
attempts.push(viaBridge);
const link = await Promise.race([viaBridge.catch(() => null), new Promise((r) => setTimeout(r, bridgeWaitMs))]);
if (link?.bridge) return;
}
for (const peer of known.values()) {
if (keyOf(peer) === keyOf(bridge || {}) || !(await dialable(peer, name))) continue;
attempts.push(opened(meshTarget(peer), name, ch));
}
if (!attempts.length) throw new Error(`no peer of ${name} takes WebRTC`);
await Promise.any(attempts).catch((error) => {
if (!ch.links.size) throw new Error(`no peer of ${name} answered: ${error.errors.map((e) => e.message).join("; ")}`);
});
}
// Who this page posts as on a channel: its bridge node, else itself.
const meOn = (ch) => (ch?.bridge ? ch.bridge.peer : { did: repo?.did, subhandle });
const api = {
// Join a channel; resolves once one link is open (history replays first).
join(channel) {
const name = normalizeChannel(channel);
let ch = channels.get(name);
if (ch) return ch.planning;
channels.set(name, (ch = { links: new Map(), dialing: new Set(), seen: new Map(), bridge: null, joined: true, retry: retryMs }));
ch.planning = plan(name, ch).catch((error) => {
if (!ch.links.size) {
ch.joined = false;
if (channels.get(name) === ch) channels.delete(name);
}
throw error;
});
return ch.planning;
},
leave(channel) {
const name = normalizeChannel(channel);
const ch = channels.get(name);
if (!ch) return;
ch.joined = false;
channels.delete(name);
for (const [key, link] of ch.links) drop(name, ch, key, link);
},
// Post on every link of the channel; the page's own line is emitted too (self: true).
async post(channel, text) {
const name = normalizeChannel(channel);
const ch = channels.get(name);
const links = [...(ch?.links.values() ?? [])];
if (!links.length) throw new Error(`not in ${name}`);
await Promise.all(links.map((link) => link.mesh.send(String(text))));
const me = meOn(ch);
emit(messageListeners, { channel: name, identity: me.subhandle, nick: nickOf(me.subhandle), did: me.did, text: String(text), ts: new Date().toISOString(), self: true });
},
onMessage(cb) {
messageListeners.add(cb);
return () => messageListeners.delete(cb);
},
onPresence(cb) {
presenceListeners.add(cb);
return () => presenceListeners.delete(cb);
},
// { me, channels, peers: [{did, subhandle}] }: the linked nodes and their members, on one channel or all.
async peers(channel) {
const names = channel ? [normalizeChannel(channel)] : [...channels.keys()];
const out = new Map();
let me;
for (const name of names) {
const ch = channels.get(name);
me ??= ch?.bridge?.peer.subhandle;
for (const link of ch?.links.values() ?? []) {
out.set(keyOf(link.peer), link.peer);
for (const [k, m] of link.members) out.set(k, m);
}
}
me ??= subhandle;
for (const [k, p] of out) if (p.subhandle === me) out.delete(k);
return { me, channels: [...channels.keys()], peers: [...out.values()] };
},
close() {
for (const name of [...channels.keys()]) api.leave(name);
},
};
return api;
}
// A repo for atirc() from an OAuth session (@atproto/oauth-client-browser's OAuthSession: did + fetchHandler):
// writes go through the session, reads are public getRecords at the record's PDS (did:plc or did:web).
export function sessionRepo(session, { fetch = globalThis.fetch, plc = "https://plc.directory" } = {}) {
const did = session.did ?? session.sub;
const pds = new Map();
const pdsOf = (repo) => {
if (!pds.has(repo)) {
const url = repo.startsWith("did:web:") ? `https://${repo.slice(8).replaceAll(":", "/")}/.well-known/did.json` : `${plc}/${repo}`;
const found = fetch(url)
.then((r) => (r.ok ? r.json() : Promise.reject(new Error(`${repo}: DID document ${r.status}`))))
.then((doc) => {
const endpoint = doc.service?.find((s) => s.id === "#atproto_pds" || s.id === `${repo}#atproto_pds`)?.serviceEndpoint;
if (!endpoint) throw new Error(`${repo} names no PDS`);
return endpoint;
});
found.catch(() => pds.delete(repo));
pds.set(repo, found);
}
return pds.get(repo);
};
const write = async (method, body) => {
const response = await session.fetchHandler(`/xrpc/com.atproto.repo.${method}`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ repo: did, ...body }),
});
if (!response.ok) throw new Error(`${method} failed with ${response.status}: ${await response.text()}`);
return response.json();
};
return {
did,
async getRecord({ repo, collection, rkey }) {
const url = new URL("/xrpc/com.atproto.repo.getRecord", await pdsOf(repo));
url.search = new URLSearchParams({ repo, collection, rkey });
const response = await fetch(url);
if (response.status === 400 || response.status === 404) return null;
if (!response.ok) throw new Error(`getRecord ${collection}/${rkey} failed with ${response.status}`);
return response.json();
},
putRecord: (input) => write("putRecord", input),
deleteRecord: (input) => write("deleteRecord", input),
};
}