Skip to content
Sign up

flagon-io/g1t

public
0
197 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Artifacts is a mode in the rail, the shell and the phone's tab bar in place of Docs: its home lists everything you can open by day or as cards under All, Yours and Shared with you, with search, filters and tiles to start a doc (slides, designs and dashboards say they are coming soon), and its sidebar holds favorites, spaces, Private, Shared, projects' docs and the trash.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 { heldOpen } from "../../lib/notify-store";
22
23const MESSAGE_SYNC = 0;
24const MESSAGE_AWARENESS = 1;
25const MESSAGE_QUERY_AWARENESS = 3;
26
27export type LiveStatus = "connecting" | "synced" | "offline" | "closed";
28
29export class FolioProvider {
30 readonly doc: Y.Doc;
31 readonly awareness: awarenessProtocol.Awareness;
32 status: LiveStatus = "connecting";
33 private socket: WebSocket | null = null;
34 private attempts = 0;
35 /** When the current connection opened; the backoff starts over only once one holds. */
36 private openedAt: number | null = null;
37 private timer: ReturnType<typeof setTimeout> | null = null;
38 private keepalive: ReturnType<typeof setInterval> | null = null;
39 private stopped = false;
40 private readonly statusListeners = new Set<(status: LiveStatus) => void>();
41 private readonly eventListeners = new Set<(event: FoliosLiveEvent) => void>();
42
43 constructor(
44 private readonly url: string,
45 doc?: Y.Doc,
46 ) {
47 this.doc = doc ?? new Y.Doc();
48 this.awareness = new awarenessProtocol.Awareness(this.doc);
49 this.doc.on("update", this.onDocUpdate);
50 this.awareness.on("update", this.onAwarenessUpdate);
51 if (typeof window !== "undefined") {
52 window.addEventListener("beforeunload", this.onUnload);
53 window.addEventListener("online", this.onOnline);
54 }
55 this.connect();
56 }
57
58 onStatus(listener: (status: LiveStatus) => void): () => void {
59 this.statusListeners.add(listener);
60 listener(this.status);
61 return () => this.statusListeners.delete(listener);
62 }
63
64 onEvent(listener: (event: FoliosLiveEvent) => void): () => void {
65 this.eventListeners.add(listener);
66 return () => this.eventListeners.delete(listener);
67 }
68
69 private setStatus(status: LiveStatus) {
70 if (this.status === status) return;
71 this.status = status;
72 for (const l of this.statusListeners) l(status);
73 }
74
75 private connect() {
76 if (this.stopped) return;
77 this.setStatus(this.attempts ? "offline" : "connecting");
78 const socket = new WebSocket(this.url);
79 socket.binaryType = "arraybuffer";
80 this.socket = socket;
81 socket.onopen = () => {
82 this.openedAt = Date.now();
83 // Our state vector: the room answers with what we lack, and asks for what it lacks.
84 const encoder = encoding.createEncoder();
85 encoding.writeVarUint(encoder, MESSAGE_SYNC);
86 syncProtocol.writeSyncStep1(encoder, this.doc);
87 socket.send(encoding.toUint8Array(encoder));
88 if (this.awareness.getLocalState() !== null) {
89 const a = encoding.createEncoder();
90 encoding.writeVarUint(a, MESSAGE_AWARENESS);
91 encoding.writeVarUint8Array(a, awarenessProtocol.encodeAwarenessUpdate(this.awareness, [this.doc.clientID]));
92 socket.send(encoding.toUint8Array(a));
93 }
94 const q = encoding.createEncoder();
95 encoding.writeVarUint(q, MESSAGE_QUERY_AWARENESS);
96 socket.send(encoding.toUint8Array(q));
97 if (this.keepalive) clearInterval(this.keepalive);
98 // Answered at the edge without waking the room.
99 this.keepalive = setInterval(() => socket.readyState === WebSocket.OPEN && socket.send("ping"), 25_000);
100 };
101 socket.onmessage = (event) => {
102 if (typeof event.data === "string") {
103 if (event.data === "pong") return;
104 try {
105 const parsed = JSON.parse(event.data) as FoliosLiveEvent;
106 for (const l of this.eventListeners) l(parsed);
107 } catch {
108 // Not ours.
109 }
110 return;
111 }
112 this.receive(new Uint8Array(event.data as ArrayBuffer));
113 };
114 socket.onclose = (event) => {
115 if (this.keepalive) clearInterval(this.keepalive);
116 this.socket = null;
117 // Others stop seeing our cursor; we stop seeing theirs.
118 awarenessProtocol.removeAwarenessStates(
119 this.awareness,
120 [...this.awareness.getStates().keys()].filter((id) => id !== this.doc.clientID),
121 this,
122 );
123 // 4403: no longer allowed; 4410: in the trash. Neither comes back by retrying.
124 if (event.code === 4403 || event.code === 4410 || this.stopped) {
125 this.setStatus("closed");
126 return;
127 }
128 this.setStatus("offline");
129 // Only a connection that held starts the backoff over.
130 if (heldOpen(this.openedAt)) this.attempts = 0;
131 this.openedAt = null;
132 const delay = Math.min(30_000, 500 * 2 ** this.attempts) + Math.random() * 500;
133 this.attempts++;
134 this.timer = setTimeout(() => this.connect(), delay);
135 };
136 }
137
138 private receive(data: Uint8Array) {
139 const decoder = decoding.createDecoder(data);
140 const type = decoding.readVarUint(decoder);
141 if (type === MESSAGE_SYNC) {
142 const encoder = encoding.createEncoder();
143 encoding.writeVarUint(encoder, MESSAGE_SYNC);
144 const step = syncProtocol.readSyncMessage(decoder, encoder, this.doc, this);
145 if (encoding.length(encoder) > 1) this.socket?.send(encoding.toUint8Array(encoder));
146 if (step === syncProtocol.messageYjsSyncStep2) this.setStatus("synced");
147 return;
148 }
149 if (type === MESSAGE_AWARENESS) {
150 awarenessProtocol.applyAwarenessUpdate(this.awareness, decoding.readVarUint8Array(decoder), this);
151 }
152 }
153
154 private onDocUpdate = (update: Uint8Array, origin: unknown) => {
155 if (origin === this) return;
156 const encoder = encoding.createEncoder();
157 encoding.writeVarUint(encoder, MESSAGE_SYNC);
158 syncProtocol.writeUpdate(encoder, update);
159 if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(encoding.toUint8Array(encoder));
160 // While offline, the update stays in the document and goes in the next sync.
161 };
162
163 private onAwarenessUpdate = ({ added, updated, removed }: { added: number[]; updated: number[]; removed: number[] }, origin: unknown) => {
164 if (origin === this) return;
165 const changed = [...added, ...updated, ...removed];
166 const encoder = encoding.createEncoder();
167 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
168 encoding.writeVarUint8Array(encoder, awarenessProtocol.encodeAwarenessUpdate(this.awareness, changed));
169 if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(encoding.toUint8Array(encoder));
170 };
171
172 private onUnload = () => {
173 awarenessProtocol.removeAwarenessStates(this.awareness, [this.doc.clientID], "unload");
174 };
175
176 private onOnline = () => {
177 if (this.socket || this.stopped) return;
178 if (this.timer) clearTimeout(this.timer);
179 this.attempts = 0;
180 this.connect();
181 };
182
183 destroy() {
184 this.stopped = true;
185 if (this.timer) clearTimeout(this.timer);
186 if (this.keepalive) clearInterval(this.keepalive);
187 awarenessProtocol.removeAwarenessStates(this.awareness, [this.doc.clientID], "destroy");
188 this.doc.off("update", this.onDocUpdate);
189 this.awareness.off("update", this.onAwarenessUpdate);
190 if (typeof window !== "undefined") {
191 window.removeEventListener("beforeunload", this.onUnload);
192 window.removeEventListener("online", this.onOnline);
193 }
194 this.socket?.close();
195 this.awareness.destroy();
196 }
197}