packages/core - y.js in contracts

packages/core/package.json

@@ -76,6 +76,7 @@
 		"@types/pg": "^8.11.11",
 		"dotenv": "^16.4.7",
 		"nanoid": "^5.0.9",
-		"prando": "^6.0.1"
+		"prando": "^6.0.1",
+		"yjs": "^13.6.23"
 	}
 }

packages/core/src/ExecutionContext.ts

@@ -1,6 +1,7 @@
 import * as cbor from "@ipld/dag-cbor"
 import { blake3 } from "@noble/hashes/blake3"
 import { bytesToHex } from "@noble/hashes/utils"
+import * as Y from "yjs"
 
 import type { Action, Session, Snapshot } from "@canvas-js/interfaces"
 
@@ -10,6 +11,13 @@ import { assert, mapValues } from "@canvas-js/utils"
 
 export const getKeyHash = (key: string) => bytesToHex(blake3(key, { dkLen: 16 }))
 
+type ApplyDocumentUpdate = {
+	type: "applyDocumentUpdate"
+	update: Uint8Array
+}
+
+type Operation = ApplyDocumentUpdate
+
 export class ExecutionContext {
 	// // recordId -> { version, value }
 	// public readonly reads: Record<string, { version: string | null; value: ModelValue | null }> = {}
@@ -19,6 +27,7 @@ export class ExecutionContext {
 
 	public readonly modelEntries: Record<string, Record<string, ModelValue | null>>
 	public readonly root: MessageId[]
+	public readonly operations: Record<string, Record<string, Operation[]>> = {}
 
 	constructor(
 		public readonly messageLog: AbstractGossipLog<Action | Session | Snapshot>,

packages/core/src/runtime/AbstractRuntime.ts

@@ -9,6 +9,7 @@ import { assert } from "@canvas-js/utils"
 
 import { ExecutionContext, getKeyHash } from "../ExecutionContext.js"
 import { isAction, isSession, isSnapshot } from "../utils.js"
+import * as Y from "yjs"
 
 export type EffectRecord = { key: string; value: Uint8Array | null; branch: number; clock: number }
 
@@ -63,8 +64,32 @@ export abstract class AbstractRuntime {
 	} satisfies ModelSchema
 
 	protected static getModelSchema(schema: ModelSchema): ModelSchema {
+		const outputSchema: ModelSchema = {}
+		for (const [modelName, modelSchema] of Object.entries(schema)) {
+			// @ts-ignore
+			if (modelSchema.content === "yjs-doc") {
+				if (
+					Object.entries(modelSchema).length !== 2 &&
+					// @ts-ignore
+					modelSchema.id !== "primary"
+				) {
+					// not valid
+					throw new Error("yjs-doc tables must have two columns, one of which is 'id'")
+				} else {
+					// this table stores the current state of the Yjs document
+					// we just need one entry per document because updates are commutative
+					outputSchema[`${modelName}:state`] = {
+						id: "primary",
+						content: "bytes",
+					}
+				}
+			} else {
+				outputSchema[modelName] = modelSchema
+			}
+		}
+
 		return {
-			...schema,
+			...outputSchema,
 			...AbstractRuntime.sessionsModel,
 			...AbstractRuntime.actionsModel,
 			...AbstractRuntime.effectsModel,
@@ -212,6 +237,23 @@ export abstract class AbstractRuntime {
 		const actionRecord: ActionRecord = { message_id: id, did, name, timestamp: context.timestamp }
 		const effects: Effect[] = [{ operation: "set", model: "$actions", value: actionRecord }]
 
+		for (const [model, entries] of Object.entries(executionContext.operations)) {
+			for (const [key, operations] of Object.entries(entries)) {
+				// load the current doc from the database or create a new one if it doesn't exist yet
+				const currentState = (await executionContext.messageLog.getYDoc(model, key)) || new Y.Doc()
+
+				// apply the actual operations to the document
+				for (const operation of operations) {
+					if (operation.type === "applyDocumentUpdate") {
+						// apply document update to the doc here
+						Y.applyUpdate(currentState, operation.update)
+					}
+				}
+
+				await this.db.set(`${model}:state`, { id: key, content: Y.encodeStateAsUpdate(currentState) })
+			}
+		}
+
 		for (const [model, entries] of Object.entries(executionContext.modelEntries)) {
 			for (const [key, value] of Object.entries(entries)) {
 				const keyHash = getKeyHash(key)

packages/core/src/runtime/ContractRuntime.ts

@@ -134,6 +134,24 @@ export class ContractRuntime extends AbstractRuntime {
 					const key = vm.context.getString(keyHandle)
 					this.context.deleteModelValue(model, key)
 				}),
+
+				applyDocumentUpdate: vm.context.newFunction("applyDocumentUpdate", (modelHandle, keyHandle, updateHandle) => {
+					const model = vm.context.getString(modelHandle)
+					const key = vm.context.getString(keyHandle)
+
+					assert(this.#context !== null, "expected this.#context !== null")
+
+					if (this.#context.operations[model] === undefined) {
+						this.#context.operations[model] = {}
+					}
+					if (this.#context.operations[model][key] === undefined) {
+						this.#context.operations[model][key] = []
+					}
+					this.#context.operations[model][key].push({
+						type: "applyDocumentUpdate",
+						update: this.vm.unwrapValue(updateHandle) as Uint8Array,
+					})
+				}),
 			})
 			.consume(vm.cache)
 	}

packages/core/src/runtime/FunctionRuntime.ts

@@ -173,6 +173,20 @@ export class FunctionRuntime<ModelsT extends ModelSchema> extends AbstractRuntim
 					this.releaseLock()
 				}
 			},
+			applyDocumentUpdate: async (model: string, key: string, update: Uint8Array) => {
+				assert(this.#context !== null, "expected this.#context !== null")
+
+				if (this.#context.operations[model] === undefined) {
+					this.#context.operations[model] = {}
+				}
+				if (this.#context.operations[model][key] === undefined) {
+					this.#context.operations[model][key] = []
+				}
+				this.#context.operations[model][key].push({
+					type: "applyDocumentUpdate",
+					update,
+				})
+			},
 		}
 	}
 

packages/core/src/types.ts

@@ -17,14 +17,18 @@ export type ActionImplementation<
 	ModelsT extends ModelSchema = ModelSchema,
 	Args extends Array<any> = any,
 	Result = any,
-> = (this: ActionContext<DeriveModelTypes<ModelsT>>, db: ModelAPI<DeriveModelTypes<ModelsT>>, ...args: Args) => Awaitable<Result>
+> = (
+	this: ActionContext<DeriveModelTypes<ModelsT>>,
+	db: ModelAPI<DeriveModelTypes<ModelsT>>,
+	...args: Args
+) => Awaitable<Result>
 
 export type Chainable<ModelTypes extends Record<string, ModelValue>> = Promise<void> & {
 	link: <T extends keyof ModelTypes & string>(
 		model: T,
 		primaryKey: string,
 		through?: { through: string },
-	) => Promise<void>,
+	) => Promise<void>
 	unlink: <T extends keyof ModelTypes & string>(
 		model: T,
 		primaryKey: string,
@@ -40,6 +44,7 @@ export type ModelAPI<ModelTypes extends Record<string, ModelValue>> = {
 	update: <T extends keyof ModelTypes & string>(model: T, value: Partial<ModelTypes[T]>) => Chainable<ModelTypes>
 	merge: <T extends keyof ModelTypes & string>(model: T, value: Partial<ModelTypes[T]>) => Chainable<ModelTypes>
 	delete: <T extends keyof ModelTypes & string>(model: T, key: string) => Promise<void>
+	applyDocumentUpdate: <T extends keyof ModelTypes & string>(model: T, key: string, update: Uint8Array) => Promise<void>
 }
 
 export type ActionContext<T extends Record<string, ModelValue>> = {

packages/core/test/canvas.test.ts

@@ -7,8 +7,16 @@ import type { Action, Message, Session } from "@canvas-js/interfaces"
 import { ed25519 } from "@canvas-js/signatures"
 import { SIWESigner, Eip712Signer } from "@canvas-js/chain-ethereum"
 import { CosmosSigner } from "@canvas-js/chain-cosmos"
-import { Canvas } from "@canvas-js/core"
+import { Canvas, Config } from "@canvas-js/core"
 import { assert } from "@canvas-js/utils"
+import * as Y from "yjs"
+
+function createUpdate(baseDoc: Y.Doc, update: (startDoc: Y.Doc) => void) {
+	const state0 = Y.encodeStateAsUpdate(baseDoc)
+	update(baseDoc)
+	const state1 = Y.encodeStateAsUpdate(baseDoc)
+	return Y.diffUpdate(state1, Y.encodeStateVectorFromUpdate(state0))
+}
 
 const contract = `
 export const models = {
@@ -499,3 +507,101 @@ test("open custom modeldb tables", async (t) => {
 	await app.db.set("widgets", { id, name: "foobar" })
 	t.deepEqual(await app.db.get("widgets", id), { id, name: "foobar" })
 })
+
+test("create a contract with a yjs text table with ContractRuntime", async (t) => {
+	await yjsTest(t, {
+		contract: `
+			export const models = {
+				posts: {
+					id: "primary",
+					content: "yjs-doc"
+				}
+			};
+			export const actions = {
+				updatePost: (db, { key, update }) => {
+					db.applyDocumentUpdate("posts", key, update)
+				}
+			};`,
+		topic: "com.example.app",
+	})
+})
+
+test("create a contract with a yjs text table with FunctionRuntime", async (t) => {
+	await yjsTest(t, {
+		contract: {
+			models: {
+				// @ts-ignore
+				posts: { id: "primary", content: "yjs-doc" },
+			},
+			actions: {
+				updatePost: (db, { key, update }: { key: string; update: Uint8Array }) => {
+					db.applyDocumentUpdate("posts", key, update)
+				},
+			},
+		},
+		topic: "com.example.app",
+	})
+})
+
+async function yjsTest(t: ExecutionContext, config: Config) {
+	// initialize two apps with the same config
+	const appA = await Canvas.initialize(config)
+	const appB = await Canvas.initialize(config)
+
+	t.teardown(() => {
+		appA.stop()
+		appB.stop()
+	})
+
+	const post1Id = "post_1"
+
+	// call an action that updates the yjs-doc item
+	const doc0A = new Y.Doc()
+
+	await appA.actions.updatePost({
+		key: post1Id,
+		update: createUpdate(doc0A, (doc) => doc.getText().insert(0, "hello")),
+	})
+
+	const doc1A = await appA.messageLog.getYDoc("posts", post1Id)
+	t.is(doc1A!.getText().toJSON(), "hello", "assert doc has been updated")
+
+	// call another action that updates the yjs-doc item
+	const update1 = await appA.actions.updatePost({
+		key: post1Id,
+		update: createUpdate(doc1A!, (doc) => doc.getText().insert(5, " world")),
+	})
+
+	const doc2A = await appA.messageLog.getYDoc("posts", post1Id)
+	t.is(doc2A!.getText().toJSON(), "hello world", "assert doc has been updated")
+
+	// sync app -> app2
+	await appA.messageLog.serve((source) => appB.messageLog.sync(source))
+
+	const doc2B = await appB.messageLog.getYDoc("posts", post1Id)
+	t.is(doc2B!.getText().toJSON(), "hello world", "assert doc on peer has been updated")
+
+	// apply an action on app
+	await appA.actions.updatePost({
+		key: post1Id,
+		update: createUpdate(doc2A!, (doc) => doc.getText().insert(11, "!")),
+	})
+	const doc3A = await appA.messageLog.getYDoc("posts", post1Id)
+	t.is(doc3A!.getText().toJSON(), "hello world!", "assert doc has been updated")
+
+	// concurrently apply an action on app2
+	await appB.actions.updatePost({
+		key: post1Id,
+		update: createUpdate(doc2B!, (doc) => doc.getText().insert(11, "?")),
+	})
+	const doc3B = await appB.messageLog.getYDoc("posts", post1Id)
+	t.is(doc3B!.getText().toJSON(), "hello world?", "assert doc on peer has been updated")
+
+	await appA.messageLog.serve((source) => appB.messageLog.sync(source))
+	await appB.messageLog.serve((source) => appA.messageLog.sync(source))
+
+	// assert that app contains both changes
+	const doc4A = await appA.messageLog.getYDoc("posts", post1Id)
+	const doc4B = await appB.messageLog.getYDoc("posts", post1Id)
+	t.is(doc4A!.getText().toJSON(), doc4B!.getText().toJSON(), "assert concurrent changes have converged")
+}

packages/gossiplog/package.json

@@ -102,7 +102,8 @@
 		"protons-runtime": "^5.5.0",
 		"uint8arraylist": "^2.4.8",
 		"uint8arrays": "^5.1.0",
-		"ws": "^8.18.0"
+		"ws": "^8.18.0",
+		"yjs": "^13.6.23"
 	},
 	"devDependencies": {
 		"@cloudflare/workers-types": "^4.20240925.0",

packages/gossiplog/src/AbstractGossipLog.ts

@@ -21,6 +21,7 @@ import { MessageSource, SignedMessage } from "./SignedMessage.js"
 import { decodeId, encodeId, MessageId, messageIdPattern } from "./MessageId.js"
 import { getNextClock } from "./schema.js"
 import { gossiplogTopicPattern } from "./utils.js"
+import * as Y from "yjs"
 
 export type GossipLogConsumer<Payload = unknown, Result = any> = (
 	this: AbstractGossipLog<Payload, Result>,
@@ -197,6 +198,20 @@ export abstract class AbstractGossipLog<Payload = unknown, Result = any> extends
 		return this.encode(signature, message, { branch })
 	}
 
+	public async getYDoc(modelName: string, id: string): Promise<Y.Doc | null> {
+		const existingStateEntries = await this.db.query<{ id: string; content: Uint8Array }>(`${modelName}:state`, {
+			where: { id: id },
+			limit: 1,
+		})
+		if (existingStateEntries.length > 0) {
+			const doc = new Y.Doc()
+			Y.applyUpdate(doc, existingStateEntries[0].content)
+			return doc
+		} else {
+			return null
+		}
+	}
+
 	public getMessages(
 		range: { lt?: string; lte?: string; gt?: string; gte?: string; reverse?: boolean; limit?: number } = {},
 	): Promise<{ id: string; signature: Signature; message: Message<Payload>; branch: number }[]> {