Skip to content
Sign up

flagon-io/g1t

public
0
196 linesCodeBlameRaw
1/**
2 * A page's live connection: the Yjs document and everyone's presence,
3 * synced over `wss://<site>/<workspace>/-/docs/live?page=<id>` with the
4 * page's room (services/docs src/room.ts). Binary frames speak the
5 * y-protocols sync and awareness messages; text frames are the service's
6 * own notices (`DocsLiveEvent`): a rename, a new suggestion, a version,
7 * a change of access.
8 *
9 * It reconnects with backoff when the socket drops, and says where it is
10 * (`status`) so the page can show "Offline, changes will sync".
11 * Browser-only.
12 */
13import type { DocsLiveEvent } from "@g1t/contracts";
14import * as decoding from "lib0/decoding";
15import * as encoding from "lib0/encoding";
16import * as awarenessProtocol from "y-protocols/awareness";
17import * as syncProtocol from "y-protocols/sync";
18import * as Y from "yjs";
19
20import { heldOpen } from "../../lib/notify-store";
21
22const MESSAGE_SYNC = 0;
23const MESSAGE_AWARENESS = 1;
24const MESSAGE_QUERY_AWARENESS = 3;
25
26export type LiveStatus = "connecting" | "synced" | "offline" | "closed";
27
28export class DocsProvider {
29 readonly doc: Y.Doc;
30 readonly awareness: awarenessProtocol.Awareness;
31 status: LiveStatus = "connecting";
32 private socket: WebSocket | null = null;
33 private attempts = 0;
34 /** When the current connection opened; the backoff starts over only once one holds. */
35 private openedAt: number | null = null;
36 private timer: ReturnType<typeof setTimeout> | null = null;
37 private keepalive: ReturnType<typeof setInterval> | null = null;
38 private stopped = false;
39 private readonly statusListeners = new Set<(status: LiveStatus) => void>();
40 private readonly eventListeners = new Set<(event: DocsLiveEvent) => void>();
41
42 constructor(
43 private readonly url: string,
44 doc?: Y.Doc,
45 ) {
46 this.doc = doc ?? new Y.Doc();
47 this.awareness = new awarenessProtocol.Awareness(this.doc);
48 this.doc.on("update", this.onDocUpdate);
49 this.awareness.on("update", this.onAwarenessUpdate);
50 if (typeof window !== "undefined") {
51 window.addEventListener("beforeunload", this.onUnload);
52 window.addEventListener("online", this.onOnline);
53 }
54 this.connect();
55 }
56
57 onStatus(listener: (status: LiveStatus) => void): () => void {
58 this.statusListeners.add(listener);
59 listener(this.status);
60 return () => this.statusListeners.delete(listener);
61 }
62
63 onEvent(listener: (event: DocsLiveEvent) => void): () => void {
64 this.eventListeners.add(listener);
65 return () => this.eventListeners.delete(listener);
66 }
67
68 private setStatus(status: LiveStatus) {
69 if (this.status === status) return;
70 this.status = status;
71 for (const l of this.statusListeners) l(status);
72 }
73
74 private connect() {
75 if (this.stopped) return;
76 this.setStatus(this.attempts ? "offline" : "connecting");
77 const socket = new WebSocket(this.url);
78 socket.binaryType = "arraybuffer";
79 this.socket = socket;
80 socket.onopen = () => {
81 this.openedAt = Date.now();
82 // Our state vector: the room answers with what we lack, and asks for what it lacks.
83 const encoder = encoding.createEncoder();
84 encoding.writeVarUint(encoder, MESSAGE_SYNC);
85 syncProtocol.writeSyncStep1(encoder, this.doc);
86 socket.send(encoding.toUint8Array(encoder));
87 if (this.awareness.getLocalState() !== null) {
88 const a = encoding.createEncoder();
89 encoding.writeVarUint(a, MESSAGE_AWARENESS);
90 encoding.writeVarUint8Array(a, awarenessProtocol.encodeAwarenessUpdate(this.awareness, [this.doc.clientID]));
91 socket.send(encoding.toUint8Array(a));
92 }
93 const q = encoding.createEncoder();
94 encoding.writeVarUint(q, MESSAGE_QUERY_AWARENESS);
95 socket.send(encoding.toUint8Array(q));
96 if (this.keepalive) clearInterval(this.keepalive);
97 // Answered at the edge without waking the room.
98 this.keepalive = setInterval(() => socket.readyState === WebSocket.OPEN && socket.send("ping"), 25_000);
99 };
100 socket.onmessage = (event) => {
101 if (typeof event.data === "string") {
102 if (event.data === "pong") return;
103 try {
104 const parsed = JSON.parse(event.data) as DocsLiveEvent;
105 for (const l of this.eventListeners) l(parsed);
106 } catch {
107 // Not ours.
108 }
109 return;
110 }
111 this.receive(new Uint8Array(event.data as ArrayBuffer));
112 };
113 socket.onclose = (event) => {
114 if (this.keepalive) clearInterval(this.keepalive);
115 this.socket = null;
116 // Others stop seeing our cursor; we stop seeing theirs.
117 awarenessProtocol.removeAwarenessStates(
118 this.awareness,
119 [...this.awareness.getStates().keys()].filter((id) => id !== this.doc.clientID),
120 this,
121 );
122 // 4403: no longer allowed; 4410: in the trash. Neither comes back by retrying.
123 if (event.code === 4403 || event.code === 4410 || this.stopped) {
124 this.setStatus("closed");
125 return;
126 }
127 this.setStatus("offline");
128 // Only a connection that held starts the backoff over.
129 if (heldOpen(this.openedAt)) this.attempts = 0;
130 this.openedAt = null;
131 const delay = Math.min(30_000, 500 * 2 ** this.attempts) + Math.random() * 500;
132 this.attempts++;
133 this.timer = setTimeout(() => this.connect(), delay);
134 };
135 }
136
137 private receive(data: Uint8Array) {
138 const decoder = decoding.createDecoder(data);
139 const type = decoding.readVarUint(decoder);
140 if (type === MESSAGE_SYNC) {
141 const encoder = encoding.createEncoder();
142 encoding.writeVarUint(encoder, MESSAGE_SYNC);
143 const step = syncProtocol.readSyncMessage(decoder, encoder, this.doc, this);
144 if (encoding.length(encoder) > 1) this.socket?.send(encoding.toUint8Array(encoder));
145 if (step === syncProtocol.messageYjsSyncStep2) this.setStatus("synced");
146 return;
147 }
148 if (type === MESSAGE_AWARENESS) {
149 awarenessProtocol.applyAwarenessUpdate(this.awareness, decoding.readVarUint8Array(decoder), this);
150 }
151 }
152
153 private onDocUpdate = (update: Uint8Array, origin: unknown) => {
154 if (origin === this) return;
155 const encoder = encoding.createEncoder();
156 encoding.writeVarUint(encoder, MESSAGE_SYNC);
157 syncProtocol.writeUpdate(encoder, update);
158 if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(encoding.toUint8Array(encoder));
159 // While offline, the update stays in the document and goes in the next sync.
160 };
161
162 private onAwarenessUpdate = ({ added, updated, removed }: { added: number[]; updated: number[]; removed: number[] }, origin: unknown) => {
163 if (origin === this) return;
164 const changed = [...added, ...updated, ...removed];
165 const encoder = encoding.createEncoder();
166 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
167 encoding.writeVarUint8Array(encoder, awarenessProtocol.encodeAwarenessUpdate(this.awareness, changed));
168 if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(encoding.toUint8Array(encoder));
169 };
170
171 private onUnload = () => {
172 awarenessProtocol.removeAwarenessStates(this.awareness, [this.doc.clientID], "unload");
173 };
174
175 private onOnline = () => {
176 if (this.socket || this.stopped) return;
177 if (this.timer) clearTimeout(this.timer);
178 this.attempts = 0;
179 this.connect();
180 };
181
182 destroy() {
183 this.stopped = true;
184 if (this.timer) clearTimeout(this.timer);
185 if (this.keepalive) clearInterval(this.keepalive);
186 awarenessProtocol.removeAwarenessStates(this.awareness, [this.doc.clientID], "destroy");
187 this.doc.off("update", this.onDocUpdate);
188 this.awareness.off("update", this.onAwarenessUpdate);
189 if (typeof window !== "undefined") {
190 window.removeEventListener("beforeunload", this.onUnload);
191 window.removeEventListener("online", this.onOnline);
192 }
193 this.socket?.close();
194 this.awareness.destroy();
195 }
196}