// 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.//// Every SNAPSHOT_EVERY versions the room also copies the note's text to the SNAPSHOTS bucket// (snapshots.ts), after the editors have their replies.import { DurableObject } from "cloudflare:workers";import type { Env } from "./index.ts";import type { Op } from "./ops.ts";import { LOG_SIZE, Room, type Note, type NoteStore } from "./room.ts";import { saveSnapshot } from "./snapshots.ts";export class NoteRoom extends DurableObject<Env> {  #room: Room;  // A snapshot is being written. Edits that arrive meanwhile don't start another one.  #saving = false;  constructor(ctx: DurableObjectState, env: 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();    this.ctx.acceptWebSocket(server);    server.send(JSON.stringify(this.#room.snapshot()));    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 out = this.#room.receive(data);    ws.send(JSON.stringify(out.reply));    if (out.broadcast) {      const frame = JSON.stringify(out.broadcast);      for (const other of this.ctx.getWebSockets()) {        if (other === ws) continue;        try {          other.send(frame);        } catch {          // The socket is closing; webSocketClose will follow.        }      }    }    if (out.saveSnapshot) await this.#snapshot(out.saveSnapshot);  }  // Copies the note to R2. A failure is logged and the snapshot stays due, so the next edit  // tries again; editing never waits on R2.  async #snapshot(note: Note): Promise<void> {    if (this.#saving) return;    this.#saving = true;    try {      await saveSnapshot(this.env.SNAPSHOTS, this.ctx.id.toString(), note);      this.#room.snapshotSaved(note.version);    } catch (e) {      console.error(JSON.stringify({ msg: "saving a snapshot failed", version: note.version, error: String(e) }));    } finally {      this.#saving = false;    }  }  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);  }}// 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);      CREATE TABLE IF NOT EXISTS snapshot (id INTEGER PRIMARY KEY CHECK (id = 1), version INTEGER NOT NULL);    `);  }  lastSnapshot(): number {    return this.#storage.sql.exec<{ version: number }>("SELECT version FROM snapshot WHERE id = 1").toArray()[0]?.version ?? 0;  }  snapshotSaved(version: number): void {    this.#storage.sql.exec("INSERT INTO snapshot (id, version) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET version = excluded.version", version);  }  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);    });  }}