diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 9c4dfb01..d3fe565b 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -99,11 +99,13 @@ const station = await CultMesh.startNode(statePath, { const stock = station.document(stationStockUiDocument, "station:starbridge:stock"); const current = await stock.latest(); -await stock.set(updatedStockFromUi); +await stock.authoritativeWriter().write(updatedStockFromUi); -const reactive = station.reactiveDocument(stationStockUiDocument, "station:starbridge:stock"); +const reactive = stock.authoritativeWriter().reactive(); await reactive.ready; -reactive.current.availableMissiles -= 4; +reactive.update(draft => { + draft.availableMissiles -= 4; +}); ``` State pointers are the same kind of managed surface for UI and tools. They can @@ -418,7 +420,8 @@ Use `node.syncDocumentFromPeerSnapshot(...)` or local node from the remote snapshot. The helper requests the raw snapshot, applies it through the node's document registry, and returns the requested typed definition. Same-schema aliases stay local after the call: the caller can -keep using `node.document(aliasDefinition, key)` or `node.reactiveDocument(...)` +keep using `node.document(aliasDefinition, key)` and explicitly select +`authoritativeWriter()`, `predictionWriter()`, or `observe()` without re-threading the RUDP peer. The same typed-definition overload is available for @@ -432,7 +435,7 @@ sync came from local persistence or a remote CultNet snapshot. Use `node.syncDocumentFromPublication(...)` or `CultMesh.syncDocumentFromPublication(...)` for the same source-agnostic shape when the publication should hydrate a local node before the caller continues -through `node.document(...)` or `node.reactiveDocument(...)`. +through `node.document(...)` and its explicit writer or observer. ## Streaming Mode diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index aa84f5f9..5d81425e 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -232,6 +232,7 @@ export interface CultMeshReactiveDocumentOptions { readonly context?: CultMeshQueryContext | string; readonly watch?: boolean; readonly debounceMs?: number; + readonly replaceDirtyCurrentOnCanonicalSnapshot?: boolean; readonly onError?: (error: unknown) => void; } @@ -248,15 +249,30 @@ export interface CultMeshReactiveDocumentReconciliation { - readonly current: TDocument; + readonly writer: CultMeshDocumentWriter; + readonly snapshot: TDocument; + readonly dirty: boolean; readonly reconciliation: CultMeshReactiveDocumentReconciliation | undefined; readonly ready: Promise; readonly disposed: boolean; + update(update: (draft: TDocument) => void | TDocument): TDocument; + replaceLocal(value: TDocument): TDocument; + flush(): Promise; clearReconciliation(): void; refresh(): Promise; dispose(): void; } +export interface CultMeshObservedDocument { + readonly current: TDocument; + readonly ready: Promise; + readonly disposed: boolean; + refresh(): Promise; + dispose(): void; +} + +export type CultMeshDocumentWriteKind = "authoritative-replacement" | "prediction"; + export interface CultMeshNativeSliceColumn { readonly name: string; readonly valueType: string; @@ -842,10 +858,6 @@ export class CultMeshDocumentHandle { return this.submitPredictionDocument !== undefined; } - public get canSet(): boolean { - return this.canSubmitPrediction || this.canReplace; - } - public latest(context: CultMeshQueryContext | string = "local"): Promise { return this.snapshotDocument( this.resolveContext(typeof context === "string" ? cultMeshQueryContext(context) : context), @@ -932,63 +944,44 @@ export class CultMeshDocumentHandle { return this.submitPredictionDocument(this.resolveContext(context), value); } - public set(value: TDocument): Promise; - public set(context: CultMeshQueryContext | string, value: TDocument): Promise; - public set( - contextOrValue: CultMeshQueryContext | string | TDocument, - maybeValue?: TDocument, - ): Promise { - const hasContext = - typeof contextOrValue === "string" || isCultMeshQueryContext(contextOrValue); - const context = hasContext - ? typeof contextOrValue === "string" - ? cultMeshQueryContext(contextOrValue) - : contextOrValue as CultMeshQueryContext - : cultMeshQueryContext("local"); - const value = hasContext ? maybeValue : contextOrValue as TDocument; - if (value === undefined) { - throw new Error(`Document '${this.documentId}' requires a value.`); + public authoritativeWriter( + context: CultMeshQueryContext | string = "local", + ): CultMeshDocumentWriter { + if (!this.replaceDocument) { + throw new Error(`Document '${this.documentId}' does not support authoritative replacement.`); } - - const resolved = this.resolveContext(context); - if (this.submitPredictionDocument) { - return this.submitPredictionDocument(resolved, value); - } - if (this.replaceDocument) { - return this.replaceDocument(resolved, value); - } - - throw new Error(`Document '${this.documentId}' does not support mutation.`); + const resolved = this.resolveContext( + typeof context === "string" ? cultMeshQueryContext(context) : context, + ); + return new CultMeshDocumentWriter( + this as unknown as CultMeshDocumentHandle, + "authoritative-replacement", + resolved, + value => this.replaceDocument!(resolved, value as unknown as TDocument), + ); } - public update(update: CultMeshDocumentUpdater): Promise; - public update( - context: CultMeshQueryContext | string, - update: CultMeshDocumentUpdater, - ): Promise; - public async update( - contextOrUpdate: CultMeshQueryContext | string | CultMeshDocumentUpdater, - maybeUpdate?: CultMeshDocumentUpdater, - ): Promise { - const hasContext = - typeof contextOrUpdate === "string" || isCultMeshQueryContext(contextOrUpdate); - const context = hasContext - ? contextOrUpdate as CultMeshQueryContext | string - : "local"; - const update = hasContext ? maybeUpdate : contextOrUpdate as CultMeshDocumentUpdater; - if (!update) { - throw new Error(`Document '${this.documentId}' requires an update function.`); + public predictionWriter( + context: CultMeshQueryContext | string = "local", + ): CultMeshDocumentWriter { + if (!this.submitPredictionDocument) { + throw new Error(`Document '${this.documentId}' does not support prediction submission.`); } - - const next = await update(await this.latest(context)); - await this.set(context, next); - return next; + const resolved = this.resolveContext( + typeof context === "string" ? cultMeshQueryContext(context) : context, + ); + return new CultMeshDocumentWriter( + this as unknown as CultMeshDocumentHandle, + "prediction", + resolved, + value => this.submitPredictionDocument!(resolved, value as unknown as TDocument), + ); } - public reactive( + public observe( options: CultMeshReactiveDocumentOptions = {}, - ): CultMeshReactiveDocument { - return new CultMeshReactiveDocumentHandle( + ): CultMeshObservedDocument { + return new CultMeshObservedDocumentHandle( this as unknown as CultMeshDocumentHandle, options, ); @@ -1041,22 +1034,40 @@ export class CultMeshDocumentHandle { } } -class CultMeshReactiveDocumentHandle - implements CultMeshReactiveDocument { +export class CultMeshDocumentWriter { + public constructor( + public readonly document: CultMeshDocumentHandle, + public readonly kind: CultMeshDocumentWriteKind, + public readonly context: CultMeshQueryContext, + private readonly writeDocument: (value: TDocument) => Promise, + ) {} + + public write(value: TDocument): Promise { + return this.writeDocument(cloneCultMeshDocument(value)); + } + + public async update(update: CultMeshDocumentUpdater): Promise { + const next = await update(cloneCultMeshDocument(await this.document.latest(this.context))); + await this.write(next); + return next; + } + + public reactive( + options: Omit = {}, + ): CultMeshReactiveDocument { + return new CultMeshReactiveDocumentHandle(this, options); + } +} + +class CultMeshObservedDocumentHandle + implements CultMeshObservedDocument { readonly #document: CultMeshDocumentHandle; readonly #context: CultMeshQueryContext | string; readonly #onError: (error: unknown) => void; - readonly #debounceMs: number; - #current: TDocument; + #canonicalVersion = 0; + #current = {} as TDocument; #disposed = false; - #applyingRemote = false; - #pendingSet: Promise = Promise.resolve(); - #flushScheduled = false; - #flushTimer: ReturnType | undefined; #unsubscribe: CultMeshUnsubscribe | undefined; - #lastPredicted: TDocument | undefined; - #reconciliation: CultMeshReactiveDocumentReconciliation | undefined; - #reconciliationVersion = 0; public readonly ready: Promise; @@ -1066,19 +1077,17 @@ class CultMeshReactiveDocumentHandle ) { this.#document = document; this.#context = options.context ?? "local"; - this.#debounceMs = options.debounceMs ?? 0; this.#onError = options.onError ?? ((error) => { queueMicrotask(() => { throw error; }); }); - this.#current = this.createProxy({} as TDocument); this.ready = this.refresh(); - if (options.watch !== false) { try { this.#unsubscribe = document.watch(this.#context, value => { - this.applySnapshot(value); + this.#current = cloneCultMeshDocument(value); + this.#canonicalVersion++; }); } catch (error) { this.#onError(error); @@ -1087,9 +1096,100 @@ class CultMeshReactiveDocumentHandle } public get current(): TDocument { + this.throwIfDisposed(); return this.#current; } + public get disposed(): boolean { + return this.#disposed; + } + + public async refresh(): Promise { + this.throwIfDisposed(); + const versionBeforeRead = this.#canonicalVersion; + const latest = cloneCultMeshDocument(await this.#document.latest(this.#context)); + if (versionBeforeRead === this.#canonicalVersion) { + this.#current = latest; + this.#canonicalVersion++; + } + return this.#current; + } + + public dispose(): void { + if (this.#disposed) return; + this.#disposed = true; + this.#unsubscribe?.(); + this.#unsubscribe = undefined; + } + + private throwIfDisposed(): void { + if (this.#disposed) { + throw new Error(`Observed document '${this.#document.documentId}' is disposed.`); + } + } +} + +class CultMeshReactiveDocumentHandle + implements CultMeshReactiveDocument { + readonly #writer: CultMeshDocumentWriter; + readonly #document: CultMeshDocumentHandle; + readonly #onError: (error: unknown) => void; + readonly #debounceMs: number; + readonly #replaceDirtyCurrentOnCanonicalSnapshot: boolean; + #current = {} as TDocument; + #disposed = false; + #pendingSet: Promise = Promise.resolve(); + #flushScheduled = false; + #flushTimer: ReturnType | undefined; + #unsubscribe: CultMeshUnsubscribe | undefined; + #dirty = false; + #lastPredicted: TDocument | undefined; + #reconciliation: CultMeshReactiveDocumentReconciliation | undefined; + #reconciliationVersion = 0; + #canonicalVersion = 0; + + public readonly ready: Promise; + + public constructor( + writer: CultMeshDocumentWriter, + options: Omit = {}, + ) { + this.#writer = writer; + this.#document = writer.document; + this.#debounceMs = options.debounceMs ?? 0; + this.#replaceDirtyCurrentOnCanonicalSnapshot = + options.replaceDirtyCurrentOnCanonicalSnapshot ?? false; + this.#onError = options.onError ?? ((error) => { + queueMicrotask(() => { + throw error; + }); + }); + this.ready = this.refresh(); + + if (options.watch !== false) { + try { + this.#unsubscribe = this.#document.watch(this.#writer.context, value => { + this.applySnapshot(value); + }); + } catch (error) { + this.#onError(error); + } + } + } + + public get writer(): CultMeshDocumentWriter { + return this.#writer; + } + + public get snapshot(): TDocument { + this.throwIfDisposed(); + return cloneCultMeshDocument(this.#current); + } + + public get dirty(): boolean { + return this.#dirty; + } + public get reconciliation(): CultMeshReactiveDocumentReconciliation | undefined { return this.#reconciliation; } @@ -1099,9 +1199,37 @@ class CultMeshReactiveDocumentHandle } public async refresh(): Promise { - const latest = await this.#document.latest(this.#context); - this.applySnapshot(latest); - return this.#current; + this.throwIfDisposed(); + const versionBeforeRead = this.#canonicalVersion; + const latest = await this.#document.latest(this.#writer.context); + if (versionBeforeRead === this.#canonicalVersion) { + this.applySnapshot(latest); + } + return this.snapshot; + } + + public update(update: (draft: TDocument) => void | TDocument): TDocument { + this.throwIfDisposed(); + const draft = cloneCultMeshDocument(this.#current); + const returned = update(draft); + this.#current = cloneCultMeshDocument(returned ?? draft); + this.#dirty = true; + this.queueSet(); + return this.snapshot; + } + + public replaceLocal(value: TDocument): TDocument { + this.throwIfDisposed(); + this.#current = cloneCultMeshDocument(value); + this.#dirty = true; + this.queueSet(); + return this.snapshot; + } + + public async flush(): Promise { + this.throwIfDisposed(); + this.flushNow(); + await this.#pendingSet; } public clearReconciliation(): void { @@ -1121,18 +1249,13 @@ class CultMeshReactiveDocumentHandle private applySnapshot(value: TDocument): void { const canonical = cloneCultMeshDocument(value); + this.#canonicalVersion++; this.captureReconciliation(canonical); - this.#applyingRemote = true; - try { - for (const key of Reflect.ownKeys(this.#current)) { - if (!Reflect.has(canonical, key)) { - Reflect.deleteProperty(this.#current, key); - } - } - Object.assign(this.#current, canonical); - } finally { - this.#applyingRemote = false; + if (this.#dirty && !this.#replaceDirtyCurrentOnCanonicalSnapshot) { + return; } + this.#current = canonical; + this.#dirty = false; } private captureReconciliation(canonical: TDocument): void { @@ -1158,27 +1281,6 @@ class CultMeshReactiveDocumentHandle this.#lastPredicted = undefined; } - private createProxy(seed: TDocument): TDocument { - return new Proxy(seed, { - set: (target, property, value) => { - const changed = !Object.is(Reflect.get(target, property), value); - const ok = Reflect.set(target, property, value); - if (ok && changed && !this.#applyingRemote) { - this.queueSet(); - } - return ok; - }, - deleteProperty: (target, property) => { - const hadProperty = Reflect.has(target, property); - const ok = Reflect.deleteProperty(target, property); - if (ok && hadProperty && !this.#applyingRemote) { - this.queueSet(); - } - return ok; - }, - }); - } - private queueSet(): void { if (this.#disposed) { return; @@ -1188,20 +1290,7 @@ class CultMeshReactiveDocumentHandle } this.#flushScheduled = true; - const flush = () => { - this.#flushScheduled = false; - this.#flushTimer = undefined; - if (this.#disposed) { - return; - } - - const snapshot = cloneCultMeshDocument(this.#current); - this.#lastPredicted = snapshot; - this.#pendingSet = this.#pendingSet - .catch(() => undefined) - .then(() => this.#document.set(this.#context, snapshot)) - .catch(error => this.#onError(error)); - }; + const flush = () => this.flushNow(); const debounceMs = this.#debounceMs; if (debounceMs > 0) { @@ -1219,6 +1308,29 @@ class CultMeshReactiveDocumentHandle this.#flushTimer = setTimeout(flush, 0); } + + private flushNow(): void { + if (!this.#flushScheduled && !this.#dirty) return; + this.#flushScheduled = false; + if (this.#flushTimer !== undefined) { + clearTimeout(this.#flushTimer); + this.#flushTimer = undefined; + } + if (this.#disposed || !this.#dirty) return; + const snapshot = cloneCultMeshDocument(this.#current); + this.#dirty = false; + this.#lastPredicted = snapshot; + this.#pendingSet = this.#pendingSet + .catch(() => undefined) + .then(() => this.#writer.write(snapshot)) + .catch(error => this.#onError(error)); + } + + private throwIfDisposed(): void { + if (this.#disposed) { + throw new Error(`Reactive document '${this.#document.documentId}' is disposed.`); + } + } } function cloneCultMeshDocument(value: TDocument): TDocument { @@ -1295,10 +1407,6 @@ export class CultMeshBoundDocumentHandle { return this.document.canSubmitPrediction; } - public get canSet(): boolean { - return this.document.canSet; - } - public latest(): Promise { return this.document.latest(cultMeshQueryContextFromVerse(this.verse)); } @@ -1319,18 +1427,18 @@ export class CultMeshBoundDocumentHandle { return this.document.submitPrediction(cultMeshQueryContextFromVerse(this.verse), value); } - public set(value: TDocument): Promise { - return this.document.set(cultMeshQueryContextFromVerse(this.verse), value); + public authoritativeWriter(): CultMeshDocumentWriter { + return this.document.authoritativeWriter(cultMeshQueryContextFromVerse(this.verse)); } - public update(update: CultMeshDocumentUpdater): Promise { - return this.document.update(cultMeshQueryContextFromVerse(this.verse), update); + public predictionWriter(): CultMeshDocumentWriter { + return this.document.predictionWriter(cultMeshQueryContextFromVerse(this.verse)); } - public reactive( + public observe( options: Omit = {}, - ): CultMeshReactiveDocument { - return this.document.reactive({ + ): CultMeshObservedDocument { + return this.document.observe({ ...options, context: cultMeshQueryContextFromVerse(this.verse), }); @@ -1451,10 +1559,6 @@ export class CultMeshDocumentCatalog { return this.tryDocument(schema)?.canSubmitPrediction ?? false; } - public canSet(schema: CultMeshDocumentSchemaDescriptor): boolean { - return this.tryDocument(schema)?.canSet ?? false; - } - public submitPrediction( schema: CultMeshDocumentSchemaDescriptor, value: TDocument, @@ -1482,60 +1586,18 @@ export class CultMeshDocumentCatalog { return this.document(schema, options).submitPrediction(context, value); } - public set( + public authoritativeWriter( schema: CultMeshDocumentSchemaDescriptor, - value: TDocument, - options?: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string }, - ): Promise; - public set( - schema: CultMeshDocumentSchemaDescriptor, - context: CultMeshQueryContext | string, - value: TDocument, - options?: { parse?: (value: unknown) => TDocument }, - ): Promise; - public set( - schema: CultMeshDocumentSchemaDescriptor, - contextOrValue: CultMeshQueryContext | string | TDocument, - valueOrOptions?: TDocument | { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string }, - maybeOptions: { parse?: (value: unknown) => TDocument } = {}, - ): Promise { - const hasContext = - typeof contextOrValue === "string" || isCultMeshQueryContext(contextOrValue); - const context = hasContext - ? contextOrValue as CultMeshQueryContext | string - : (valueOrOptions as { context?: CultMeshQueryContext | string } | undefined)?.context ?? "local"; - const value = hasContext ? valueOrOptions as TDocument : contextOrValue as TDocument; - const options = hasContext ? maybeOptions : valueOrOptions as { parse?: (value: unknown) => TDocument } | undefined; - return this.document(schema, options).set(context, value); + options: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string } = {}, + ): CultMeshDocumentWriter { + return this.document(schema, options).authoritativeWriter(options.context ?? "local"); } - public update( + public predictionWriter( schema: CultMeshDocumentSchemaDescriptor, - update: CultMeshDocumentUpdater, - options?: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string }, - ): Promise; - public update( - schema: CultMeshDocumentSchemaDescriptor, - context: CultMeshQueryContext | string, - update: CultMeshDocumentUpdater, - options?: { parse?: (value: unknown) => TDocument }, - ): Promise; - public update( - schema: CultMeshDocumentSchemaDescriptor, - contextOrUpdate: CultMeshQueryContext | string | CultMeshDocumentUpdater, - updateOrOptions?: CultMeshDocumentUpdater | { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string }, - maybeOptions: { parse?: (value: unknown) => TDocument } = {}, - ): Promise { - const hasContext = - typeof contextOrUpdate === "string" || isCultMeshQueryContext(contextOrUpdate); - const context = hasContext - ? contextOrUpdate as CultMeshQueryContext | string - : (updateOrOptions as { context?: CultMeshQueryContext | string } | undefined)?.context ?? "local"; - const update = hasContext - ? updateOrOptions as CultMeshDocumentUpdater - : contextOrUpdate as CultMeshDocumentUpdater; - const options = hasContext ? maybeOptions : updateOrOptions as { parse?: (value: unknown) => TDocument } | undefined; - return this.document(schema, options).update(context, update); + options: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string } = {}, + ): CultMeshDocumentWriter { + return this.document(schema, options).predictionWriter(options.context ?? "local"); } public watch( @@ -3927,29 +3989,6 @@ export class CultMeshNode { ); } - public reactiveDocument( - definition: TDefinition, - key: string, - options: { - context?: CultMeshQueryContext | string; - watch?: boolean; - debounceMs?: number; - onError?: (error: unknown) => void; - documentId?: string; - routeHint?: CultMeshRouteHint; - pollMs?: number; - } = {}, - ): CultMeshReactiveDocument & object> { - const { - documentId, - routeHint, - pollMs, - ...reactiveOptions - } = options; - return this.document(definition, key, { documentId, routeHint, pollMs }) - .reactive & object>(reactiveOptions); - } - public async syncDocumentFromPeerSnapshot( peer: CultNetPeer | (() => CultNetPeer | Promise), definition: TDefinition, @@ -5005,23 +5044,6 @@ export class CultMesh { return cultMeshGlobalDocumentFromCache(cache, definition, options); } - public static reactiveDocument( - node: CultMeshNode, - definition: TDefinition, - key: string, - options: { - context?: CultMeshQueryContext | string; - watch?: boolean; - debounceMs?: number; - onError?: (error: unknown) => void; - documentId?: string; - routeHint?: CultMeshRouteHint; - pollMs?: number; - } = {}, - ): CultMeshReactiveDocument & object> { - return node.reactiveDocument(definition, key, options); - } - public static syncDocumentFromPeerSnapshot( node: CultMeshNode, peer: CultNetPeer | (() => CultNetPeer | Promise), diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index e99bc454..709dc5b8 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -114,7 +114,6 @@ test("CultMesh TS document handles hide local cache plumbing behind typed reacti assert.equal(document.documentId, "cultmesh.note:note:1"); assert.equal(document.canReplace, true); assert.equal(document.canSubmitPrediction, false); - assert.equal(document.canSet, true); assert.equal((await bound.latest()).body, "initial"); await bound.replace({ @@ -131,13 +130,12 @@ test("CultMesh TS document handles hide local cache plumbing behind typed reacti const catalog = CultMesh.documents(document); assert.equal(catalog.canReplace(noteAliasDocument), true); - assert.equal(catalog.canSet(noteAliasDocument), true); - await catalog.set(noteAliasDocument, { - noteId: "note:1", - body: "catalog-updated", - }, { + await catalog.authoritativeWriter(noteAliasDocument, { parse: value => noteAliasDocument.schema.parse(value), context: "browser-client", + }).write({ + noteId: "note:1", + body: "catalog-updated", }); assert.equal( (await catalog.latest(noteAliasDocument, "browser-client", { @@ -145,7 +143,7 @@ test("CultMesh TS document handles hide local cache plumbing behind typed reacti })).body, "catalog-updated", ); - const incremented = await bound.update(value => ({ + const incremented = await bound.authoritativeWriter().update(value => ({ ...value, body: `${value.body}:updated-through-bound-set`, })); @@ -183,17 +181,15 @@ test("CultMesh TS document handles submit predictions through configured authori assert.equal(document.canReplace, false); assert.equal(document.canSubmitPrediction, true); - assert.equal(document.canSet, true); assert.equal(bound.canSubmitPrediction, true); - assert.equal(bound.canSet, true); assert.equal(catalog.canSubmitPrediction(noteAliasDocument), true); - assert.equal(catalog.canSet(noteAliasDocument), true); - await catalog.set(noteAliasDocument, "pilot-a", { + await catalog.predictionWriter(noteAliasDocument, { + parse: value => noteAliasDocument.schema.parse(value), + context: "pilot-a", + }).write({ noteId: "note:prediction", body: "predicted", - }, { - parse: value => noteAliasDocument.schema.parse(value), }); assert.deepEqual(contexts, ["pilot-a:network"]); @@ -204,14 +200,14 @@ test("CultMesh TS document handles submit predictions through configured authori parse: value => noteAliasDocument.schema.parse(value), }); assert.equal(alias.canSubmitPrediction, true); - await alias.set({ + await alias.predictionWriter().write({ noteId: "note:prediction", body: "alias-predicted", }); assert.deepEqual(predictions, ["predicted", "alias-predicted"]); assert.equal((await document.latest("pilot-a")).body, "alias-predicted"); - const updated = await document.update("pilot-a", value => ({ + const updated = await document.predictionWriter("pilot-a").update(value => ({ ...value, body: "updated-as-prediction", })); @@ -227,6 +223,34 @@ test("CultMesh TS document handles submit predictions through configured authori ); }); +test("CultMesh TS documents with both capabilities require an immutable writer choice", async () => { + let current = { noteId: "note:dual", body: "initial" }; + const writes: string[] = []; + const document = CultMesh.document( + "cultmesh.note:dual", + noteDocument, + async () => current, + { + replaceDocument: async (_context, value) => { + writes.push(`authoritative:${value.body}`); + current = value; + }, + submitPrediction: async (_context, value) => { + writes.push(`prediction:${value.body}`); + current = value; + }, + }, + ); + + assert.equal((document as unknown as { set?: unknown }).set, undefined); + await document.authoritativeWriter().write({ noteId: "note:dual", body: "canonical" }); + const predictionWriter = document.predictionWriter(); + await predictionWriter.write({ noteId: "note:dual", body: "predicted" }); + + assert.equal(predictionWriter.kind, "prediction"); + assert.deepEqual(writes, ["authoritative:canonical", "prediction:predicted"]); +}); + test("CultMesh TS store document handles choose compatible foreign schema payloads", async () => { const store = { async pullAll() { @@ -280,7 +304,7 @@ test("CultMesh TS store document handles choose compatible foreign schema payloa }); }); -test("CultMesh TS reactive documents submit predictions from member writes", async () => { +test("CultMesh TS reactive documents submit predictions from explicit updates", async () => { let current = { noteId: "note:reactive", body: "initial", @@ -299,23 +323,71 @@ test("CultMesh TS reactive documents submit predictions from member writes", asy }, ); - const reactive = document.reactive({ + const reactive = document.predictionWriter().reactive({ watch: false, }); await reactive.ready; - reactive.current.body = "member-write-prediction"; - reactive.current.body = "member-write-prediction-final"; + reactive.update(value => { + value.body = "explicit-update-prediction"; + }); + reactive.update(value => { + value.body = "explicit-update-prediction-final"; + }); await waitFor( - () => predictions.includes("member-write-prediction-final"), - "reactive member write prediction", + () => predictions.includes("explicit-update-prediction-final"), + "reactive explicit update prediction", ); - assert.equal(current.body, "member-write-prediction-final"); - assert.deepEqual(predictions, ["member-write-prediction-final"]); + assert.equal(current.body, "explicit-update-prediction-final"); + assert.deepEqual(predictions, ["explicit-update-prediction-final"]); reactive.dispose(); - reactive.current.body = "after-dispose"; + assert.throws(() => reactive.update(value => { + value.body = "after-dispose"; + }), /disposed/); + assert.deepEqual(predictions, ["explicit-update-prediction-final"]); +}); + +test("CultMesh TS reactive updates publish nested edits while snapshot mutation stays local", async () => { + let current = { + shipId: "ship:nested", + loadout: { + weapon: { + key: "weapon:laser", + tuning: { damage: 10 }, + }, + }, + }; + const predictions: typeof current[] = []; + const document = CultMesh.document( + "cultmesh.ship:nested", + { schemaId: "cultmesh.ship.nested.v0" }, + async () => current, + { + submitPrediction: async (_context, value) => { + current = structuredClone(value); + predictions.push(structuredClone(value)); + }, + }, + ); + const reactive = document.predictionWriter().reactive({ watch: false }); + await reactive.ready; + + reactive.update(draft => { + draft.loadout.weapon.tuning.damage = 12; + }); + await reactive.flush(); + + assert.equal(current.loadout.weapon.tuning.damage, 12); + assert.equal(predictions.length, 1); + + const borrowed = reactive.snapshot; + borrowed.loadout.weapon.tuning.damage = 99; await new Promise(resolve => setTimeout(resolve, 20)); - assert.deepEqual(predictions, ["member-write-prediction-final"]); + + assert.equal(reactive.snapshot.loadout.weapon.tuning.damage, 12); + assert.equal(current.loadout.weapon.tuning.damage, 12); + assert.equal(predictions.length, 1); + reactive.dispose(); }); test("CultMesh TS node opens same-schema aliases with one call", async () => { @@ -333,7 +405,7 @@ test("CultMesh TS node opens same-schema aliases with one call", async () => { }); assert.equal(aliasHandle.documentId, "cultmesh.note.ui:note:node-reactive"); assert.equal((await aliasHandle.latest()).body, "initial"); - await aliasHandle.set({ + await aliasHandle.authoritativeWriter().write({ noteId: "note:node-reactive", body: "edited-through-alias-handle", }); @@ -355,26 +427,29 @@ test("CultMesh TS node opens same-schema aliases with one call", async () => { ["edited-through-alias-put"], ); - const reactive = node.reactiveDocument(noteAliasDocument, "note:node-reactive", { - context: CultMesh.queryContext("browser-client"), + const reactive = aliasHandle.authoritativeWriter( + CultMesh.queryContext("browser-client"), + ).reactive({ watch: false, }); await reactive.ready; - reactive.current.body = "edited-through-node-helper"; + reactive.update(value => { + value.body = "edited-through-node-helper"; + }); await waitFor( () => node.getRequired(noteDocument, "note:node-reactive").body === "edited-through-node-helper", "node-level reactive alias write", ); - assert.equal(reactive.current.noteId, "note:node-reactive"); + assert.equal(reactive.snapshot.noteId, "note:node-reactive"); assert.equal( node.document(noteDocument, "note:node-reactive").asSchemaAlias(noteAliasDocument, { parse: value => noteAliasDocument.schema.parse(value), - }).canSet, + }).canReplace, true, ); assert.equal( - (await CultMesh.reactiveDocument(node, noteAliasDocument, "note:node-reactive", { + (await node.document(noteAliasDocument, "note:node-reactive").observe({ watch: false, }).ready).body, "edited-through-node-helper", @@ -409,10 +484,12 @@ test("CultMesh TS reactive documents store reconciliation deltas after mispredic }, ); - const reactive = document.reactive(); + const reactive = document.predictionWriter().reactive(); await reactive.ready; - reactive.current.body = "predicted"; - reactive.current.revision = 7; + reactive.update(value => { + value.body = "predicted"; + value.revision = 7; + }); await waitFor(() => predictions.length === 1, "reactive reconciliation prediction"); watcher?.({ @@ -421,8 +498,8 @@ test("CultMesh TS reactive documents store reconciliation deltas after mispredic revision: 5, }); - assert.equal(reactive.current.body, "authoritative"); - assert.equal(reactive.current.revision, 5); + assert.equal(reactive.snapshot.body, "authoritative"); + assert.equal(reactive.snapshot.revision, 5); assert.equal(reactive.reconciliation?.version, 1); assert.deepEqual(reactive.reconciliation?.canonical, { noteId: "note:reconcile", @@ -444,7 +521,7 @@ test("CultMesh TS reactive documents store reconciliation deltas after mispredic reactive.dispose(); }); -test("CultMesh TS reactive documents coalesce same-frame member writes", async () => { +test("CultMesh TS reactive documents coalesce same-frame explicit updates", async () => { let current = { noteId: "note:reactive-frame", body: "initial", @@ -468,14 +545,14 @@ test("CultMesh TS reactive documents coalesce same-frame member writes", async ( }, ); - const reactive = document.reactive({ + const reactive = document.predictionWriter().reactive({ watch: false, }); try { await reactive.ready; - reactive.current.body = "frame-1"; - reactive.current.body = "frame-2"; - reactive.current.body = "frame-3"; + reactive.update(value => { value.body = "frame-1"; }); + reactive.update(value => { value.body = "frame-2"; }); + reactive.update(value => { value.body = "frame-3"; }); assert.equal(frameCallbacks.length, 1); frameCallbacks[0](performance.now()); @@ -524,14 +601,14 @@ test("CultMesh TS reactive documents queue edits made while a prediction is in f }, ); - const reactive = document.reactive({ + const reactive = document.predictionWriter().reactive({ watch: false, debounceMs: 0, }); await reactive.ready; - reactive.current.body = "first"; + reactive.update(value => { value.body = "first"; }); await firstPredictionStarted; - reactive.current.body = "second"; + reactive.update(value => { value.body = "second"; }); releaseFirstPrediction(); await waitFor(() => predictions.length === 2, "in-flight queued prediction"); @@ -540,13 +617,13 @@ test("CultMesh TS reactive documents queue edits made while a prediction is in f reactive.dispose(); }); -test("CultMesh TS reactive read-only documents surface mutation failures and keep watching", async () => { +test("CultMesh TS observed documents never publish borrowed snapshot mutation and keep watching", async () => { let current = { noteId: "note:reactive-readonly", body: "initial", }; let watcher: ((value: typeof current) => void) | undefined; - const errors: string[] = []; + const writes: typeof current[] = []; const document = CultMesh.document( "cultmesh.note:reactive-readonly", noteDocument, @@ -558,15 +635,17 @@ test("CultMesh TS reactive read-only documents surface mutation failures and kee watcher = undefined; }; }, + submitPrediction: async (_context, value) => { + writes.push(value); + }, }, ); - const reactive = document.reactive({ - onError: error => errors.push(String((error as Error).message ?? error)), - }); - await reactive.ready; - reactive.current.body = "local-readonly-edit"; - await waitFor(() => errors.length === 1, "read-only mutation error"); + const observed = document.observe(); + await observed.ready; + observed.current.body = "borrowed-local-edit"; + await new Promise(resolve => setTimeout(resolve, 20)); + assert.deepEqual(writes, []); current = { noteId: "note:reactive-readonly", @@ -574,9 +653,8 @@ test("CultMesh TS reactive read-only documents surface mutation failures and kee }; watcher?.(current); - assert.match(errors[0], /does not support mutation/); - assert.equal(reactive.current.body, "canonical-after-error"); - reactive.dispose(); + assert.equal(observed.current.body, "canonical-after-error"); + observed.dispose(); }); test("CultMesh TS document handles read schema publications from single-file stores", async () => { @@ -796,7 +874,7 @@ test("CultMesh TS syncs configured publications into local node aliases", async assert.deepEqual(facadeSynced, synced); assert.equal(target.getRequired(noteDocument, "note:published").body, "publication source hydrates local alias"); assert.equal(target.getRequired(noteAliasDocument, "note:published").body, "publication source hydrates local alias"); - assert.equal((await target.reactiveDocument(noteAliasDocument, "note:published", { watch: false }).ready).body, "publication source hydrates local alias"); + assert.equal((await target.document(noteAliasDocument, "note:published").observe({ watch: false }).ready).body, "publication source hydrates local alias"); }); test("CultMesh TS syncs configured publication catalogs into local node aliases", async () => {