Skip to content
Sign up

flagon-io/g1t

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