diff --git a/packages/liveblocks-server/package.json b/packages/liveblocks-server/package.json index 95535b9492..7673ea36a0 100644 --- a/packages/liveblocks-server/package.json +++ b/packages/liveblocks-server/package.json @@ -69,7 +69,7 @@ }, "sideEffects": false, "dependencies": { - "@liveblocks/core": "3.23.0-file1", + "@liveblocks/core": "3.23.1-exp3", "async-mutex": "^0.4.0", "decoders": "^2.9.0", "itertools": "^2.7.1", diff --git a/packages/liveblocks-server/src/Room.ts b/packages/liveblocks-server/src/Room.ts index 9c17fc39eb..27c14cc7e6 100644 --- a/packages/liveblocks-server/src/Room.ts +++ b/packages/liveblocks-server/src/Room.ts @@ -1636,12 +1636,15 @@ export class Room { case "rectified": // The op was already applied earlier; re-acknowledge it with - // its stored, authoritative position. - return [r.ackOp, r.fix]; + // its stored, authoritative fields and optional correction. + return r.fix !== undefined ? [r.ackOp, r.fix] : [r.ackOp]; case "accepted": return r.fix !== undefined ? [r.fix] : []; + case "rejected": + return []; + // istanbul ignore next default: return assertNever(r, "Unhandled case"); @@ -1667,6 +1670,14 @@ export class Room { }); } + for (const rejected of result.filter((r) => r.action === "rejected")) { + replyImmediately({ + type: ServerMsgCode.REJECT_STORAGE_OP, + opIds: rejected.opIds, + reason: rejected.reason, + }); + } + if (opsToForward.length > 0) { // NOTE! These are being called after *every* handleOne() call // currently. Should we not just call these once at the end of diff --git a/packages/liveblocks-server/src/Storage.ts b/packages/liveblocks-server/src/Storage.ts index 9f57bdf830..f741e85c43 100644 --- a/packages/liveblocks-server/src/Storage.ts +++ b/packages/liveblocks-server/src/Storage.ts @@ -17,11 +17,14 @@ import type { SerializedChild, SerializedCrdt } from "@liveblocks/core"; import { + applyLiveTextOperations, asPos, assertNever, CrdtType, makePosition, + normalizeLiveTextOperations, OpCode, + transformTextOperations, } from "@liveblocks/core"; import type { IStorageDriver } from "~/interfaces"; @@ -34,21 +37,27 @@ import type { HasOpId, SetParentKeyOp, UpdateObjectOp, + UpdateTextOp, } from "~/protocol"; import type { Pos } from "~/types"; +const LIVE_TEXT_HISTORY_LIMIT = 1000; +const LIVE_TEXT_HISTORY_TOO_OLD_REASON = + "LiveText operation is older than retained history"; + /** - * The three possible outcomes of applying a client op. They differ along + * The possible outcomes of applying a client op. They differ along * when the op (first) changed storage state, who hears about it, and what * gets sent back to the originating client: * - * | | state change | fan out to others | reply to sender | - * |-------------|--------------|-------------------|------------------| - * | OpAccepted | now | yes | ack echo (+ fix) | - * | OpRectified | in the past | no | ack echo + fix | - * | OpIgnored | never | no | bare (H)Ack | + * | | state change | fan out to others | reply to sender | + * |------------|--------------|-------------------|------------------| + * | OpAccepted | now | yes | ack echo (+ fix) | + * | OpRectified| in the past | no | ack echo (+ fix) | + * | OpIgnored | never | no | bare (H)Ack | + * | OpRejected | never | no | rejection | */ -type ApplyOpResult = OpAccepted | OpIgnored | OpRectified; +type ApplyOpResult = OpAccepted | OpIgnored | OpRectified | OpRejected; export type OpAccepted = { action: "accepted"; @@ -64,20 +73,22 @@ export type OpIgnored = { export type OpRectified = { action: "rectified"; /** - * Echo of the client's op, with the stored, authoritative parentKey. Sent - * back to the originating client as the acknowledgement, instead of the - * bare (H)Ack. Used for re-sent CREATE ops whose node the server already - * stored: the echo carries the authoritative position, so the client can - * correct any optimistic local position it may have predicted while the op - * was pending. Never fanned out to others: they already received the op - * when it was originally accepted. + * Echo of the client's op with authoritative server fields. Sent back to the + * originating client as the acknowledgement, instead of the bare (H)Ack. + * Never fanned out to others: they already received the op when it was + * originally accepted. */ - ackOp: CreateOp & HasOpId; + ackOp: ClientWireOp; /** - * A corrective op to send back to the originating client, stating that - * same authoritative position (see ackOp). + * Optional corrective op to send back to the originating client. */ - fix: FixOp; + fix?: FixOp; +}; + +export type OpRejected = { + action: "rejected"; + opIds: string[]; + reason: string; }; function accept(op: ClientWireOp, fix?: FixOp): OpAccepted { @@ -88,12 +99,12 @@ function ignore(ignoredOp: ClientWireOp): OpIgnored { return { action: "ignored", ignoredOpId: ignoredOp.opId }; } -function rectify(op: CreateOp & HasOpId, parentKey: string): OpRectified { - return { - action: "rectified", - ackOp: { ...op, parentKey }, - fix: { type: OpCode.SET_PARENT_KEY, id: op.id, parentKey }, - }; +function reject(op: ClientWireOp, reason: string): OpRejected { + return { action: "rejected", opIds: [op.opId], reason }; +} + +function rectify(ackOp: ClientWireOp, fix?: FixOp): OpRectified { + return { action: "rectified", ackOp, fix }; } function nodeFromCreateChildOp(op: CreateOp): SerializedChild { @@ -128,6 +139,14 @@ function nodeFromCreateChildOp(op: CreateOp): SerializedChild { data: op.data, }; + case OpCode.CREATE_TEXT: + return { + type: CrdtType.TEXT, + parentId: op.parentId, + parentKey: op.parentKey, + data: op.data, + version: op.version, + }; case OpCode.CREATE_FILE: return { type: CrdtType.FILE, @@ -184,12 +203,16 @@ export class Storage { case OpCode.CREATE_MAP: case OpCode.CREATE_REGISTER: case OpCode.CREATE_OBJECT: + case OpCode.CREATE_TEXT: case OpCode.CREATE_FILE: return this.applyCreateOp(op); case OpCode.UPDATE_OBJECT: return this.applyUpdateObjectOp(op); + case OpCode.UPDATE_TEXT: + return this.applyUpdateTextOp(op); + case OpCode.SET_PARENT_KEY: return this.applySetParentKeyOp(op); @@ -227,7 +250,14 @@ export class Storage { stored?.parentId !== undefined && this.driver.get_node(stored.parentId)?.type === CrdtType.LIST ) { - return rectify(op, stored.parentKey); + return rectify( + { ...op, parentKey: stored.parentKey }, + { + type: OpCode.SET_PARENT_KEY, + id: op.id, + parentKey: stored.parentKey, + } + ); } } return ignore(op); @@ -264,6 +294,7 @@ export class Storage { return this.createChildAsListItem(op, node); case CrdtType.REGISTER: + case CrdtType.TEXT: case CrdtType.FILE: // It's illegal for leaf nodes to have children return ignore(op); @@ -368,6 +399,80 @@ export class Storage { return accept(op); } + private applyUpdateTextOp(op: UpdateTextOp & HasOpId): ApplyOpResult { + const node = this.driver.get_node(op.id); + if (node?.type !== CrdtType.TEXT) { + return ignore(op); + } + + const duplicate = this.driver.get_live_text_history_by_op_id( + op.id, + op.opId + ); + if (duplicate !== undefined) { + return rectify({ + ...op, + baseVersion: duplicate.baseVersion, + version: duplicate.version, + ops: [...duplicate.ops], + }); + } + + if (op.ops.length === 0) { + // Empty updates are pure acknowledgement vehicles (e.g. an undo whose + // content was queued behind another in-flight op on the client). Ack + // without applying or bumping the version. + return ignore(op); + } + + if (op.baseVersion > node.version) { + return reject(op, "LiveText operation base version is ahead of storage"); + } + + const history = + op.baseVersion < node.version + ? this.driver.get_live_text_history_since(op.id, op.baseVersion) + : []; + if ( + op.baseVersion < node.version && + history.length !== node.version - op.baseVersion + ) { + return reject(op, LIVE_TEXT_HISTORY_TOO_OLD_REASON); + } + + const acceptedOps = history.flatMap((entry) => entry.ops); + const transformedOps = + acceptedOps.length > 0 + ? transformTextOperations(op.ops, acceptedOps, "after") + : op.ops; + const ops = normalizeLiveTextOperations(node.data, transformedOps); + const version = node.version + 1; + const data = applyLiveTextOperations(node.data, ops); + this.driver.set_child( + op.id, + { + ...node, + data, + version, + }, + true + ); + this.driver.append_live_text_history({ + nodeId: op.id, + baseVersion: node.version, + version, + opId: op.opId, + ops: [...ops], + }); + // Keep the retained history bounded to LIVE_TEXT_HISTORY_LIMIT + this.driver.purge_live_text_history_before( + op.id, + Math.max(0, version - LIVE_TEXT_HISTORY_LIMIT + 1) + ); + + return accept({ ...op, baseVersion: node.version, version, ops: [...ops] }); + } + private applyDeleteCrdtOp(op: DeleteCrdtOp & HasOpId): ApplyOpResult { this.driver.delete_node(op.id); return accept(op); diff --git a/packages/liveblocks-server/src/decoders/Op.ts b/packages/liveblocks-server/src/decoders/Op.ts index 4977ac471e..5527918bce 100644 --- a/packages/liveblocks-server/src/decoders/Op.ts +++ b/packages/liveblocks-server/src/decoders/Op.ts @@ -15,10 +15,13 @@ * along with this program. If not, see . */ +import type { LiveTextData, LiveTextSegment } from "@liveblocks/core"; import { OpCode } from "@liveblocks/core"; import type { Decoder } from "decoders"; import { + array, constant, + either, number, object, oneOf, @@ -27,6 +30,7 @@ import { startsWith, string, taggedUnion, + tuple, } from "decoders"; import type { @@ -36,10 +40,12 @@ import type { CreateMapOp, CreateObjectOp, CreateRegisterOp, + CreateTextOp, DeleteCrdtOp, DeleteObjectKeyOp, SetParentKeyOp, UpdateObjectOp, + UpdateTextOp, } from "~/protocol"; import { jsonObjectYolo, jsonYolo } from "./jsonYolo"; @@ -53,6 +59,12 @@ const fileSize = number.refine( "Must be a valid file size" ); +const liveTextVersion = number.reject((value) => + Number.isSafeInteger(value) && value >= 0 + ? null + : "Must be a non-negative safe integer" +); + const updateObjectOp: Decoder = object({ type: constant(OpCode.UPDATE_OBJECT), opId: string, @@ -102,6 +114,45 @@ const createRegisterOp: Decoder = object({ deletedId: optional(string), }); +const liveTextSegment: Decoder = either( + tuple(string), + tuple(string, jsonObjectYolo) +); + +const liveTextData: Decoder = array(liveTextSegment); + +const textOperation = taggedUnion("type", { + insert: object({ + type: constant("insert"), + index: number, + text: string, + attributes: optional(jsonObjectYolo), + }), + delete: object({ + type: constant("delete"), + index: number, + length: number, + }), + format: object({ + type: constant("format"), + index: number, + length: number, + attributes: jsonObjectYolo, + }), +}); + +const createTextOp: Decoder = object({ + type: constant(OpCode.CREATE_TEXT), + opId: string, + id: string, + parentId: string, + parentKey: string, + data: liveTextData, + version: liveTextVersion, + intent: optional(intent), + deletedId: optional(string), +}); + const createFileOp: Decoder = object({ type: constant(OpCode.CREATE_FILE), opId: string, @@ -118,6 +169,15 @@ const createFileOp: Decoder = object({ deletedId: optional(string), }); +const updateTextOp: Decoder = object({ + type: constant(OpCode.UPDATE_TEXT), + opId: string, + id: string, + baseVersion: liveTextVersion, + version: optional(liveTextVersion), + ops: array(textOperation), +}); + const deleteCrdtOp: Decoder = object({ type: constant(OpCode.DELETE_CRDT), opId: string, @@ -144,6 +204,8 @@ export const op: Decoder = taggedUnion("type", { [OpCode.CREATE_LIST]: createListOp, [OpCode.CREATE_MAP]: createMapOp, [OpCode.CREATE_REGISTER]: createRegisterOp, + [OpCode.CREATE_TEXT]: createTextOp, + [OpCode.UPDATE_TEXT]: updateTextOp, [OpCode.CREATE_FILE]: createFileOp, [OpCode.DELETE_CRDT]: deleteCrdtOp, [OpCode.SET_PARENT_KEY]: setParentKeyOp, diff --git a/packages/liveblocks-server/src/formats/LossyJson.ts b/packages/liveblocks-server/src/formats/LossyJson.ts index 80cd6824af..f8164da3c5 100644 --- a/packages/liveblocks-server/src/formats/LossyJson.ts +++ b/packages/liveblocks-server/src/formats/LossyJson.ts @@ -49,6 +49,8 @@ function buildNode(snapshot: IReadableSnapshot, id: string): Json { return buildList(snapshot, id); } else if (node.type === CrdtType.MAP) { return buildMap(snapshot, id); + } else if (node.type === CrdtType.TEXT) { + return node.data as Json; } else if (node.type === CrdtType.FILE) { return node.data; } else { @@ -122,6 +124,8 @@ function* emit(snapshot: IReadableSnapshot, id: string): StringGen { yield* emitMap(snapshot, id); } else if (node.type === CrdtType.REGISTER) { yield JSON.stringify(node.data); + } else if (node.type === CrdtType.TEXT) { + yield JSON.stringify(node.data); } else if (node.type === CrdtType.FILE) { yield JSON.stringify(node.data); } diff --git a/packages/liveblocks-server/src/formats/PlainLson.ts b/packages/liveblocks-server/src/formats/PlainLson.ts index d06c32046a..a69a610fc9 100644 --- a/packages/liveblocks-server/src/formats/PlainLson.ts +++ b/packages/liveblocks-server/src/formats/PlainLson.ts @@ -24,6 +24,7 @@ import type { PlainLsonList, PlainLsonMap, PlainLsonObject, + PlainLsonText, RootStorageNode, SerializedList, StorageNode, @@ -45,7 +46,12 @@ function generateId(state: { clock: number }) { function isSpecialPlainLsonValue( value: PlainLson -): value is PlainLsonObject | PlainLsonMap | PlainLsonList | PlainLsonFile { +): value is + | PlainLsonObject + | PlainLsonMap + | PlainLsonList + | PlainLsonText + | PlainLsonFile { return isJsonObject(value) && value.liveblocksType !== undefined; } @@ -73,6 +79,19 @@ function* iterJson( yield* iterMap(key, data.data, parent, state); return; + case "LiveText": + yield [ + generateId(state), + { + type: CrdtType.TEXT, + data: data.data, + version: data.version ?? 0, + parentId: parent[0], + parentKey: key, + }, + ]; + return; + case "LiveFile": yield [ generateId(state), @@ -244,6 +263,11 @@ function buildNode(snapshot: IReadableSnapshot, id: string): PlainLson { return buildList(snapshot, id); } else if (node.type === CrdtType.MAP) { return buildMap(snapshot, id); + } else if (node.type === CrdtType.TEXT) { + return { + liveblocksType: "LiveText", + data: node.data, + }; } else if (node.type === CrdtType.FILE) { return { liveblocksType: "LiveFile", data: node.data }; } else { @@ -323,6 +347,11 @@ function* emit(snapshot: IReadableSnapshot, id: string): StringGen { // TEMPORARY: see buildNode — remove `?? null` once all rooms are // migrated to SQLite. yield JSON.stringify(node.data ?? null); + } else if (node.type === CrdtType.TEXT) { + yield JSON.stringify({ + liveblocksType: "LiveText", + data: node.data, + }); } else if (node.type === CrdtType.FILE) { yield '{"liveblocksType":"LiveFile","data":'; yield JSON.stringify(node.data); diff --git a/packages/liveblocks-server/src/interfaces/IStorageDriver.ts b/packages/liveblocks-server/src/interfaces/IStorageDriver.ts index 1346eb678a..5d6b2c68a1 100644 --- a/packages/liveblocks-server/src/interfaces/IStorageDriver.ts +++ b/packages/liveblocks-server/src/interfaces/IStorageDriver.ts @@ -24,6 +24,7 @@ import type { SerializedCrdt, SerializedRootObject, StorageNode, + TextOperation, } from "@liveblocks/core"; import type { YDocId } from "~/decoders/y-types"; @@ -64,6 +65,14 @@ export type ListFeedMessagesResult = { nextCursor?: string; // Cursor for next page, undefined if no more pages }; +export type LiveTextHistoryEntry = { + nodeId: string; + version: number; + baseVersion: number; + opId: string; + ops: TextOperation[]; +}; + /** * An isolated, read-only copy of the storage document at a point in time. * @@ -268,6 +277,40 @@ export interface IStorageDriver { raw_iter_nodes(): Iterable<[string, SerializedCrdt]>; // --------------------------------------------------------------------------- + // LiveText history APIs + // --------------------------------------------------------------------------- + + /** + * Return authoritative LiveText operations accepted after the given version, + * sorted by ascending version. + */ + get_live_text_history_since( + nodeId: string, + version: number + ): LiveTextHistoryEntry[]; + + /** + * Return the authoritative history entry for an opId, if it was already + * accepted for this LiveText node. + */ + get_live_text_history_by_op_id( + nodeId: string, + opId: string + ): LiveTextHistoryEntry | undefined; + + /** + * Append the authoritative, already-rebased operations for an accepted + * LiveText update. + */ + append_live_text_history(entry: LiveTextHistoryEntry): void; + + /** + * Drop retained history older than minVersionToKeep for this LiveText node. + */ + purge_live_text_history_before( + nodeId: string, + minVersionToKeep: number + ): void; // LiveFile upload receipt APIs // --------------------------------------------------------------------------- diff --git a/packages/liveblocks-server/src/interfaces/index.ts b/packages/liveblocks-server/src/interfaces/index.ts index aa055a0e05..663f5e5593 100644 --- a/packages/liveblocks-server/src/interfaces/index.ts +++ b/packages/liveblocks-server/src/interfaces/index.ts @@ -30,5 +30,6 @@ export type { ListFeedMessagesResult, ListFeedsOptions, ListFeedsResult, + LiveTextHistoryEntry, } from "./IStorageDriver"; export type { LeasedSession } from "~/types"; diff --git a/packages/liveblocks-server/src/plugins/InMemoryDriver.ts b/packages/liveblocks-server/src/plugins/InMemoryDriver.ts index a4a52dd794..286eba2cc6 100644 --- a/packages/liveblocks-server/src/plugins/InMemoryDriver.ts +++ b/packages/liveblocks-server/src/plugins/InMemoryDriver.ts @@ -47,6 +47,7 @@ import type { ListFeedMessagesResult, ListFeedsOptions, ListFeedsResult, + LiveTextHistoryEntry, } from "~/interfaces"; import { NestedMap } from "~/lib/NestedMap"; import { quote } from "~/lib/text"; @@ -89,7 +90,9 @@ function buildReverseLookup(nodes: NodeMap) { } const isLeafNode = - node.type === CrdtType.REGISTER || node.type === CrdtType.FILE; + node.type === CrdtType.REGISTER || + node.type === CrdtType.FILE || + node.type === CrdtType.TEXT; if (!isLeafNode) { queue.push(...revNodes.valuesAt(nodeId)); } else if (node.type === CrdtType.REGISTER) { @@ -162,6 +165,7 @@ export class InMemoryDriver implements IStorageDriver { private _leasedSessions: Map; private _feeds: Map; private _feedMessages: Map; // Key: `${feedId}:${messageId}` + private _liveTextHistory: Map; private _livefileUploads: Map; // Key: fileId, value: size constructor(options?: { @@ -174,6 +178,7 @@ export class InMemoryDriver implements IStorageDriver { this._leasedSessions = new Map(); this._feeds = new Map(); this._feedMessages = new Map(); + this._liveTextHistory = new Map(); this._livefileUploads = new Map(); this._nextActor = options?.initialActor ?? -1; @@ -198,11 +203,58 @@ export class InMemoryDriver implements IStorageDriver { this.reinitialize(); this._nodes.clear(); + this._liveTextHistory.clear(); for (const [id, node] of plainLsonToNodeStream(doc)) { this._nodes.set(id, this._withUploadedSize(node)); } } + get_live_text_history_since( + nodeId: string, + version: number + ): LiveTextHistoryEntry[] { + return (this._liveTextHistory.get(nodeId) ?? []) + .filter((entry) => entry.version > version) + .sort((left, right) => left.version - right.version) + .map((entry) => ({ ...entry, ops: [...entry.ops] })); + } + + get_live_text_history_by_op_id( + nodeId: string, + opId: string + ): LiveTextHistoryEntry | undefined { + const entry = (this._liveTextHistory.get(nodeId) ?? []).find( + (item) => item.opId === opId + ); + return entry === undefined ? undefined : { ...entry, ops: [...entry.ops] }; + } + + append_live_text_history(entry: LiveTextHistoryEntry): void { + const history = this._liveTextHistory.get(entry.nodeId) ?? []; + history.push({ ...entry, ops: [...entry.ops] }); + history.sort((left, right) => left.version - right.version); + this._liveTextHistory.set(entry.nodeId, history); + } + + purge_live_text_history_before( + nodeId: string, + minVersionToKeep: number + ): void { + const history = this._liveTextHistory.get(nodeId); + if (history === undefined) { + return; + } + + const retained = history.filter( + (entry) => entry.version >= minVersionToKeep + ); + if (retained.length === 0) { + this._liveTextHistory.delete(nodeId); + } else { + this._liveTextHistory.set(nodeId, retained); + } + } + put_livefile_upload(fileId: string, size: number): void { this._livefileUploads.set(fileId, size); } @@ -607,6 +659,7 @@ export class InMemoryDriver implements IStorageDriver { // For the in-memory backend, this._nodes IS the "on-disk" storage, // so we operate on it directly (no separate cache needed). const nodes = this._nodes; + const liveTextHistory = this._liveTextHistory; if (!nodes.has("root")) { nodes.set("root", { type: CrdtType.OBJECT, data: {} }); } @@ -765,6 +818,7 @@ export class InMemoryDriver implements IStorageDriver { const currid = queue.pop()!; queue.push(...revNodes.valuesAt(currid)); nodes.delete(currid); + liveTextHistory.delete(currid); revNodes.deleteAll(currid); } } diff --git a/packages/liveblocks-server/src/protocol/vNEXT.ts b/packages/liveblocks-server/src/protocol/vNEXT.ts index 5cfe12af6c..ca9e2f813d 100644 --- a/packages/liveblocks-server/src/protocol/vNEXT.ts +++ b/packages/liveblocks-server/src/protocol/vNEXT.ts @@ -43,6 +43,7 @@ export type { CreateObjectOp, CreateOp, CreateRegisterOp, + CreateTextOp, DeleteCrdtOp, DeleteObjectKeyOp, HasOpId, @@ -51,4 +52,5 @@ export type { ServerWireOp, SetParentKeyOp, UpdateObjectOp, + UpdateTextOp, } from "@liveblocks/core"; diff --git a/packages/liveblocks-server/test/Storage.liveText.test.ts b/packages/liveblocks-server/test/Storage.liveText.test.ts new file mode 100644 index 0000000000..281b83e916 --- /dev/null +++ b/packages/liveblocks-server/test/Storage.liveText.test.ts @@ -0,0 +1,560 @@ +/** + * Copyright (c) Liveblocks Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import type { LiveTextData, NodeMap, TextOperation } from "@liveblocks/core"; +import { CrdtType, OpCode } from "@liveblocks/core"; +import { describe, expect, test } from "vitest"; + +import { InMemoryDriver } from "~/plugins/InMemoryDriver"; +import { Storage } from "~/Storage"; + +function initInMemory(driver: InMemoryDriver, rawNodes: NodeMap): void { + const internalNodes = (driver as unknown as { _nodes: NodeMap })._nodes; + for (const [id, node] of rawNodes) { + internalNodes.set(id, node); + } +} + +function updateTextOp( + id: string, + baseVersion: number, + ops: TextOperation[], + opId = `op:${id}:${baseVersion}` +) { + return { + opId, + id, + type: OpCode.UPDATE_TEXT, + baseVersion, + ops, + } as const; +} + +describe("Storage LiveText", () => { + test("applies insert, delete, and format operations with version increments", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + + const insertResult = storage.applyOps([ + updateTextOp("0:1", 0, [{ type: "insert", index: 5, text: "!" }]), + ]); + expect(insertResult[0]?.action).toBe("accepted"); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello!"]], + version: 1, + }); + + const formatResult = storage.applyOps([ + updateTextOp("0:1", 1, [ + { + type: "format", + index: 0, + length: 5, + attributes: { bold: true }, + }, + ]), + ]); + expect(formatResult[0]?.action).toBe("accepted"); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello", { bold: true }], ["!"]], + version: 2, + }); + + const deleteResult = storage.applyOps([ + updateTextOp("0:1", 2, [{ type: "delete", index: 5, length: 1 }]), + ]); + expect(deleteResult[0]?.action).toBe("accepted"); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello", { bold: true }]], + version: 3, + }); + }); + + test.each<{ + name: string; + operation: TextOperation; + normalizedOperation: TextOperation; + expectedData: LiveTextData; + }>([ + { + name: "insert", + operation: { type: "insert", index: 2, text: "X" }, + normalizedOperation: { type: "insert", index: 1, text: "X" }, + expectedData: [["aX😀b"]], + }, + { + name: "delete", + operation: { type: "delete", index: 2, length: 1 }, + normalizedOperation: { type: "delete", index: 1, length: 2 }, + expectedData: [["ab"]], + }, + { + name: "format", + operation: { + type: "format", + index: 2, + length: 1, + attributes: { bold: true }, + }, + normalizedOperation: { + type: "format", + index: 1, + length: 2, + attributes: { bold: true }, + }, + expectedData: [["a"], ["😀", { bold: true }], ["b"]], + }, + ])( + "normalizes a surrogate-splitting $name before accepting it", + (testCase) => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["a😀b"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + const result = storage.applyOps([ + updateTextOp("0:1", 0, [testCase.operation]), + ]); + + expect(result[0]).toMatchObject({ + action: "accepted", + op: { ops: [testCase.normalizedOperation] }, + }); + expect(storage.driver.get_node("0:1")).toMatchObject({ + data: testCase.expectedData, + version: 1, + }); + expect( + storage.driver.get_live_text_history_since("0:1", 0)[0]?.ops + ).toEqual([testCase.normalizedOperation]); + } + ); + + test("normalizes surrogate boundaries after rebasing stale operations", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["a😀b"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + expect( + storage.applyOps([ + updateTextOp( + "0:1", + 0, + [{ type: "insert", index: 0, text: "Z" }], + "first" + ), + ])[0]?.action + ).toBe("accepted"); + + const staleResult = storage.applyOps([ + updateTextOp( + "0:1", + 0, + [{ type: "insert", index: 2, text: "X" }], + "stale" + ), + ]); + + expect(staleResult[0]).toMatchObject({ + action: "accepted", + op: { + baseVersion: 1, + version: 2, + ops: [{ type: "insert", index: 2, text: "X" }], + }, + }); + expect(storage.driver.get_node("0:1")).toMatchObject({ + data: [["ZaX😀b"]], + version: 2, + }); + }); + + test("normalizes segments with attribute key order differences", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [ + ["He", { bold: true, italic: true }], + ["llo", { italic: true, bold: true }], + ], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + + const result = storage.applyOps([ + updateTextOp("0:1", 0, [{ type: "insert", index: 5, text: "!" }]), + ]); + expect(result[0]?.action).toBe("accepted"); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello", { bold: true, italic: true }], ["!"]], + version: 1, + }); + }); + + test("rebases stale operations over authoritative history", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + + const first = storage.applyOps([ + updateTextOp( + "0:1", + 0, + [{ type: "insert", index: 0, text: "A" }], + "client:a" + ), + ]); + expect(first[0]).toMatchObject({ + action: "accepted", + op: { + baseVersion: 0, + version: 1, + ops: [{ type: "insert", index: 0, text: "A" }], + }, + }); + + const second = storage.applyOps([ + updateTextOp( + "0:1", + 0, + [{ type: "delete", index: 0, length: 2 }], + "client:b" + ), + ]); + + expect(second[0]).toMatchObject({ + action: "accepted", + op: { + baseVersion: 1, + version: 2, + ops: [{ type: "delete", index: 1, length: 2 }], + }, + }); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Allo"]], + version: 2, + }); + }); + + test("acknowledges duplicate LiveText opIds without applying twice", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + const op = updateTextOp( + "0:1", + 0, + [{ type: "insert", index: 5, text: "!" }], + "client:duplicate" + ); + + expect(storage.applyOps([op])[0]?.action).toBe("accepted"); + const duplicate = storage.applyOps([op]); + + expect(duplicate[0]).toMatchObject({ + action: "rectified", + ackOp: { + opId: "client:duplicate", + baseVersion: 0, + version: 1, + ops: [{ type: "insert", index: 5, text: "!" }], + }, + }); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello!"]], + version: 1, + }); + }); + + test("rejects stale operations when history does not cover the version gap", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["AHello"]], + version: 1, + }, + ], + ]) + ); + + const storage = new Storage(driver); + const result = storage.applyOps([ + updateTextOp( + "0:1", + 0, + [{ type: "delete", index: 0, length: 2 }], + "client:stale" + ), + ]); + + expect(result[0]).toMatchObject({ + action: "rejected", + opIds: ["client:stale"], + reason: "LiveText operation is older than retained history", + }); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["AHello"]], + version: 1, + }); + }); + + test("purges retained LiveText history to the configured window", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [[""]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + for (let version = 0; version < 1002; version++) { + const result = storage.applyOps([ + updateTextOp( + "0:1", + version, + [{ type: "insert", index: version, text: "x" }], + `client:${version}` + ), + ]); + expect(result[0]?.action).toBe("accepted"); + } + + // History is purged continuously as ops are appended, not just when a + // room (re)loads. + const history = storage.driver.get_live_text_history_since("0:1", 0); + expect(history).toHaveLength(1000); + expect(history[0]?.version).toBe(3); + expect(history[history.length - 1]?.version).toBe(1002); + + // Reloading keeps the same bounded window. + const reloadedStorage = new Storage(driver); + const reloadedHistory = reloadedStorage.driver.get_live_text_history_since( + "0:1", + 0 + ); + expect(reloadedHistory).toHaveLength(1000); + expect(reloadedHistory[0]?.version).toBe(3); + expect(reloadedHistory[reloadedHistory.length - 1]?.version).toBe(1002); + }); + + test("ignores empty updates without bumping the version", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + const result = storage.applyOps([updateTextOp("0:1", 0, [])]); + + expect(result[0]?.action).toBe("ignored"); + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }); + expect(storage.driver.get_live_text_history_since("0:1", 0)).toHaveLength( + 0 + ); + }); + + test("a delete rebased over a concurrent interior insert preserves the insert", () => { + const driver = new InMemoryDriver(); + initInMemory( + driver, + new Map([ + ["root", { type: CrdtType.OBJECT, data: {} }], + [ + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["abcdef"]], + version: 0, + }, + ], + ]) + ); + + const storage = new Storage(driver); + + // Client A inserts "ZZ" at index 3 (accepted first) + const insertResult = storage.applyOps([ + updateTextOp("0:1", 0, [{ type: "insert", index: 3, text: "ZZ" }], "a:1"), + ]); + expect(insertResult[0]?.action).toBe("accepted"); + + // Client B concurrently deletes [1, 5) ("bcde"), based on version 0 + const deleteResult = storage.applyOps([ + updateTextOp("0:1", 0, [{ type: "delete", index: 1, length: 4 }], "b:1"), + ]); + expect(deleteResult[0]?.action).toBe("accepted"); + + // The concurrently inserted text survives the spanning delete + expect(storage.driver.get_node("0:1")).toEqual({ + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["aZZf"]], + version: 2, + }); + }); +}); diff --git a/packages/liveblocks-server/test/decoders/Op.test.ts b/packages/liveblocks-server/test/decoders/Op.test.ts new file mode 100644 index 0000000000..4c9899b599 --- /dev/null +++ b/packages/liveblocks-server/test/decoders/Op.test.ts @@ -0,0 +1,81 @@ +/** + * Copyright (c) Liveblocks Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { OpCode } from "@liveblocks/core"; +import { describe, expect, test } from "vitest"; + +import { op } from "~/decoders/Op"; + +describe("LiveText op decoder", () => { + test.each([-1, 0.5, Number.MAX_SAFE_INTEGER + 1])( + "rejects unsafe CREATE_TEXT version %s", + (version) => { + expect( + op.decode({ + type: OpCode.CREATE_TEXT, + opId: "1:1", + id: "1:1", + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version, + }).ok + ).toBe(false); + } + ); + + test.each([-1, 0.5, Number.MAX_SAFE_INTEGER + 1])( + "rejects unsafe UPDATE_TEXT base version %s", + (baseVersion) => { + expect( + op.decode({ + type: OpCode.UPDATE_TEXT, + opId: "1:2", + id: "1:1", + baseVersion, + ops: [{ type: "insert", index: 0, text: "Hello" }], + }).ok + ).toBe(false); + } + ); + + test("rejects an unsafe authoritative UPDATE_TEXT version", () => { + expect( + op.decode({ + type: OpCode.UPDATE_TEXT, + opId: "1:2", + id: "1:1", + baseVersion: 0, + version: 0.5, + ops: [{ type: "insert", index: 0, text: "Hello" }], + }).ok + ).toBe(false); + }); + + test("accepts non-negative safe LiveText versions", () => { + expect( + op.decode({ + type: OpCode.UPDATE_TEXT, + opId: "1:2", + id: "1:1", + baseVersion: 0, + version: Number.MAX_SAFE_INTEGER, + ops: [{ type: "insert", index: 0, text: "Hello" }], + }).ok + ).toBe(true); + }); +}); diff --git a/packages/liveblocks-server/test/formats/PlainLson.test.ts b/packages/liveblocks-server/test/formats/PlainLson.test.ts index da82727a36..e78235dfb8 100644 --- a/packages/liveblocks-server/test/formats/PlainLson.test.ts +++ b/packages/liveblocks-server/test/formats/PlainLson.test.ts @@ -271,6 +271,49 @@ describe("Serialization of nodes (to PlainLson format)", () => { convertedNodes.delete("root"); expect(convertedNodes).toEqual(new Map(nodes)); }); + + test("LiveText serializes without exposing version", () => { + const nodes: StorageNode[] = [ + [ + "si:1", + { + parentId: "root", + parentKey: "text", + type: CrdtType.TEXT, + data: [["Hello", { bold: true }], [" world"]], + version: 2, + }, + ], + ]; + + const plainLson = snapshotToPlainLson(makeSnapshot(nodes)); + expect(plainLson).toEqual({ + liveblocksType: "LiveObject", + data: { + text: { + liveblocksType: "LiveText", + data: [["Hello", { bold: true }], [" world"]], + }, + }, + }); + + const convertedNodes = plainLsonToNodeMap(plainLson); + convertedNodes.delete("root"); + expect(convertedNodes).toEqual( + new Map([ + [ + "si:1", + { + parentId: "root", + parentKey: "text", + type: CrdtType.TEXT, + data: [["Hello", { bold: true }], [" world"]], + version: 0, + }, + ], + ]) + ); + }); }); describe("Deserialization of nodes (from PlainLson format)", () => { @@ -389,6 +432,36 @@ describe("Deserialization of nodes (from PlainLson format)", () => { ); }); + test("LiveText initializes with internal version 0", () => { + const tree: PlainLsonObject = { + liveblocksType: "LiveObject", + data: { + text: { + liveblocksType: "LiveText", + data: [["Hello"]], + }, + }, + }; + + const convertedNodes = plainLsonToNodeMap(tree); + convertedNodes.delete("root"); + + expect(convertedNodes).toEqual( + new Map([ + [ + "si:1", + { + parentId: "root", + parentKey: "text", + type: CrdtType.TEXT, + data: [["Hello"]], + version: 0, + }, + ], + ]) + ); + }); + test("Invalid liveblocksType should throw error", () => { const tree = { liveblocksType: "LiveObject", @@ -467,6 +540,22 @@ describe("streaming === nonstreaming equivalence", () => { expect(streamingResult(nodes)).toBe(nonstreamingResult(nodes)); }); + test("LiveText", () => { + const nodes: StorageNode[] = [ + [ + "si:1", + { + parentId: "root", + parentKey: "text", + type: CrdtType.TEXT, + data: [["Hello", { bold: true }], [" world"]], + version: 2, + }, + ], + ]; + expect(streamingResult(nodes)).toBe(nonstreamingResult(nodes)); + }); + test("deeply nested: Object > List > Object > Map > Object", () => { // prettier-ignore const nodes: StorageNode[] = [ diff --git a/packages/liveblocks-server/test/lib/liveTextOps.test.ts b/packages/liveblocks-server/test/lib/liveTextOps.test.ts new file mode 100644 index 0000000000..e4fe1c227d --- /dev/null +++ b/packages/liveblocks-server/test/lib/liveTextOps.test.ts @@ -0,0 +1,33 @@ +/** + * Copyright (c) Liveblocks Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { applyLiveTextOperations } from "@liveblocks/core"; +import { describe, expect, test } from "vitest"; + +describe("liveTextOps", () => { + test("applyLiveTextOperations formats and inserts text", () => { + expect( + applyLiveTextOperations([["Hello"]], [ + { type: "format", index: 0, length: 5, attributes: { bold: true } }, + { type: "insert", index: 5, text: "!" }, + ]) + ).toEqual([ + ["Hello", { bold: true }], + ["!"], + ]); + }); +}); diff --git a/packages/liveblocks-server/test/storage/model-based/live-text-model.ts b/packages/liveblocks-server/test/storage/model-based/live-text-model.ts new file mode 100644 index 0000000000..03399ffaf9 --- /dev/null +++ b/packages/liveblocks-server/test/storage/model-based/live-text-model.ts @@ -0,0 +1,527 @@ +/** + * Copyright (c) Liveblocks Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import type { + JsonObject, + LiveTextData, + NodeStream, + SerializedCrdt, + TextOperation, +} from "@liveblocks/core"; +import { + applyLiveTextOperations, + CrdtType, + OpCode, + transformTextOperations, +} from "@liveblocks/core"; +import * as fc from "fast-check"; +import { expect } from "vitest"; + +import type { ClientWireOp } from "~/protocol"; +import type { Storage as RealStorage } from "~/Storage"; +import { selfCheck } from "~test/plugins/_generateFullTestSuite"; + +// ----------------------------------------------------------------------------- +// Arbitraries +// ----------------------------------------------------------------------------- + +const ALPHABET = "abcdefgh"; + +const textArb = fc + .array(fc.integer({ min: 0, max: ALPHABET.length - 1 }), { + minLength: 1, + maxLength: 4, + }) + .map((indexes) => indexes.map((i) => ALPHABET[i]).join("")); + +const attributesArb: fc.Arbitrary = fc.oneof( + fc.constant({ bold: true }), + fc.constant({ bold: null }), + fc.constant({ italic: 1 }), + fc.constant({ bold: true, italic: null }), + fc.constant({ color: "red" }) +); + +/** + * Seeds that get concretized against the document length at the op's base + * version, at command run time. Commands cannot carry concrete TextOperations, + * because the document they will apply to is only known once all preceding + * commands in the sequence have run. + */ +type EditSeed = + | { type: "insert"; at: number; text: string; attrs: boolean } + | { type: "delete"; at: number; len: number } + | { type: "format"; at: number; len: number; attrs: JsonObject }; + +const editSeedArb: fc.Arbitrary = fc.oneof( + fc.record({ + type: fc.constant("insert" as const), + at: fc.nat(1000), + text: textArb, + attrs: fc.boolean(), + }), + fc.record({ + type: fc.constant("delete" as const), + at: fc.nat(1000), + len: fc.integer({ min: 1, max: 4 }), + }), + fc.record({ + type: fc.constant("format" as const), + at: fc.nat(1000), + len: fc.integer({ min: 1, max: 4 }), + attrs: attributesArb, + }) +); + +/** Arbitrary initial LiveText document (0–3 segments). */ +export const docArb: fc.Arbitrary = fc + .array( + fc.record({ + text: textArb, + attrs: fc.option(attributesArb, { nil: undefined }), + }), + { minLength: 0, maxLength: 3 } + ) + .map((segments) => + segments.map(({ text, attrs }) => + attrs === undefined ? [text] : ([text, attrs] as const) + ) + ) as fc.Arbitrary; + +function concretize(seed: EditSeed, length: number): TextOperation | undefined { + if (seed.type === "insert") { + return { + type: "insert", + index: seed.at % (length + 1), + text: seed.text, + ...(seed.attrs ? { attributes: { bold: true } } : {}), + }; + } + if (length === 0) { + return undefined; + } + const index = seed.at % length; + const len = Math.min(seed.len, length - index); + if (len <= 0) { + return undefined; + } + if (seed.type === "delete") { + return { type: "delete", index, length: len }; + } + return { type: "format", index, length: len, attributes: seed.attrs }; +} + +/** Generate a sequential op list valid against a doc of the given length. */ +function concretizeSequence( + seeds: readonly EditSeed[], + initialLength: number +): TextOperation[] { + const ops: TextOperation[] = []; + let length = initialLength; + for (const seed of seeds) { + const op = concretize(seed, length); + if (op === undefined) { + continue; + } + ops.push(op); + if (op.type === "insert") { + length += op.text.length; + } else if (op.type === "delete") { + length -= op.length; + } + } + return ops; +} + +function dataLength(data: LiveTextData): number { + let total = 0; + for (const segment of data) { + total += segment[0].length; + } + return total; +} + +// ----------------------------------------------------------------------------- +// Reference model +// ----------------------------------------------------------------------------- + +type AcceptedEntry = { + opId: string; + baseVersion: number; + version: number; + ops: TextOperation[]; +}; + +class TextNodeModel { + /** Expected document contents (kept in sync via the server's authoritative ops). */ + data: LiveTextData; + /** Expected node version. */ + version: number; + /** + * The lowest baseVersion the server can still rebase against. Initial nodes + * are seeded without history, so this is the node's initial version: ops + * based on anything older must be rejected as "older than retained history". + * (LIVE_TEXT_HISTORY_LIMIT never kicks in here, since model runs are far + * shorter than 1000 commands.) + */ + minBaseVersion: number; + /** Document length at each version >= minBaseVersion, to concretize stale ops. */ + lengthAt: Map; + /** Accepted history entries, used to replay duplicates. */ + history: AcceptedEntry[]; + + constructor(data: LiveTextData, version: number) { + // Deep-copy, so the model never aliases the driver's node data + this.data = JSON.parse(JSON.stringify(data)) as LiveTextData; + this.version = version; + this.minBaseVersion = version; + this.lengthAt = new Map([[version, dataLength(this.data)]]); + this.history = []; + } +} + +export class Model { + texts: Map; + nextOpId: number; + + constructor(nodeStream: NodeStream) { + this.texts = new Map(); + this.nextOpId = 1; + for (const [id, node] of nodeStream as Iterable<[string, SerializedCrdt]>) { + if (node.type === CrdtType.TEXT) { + this.texts.set(id, new TextNodeModel(node.data, node.version)); + } + } + } + + pickTextNode(seed: number): [string, TextNodeModel] { + const entries = Array.from(this.texts.entries()); + const picked = entries[seed % entries.length]; + /* istanbul ignore next */ + if (picked === undefined) { + throw new Error("No text nodes available (check() should prevent this)"); + } + return picked; + } +} + +// ----------------------------------------------------------------------------- +// Assertion helpers +// ----------------------------------------------------------------------------- + +function expectNodeMatchesModel( + real: RealStorage, + nodeId: string, + expected: TextNodeModel +): void { + const node = real.driver.get_node(nodeId); + expect(node?.type).toBe(CrdtType.TEXT); + if (node?.type !== CrdtType.TEXT) { + return; // unreachable; narrows the type + } + expect(node.version).toBe(expected.version); + expect(node.data).toEqual(expected.data); +} + +function makeUpdateTextOp( + opId: string, + id: string, + baseVersion: number, + ops: TextOperation[] +): ClientWireOp { + return { opId, id, type: OpCode.UPDATE_TEXT, baseVersion, ops }; +} + +// ----------------------------------------------------------------------------- +// Commands +// ----------------------------------------------------------------------------- + +/** + * The bread-and-butter command: sends an UpdateTextOp with 1–3 text + * operations, based at a (possibly stale, possibly too-old) base version. + * + * - baseVersion === current version → plain accept + * - minBaseVersion <= baseVersion < version → accept, rebased over history + * - baseVersion < minBaseVersion → rejected ("older than retained history") + * - concretized ops turn out empty → ignored + * + * The oracle: the model tracks the authoritative ops accepted at every + * version, and independently computes the expected rebased ops with + * transformTextOperations over that history. The server's echoed ops must + * match exactly, and so must the persisted node data after applying them to + * the reference document. (Correctness of the transform primitives themselves + * is covered by the TP1 fuzz tests in @liveblocks/core; this test verifies + * that Storage feeds them the right history, in the right order.) + */ +class EditTextCommand implements fc.Command { + constructor( + readonly nodeSeed: number, + readonly lagSeed: number, + readonly seeds: readonly EditSeed[] + ) {} + + check(model: Model): boolean { + return model.texts.size > 0; + } + + run(model: Model, real: RealStorage): void { + const [nodeId, text] = model.pickTextNode(this.nodeSeed); + + // Pick a baseVersion in [max(0, minBaseVersion - 2), version], so we + // occasionally dip below the retained history floor (when the node was + // seeded with version > 0) and trigger a rejection. + const lowest = Math.max(0, text.minBaseVersion - 2); + const span = text.version - lowest; + const baseVersion = text.version - (this.lagSeed % (span + 1)); + + // Concretize against the document length at the base version, like a real + // client editing an older snapshot would. + const lengthAtBase = + text.lengthAt.get(Math.max(baseVersion, text.minBaseVersion)) ?? 0; + const ops = concretizeSequence(this.seeds, lengthAtBase); + + const opId = `op:${model.nextOpId++}`; + const result = real.applyOps([ + makeUpdateTextOp(opId, nodeId, baseVersion, ops), + ])[0]; + + if (ops.length === 0) { + expect(result).toMatchObject({ action: "ignored" }); + } else if (baseVersion < text.minBaseVersion) { + expect(result).toMatchObject({ + action: "rejected", + opIds: [opId], + reason: "LiveText operation is older than retained history", + }); + } else { + expect(result?.action).toBe("accepted"); + if (result?.action !== "accepted") { + return; // unreachable; narrows the type + } + expect(result.op.type).toBe(OpCode.UPDATE_TEXT); + if (result.op.type !== OpCode.UPDATE_TEXT) { + return; // unreachable; narrows the type + } + + // The server echoes the op with authoritative fields + expect(result.op.opId).toBe(opId); + expect(result.op.baseVersion).toBe(text.version); + expect(result.op.version).toBe(text.version + 1); + + // Independently rebase the ops over the accepted history since + // baseVersion: the server must produce exactly these + const opsSinceBase = text.history + .filter((entry) => entry.version > baseVersion) + .flatMap((entry) => entry.ops); + const authoritativeOps = + opsSinceBase.length > 0 + ? transformTextOperations(ops, opsSinceBase, "after") + : ops; + expect(result.op.ops).toEqual(authoritativeOps); + + // Applying the authoritative (rebased) ops to the reference document + // must yield exactly the persisted node data + text.data = applyLiveTextOperations(text.data, authoritativeOps); + text.version += 1; + text.lengthAt.set(text.version, dataLength(text.data)); + text.history.push({ + opId, + baseVersion: text.version - 1, + version: text.version, + ops: authoritativeOps.map((op) => ({ ...op })), + }); + } + + // In all cases (accepted or not), the persisted node must match the model + expectNodeMatchesModel(real, nodeId, text); + selfCheck(real); + } + + toString(): string { + return `\n\n/* EditTextCommand */\n${JSON.stringify({ + nodeSeed: this.nodeSeed, + lagSeed: this.lagSeed, + seeds: this.seeds, + })}`; + } +} + +/** + * Re-sends a previously accepted op (same opId). The server must answer with + * a "rectified" ack carrying the originally stored authoritative fields, and + * must not apply the op twice. + */ +class DuplicateOpCommand implements fc.Command { + constructor( + readonly nodeSeed: number, + readonly pickSeed: number + ) {} + + check(model: Model): boolean { + return Array.from(model.texts.values()).some( + (text) => text.history.length > 0 + ); + } + + run(model: Model, real: RealStorage): void { + const candidates = Array.from(model.texts.entries()).filter( + ([, text]) => text.history.length > 0 + ); + const [nodeId, text] = candidates[this.nodeSeed % candidates.length]!; + const entry = text.history[this.pickSeed % text.history.length]!; + + const result = real.applyOps([ + makeUpdateTextOp(entry.opId, nodeId, entry.baseVersion, entry.ops), + ])[0]; + + expect(result).toMatchObject({ + action: "rectified", + ackOp: { + opId: entry.opId, + id: nodeId, + baseVersion: entry.baseVersion, + version: entry.version, + ops: entry.ops, + }, + }); + + // Nothing must have been applied twice + expectNodeMatchesModel(real, nodeId, text); + selfCheck(real); + } + + toString(): string { + return `\n\n/* DuplicateOpCommand */\n${JSON.stringify({ + nodeSeed: this.nodeSeed, + pickSeed: this.pickSeed, + })}`; + } +} + +/** + * Sends an op with an empty ops list. The server must acknowledge it as + * "ignored" without bumping the version or appending history. + */ +class EmptyOpsCommand implements fc.Command { + constructor(readonly nodeSeed: number) {} + + check(model: Model): boolean { + return model.texts.size > 0; + } + + run(model: Model, real: RealStorage): void { + const [nodeId, text] = model.pickTextNode(this.nodeSeed); + + const opId = `op:${model.nextOpId++}`; + const result = real.applyOps([ + makeUpdateTextOp(opId, nodeId, text.version, []), + ])[0]; + + expect(result).toMatchObject({ action: "ignored", ignoredOpId: opId }); + expectNodeMatchesModel(real, nodeId, text); + selfCheck(real); + } + + toString(): string { + return `\n\n/* EmptyOpsCommand */\n${JSON.stringify({ + nodeSeed: this.nodeSeed, + })}`; + } +} + +/** + * Sends an op whose baseVersion is ahead of the server's version. The server + * must reject it and leave the node untouched. + */ +class FutureBaseVersionCommand implements fc.Command { + constructor( + readonly nodeSeed: number, + readonly ahead: number, + readonly text: string + ) {} + + check(model: Model): boolean { + return model.texts.size > 0; + } + + run(model: Model, real: RealStorage): void { + const [nodeId, text] = model.pickTextNode(this.nodeSeed); + + const opId = `op:${model.nextOpId++}`; + const result = real.applyOps([ + makeUpdateTextOp(opId, nodeId, text.version + this.ahead, [ + { type: "insert", index: 0, text: this.text }, + ]), + ])[0]; + + expect(result).toMatchObject({ + action: "rejected", + opIds: [opId], + reason: "LiveText operation base version is ahead of storage", + }); + expectNodeMatchesModel(real, nodeId, text); + selfCheck(real); + } + + toString(): string { + return `\n\n/* FutureBaseVersionCommand */\n${JSON.stringify({ + nodeSeed: this.nodeSeed, + ahead: this.ahead, + text: this.text, + })}`; + } +} + +// ----------------------------------------------------------------------------- +// Command sequence generator +// ----------------------------------------------------------------------------- + +export function commands(options?: { + size?: fc.SizeForArbitrary; + replayPath?: string; +}) { + return fc.commands( + [ + // Normal/stale/too-old edits make up the bulk of the traffic + ...Array.from({ length: 8 }, () => + fc + .tuple( + fc.nat(1000), + fc.nat(1000), + fc.array(editSeedArb, { minLength: 1, maxLength: 3 }) + ) + .map( + ([nodeSeed, lagSeed, seeds]) => + new EditTextCommand(nodeSeed, lagSeed, seeds) + ) + ), + fc + .tuple(fc.nat(1000), fc.nat(1000)) + .map( + ([nodeSeed, pickSeed]) => new DuplicateOpCommand(nodeSeed, pickSeed) + ), + fc.nat(1000).map((nodeSeed) => new EmptyOpsCommand(nodeSeed)), + fc + .tuple(fc.nat(1000), fc.integer({ min: 1, max: 5 }), textArb) + .map( + ([nodeSeed, ahead, text]) => + new FutureBaseVersionCommand(nodeSeed, ahead, text) + ), + ], + options + ); +} diff --git a/packages/liveblocks-server/test/storage/model-based/live-text.model.test.ts b/packages/liveblocks-server/test/storage/model-based/live-text.model.test.ts new file mode 100644 index 0000000000..48616c4aa6 --- /dev/null +++ b/packages/liveblocks-server/test/storage/model-based/live-text.model.test.ts @@ -0,0 +1,119 @@ +/** + * Copyright (c) Liveblocks Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import type { LiveTextData, StorageNode } from "@liveblocks/core"; +import { CrdtType } from "@liveblocks/core"; +import fc from "fast-check"; +import { describe, test } from "vitest"; + +import { selfCheck } from "~test/plugins/_generateFullTestSuite"; + +import { runWithStorage } from "../utils"; +import { commands, docArb, Model } from "./live-text-model"; + +type InitialText = { + data: LiveTextData; + version: number; +}; + +const initialTextsArb = fc.array( + fc.record({ + data: docArb, + version: fc.nat(3), + }), + { minLength: 1, maxLength: 3 } +); + +/** + * Builds an initial node tree with a root object and the given LiveText nodes + * attached to it (under keys text0, text1, ...). + */ +function makeNodeStream(texts: InitialText[]): StorageNode[] { + const nodes: StorageNode[] = [ + ["root", { type: CrdtType.OBJECT, data: {} }], + ]; + for (const [i, { data, version }] of texts.entries()) { + nodes.push([ + `text:${i}`, + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: `text${i}`, + data, + version, + }, + ]); + } + return nodes; +} + +describe("Storage LiveText (model-based test)", () => { + test( + "matches the reference model no matter what UpdateTextOps are applied", + { timeout: 22_000 }, + async () => + fc.assert( + fc.asyncProperty( + fc.record({ + initialTexts: initialTextsArb, + // NOTE: "+2" (like the generic storage model test uses) is too + // slow here: per-command cost grows with the accepted history + // (stale ops rebase over everything since their baseVersion), so + // long sequences get quadratically expensive and starve the + // min-iterations budget on slower CI runners. + commands: commands({ + size: "+1", + // replayPath: "" + }), + }), + + ({ initialTexts, commands }) => + // Set up real system and reference model. The model tracks the + // expected document data and version per LiveText node, updated by + // applying the server's authoritative (rebased) ops. Each command + // also predicts the result action (accepted / rectified / ignored + // / rejected) and runs the internal-consistency selfCheck. + + runWithStorage(makeNodeStream(initialTexts), ({ storage: real }) => { + selfCheck(real); + + const model = new Model(real.driver.iter_nodes()); + + // Tries running randomized sequences of UpdateTextOps (normal, + // stale, too-old, duplicate, empty, and future-based ops) + fc.modelRun(() => ({ model, real }), commands); + }) + ), + { + numRuns: 200, // Stop after 200 iterations, or... + interruptAfterTimeLimit: 20_000, // ...after 20 seconds (whichever comes first) + reporter: (out) => { + if (out.failed) { + throw new Error(fc.defaultReportMessage(out)); + } + // Expect at least 50 iterations, though + const MIN_ITERATIONS = 50; + if (out.numRuns < MIN_ITERATIONS) { + throw new Error( + `Expected at least ${MIN_ITERATIONS} iterations, but only ran ${out.numRuns} (why so slow?)` + ); + } + }, + } + ) + ); +}); diff --git a/packages/liveblocks-server/test/storage/model-based/storage-model.ts b/packages/liveblocks-server/test/storage/model-based/storage-model.ts index c674b95449..905f454b46 100644 --- a/packages/liveblocks-server/test/storage/model-based/storage-model.ts +++ b/packages/liveblocks-server/test/storage/model-based/storage-model.ts @@ -39,6 +39,7 @@ function wsApiVersion(): fc.Arbitrary { export class Model { availableParentNodeIds: Set; availableObjectNodeIds: Set; + availableTextNodeIds: Set; constructor(nodeStream: NodeStream) { const nodeMap = new Map(nodeStream); @@ -46,9 +47,13 @@ export class Model { const objIds = Array.from(nodeMap.entries()) .filter(([_, node]) => node.type === CrdtType.OBJECT) .map(([id]) => id); + const textIds = Array.from(nodeMap.entries()) + .filter(([_, node]) => node.type === CrdtType.TEXT) + .map(([id]) => id); this.availableParentNodeIds = new Set(allIds); this.availableObjectNodeIds = new Set(objIds); + this.availableTextNodeIds = new Set(textIds); } } @@ -70,6 +75,7 @@ class ApplyOpCommand implements fc.Command { case OpCode.CREATE_LIST: case OpCode.CREATE_MAP: case OpCode.CREATE_REGISTER: + case OpCode.CREATE_TEXT: case OpCode.CREATE_FILE: return model.availableParentNodeIds.has(this.op.parentId); @@ -81,6 +87,9 @@ class ApplyOpCommand implements fc.Command { case OpCode.DELETE_OBJECT_KEY: return model.availableObjectNodeIds.has(this.op.id); + case OpCode.UPDATE_TEXT: + return model.availableTextNodeIds.has(this.op.id); + default: return assertNever(this.op, "Unhandled case"); } @@ -105,6 +114,7 @@ class ApplyOpCommand implements fc.Command { break; case OpCode.CREATE_REGISTER: + case OpCode.CREATE_TEXT: case OpCode.CREATE_FILE: // Don't register leaf nodes as potential parent IDs break; @@ -128,6 +138,7 @@ class ApplyOpCommand implements fc.Command { case OpCode.SET_PARENT_KEY: case OpCode.UPDATE_OBJECT: case OpCode.DELETE_OBJECT_KEY: + case OpCode.UPDATE_TEXT: break; default: @@ -162,6 +173,11 @@ class ApplyOpCommand implements fc.Command { return "UpdateObjectOp"; case OpCode.DELETE_OBJECT_KEY: return "DeleteObjectKeyOp"; + case OpCode.CREATE_TEXT: + return "CreateTextOp"; + case OpCode.UPDATE_TEXT: + return "UpdateTextOp"; + default: return assertNever(op, "Unhandled case"); } diff --git a/tools/liveblocks-cli/package.json b/tools/liveblocks-cli/package.json index 9f107f05d2..e298141271 100644 --- a/tools/liveblocks-cli/package.json +++ b/tools/liveblocks-cli/package.json @@ -46,7 +46,7 @@ "typescript": "^5.9.3" }, "dependencies": { - "@liveblocks/core": "3.23.0-file1", + "@liveblocks/core": "3.23.1-exp3", "@liveblocks/query-parser": "workspace:^", "@liveblocks/server": "workspace:^", "@liveblocks/zenrouter": "^1.2.0", diff --git a/tools/liveblocks-cli/src/dev-server/db/BunSQLiteDriver.ts b/tools/liveblocks-cli/src/dev-server/db/BunSQLiteDriver.ts index 5564012898..c74f47678c 100644 --- a/tools/liveblocks-cli/src/dev-server/db/BunSQLiteDriver.ts +++ b/tools/liveblocks-cli/src/dev-server/db/BunSQLiteDriver.ts @@ -22,6 +22,7 @@ import type { Json, JsonObject, LiveFileData, + LiveTextData, NodeStream, PlainLsonObject, Relax, @@ -29,6 +30,7 @@ import type { SerializedCrdt, SerializedObject, SerializedRootObject, + TextOperation, } from "@liveblocks/core"; import { asPos, CrdtType, raise } from "@liveblocks/core"; import type { @@ -52,6 +54,14 @@ import { } from "@liveblocks/server"; import { Database, type SQLQueryBindings } from "bun:sqlite"; +type LiveTextHistoryEntry = { + nodeId: string; + version: number; + baseVersion: number; + opId: string; + ops: TextOperation[]; +}; + function tryParseJson( value: string | undefined ): J | undefined { @@ -74,6 +84,7 @@ type NodeRow = parent_id: null; parent_key: null; jdata: jstring; + version: null; } | { id: string; @@ -81,6 +92,15 @@ type NodeRow = parent_id: string; parent_key: string; jdata: jstring; + version: null; + } + | { + id: string; + type: CrdtType.TEXT; + parent_id: string; + parent_key: string; + jdata: jstring; + version: number; } | { id: string; @@ -88,8 +108,29 @@ type NodeRow = parent_id: string; parent_key: string; jdata: null; + version: null; }; +type LiveTextHistoryRow = { + node_id: string; + version: number; + base_version: number; + op_id: string; + ops: jstring; +}; + +function rowToLiveTextHistoryEntry( + row: LiveTextHistoryRow +): LiveTextHistoryEntry { + return { + nodeId: row.node_id, + version: row.version, + baseVersion: row.base_version, + opId: row.op_id, + ops: parseJson(row.ops), + }; +} + function rowToSerializedCrdt(row: NodeRow): SerializedCrdt { const { type, parent_id: parentId, parent_key: parentKey, jdata } = row; switch (type) { @@ -101,6 +142,14 @@ function rowToSerializedCrdt(row: NodeRow): SerializedCrdt { case CrdtType.REGISTER: return { type, parentId, parentKey, data: parseJson(jdata) }; + case CrdtType.TEXT: + return { + type, + parentId, + parentKey, + data: parseJson(jdata), + version: row.version, + }; case CrdtType.FILE: return { type, @@ -186,15 +235,15 @@ function nparams(count: number): string { */ function sanitize_missingRoot(db: Database): void { db.run( - `INSERT OR IGNORE INTO nodes (id, type, parent_id, parent_key, jdata) - VALUES ('root', 0, NULL, NULL, '{}')` + `INSERT OR IGNORE INTO nodes (id, type, parent_id, parent_key, jdata, version) + VALUES ('root', 0, NULL, NULL, '{}', NULL)` ); } /** * Deletes illegal tree nodes and their subtrees: * 1. Registers under Objects - * 2. Any child node under a Register + * 2. Any child node under a Register or Text * * Common case is also the happy path: no rows match, this is a no-op. */ @@ -208,7 +257,7 @@ function sanitize_illegalNodes(db: Database): void { WHERE (c.type = ${CrdtType.REGISTER} AND p.type = ${CrdtType.OBJECT}) OR - p.type IN (${CrdtType.REGISTER}, ${CrdtType.FILE})` + p.type IN (${CrdtType.REGISTER}, ${CrdtType.TEXT}, ${CrdtType.FILE})` ) .all(); @@ -264,7 +313,7 @@ function get_node(db: Database, id: string): SerializedCrdt | undefined { .query< NodeRow, [string] - >("SELECT id, type, parent_id, parent_key, jdata FROM nodes WHERE id = ?") + >("SELECT id, type, parent_id, parent_key, jdata, version FROM nodes WHERE id = ?") .get(id); return row ? rowToSerializedCrdt(row) : undefined; } @@ -274,7 +323,7 @@ function iter_nodes(db: Database): Iterable<[string, SerializedCrdt]> { .query< NodeRow, [] - >("SELECT id, type, parent_id, parent_key, jdata FROM nodes") + >("SELECT id, type, parent_id, parent_key, jdata, version FROM nodes") .all() .map(rowToIdTuple); } @@ -295,6 +344,10 @@ function iter_nodes_optimized(db: Database): Iterable> { WHEN jdata IS NULL THEN '[' || json_quote(id) || ',' || type || ',' || json_quote(parent_id) || ',' || json_quote(parent_key) || ']' + WHEN type = ${CrdtType.TEXT} THEN + '[' || json_quote(id) || ',' || type || ',' || + json_quote(parent_id) || ',' || json_quote(parent_key) || ',' || + jdata || ',' || version || ']' ELSE '[' || json_quote(id) || ',' || type || ',' || json_quote(parent_id) || ',' || json_quote(parent_key) || ',' || jdata || ']' @@ -359,16 +412,18 @@ function upsert_node(db: Database, id: string, node: SerializedCrdt): void { const jdata = node.type === CrdtType.OBJECT || node.type === CrdtType.REGISTER || + node.type === CrdtType.TEXT || node.type === CrdtType.FILE ? JSON.stringify(node.data) : null; + const version = node.type === CrdtType.TEXT ? node.version : null; db.query( - `INSERT INTO nodes (id, type, parent_id, parent_key, jdata) - VALUES (?, ?, ?, ?, ?) + `INSERT INTO nodes (id, type, parent_id, parent_key, jdata, version) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (id) DO - UPDATE SET type = excluded.type, parent_id = excluded.parent_id, parent_key = excluded.parent_key, jdata = excluded.jdata` - ).run(id, node.type, parentId, parentKey, jdata); + UPDATE SET type = excluded.type, parent_id = excluded.parent_id, parent_key = excluded.parent_key, jdata = excluded.jdata, version = excluded.version` + ).run(id, node.type, parentId, parentKey, jdata, version); } /** @@ -568,7 +623,8 @@ export class BunSQLiteDriver implements IStorageDriver { -- ^^^^^^^ 0=LiveObject, 1=LiveList, 2=LiveMap, 3=Register, 4=LiveText, 5=LiveFile parent_id TEXT, -- NULL only for root parent_key TEXT, -- NULL only for root - jdata TEXT, -- JSON data for LiveObject, Register, LiveText, and LiveFile; NULL for LiveList/LiveMap + jdata TEXT, -- JSON data for LiveObject, Register, and LiveText; NULL for LiveList/LiveMap + version INTEGER, -- LiveText version; NULL for all other types UNIQUE (parent_id, parent_key), @@ -585,6 +641,10 @@ export class BunSQLiteDriver implements IStorageDriver { CHECK (type != 2 OR jdata IS NULL), -- LiveMap must NOT have jdata CHECK (type != 3 OR jdata IS NOT NULL), -- Register must have jdata (even "null" is stored as JSON string) CHECK (type != 4 OR jdata IS NOT NULL), -- LiveText must have jdata + + -- Only LiveText carries a version + CHECK (type = 4 OR version IS NULL), -- non-LiveText must NOT have a version + CHECK (type != 4 OR version IS NOT NULL), -- LiveText must have a version CHECK (type != 5 OR jdata IS NOT NULL), -- LiveFile must have jdata -- Foreign key: parent_id must reference an existing node @@ -592,6 +652,21 @@ export class BunSQLiteDriver implements IStorageDriver { ) STRICT` ); + db.run( + `CREATE TABLE IF NOT EXISTS live_text_op_history ( + node_id TEXT NOT NULL, + version INTEGER NOT NULL, + base_version INTEGER NOT NULL, + op_id TEXT NOT NULL, + ops TEXT NOT NULL, + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + + PRIMARY KEY (node_id, version), + UNIQUE (node_id, op_id), + FOREIGN KEY (node_id) REFERENCES nodes (id) ON DELETE CASCADE + ) STRICT` + ); + // Create a table to read/write JSON values db.run( `CREATE TABLE IF NOT EXISTS metadata ( @@ -777,6 +852,64 @@ export class BunSQLiteDriver implements IStorageDriver { return this.loadedApi.get_snapshot(lowMemory); } + get_live_text_history_since( + nodeId: string, + version: number + ): LiveTextHistoryEntry[] { + const rows = this.db + .query( + `SELECT node_id, version, base_version, op_id, ops + FROM live_text_op_history + WHERE node_id = ? AND version > ? + ORDER BY version ASC` + ) + .all(nodeId, version); + return rows.map(rowToLiveTextHistoryEntry); + } + + get_live_text_history_by_op_id( + nodeId: string, + opId: string + ): LiveTextHistoryEntry | undefined { + const row = this.db + .query( + `SELECT node_id, version, base_version, op_id, ops + FROM live_text_op_history + WHERE node_id = ? AND op_id = ? + LIMIT 1` + ) + .get(nodeId, opId); + return row === null ? undefined : rowToLiveTextHistoryEntry(row); + } + + append_live_text_history(entry: LiveTextHistoryEntry): void { + this.db + .query( + `INSERT INTO live_text_op_history + (node_id, version, base_version, op_id, ops) + VALUES (?, ?, ?, ?, ?)` + ) + .run( + entry.nodeId, + entry.version, + entry.baseVersion, + entry.opId, + JSON.stringify(entry.ops) + ); + } + + purge_live_text_history_before( + nodeId: string, + minVersionToKeep: number + ): void { + this.db + .query( + `DELETE FROM live_text_op_history + WHERE node_id = ? AND version < ?` + ) + .run(nodeId, minVersionToKeep); + } + private _loadNodesApi(): NodesAPI { const db = this.db; @@ -826,7 +959,7 @@ export class BunSQLiteDriver implements IStorageDriver { this.reinitialize(); const insertStm = this.db.prepare( - "INSERT INTO nodes (id, type, parent_id, parent_key, jdata) VALUES (?, ?, ?, ?, ?)" + "INSERT INTO nodes (id, type, parent_id, parent_key, jdata, version) VALUES (?, ?, ?, ?, ?, ?)" ); const resetNodes = this.db.transaction(() => { // Defer FK checks until the end of the transaction so the bulk DELETE @@ -840,10 +973,12 @@ export class BunSQLiteDriver implements IStorageDriver { const jdata = node.type === CrdtType.OBJECT || node.type === CrdtType.REGISTER || + node.type === CrdtType.TEXT || node.type === CrdtType.FILE ? JSON.stringify(node.data) : null; - insertStm.run(id, node.type, parentId, parentKey, jdata); + const version = node.type === CrdtType.TEXT ? node.version : null; + insertStm.run(id, node.type, parentId, parentKey, jdata, version); } }); resetNodes(); @@ -855,7 +990,7 @@ export class BunSQLiteDriver implements IStorageDriver { .query< NodeRow, [] - >("SELECT id, type, parent_id, parent_key, jdata FROM nodes") + >("SELECT id, type, parent_id, parent_key, jdata, version FROM nodes") .all() .map(rowToIdTuple); } diff --git a/tools/liveblocks-cli/src/dev-server/db/rooms.ts b/tools/liveblocks-cli/src/dev-server/db/rooms.ts index 1f617d06e0..c54f9692ec 100644 --- a/tools/liveblocks-cli/src/dev-server/db/rooms.ts +++ b/tools/liveblocks-cli/src/dev-server/db/rooms.ts @@ -55,9 +55,7 @@ export type ClientMeta = JsonObject; // Module state // --------------------------------------------------------------------------- -// Bumped to v3 when the per-room node storage schema added new CRDT node -// types. Old v1/v2 data is left untouched on disk. -const DEFAULT_BASE_PATH = ".liveblocks/v3"; +const DEFAULT_BASE_PATH = ".liveblocks/v2"; let basePath = DEFAULT_BASE_PATH; let isEphemeral = false; let _initializedDb: Database | null = null; diff --git a/tools/liveblocks-cli/test/plugins/BunSQLiteDriver.test.ts b/tools/liveblocks-cli/test/plugins/BunSQLiteDriver.test.ts index 5c5204d7e9..700d83f321 100644 --- a/tools/liveblocks-cli/test/plugins/BunSQLiteDriver.test.ts +++ b/tools/liveblocks-cli/test/plugins/BunSQLiteDriver.test.ts @@ -22,7 +22,7 @@ import stdPath from "node:path"; import type { NodeMap } from "@liveblocks/core"; import { CrdtType } from "@liveblocks/core"; import { Database } from "bun:sqlite"; -import { describe } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { BunSQLiteDriver } from "~/dev-server/db/BunSQLiteDriver"; @@ -42,13 +42,15 @@ function initBunSQLite(dbPath: string, rawNodes: NodeMap): void { const jdata = node.type === CrdtType.OBJECT || node.type === CrdtType.REGISTER || + node.type === CrdtType.TEXT || node.type === CrdtType.FILE ? JSON.stringify(node.data) : null; + const version = node.type === CrdtType.TEXT ? node.version : null; db.run( - "INSERT INTO nodes (id, type, parent_id, parent_key, jdata) VALUES (?, ?, ?, ?, ?)", - [id, node.type, parentId, parentKey, jdata] + "INSERT INTO nodes (id, type, parent_id, parent_key, jdata, version) VALUES (?, ?, ?, ?, ?, ?)", + [id, node.type, parentId, parentKey, jdata, version] ); } @@ -77,3 +79,65 @@ describe("Bun SQLite driver", () => { }, }); }); + +describe("Bun SQLite driver LiveText history", () => { + test("stores, purges, and deletes LiveText operation history", () => { + const tmpdir = fs.mkdtempSync(stdPath.join(os.tmpdir(), "lb-sqlite-test-")); + const dbPath = stdPath.join(tmpdir, "my-test-room.db"); + const driver = new BunSQLiteDriver(dbPath); + + try { + driver.set_child( + "0:1", + { + type: CrdtType.TEXT, + parentId: "root", + parentKey: "text", + data: [["Hello"]], + version: 0, + }, + true + ); + + driver.append_live_text_history({ + nodeId: "0:1", + baseVersion: 0, + version: 1, + opId: "op:1", + ops: [{ type: "insert", index: 5, text: "!" }], + }); + driver.append_live_text_history({ + nodeId: "0:1", + baseVersion: 1, + version: 2, + opId: "op:2", + ops: [{ type: "delete", index: 0, length: 1 }], + }); + + expect(driver.get_live_text_history_by_op_id("0:1", "op:1")).toEqual({ + nodeId: "0:1", + baseVersion: 0, + version: 1, + opId: "op:1", + ops: [{ type: "insert", index: 5, text: "!" }], + }); + expect(driver.get_live_text_history_since("0:1", 0)).toHaveLength(2); + + driver.purge_live_text_history_before("0:1", 2); + expect(driver.get_live_text_history_since("0:1", 0)).toEqual([ + { + nodeId: "0:1", + baseVersion: 1, + version: 2, + opId: "op:2", + ops: [{ type: "delete", index: 0, length: 1 }], + }, + ]); + + driver.delete_node("0:1"); + expect(driver.get_live_text_history_since("0:1", 0)).toEqual([]); + } finally { + driver.close(); + } + }); +});