Skip to main content
Sign In
Realtime

Collaborative Text Editor (OT)

Build a collaborative text editor with operational transformation, where one Actor per document orders every edit and keeps the operation log.

See the full example on GitHub. View on GitHub
Editor Asends an operationEditor Bapplies itdocumenttransforms and ordersSQLiteoperations, snapshotgetSnapshot on connect1submitOperation2transform, append3operation (ack)4operation to all4One per documentRivet ActorCallReply

Operational transformation (OT) keeps one authoritative order of edits. Every edit goes through one place that decides its position in the history and rewrites it so it still makes sense after the edits that got there first. On Rivet, that place is a Rivet Actor per document.

What we’ll build:

  • A small OT library for plain text with insert and delete, written from scratch with no dependencies.
  • A document Actor per document that is the single sequencer for its edits.
  • An operation log and the latest text in the Actor’s SQLite database, so late joiners load one row instead of replaying history.
  • A browser client that applies its own edits right away, keeps one operation in flight, and rebases pending edits on top of edits from others.

This guide uses OT. The other approach is a CRDT, where every client can merge edits in any order and the server only relays and stores them. A CRDT needs no central ordering and handles long offline sessions well, at the cost of larger document metadata. See Collaborative Text Editor (CRDT).

Install

npm add rivetkit @rivetkit/react

The OT code is part of the example, so there is no OT package to install.

Represent edits as operations

An operation describes one edit as a walk over the whole document:

export type Component =
	| { retain: number }
	| { insert: string }
	| { delete: number };

export type Operation = Component[];

/** Applies an operation to the text it was made against. */
export function apply(text: string, op: Operation): string {
	if (baseLength(op) !== text.length) {
		throw new Error("operation does not match the document length");
	}
	let result = "";
	let index = 0;
	for (const component of op) {
		if ("retain" in component) {
			result += text.slice(index, index + component.retain);
			index += component.retain;
		} else if ("insert" in component) {
			result += component.insert;
		} else {
			index += component.delete;
		}
	}
	return result;
}
  • retain keeps characters, insert adds text at the current position, and delete removes characters. Turning "hello" into "hello world" is [{ retain: 5 }, { insert: " world" }].
  • The retained and deleted counts always add up to the length of the text the operation was made against. apply checks this and throws on a mismatch, so an operation can never be applied to the wrong version of the text.
  • Operations are plain JSON, so they cross the action boundary as is and can be stored in SQLite as a string.

Transform concurrent operations

transform is the heart of OT. It takes two operations made against the same text and rewrites each one so it can be applied after the other:

/**
 * Transforms two operations made against the same text.
 *
 * Returns `[a2, b2]` where applying `a` then `b2` gives the same text as
 * applying `b` then `a2`. When both insert at the same position, the text of
 * `a` goes first.
 */
export function transform(a: Operation, b: Operation): [Operation, Operation] {
	if (baseLength(a) !== baseLength(b)) {
		throw new Error("operations were not made against the same text");
	}
	const a2: Operation = [];
	const b2: Operation = [];
	let i = 0;
	let j = 0;
	let ca: Component | undefined = a[i++];
	let cb: Component | undefined = b[j++];

	while (ca || cb) {
		// Inserted text is new on the other side, so the other side retains it.
		if (ca && "insert" in ca) {
			push(a2, ca);
			push(b2, { retain: ca.insert.length });
			ca = a[i++];
			continue;
		}
		if (cb && "insert" in cb) {
			push(a2, { retain: cb.insert.length });
			push(b2, cb);
			cb = b[j++];
			continue;
		}
		if (!ca || !cb) {
			throw new Error("operations were not made against the same text");
		}

		// Both components now cover existing text. Consume the shorter one.
		const n = Math.min(size(ca), size(cb));
		if ("retain" in ca && "retain" in cb) {
			push(a2, { retain: n });
			push(b2, { retain: n });
		} else if ("delete" in ca && "retain" in cb) {
			push(a2, { delete: n });
		} else if ("retain" in ca && "delete" in cb) {
			push(b2, { delete: n });
		}
		// When both delete the same text, neither side deletes it again.
		ca = shrink(ca, n) ?? a[i++];
		cb = shrink(cb, n) ?? b[j++];
	}
	return [a2, b2];
}
  • It walks both operations at once. Text that one side inserted is new to the other side, so the other side retains it. That is how two inserts in different places both survive.
  • When both sides insert at the same position, the text of a goes first. The rule only has to be deterministic, and the Actor and the clients always pass the uncommitted operation as a, so every copy picks the same order.
  • When one side deletes text the other side retained, only the delete is kept. When both sides deleted the same text, neither deletes it again. An insert inside a range that someone else deleted is kept, because inserts are handled before deletes.
  • The guarantee is that applying a then b2 gives the same text as applying b then a2. The example’s tests check this on hundreds of random edit pairs.

Create the document Actor

Each document is its own document Actor, keyed by the document id. Its SQLite database holds two tables:

export const document = actor({
	// Each document keeps its operation log and latest text in its own SQLite database: https://rivet.dev/docs/actors/sqlite
	db: db({
		onMigrate: async (db) => {
			await db.execute(`
				CREATE TABLE IF NOT EXISTS operations (
					revision INTEGER PRIMARY KEY,
					client_id TEXT NOT NULL,
					operation TEXT NOT NULL,
					created_at INTEGER NOT NULL
				)
			`);
			await db.execute(`
				CREATE TABLE IF NOT EXISTS snapshot (
					id INTEGER PRIMARY KEY CHECK (id = 1),
					revision INTEGER NOT NULL,
					text TEXT NOT NULL
				)
			`);
			await db.execute(
				"INSERT OR IGNORE INTO snapshot (id, revision, text) VALUES (1, 0, '')",
			);
		},
	}),
	events: {
		operation: event<OperationEvent>(),
	},

	actions: {
		getSnapshot: async (c): Promise<Snapshot> => {
			const [row] = await c.db.execute<Snapshot>(
				"SELECT text, revision FROM snapshot WHERE id = 1",
			);
			return { text: row.text, revision: row.revision };
		},

		submitOperation: async (
			c,
			clientId: string,
			baseRevision: number,
			operation: Operation,
		) => {
			try {
				assertOperation(operation);
			} catch (error) {
				throw new UserError((error as Error).message, {
					code: "invalid_operation",
				});
			}

			// The transaction runs one submission at a time, so no other
			// operation can commit between reading the log and appending to it.
			const committed = await c.db.transaction(
				async (tx) => {
					const [current] = await tx.execute<Snapshot>(
						"SELECT text, revision FROM snapshot WHERE id = 1",
					);
					if (
						!Number.isInteger(baseRevision) ||
						baseRevision < 0 ||
						baseRevision > current.revision
					) {
						throw new UserError("unknown base revision", {
							code: "invalid_revision",
						});
					}

					// Rewrite the operation past everything committed since the
					// revision the client based it on.
					const concurrent = await tx.execute<{ operation: string }>(
						"SELECT operation FROM operations WHERE revision > ? ORDER BY revision ASC",
						baseRevision,
					);
					let transformed = operation;
					for (const row of concurrent) {
						const committedOp = JSON.parse(
							row.operation,
						) as Operation;
						[transformed] = transform(transformed, committedOp);
					}

					let text: string;
					try {
						text = apply(current.text, transformed);
					} catch (error) {
						throw new UserError((error as Error).message, {
							code: "invalid_operation",
						});
					}
					const revision = current.revision + 1;
					await tx.execute(
						"INSERT INTO operations (revision, client_id, operation, created_at) VALUES (?, ?, ?, ?)",
						revision,
						clientId,
						JSON.stringify(transformed),
						Date.now(),
					);
					await tx.execute(
						"UPDATE snapshot SET revision = ?, text = ? WHERE id = 1",
						revision,
						text,
					);
					return { revision, clientId, operation: transformed };
				},
				{ name: "submit-operation" },
			);

			// Every editor gets the transformed operation, including the sender,
			// which treats it as the acknowledgment.
			c.broadcast("operation", committed);
			return committed.revision;
		},
	},
});
  • operations is the operation log. Each row is one committed operation, already transformed, at its revision. The Actor reads it to transform operations that were based on an older revision.
  • snapshot is a single row with the current text and revision. It is updated with every operation, so getSnapshot is one read no matter how long the history is, and nothing is replayed when the Actor restarts.
  • onMigrate creates the tables and the empty snapshot the first time the Actor starts. See SQLite.
  • events declares the operation event that carries every committed operation to the editors.

Transform and append each operation

Clients call submitOperation with their id, the revision their operation was based on, and the operation:

submitOperation: async (
	c,
	clientId: string,
	baseRevision: number,
	operation: Operation,
) => {
	try {
		assertOperation(operation);
	} catch (error) {
		throw new UserError((error as Error).message, {
			code: "invalid_operation",
		});
	}

	// The transaction runs one submission at a time, so no other
	// operation can commit between reading the log and appending to it.
	const committed = await c.db.transaction(
		async (tx) => {
			const [current] = await tx.execute<Snapshot>(
				"SELECT text, revision FROM snapshot WHERE id = 1",
			);
			if (
				!Number.isInteger(baseRevision) ||
				baseRevision < 0 ||
				baseRevision > current.revision
			) {
				throw new UserError("unknown base revision", {
					code: "invalid_revision",
				});
			}

			// Rewrite the operation past everything committed since the
			// revision the client based it on.
			const concurrent = await tx.execute<{ operation: string }>(
				"SELECT operation FROM operations WHERE revision > ? ORDER BY revision ASC",
				baseRevision,
			);
			let transformed = operation;
			for (const row of concurrent) {
				const committedOp = JSON.parse(
					row.operation,
				) as Operation;
				[transformed] = transform(transformed, committedOp);
			}

			let text: string;
			try {
				text = apply(current.text, transformed);
			} catch (error) {
				throw new UserError((error as Error).message, {
					code: "invalid_operation",
				});
			}
			const revision = current.revision + 1;
			await tx.execute(
				"INSERT INTO operations (revision, client_id, operation, created_at) VALUES (?, ?, ?, ?)",
				revision,
				clientId,
				JSON.stringify(transformed),
				Date.now(),
			);
			await tx.execute(
				"UPDATE snapshot SET revision = ?, text = ? WHERE id = 1",
				revision,
				text,
			);
			return { revision, clientId, operation: transformed };
		},
		{ name: "submit-operation" },
	);

	// Every editor gets the transformed operation, including the sender,
	// which treats it as the acknowledgment.
	c.broadcast("operation", committed);
	return committed.revision;
},
  • assertOperation checks the shape of the client’s operation, and the base revision must be one the Actor has already reached.
  • The Actor loads every operation committed after baseRevision and transforms the incoming operation past each one in order. The result applies cleanly to the current text.
  • It appends the transformed operation at the next revision and updates the snapshot.
  • All of this runs inside c.db.transaction. Actions run concurrently, but transactions on one Actor run one at a time, so no other operation can commit between reading the log and appending to it. That makes the Actor a single sequencer without any lock of its own.

clientId comes from the client and is only used to recognize acknowledgments, so a client could claim another client’s id. The example also has no authentication, no limit on operation size, and no rate limit. In production, use a connection or user id the server owns, cap the operation size, and check who may edit each document. See Authentication.

The log grows by one row per edit. It is only read for operations newer than the oldest base revision a client still has, so you can delete rows older than a cutoff, such as the last 1,000 revisions, and reject operations based on anything older. Such a client reloads the snapshot.

Broadcast the transformed operation

Once the transaction commits, the Actor tells every editor:

submitOperation: async (
	c,
	clientId: string,
	baseRevision: number,
	operation: Operation,
) => {
	try {
		assertOperation(operation);
	} catch (error) {
		throw new UserError((error as Error).message, {
			code: "invalid_operation",
		});
	}

	// The transaction runs one submission at a time, so no other
	// operation can commit between reading the log and appending to it.
	const committed = await c.db.transaction(
		async (tx) => {
			const [current] = await tx.execute<Snapshot>(
				"SELECT text, revision FROM snapshot WHERE id = 1",
			);
			if (
				!Number.isInteger(baseRevision) ||
				baseRevision < 0 ||
				baseRevision > current.revision
			) {
				throw new UserError("unknown base revision", {
					code: "invalid_revision",
				});
			}

			// Rewrite the operation past everything committed since the
			// revision the client based it on.
			const concurrent = await tx.execute<{ operation: string }>(
				"SELECT operation FROM operations WHERE revision > ? ORDER BY revision ASC",
				baseRevision,
			);
			let transformed = operation;
			for (const row of concurrent) {
				const committedOp = JSON.parse(
					row.operation,
				) as Operation;
				[transformed] = transform(transformed, committedOp);
			}

			let text: string;
			try {
				text = apply(current.text, transformed);
			} catch (error) {
				throw new UserError((error as Error).message, {
					code: "invalid_operation",
				});
			}
			const revision = current.revision + 1;
			await tx.execute(
				"INSERT INTO operations (revision, client_id, operation, created_at) VALUES (?, ?, ?, ?)",
				revision,
				clientId,
				JSON.stringify(transformed),
				Date.now(),
			);
			await tx.execute(
				"UPDATE snapshot SET revision = ?, text = ? WHERE id = 1",
				revision,
				text,
			);
			return { revision, clientId, operation: transformed };
		},
		{ name: "submit-operation" },
	);

	// Every editor gets the transformed operation, including the sender,
	// which treats it as the acknowledgment.
	c.broadcast("operation", committed);
	return committed.revision;
},
  • The operation event carries the new revision, the sender’s id, and the transformed operation. Every editor applies events in revision order, so they all apply the same operations in the same order.
  • The event goes to the sender too. The sender does not apply it again, because it already applied the edit locally. It uses the event as the acknowledgment that its operation was accepted.
  • The broadcast happens after the commit, so an operation that anyone has seen is already saved.

Send edits from the client

DocumentSync keeps one editor’s copy of the document in sync. It turns every change into an operation and sends at most one at a time:

import type { OperationEvent, Snapshot } from "../src/index.ts";
import { apply, diff, isNoop, type Operation, transform } from "../src/ot.ts";

type Send = (baseRevision: number, operation: Operation) => Promise<unknown>;

/**
 * Keeps one editor's copy of a document in sync with the `document` Actor.
 *
 * Local edits apply right away and wait in `pending`. Only the first pending
 * operation is sent, and the next one goes out when the Actor broadcasts it
 * back. Operations from other editors are transformed past every pending
 * operation before they are applied, so local edits never get lost.
 */
export class DocumentSync {
	text = "";
	revision = 0;
	pending: Operation[] = [];
	private loaded = false;
	private inFlight = false;
	private buffered = new Map<number, OperationEvent>();

	constructor(
		private readonly clientId: string,
		private readonly send: Send,
		private readonly onChange: (remote?: Operation) => void,
		private readonly onRejected: () => void,
	) {}

	/** Starts from a snapshot, then applies any events that arrived first. */
	load(snapshot: Snapshot) {
		this.text = snapshot.text;
		this.revision = snapshot.revision;
		this.pending = [];
		this.inFlight = false;
		this.loaded = true;
		this.drain();
		this.onChange();
	}

	/** Queues an event and applies events strictly in revision order. */
	receive(event: OperationEvent) {
		this.buffered.set(event.revision, event);
		if (this.loaded) this.drain();
	}

	private drain() {
		for (const revision of this.buffered.keys()) {
			if (revision <= this.revision) this.buffered.delete(revision);
		}
		let next = this.buffered.get(this.revision + 1);
		while (next) {
			this.buffered.delete(next.revision);
			this.applyEvent(next);
			next = this.buffered.get(this.revision + 1);
		}
	}

	/** Records a local edit and sends it if nothing is in flight. */
	edit(newText: string) {
		const op = diff(this.text, newText);
		if (isNoop(op)) return;
		this.text = newText;
		this.pending.push(op);
		this.flush();
		this.onChange();
	}

	private flush() {
		if (this.inFlight || !this.loaded || this.pending.length === 0) return;
		this.inFlight = true;
		this.send(this.revision, this.pending[0]).catch((error) => {
			// The local text no longer matches the Actor, so start over from a
			// fresh snapshot.
			console.error("operation rejected", error);
			this.loaded = false;
			this.onRejected();
		});
	}

	private applyEvent(event: OperationEvent) {
		this.revision = event.revision;

		// Our own operation came back, so the Actor accepted it.
		if (this.inFlight && event.clientId === this.clientId) {
			this.pending.shift();
			this.inFlight = false;
			this.flush();
			this.onChange();
			return;
		}

		// Another editor's operation: transform it past every local operation
		// the Actor has not committed yet, and rebase those on top of it.
		let remote = event.operation;
		this.pending = this.pending.map((local) => {
			const [rebased, remoteAfter] = transform(local, remote);
			remote = remoteAfter;
			return rebased;
		});
		this.text = apply(this.text, remote);
		this.onChange(remote);
	}
}
  • edit builds an operation from the old and new text with diff, applies it locally right away, and queues it in pending. Typing never waits for the network.
  • flush sends only pending[0], with the last revision the client has seen as its base. The next one goes out when the acknowledgment arrives, already rebased on everything the client received in the meantime, so the client never has two operations in flight to reconcile.
  • When another editor’s operation arrives, applyEvent transforms it past every pending operation and rebases those operations on top of it. The client then applies the transformed remote operation, so local edits that are still in flight are never lost.
  • If the Actor rejects an operation, the client throws away its pending edits and reloads the snapshot.

Connect the editor

The React editor creates a DocumentSync for every connection and feeds it events:

const documentActor = useActor({ name: "document", key: [documentId] });

useEffect(() => {
	const connection = documentActor.connection;
	if (!connection) return;

	const clientId = crypto.randomUUID();
	const resync = () => {
		connection.getSnapshot().then((snapshot) => sync.load(snapshot));
	};
	const sync = new DocumentSync(
		clientId,
		(baseRevision, operation) =>
			connection.submitOperation(clientId, baseRevision, operation),
		(remote) => {
			if (remote) {
				selectionRef.current = rebaseSelection(
					textareaRef.current,
					selectionRef.current,
					remote,
				);
			}
			setView({
				text: sync.text,
				revision: sync.revision,
				pending: sync.pending.length,
			});
		},
		resync,
	);
	syncRef.current = sync;
	resync();

	return () => {
		syncRef.current = null;
	};
}, [documentActor.connection]);

documentActor.useEvent("operation", (event) => {
	syncRef.current?.receive(event as OperationEvent);
});

/** Starts from a snapshot, then applies any events that arrived first. */
load(snapshot: Snapshot) {
	this.text = snapshot.text;
	this.revision = snapshot.revision;
	this.pending = [];
	this.inFlight = false;
	this.loaded = true;
	this.drain();
	this.onChange();
}

/** Queues an event and applies events strictly in revision order. */
receive(event: OperationEvent) {
	this.buffered.set(event.revision, event);
	if (this.loaded) this.drain();
}

private drain() {
	for (const revision of this.buffered.keys()) {
		if (revision <= this.revision) this.buffered.delete(revision);
	}
	let next = this.buffered.get(this.revision + 1);
	while (next) {
		this.buffered.delete(next.revision);
		this.applyEvent(next);
		next = this.buffered.get(this.revision + 1);
	}
}
  • useActor({ name: "document", key: [documentId] }) connects to the document’s Actor and creates it if needed.
  • Each connection gets a new client id and calls getSnapshot. Events that arrive before the snapshot are buffered, and load drops the ones the snapshot already includes.
  • receive applies events strictly in revision order and holds any that arrive early. The textarea calls edit on every change, and the editor moves the caret past remote inserts and deletes with transformIndex, so it does not jump while someone else types.
  • On reconnect, the client starts over from a fresh snapshot. Edits that were still pending when the connection dropped are discarded.

Try it

git clone https://github.com/rivet-dev/rivet.git
cd rivet/examples/collaborative-document-ot
pnpm install
pnpm dev

Open http://localhost:5173 in two tabs. Type in both at the same time, even at the same spot, and both tabs end up with the same text. The header shows the current revision and how many of your operations are still waiting for the Actor. Reload a tab and the text comes back from the snapshot. Change the Document field to open a different document.

Next steps

Edit this page