examples/live-notes Public
Add the room that sequences edits to a note
One room per note applies edits one at a time: it transforms each op past the ops its sender had not seen yet, saves it, acks the sender and hands the op on for the other editors. It keeps the last 200 ops; an editor further behind gets a snapshot instead. Storage sits behind a small interface so the tests run in plain Node.
Changed files
5 files changed252 additions0 deletionsagainst 50b1d73
Addedsrc/protocol.ts+41−0
| @@ -0,0 +1,41 @@ | ||
| +// The messages editors and a note's room exchange over the WebSocket, as JSON text frames. | ||
| +// Every version is the room's: the number of ops it has applied to the note. | ||
| + | ||
| +import type { Op } from "./ops.ts"; | ||
| + | ||
| +// An edit, written against the note as it was at version `base`. | ||
| +export type ClientMessage = { type: "op"; base: number; op: Op }; | ||
| + | ||
| +export type ServerMessage = | ||
| + // The whole note: on connect, and whenever an editor's view can't be repaired. | ||
| + | { type: "snapshot"; text: string; version: number } | ||
| + // Your op was applied (or turned out to change nothing) and the note is now at `version`. | ||
| + | { type: "ack"; version: number } | ||
| + // Someone else's op, already transformed, which took the note to `version`. | ||
| + | { type: "op"; version: number; op: Op } | ||
| + | { type: "error"; message: string }; | ||
| + | ||
| +// Parses and checks a frame from an editor. Returns null for anything malformed; the room | ||
| +// never trusts the shape of what arrives on a socket. | ||
| +export function parseClientMessage(raw: string): ClientMessage | null { | ||
| + let m: unknown; | ||
| + try { | ||
| + m = JSON.parse(raw); | ||
| + } catch { | ||
| + return null; | ||
| + } | ||
| + if (!isObject(m) || m.type !== "op" || !isCount(m.base) || !isObject(m.op)) return null; | ||
| + const op = m.op; | ||
| + if (!isCount(op.pos)) return null; | ||
| + if (op.type === "insert" && typeof op.text === "string") return { type: "op", base: m.base, op: { type: "insert", pos: op.pos, text: op.text } }; | ||
| + if (op.type === "delete" && isCount(op.count)) return { type: "op", base: m.base, op: { type: "delete", pos: op.pos, count: op.count } }; | ||
| + return null; | ||
| +} | ||
| + | ||
| +function isObject(v: unknown): v is Record<string, unknown> { | ||
| + return typeof v === "object" && v !== null && !Array.isArray(v); | ||
| +} | ||
| + | ||
| +function isCount(v: unknown): v is number { | ||
| + return Number.isSafeInteger(v) && (v as number) >= 0; | ||
| +} | ||
Addedsrc/room.ts+71−0
| @@ -0,0 +1,71 @@ | ||
| +// The sequencer for one note. Every edit to a note passes through its room, one at a time, so | ||
| +// the room decides the order and every editor applies the same ops in that order. | ||
| +// | ||
| +// Room knows nothing about Durable Objects or sockets: it reads and writes through a NoteStore | ||
| +// and answers each message with an Outcome. NoteRoom (note-room.ts) supplies SQLite storage and | ||
| +// does the sending, and the tests supply an in-memory store. | ||
| + | ||
| +import { OpError, apply, isNoop, transformPast, type Op } from "./ops.ts"; | ||
| +import { parseClientMessage, type ServerMessage } from "./protocol.ts"; | ||
| + | ||
| +// How many recent ops the room keeps. An editor whose base version is older than that can't be | ||
| +// caught up op by op and gets a fresh snapshot instead. | ||
| +export const LOG_SIZE = 200; | ||
| + | ||
| +export type Note = { text: string; version: number }; | ||
| + | ||
| +export interface NoteStore { | ||
| + // The saved note, or { text: "", version: 0 } for one nobody has written yet. | ||
| + load(): Note; | ||
| + // The logged ops after `version`, oldest first. The room only asks for the last LOG_SIZE. | ||
| + opsSince(version: number): Op[]; | ||
| + // Saves the note at its new version together with the op that produced it, and trims the | ||
| + // log to the last LOG_SIZE ops. Must be atomic. | ||
| + commit(note: Note, op: Op): void; | ||
| +} | ||
| + | ||
| +// What to send after a message: `reply` goes back to the sender, `broadcast` to every other | ||
| +// editor of the note. | ||
| +export type Outcome = { reply: ServerMessage; broadcast?: ServerMessage }; | ||
| + | ||
| +export class Room { | ||
| + #store: NoteStore; | ||
| + #note: Note; | ||
| + | ||
| + constructor(store: NoteStore) { | ||
| + this.#store = store; | ||
| + this.#note = store.load(); | ||
| + } | ||
| + | ||
| + get note(): Note { | ||
| + return this.#note; | ||
| + } | ||
| + | ||
| + snapshot(): ServerMessage { | ||
| + return { type: "snapshot", text: this.#note.text, version: this.#note.version }; | ||
| + } | ||
| + | ||
| + receive(raw: string): Outcome { | ||
| + const msg = parseClientMessage(raw); | ||
| + if (!msg) return { reply: { type: "error", message: "expected an op message" } }; | ||
| + const { text, version } = this.#note; | ||
| + | ||
| + // A base from the future, or one older than the log reaches, can't be transformed. | ||
| + if (msg.base > version || version - msg.base > LOG_SIZE) return { reply: this.snapshot() }; | ||
| + | ||
| + const op = transformPast(msg.op, this.#store.opsSince(msg.base)); | ||
| + if (isNoop(op)) return { reply: { type: "ack", version } }; | ||
| + let next: string; | ||
| + try { | ||
| + next = apply(text, op); | ||
| + } catch (e) { | ||
| + if (e instanceof OpError) return { reply: this.snapshot() }; | ||
| + throw e; | ||
| + } | ||
| + | ||
| + const note = { text: next, version: version + 1 }; | ||
| + this.#store.commit(note, op); | ||
| + this.#note = note; | ||
| + return { reply: { type: "ack", version: note.version }, broadcast: { type: "op", version: note.version, op } }; | ||
| + } | ||
| +} | ||
Addedtest/helpers/memory-store.ts+26−0
| @@ -0,0 +1,26 @@ | ||
| +import type { Op } from "../../src/ops.ts"; | ||
| +import { LOG_SIZE, type Note, type NoteStore } from "../../src/room.ts"; | ||
| + | ||
| +// A NoteStore in memory, with the same trimming rule as the SQLite one. | ||
| +export class MemoryStore implements NoteStore { | ||
| + note: Note; | ||
| + log: { version: number; op: Op }[] = []; | ||
| + | ||
| + constructor(note: Note = { text: "", version: 0 }) { | ||
| + this.note = note; | ||
| + } | ||
| + | ||
| + load(): Note { | ||
| + return { ...this.note }; | ||
| + } | ||
| + | ||
| + opsSince(version: number): Op[] { | ||
| + return this.log.filter((e) => e.version > version).map((e) => e.op); | ||
| + } | ||
| + | ||
| + commit(note: Note, op: Op): void { | ||
| + this.note = { ...note }; | ||
| + this.log.push({ version: note.version, op }); | ||
| + this.log = this.log.filter((e) => e.version > note.version - LOG_SIZE); | ||
| + } | ||
| +} | ||
Addedtest/protocol.test.ts+37−0
| @@ -0,0 +1,37 @@ | ||
| +import assert from "node:assert/strict"; | ||
| +import { test } from "node:test"; | ||
| +import { parseClientMessage } from "../src/protocol.ts"; | ||
| + | ||
| +test("parses insert and delete messages", () => { | ||
| + assert.deepEqual(parseClientMessage('{"type":"op","base":3,"op":{"type":"insert","pos":1,"text":"a"}}'), { | ||
| + type: "op", | ||
| + base: 3, | ||
| + op: { type: "insert", pos: 1, text: "a" }, | ||
| + }); | ||
| + assert.deepEqual(parseClientMessage('{"type":"op","base":0,"op":{"type":"delete","pos":0,"count":2}}'), { | ||
| + type: "op", | ||
| + base: 0, | ||
| + op: { type: "delete", pos: 0, count: 2 }, | ||
| + }); | ||
| +}); | ||
| + | ||
| +test("drops fields it does not know", () => { | ||
| + const msg = parseClientMessage('{"type":"op","base":0,"extra":1,"op":{"type":"insert","pos":0,"text":"a","color":"red"}}'); | ||
| + assert.deepEqual(msg, { type: "op", base: 0, op: { type: "insert", pos: 0, text: "a" } }); | ||
| +}); | ||
| + | ||
| +test("rejects anything malformed", () => { | ||
| + for (const raw of [ | ||
| + "", | ||
| + "null", | ||
| + "[]", | ||
| + '{"type":"hello"}', | ||
| + '{"type":"op","base":-1,"op":{"type":"insert","pos":0,"text":"a"}}', | ||
| + '{"type":"op","base":0,"op":{"type":"insert","pos":0.5,"text":"a"}}', | ||
| + '{"type":"op","base":0,"op":{"type":"insert","pos":0,"text":5}}', | ||
| + '{"type":"op","base":0,"op":{"type":"delete","pos":0}}', | ||
| + '{"type":"op","base":"1","op":{"type":"delete","pos":0,"count":1}}', | ||
| + ]) { | ||
| + assert.equal(parseClientMessage(raw), null, raw); | ||
| + } | ||
| +}); | ||
Addedtest/room.test.ts+77−0
| @@ -0,0 +1,77 @@ | ||
| +import assert from "node:assert/strict"; | ||
| +import { describe, test } from "node:test"; | ||
| +import { del, insert, type Op } from "../src/ops.ts"; | ||
| +import { LOG_SIZE, Room } from "../src/room.ts"; | ||
| +import { MemoryStore } from "./helpers/memory-store.ts"; | ||
| + | ||
| +const send = (base: number, op: Op) => JSON.stringify({ type: "op", base, op }); | ||
| + | ||
| +describe("Room", () => { | ||
| + test("a new note is empty at version 0", () => { | ||
| + const room = new Room(new MemoryStore()); | ||
| + assert.deepEqual(room.snapshot(), { type: "snapshot", text: "", version: 0 }); | ||
| + }); | ||
| + | ||
| + test("applies an op, acks the sender and broadcasts it to the others", () => { | ||
| + const store = new MemoryStore(); | ||
| + const room = new Room(store); | ||
| + const out = room.receive(send(0, insert(0, "milk"))); | ||
| + assert.deepEqual(out.reply, { type: "ack", version: 1 }); | ||
| + assert.deepEqual(out.broadcast, { type: "op", version: 1, op: insert(0, "milk") }); | ||
| + assert.deepEqual(store.note, { text: "milk", version: 1 }); | ||
| + }); | ||
| + | ||
| + test("transforms an op past the ops its sender had not seen", () => { | ||
| + const room = new Room(new MemoryStore({ text: "milk\n", version: 0 })); | ||
| + // Two editors at version 0: one adds a line at the top, the other appends. | ||
| + room.receive(send(0, insert(0, "eggs\n"))); | ||
| + const out = room.receive(send(0, insert(5, "bread\n"))); | ||
| + assert.deepEqual(out.broadcast, { type: "op", version: 2, op: insert(10, "bread\n") }); | ||
| + assert.equal(room.note.text, "eggs\nmilk\nbread\n"); | ||
| + }); | ||
| + | ||
| + test("acks an op that another edit already made pointless without bumping the version", () => { | ||
| + const room = new Room(new MemoryStore({ text: "milk eggs", version: 0 })); | ||
| + room.receive(send(0, del(4, 5))); | ||
| + const out = room.receive(send(0, del(5, 4))); | ||
| + assert.deepEqual(out, { reply: { type: "ack", version: 1 } }); | ||
| + assert.equal(room.note.text, "milk"); | ||
| + }); | ||
| + | ||
| + test("sends a snapshot to an editor further behind than the log reaches", () => { | ||
| + const room = new Room(new MemoryStore()); | ||
| + for (let v = 0; v < LOG_SIZE + 1; v++) room.receive(send(v, insert(v, "x"))); | ||
| + const out = room.receive(send(0, insert(0, "late"))); | ||
| + assert.equal(out.reply.type, "snapshot"); | ||
| + assert.equal(out.broadcast, undefined); | ||
| + // One version less behind is still within the log. | ||
| + assert.equal(room.receive(send(1, insert(0, "y"))).reply.type, "ack"); | ||
| + }); | ||
| + | ||
| + test("sends a snapshot for a base version from the future", () => { | ||
| + const room = new Room(new MemoryStore()); | ||
| + assert.deepEqual(room.receive(send(3, insert(0, "x"))).reply, { type: "snapshot", text: "", version: 0 }); | ||
| + }); | ||
| + | ||
| + test("sends a snapshot when an op does not fit the note", () => { | ||
| + const store = new MemoryStore({ text: "abc", version: 4 }); | ||
| + const room = new Room(store); | ||
| + assert.equal(room.receive(send(4, del(2, 5))).reply.type, "snapshot"); | ||
| + assert.deepEqual(store.note, { text: "abc", version: 4 }); | ||
| + }); | ||
| + | ||
| + test("answers a malformed frame with an error", () => { | ||
| + const room = new Room(new MemoryStore()); | ||
| + assert.equal(room.receive("{").reply.type, "error"); | ||
| + assert.equal(room.receive(JSON.stringify({ type: "op", base: 0, op: { type: "upsert", pos: 0 } })).reply.type, "error"); | ||
| + }); | ||
| + | ||
| + test("picks up where it left off from the store", () => { | ||
| + const store = new MemoryStore(); | ||
| + new Room(store).receive(send(0, insert(0, "kept"))); | ||
| + // A Durable Object woken from hibernation builds a new Room over the same storage. | ||
| + const woken = new Room(store); | ||
| + assert.deepEqual(woken.note, { text: "kept", version: 1 }); | ||
| + assert.deepEqual(woken.receive(send(1, insert(0, "> "))).broadcast, { type: "op", version: 2, op: insert(0, "> ") }); | ||
| + }); | ||
| +}); | ||
Checks
No checks reported for this commit.
Runs
No runs for this commit.