diff --git a/packages/cultcache-ts/src/document.ts b/packages/cultcache-ts/src/document.ts index e6cd080e..50e08afb 100644 --- a/packages/cultcache-ts/src/document.ts +++ b/packages/cultcache-ts/src/document.ts @@ -7,6 +7,7 @@ import type { CultCacheSchemaCatalogMember, CultCacheDocumentDefinition, CultCacheDocumentRegistry, + CultCacheDocumentValue, CultCacheSchema, } from "./types"; @@ -90,10 +91,13 @@ const plainDocumentFormatter: CultCacheDocumentFormatter = { decode: (payload: Uint8Array) => decode(payload), }; -// The one reader of a document payload outside the cache: through the definition's formatter, or as a plain msgpack -// value when the caller holds no definition for the schema. Every package above the cache reads a payload through here. -export function decodeDocumentPayload(definition: AnyCultCacheDocumentDefinition | undefined, payload: Uint8Array): unknown { - return (definition ? resolveDocumentFormatter(definition) : plainDocumentFormatter).decode(payload); +// The one reader of a document payload: through the definition's formatter, codec or plain map, then its schema. A +// reader holds the definition or it has no business reading the bytes, so there is no definition-less form. +export function decodeDocumentPayload( + definition: TDefinition, + payload: Uint8Array, +): CultCacheDocumentValue { + return definition.schema.parse(resolveDocumentFormatter(definition).decode(payload)) as CultCacheDocumentValue; } // The schema members of a document: the ones it declares, else those its codec derives, else none. diff --git a/packages/cultcache-ts/test/decode-document-payload.test.ts b/packages/cultcache-ts/test/decode-document-payload.test.ts index a63859ee..e212c54a 100644 --- a/packages/cultcache-ts/test/decode-document-payload.test.ts +++ b/packages/cultcache-ts/test/decode-document-payload.test.ts @@ -23,13 +23,10 @@ const formatted = defineDocumentType({ formatter: { encode: (value) => encode([(value as { id: string }).id]), decode: () => ({ id: "from-formatter", count: 0 }) }, }); -test("a payload is read by its definition's formatter, a codec's, or as a plain value without a definition", () => { - assert.deepEqual(decodeDocumentPayload(undefined, encode({ id: "a", count: 1 })), { id: "a", count: 1 }); +test("a payload is read by its definition's formatter, a codec's, or a plain map", () => { assert.deepEqual(decodeDocumentPayload(plain, encode({ id: "a", count: 1 })), { id: "a", count: 1 }); assert.deepEqual(decodeDocumentPayload(codec, encode(["a", 1])), { id: "a", count: 1 }); assert.deepEqual(decodeDocumentPayload(formatted, new Uint8Array([0xc0])), { id: "from-formatter", count: 0 }); - // Without the definition the codec payload is its positional array, which is why a caller that holds one passes it. - assert.deepEqual(decodeDocumentPayload(undefined, encode(["a", 1])), ["a", 1]); }); test("a definition that declares both a codec and a formatter is refused where its payload is read", () => { diff --git a/packages/cultmesh-browser/test/payload-decode-owner.test.ts b/packages/cultmesh-browser/test/payload-decode-owner.test.ts deleted file mode 100644 index e5a23f78..00000000 --- a/packages/cultmesh-browser/test/payload-decode-owner.test.ts +++ /dev/null @@ -1,33 +0,0 @@ -import assert from "node:assert/strict"; -import { existsSync, readdirSync, readFileSync } from "node:fs"; -import { dirname, join } from "node:path"; -import { fileURLToPath } from "node:url"; -import test from "node:test"; - -// A document payload is decoded by cultcache-ts alone (decodeDocumentPayload); every other value on a wire goes through -// cultnet-ts's decodeWireValue / encodeWireValue. So no file of this package imports msgpack, and the manifest does not -// list it, which keeps a new direct decode from arriving unnoticed. -// Limits: a text scan. It reads the package's src files for the module specifier and the manifest for the dependency -// name. It does not follow a re-export from another package, a specifier assembled at run time, or a hand-written -// decoder; the dependency absence is what makes the specifier unresolvable here when the workspace does not hoist it. -// The package root: the nearest directory above this file (compiled or run in place) that holds a package.json. -function packageRoot(from: string): string { - return existsSync(join(from, "package.json")) ? from : packageRoot(dirname(from)); -} -const root = packageRoot(dirname(fileURLToPath(import.meta.url))); - -function sourceFiles(dir: string): string[] { - return readdirSync(dir, { withFileTypes: true }).flatMap((entry) => - entry.isDirectory() ? sourceFiles(join(dir, entry.name)) : entry.name.endsWith(".ts") ? [join(dir, entry.name)] : []); -} - -test("no source file imports msgpack and the manifest does not list it", () => { - const manifest = JSON.parse(readFileSync(join(root, "package.json"), "utf8")) as Record | undefined>; - for (const section of ["dependencies", "devDependencies", "peerDependencies", "optionalDependencies"]) { - assert.equal(manifest[section]?.["@msgpack/msgpack"], undefined, section); - } - const files = sourceFiles(join(root, "src")); - assert.ok(files.length >= 1, "fixture: the scan reads the package's sources"); - const offenders = files.filter((file) => /@msgpack\/msgpack/.test(readFileSync(file, "utf8"))); - assert.deepEqual(offenders, []); -}); diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index f738cdcd..8dee1a39 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -127,7 +127,7 @@ export type CultMeshDocumentPublicationSource = }; export interface CultMeshPublicationDocumentBinding { - readonly schema: string | AnyCultCacheDocumentDefinition; + readonly schema: AnyCultCacheDocumentDefinition; readonly recordKey: string; readonly documentId?: string; readonly sourceId?: string; @@ -2591,54 +2591,29 @@ export function cultMeshDocumentFromCache; export function cultMeshDocumentFromStore( store: CacheBackingStore, definition: TDefinition, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, -): CultMeshDocumentHandle>; -export function cultMeshDocumentFromStore( - store: CacheBackingStore, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, options: { documentId?: string; routeHint?: CultMeshRouteHint; sourceId?: string; pollMs?: number; } = {}, -): CultMeshDocumentHandle { - const schema = typeof schemaOrDefinition === "string" - ? { schemaId: schemaOrDefinition } - : cultMeshSchemaFromDefinition(schemaOrDefinition); +): CultMeshDocumentHandle> { + const schema = cultMeshSchemaFromDefinition(definition); const resolvedSchemaId = schema.schemaId ?? schema.type; if (!resolvedSchemaId) { throw new Error("CultMesh store document requires a schema id or document type."); } - if (typeof schemaOrDefinition === "string") { - requireNonEmpty(schemaOrDefinition, "schemaId"); - } const documentId = options.documentId ?? resolvedSchemaId; - const read = async () => { + const read = async (): Promise> => { const records = await store.pullAll(); - const record = resolveCultMeshStoreDocumentRecord(records, schema, schemaOrDefinition); + const record = resolveCultMeshStoreDocumentRecord(records, schema, definition); if (!record) { throw new Error(`CultMesh store document '${documentId}' did not contain schema ${resolvedSchemaId}.`); } - return parseCultMeshDocumentValue(schemaOrDefinition, decodeCultMeshPayload(schemaOrDefinition, record.payload)); + return decodeDocumentPayload(definition, record.payload); }; return cultMeshDocument( @@ -2655,77 +2630,28 @@ export function cultMeshDocumentFromStore( ); } -export function cultMeshDocumentFromSingleFile( - path: string, - schemaId: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, -): CultMeshDocumentHandle; export function cultMeshDocumentFromSingleFile( path: string, definition: TDefinition, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, -): CultMeshDocumentHandle>; -export function cultMeshDocumentFromSingleFile( - path: string, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, options: { documentId?: string; routeHint?: CultMeshRouteHint; sourceId?: string; pollMs?: number; } = {}, -): CultMeshDocumentHandle { +): CultMeshDocumentHandle> { requireNonEmpty(path, "path"); const store = new SingleFileMessagePackBackingStore(path); - const resolvedOptions = { + return cultMeshDocumentFromStore(store, definition, { ...options, routeHint: options.routeHint ?? cultMeshRouteHint("shared-memory", path), - }; - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromStore(store, schemaOrDefinition, resolvedOptions) - : cultMeshDocumentFromStore(store, schemaOrDefinition, resolvedOptions); + }); } -export function cultMeshDocumentFromPeerSnapshot( - peer: CultNetPeer | (() => CultNetPeer | Promise), - schemaId: string, - recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, -): CultMeshDocumentHandle; export function cultMeshDocumentFromPeerSnapshot( peer: CultNetPeer | (() => CultNetPeer | Promise), definition: TDefinition, recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, -): CultMeshDocumentHandle>; -export function cultMeshDocumentFromPeerSnapshot( - peer: CultNetPeer | (() => CultNetPeer | Promise), - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, - recordKey: string, options: { documentId?: string; routeHint?: CultMeshRouteHint; @@ -2734,45 +2660,34 @@ export function cultMeshDocumentFromPeerSnapshot( pollMs?: number; messageIdPrefix?: string; } = {}, -): CultMeshDocumentHandle { - const schema = typeof schemaOrDefinition === "string" - ? { schemaId: schemaOrDefinition } - : cultMeshSchemaFromDefinition(schemaOrDefinition); +): CultMeshDocumentHandle> { + const schema = cultMeshSchemaFromDefinition(definition); const resolvedSchemaId = schema.schemaId ?? schema.type; if (!resolvedSchemaId) { throw new Error("CultMesh peer snapshot document requires a schema id or document type."); } - if (typeof schemaOrDefinition === "string") { - requireNonEmpty(schemaOrDefinition, "schemaId"); - } requireNonEmpty(recordKey, "recordKey"); const documentId = options.documentId ?? recordKey; - const read = async () => - parseCultMeshDocumentValue( - schemaOrDefinition, - decodeCultNetRawDocumentPayload( - schemaOrDefinition, - await requestCultNetRawSnapshotDocument( - peer, - resolvedSchemaId, - recordKey, - { - timeoutMs: options.timeoutMs, - messageIdPrefix: options.messageIdPrefix ?? documentId, - accepts: typeof schemaOrDefinition === "string" - ? undefined - : record => { - try { - schemaOrDefinition.schema.parse(decodeCultNetRawDocumentPayload(schemaOrDefinition, record)); - return true; - } catch { - return false; - } - }, - }, - ), - ), + const read = async (): Promise> => { + const record = await requestCultNetRawSnapshotDocument( + peer, + resolvedSchemaId, + recordKey, + { + timeoutMs: options.timeoutMs, + messageIdPrefix: options.messageIdPrefix ?? documentId, + accepts: candidate => { + try { + decodeDocumentPayload(definition, candidate.payload); + return true; + } catch { + return false; + } + }, + }, ); + return decodeDocumentPayload(definition, record.payload); + }; return cultMeshDocument( documentId, @@ -2788,36 +2703,10 @@ export function cultMeshDocumentFromPeerSnapshot( ); } -export function cultMeshDocumentFromPublication( - source: CultMeshDocumentPublicationSource, - schemaId: string, - recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, -): CultMeshDocumentHandle; export function cultMeshDocumentFromPublication( source: CultMeshDocumentPublicationSource, definition: TDefinition, recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, -): CultMeshDocumentHandle>; -export function cultMeshDocumentFromPublication( - source: CultMeshDocumentPublicationSource, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, - recordKey: string, options: { documentId?: string; routeHint?: CultMeshRouteHint; @@ -2826,7 +2715,7 @@ export function cultMeshDocumentFromPublication( pollMs?: number; messageIdPrefix?: string; } = {}, -): CultMeshDocumentHandle { +): CultMeshDocumentHandle> { if (!source) { throw new Error("CultMesh publication source is required."); } @@ -2834,14 +2723,12 @@ export function cultMeshDocumentFromPublication( switch (source.kind) { case "live-publication": { - const schema = typeof schemaOrDefinition === "string" - ? { schemaId: schemaOrDefinition } - : cultMeshSchemaFromDefinition(schemaOrDefinition); + const schema = cultMeshSchemaFromDefinition(definition); const schemaId = schema.schemaId ?? schema.type; if (!schemaId) { throw new Error("CultMesh live publication document requires a schema id or document type."); } - const parse = (value: unknown) => parseCultMeshDocumentValue(schemaOrDefinition, value); + const parse = (value: unknown) => definition.schema.parse(value) as CultCacheDocumentValue; return cultMeshDocument( options.documentId ?? recordKey, schema, @@ -2861,51 +2748,28 @@ export function cultMeshDocumentFromPublication( ); } case "single-file": - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromSingleFile(source.path, schemaOrDefinition, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint, - sourceId: options.sourceId ?? recordKey, - pollMs: options.pollMs, - }) - : cultMeshDocumentFromSingleFile(source.path, schemaOrDefinition, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint, - sourceId: options.sourceId ?? recordKey, - pollMs: options.pollMs, - }); + return cultMeshDocumentFromSingleFile(source.path, definition, { + documentId: options.documentId ?? recordKey, + routeHint: options.routeHint, + sourceId: options.sourceId ?? recordKey, + pollMs: options.pollMs, + }); case "store": - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromStore(source.store, schemaOrDefinition, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint, - sourceId: options.sourceId ?? recordKey, - pollMs: options.pollMs, - }) - : cultMeshDocumentFromStore(source.store, schemaOrDefinition, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint, - sourceId: options.sourceId ?? recordKey, - pollMs: options.pollMs, - }); + return cultMeshDocumentFromStore(source.store, definition, { + documentId: options.documentId ?? recordKey, + routeHint: options.routeHint, + sourceId: options.sourceId ?? recordKey, + pollMs: options.pollMs, + }); case "peer-snapshot": - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromPeerSnapshot(source.peer, schemaOrDefinition, recordKey, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint ?? cultMeshRouteHint("network", source.endpoint ?? "CultNet snapshot"), - sourceId: options.sourceId ?? recordKey, - timeoutMs: options.timeoutMs, - pollMs: options.pollMs, - messageIdPrefix: options.messageIdPrefix, - }) - : cultMeshDocumentFromPeerSnapshot(source.peer, schemaOrDefinition, recordKey, { - documentId: options.documentId ?? recordKey, - routeHint: options.routeHint ?? cultMeshRouteHint("network", source.endpoint ?? "CultNet snapshot"), - sourceId: options.sourceId ?? recordKey, - timeoutMs: options.timeoutMs, - pollMs: options.pollMs, - messageIdPrefix: options.messageIdPrefix, - }); + return cultMeshDocumentFromPeerSnapshot(source.peer, definition, recordKey, { + documentId: options.documentId ?? recordKey, + routeHint: options.routeHint ?? cultMeshRouteHint("network", source.endpoint ?? "CultNet snapshot"), + sourceId: options.sourceId ?? recordKey, + timeoutMs: options.timeoutMs, + pollMs: options.pollMs, + messageIdPrefix: options.messageIdPrefix, + }); default: throw new Error(`Unsupported CultMesh publication source kind '${(source as { kind?: string }).kind}'.`); } @@ -2967,11 +2831,6 @@ export async function cultMeshSyncDocumentsFromPublication( 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( @@ -3029,7 +2888,7 @@ export function cultMeshDocuments( } export function cultMeshPublicationDocument( - schema: string | AnyCultCacheDocumentDefinition, + schema: AnyCultCacheDocumentDefinition, recordKey: string, options: { documentId?: string; @@ -3071,33 +2930,19 @@ export function cultMeshDocumentsFromPublication( } const resolvedSource = binding.source ?? (typeof source === "function" ? source(binding) : source); - return typeof binding.schema === "string" - ? cultMeshDocumentFromPublication( - resolvedSource, - binding.schema, - binding.recordKey, - { - documentId: binding.documentId ?? binding.recordKey, - sourceId: binding.sourceId ?? binding.documentId ?? binding.recordKey, - routeHint: options.routeHint, - timeoutMs: options.timeoutMs, - pollMs: options.pollMs, - messageIdPrefix: options.messageIdPrefix, - }, - ) - : cultMeshDocumentFromPublication( - resolvedSource, - binding.schema, - binding.recordKey, - { - documentId: binding.documentId ?? binding.recordKey, - sourceId: binding.sourceId ?? binding.documentId ?? binding.recordKey, - routeHint: options.routeHint, - timeoutMs: options.timeoutMs, - pollMs: options.pollMs, - messageIdPrefix: options.messageIdPrefix, - }, - ); + return cultMeshDocumentFromPublication( + resolvedSource, + binding.schema, + binding.recordKey, + { + documentId: binding.documentId ?? binding.recordKey, + sourceId: binding.sourceId ?? binding.documentId ?? binding.recordKey, + routeHint: options.routeHint, + timeoutMs: options.timeoutMs, + pollMs: options.pollMs, + messageIdPrefix: options.messageIdPrefix, + }, + ); }), ); } @@ -4108,7 +3953,7 @@ export class CultMeshNode { messageIdPrefix: options.messageIdPrefix ?? key, accepts: record => { try { - definition.schema.parse(decodeCultNetRawDocumentPayload(definition, record)); + decodeDocumentPayload(definition, record.payload); return true; } catch { return false; @@ -5037,106 +4882,36 @@ export class CultMesh { return cultMeshDocumentFromCache(cache, definition, key, options); } - public static documentFromStore( - store: CacheBackingStore, - schemaId: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, - ): CultMeshDocumentHandle; public static documentFromStore( store: CacheBackingStore, definition: TDefinition, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, - ): CultMeshDocumentHandle>; - public static documentFromStore( - store: CacheBackingStore, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, options: { documentId?: string; routeHint?: CultMeshRouteHint; sourceId?: string; pollMs?: number; } = {}, - ): CultMeshDocumentHandle { - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromStore(store, schemaOrDefinition, options) - : cultMeshDocumentFromStore(store, schemaOrDefinition, options); + ): CultMeshDocumentHandle> { + return cultMeshDocumentFromStore(store, definition, options); } - public static documentFromSingleFile( - path: string, - schemaId: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, - ): CultMeshDocumentHandle; public static documentFromSingleFile( path: string, definition: TDefinition, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - pollMs?: number; - }, - ): CultMeshDocumentHandle>; - public static documentFromSingleFile( - path: string, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, options: { documentId?: string; routeHint?: CultMeshRouteHint; sourceId?: string; pollMs?: number; } = {}, - ): CultMeshDocumentHandle { - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromSingleFile(path, schemaOrDefinition, options) - : cultMeshDocumentFromSingleFile(path, schemaOrDefinition, options); + ): CultMeshDocumentHandle> { + return cultMeshDocumentFromSingleFile(path, definition, options); } - public static documentFromPeerSnapshot( - peer: CultNetPeer | (() => CultNetPeer | Promise), - schemaId: string, - recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, - ): CultMeshDocumentHandle; public static documentFromPeerSnapshot( peer: CultNetPeer | (() => CultNetPeer | Promise), definition: TDefinition, recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, - ): CultMeshDocumentHandle>; - public static documentFromPeerSnapshot( - peer: CultNetPeer | (() => CultNetPeer | Promise), - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, - recordKey: string, options: { documentId?: string; routeHint?: CultMeshRouteHint; @@ -5145,42 +4920,14 @@ export class CultMesh { pollMs?: number; messageIdPrefix?: string; } = {}, - ): CultMeshDocumentHandle { - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromPeerSnapshot(peer, schemaOrDefinition, recordKey, options) - : cultMeshDocumentFromPeerSnapshot(peer, schemaOrDefinition, recordKey, options); + ): CultMeshDocumentHandle> { + return cultMeshDocumentFromPeerSnapshot(peer, definition, recordKey, options); } - public static documentFromPublication( - source: CultMeshDocumentPublicationSource, - schemaId: string, - recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, - ): CultMeshDocumentHandle; public static documentFromPublication( source: CultMeshDocumentPublicationSource, definition: TDefinition, recordKey: string, - options?: { - documentId?: string; - routeHint?: CultMeshRouteHint; - sourceId?: string; - timeoutMs?: number; - pollMs?: number; - messageIdPrefix?: string; - }, - ): CultMeshDocumentHandle>; - public static documentFromPublication( - source: CultMeshDocumentPublicationSource, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, - recordKey: string, options: { documentId?: string; routeHint?: CultMeshRouteHint; @@ -5189,10 +4936,8 @@ export class CultMesh { pollMs?: number; messageIdPrefix?: string; } = {}, - ): CultMeshDocumentHandle { - return typeof schemaOrDefinition === "string" - ? cultMeshDocumentFromPublication(source, schemaOrDefinition, recordKey, options) - : cultMeshDocumentFromPublication(source, schemaOrDefinition, recordKey, options); + ): CultMeshDocumentHandle> { + return cultMeshDocumentFromPublication(source, definition, recordKey, options); } public static globalDocumentFromCache( @@ -5254,7 +4999,7 @@ export class CultMesh { } public static publicationDocument( - schema: string | AnyCultCacheDocumentDefinition, + schema: AnyCultCacheDocumentDefinition, recordKey: string, options: { documentId?: string; @@ -6331,7 +6076,7 @@ export class CultMesh { const onMessage = (message: CultNetMessage): void => { if (message.schemaVersion !== "cultnet.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return; try { - const decoded = decodeCultMeshPayload(receiptBinding.definition, new Uint8Array(message.document.payload)); + const decoded = decodeDocumentPayload(receiptBinding.definition, new Uint8Array(message.document.payload)); cleanup(); resolve(receiptBinding.definition.schema.parse(decoded) as CultCacheDocumentValue); } catch (error) { cleanup(); reject(error); } @@ -6597,18 +6342,14 @@ function inferCultMeshSchemaName(schemaId: string | undefined): string | undefin function resolveCultMeshStoreDocumentRecord( records: readonly CultCacheEnvelope[], schema: CultMeshDocumentSchemaDescriptor, - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, + definition: AnyCultCacheDocumentDefinition, ): CultCacheEnvelope | undefined { const schemaId = schema.schemaId ?? schema.type; return ( records.find(candidate => schemaId !== undefined && candidate.schemaId === schemaId) ?? records.find(candidate => - cultMeshEnvelopeMatchesSchema(candidate, schema) && - (typeof schemaOrDefinition === "string" || - cultMeshEnvelopeParsesAsDefinition(candidate, schemaOrDefinition))) ?? - (typeof schemaOrDefinition === "string" - ? undefined - : records.find(candidate => cultMeshEnvelopeParsesAsDefinition(candidate, schemaOrDefinition))) + cultMeshEnvelopeMatchesSchema(candidate, schema) && cultMeshEnvelopeParsesAsDefinition(candidate, definition)) ?? + records.find(candidate => cultMeshEnvelopeParsesAsDefinition(candidate, definition)) ); } @@ -6640,7 +6381,7 @@ function cultMeshEnvelopeParsesAsDefinition( definition: AnyCultCacheDocumentDefinition, ): boolean { try { - definition.schema.parse(decodeCultMeshPayload(definition, envelope.payload)); + decodeDocumentPayload(definition, envelope.payload); return true; } catch { return false; @@ -6675,11 +6416,15 @@ function normalizeRudpDocumentPut( if (document.payloadEncoding !== "messagepack") { throw new Error(`Unsupported raw payload encoding ${document.payloadEncoding}.`); } + const definition = documents.getBySchemaId(document.schemaId)?.definition; + if (!definition) { + throw new Error("Raw document put names a schema with no registered definition."); + } return { schemaId: document.schemaId, recordKey: document.recordKey, storedAt: document.storedAt ?? new Date().toISOString(), - payload: decodeDocumentPayload(documents.getBySchemaId(document.schemaId)?.definition, document.payload), + payload: decodeDocumentPayload(definition, document.payload), sourceRuntimeId: document.sourceRuntimeId ?? null, sourceAgentId: document.sourceAgentId ?? null, sourceRole: document.sourceRole ?? null, @@ -6777,45 +6522,10 @@ async function resolveCultNetPeer( : peerOrProvider; } -// A document payload is read the way its definition's cache reads it (a codec or formatter document is not a plain -// msgpack map); a payload with no definition is plain msgpack. Both are cultcache-ts's decision. -function decodeCultMeshPayload(schemaOrDefinition: string | AnyCultCacheDocumentDefinition, payload: Uint8Array): unknown { - return decodeDocumentPayload(typeof schemaOrDefinition === "string" ? undefined : schemaOrDefinition, payload); -} - -function decodeCultNetRawDocumentPayload( - schemaOrDefinition: string | AnyCultCacheDocumentDefinition, - document: CultNetRawDocumentRecord, -): unknown { - if (document.payloadEncoding !== "messagepack") { - throw new Error(`Unsupported CultNet raw document payload encoding ${document.payloadEncoding}.`); - } - return decodeCultMeshPayload(schemaOrDefinition, toUint8Array(document.payload)); -} - -function parseCultMeshDocumentValue( - schemaOrDefinition: string | TDefinition, - value: unknown, -): unknown | CultCacheDocumentValue { - return typeof schemaOrDefinition === "string" - ? value - : schemaOrDefinition.schema.parse(value); -} - function isCultNetErrorMessage(message: CultNetMessage): message is CultNetErrorMessage { return message.schemaVersion === "cultnet.error.v0"; } -function toUint8Array(value: unknown): Uint8Array { - if (value instanceof Uint8Array) { - return value; - } - if (Array.isArray(value)) { - return Uint8Array.from(value); - } - throw new Error("CultNet raw document payload was not binary."); -} - async function bindRudpSocket(host: string, options: CultMeshRudpSocketOptions): Promise { const socket = createSocket(host.includes(":") ? "udp6" : "udp4"); const port = options.bindPort ?? 0; diff --git a/packages/cultmesh-ts/test/cultmesh.test.ts b/packages/cultmesh-ts/test/cultmesh.test.ts index 4cdce8df..e1c4ed49 100644 --- a/packages/cultmesh-ts/test/cultmesh.test.ts +++ b/packages/cultmesh-ts/test/cultmesh.test.ts @@ -3018,7 +3018,12 @@ test("CultMesh TS reads remote RUDP snapshots through document handles", async ( const alias = CultMesh.documentFromPeerSnapshot( peer, - "cultmesh.note.alias.v1", + defineDocumentType({ + type: "cultmesh.note.alias", + schemaId: "cultmesh.note.alias.v1", + schema: noteDocument.schema, + name: "noteId", + }), "note:remote", { documentId: "note:remote.alias", diff --git a/packages/cultmesh-ts/test/payload-decode-owner.test.ts b/packages/cultmesh-ts/test/payload-decode-owner.test.ts deleted file mode 100644 index f6471bb4..00000000 --- a/packages/cultmesh-ts/test/payload-decode-owner.test.ts +++ /dev/null @@ -1,32 +0,0 @@ -import assert from "node:assert/strict"; -import { existsSync, readdirSync, readFileSync } from "node:fs"; -import { dirname, join } from "node:path"; -import test from "node:test"; - -// A document payload is decoded by cultcache-ts alone (decodeDocumentPayload); every other value on a wire goes through -// cultnet-ts's decodeWireValue / encodeWireValue. So no file of this package imports msgpack, and the manifest does not -// list it, which keeps a new direct decode from arriving unnoticed. -// Limits: a text scan. It reads the package's src files for the module specifier and the manifest for the dependency -// name. It does not follow a re-export from another package, a specifier assembled at run time, or a hand-written -// decoder; the dependency absence is what makes the specifier unresolvable here when the workspace does not hoist it. -// The package root: the nearest directory above this file (compiled or run in place) that holds a package.json. -function packageRoot(from: string): string { - return existsSync(join(from, "package.json")) ? from : packageRoot(dirname(from)); -} -const root = packageRoot(__dirname); - -function sourceFiles(dir: string): string[] { - return readdirSync(dir, { withFileTypes: true }).flatMap((entry) => - entry.isDirectory() ? sourceFiles(join(dir, entry.name)) : entry.name.endsWith(".ts") ? [join(dir, entry.name)] : []); -} - -test("no source file imports msgpack and the manifest does not list it", () => { - const manifest = JSON.parse(readFileSync(join(root, "package.json"), "utf8")) as Record | undefined>; - for (const section of ["dependencies", "devDependencies", "peerDependencies", "optionalDependencies"]) { - assert.equal(manifest[section]?.["@msgpack/msgpack"], undefined, section); - } - const files = sourceFiles(join(root, "src")); - assert.ok(files.length >= 1, "fixture: the scan reads the package's sources"); - const offenders = files.filter((file) => /@msgpack\/msgpack/.test(readFileSync(file, "utf8"))); - assert.deepEqual(offenders, []); -}); diff --git a/packages/cultnet-ts/src/replication.ts b/packages/cultnet-ts/src/replication.ts index fb970906..07652497 100644 --- a/packages/cultnet-ts/src/replication.ts +++ b/packages/cultnet-ts/src/replication.ts @@ -5,7 +5,7 @@ import type { CultCacheDocumentValue, CultCacheEnvelope, } from "@gamecult/cultcache-ts"; -import { resolveDocumentFormatter } from "@gamecult/cultcache-ts"; +import { decodeDocumentPayload, resolveDocumentFormatter } from "@gamecult/cultcache-ts"; import { type CultNetDocumentDeleteMessage, @@ -174,7 +174,7 @@ export class CultNetDocumentRegistry { continue; } - const payload = decodeDocumentValue(binding.definition, envelope.payload); + const payload = decodeDocumentPayload(binding.definition, envelope.payload); documents.push({ schemaId, recordKey: envelope.key, @@ -343,7 +343,7 @@ export class CultNetDocumentRegistry { const payload = new Uint8Array(document.payload); for (const binding of this.#uniqueBindings()) { try { - decodeDocumentValue(binding.definition, payload); + decodeDocumentPayload(binding.definition, payload); return { binding, preserveIncomingSchemaId: false }; } catch { // Keep looking; a foreign schema id is acceptable only when the payload validates locally. @@ -429,12 +429,3 @@ function resolvePayloadSchemaVersion( - definition: TDefinition, - payload: Uint8Array, -): CultCacheDocumentValue { - const formatter = resolveDocumentFormatter(definition); - const decoded = formatter.decode(payload); - return definition.schema.parse(decoded) as CultCacheDocumentValue; -}