jakelazaroff.com
a proof of concept realtime collaborative todo list
2.7 kB
92 lines
1// Subscribe to multiple Jetstream instances at once, filtered server-side to
2// the DIDs + collections we care about. Each commit comes through as JSON
3// with the record inline, so we don't need to re-fetch.
4//
5// A single instance can silently drop or lag commits for some repos, so we
6// fan out to several and dedupe: each commit arrives once per healthy
7// instance, but downstream handlers only need to see it once.
8
9const JETSTREAMS = [
10 "wss://jetstream1.us-east.bsky.network/subscribe",
11 "wss://jetstream2.us-east.bsky.network/subscribe",
12 "wss://jetstream1.us-west.bsky.network/subscribe",
13 "wss://jetstream2.fr.hose.cam/subscribe",
14];
15
16export function subscribe({ dids, collections, onCommit }) {
17 let closed = false;
18 const sockets = new Set();
19
20 const seen = new Set();
21 const seenOrder = [];
22 const once = (key) => {
23 if (seen.has(key)) return false;
24 seen.add(key);
25 seenOrder.push(key);
26 if (seenOrder.length > 1000) seen.delete(seenOrder.shift());
27 return true;
28 };
29
30 const connect = (endpoint) => {
31 if (closed) return;
32 let retry = 0;
33 let cursor = null;
34
35 const open = () => {
36 if (closed) return;
37 const url = new URL(endpoint);
38 for (const d of dids) url.searchParams.append("wantedDids", d);
39 for (const c of collections) url.searchParams.append("wantedCollections", c);
40 // Resume from the last seen event on reconnect, otherwise Jetstream
41 // starts a fresh live tail and commits during the downtime are lost
42 // for good.
43 if (cursor) url.searchParams.set("cursor", String(cursor));
44
45 const ws = new WebSocket(url);
46 sockets.add(ws);
47 ws.onopen = () => (retry = 0);
48 ws.onmessage = (e) => {
49 let msg;
50 try {
51 msg = JSON.parse(e.data);
52 } catch {
53 return;
54 }
55 if (msg.time_us) cursor = msg.time_us;
56 if (msg.kind !== "commit" || !msg.commit) return;
57
58 const c = msg.commit;
59 if (!once(`${msg.did}/${c.collection}/${c.rkey}/${c.cid ?? c.operation}`)) return;
60
61 onCommit({
62 did: msg.did,
63 collection: c.collection,
64 rkey: c.rkey,
65 cid: c.cid ?? null,
66 action: c.operation,
67 record: c.record ?? null,
68 });
69 };
70 ws.onclose = () => {
71 sockets.delete(ws);
72 if (closed) return;
73 retry++;
74 setTimeout(open, Math.min(1000 * 2 ** retry, 30000));
75 };
76 ws.onerror = () => {};
77 };
78
79 open();
80 };
81
82 for (const endpoint of JETSTREAMS) connect(endpoint);
83
84 return () => {
85 closed = true;
86 for (const ws of sockets) {
87 ws.onclose = null;
88 ws.close();
89 }
90 sockets.clear();
91 };
92}