jakelazaroff.com
a proof of concept realtime collaborative text editor using atproto as a sync server
jake.tngl.io/y-pds/
subscribe to multiple jetstreams
author
Jake Lazaroff
date
(Jun 20, 2026, 5:12 PM -0400)
commit
ccfd7bb0
ccfd7bb02cab6f4bbe299912a792e6827ea98ee9
parent
2b39b05d
2b39b05de262121df2e6d4b0327c07322ddac480
+68
-17
+68
-17
y-pds.js
+68
-17
y-pds.js
···
121
121
/** @type {Map<string, {repo: string, uri: string, cid: string, value: object}>} */
122
122
#awarenessMap = new Map();
123
123
124
-
/** @type {WebSocket | null} */
125
-
#ws = null;
124
+
/** @type {Set<WebSocket>} */
125
+
#sockets = new Set();
126
126
127
-
#jetstream = "wss://jetstream2.us-east.bsky.network/subscribe";
127
+
// Subscribe to multiple Jetstream instances at once: any single instance can
128
+
// silently drop or lag commits from some repos (e.g. jetstream2.us-east
129
+
// missing us-west repos), so we fan out and dedupe. Yjs updates are
130
+
// idempotent, but deduping keeps side effects (record list, load(), WebRTC
131
+
// signaling) from firing once per instance.
132
+
#jetstreams = [
133
+
"wss://jetstream1.us-east.bsky.network/subscribe",
134
+
"wss://jetstream2.us-east.bsky.network/subscribe",
135
+
"wss://jetstream1.us-west.bsky.network/subscribe",
136
+
];
137
+
138
+
/** @type {Map<string, number>} per-endpoint cursor (time_us of last event) */
139
+
#cursors = new Map();
140
+
141
+
/** @type {Set<string>} commit keys already processed, for cross-instance dedupe */
142
+
#seen = new Set();
143
+
/** @type {string[]} insertion order for bounding #seen */
144
+
#seenOrder = [];
128
145
129
146
/** @type {{ editors: string[], [key: string]: unknown } | null} */
130
147
#doc = null;
131
148
132
-
#cursor = 0;
133
149
#destroyed = false;
134
150
135
151
/**
···
174
190
#awarenessDirty = false;
175
191
#awarenessTimerId = null;
176
192
177
-
constructor({ ydoc, pds, atUri, did, awareness, jetstream, throttle }) {
193
+
constructor({ ydoc, pds, atUri, did, awareness, jetstream, jetstreams, throttle }) {
178
194
this.#pds.set(did, pds);
179
195
this.#ydoc = ydoc;
180
196
···
183
199
this.#rkey = rkey;
184
200
this.did = did;
185
201
this.awareness = awareness ?? new Awareness(ydoc);
186
-
this.#jetstream = jetstream ?? this.#jetstream;
202
+
if (jetstreams) this.#jetstreams = jetstreams;
203
+
else if (jetstream) this.#jetstreams = [jetstream];
187
204
if (throttle !== undefined) this.#throttle = throttle;
188
205
189
206
this.awareness.on("change", this.#onAwarenessChange);
···
232
249
// even if other editors later delete their records.
233
250
await this.#consolidate();
234
251
235
-
this.#ws?.close();
236
252
this.#subscribe();
237
253
}
238
254
···
519
535
}
520
536
521
537
#subscribe() {
522
-
const url = new URL(this.#jetstream);
538
+
this.#closeSockets();
539
+
for (const endpoint of this.#jetstreams) this.#connect(endpoint);
540
+
}
541
+
542
+
/** Close every socket without triggering its reconnect (intentional teardown). */
543
+
#closeSockets() {
544
+
for (const ws of this.#sockets) {
545
+
ws.onclose = null;
546
+
ws.close();
547
+
}
548
+
this.#sockets.clear();
549
+
}
550
+
551
+
/** Record a commit key; returns false if it was already seen on another instance. */
552
+
#once(key) {
553
+
if (this.#seen.has(key)) return false;
554
+
this.#seen.add(key);
555
+
this.#seenOrder.push(key);
556
+
if (this.#seenOrder.length > 2000) this.#seen.delete(this.#seenOrder.shift());
557
+
return true;
558
+
}
559
+
560
+
#connect(endpoint) {
561
+
const url = new URL(endpoint);
523
562
url.searchParams.append("wantedCollections", UPDATE_COLLECTION);
524
563
url.searchParams.append("wantedCollections", AWARENESS_COLLECTION);
525
564
url.searchParams.append("wantedCollections", DOC_COLLECTION);
···
527
566
url.searchParams.append("wantedDids", this.#ownerDid);
528
567
for (const editor of this.#doc.editors) url.searchParams.append("wantedDids", editor);
529
568
530
-
if (this.#cursor) url.searchParams.set("cursor", this.#cursor);
569
+
const cursor = this.#cursors.get(endpoint);
570
+
if (cursor) url.searchParams.set("cursor", String(cursor));
531
571
532
-
this.#ws = new WebSocket(url);
533
-
this.#ws.onclose = e => {
534
-
if (e.target !== this.#ws || this.#destroyed) return;
572
+
const ws = new WebSocket(url);
573
+
this.#sockets.add(ws);
574
+
ws.onerror = () => {};
575
+
ws.onclose = () => {
576
+
this.#sockets.delete(ws);
577
+
if (this.#destroyed) return;
535
578
setTimeout(() => {
536
-
if (!this.#destroyed) this.#subscribe();
579
+
if (!this.#destroyed) this.#connect(endpoint);
537
580
}, 1000);
538
581
};
539
582
540
-
this.#ws.onmessage = async e => {
583
+
ws.onmessage = e => {
541
584
const event = JSON.parse(e.data);
542
-
this.#cursor = event.time_us;
585
+
this.#cursors.set(endpoint, event.time_us);
543
586
if (event.kind !== "commit") return;
544
587
588
+
// The same commit arrives from every instance; process it only once.
589
+
const c = event.commit;
590
+
if (!this.#once(`${event.did}/${c.collection}/${c.rkey}/${c.cid ?? c.operation}`)) return;
591
+
592
+
this.#handleCommit(event);
593
+
};
594
+
}
595
+
596
+
async #handleCommit(event) {
545
597
switch (event.commit.collection) {
546
598
case DOC_COLLECTION: {
547
599
if (event.did !== this.#ownerDid || event.commit.rkey !== this.#rkey) return;
···
601
653
break;
602
654
}
603
655
}
604
-
};
605
656
}
606
657
607
658
destroy() {
···
612
663
clearTimeout(this.#awarenessTimerId);
613
664
this.#flushUpdate();
614
665
this.awareness.destroy();
615
-
this.#ws?.close();
666
+
this.#closeSockets();
616
667
for (const { pc } of this.#peers.values()) pc.close();
617
668
this.#peers.clear();
618
669
}