From bcb15d461e81763f401060476c1bf294680cf7bb Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 23:57:10 +0100 Subject: [PATCH] Add TS publication catalog sync --- packages/cultmesh-ts/src/index.ts | 78 ++++++++++++++++++++ packages/cultmesh-ts/test/cultmesh.test.ts | 83 ++++++++++++++++++++++ 2 files changed, 161 insertions(+) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index d817370f..c61dfee1 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -2787,6 +2787,57 @@ export async function cultMeshSyncDocumentFromPublication { + if (!node) { + throw new Error("CultMesh node is required."); + } + if (!source) { + throw new Error("CultMesh publication source is required."); + } + if (!bindings) { + throw new Error("CultMesh publication bindings are required."); + } + + const documents = await Promise.all(bindings.map(async binding => { + if (!binding) { + throw new Error("CultMesh publication binding is required."); + } + if (typeof binding.schema === "string") { + throw new Error( + `CultMesh synced publication binding '${binding.recordKey}' requires a document definition, not only a schema id.`, + ); + } + const resolvedSource = binding.source ?? + (typeof source === "function" ? source(binding) : source); + await node.syncDocumentFromPublication( + resolvedSource, + binding.schema, + binding.recordKey, + { + timeoutMs: options.timeoutMs, + messageIdPrefix: options.messageIdPrefix, + }, + ); + return node.document(binding.schema, binding.recordKey, { + documentId: binding.documentId ?? binding.recordKey, + routeHint: options.routeHint, + pollMs: options.pollMs, + }); + })); + + return cultMeshDocuments(...documents); +} + export function cultMeshGlobalDocumentFromCache( cache: CultCache, definition: TDefinition, @@ -3898,6 +3949,19 @@ export class CultMeshNode { return cultMeshSyncDocumentFromPublication(this, source, definition, key, options); } + public syncDocumentsFromPublication( + source: CultMeshDocumentPublicationSourceResolver, + bindings: readonly CultMeshPublicationDocumentBinding[], + options: { + routeHint?: CultMeshRouteHint; + timeoutMs?: number; + pollMs?: number; + messageIdPrefix?: string; + } = {}, + ): Promise { + return cultMeshSyncDocumentsFromPublication(this, source, bindings, options); + } + public globalDocument( definition: TDefinition, options: { @@ -4901,6 +4965,20 @@ export class CultMesh { return node.syncDocumentFromPublication(source, definition, key, options); } + public static syncDocumentsFromPublication( + node: CultMeshNode, + source: CultMeshDocumentPublicationSourceResolver, + bindings: readonly CultMeshPublicationDocumentBinding[], + options: { + routeHint?: CultMeshRouteHint; + timeoutMs?: number; + pollMs?: number; + messageIdPrefix?: string; + } = {}, + ): Promise { + return node.syncDocumentsFromPublication(source, bindings, 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 7bd1d8e0..f4e8f9cc 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -722,6 +722,89 @@ test("CultMesh TS syncs configured publications into local node aliases", async assert.equal((await target.reactiveDocument(noteAliasDocument, "note:published", { watch: false }).ready).body, "publication source hydrates local alias"); }); +test("CultMesh TS syncs configured publication catalogs into local node aliases", async () => { + const firstPath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-publication-sync-catalog-first-")), "first.ccmp"); + const secondPath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-publication-sync-catalog-second-")), "second.ccmp"); + const targetPath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-publication-sync-catalog-target-")), "target.ccmp"); + const first = await CultMesh.startNode(firstPath, { + documents: [noteDocument], + }); + const second = await CultMesh.startNode(secondPath, { + documents: [noteDocument], + }); + await first.put(noteDocument, "note:first-sync", { + noteId: "note:first-sync", + body: "first synced publication", + }); + await second.put(noteDocument, "note:second-sync", { + noteId: "note:second-sync", + body: "second synced publication", + }); + await first.flush(); + await second.flush(); + const target = await CultMesh.startNode(targetPath, { + documents: [noteDocument], + }); + const bindings = [ + CultMesh.publicationDocument(noteDocument, "note:first-sync", { + documentId: "local:first", + }), + CultMesh.publicationDocument(noteAliasDocument, "note:second-sync", { + documentId: "local:second", + source: { + kind: "single-file", + path: secondPath, + }, + }), + ]; + + const catalog = await target.syncDocumentsFromPublication( + { + kind: "single-file", + path: firstPath, + }, + bindings, + { + routeHint: CultMesh.routeHint("shared-memory", "publication sync catalog"), + pollMs: 5, + }, + ); + const facadeCatalog = await CultMesh.syncDocumentsFromPublication( + target, + { + kind: "single-file", + path: firstPath, + }, + bindings, + ); + + assert.deepEqual( + catalog.documents.map(document => document.documentId), + ["local:first", "local:second"], + ); + assert.equal( + await catalog.document(noteDocument, { + parse: value => noteDocument.schema.parse(value), + }).latest().then(value => value.body), + "first synced publication", + ); + assert.equal( + await catalog.document(noteAliasDocument, { + parse: value => noteAliasDocument.schema.parse(value), + }).latest().then(value => value.body), + "second synced publication", + ); + assert.equal(catalog.document(noteAliasDocument).routeHint.description, "publication sync catalog"); + assert.equal(target.getRequired(noteDocument, "note:second-sync").body, "second synced publication"); + assert.equal(target.getRequired(noteAliasDocument, "note:second-sync").body, "second synced publication"); + assert.equal( + await facadeCatalog.document(noteAliasDocument, { + parse: value => noteAliasDocument.schema.parse(value), + }).latest().then(value => value.body), + "second synced publication", + ); +}); + test("CultMesh TS document catalogs resolve semantic schema versions passed as schema ids", async () => { const current = { noteId: "note:semantic",