add MessageType and Updates types

packages/chain-ethereum/src/eip712/Secp256k1DelegateSigner.ts

@@ -11,7 +11,7 @@ import {
 
 import { AbiCoder } from "ethers/abi"
 
-import type { Action, Message, Session, Snapshot, Signature, SignatureScheme, Signer } from "@canvas-js/interfaces"
+import type { Message, Signature, SignatureScheme, Signer, MessageType } from "@canvas-js/interfaces"
 import { decodeURI, encodeURI } from "@canvas-js/signatures"
 import { assert, prepareMessage, signalInvalidType } from "@canvas-js/utils"
 
@@ -28,7 +28,7 @@ export const codecs = {
  * - canvas-action-eip712
  * - canvas-session-eip712
  */
-export class Secp256k1DelegateSigner implements Signer<Action | Session<Eip712SessionData> | Snapshot> {
+export class Secp256k1DelegateSigner implements Signer<MessageType<Eip712SessionData>> {
 	public static eip712ActionTypes = {
 		Message: [
 			{ name: "topic", type: "string" },
@@ -63,7 +63,7 @@ export class Secp256k1DelegateSigner implements Signer<Action | Session<Eip712Se
 		AuthorizationData: [{ name: "signature", type: "bytes" }],
 	} satisfies Record<string, TypedDataField[]>
 
-	public readonly scheme: SignatureScheme<Action | Session<Eip712SessionData> | Snapshot> = Secp256k1SignatureScheme
+	public readonly scheme: SignatureScheme<MessageType<Eip712SessionData>> = Secp256k1SignatureScheme
 	public readonly publicKey: string
 
 	readonly #wallet: BaseWallet
@@ -80,7 +80,7 @@ export class Secp256k1DelegateSigner implements Signer<Action | Session<Eip712Se
 		this.publicKey = encodeURI(Secp256k1SignatureScheme.type, publicKey)
 	}
 
-	public async sign(message: Message<Action | Session<Eip712SessionData> | Snapshot>): Promise<Signature> {
+	public async sign(message: Message<MessageType<Eip712SessionData>>): Promise<Signature> {
 		const { topic, clock, parents, payload } = prepareMessage(message)
 
 		if (payload.type === "action") {
@@ -128,6 +128,8 @@ export class Secp256k1DelegateSigner implements Signer<Action | Session<Eip712Se
 		} else if (payload.type === "snapshot") {
 			throw new Error("snapshots not supported for Secp256k1DelegateSigner")
 			// snapshots must be in an EVM-friendly onchain data format before we can support them here
+		} else if (payload.type === "updates") {
+			throw new Error("updates not supported for Secp256k1DelegateSigner")
 		} else {
 			signalInvalidType(payload)
 		}
@@ -182,10 +184,10 @@ function getAbiTypeForValue(value: any) {
 	throw new TypeError(`invalid type ${typeof value}: ${JSON.stringify(value)}`)
 }
 
-export const Secp256k1SignatureScheme: SignatureScheme<Action | Session<Eip712SessionData> | Snapshot> = {
+export const Secp256k1SignatureScheme: SignatureScheme<MessageType<Eip712SessionData>> = {
 	type: "secp256k1",
 	codecs: [codecs.action, codecs.session],
-	verify(signature: Signature, message: Message<Action | Session<Eip712SessionData> | Snapshot>) {
+	verify(signature: Signature, message: Message<MessageType<Eip712SessionData>>) {
 		const { type, publicKey } = decodeURI(signature.publicKey)
 		assert(type === Secp256k1SignatureScheme.type)
 
@@ -244,6 +246,8 @@ export const Secp256k1SignatureScheme: SignatureScheme<Action | Session<Eip712Se
 			assert(recoveredAddress === sessionAddress, "invalid EIP-712 session signature")
 		} else if (payload.type === "snapshot") {
 			throw new Error("snapshots not supported for Secp256k1DelegateSigner")
+		} else if (payload.type === "updates") {
+			throw new Error("updates not supported for Secp256k1DelegateSigner")
 		} else {
 			signalInvalidType(payload)
 		}

packages/chain-ethereum/src/siwf/SIWFSigner.ts

@@ -2,7 +2,15 @@ import { verifyMessage, hexlify, getBytes } from "ethers"
 import * as siwe from "siwe"
 import * as json from "@ipld/dag-json"
 
-import type { Action, Session, Snapshot, AbstractSessionData, DidIdentifier, Signer } from "@canvas-js/interfaces"
+import type {
+	Action,
+	Session,
+	Snapshot,
+	AbstractSessionData,
+	DidIdentifier,
+	Signer,
+	MessageType,
+} from "@canvas-js/interfaces"
 import { AbstractSessionSigner, ed25519 } from "@canvas-js/signatures"
 import { assert, DAYS } from "@canvas-js/utils"
 
@@ -75,7 +83,7 @@ export class SIWFSigner extends AbstractSessionSigner<SIWFSessionData> {
 		authorizationData: SIWFSessionData,
 		timestamp: number,
 		privateKey: Uint8Array,
-	): Promise<{ payload: Session<SIWFSessionData>; signer: Signer<Action | Session<SIWFSessionData> | Snapshot> }> {
+	): Promise<{ payload: Session<SIWFSessionData>; signer: Signer<MessageType<SIWFSessionData>> }> {
 		const signer = this.scheme.create({ type: ed25519.type, privateKey })
 		const did = await this.getDid()
 

packages/cli/src/commands/import.ts

@@ -5,7 +5,7 @@ import type { Argv } from "yargs"
 import chalk from "chalk"
 import * as json from "@ipld/dag-json"
 
-import type { Action, Message, Session, Signature, Snapshot } from "@canvas-js/interfaces"
+import type { Action, Message, MessageType, Session, Signature, Snapshot } from "@canvas-js/interfaces"
 import { Canvas } from "@canvas-js/core"
 
 import { getContractLocation } from "../utils.js"
@@ -52,7 +52,7 @@ export async function handler(args: Args) {
 		const { id, signature, message } = json.parse<{
 			id: string
 			signature: Signature
-			message: Message<Action | Session | Snapshot>
+			message: Message<MessageType>
 		}>(line)
 
 		try {

packages/core/src/Canvas.ts

@@ -4,13 +4,23 @@ import { logger } from "@libp2p/logger"
 import type pg from "pg"
 import type { SqlStorage } from "@cloudflare/workers-types"
 
-import { Signature, Action, Session, Message, Snapshot, SessionSigner, SignerCache } from "@canvas-js/interfaces"
+import {
+	Signature,
+	Action,
+	Session,
+	Message,
+	MessageType,
+	Snapshot,
+	SessionSigner,
+	SignerCache,
+	Updates,
+} from "@canvas-js/interfaces"
 import { AbstractModelDB, Model, ModelSchema, Effect } from "@canvas-js/modeldb"
 import { SIWESigner } from "@canvas-js/chain-ethereum"
 import { AbstractGossipLog, GossipLogEvents, SignedMessage } from "@canvas-js/gossiplog"
 import type { ServiceMap, NetworkConfig } from "@canvas-js/gossiplog/libp2p"
 
-import { assert, mapValues } from "@canvas-js/utils"
+import { assert, mapValues, signalInvalidType } from "@canvas-js/utils"
 
 import target from "#target"
 
@@ -49,14 +59,14 @@ export type ActionResult<Result = any> = { id: string; signature: Signature; mes
 
 export type ActionAPI<Args extends Array<any> = any, Result = any> = (...args: Args) => Promise<ActionResult<Result>>
 
-export interface CanvasEvents extends GossipLogEvents<Action | Session | Snapshot> {
+export interface CanvasEvents extends GossipLogEvents<MessageType> {
 	stop: Event
 }
 
 export type CanvasLogEvent = CustomEvent<{
 	id: string
 	signature: unknown
-	message: Message<Action | Session | Snapshot>
+	message: Message<MessageType>
 }>
 
 export type ApplicationData = {
@@ -88,7 +98,7 @@ export class Canvas<
 
 		const signers = new SignerCache(initSigners.length === 0 ? [new SIWESigner()] : initSigners)
 
-		const verifySignature = (signature: Signature, message: Message<Action | Session | Snapshot>) => {
+		const verifySignature = (signature: Signature, message: Message<MessageType>) => {
 			const signer = signers.getAll().find((signer) => signer.scheme.codecs.includes(signature.codec))
 			assert(signer !== undefined, "no matching signer found")
 			return signer.scheme.verify(signature, message)
@@ -149,9 +159,9 @@ export class Canvas<
 			let resultCount: number
 			let start: string | undefined = undefined
 			do {
-				const results: { id: string; message: Message<Action | Session> }[] = await db.query<{
+				const results: { id: string; message: Message<Action | Session | Updates> }[] = await db.query<{
 					id: string
-					message: Message<Action | Session>
+					message: Message<Action | Session | Updates>
 				}>("$messages", {
 					limit,
 					select: { id: true, message: true },
@@ -176,6 +186,10 @@ export class Canvas<
 						app.log("indexing user %s (did: %s)", publicKey, did)
 						const record = { did }
 						effects.push({ operation: "set", model: "$dids", value: record })
+					} else if (message.payload.type === "updates") {
+						// TODO: handle updates
+					} else {
+						signalInvalidType(message.payload)
 					}
 					start = id
 				}
@@ -216,7 +230,7 @@ export class Canvas<
 
 	private constructor(
 		public readonly signers: SignerCache,
-		public readonly messageLog: AbstractGossipLog<Action | Session | Snapshot>,
+		public readonly messageLog: AbstractGossipLog<MessageType | Updates>,
 		private readonly runtime: Runtime,
 	) {
 		super()
@@ -302,7 +316,7 @@ export class Canvas<
 		await target.listen(this, port, options)
 	}
 
-	public async startLibp2p(config: NetworkConfig): Promise<Libp2p<ServiceMap<Action | Session | Snapshot>>> {
+	public async startLibp2p(config: NetworkConfig): Promise<Libp2p<ServiceMap<MessageType>>> {
 		this.networkConfig = config
 		return await this.messageLog.startLibp2p(config)
 	}
@@ -382,23 +396,23 @@ export class Canvas<
 	 * Low-level utility method for internal and debugging use.
 	 * The normal way to apply actions is to use the `Canvas.actions[name](...)` functions.
 	 */
-	public async insert(signature: Signature, message: Message<Session | Action | Snapshot>): Promise<{ id: string }> {
+	public async insert(signature: Signature, message: Message<MessageType>): Promise<{ id: string }> {
 		assert(message.topic === this.topic, "invalid message topic")
 
 		const signedMessage = this.messageLog.encode(signature, message)
 		await this.messageLog.insert(signedMessage)
 		return { id: signedMessage.id }
 	}
 
-	public async getMessage(id: string): Promise<SignedMessage<Action | Session | Snapshot> | null> {
+	public async getMessage(id: string): Promise<SignedMessage<MessageType> | null> {
 		return await this.messageLog.get(id)
 	}
 
 	public async *getMessages(
 		lowerBound: { id: string; inclusive: boolean } | null = null,
 		upperBound: { id: string; inclusive: boolean } | null = null,
 		options: { reverse?: boolean } = {},
-	): AsyncIterable<SignedMessage<Action | Session | Snapshot>> {
+	): AsyncIterable<SignedMessage<MessageType>> {
 		const range: { lt?: string; lte?: string; gt?: string; gte?: string; reverse?: boolean; limit?: number } = {}
 		if (lowerBound) {
 			if (lowerBound.inclusive) range.gte = lowerBound.id

packages/core/src/ExecutionContext.ts

@@ -2,7 +2,7 @@ import * as cbor from "@ipld/dag-cbor"
 import { blake3 } from "@noble/hashes/blake3"
 import { bytesToHex } from "@noble/hashes/utils"
 
-import type { Action, Session, Snapshot } from "@canvas-js/interfaces"
+import type { Action, MessageType } from "@canvas-js/interfaces"
 
 import { ModelValue, PropertyValue, validateModelValue, updateModelValues, mergeModelValues } from "@canvas-js/modeldb"
 import { AbstractGossipLog, SignedMessage, MessageId } from "@canvas-js/gossiplog"
@@ -21,7 +21,8 @@ export class ExecutionContext {
 	public readonly root: MessageId[]
 
 	constructor(
-		public readonly messageLog: AbstractGossipLog<Action | Session | Snapshot>,
+		public readonly messageLog: AbstractGossipLog<MessageType>,
+		// TODO: why is this just action? should it be `SignedMessage<Action | Updates>`?
 		public readonly signedMessage: SignedMessage<Action>,
 		public readonly address: string,
 	) {

packages/core/src/runtime/AbstractRuntime.ts

@@ -1,14 +1,14 @@
 import * as cbor from "@ipld/dag-cbor"
 import { logger } from "@libp2p/logger"
 
-import type { Action, Session, Snapshot, SignerCache, Awaitable } from "@canvas-js/interfaces"
+import type { Action, Session, Snapshot, SignerCache, Awaitable, MessageType } from "@canvas-js/interfaces"
 
 import { AbstractModelDB, Effect, ModelSchema } from "@canvas-js/modeldb"
 import { GossipLogConsumer, MAX_MESSAGE_ID, AbstractGossipLog, SignedMessage } from "@canvas-js/gossiplog"
-import { assert } from "@canvas-js/utils"
+import { assert, signalInvalidType } from "@canvas-js/utils"
 
 import { ExecutionContext, getKeyHash } from "../ExecutionContext.js"
-import { isAction, isSession, isSnapshot } from "../utils.js"
+import { isAction, isSession, isSnapshot, isUpdates } from "../utils.js"
 
 export type EffectRecord = { key: string; value: Uint8Array | null; branch: number; clock: number }
 
@@ -95,28 +95,27 @@ export abstract class AbstractRuntime {
 		this.#db = db
 	}
 
-	public getConsumer(): GossipLogConsumer<Action | Session | Snapshot> {
+	public getConsumer(): GossipLogConsumer<MessageType> {
 		const handleSession = this.handleSession.bind(this)
 		const handleAction = this.handleAction.bind(this)
 		const handleSnapshot = this.handleSnapshot.bind(this)
 
-		return async function (this: AbstractGossipLog<Action | Session | Snapshot>, signedMessage) {
+		return async function (this: AbstractGossipLog<MessageType>, signedMessage) {
 			if (isSession(signedMessage)) {
 				return await handleSession(signedMessage)
 			} else if (isAction(signedMessage)) {
 				return await handleAction(signedMessage, this)
 			} else if (isSnapshot(signedMessage)) {
 				return await handleSnapshot(signedMessage, this)
+			} else if (isUpdates(signedMessage)) {
+				// TODO: handle updates
 			} else {
 				throw new Error("invalid message payload type")
 			}
 		}
 	}
 
-	private async handleSnapshot(
-		signedMessage: SignedMessage<Snapshot>,
-		messageLog: AbstractGossipLog<Action | Session | Snapshot>,
-	) {
+	private async handleSnapshot(signedMessage: SignedMessage<Snapshot>, messageLog: AbstractGossipLog<MessageType>) {
 		const { models, effects } = signedMessage.message.payload
 
 		const messages = await messageLog.getMessages()
@@ -167,10 +166,7 @@ export abstract class AbstractRuntime {
 		await this.db.apply(effects)
 	}
 
-	private async handleAction(
-		signedMessage: SignedMessage<Action>,
-		messageLog: AbstractGossipLog<Action | Session | Snapshot>,
-	) {
+	private async handleAction(signedMessage: SignedMessage<Action>, messageLog: AbstractGossipLog<MessageType>) {
 		const { id, signature, message } = signedMessage
 		const { did, name, context } = message.payload
 

packages/core/src/schema.ts

@@ -1,7 +1,7 @@
 import { fromDSL } from "@ipld/schema/from-dsl.js"
 import { create } from "@ipld/schema/typed.js"
 
-import type { Action, Session, Snapshot } from "@canvas-js/interfaces"
+import type { Action, MessageType, Session, Snapshot, Updates } from "@canvas-js/interfaces"
 
 const schema = `
 type ActionContext struct {
@@ -39,18 +39,35 @@ type Snapshot struct {
   effects   [SnapshotEffect]
 }
 
+type Update = {
+	model: string
+	key: string
+	diff: Uint8Array
+}
+
+type Updates = {
+	type: "updates"
+	updates: Update[]
+}
+
 type Payload union {
   | Action "action"
   | Session "session"
   | Snapshot "snapshot"
+  | Updates "updates"
 } representation inline {
   discriminantKey "type"
 }
 `
 
 const { toTyped } = create(fromDSL(schema), "Payload")
 
-export function validatePayload(payload: unknown): payload is Action | Session | Snapshot {
-	const result = toTyped(payload) as { Action: Omit<Action, "type"> } | { Session: Omit<Session, "type"> } | undefined
+export function validatePayload(payload: unknown): payload is MessageType {
+	const result = toTyped(payload) as
+		| { Action: Omit<Action, "type"> }
+		| { Session: Omit<Session, "type"> }
+		| { Snapshot: Omit<Snapshot, "type"> }
+		| { Updates: Omit<Updates, "type"> }
+		| undefined
 	return result !== undefined
 }

packages/core/src/targets/interface.ts

@@ -1,15 +1,15 @@
 import type pg from "pg"
 
-import type { Action, Session, Snapshot } from "@canvas-js/interfaces"
+import type { MessageType } from "@canvas-js/interfaces"
 import type { AbstractGossipLog, GossipLogInit } from "@canvas-js/gossiplog"
 import type { Canvas } from "@canvas-js/core"
 import type { SqlStorage } from "@cloudflare/workers-types"
 
 export interface PlatformTarget {
 	openGossipLog: (
 		location: { path: string | pg.ConnectionConfig | SqlStorage | null; topic: string; clear?: boolean },
-		init: GossipLogInit<Action | Session | Snapshot>,
-	) => Promise<AbstractGossipLog<Action | Session | Snapshot>>
+		init: GossipLogInit<MessageType>,
+	) => Promise<AbstractGossipLog<MessageType>>
 
 	listen: (app: Canvas, port: number, options?: { signal?: AbortSignal }) => Promise<void>
 }