jakelazaroff.com
a proof of concept realtime collaborative text editor using atproto as a sync server
jake.tngl.io/y-pds/
21 kB
670 lines
1import * as Y from "yjs";
2import { Awareness, encodeAwarenessUpdate, applyAwarenessUpdate } from "y-protocols/awareness";
3
4export const DOC_COLLECTION = "com.jakelazaroff.y-pds.doc";
5export const UPDATE_COLLECTION = "com.jakelazaroff.y-pds.update";
6export const AWARENESS_COLLECTION = "com.jakelazaroff.y-pds.awareness";
7export const WEBRTC_COLLECTION = "com.jakelazaroff.y-pds.webrtc";
8
9const encode = (/** @type {Uint8Array} */ update) => btoa(String.fromCharCode(...update));
10const decode = (/** @type {string} */ b64) => Uint8Array.from(atob(b64), c => c.charCodeAt(0));
11
12const TID_CHARS = "234567abcdefghijklmnopqrstuvwxyz";
13
14const s32encode = (/** @type {number} */ i) => {
15 let s = "";
16 while (i) {
17 s = TID_CHARS[i % 32] + s;
18 i = Math.floor(i / 32);
19 }
20 return s;
21};
22
23/** Generate an AT Protocol TID: a 13-character, lexicographically sortable record key. */
24export const genTid = () => {
25 const timestamp = Date.now() * 1000 + Math.floor(Math.random() * 1000);
26 const clockId = Math.floor(Math.random() * 1024);
27 return s32encode(timestamp).padStart(11, "2") + s32encode(clockId).padStart(2, "2");
28};
29
30const MSG_DOC = 0;
31const MSG_AWARENESS = 1;
32
33const ICE_SERVERS = [
34 { urls: "stun:stun.l.google.com:19302" },
35];
36
37export class YPdsProvider {
38 /** @type {Y.Doc} */
39 #ydoc;
40
41 /** @type {Map<string, string>} */
42 #pds = new Map();
43
44 /** @param {string} did */
45 async #getPDS(did) {
46 let pds = this.#pds.get(did);
47 if (pds) return pds;
48
49 const url = did.startsWith("did:web:")
50 ? `https://${did.slice("did:web:".length)}/.well-known/did.json`
51 : `https://plc.directory/${did}`;
52 const doc = await fetch(url).then(r => r.json());
53
54 pds = doc.service?.find(s => s.type === "AtprotoPersonalDataServer")?.serviceEndpoint;
55 if (!pds) throw new Error(`no PDS found for ${did}`);
56
57 this.#pds.set(did, pds);
58 return pds;
59 }
60
61 /**
62 * @params {object} options
63 * @params {string} [options.rkey]
64 * @params {string} options.repo
65 * @params {string} options.collection
66 */
67 async #getRecord({ repo, collection, rkey }) {
68 const pds = await this.#getPDS(repo);
69 const params = new URLSearchParams({ repo, collection, rkey });
70 const res = await fetch(`${pds}/xrpc/com.atproto.repo.getRecord?${params}`);
71 const data = await res.json();
72 if (!res.ok) throw new Error(data.message);
73 return data;
74 }
75
76 /**
77 * @params {object} options
78 * @params {string} options.repo
79 * @params {string} options.collection
80 * @params {string} [options.cursor]
81 */
82 async #listRecords({ repo, collection, cursor }) {
83 const pds = await this.#getPDS(repo);
84 const params = new URLSearchParams({ repo, collection, limit: "100" });
85 if (cursor) params.set("cursor", cursor);
86 const res = await fetch(`${pds}/xrpc/com.atproto.repo.listRecords?${params}`);
87 const data = await res.json();
88 if (!res.ok) throw new Error(data.message);
89 return data;
90 }
91
92 /**
93 * @params {object} options
94 * @params {string} options.repo
95 * @params {string} options.collection
96 * @params {object} options.record
97 * @params {string} [options.rkey]
98 */
99 async #upsertRecord({ repo, collection, rkey, record }) {
100 const pds = await this.#getPDS(repo);
101 const verb = rkey ? "com.atproto.repo.putRecord" : "com.atproto.repo.createRecord";
102 const res = await fetch(`${pds}/xrpc/${verb}`, {
103 method: "POST",
104 headers: {
105 "content-type": "application/json",
106 "x-atsw-did": repo,
107 },
108 body: JSON.stringify({ repo, collection, rkey, record: { $type: collection, ...record } }),
109 });
110 const data = await res.json();
111 if (!res.ok) throw new Error(data.message);
112 return data;
113 }
114
115 #ownerDid = "";
116 #rkey = "";
117
118 /** @type {Array<{uri: string, cid: string, repo: string, value: {docId: string, update: string, createdAt: string}}>} */
119 #allRecords = [];
120
121 /** @type {Map<string, {repo: string, uri: string, cid: string, value: object}>} */
122 #awarenessMap = new Map();
123
124 /** @type {Set<WebSocket>} */
125 #sockets = new Set();
126
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 = [];
145
146 /** @type {{ editors: string[], [key: string]: unknown } | null} */
147 #doc = null;
148
149 #destroyed = false;
150
151 /**
152 * @param {object} options
153 * @param {Y.Doc} options.ydoc
154 * @param {string} options.pds
155 * @param {string} options.atUri
156 * @param {string} options.did
157 * @param {Awareness} [options.awareness]
158 * @param {string} [options.jetstream]
159 */
160 /** @type {Map<string, { pc: RTCPeerConnection }>} */
161 #peers = new Map();
162
163 /** @type {((editors: string[]) => void) | null} */
164 onMembersChange = null;
165
166 /** @type {(() => void) | null} */
167 onRecordAdded = null;
168
169 /** @type {(() => void) | null} */
170 onAwarenessChange = null;
171
172 /** @type {(() => void) | null} */
173 onPauseChange = null;
174
175 /** @type {((did: string) => void) | null} */
176 onPeerConnect = null;
177
178 /** @type {((did: string) => void) | null} */
179 onPeerDisconnect = null;
180
181 #paused = false;
182 get paused() { return this.#paused; }
183
184 #throttle = 5000;
185
186 /** @type {Uint8Array[]} */
187 #pendingUpdates = [];
188 #updateTimerId = null;
189
190 #awarenessDirty = false;
191 #awarenessTimerId = null;
192
193 constructor({ ydoc, pds, atUri, did, awareness, jetstream, jetstreams, throttle }) {
194 this.#pds.set(did, pds);
195 this.#ydoc = ydoc;
196
197 const [ownerDid, , rkey] = atUri.slice("at://".length).split("/");
198 this.#ownerDid = ownerDid;
199 this.#rkey = rkey;
200 this.did = did;
201 this.awareness = awareness ?? new Awareness(ydoc);
202 if (jetstreams) this.#jetstreams = jetstreams;
203 else if (jetstream) this.#jetstreams = [jetstream];
204 if (throttle !== undefined) this.#throttle = throttle;
205
206 this.awareness.on("change", this.#onAwarenessChange);
207 this.#ydoc.on("update", this.#onUpdate);
208 }
209
210 async load() {
211 try {
212 const data = await this.#getRecord({
213 repo: this.#ownerDid,
214 collection: DOC_COLLECTION,
215 rkey: this.#rkey,
216 });
217 this.#doc = data.value;
218 this.onMembersChange?.(this.#doc.editors);
219 } catch {
220 // if the fetch fails and the document is not in our own repo,
221 // throw an error rather than attempting to create it
222 if (this.did !== this.#ownerDid) throw new Error("Document not found");
223
224 // create the document
225 this.#doc = { docId: this.#rkey, editors: [], createdAt: new Date().toISOString() };
226 await this.#upsertRecord({
227 repo: this.did,
228 collection: DOC_COLLECTION,
229 rkey: this.#rkey,
230 record: this.#doc,
231 });
232 this.onMembersChange?.(this.#doc.editors);
233 }
234
235 // fetch updates from the owner and all editors
236 const repos = [this.#ownerDid].concat(this.#doc.editors);
237 const updates = await Promise.all(repos.map(repo => this.#fetchUpdates(repo)));
238
239 // apply all the updates
240 Y.transact(this.#ydoc, () => {
241 for (const { value } of updates.flat()) {
242 Y.applyUpdate(this.#ydoc, decode(value.update));
243 }
244 }, this);
245
246 this.#allRecords = updates.flat();
247
248 // Persist a consolidated copy in our own PDS so we keep the full document
249 // even if other editors later delete their records.
250 await this.#consolidate();
251
252 this.#subscribe();
253 }
254
255 async #consolidate() {
256 // Nothing to back up if every record is already ours.
257 if (!this.#allRecords.some(r => r.repo !== this.did)) return;
258
259 const merged = Y.encodeStateAsUpdate(this.#ydoc);
260 await this.#upsertRecord({
261 repo: this.did,
262 collection: UPDATE_COLLECTION,
263 // Fixed rkey = the doc's own key, so putRecord overwrites rather than
264 // accumulating a new record on every load.
265 rkey: this.#rkey,
266 record: { docId: this.#rkey, update: encode(merged), createdAt: new Date().toISOString() },
267 });
268 }
269
270 async #fetchUpdates(repo) {
271 const records = [];
272 let cursor;
273
274 do {
275 const res = await this.#listRecords({ repo, collection: UPDATE_COLLECTION, cursor });
276 records.push(
277 ...res.records
278 .filter(r => r.value.docId === this.#rkey)
279 .map(r => ({ ...r, repo })),
280 );
281 cursor = res.cursor;
282 } while (cursor);
283
284 return records;
285 }
286
287 /**
288 * @param {Uint8Array} update
289 * @param {unknown} origin
290 */
291 #flushUpdate = async () => {
292 this.#updateTimerId = null;
293 if (this.#pendingUpdates.length === 0) return;
294 const merged = this.#pendingUpdates.length === 1
295 ? this.#pendingUpdates[0]
296 : Y.mergeUpdates(this.#pendingUpdates);
297 this.#pendingUpdates = [];
298 const createdAt = new Date().toISOString();
299 const data = await this.#upsertRecord({
300 repo: this.did,
301 collection: UPDATE_COLLECTION,
302 record: { docId: this.#rkey, update: encode(merged), createdAt },
303 });
304 this.#allRecords.push({
305 uri: data.uri,
306 cid: data.cid,
307 repo: this.did,
308 value: { docId: this.#rkey, update: encode(merged), createdAt },
309 });
310 this.onRecordAdded?.();
311 };
312
313 #onUpdate = (update, origin) => {
314 if (origin === this) return;
315 this.#pendingUpdates.push(update);
316 if (!this.#paused && this.#updateTimerId == null) {
317 this.#updateTimerId = setTimeout(this.#flushUpdate, this.#throttle);
318 }
319 };
320
321 /** @param {{ added: number[], updated: number[] }} changes */
322 #flushAwareness = async () => {
323 this.#awarenessTimerId = null;
324 if (!this.#awarenessDirty) return;
325 this.#awarenessDirty = false;
326 const update = encodeAwarenessUpdate(this.awareness, [this.awareness.clientID]);
327 const createdAt = new Date().toISOString();
328 const data = await this.#upsertRecord({
329 repo: this.did,
330 collection: AWARENESS_COLLECTION,
331 rkey: this.#rkey,
332 record: { docId: this.#rkey, awareness: encode(update), createdAt },
333 });
334 this.#awarenessMap.set(this.did, {
335 repo: this.did,
336 uri: data.uri,
337 cid: data.cid,
338 value: { docId: this.#rkey, awareness: encode(update), createdAt },
339 });
340 this.onAwarenessChange?.();
341 };
342
343 #onAwarenessChange = ({ added, updated }) => {
344 const ours = this.awareness.clientID;
345 if (!added.includes(ours) && !updated.includes(ours)) return;
346 this.#awarenessDirty = true;
347 if (!this.#paused && this.#awarenessTimerId == null) {
348 this.#awarenessTimerId = setTimeout(this.#flushAwareness, this.#throttle);
349 }
350 };
351
352 getMembers() {
353 return this.#doc.editors;
354 }
355
356 getAwarenessRecords() {
357 return [...this.#awarenessMap.values()];
358 }
359
360 getDebugInfo() {
361 return {
362 doc: this.#doc,
363 docUri: `at://${this.#ownerDid}/${DOC_COLLECTION}/${this.#rkey}`,
364 ownerDid: this.#ownerDid,
365 rkey: this.#rkey,
366 records: [...this.#allRecords],
367 };
368 }
369
370 pause() {
371 if (this.#paused) return;
372 this.#paused = true;
373 clearTimeout(this.#updateTimerId);
374 this.#updateTimerId = null;
375 clearTimeout(this.#awarenessTimerId);
376 this.#awarenessTimerId = null;
377 this.onPauseChange?.();
378 }
379
380 async play() {
381 if (!this.#paused) return;
382 this.#paused = false;
383
384 // Re-fetch all repos and apply only records we haven't seen yet,
385 // merged into one update so the ydoc sees a single transaction.
386 const repos = [this.#ownerDid, ...this.#doc.editors];
387 const fetched = (await Promise.all(repos.map(r => this.#fetchUpdates(r)))).flat();
388 const seenUris = new Set(this.#allRecords.map(r => r.uri));
389 const newRecords = fetched.filter(r => !seenUris.has(r.uri));
390 if (newRecords.length > 0) {
391 const merged = Y.mergeUpdates(newRecords.map(r => decode(r.value.update)));
392 Y.applyUpdate(this.#ydoc, merged, this);
393 this.#allRecords.push(...newRecords);
394 this.onRecordAdded?.();
395 }
396
397 // Flush buffered local edits as one record.
398 await this.#flushUpdate();
399
400 this.onPauseChange?.();
401 }
402
403 getPeers() {
404 return [...this.#peers.keys()];
405 }
406
407 async connect(dids) {
408 await Promise.all(dids.map(async did => {
409 if (did === this.did || this.#peers.has(did)) return;
410
411 const pc = new RTCPeerConnection({ iceServers: ICE_SERVERS });
412 const channel = pc.createDataChannel("yjs");
413 this.#peers.set(did, { pc });
414 this.#setupDataChannel(did, channel);
415
416 const icePromise = this.#waitForIce(pc);
417 const offer = await pc.createOffer();
418 await pc.setLocalDescription(offer);
419 await icePromise;
420
421 await this.#upsertRecord({
422 repo: this.did,
423 collection: WEBRTC_COLLECTION,
424 record: {
425 target: did,
426 docId: this.#rkey,
427 type: "offer",
428 sdp: pc.localDescription.sdp,
429 createdAt: new Date().toISOString(),
430 },
431 });
432 }));
433 }
434
435 /** Wait for ICE gathering to finish. Call before setLocalDescription to avoid missing candidates. */
436 #waitForIce(pc) {
437 return new Promise(resolve => {
438 if (pc.iceGatheringState === "complete") { resolve(); return; }
439 const timer = setTimeout(resolve, 10_000);
440 pc.addEventListener("icecandidate", function check(e) {
441 if (!e.candidate) {
442 clearTimeout(timer);
443 pc.removeEventListener("icecandidate", check);
444 resolve();
445 }
446 });
447 });
448 }
449
450 /** @param {string} did @param {RTCDataChannel} channel */
451 #setupDataChannel(did, channel) {
452 channel.binaryType = "arraybuffer";
453
454 const send = (type, data) => {
455 if (channel.readyState !== "open") return;
456 const msg = new Uint8Array(1 + data.byteLength);
457 msg[0] = type;
458 msg.set(data, 1);
459 channel.send(msg);
460 };
461
462 channel.onopen = () => {
463 this.onPeerConnect?.(did);
464 send(MSG_DOC, Y.encodeStateAsUpdate(this.#ydoc));
465 send(MSG_AWARENESS, encodeAwarenessUpdate(this.awareness, [this.awareness.clientID]));
466 };
467
468 channel.onmessage = e => {
469 const msg = new Uint8Array(e.data);
470 const payload = msg.slice(1);
471 if (msg[0] === MSG_DOC) Y.applyUpdate(this.#ydoc, payload, "webrtc");
472 else if (msg[0] === MSG_AWARENESS) applyAwarenessUpdate(this.awareness, payload, "webrtc");
473 };
474
475 const onUpdate = (update, origin) => {
476 if (origin === "webrtc" || origin === this) return;
477 send(MSG_DOC, update);
478 };
479 this.#ydoc.on("update", onUpdate);
480
481 const onAwarenessChange = ({ added, updated }) => {
482 const own = [...added, ...updated].filter(id => id === this.awareness.clientID);
483 if (own.length > 0) send(MSG_AWARENESS, encodeAwarenessUpdate(this.awareness, own));
484 };
485 this.awareness.on("change", onAwarenessChange);
486
487 channel.onclose = () => {
488 this.#ydoc.off("update", onUpdate);
489 this.awareness.off("change", onAwarenessChange);
490 this.#peers.delete(did);
491 this.onPeerDisconnect?.(did);
492 };
493 }
494
495 async #handleOffer(did, record) {
496 if (this.#peers.has(did)) return;
497
498 const pc = new RTCPeerConnection({ iceServers: ICE_SERVERS });
499 this.#peers.set(did, { pc });
500 pc.ondatachannel = e => this.#setupDataChannel(did, e.channel);
501
502 const icePromise = this.#waitForIce(pc);
503 await pc.setRemoteDescription({ type: "offer", sdp: record.sdp });
504 const answer = await pc.createAnswer();
505 await pc.setLocalDescription(answer);
506 await icePromise;
507
508 await this.#upsertRecord({
509 repo: this.did,
510 collection: WEBRTC_COLLECTION,
511 record: {
512 target: did,
513 docId: this.#rkey,
514 type: "answer",
515 sdp: pc.localDescription.sdp,
516 createdAt: new Date().toISOString(),
517 },
518 });
519 }
520
521 async #handleAnswer(did, record) {
522 const peer = this.#peers.get(did);
523 if (!peer || peer.pc.signalingState !== "have-local-offer") return;
524 await peer.pc.setRemoteDescription({ type: "answer", sdp: record.sdp });
525 }
526
527 async setMembers(editors) {
528 this.#doc = { ...this.#doc, editors };
529 await this.#upsertRecord({
530 repo: this.did,
531 collection: DOC_COLLECTION,
532 rkey: this.#rkey,
533 record: this.#doc,
534 });
535 }
536
537 #subscribe() {
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);
562 url.searchParams.append("wantedCollections", UPDATE_COLLECTION);
563 url.searchParams.append("wantedCollections", AWARENESS_COLLECTION);
564 url.searchParams.append("wantedCollections", DOC_COLLECTION);
565 url.searchParams.append("wantedCollections", WEBRTC_COLLECTION);
566 url.searchParams.append("wantedDids", this.#ownerDid);
567 for (const editor of this.#doc.editors) url.searchParams.append("wantedDids", editor);
568
569 const cursor = this.#cursors.get(endpoint);
570 if (cursor) url.searchParams.set("cursor", String(cursor));
571
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;
578 setTimeout(() => {
579 if (!this.#destroyed) this.#connect(endpoint);
580 }, 1000);
581 };
582
583 ws.onmessage = e => {
584 const event = JSON.parse(e.data);
585 this.#cursors.set(endpoint, event.time_us);
586 if (event.kind !== "commit") return;
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) {
597 switch (event.commit.collection) {
598 case DOC_COLLECTION: {
599 if (event.did !== this.#ownerDid || event.commit.rkey !== this.#rkey) return;
600 await this.load();
601 break;
602 }
603
604 case UPDATE_COLLECTION: {
605 if (event.commit.record.docId !== this.#rkey) return;
606 if (event.did !== this.#ownerDid && !this.#doc.editors.includes(event.did)) return;
607
608 if (event.commit.operation !== "create") return;
609
610 if (this.#paused) break;
611
612 // Own writes are already tracked optimistically in #flushUpdate
613 if (event.did !== this.did) {
614 this.#allRecords.push({
615 uri: `at://${event.did}/${UPDATE_COLLECTION}/${event.commit.rkey}`,
616 cid: event.commit.cid,
617 repo: event.did,
618 value: event.commit.record,
619 });
620 this.onRecordAdded?.();
621 }
622 Y.applyUpdate(this.#ydoc, decode(event.commit.record.update), this);
623 break;
624 }
625
626 case AWARENESS_COLLECTION: {
627 if (event.commit.record.docId !== this.#rkey) return;
628 if (event.did !== this.#ownerDid && !this.#doc.editors.includes(event.did)) return;
629
630 if (this.#paused) break;
631
632 this.#awarenessMap.set(event.did, {
633 repo: event.did,
634 uri: `at://${event.did}/${AWARENESS_COLLECTION}/${event.commit.rkey}`,
635 cid: event.commit.cid,
636 value: event.commit.record,
637 });
638 this.onAwarenessChange?.();
639 if (event.did === this.did) break;
640 applyAwarenessUpdate(this.awareness, decode(event.commit.record.awareness), "remote");
641 break;
642 }
643
644 case WEBRTC_COLLECTION: {
645 const rec = event.commit.record;
646 if (rec.target !== this.did) return;
647 if (rec.docId !== this.#rkey) return;
648 if (event.did !== this.#ownerDid && !this.#doc.editors.includes(event.did)) return;
649 if (Date.now() - new Date(rec.createdAt).getTime() > 120_000) return;
650
651 if (rec.type === "offer") this.#handleOffer(event.did, rec).catch(console.error);
652 else if (rec.type === "answer") this.#handleAnswer(event.did, rec).catch(console.error);
653 break;
654 }
655 }
656 }
657
658 destroy() {
659 this.#destroyed = true;
660 this.#ydoc.off("update", this.#onUpdate);
661 this.awareness.off("change", this.#onAwarenessChange);
662 clearTimeout(this.#updateTimerId);
663 clearTimeout(this.#awarenessTimerId);
664 this.#flushUpdate();
665 this.awareness.destroy();
666 this.#closeSockets();
667 for (const { pc } of this.#peers.values()) pc.close();
668 this.#peers.clear();
669 }
670}