Skip to content
Sign up

flagon-io/g1t

public
0
495 linesCodeBlameRaw
1/**
2 * One folio's live room, as a Durable Object named by the folio id: the
3 * same socket protocol, hibernation, roles, save alarm and index alarm as
4 * a Docs page's room (src/room.ts, PageRoom), for every kind of folio
5 * through its kind's model (src/kinds/). The kind is kept in `meta`.
6 *
7 * It owns the folio's Yjs document: every editor's socket syncs with it
8 * (y-protocols sync and awareness, binary frames), and every change made
9 * on the server (an agent's edit, an accepted suggestion, a restore, a
10 * comment) is applied here, so all of them merge as one CRDT. A few
11 * seconds after a burst of edits (an alarm) it saves the kind's text
12 * rendition, card, links, citations and history to D1 (src/persist.ts
13 * `saveFolio`); half a minute after the text first changes, it brings
14 * the folio's passages in the index up to date (src/indexer.ts
15 * `indexFolio`).
16 *
17 * The room authorizes nothing about who may open the folio: the Worker
18 * checks the viewer's role before forwarding a socket
19 * (src/folios/service.ts, `live`) and puts it in ROOM_MEMBER_HEADER. The
20 * room enforces that role: a socket that may only view or comment never
21 * changes the document. Access changes reach open sockets through
22 * `setRole`; a socket whose access ended closes with 4403.
23 */
24import { DurableObject } from "cloudflare:workers";
25
26import type { DocEditTarget, DocRole, DocThreadAction, FolioAgentEdit, FolioKind, FoliosLiveEvent, MemberProfile, ServiceBinding } from "@g1t/contracts";
27import * as decoding from "lib0/decoding";
28import * as encoding from "lib0/encoding";
29import * as syncProtocol from "y-protocols/sync";
30import * as Y from "yjs";
31
32import { atLeast } from "../access.ts";
33import { anchorThread, unanchorThread } from "../edits.ts";
34import type { FileStoreEnv } from "../files.ts";
35import { indexFolio, type IndexEnv } from "../indexer.ts";
36import { docFragment, docTarget, docTargets } from "../kinds/doc/index.ts";
37import { kindModel } from "../kinds/index.ts";
38import type { AgentForm, FolioOrigin, KindModel } from "../kinds/types.ts";
39import { saveFolio } from "../persist.ts";
40import { ROOM_MEMBER_HEADER, awarenessEntries } from "../room.ts";
41import { applyThreadAction, listThreads, setQuote, type ThreadResult } from "../threads.ts";
42import { notifyFolioMentions } from "./service.ts";
43
44export { ROOM_MEMBER_HEADER };
45
46export type FolioRoomMember = {
47 folio_id: string;
48 workspace_slug: string;
49 /** `user:<id>`. */
50 key: string;
51 member: MemberProfile;
52 role: DocRole;
53};
54
55type Attachment = FolioRoomMember & { clients: number[] };
56
57export type FolioRoomEnv = IndexEnv &
58 FileStoreEnv & {
59 DB: D1Database;
60 NOTIFY?: ServiceBinding;
61 EVENTS?: ServiceBinding;
62 IDENTITY: ServiceBinding;
63 REPOS?: ServiceBinding;
64 /** The rooms themselves, as the Worker binds them (mentions are checked through the service). */
65 FOLIOS: DurableObjectNamespace<FolioRoom>;
66 };
67
68const MESSAGE_SYNC = 0;
69const MESSAGE_AWARENESS = 1;
70const MESSAGE_QUERY_AWARENESS = 3;
71/** Save this long after the last change. */
72const SAVE_AFTER_MS = 4_000;
73/** Index this long after the text first changed: at most twice a minute while someone types. */
74const INDEX_AFTER_MS = 30_000;
75/** Compact the stored updates into one snapshot past this many. */
76const COMPACT_AT = 300;
77
78export class FolioRoom extends DurableObject<FolioRoomEnv> {
79 private doc: Y.Doc | null = null;
80 /** The last awareness update each client sent. Lost on hibernation; clients resend every 15 s. */
81 private awareness = new Map<number, Uint8Array>();
82 private clocks = new Map<number, number>();
83
84 constructor(ctx: DurableObjectState, env: FolioRoomEnv) {
85 super(ctx, env);
86 ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
87 ctx.blockConcurrencyWhile(async () => {
88 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS updates (seq INTEGER PRIMARY KEY AUTOINCREMENT, data BLOB NOT NULL)");
89 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS snapshot (id INTEGER PRIMARY KEY CHECK (id = 1), data BLOB NOT NULL)");
90 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)");
91 });
92 }
93
94 // ── State ──────────────────────────────────────────────────────────────
95
96 private meta<T>(key: string, fallback: T): T {
97 const row = this.ctx.storage.sql.exec<{ value: string }>("SELECT value FROM meta WHERE key = ?", key).toArray()[0];
98 if (!row) return fallback;
99 try {
100 return JSON.parse(row.value) as T;
101 } catch {
102 return fallback;
103 }
104 }
105
106 private setMeta(key: string, value: unknown): void {
107 this.ctx.storage.sql.exec("INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", key, JSON.stringify(value));
108 }
109
110 /** The kind's model. Every room is named and given its kind by `ensure` (or a socket) before anything else. */
111 private model(): KindModel {
112 const model = kindModel(this.meta<string | null>("kind", null));
113 if (!model) throw new Error(`folio room ${this.meta<string>("folio_id", "?")} has no kind it knows`);
114 return model;
115 }
116
117 private load(): Y.Doc {
118 if (this.doc) return this.doc;
119 const doc = new Y.Doc({ gc: true });
120 const snapshot = this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM snapshot WHERE id = 1").toArray()[0];
121 if (snapshot) Y.applyUpdate(doc, new Uint8Array(snapshot.data));
122 for (const row of this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM updates ORDER BY seq")) {
123 Y.applyUpdate(doc, new Uint8Array(row.data));
124 }
125 doc.on("update", (update: Uint8Array, origin: unknown) => this.onUpdate(update, origin));
126 this.doc = doc;
127 return doc;
128 }
129
130 private onUpdate(update: Uint8Array, origin: unknown): void {
131 this.ctx.storage.sql.exec("INSERT INTO updates (data) VALUES (?)", update);
132 const count = this.ctx.storage.sql.exec<{ n: number }>("SELECT COUNT(*) AS n FROM updates").one().n;
133 if (count >= COMPACT_AT) this.compact();
134 const member = origin instanceof WebSocket ? (origin.deserializeAttachment() as Attachment | null) : null;
135 const key = member ? member.key : (origin as FolioOrigin | null)?.key;
136 if (member?.member.kind === "user") {
137 const names = this.meta<string[]>("editor_names", []);
138 const name = member.member.name.toLowerCase();
139 if (!names.includes(name)) this.setMeta("editor_names", [...names, name]);
140 }
141 if (key) {
142 const editors = this.meta<string[]>("editors", []).filter((k) => k !== key);
143 editors.push(key);
144 this.setMeta("editors", editors);
145 const pending = this.meta<string[]>("pending_authors", []);
146 if (!pending.includes(key)) this.setMeta("pending_authors", [...pending, key]);
147 }
148 const encoder = encoding.createEncoder();
149 encoding.writeVarUint(encoder, MESSAGE_SYNC);
150 syncProtocol.writeUpdate(encoder, update);
151 this.send(encoding.toUint8Array(encoder), origin instanceof WebSocket ? origin : null);
152 this.alarmBy(Date.now() + SAVE_AFTER_MS);
153 }
154
155 private compact(): void {
156 const doc = this.load();
157 this.ctx.storage.transactionSync(() => {
158 this.ctx.storage.sql.exec("INSERT INTO snapshot (id, data) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET data = excluded.data", Y.encodeStateAsUpdate(doc));
159 this.ctx.storage.sql.exec("DELETE FROM updates");
160 });
161 }
162
163 private alarmBy(when: number): void {
164 void this.ctx.storage.getAlarm().then((at) => {
165 if (at == null || at > when) return this.ctx.storage.setAlarm(when);
166 });
167 }
168
169 private send(message: Uint8Array | string, except: WebSocket | null = null): void {
170 for (const socket of this.ctx.getWebSockets()) {
171 if (socket === except) continue;
172 try {
173 socket.send(message);
174 } catch {
175 // Closing already.
176 }
177 }
178 }
179
180 /** Saves to D1 now: the rendition, search, links, citations, and a version if one is due or asked for. */
181 private async persist(version: FolioOrigin | null = null): Promise<string | null> {
182 const folioId = this.meta<string | null>("folio_id", null);
183 if (!folioId || !kindModel(this.meta<string | null>("kind", null))) return null;
184 const doc = this.load();
185 const pending = this.meta<string[]>("pending_authors", []);
186 const result = await saveFolio(this.env, {
187 folio_id: folioId,
188 rendition: this.model().render(doc),
189 editors: this.meta<string[]>("editors", []),
190 editor_names: this.meta<string[]>("editor_names", []),
191 state: Y.encodeStateAsUpdate(doc),
192 version: version ? { kind: version.kind, note: version.note, authors: version.authors ?? [version.key] } : null,
193 pending_authors: pending,
194 last_version_at: this.meta<number>("last_version_at", 0),
195 workspace_slug: this.meta<string | null>("workspace_slug", null),
196 });
197 this.setMeta("editors", []);
198 this.setMeta("editor_names", []);
199 if (result.changed && this.meta<number | null>("index_at", null) == null) {
200 const when = Date.now() + INDEX_AFTER_MS;
201 this.setMeta("index_at", when);
202 this.alarmBy(when);
203 }
204 if (result.version_id) {
205 this.setMeta("last_version_at", Date.now());
206 this.setMeta("pending_authors", []);
207 }
208 if (result.mentioned.length) {
209 // Told only if they can read it (the service checks), so this waits on nothing.
210 const slug = this.meta<string | null>("workspace_slug", null);
211 if (slug) {
212 this.ctx.waitUntil(notifyFolioMentions(this.env, slug, folioId, result.mentioned, result.last).catch((error: unknown) => console.error("folios could not tell people they were mentioned", String(error))));
213 }
214 }
215 return result.version_id;
216 }
217
218 override async alarm(): Promise<void> {
219 await this.persist();
220 const due = this.meta<number | null>("index_at", null);
221 if (due == null) return;
222 if (Date.now() < due) return this.alarmBy(due);
223 this.setMeta("index_at", null);
224 const folioId = this.meta<string | null>("folio_id", null);
225 if (folioId) await indexFolio(this.env, folioId);
226 }
227
228 // ── Calls from the Worker ──────────────────────────────────────────────
229
230 /**
231 * Names the folio and its kind, and fills an empty document: from a
232 * Yjs state (a duplicate, or what D1 saved), or the kind's seed from
233 * text or a spec (a template, an agent's new folio, or blank). Returns
234 * whether the document has content afterwards: called with nothing to
235 * fill it with, an empty room says false and the service reads what
236 * was saved, so the usual open costs one call and no D1 read.
237 */
238 async ensure(init: { folio_id: string; kind: FolioKind; workspace_slug: string; text?: string | null; spec?: unknown; state?: Uint8Array | null }): Promise<boolean> {
239 this.setMeta("folio_id", init.folio_id);
240 this.setMeta("kind", init.kind);
241 if (init.workspace_slug) this.setMeta("workspace_slug", init.workspace_slug);
242 const doc = this.load();
243 const model = this.model();
244 if (!model.isEmpty(doc)) return true;
245 if (init.text === undefined && init.spec === undefined && init.state == null) return false;
246 // No origin: filling a new room is its "created" version (written by the service), nobody's edit.
247 if (init.state) Y.applyUpdate(doc, init.state);
248 else doc.transact(() => model.seed(doc, { text: init.text ?? null, spec: init.spec }));
249 return true;
250 }
251
252 /** The folio in its agent form. */
253 async read(): Promise<AgentForm> {
254 return this.model().read(this.load());
255 }
256
257 /** The text rendition now (unsaved edits included). */
258 async text(): Promise<string> {
259 return this.model().render(this.load()).text;
260 }
261
262 /** The whole document's state, for a duplicate. */
263 async state(): Promise<Uint8Array> {
264 return Y.encodeStateAsUpdate(this.load());
265 }
266
267 /** A doc's target now, or null when it is gone (doc kind only). */
268 async target(target: DocEditTarget): Promise<{ markdown: string; block_ids: string[] } | null> {
269 if (this.model().kind !== "doc") return null;
270 return docTarget(this.load(), target);
271 }
272
273 async targets(targets: DocEditTarget[]): Promise<(string[] | null)[]> {
274 if (this.model().kind !== "doc") return targets.map(() => null);
275 return docTargets(this.load(), targets);
276 }
277
278 /** Applies an edit in the kind's terms and records a version for it. */
279 async edit(edit: FolioAgentEdit, origin: FolioOrigin): Promise<{ applied: boolean; version_id: string | null; summary: string }> {
280 const doc = this.load();
281 const model = this.model();
282 const result = model.applyAgentEdit(doc, edit, origin);
283 if (!result.applied) return { applied: false, version_id: null, summary: result.summary };
284 const version_id = await this.persist({ ...origin, note: origin.note ?? result.summary });
285 return { applied: true, version_id, summary: result.summary };
286 }
287
288 /** Makes the document what a version's was, as a new version. */
289 async restore(input: { state: Uint8Array | null; text: string }, origin: FolioOrigin): Promise<string | null> {
290 const doc = this.load();
291 const model = this.model();
292 if (input.state) {
293 const old = new Y.Doc();
294 Y.applyUpdate(old, input.state);
295 model.restore(doc, old, origin);
296 } else model.restoreText(doc, input.text, origin);
297 return this.persist(origin);
298 }
299
300 /** A comment operation from `actor` with `role`. Anchoring in the text is the doc kind's. */
301 async thread(actor: string, role: DocRole, action: DocThreadAction): Promise<ThreadResult> {
302 const doc = this.load();
303 const isDoc = this.model().kind === "doc";
304 if (action.op === "anchor") {
305 if (!atLeast(role, "comment")) return { ok: false, code: "forbidden", message: "You can read this but not comment on it." };
306 if (!isDoc) return { ok: false, code: "invalid", message: "Comments on this kind of artifact are pinned, not anchored in text." };
307 const thread = doc.getMap<Y.Map<unknown>>("threads").get(action.thread_id);
308 if (!thread) return { ok: false, code: "not_found", message: "That thread is gone." };
309 const quote = anchorThread(doc, docFragment(doc), action.anchor, action.head, action.thread_id);
310 if (quote) setQuote(doc, action.thread_id, quote);
311 return { ok: true, value: { quote } };
312 }
313 const result = applyThreadAction(doc, actor, role, action);
314 if (result.ok && action.op === "delete_thread" && isDoc) unanchorThread(doc, docFragment(doc), action.thread_id);
315 return result;
316 }
317
318 async threads(): Promise<ReturnType<typeof listThreads>> {
319 return listThreads(this.load());
320 }
321
322 /** Tells everyone with the folio open. */
323 async notice(event: FoliosLiveEvent): Promise<void> {
324 this.send(JSON.stringify(event));
325 }
326
327 /** Who has it open: each socket's member key and username, for re-checking access. */
328 async members(): Promise<{ key: string; name: string }[]> {
329 const out = new Map<string, string>();
330 for (const socket of this.ctx.getWebSockets()) {
331 const who = socket.deserializeAttachment() as Attachment | null;
332 if (who) out.set(who.key, who.member.name);
333 }
334 return [...out].map(([key, name]) => ({ key, name }));
335 }
336
337 /** A member's role changed (null: they can no longer read it): their sockets follow, or close with 4403. */
338 async setRole(key: string, role: DocRole | null): Promise<void> {
339 for (const socket of this.ctx.getWebSockets(key)) {
340 const who = socket.deserializeAttachment() as Attachment | null;
341 if (!who) continue;
342 if (!role) {
343 try {
344 socket.send(JSON.stringify({ type: "access", role: null } satisfies FoliosLiveEvent));
345 socket.close(4403, "No longer allowed");
346 } catch {
347 // Already closed.
348 }
349 continue;
350 }
351 if (who.role === role) continue;
352 socket.serializeAttachment({ ...who, role });
353 try {
354 socket.send(JSON.stringify({ type: "access", role } satisfies FoliosLiveEvent));
355 } catch {
356 // Closing.
357 }
358 }
359 }
360
361 /** Saves now, as before a version list, an export or the trash. */
362 async flush(): Promise<void> {
363 if (this.doc) await this.persist();
364 }
365
366 /** Closes every socket: the folio went to the trash. */
367 async closeAll(reason: string): Promise<void> {
368 for (const socket of this.ctx.getWebSockets()) {
369 try {
370 socket.send(JSON.stringify({ type: "folio.trashed", folio_id: this.meta<string>("folio_id", "") } satisfies FoliosLiveEvent));
371 socket.close(4410, reason);
372 } catch {
373 // Already closed.
374 }
375 }
376 }
377
378 /** Forgets everything: the folio was deleted for good. */
379 async destroy(): Promise<void> {
380 await this.closeAll("Deleted");
381 this.doc = null;
382 await this.ctx.storage.deleteAlarm();
383 await this.ctx.storage.deleteAll();
384 }
385
386 // ── Sockets ────────────────────────────────────────────────────────────
387
388 override async fetch(request: Request): Promise<Response> {
389 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") return new Response("Expected a WebSocket upgrade\n", { status: 426 });
390 let who: FolioRoomMember;
391 try {
392 who = JSON.parse(request.headers.get(ROOM_MEMBER_HEADER) ?? "") as FolioRoomMember;
393 } catch {
394 return new Response("Missing member\n", { status: 400 });
395 }
396 if (!kindModel(this.meta<string | null>("kind", null))) return new Response("Not ready\n", { status: 409 });
397 if (who.workspace_slug) this.setMeta("workspace_slug", who.workspace_slug);
398 const pair = new WebSocketPair();
399 const [client, server] = Object.values(pair) as [WebSocket, WebSocket];
400 this.ctx.acceptWebSocket(server, [who.key]);
401 server.serializeAttachment({ ...who, clients: [] } satisfies Attachment);
402 const doc = this.load();
403 const encoder = encoding.createEncoder();
404 encoding.writeVarUint(encoder, MESSAGE_SYNC);
405 syncProtocol.writeSyncStep1(encoder, doc);
406 server.send(encoding.toUint8Array(encoder));
407 for (const update of this.awareness.values()) {
408 const e = encoding.createEncoder();
409 encoding.writeVarUint(e, MESSAGE_AWARENESS);
410 encoding.writeVarUint8Array(e, update);
411 server.send(encoding.toUint8Array(e));
412 }
413 return new Response(null, { status: 101, webSocket: client });
414 }
415
416 override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
417 if (typeof message === "string") return;
418 const who = socket.deserializeAttachment() as Attachment | null;
419 if (!who) return;
420 const data = new Uint8Array(message);
421 const decoder = decoding.createDecoder(data);
422 const type = decoding.readVarUint(decoder);
423 if (type === MESSAGE_SYNC) {
424 // Viewers and commenters may ask for the document (step 1) but never change it.
425 const peek = decoding.createDecoder(data);
426 decoding.readVarUint(peek);
427 const step = decoding.readVarUint(peek);
428 if (step !== syncProtocol.messageYjsSyncStep1 && !atLeast(who.role, "edit")) return;
429 const doc = this.load();
430 const encoder = encoding.createEncoder();
431 encoding.writeVarUint(encoder, MESSAGE_SYNC);
432 syncProtocol.readSyncMessage(decoder, encoder, doc, socket);
433 if (encoding.length(encoder) > 1) socket.send(encoding.toUint8Array(encoder));
434 return;
435 }
436 if (type === MESSAGE_AWARENESS) {
437 const update = decoding.readVarUint8Array(decoder);
438 const entries = awarenessEntries(update);
439 const clients = entries.map((e) => e.id);
440 for (const e of entries) {
441 this.clocks.set(e.id, e.clock);
442 if (e.gone) this.awareness.delete(e.id);
443 else this.awareness.set(e.id, update);
444 }
445 const known = new Set(who.clients);
446 if (clients.some((id) => !known.has(id))) socket.serializeAttachment({ ...who, clients: [...new Set([...who.clients, ...clients])] });
447 const encoder = encoding.createEncoder();
448 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
449 encoding.writeVarUint8Array(encoder, update);
450 this.send(encoding.toUint8Array(encoder), socket);
451 return;
452 }
453 if (type === MESSAGE_QUERY_AWARENESS) {
454 for (const update of this.awareness.values()) {
455 const encoder = encoding.createEncoder();
456 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
457 encoding.writeVarUint8Array(encoder, update);
458 socket.send(encoding.toUint8Array(encoder));
459 }
460 }
461 }
462
463 override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> {
464 this.leave(socket);
465 try {
466 socket.close(code, reason);
467 } catch {
468 // Already closed.
469 }
470 }
471
472 override async webSocketError(socket: WebSocket): Promise<void> {
473 this.leave(socket);
474 }
475
476 private leave(socket: WebSocket): void {
477 const who = socket.deserializeAttachment() as Attachment | null;
478 if (!who?.clients.length) return;
479 for (const id of who.clients) this.awareness.delete(id);
480 const gone = who.clients.filter((id) => this.clocks.has(id));
481 if (!gone.length) return;
482 const encoder = encoding.createEncoder();
483 encoding.writeVarUint(encoder, gone.length);
484 for (const id of gone) {
485 encoding.writeVarUint(encoder, id);
486 encoding.writeVarUint(encoder, this.clocks.get(id)! + 1);
487 encoding.writeVarString(encoder, "null");
488 this.clocks.delete(id);
489 }
490 const message = encoding.createEncoder();
491 encoding.writeVarUint(message, MESSAGE_AWARENESS);
492 encoding.writeVarUint8Array(message, encoding.toUint8Array(encoder));
493 this.send(encoding.toUint8Array(message), socket);
494 }
495}