Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
558 linesCodeBlameRaw
1/**
2 * One page's live room, as a Durable Object named by the page id.
3 *
4 * It owns the page's Yjs document: every editor's socket syncs with it
5 * (the y-protocols sync and awareness messages, as binary frames), and
6 * every change made on the server (an agent's edit, an accepted
7 * suggestion, a restore, a comment) is applied here, so all of them merge
8 * as one CRDT. It holds sockets with the WebSocket Hibernation API, so a
9 * page people have open but aren't typing in costs nothing between
10 * keystrokes; the document is read back from storage when it wakes.
11 *
12 * Storage (the object's own SQLite): the document as a snapshot plus the
13 * updates since, compacted every so often. A few seconds after a burst of
14 * edits (an alarm), the room saves the Markdown rendition, the search
15 * index, backlinks and history to D1 (src/persist.ts). Half a minute after
16 * the Markdown first changes, it brings the page's passages in the
17 * semantic index up to date (src/indexer.ts), on the same alarm, so typing
18 * never waits on embedding.
19 *
20 * The room authorizes nothing about who may open the page: the Worker
21 * checks the viewer's role before forwarding a socket (src/index.ts,
22 * `live`) and puts it in ROOM_MEMBER_HEADER. The room enforces that role:
23 * a socket that may only view or comment never changes the document.
24 */
25import { DurableObject } from "cloudflare:workers";
26
27import type { DocEditTarget, DocRole, DocThreadAction, DocVersionKind, DocsLiveEvent, MemberProfile, ServiceBinding } from "@g1t/contracts";
28import * as decoding from "lib0/decoding";
29import * as encoding from "lib0/encoding";
30import * as syncProtocol from "y-protocols/sync";
31import * as Y from "yjs";
32
33import { atLeast } from "./access.ts";
34import { seed } from "./blocks.ts";
35import { anchorThread, applyEdit, findTarget, rangeIds, rangeMarkdown, restoreFrom, unanchorThread } from "./edits.ts";
36import { bodyCitations } from "./citations.ts";
37import { citationNodes, documentMarkdown, mentionedIds, outline, type Outline } from "./markdown.ts";
38import { indexPage, type DocsJob } from "./indexer.ts";
39import { save } from "./persist.ts";
40import { applyThreadAction, listThreads, setQuote, type ThreadResult } from "./threads.ts";
41
42/** Header the Worker sets on a socket it forwards: who it is, as JSON (`RoomMember`). */
43export const ROOM_MEMBER_HEADER = "x-g1t-docs-member";
44
45export type RoomMember = {
46 page_id: string;
47 workspace_slug: string;
48 /** `user:<id>`. */
49 key: string;
50 member: MemberProfile;
51 role: DocRole;
52};
53
54/** Who made a change on the server, for history. */
55export type Origin = { key: string; kind: DocVersionKind; note: string | null; authors?: string[] };
56
57type Attachment = RoomMember & { clients: number[] };
58
59type Env = { DB: D1Database; NOTIFY?: ServiceBinding; EVENTS?: ServiceBinding; AI?: Ai; VECTORS?: Vectorize; JOBS?: Queue<DocsJob> };
60
61const FRAGMENT = "document-store";
62const MESSAGE_SYNC = 0;
63const MESSAGE_AWARENESS = 1;
64const MESSAGE_QUERY_AWARENESS = 3;
65/** Save this long after the last change. */
66const SAVE_AFTER_MS = 4_000;
67/** Index the page's passages this long after its Markdown first changed: at most twice a minute while someone types. */
68const INDEX_AFTER_MS = 30_000;
69/** Compact the stored updates into one snapshot past this many. */
70const COMPACT_AT = 300;
71
72export class PageRoom extends DurableObject<Env> {
73 private doc: Y.Doc | null = null;
74 /** Saves and index runs in flight: destroy() waits for them, so none writes after it. */
75 private saves = new Set<Promise<unknown>>();
76 /** The last awareness update each client sent, so newcomers see everyone at once. Lost on hibernation; clients resend every 15 s. */
77 private awareness = new Map<number, Uint8Array>();
78 /** Each client's last awareness clock, to mark it gone with the next one. */
79 private clocks = new Map<number, number>();
80
81 constructor(ctx: DurableObjectState, env: Env) {
82 super(ctx, env);
83 ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
84 ctx.blockConcurrencyWhile(async () => {
85 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS updates (seq INTEGER PRIMARY KEY AUTOINCREMENT, data BLOB NOT NULL)");
86 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS snapshot (id INTEGER PRIMARY KEY CHECK (id = 1), data BLOB NOT NULL)");
87 this.ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)");
88 });
89 }
90
91 // ── State ──────────────────────────────────────────────────────────────
92
93 private meta<T>(key: string, fallback: T): T {
94 const row = this.ctx.storage.sql.exec<{ value: string }>("SELECT value FROM meta WHERE key = ?", key).toArray()[0];
95 if (!row) return fallback;
96 try {
97 return JSON.parse(row.value) as T;
98 } catch {
99 return fallback;
100 }
101 }
102
103 private setMeta(key: string, value: unknown): void {
104 if (key !== "erased" && this.meta("erased", false)) return;
105 this.ctx.storage.sql.exec("INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", key, JSON.stringify(value));
106 }
107
108 /** The document, read from storage the first time it is needed after waking. */
109 async forgetPerson(key: string, username: string): Promise<void> {
110 if (this.meta("erased", false)) return;
111 this.setMeta(`erased:${key}`, true);
112 await this.setRole(key, null);
113 await Promise.allSettled([...this.saves]);
114 for (const field of ["editors", "pending_authors"])
115 this.setMeta(
116 field,
117 this.meta<string[]>(field, []).filter((value) => value !== key),
118 );
119 this.setMeta(
120 "editor_names",
121 this.meta<string[]>("editor_names", []).filter((value) => value.toLowerCase() !== username.toLowerCase()),
122 );
123 }
124
125 async destroy(): Promise<void> {
126 // Synchronous tombstone comes first; no reentrant save can recreate state.
127 this.setMeta("erased", true);
128 for (const socket of this.ctx.getWebSockets()) {
129 try {
130 socket.close(4403, "Document erased");
131 } catch {}
132 }
133 await Promise.allSettled([...this.saves]);
134 this.doc?.destroy();
135 this.doc = null;
136 this.awareness.clear();
137 this.clocks.clear();
138 this.ctx.storage.transactionSync(() => {
139 this.ctx.storage.sql.exec("DELETE FROM updates");
140 this.ctx.storage.sql.exec("DELETE FROM snapshot");
141 this.ctx.storage.sql.exec("DELETE FROM meta WHERE key != 'erased'");
142 });
143 await this.ctx.storage.deleteAlarm();
144 }
145
146 private load(): Y.Doc {
147 if (this.meta("erased", false)) throw new Error("Document erased");
148 if (this.doc) return this.doc;
149 const doc = new Y.Doc({ gc: true });
150 const snapshot = this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM snapshot WHERE id = 1").toArray()[0];
151 if (snapshot) Y.applyUpdate(doc, new Uint8Array(snapshot.data));
152 for (const row of this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM updates ORDER BY seq")) {
153 Y.applyUpdate(doc, new Uint8Array(row.data));
154 }
155 doc.on("update", (update: Uint8Array, origin: unknown) => this.onUpdate(update, origin));
156 this.doc = doc;
157 return doc;
158 }
159
160 private fragment(): Y.XmlFragment {
161 return this.load().getXmlFragment(FRAGMENT);
162 }
163
164 private onUpdate(update: Uint8Array, origin: unknown): void {
165 if (this.meta("erased", false)) return;
166 this.ctx.storage.sql.exec("INSERT INTO updates (data) VALUES (?)", update);
167 const count = this.ctx.storage.sql.exec<{ n: number }>("SELECT COUNT(*) AS n FROM updates").one().n;
168 if (count >= COMPACT_AT) this.compact();
169 // Who changed it: a socket's member, or a change made here.
170 const member = origin instanceof WebSocket ? (origin.deserializeAttachment() as Attachment | null) : null;
171 const key = member ? member.key : (origin as Origin | null)?.key;
172 if (member?.member.kind === "user") {
173 const names = this.meta<string[]>("editor_names", []);
174 const name = member.member.name.toLowerCase();
175 if (!names.includes(name)) this.setMeta("editor_names", [...names, name]);
176 }
177 if (key) {
178 const editors = this.meta<string[]>("editors", []).filter((k) => k !== key);
179 editors.push(key);
180 this.setMeta("editors", editors);
181 const pending = this.meta<string[]>("pending_authors", []);
182 if (!pending.includes(key)) this.setMeta("pending_authors", [...pending, key]);
183 }
184 // Everyone else sees it at once.
185 const encoder = encoding.createEncoder();
186 encoding.writeVarUint(encoder, MESSAGE_SYNC);
187 syncProtocol.writeUpdate(encoder, update);
188 this.send(encoding.toUint8Array(encoder), origin instanceof WebSocket ? origin : null);
189 this.scheduleSave();
190 }
191
192 private compact(): void {
193 const doc = this.load();
194 this.ctx.storage.transactionSync(() => {
195 this.ctx.storage.sql.exec(
196 "INSERT INTO snapshot (id, data) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET data = excluded.data",
197 Y.encodeStateAsUpdate(doc),
198 );
199 this.ctx.storage.sql.exec("DELETE FROM updates");
200 });
201 }
202
203 private scheduleSave(): void {
204 this.alarmBy(Date.now() + SAVE_AFTER_MS);
205 }
206
207 /** Makes sure the alarm rings by `when`: the save's, or the index's, whichever is first. */
208 private alarmBy(when: number): void {
209 void this.ctx.storage.getAlarm().then((at) => {
210 if (this.meta("erased", false)) return;
211 if (at == null || at > when) return this.ctx.storage.setAlarm(when);
212 });
213 }
214
215 private send(message: Uint8Array | string, except: WebSocket | null = null): void {
216 for (const socket of this.ctx.getWebSockets()) {
217 if (socket === except) continue;
218 try {
219 socket.send(message);
220 } catch {
221 // Closing already.
222 }
223 }
224 }
225
226 /** Saves to D1 now: the Markdown, search, links, mentions, and a version if one is due or asked for. */
227 private async persist(version: Origin | null = null): Promise<string | null> {
228 if (this.meta("erased", false)) return null;
229 const saving = this.save(version);
230 this.saves.add(saving);
231 try {
232 return await saving;
233 } finally {
234 this.saves.delete(saving);
235 }
236 }
237
238 private async save(version: Origin | null = null): Promise<string | null> {
239 const pageId = this.meta<string | null>("page_id", null);
240 if (!pageId) return null;
241 const fragment = this.fragment();
242 const editors = this.meta<string[]>("editors", []);
243 const pending = this.meta<string[]>("pending_authors", []);
244 const markdown = documentMarkdown(fragment);
245 const result = await save(this.env, {
246 page_id: pageId,
247 markdown,
248 citations: bodyCitations(citationNodes(fragment), markdown),
249 editors,
250 mentioned: mentionedIds(fragment).users,
251 editor_names: this.meta<string[]>("editor_names", []),
252 state: Y.encodeStateAsUpdate(this.load()),
253 version: version ? { kind: version.kind, note: version.note, authors: version.authors ?? [version.key] } : null,
254 pending_authors: pending,
255 last_version_at: this.meta<number>("last_version_at", 0),
256 workspace_slug: this.meta<string | null>("workspace_slug", null),
257 });
258 this.setMeta("editors", []);
259 this.setMeta("editor_names", []);
260 if (result.changed && this.meta<number | null>("index_at", null) == null) {
261 const when = Date.now() + INDEX_AFTER_MS;
262 this.setMeta("index_at", when);
263 this.alarmBy(when);
264 }
265 if (result.version_id) {
266 this.setMeta("last_version_at", Date.now());
267 this.setMeta("pending_authors", []);
268 }
269 return result.version_id;
270 }
271
272 override async alarm(): Promise<void> {
273 if (this.meta("erased", false)) return;
274 await this.persist();
275 const due = this.meta<number | null>("index_at", null);
276 if (due == null) return;
277 if (Date.now() < due) return this.alarmBy(due);
278 this.setMeta("index_at", null);
279 const pageId = this.meta<string | null>("page_id", null);
280 // Never throws; a failure is retried on the next change.
281 if (pageId) {
282 const indexing = indexPage(this.env, pageId);
283 this.saves.add(indexing);
284 try {
285 await indexing;
286 } finally {
287 this.saves.delete(indexing);
288 }
289 }
290 }
291
292 // ── Calls from the Worker ──────────────────────────────────────────────
293
294 /**
295 * Names the page this room is for, and fills an empty document from
296 * Markdown (a template, an agent's new page, or a page made before its
297 * room existed) or from a Yjs state (a duplicate).
298 */
299 async ensure(init: { page_id: string; workspace_slug: string; markdown?: string | null; state?: Uint8Array | null }): Promise<void> {
300 this.setMeta("page_id", init.page_id);
301 if (init.workspace_slug) this.setMeta("workspace_slug", init.workspace_slug);
302 const doc = this.load();
303 const fragment = this.fragment();
304 if (fragment.length > 0) return;
305 if (init.state) Y.applyUpdate(doc, init.state, { key: "system", kind: "created", note: null } satisfies Origin);
306 else seed(doc, fragment, init.markdown ?? "");
307 }
308
309 /** The document as Markdown, and its top-level blocks. */
310 async read(): Promise<{ markdown: string; blocks: Outline[] }> {
311 const fragment = this.fragment();
312 return { markdown: documentMarkdown(fragment), blocks: outline(fragment) };
313 }
314
315 /** The whole document's state, for a duplicate. */
316 async state(): Promise<Uint8Array> {
317 return Y.encodeStateAsUpdate(this.load());
318 }
319
320 /** A target's current Markdown and blocks, or null when it is gone. */
321 async target(target: DocEditTarget): Promise<{ markdown: string; block_ids: string[] } | null> {
322 const fragment = this.fragment();
323 const range = findTarget(fragment, target);
324 if (!range) return null;
325 return { markdown: rangeMarkdown(fragment, range), block_ids: rangeIds(fragment, range) };
326 }
327
328 /** Where each target is now, for marking open suggestions in the editor. */
329 async targets(targets: DocEditTarget[]): Promise<(string[] | null)[]> {
330 const fragment = this.fragment();
331 return targets.map((t) => {
332 const range = findTarget(fragment, t);
333 return range ? rangeIds(fragment, range) : null;
334 });
335 }
336
337 /** Applies an edit and records a version for it. False when the target is gone. */
338 async edit(target: DocEditTarget, markdown: string, origin: Origin): Promise<{ applied: boolean; version_id: string | null }> {
339 const doc = this.load();
340 const applied = applyEdit(doc, this.fragment(), target, markdown, origin);
341 if (!applied) return { applied: false, version_id: null };
342 const version_id = await this.persist(origin);
343 return { applied, version_id };
344 }
345
346 /** Makes the document what a version's was, as a new version. */
347 async restore(input: { state: Uint8Array | null; markdown: string }, origin: Origin): Promise<string | null> {
348 const doc = this.load();
349 if (input.state) {
350 const old = new Y.Doc();
351 Y.applyUpdate(old, input.state);
352 restoreFrom(doc, this.fragment(), old.getXmlFragment(FRAGMENT), origin);
353 } else applyEdit(doc, this.fragment(), { kind: "document" }, input.markdown, origin);
354 return this.persist(origin);
355 }
356
357 /** A comment operation from `actor` with `role`. */
358 async thread(actor: string, role: DocRole, action: DocThreadAction): Promise<ThreadResult> {
359 const doc = this.load();
360 if (action.op === "anchor") {
361 if (!atLeast(role, "comment")) return { ok: false, code: "forbidden", message: "You can read this page but not comment on it." };
362 const thread = doc.getMap<Y.Map<unknown>>("threads").get(action.thread_id);
363 if (!thread) return { ok: false, code: "not_found", message: "That thread is gone." };
364 const quote = anchorThread(doc, this.fragment(), action.anchor, action.head, action.thread_id);
365 if (quote) setQuote(doc, action.thread_id, quote);
366 return { ok: true, value: { quote } };
367 }
368 const result = applyThreadAction(doc, actor, role, action);
369 if (result.ok && action.op === "delete_thread") unanchorThread(doc, this.fragment(), action.thread_id);
370 return result;
371 }
372
373 /** The page's threads, authors as member keys. */
374 async threads(): Promise<ReturnType<typeof listThreads>> {
375 return listThreads(this.load());
376 }
377
378 /** Tells everyone with the page open. */
379 async notice(event: DocsLiveEvent): Promise<void> {
380 this.send(JSON.stringify(event));
381 }
382
383 /** A member's role changed (null: they can no longer read it): their sockets follow. */
384 async setRole(key: string, role: DocRole | null): Promise<void> {
385 for (const socket of this.ctx.getWebSockets(key)) {
386 const who = socket.deserializeAttachment() as Attachment | null;
387 if (!who) continue;
388 if (!role) {
389 try {
390 socket.close(4403, "No longer allowed");
391 } catch {
392 // Already closed.
393 }
394 continue;
395 }
396 socket.serializeAttachment({ ...who, role });
397 try {
398 socket.send(JSON.stringify({ type: "access", role } satisfies DocsLiveEvent));
399 } catch {
400 // Closing.
401 }
402 }
403 }
404
405 /** Saves now, as before the page is archived or exported. */
406 async flush(): Promise<void> {
407 if (this.doc) await this.persist();
408 }
409
410 /** Closes every socket: the page went to the trash. */
411 async closeAll(reason: string): Promise<void> {
412 for (const socket of this.ctx.getWebSockets()) {
413 try {
414 socket.send(JSON.stringify({ type: "page.archived", page_id: this.meta<string>("page_id", "") } satisfies DocsLiveEvent));
415 socket.close(4410, reason);
416 } catch {
417 // Already closed.
418 }
419 }
420 }
421
422 // ── Sockets ────────────────────────────────────────────────────────────
423
424 override async fetch(request: Request): Promise<Response> {
425 if (this.meta("erased", false)) return new Response("Document erased", { status: 410 });
426 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
427 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
428 }
429 let who: RoomMember;
430 try {
431 who = JSON.parse(request.headers.get(ROOM_MEMBER_HEADER) ?? "") as RoomMember;
432 } catch {
433 return new Response("Missing member\n", { status: 400 });
434 }
435 if (this.meta(`erased:${who.key}`, false)) return new Response("Account erased", { status: 410 });
436 if (who.workspace_slug) this.setMeta("workspace_slug", who.workspace_slug);
437 if (who.page_id) this.setMeta("page_id", who.page_id);
438 const pair = new WebSocketPair();
439 const [client, server] = Object.values(pair) as [WebSocket, WebSocket];
440 this.ctx.acceptWebSocket(server, [who.key]);
441 server.serializeAttachment({ ...who, clients: [] } satisfies Attachment);
442 // Start syncing: our state vector, and everyone's presence so far.
443 const doc = this.load();
444 const encoder = encoding.createEncoder();
445 encoding.writeVarUint(encoder, MESSAGE_SYNC);
446 syncProtocol.writeSyncStep1(encoder, doc);
447 server.send(encoding.toUint8Array(encoder));
448 for (const update of this.awareness.values()) {
449 const e = encoding.createEncoder();
450 encoding.writeVarUint(e, MESSAGE_AWARENESS);
451 encoding.writeVarUint8Array(e, update);
452 server.send(encoding.toUint8Array(e));
453 }
454 return new Response(null, { status: 101, webSocket: client });
455 }
456
457 override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
458 // Text frames are for the Worker's notices only; clients speak binary.
459 if (typeof message === "string") return;
460 const who = socket.deserializeAttachment() as Attachment | null;
461 if (!who || this.meta("erased", false) || this.meta(`erased:${who.key}`, false)) return;
462 const data = new Uint8Array(message);
463 const decoder = decoding.createDecoder(data);
464 const type = decoding.readVarUint(decoder);
465 if (type === MESSAGE_SYNC) {
466 // Peek at the sync message: viewers and commenters may ask for the
467 // document (step 1) but never change it (step 2, update).
468 const peek = decoding.createDecoder(data);
469 decoding.readVarUint(peek);
470 const step = decoding.readVarUint(peek);
471 if (step !== syncProtocol.messageYjsSyncStep1 && !atLeast(who.role, "edit")) return;
472 const doc = this.load();
473 const encoder = encoding.createEncoder();
474 encoding.writeVarUint(encoder, MESSAGE_SYNC);
475 syncProtocol.readSyncMessage(decoder, encoder, doc, socket);
476 if (encoding.length(encoder) > 1) socket.send(encoding.toUint8Array(encoder));
477 return;
478 }
479 if (type === MESSAGE_AWARENESS) {
480 const update = decoding.readVarUint8Array(decoder);
481 // Remember which clients this socket speaks for, to clear them when it closes.
482 const entries = awarenessEntries(update);
483 const clients = entries.map((e) => e.id);
484 for (const e of entries) {
485 this.clocks.set(e.id, e.clock);
486 if (e.gone) this.awareness.delete(e.id);
487 else this.awareness.set(e.id, update);
488 }
489 const known = new Set(who.clients);
490 if (clients.some((id) => !known.has(id))) socket.serializeAttachment({ ...who, clients: [...new Set([...who.clients, ...clients])] });
491 const encoder = encoding.createEncoder();
492 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
493 encoding.writeVarUint8Array(encoder, update);
494 this.send(encoding.toUint8Array(encoder), socket);
495 return;
496 }
497 if (type === MESSAGE_QUERY_AWARENESS) {
498 for (const update of this.awareness.values()) {
499 const encoder = encoding.createEncoder();
500 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
501 encoding.writeVarUint8Array(encoder, update);
502 socket.send(encoding.toUint8Array(encoder));
503 }
504 }
505 }
506
507 override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> {
508 this.leave(socket);
509 try {
510 socket.close(code, reason);
511 } catch {
512 // Already closed.
513 }
514 }
515
516 override async webSocketError(socket: WebSocket): Promise<void> {
517 this.leave(socket);
518 }
519
520 /** Tells everyone a closed socket's people left: their cursors go. */
521 private leave(socket: WebSocket): void {
522 const who = socket.deserializeAttachment() as Attachment | null;
523 if (!who?.clients.length) return;
524 for (const id of who.clients) this.awareness.delete(id);
525 // Marks each client gone (state null, the next clock). One whose clock
526 // was lost to hibernation is left to time out in the others' editors
527 // (30 s), as y-protocols does for any client that stops renewing.
528 const gone = who.clients.filter((id) => this.clocks.has(id));
529 if (!gone.length) return;
530 const encoder = encoding.createEncoder();
531 encoding.writeVarUint(encoder, gone.length);
532 for (const id of gone) {
533 encoding.writeVarUint(encoder, id);
534 encoding.writeVarUint(encoder, this.clocks.get(id)! + 1);
535 encoding.writeVarString(encoder, "null");
536 this.clocks.delete(id);
537 }
538 const update = encoding.toUint8Array(encoder);
539 const message = encoding.createEncoder();
540 encoding.writeVarUint(message, MESSAGE_AWARENESS);
541 encoding.writeVarUint8Array(message, update);
542 this.send(encoding.toUint8Array(message), socket);
543 }
544}
545
546/** The clients in an awareness update (its format: count, then id, clock, JSON state). */
547export function awarenessEntries(update: Uint8Array): { id: number; clock: number; gone: boolean }[] {
548 const decoder = decoding.createDecoder(update);
549 const count = decoding.readVarUint(decoder);
550 const out: { id: number; clock: number; gone: boolean }[] = [];
551 for (let i = 0; i < count; i++) {
552 const id = decoding.readVarUint(decoder);
553 const clock = decoding.readVarUint(decoder);
554 const state = decoding.readVarString(decoder);
555 out.push({ id, clock, gone: state === "null" });
556 }
557 return out;
558}