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.

g1t.sh
gitagentscloudflarerustforgemcp
537 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"]) this.setMeta(field, this.meta<string[]>(field, []).filter(value => value !== key));
115 this.setMeta("editor_names", this.meta<string[]>("editor_names", []).filter(value => value.toLowerCase() !== username.toLowerCase()));
116 }
117
118 async destroy(): Promise<void> {
119 // Synchronous tombstone comes first; no reentrant save can recreate state.
120 this.setMeta("erased", true);
121 for (const socket of this.ctx.getWebSockets()) { try { socket.close(4403, "Document erased"); } catch {} }
122 await Promise.allSettled([...this.saves]);
123 this.doc?.destroy();
124 this.doc = null;
125 this.awareness.clear();
126 this.clocks.clear();
127 this.ctx.storage.transactionSync(() => {
128 this.ctx.storage.sql.exec("DELETE FROM updates");
129 this.ctx.storage.sql.exec("DELETE FROM snapshot");
130 this.ctx.storage.sql.exec("DELETE FROM meta WHERE key != 'erased'");
131 });
132 await this.ctx.storage.deleteAlarm();
133 }
134
135 private load(): Y.Doc {
136 if (this.meta("erased", false)) throw new Error("Document erased");
137 if (this.doc) return this.doc;
138 const doc = new Y.Doc({ gc: true });
139 const snapshot = this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM snapshot WHERE id = 1").toArray()[0];
140 if (snapshot) Y.applyUpdate(doc, new Uint8Array(snapshot.data));
141 for (const row of this.ctx.storage.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM updates ORDER BY seq")) {
142 Y.applyUpdate(doc, new Uint8Array(row.data));
143 }
144 doc.on("update", (update: Uint8Array, origin: unknown) => this.onUpdate(update, origin));
145 this.doc = doc;
146 return doc;
147 }
148
149 private fragment(): Y.XmlFragment {
150 return this.load().getXmlFragment(FRAGMENT);
151 }
152
153 private onUpdate(update: Uint8Array, origin: unknown): void {
154 if (this.meta("erased", false)) return;
155 this.ctx.storage.sql.exec("INSERT INTO updates (data) VALUES (?)", update);
156 const count = this.ctx.storage.sql.exec<{ n: number }>("SELECT COUNT(*) AS n FROM updates").one().n;
157 if (count >= COMPACT_AT) this.compact();
158 // Who changed it: a socket's member, or a change made here.
159 const member = origin instanceof WebSocket ? (origin.deserializeAttachment() as Attachment | null) : null;
160 const key = member ? member.key : (origin as Origin | null)?.key;
161 if (member?.member.kind === "user") {
162 const names = this.meta<string[]>("editor_names", []);
163 const name = member.member.name.toLowerCase();
164 if (!names.includes(name)) this.setMeta("editor_names", [...names, name]);
165 }
166 if (key) {
167 const editors = this.meta<string[]>("editors", []).filter((k) => k !== key);
168 editors.push(key);
169 this.setMeta("editors", editors);
170 const pending = this.meta<string[]>("pending_authors", []);
171 if (!pending.includes(key)) this.setMeta("pending_authors", [...pending, key]);
172 }
173 // Everyone else sees it at once.
174 const encoder = encoding.createEncoder();
175 encoding.writeVarUint(encoder, MESSAGE_SYNC);
176 syncProtocol.writeUpdate(encoder, update);
177 this.send(encoding.toUint8Array(encoder), origin instanceof WebSocket ? origin : null);
178 this.scheduleSave();
179 }
180
181 private compact(): void {
182 const doc = this.load();
183 this.ctx.storage.transactionSync(() => {
184 this.ctx.storage.sql.exec("INSERT INTO snapshot (id, data) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET data = excluded.data", Y.encodeStateAsUpdate(doc));
185 this.ctx.storage.sql.exec("DELETE FROM updates");
186 });
187 }
188
189 private scheduleSave(): void {
190 this.alarmBy(Date.now() + SAVE_AFTER_MS);
191 }
192
193 /** Makes sure the alarm rings by `when`: the save's, or the index's, whichever is first. */
194 private alarmBy(when: number): void {
195 void this.ctx.storage.getAlarm().then((at) => {
196 if (this.meta("erased", false)) return;
197 if (at == null || at > when) return this.ctx.storage.setAlarm(when);
198 });
199 }
200
201 private send(message: Uint8Array | string, except: WebSocket | null = null): void {
202 for (const socket of this.ctx.getWebSockets()) {
203 if (socket === except) continue;
204 try {
205 socket.send(message);
206 } catch {
207 // Closing already.
208 }
209 }
210 }
211
212 /** Saves to D1 now: the Markdown, search, links, mentions, and a version if one is due or asked for. */
213 private async persist(version: Origin | null = null): Promise<string | null> {
214 if (this.meta("erased", false)) return null;
215 const saving = this.save(version);
216 this.saves.add(saving);
217 try { return await saving; } finally { this.saves.delete(saving); }
218 }
219
220 private async save(version: Origin | null = null): Promise<string | null> {
221 const pageId = this.meta<string | null>("page_id", null);
222 if (!pageId) return null;
223 const fragment = this.fragment();
224 const editors = this.meta<string[]>("editors", []);
225 const pending = this.meta<string[]>("pending_authors", []);
226 const markdown = documentMarkdown(fragment);
227 const result = await save(this.env, {
228 page_id: pageId,
229 markdown,
230 citations: bodyCitations(citationNodes(fragment), markdown),
231 editors,
232 mentioned: mentionedIds(fragment).users,
233 editor_names: this.meta<string[]>("editor_names", []),
234 state: Y.encodeStateAsUpdate(this.load()),
235 version: version ? { kind: version.kind, note: version.note, authors: version.authors ?? [version.key] } : null,
236 pending_authors: pending,
237 last_version_at: this.meta<number>("last_version_at", 0),
238 workspace_slug: this.meta<string | null>("workspace_slug", null),
239 });
240 this.setMeta("editors", []);
241 this.setMeta("editor_names", []);
242 if (result.changed && this.meta<number | null>("index_at", null) == null) {
243 const when = Date.now() + INDEX_AFTER_MS;
244 this.setMeta("index_at", when);
245 this.alarmBy(when);
246 }
247 if (result.version_id) {
248 this.setMeta("last_version_at", Date.now());
249 this.setMeta("pending_authors", []);
250 }
251 return result.version_id;
252 }
253
254 override async alarm(): Promise<void> {
255 if (this.meta("erased", false)) return;
256 await this.persist();
257 const due = this.meta<number | null>("index_at", null);
258 if (due == null) return;
259 if (Date.now() < due) return this.alarmBy(due);
260 this.setMeta("index_at", null);
261 const pageId = this.meta<string | null>("page_id", null);
262 // Never throws; a failure is retried on the next change.
263 if (pageId) {
264 const indexing = indexPage(this.env, pageId);
265 this.saves.add(indexing);
266 try { await indexing; } finally { this.saves.delete(indexing); }
267 }
268 }
269
270 // ── Calls from the Worker ──────────────────────────────────────────────
271
272 /**
273 * Names the page this room is for, and fills an empty document from
274 * Markdown (a template, an agent's new page, or a page made before its
275 * room existed) or from a Yjs state (a duplicate).
276 */
277 async ensure(init: { page_id: string; workspace_slug: string; markdown?: string | null; state?: Uint8Array | null }): Promise<void> {
278 this.setMeta("page_id", init.page_id);
279 if (init.workspace_slug) this.setMeta("workspace_slug", init.workspace_slug);
280 const doc = this.load();
281 const fragment = this.fragment();
282 if (fragment.length > 0) return;
283 if (init.state) Y.applyUpdate(doc, init.state, { key: "system", kind: "created", note: null } satisfies Origin);
284 else seed(doc, fragment, init.markdown ?? "");
285 }
286
287 /** The document as Markdown, and its top-level blocks. */
288 async read(): Promise<{ markdown: string; blocks: Outline[] }> {
289 const fragment = this.fragment();
290 return { markdown: documentMarkdown(fragment), blocks: outline(fragment) };
291 }
292
293 /** The whole document's state, for a duplicate. */
294 async state(): Promise<Uint8Array> {
295 return Y.encodeStateAsUpdate(this.load());
296 }
297
298 /** A target's current Markdown and blocks, or null when it is gone. */
299 async target(target: DocEditTarget): Promise<{ markdown: string; block_ids: string[] } | null> {
300 const fragment = this.fragment();
301 const range = findTarget(fragment, target);
302 if (!range) return null;
303 return { markdown: rangeMarkdown(fragment, range), block_ids: rangeIds(fragment, range) };
304 }
305
306 /** Where each target is now, for marking open suggestions in the editor. */
307 async targets(targets: DocEditTarget[]): Promise<(string[] | null)[]> {
308 const fragment = this.fragment();
309 return targets.map((t) => {
310 const range = findTarget(fragment, t);
311 return range ? rangeIds(fragment, range) : null;
312 });
313 }
314
315 /** Applies an edit and records a version for it. False when the target is gone. */
316 async edit(target: DocEditTarget, markdown: string, origin: Origin): Promise<{ applied: boolean; version_id: string | null }> {
317 const doc = this.load();
318 const applied = applyEdit(doc, this.fragment(), target, markdown, origin);
319 if (!applied) return { applied: false, version_id: null };
320 const version_id = await this.persist(origin);
321 return { applied, version_id };
322 }
323
324 /** Makes the document what a version's was, as a new version. */
325 async restore(input: { state: Uint8Array | null; markdown: string }, origin: Origin): Promise<string | null> {
326 const doc = this.load();
327 if (input.state) {
328 const old = new Y.Doc();
329 Y.applyUpdate(old, input.state);
330 restoreFrom(doc, this.fragment(), old.getXmlFragment(FRAGMENT), origin);
331 } else applyEdit(doc, this.fragment(), { kind: "document" }, input.markdown, origin);
332 return this.persist(origin);
333 }
334
335 /** A comment operation from `actor` with `role`. */
336 async thread(actor: string, role: DocRole, action: DocThreadAction): Promise<ThreadResult> {
337 const doc = this.load();
338 if (action.op === "anchor") {
339 if (!atLeast(role, "comment")) return { ok: false, code: "forbidden", message: "You can read this page but not comment on it." };
340 const thread = doc.getMap<Y.Map<unknown>>("threads").get(action.thread_id);
341 if (!thread) return { ok: false, code: "not_found", message: "That thread is gone." };
342 const quote = anchorThread(doc, this.fragment(), action.anchor, action.head, action.thread_id);
343 if (quote) setQuote(doc, action.thread_id, quote);
344 return { ok: true, value: { quote } };
345 }
346 const result = applyThreadAction(doc, actor, role, action);
347 if (result.ok && action.op === "delete_thread") unanchorThread(doc, this.fragment(), action.thread_id);
348 return result;
349 }
350
351 /** The page's threads, authors as member keys. */
352 async threads(): Promise<ReturnType<typeof listThreads>> {
353 return listThreads(this.load());
354 }
355
356 /** Tells everyone with the page open. */
357 async notice(event: DocsLiveEvent): Promise<void> {
358 this.send(JSON.stringify(event));
359 }
360
361 /** A member's role changed (null: they can no longer read it): their sockets follow. */
362 async setRole(key: string, role: DocRole | null): Promise<void> {
363 for (const socket of this.ctx.getWebSockets(key)) {
364 const who = socket.deserializeAttachment() as Attachment | null;
365 if (!who) continue;
366 if (!role) {
367 try {
368 socket.close(4403, "No longer allowed");
369 } catch {
370 // Already closed.
371 }
372 continue;
373 }
374 socket.serializeAttachment({ ...who, role });
375 try {
376 socket.send(JSON.stringify({ type: "access", role } satisfies DocsLiveEvent));
377 } catch {
378 // Closing.
379 }
380 }
381 }
382
383 /** Saves now, as before the page is archived or exported. */
384 async flush(): Promise<void> {
385 if (this.doc) await this.persist();
386 }
387
388 /** Closes every socket: the page went to the trash. */
389 async closeAll(reason: string): Promise<void> {
390 for (const socket of this.ctx.getWebSockets()) {
391 try {
392 socket.send(JSON.stringify({ type: "page.archived", page_id: this.meta<string>("page_id", "") } satisfies DocsLiveEvent));
393 socket.close(4410, reason);
394 } catch {
395 // Already closed.
396 }
397 }
398 }
399
400 // ── Sockets ────────────────────────────────────────────────────────────
401
402 override async fetch(request: Request): Promise<Response> {
403 if (this.meta("erased", false)) return new Response("Document erased", { status: 410 });
404 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
405 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
406 }
407 let who: RoomMember;
408 try {
409 who = JSON.parse(request.headers.get(ROOM_MEMBER_HEADER) ?? "") as RoomMember;
410 } catch {
411 return new Response("Missing member\n", { status: 400 });
412 }
413 if (this.meta(`erased:${who.key}`, false)) return new Response("Account erased", { status: 410 });
414 if (who.workspace_slug) this.setMeta("workspace_slug", who.workspace_slug);
415 if (who.page_id) this.setMeta("page_id", who.page_id);
416 const pair = new WebSocketPair();
417 const [client, server] = Object.values(pair) as [WebSocket, WebSocket];
418 this.ctx.acceptWebSocket(server, [who.key]);
419 server.serializeAttachment({ ...who, clients: [] } satisfies Attachment);
420 // Start syncing: our state vector, and everyone's presence so far.
421 const doc = this.load();
422 const encoder = encoding.createEncoder();
423 encoding.writeVarUint(encoder, MESSAGE_SYNC);
424 syncProtocol.writeSyncStep1(encoder, doc);
425 server.send(encoding.toUint8Array(encoder));
426 for (const update of this.awareness.values()) {
427 const e = encoding.createEncoder();
428 encoding.writeVarUint(e, MESSAGE_AWARENESS);
429 encoding.writeVarUint8Array(e, update);
430 server.send(encoding.toUint8Array(e));
431 }
432 return new Response(null, { status: 101, webSocket: client });
433 }
434
435 override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
436 // Text frames are for the Worker's notices only; clients speak binary.
437 if (typeof message === "string") return;
438 const who = socket.deserializeAttachment() as Attachment | null;
439 if (!who || this.meta("erased", false) || this.meta(`erased:${who.key}`, false)) return;
440 const data = new Uint8Array(message);
441 const decoder = decoding.createDecoder(data);
442 const type = decoding.readVarUint(decoder);
443 if (type === MESSAGE_SYNC) {
444 // Peek at the sync message: viewers and commenters may ask for the
445 // document (step 1) but never change it (step 2, update).
446 const peek = decoding.createDecoder(data);
447 decoding.readVarUint(peek);
448 const step = decoding.readVarUint(peek);
449 if (step !== syncProtocol.messageYjsSyncStep1 && !atLeast(who.role, "edit")) return;
450 const doc = this.load();
451 const encoder = encoding.createEncoder();
452 encoding.writeVarUint(encoder, MESSAGE_SYNC);
453 syncProtocol.readSyncMessage(decoder, encoder, doc, socket);
454 if (encoding.length(encoder) > 1) socket.send(encoding.toUint8Array(encoder));
455 return;
456 }
457 if (type === MESSAGE_AWARENESS) {
458 const update = decoding.readVarUint8Array(decoder);
459 // Remember which clients this socket speaks for, to clear them when it closes.
460 const entries = awarenessEntries(update);
461 const clients = entries.map((e) => e.id);
462 for (const e of entries) {
463 this.clocks.set(e.id, e.clock);
464 if (e.gone) this.awareness.delete(e.id);
465 else this.awareness.set(e.id, update);
466 }
467 const known = new Set(who.clients);
468 if (clients.some((id) => !known.has(id))) socket.serializeAttachment({ ...who, clients: [...new Set([...who.clients, ...clients])] });
469 const encoder = encoding.createEncoder();
470 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
471 encoding.writeVarUint8Array(encoder, update);
472 this.send(encoding.toUint8Array(encoder), socket);
473 return;
474 }
475 if (type === MESSAGE_QUERY_AWARENESS) {
476 for (const update of this.awareness.values()) {
477 const encoder = encoding.createEncoder();
478 encoding.writeVarUint(encoder, MESSAGE_AWARENESS);
479 encoding.writeVarUint8Array(encoder, update);
480 socket.send(encoding.toUint8Array(encoder));
481 }
482 }
483 }
484
485 override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> {
486 this.leave(socket);
487 try {
488 socket.close(code, reason);
489 } catch {
490 // Already closed.
491 }
492 }
493
494 override async webSocketError(socket: WebSocket): Promise<void> {
495 this.leave(socket);
496 }
497
498 /** Tells everyone a closed socket's people left: their cursors go. */
499 private leave(socket: WebSocket): void {
500 const who = socket.deserializeAttachment() as Attachment | null;
501 if (!who?.clients.length) return;
502 for (const id of who.clients) this.awareness.delete(id);
503 // Marks each client gone (state null, the next clock). One whose clock
504 // was lost to hibernation is left to time out in the others' editors
505 // (30 s), as y-protocols does for any client that stops renewing.
506 const gone = who.clients.filter((id) => this.clocks.has(id));
507 if (!gone.length) return;
508 const encoder = encoding.createEncoder();
509 encoding.writeVarUint(encoder, gone.length);
510 for (const id of gone) {
511 encoding.writeVarUint(encoder, id);
512 encoding.writeVarUint(encoder, this.clocks.get(id)! + 1);
513 encoding.writeVarString(encoder, "null");
514 this.clocks.delete(id);
515 }
516 const update = encoding.toUint8Array(encoder);
517 const message = encoding.createEncoder();
518 encoding.writeVarUint(message, MESSAGE_AWARENESS);
519 encoding.writeVarUint8Array(message, update);
520 this.send(encoding.toUint8Array(message), socket);
521 }
522}
523
524/** The clients in an awareness update (its format: count, then id, clock, JSON state). */
525export function awarenessEntries(update: Uint8Array): { id: number; clock: number; gone: boolean }[] {
526 const decoder = decoding.createDecoder(update);
527 const count = decoding.readVarUint(decoder);
528 const out: { id: number; clock: number; gone: boolean }[] = [];
529 for (let i = 0; i < count; i++) {
530 const id = decoding.readVarUint(decoder);
531 const clock = decoding.readVarUint(decoder);
532 const state = decoding.readVarString(decoder);
533 out.push({ id, clock, gone: state === "null" });
534 }
535 return out;
536}
537