Add TS publication catalog sync

This commit is contained in:
Meta Vens 2026-06-27 23:57:10 +01:00
commit bcb15d461e
2 changed files with 161 additions and 0 deletions

View file

@ -2787,6 +2787,57 @@ export async function cultMeshSyncDocumentFromPublication<TDefinition extends An
return node.put(definition, recordKey, value);
}
export async function cultMeshSyncDocumentsFromPublication(
node: CultMeshNode,
source: CultMeshDocumentPublicationSourceResolver,
bindings: readonly CultMeshPublicationDocumentBinding[],
options: {
routeHint?: CultMeshRouteHint;
timeoutMs?: number;
pollMs?: number;
messageIdPrefix?: string;
} = {},
): Promise<CultMeshDocumentCatalog> {
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<TDefinition extends AnyCultCacheDocumentDefinition>(
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<CultMeshDocumentCatalog> {
return cultMeshSyncDocumentsFromPublication(this, source, bindings, options);
}
public globalDocument<TDefinition extends AnyCultCacheDocumentDefinition>(
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<CultMeshDocumentCatalog> {
return node.syncDocumentsFromPublication(source, bindings, options);
}
public static documents(
...documents: readonly CultMeshDocumentHandle<any>[]
): CultMeshDocumentCatalog {

View file

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