examples/live-notes Public
// NoteRoom: the Durable Object behind one note. It owns the note's storage (SQLite) and its// editors' WebSockets, and hands every edit to a Room (room.ts), which decides what happens.//// Sockets are accepted with the Hibernation API, so an idle note costs nothing: the runtime// keeps the connections open while the object is evicted, and wakes it with webSocketMessage// when a frame arrives. Everything the room needs lives in SQLite, so the constructor below is// also the wake-up path. Who is connected lives on the sockets themselves: each one carries// its editor's name and color as an attachment, which hibernation keeps.import { DurableObject } from "cloudflare:workers";import type { Op } from "./ops.ts";import { asEditor, joinMessages, pickEditor, presence } from "./presence.ts";import type { Editor } from "./protocol.ts";import { LOG_SIZE, Room, type Note, type NoteStore } from "./room.ts";export class NoteRoom extends DurableObject { #room: Room; constructor(ctx: DurableObjectState, env: Cloudflare.Env) { super(ctx, env); this.#room = new Room(new SqlNoteStore(ctx.storage)); } // GET /notes/:id, over RPC from the Worker. read(): Note { return this.#room.note; } // GET /notes/:id/ws: the Worker forwards the upgrade request here. async fetch(request: Request): Promise<Response> { if (request.headers.get("Upgrade") !== "websocket") return new Response("expected a WebSocket upgrade", { status: 426 }); const { 0: client, 1: server } = new WebSocketPair(); const here = this.#editors(); const editor = pickEditor(here, crypto.randomUUID().slice(0, 8)); this.ctx.acceptWebSocket(server); server.serializeAttachment(editor); const { you, others } = joinMessages(here, editor); server.send(JSON.stringify(this.#room.snapshot())); server.send(JSON.stringify(you)); this.#broadcast(JSON.stringify(others), server); return new Response(null, { status: 101, webSocket: client }); } async webSocketMessage(ws: WebSocket, data: string | ArrayBuffer): Promise<void> { if (typeof data !== "string") { ws.send(JSON.stringify({ type: "error", message: "frames are JSON text" })); return; } const by = asEditor(ws.deserializeAttachment())?.id ?? ""; const out = this.#room.receive(data, by); ws.send(JSON.stringify(out.reply)); if (out.broadcast) this.#broadcast(JSON.stringify(out.broadcast), ws); } async webSocketClose(ws: WebSocket, code: number): Promise<void> { // 1005 and 1006 report that no close code was sent; they can't be sent back. ws.close(code === 1005 || code === 1006 ? 1000 : code); this.#left(ws); } async webSocketError(ws: WebSocket): Promise<void> { this.#left(ws); } #left(ws: WebSocket): void { this.#broadcast(JSON.stringify(presence(this.#editors(ws))), ws); } // The editors on the open sockets, oldest first, leaving out `except`. #editors(except?: WebSocket): Editor[] { const here: Editor[] = []; for (const ws of this.ctx.getWebSockets()) { if (ws === except || ws.readyState !== WebSocket.OPEN) continue; const editor = asEditor(ws.deserializeAttachment()); if (editor) here.push(editor); } return here; } #broadcast(frame: string, except: WebSocket): void { for (const ws of this.ctx.getWebSockets()) { if (ws === except) continue; try { ws.send(frame); } catch { // The socket is closing; webSocketClose will follow. } } }}// The note's row and its op log in the object's SQLite database.class SqlNoteStore implements NoteStore { #storage: DurableObjectStorage; constructor(storage: DurableObjectStorage) { this.#storage = storage; storage.sql.exec(` CREATE TABLE IF NOT EXISTS note (id INTEGER PRIMARY KEY CHECK (id = 1), text TEXT NOT NULL, version INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS ops (version INTEGER PRIMARY KEY, op TEXT NOT NULL); `); } load(): Note { const row = this.#storage.sql.exec<{ text: string; version: number }>("SELECT text, version FROM note WHERE id = 1").toArray()[0]; return row ? { text: row.text, version: row.version } : { text: "", version: 0 }; } opsSince(version: number): Op[] { return this.#storage.sql .exec<{ op: string }>("SELECT op FROM ops WHERE version > ? ORDER BY version", version) .toArray() .map((r) => JSON.parse(r.op) as Op); } commit(note: Note, op: Op): void { this.#storage.transactionSync(() => { const sql = this.#storage.sql; sql.exec( "INSERT INTO note (id, text, version) VALUES (1, ?, ?) ON CONFLICT (id) DO UPDATE SET text = excluded.text, version = excluded.version", note.text, note.version, ); sql.exec("INSERT INTO ops (version, op) VALUES (?, ?)", note.version, JSON.stringify(op)); sql.exec("DELETE FROM ops WHERE version <= ?", note.version - LOG_SIZE); }); }}