Make CultMesh TypeScript document writes explicit

This commit is contained in:
Meta Vens 2026-08-17 19:58:54 +02:00
commit 425b430fe6
3 changed files with 384 additions and 281 deletions

View file

@ -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

View file

@ -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<TDocument extends object
}
export interface CultMeshReactiveDocument<TDocument extends object> {
readonly current: TDocument;
readonly writer: CultMeshDocumentWriter<TDocument>;
readonly snapshot: TDocument;
readonly dirty: boolean;
readonly reconciliation: CultMeshReactiveDocumentReconciliation<TDocument> | undefined;
readonly ready: Promise<TDocument>;
readonly disposed: boolean;
update(update: (draft: TDocument) => void | TDocument): TDocument;
replaceLocal(value: TDocument): TDocument;
flush(): Promise<void>;
clearReconciliation(): void;
refresh(): Promise<TDocument>;
dispose(): void;
}
export interface CultMeshObservedDocument<TDocument extends object> {
readonly current: TDocument;
readonly ready: Promise<TDocument>;
readonly disposed: boolean;
refresh(): Promise<TDocument>;
dispose(): void;
}
export type CultMeshDocumentWriteKind = "authoritative-replacement" | "prediction";
export interface CultMeshNativeSliceColumn {
readonly name: string;
readonly valueType: string;
@ -842,10 +858,6 @@ export class CultMeshDocumentHandle<TDocument> {
return this.submitPredictionDocument !== undefined;
}
public get canSet(): boolean {
return this.canSubmitPrediction || this.canReplace;
}
public latest(context: CultMeshQueryContext | string = "local"): Promise<TDocument> {
return this.snapshotDocument(
this.resolveContext(typeof context === "string" ? cultMeshQueryContext(context) : context),
@ -932,63 +944,44 @@ export class CultMeshDocumentHandle<TDocument> {
return this.submitPredictionDocument(this.resolveContext(context), value);
}
public set(value: TDocument): Promise<void>;
public set(context: CultMeshQueryContext | string, value: TDocument): Promise<void>;
public set(
contextOrValue: CultMeshQueryContext | string | TDocument,
maybeValue?: TDocument,
): Promise<void> {
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<TDocument & object> {
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<TDocument & object>,
"authoritative-replacement",
resolved,
value => this.replaceDocument!(resolved, value as unknown as TDocument),
);
}
public update(update: CultMeshDocumentUpdater<TDocument>): Promise<TDocument>;
public update(
context: CultMeshQueryContext | string,
update: CultMeshDocumentUpdater<TDocument>,
): Promise<TDocument>;
public async update(
contextOrUpdate: CultMeshQueryContext | string | CultMeshDocumentUpdater<TDocument>,
maybeUpdate?: CultMeshDocumentUpdater<TDocument>,
): Promise<TDocument> {
const hasContext =
typeof contextOrUpdate === "string" || isCultMeshQueryContext(contextOrUpdate);
const context = hasContext
? contextOrUpdate as CultMeshQueryContext | string
: "local";
const update = hasContext ? maybeUpdate : contextOrUpdate as CultMeshDocumentUpdater<TDocument>;
if (!update) {
throw new Error(`Document '${this.documentId}' requires an update function.`);
public predictionWriter(
context: CultMeshQueryContext | string = "local",
): CultMeshDocumentWriter<TDocument & object> {
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<TDocument & object>,
"prediction",
resolved,
value => this.submitPredictionDocument!(resolved, value as unknown as TDocument),
);
}
public reactive<TObject extends TDocument & object>(
public observe<TObject extends TDocument & object>(
options: CultMeshReactiveDocumentOptions = {},
): CultMeshReactiveDocument<TObject> {
return new CultMeshReactiveDocumentHandle<TObject>(
): CultMeshObservedDocument<TObject> {
return new CultMeshObservedDocumentHandle<TObject>(
this as unknown as CultMeshDocumentHandle<TObject>,
options,
);
@ -1041,22 +1034,40 @@ export class CultMeshDocumentHandle<TDocument> {
}
}
class CultMeshReactiveDocumentHandle<TDocument extends object>
implements CultMeshReactiveDocument<TDocument> {
export class CultMeshDocumentWriter<TDocument extends object> {
public constructor(
public readonly document: CultMeshDocumentHandle<TDocument>,
public readonly kind: CultMeshDocumentWriteKind,
public readonly context: CultMeshQueryContext,
private readonly writeDocument: (value: TDocument) => Promise<void>,
) {}
public write(value: TDocument): Promise<void> {
return this.writeDocument(cloneCultMeshDocument(value));
}
public async update(update: CultMeshDocumentUpdater<TDocument>): Promise<TDocument> {
const next = await update(cloneCultMeshDocument(await this.document.latest(this.context)));
await this.write(next);
return next;
}
public reactive(
options: Omit<CultMeshReactiveDocumentOptions, "context"> = {},
): CultMeshReactiveDocument<TDocument> {
return new CultMeshReactiveDocumentHandle(this, options);
}
}
class CultMeshObservedDocumentHandle<TDocument extends object>
implements CultMeshObservedDocument<TDocument> {
readonly #document: CultMeshDocumentHandle<TDocument>;
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<void> = Promise.resolve();
#flushScheduled = false;
#flushTimer: ReturnType<typeof setTimeout> | undefined;
#unsubscribe: CultMeshUnsubscribe | undefined;
#lastPredicted: TDocument | undefined;
#reconciliation: CultMeshReactiveDocumentReconciliation<TDocument> | undefined;
#reconciliationVersion = 0;
public readonly ready: Promise<TDocument>;
@ -1066,19 +1077,17 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
) {
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<TDocument extends object>
}
public get current(): TDocument {
this.throwIfDisposed();
return this.#current;
}
public get disposed(): boolean {
return this.#disposed;
}
public async refresh(): Promise<TDocument> {
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<TDocument extends object>
implements CultMeshReactiveDocument<TDocument> {
readonly #writer: CultMeshDocumentWriter<TDocument>;
readonly #document: CultMeshDocumentHandle<TDocument>;
readonly #onError: (error: unknown) => void;
readonly #debounceMs: number;
readonly #replaceDirtyCurrentOnCanonicalSnapshot: boolean;
#current = {} as TDocument;
#disposed = false;
#pendingSet: Promise<void> = Promise.resolve();
#flushScheduled = false;
#flushTimer: ReturnType<typeof setTimeout> | undefined;
#unsubscribe: CultMeshUnsubscribe | undefined;
#dirty = false;
#lastPredicted: TDocument | undefined;
#reconciliation: CultMeshReactiveDocumentReconciliation<TDocument> | undefined;
#reconciliationVersion = 0;
#canonicalVersion = 0;
public readonly ready: Promise<TDocument>;
public constructor(
writer: CultMeshDocumentWriter<TDocument>,
options: Omit<CultMeshReactiveDocumentOptions, "context"> = {},
) {
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<TDocument> {
return this.#writer;
}
public get snapshot(): TDocument {
this.throwIfDisposed();
return cloneCultMeshDocument(this.#current);
}
public get dirty(): boolean {
return this.#dirty;
}
public get reconciliation(): CultMeshReactiveDocumentReconciliation<TDocument> | undefined {
return this.#reconciliation;
}
@ -1099,9 +1199,37 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
}
public async refresh(): Promise<TDocument> {
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<void> {
this.throwIfDisposed();
this.flushNow();
await this.#pendingSet;
}
public clearReconciliation(): void {
@ -1121,18 +1249,13 @@ class CultMeshReactiveDocumentHandle<TDocument extends object>
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<TDocument extends object>
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<TDocument extends object>
}
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<TDocument extends object>
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<TDocument>(value: TDocument): TDocument {
@ -1295,10 +1407,6 @@ export class CultMeshBoundDocumentHandle<TDocument> {
return this.document.canSubmitPrediction;
}
public get canSet(): boolean {
return this.document.canSet;
}
public latest(): Promise<TDocument> {
return this.document.latest(cultMeshQueryContextFromVerse(this.verse));
}
@ -1319,18 +1427,18 @@ export class CultMeshBoundDocumentHandle<TDocument> {
return this.document.submitPrediction(cultMeshQueryContextFromVerse(this.verse), value);
}
public set(value: TDocument): Promise<void> {
return this.document.set(cultMeshQueryContextFromVerse(this.verse), value);
public authoritativeWriter(): CultMeshDocumentWriter<TDocument & object> {
return this.document.authoritativeWriter(cultMeshQueryContextFromVerse(this.verse));
}
public update(update: CultMeshDocumentUpdater<TDocument>): Promise<TDocument> {
return this.document.update(cultMeshQueryContextFromVerse(this.verse), update);
public predictionWriter(): CultMeshDocumentWriter<TDocument & object> {
return this.document.predictionWriter(cultMeshQueryContextFromVerse(this.verse));
}
public reactive<TObject extends TDocument & object>(
public observe<TObject extends TDocument & object>(
options: Omit<CultMeshReactiveDocumentOptions, "context"> = {},
): CultMeshReactiveDocument<TObject> {
return this.document.reactive<TObject>({
): CultMeshObservedDocument<TObject> {
return this.document.observe<TObject>({
...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<TDocument>(
schema: CultMeshDocumentSchemaDescriptor,
value: TDocument,
@ -1482,60 +1586,18 @@ export class CultMeshDocumentCatalog {
return this.document(schema, options).submitPrediction(context, value);
}
public set<TDocument>(
public authoritativeWriter<TDocument extends object>(
schema: CultMeshDocumentSchemaDescriptor,
value: TDocument,
options?: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string },
): Promise<void>;
public set<TDocument>(
schema: CultMeshDocumentSchemaDescriptor,
context: CultMeshQueryContext | string,
value: TDocument,
options?: { parse?: (value: unknown) => TDocument },
): Promise<void>;
public set<TDocument>(
schema: CultMeshDocumentSchemaDescriptor,
contextOrValue: CultMeshQueryContext | string | TDocument,
valueOrOptions?: TDocument | { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string },
maybeOptions: { parse?: (value: unknown) => TDocument } = {},
): Promise<void> {
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<TDocument> {
return this.document(schema, options).authoritativeWriter(options.context ?? "local");
}
public update<TDocument>(
public predictionWriter<TDocument extends object>(
schema: CultMeshDocumentSchemaDescriptor,
update: CultMeshDocumentUpdater<TDocument>,
options?: { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string },
): Promise<TDocument>;
public update<TDocument>(
schema: CultMeshDocumentSchemaDescriptor,
context: CultMeshQueryContext | string,
update: CultMeshDocumentUpdater<TDocument>,
options?: { parse?: (value: unknown) => TDocument },
): Promise<TDocument>;
public update<TDocument>(
schema: CultMeshDocumentSchemaDescriptor,
contextOrUpdate: CultMeshQueryContext | string | CultMeshDocumentUpdater<TDocument>,
updateOrOptions?: CultMeshDocumentUpdater<TDocument> | { parse?: (value: unknown) => TDocument; context?: CultMeshQueryContext | string },
maybeOptions: { parse?: (value: unknown) => TDocument } = {},
): Promise<TDocument> {
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<TDocument>
: contextOrUpdate as CultMeshDocumentUpdater<TDocument>;
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<TDocument> {
return this.document(schema, options).predictionWriter(options.context ?? "local");
}
public watch<TDocument>(
@ -3927,29 +3989,6 @@ export class CultMeshNode {
);
}
public reactiveDocument<TDefinition extends AnyCultCacheDocumentDefinition>(
definition: TDefinition,
key: string,
options: {
context?: CultMeshQueryContext | string;
watch?: boolean;
debounceMs?: number;
onError?: (error: unknown) => void;
documentId?: string;
routeHint?: CultMeshRouteHint;
pollMs?: number;
} = {},
): CultMeshReactiveDocument<CultCacheDocumentValue<TDefinition> & object> {
const {
documentId,
routeHint,
pollMs,
...reactiveOptions
} = options;
return this.document(definition, key, { documentId, routeHint, pollMs })
.reactive<CultCacheDocumentValue<TDefinition> & object>(reactiveOptions);
}
public async syncDocumentFromPeerSnapshot<TDefinition extends AnyCultCacheDocumentDefinition>(
peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
definition: TDefinition,
@ -5005,23 +5044,6 @@ export class CultMesh {
return cultMeshGlobalDocumentFromCache(cache, definition, options);
}
public static reactiveDocument<TDefinition extends AnyCultCacheDocumentDefinition>(
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<CultCacheDocumentValue<TDefinition> & object> {
return node.reactiveDocument(definition, key, options);
}
public static syncDocumentFromPeerSnapshot<TDefinition extends AnyCultCacheDocumentDefinition>(
node: CultMeshNode,
peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),

View file

@ -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<typeof current>({
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<typeof current>();
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<typeof current>({
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<typeof current>({
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<typeof current>({
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<typeof current>();
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 () => {