diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 5ffc1033..714ac2fa 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -316,6 +316,18 @@ const health = CultMesh.documentFromPeerSnapshot( ); const latestHealth = await health.latest(); + +const syncedHealth = await station.syncDocumentFromPeerSnapshot( + () => client, + daemonHealthUiDocument, + "daemon:aetheria.health.v1", + { + timeoutMs: 5_000, + }, +); +const localHealth = await station + .document(daemonHealthUiDocument, "daemon:aetheria.health.v1") + .latest(); ``` `CultMesh.createRudpPeerForPeer(...)` remains available for already trusted @@ -336,6 +348,14 @@ still read the publication with one CultMesh call. Pass a CultCache document definition instead of a raw schema id when you want a typed handle whose `latest()` result is parsed through that definition. +Use `node.syncDocumentFromPeerSnapshot(...)` or +`CultMesh.syncDocumentFromPeerSnapshot(...)` when a runtime should hydrate its +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(...)` +without re-threading the RUDP peer. + The same typed-definition overload is available for `CultMesh.documentFromStore(...)` and `CultMesh.documentFromSingleFile(...)`, so local schema publications can be read as typed reactive documents without diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 0081ec32..ab1b4779 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -3815,6 +3815,46 @@ export class CultMeshNode { .reactive & object>(reactiveOptions); } + public async syncDocumentFromPeerSnapshot( + peer: CultNetPeer | (() => CultNetPeer | Promise), + definition: TDefinition, + key: string, + options: { + timeoutMs?: number; + messageIdPrefix?: string; + } = {}, + ): Promise> { + const registered = this.resolveCacheDefinition(definition); + const schema = cultMeshSchemaFromDefinition(definition); + const schemaId = schema.schemaId ?? schema.type; + if (!schemaId) { + throw new Error("CultMesh synced document requires a schema id or document type."); + } + const document = await requestCultNetRawSnapshotDocument( + peer, + schemaId, + key, + { + timeoutMs: options.timeoutMs, + messageIdPrefix: options.messageIdPrefix ?? key, + accepts: record => { + try { + definition.schema.parse(decodeCultNetRawDocumentPayload(record)); + return true; + } catch { + return false; + } + }, + }, + ); + await this.documents.applyRawDocumentPutMessage(this.cache, { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: options.messageIdPrefix ?? `sync:${key}`, + document, + }); + return this.parseDocumentValue(definition, this.cache.getRequired(registered, key)); + } + public globalDocument( definition: TDefinition, options: { @@ -4792,6 +4832,19 @@ export class CultMesh { return node.reactiveDocument(definition, key, options); } + public static syncDocumentFromPeerSnapshot( + node: CultMeshNode, + peer: CultNetPeer | (() => CultNetPeer | Promise), + definition: TDefinition, + key: string, + options: { + timeoutMs?: number; + messageIdPrefix?: string; + } = {}, + ): Promise> { + return node.syncDocumentFromPeerSnapshot(peer, definition, key, options); + } + public static documents( ...documents: readonly CultMeshDocumentHandle[] ): CultMeshDocumentCatalog { diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 993ef62d..0a859861 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -2530,6 +2530,90 @@ test("CultMesh TS reads remote RUDP snapshots through document handles", async ( } }); +test("CultMesh TS syncs remote RUDP snapshots into a local node with same-schema aliases", async () => { + const connectionId = 0x1020304a; + const source = await CultMesh.startNode( + join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-sync-source-")), "node.ccmp"), + { + documents: [noteDocument], + }, + ); + await source.put(noteDocument, "note:sync-remote", { + noteId: "note:sync-remote", + body: "synced once, read as local", + }); + const target = await CultMesh.startNode( + join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-sync-target-")), "node.ccmp"), + { + documents: [noteDocument], + }, + ); + + const server = CultMesh.createRudpDocumentServer( + "cultmesh-ts-rudp-sync-server", + connectionId, + { + documents: new CultNetDocumentRegistry([ + defineCultNetDocumentBinding({ definition: noteDocument }), + ]), + getCache: () => source.cache, + bindHost: "127.0.0.1", + bindPort: 0, + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 1024, + maxPendingReliablePackets: 16, + }, + ); + + let peer: CultNetPeer | undefined; + try { + await server.start(); + peer = await CultMesh.createRudpPeer( + "cultmesh-ts-rudp-sync-client", + connectionId, + `rudp://127.0.0.1:${server.bind.port}`, + { + resendDelayMs: 25, + resendPollMs: 5, + maxFragmentBytes: 1024, + maxPendingReliablePackets: 16, + connectTimeoutMs: 1_000, + }, + ); + + const synced = await target.syncDocumentFromPeerSnapshot( + peer, + noteAliasDocument, + "note:sync-remote", + { + timeoutMs: 1_000, + }, + ); + const facadeSynced = await CultMesh.syncDocumentFromPeerSnapshot( + target, + peer, + noteAliasDocument, + "note:sync-remote", + { + timeoutMs: 1_000, + }, + ); + + assert.deepEqual(synced, { + noteId: "note:sync-remote", + body: "synced once, read as local", + }); + assert.deepEqual(facadeSynced, synced); + assert.equal(target.getRequired(noteDocument, "note:sync-remote").body, "synced once, read as local"); + assert.equal(target.getRequired(noteAliasDocument, "note:sync-remote").body, "synced once, read as local"); + assert.equal((await target.document(noteAliasDocument, "note:sync-remote").latest()).body, "synced once, read as local"); + } finally { + peer?.close(); + server.close(); + } +}); + test("CultMesh TS peer snapshot handles choose compatible foreign schema payloads", async () => { const connectionId = 0x10203049; const node = await CultMesh.startNode(