From 30934fc7769bc89f280ac44c2c832f155a16621d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 13:36:34 +0100 Subject: [PATCH] Add TypeScript schema catalog sync helpers --- docs/cultnet-transport-parity.md | 5 + docs/runtime-parity-scope.md | 2 +- packages/cultnet-ts/README.md | 8 ++ packages/cultnet-ts/src/index.ts | 3 +- packages/cultnet-ts/src/peer.ts | 70 +++++++++++++ packages/cultnet-ts/src/schema-discovery.ts | 108 ++++++++++++++++++++ packages/cultnet-ts/test/cultnet.test.ts | 61 +++++++++++ 7 files changed, 255 insertions(+), 2 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 5f68e988..70bfb529 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -302,6 +302,11 @@ Current progress: lookup, unsubscribe-able `watch(...)` callbacks for Verse and peer catalogs, watch callbacks for authority leases and stream declarations/frame cursors, plus lease listing and lookup beside the authority check. +- TypeScript now also has a `CultNetSchemaCatalog` for remote schema + descriptors plus `CultNetPeer` request/fetch/sync helpers for + `cultnet.schema_catalog_request.v0` / `cultnet.schema_catalog_response.v0`. + `CultNetSchemaRegistry` remains the local definition owner; the catalog owns + imported descriptors so callers do not treat raw response objects as state. - Kotlin now mirrors the CultMesh stream catalog shape for Android-adjacent media, sensor, tensor, and byte streams: stream declaration, clock metadata, preferred body transports, consumer negotiation, copy-budget classification, diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 12502592..014591ee 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -32,7 +32,7 @@ CultMesh. | --- | --- | --- | --- | | C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | | Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | -| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | +| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, schema catalog request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs with transport fetch/sync helpers, authority leases, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | diff --git a/packages/cultnet-ts/README.md b/packages/cultnet-ts/README.md index af982089..287099fd 100644 --- a/packages/cultnet-ts/README.md +++ b/packages/cultnet-ts/README.md @@ -159,6 +159,7 @@ exchange without resorting to folklore. - `cultnet.schema_catalog_request.v0` - `cultnet.schema_catalog_response.v0` - `CultNetSchemaRegistry` +- `CultNetSchemaCatalog` - `cultNetBuiltinSchemaRegistry` Descriptors include: @@ -175,6 +176,13 @@ The inline schema body travels as canonical JSON text, not as a loose object graph. Discovery is supposed to reduce ambiguity, not smuggle in a cute new polymorphic attack surface. +`CultNetSchemaRegistry` owns local schema definitions and can answer catalog +requests. `CultNetSchemaCatalog` owns remote descriptors and applies catalog +responses. `CultNetPeer.requestSchemaCatalog(...)`, +`fetchSchemaDescriptors(...)`, and `syncSchemaCatalog(...)` perform the common +message-id-matched request/response cycle so callers do not need to hand-roll +listener cleanup around every discovery exchange. + ## Quick Example ```ts diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 9a4eb739..3e82ba02 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -55,7 +55,7 @@ export { type GhostlightWorldState, } from "./contracts"; export { encodeFrame, LengthPrefixedMessageFramer } from "./framing"; -export { CultNetPeer, createTcpFramedCultNetPeer, type CultNetPeerEvents, type TcpFramedCultNetPeerOptions } from "./peer"; +export { CultNetPeer, createTcpFramedCultNetPeer, type CultNetPeerEvents, type CultNetPeerSchemaCatalogRequestOptions, type TcpFramedCultNetPeerOptions } from "./peer"; export { CultNetReconnectController, TcpFramedTransportConnection, @@ -100,6 +100,7 @@ export { } from "./replication"; export { CultNetSchemaRegistry, + CultNetSchemaCatalog, cultNetBuiltinSchemaRegistry, type CultNetSchemaCatalogOptions, type CultNetSchemaRegistration, diff --git a/packages/cultnet-ts/src/peer.ts b/packages/cultnet-ts/src/peer.ts index 4d3f064a..c54984b9 100644 --- a/packages/cultnet-ts/src/peer.ts +++ b/packages/cultnet-ts/src/peer.ts @@ -5,6 +5,7 @@ import { decode, encode } from "@msgpack/msgpack"; import { encodeCultNetMessageForWire, parseCultNetMessage, type CultNetDocumentDeleteMessage, type CultNetDocumentPutMessage, type CultNetErrorMessage, type CultNetHelloMessage, type CultNetLoginMessage, type CultNetLoginSuccessMessage, type CultNetMessage, type CultNetRegisterMessage, type CultNetSampleChangeNameMessage, type CultNetSampleChatMessage, type CultNetSchemaCatalogRequestMessage, type CultNetSchemaCatalogResponseMessage, type CultNetSnapshotRequestMessage, type CultNetSnapshotResponseMessage, type CultNetTransportProfile, type CultNetVerifyMessage, type CultNetWireContract } from "./contracts"; import { encodeFrame, LengthPrefixedMessageFramer } from "./framing"; +import { CultNetSchemaCatalog, type CultNetSchemaCatalogOptions } from "./schema-discovery"; import { createTcpFramedTransportProfile, TcpFramedTransportConnection, type CultNetTransportConnection, type TcpFramedTransportProfileOptions } from "./transport"; export interface CultNetPeerEvents { @@ -22,6 +23,11 @@ export interface TcpFramedCultNetPeerOptions extends CultNetPeerOptions, TcpFram runtimeId: string; } +export interface CultNetPeerSchemaCatalogRequestOptions extends CultNetSchemaCatalogOptions { + messageId?: string; + timeoutMs?: number; +} + export function createTcpFramedCultNetPeer( stream: Duplex, options: TcpFramedCultNetPeerOptions, @@ -150,6 +156,66 @@ export class CultNetPeer extends EventEmitter { this.send(message); } + requestSchemaCatalog( + options: CultNetPeerSchemaCatalogRequestOptions = {}, + ): Promise { + const messageId = options.messageId ?? createMessageId("cultnet-ts-schema-catalog"); + const timeoutMs = options.timeoutMs ?? 4_000; + + return new Promise((resolve, reject) => { + const cleanup = (): void => { + clearTimeout(timer); + this.off("message", onMessage); + this.off("invalidMessage", onInvalidMessage); + this.off("error", onError); + this.off("close", onClose); + }; + const rejectWith = (error: Error): void => { + cleanup(); + reject(error); + }; + const onMessage = (message: CultNetMessage): void => { + if (message.schemaVersion !== "cultnet.schema_catalog_response.v0" || message.messageId !== messageId) { + return; + } + cleanup(); + resolve(message); + }; + const onInvalidMessage = (error: Error): void => rejectWith(error); + const onError = (error: Error): void => rejectWith(error); + const onClose = (): void => rejectWith(new Error("CultNet peer closed before schema catalog response.")); + const timer = setTimeout( + () => rejectWith(new Error(`Timed out waiting for CultNet schema catalog response ${messageId}.`)), + timeoutMs, + ); + + this.on("message", onMessage); + this.on("invalidMessage", onInvalidMessage); + this.on("error", onError); + this.on("close", onClose); + this.sendSchemaCatalogRequest({ + schemaVersion: "cultnet.schema_catalog_request.v0", + messageId, + includeSchemaJson: options.includeSchemaJson, + schemaIds: options.schemaIds !== undefined ? [...options.schemaIds] : undefined, + kinds: options.kinds !== undefined ? [...options.kinds] : undefined, + }); + }); + } + + async fetchSchemaDescriptors( + options: CultNetPeerSchemaCatalogRequestOptions = {}, + ) { + return new CultNetSchemaCatalog().applyResponse(await this.requestSchemaCatalog(options)); + } + + async syncSchemaCatalog( + catalog: CultNetSchemaCatalog, + options: CultNetPeerSchemaCatalogRequestOptions = {}, + ) { + return catalog.applyResponse(await this.requestSchemaCatalog(options)); + } + close(): void { this.#transport?.close(); this.#stream?.end(); @@ -166,6 +232,10 @@ export class CultNetPeer extends EventEmitter { } } +function createMessageId(prefix: string): string { + return `${prefix}-${Date.now().toString(36)}-${Math.random().toString(16).slice(2)}`; +} + function isCultNetTransportConnection(value: Duplex | CultNetTransportConnection): value is CultNetTransportConnection { return typeof (value as CultNetTransportConnection).send === "function" && typeof (value as CultNetTransportConnection).close === "function" diff --git a/packages/cultnet-ts/src/schema-discovery.ts b/packages/cultnet-ts/src/schema-discovery.ts index 103572c6..782c42ee 100644 --- a/packages/cultnet-ts/src/schema-discovery.ts +++ b/packages/cultnet-ts/src/schema-discovery.ts @@ -25,6 +25,89 @@ export interface CultNetSchemaCatalogOptions { kinds?: readonly CultNetSchemaKind[]; } +export class CultNetSchemaCatalog { + readonly #descriptors = new Map(); + readonly #subscribers = new Set<(descriptor: CultNetSchemaDescriptor) => void>(); + + get schemas(): CultNetSchemaDescriptor[] { + return Array.from(this.#descriptors.values()) + .sort((left, right) => left.schemaId.localeCompare(right.schemaId)) + .map((descriptor) => cloneSchemaDescriptor(descriptor, true)); + } + + watch(callback: (descriptor: CultNetSchemaDescriptor) => void): () => void { + this.#subscribers.add(callback); + return () => { + this.#subscribers.delete(callback); + }; + } + + upsert(descriptor: CultNetSchemaDescriptor): CultNetSchemaDescriptor { + if (!descriptor.schemaId) { + throw new Error("CultNet schema descriptor requires schemaId."); + } + if (!descriptor.kind) { + throw new Error("CultNet schema descriptor requires kind."); + } + if (!descriptor.wireContracts.length) { + throw new Error("CultNet schema descriptor requires at least one wire contract."); + } + + const stored = cloneSchemaDescriptor(descriptor, true); + this.#descriptors.set(stored.schemaId, stored); + for (const subscriber of Array.from(this.#subscribers)) { + subscriber(cloneSchemaDescriptor(stored, true)); + } + return cloneSchemaDescriptor(stored, true); + } + + get(schemaId: string, options: CultNetSchemaCatalogOptions = {}): CultNetSchemaDescriptor | undefined { + const descriptor = this.#descriptors.get(schemaId); + if (!descriptor) { + return undefined; + } + return cloneSchemaDescriptor(descriptor, options.includeSchemaJson === true); + } + + list(options: CultNetSchemaCatalogOptions = {}): CultNetSchemaDescriptor[] { + const requestedSchemaIds = options.schemaIds ? new Set(options.schemaIds) : null; + const requestedKinds = options.kinds ? new Set(options.kinds) : null; + + return this.schemas + .filter((descriptor) => { + if (requestedSchemaIds && !requestedSchemaIds.has(descriptor.schemaId)) { + return false; + } + if (requestedKinds && !requestedKinds.has(descriptor.kind)) { + return false; + } + return true; + }) + .map((descriptor) => cloneSchemaDescriptor(descriptor, options.includeSchemaJson === true)); + } + + createCatalogResponse( + request: CultNetSchemaCatalogRequestMessage, + ): CultNetSchemaCatalogResponseMessage { + return { + schemaVersion: "cultnet.schema_catalog_response.v0", + messageId: request.messageId, + schemas: this.list({ + includeSchemaJson: request.includeSchemaJson, + schemaIds: request.schemaIds, + kinds: request.kinds, + }), + }; + } + + applyResponse(response: CultNetSchemaCatalogResponseMessage): CultNetSchemaDescriptor[] { + if (response.schemaVersion !== "cultnet.schema_catalog_response.v0") { + throw new Error(`Expected cultnet.schema_catalog_response.v0, received ${response.schemaVersion}.`); + } + return response.schemas.map((descriptor) => this.upsert(descriptor)); + } +} + interface CultNetRegisteredSchemaRecord extends CultNetSchemaRegistration { canonicalSchemaJson: string; contentHash: string; @@ -97,6 +180,31 @@ export class CultNetSchemaRegistry { } } +function cloneSchemaDescriptor( + descriptor: CultNetSchemaDescriptor, + includeSchemaJson: boolean, +): CultNetSchemaDescriptor { + const clone: CultNetSchemaDescriptor = { + schemaId: descriptor.schemaId, + kind: descriptor.kind, + wireContracts: [...descriptor.wireContracts], + contentHash: descriptor.contentHash, + }; + if (descriptor.schemaVersion !== undefined) { + clone.schemaVersion = descriptor.schemaVersion; + } + if (descriptor.documentType !== undefined) { + clone.documentType = descriptor.documentType; + } + if (descriptor.title !== undefined) { + clone.title = descriptor.title; + } + if (includeSchemaJson && descriptor.schemaJson !== undefined) { + clone.schemaJson = descriptor.schemaJson; + } + return clone; +} + function toSchemaDescriptor( entry: CultNetRegisteredSchemaRecord, includeSchemaJson: boolean, diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 5fe28333..846ffa82 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -21,6 +21,7 @@ import { CultNetRudpReconnectLoop, CultNetRudpSession, CultNetRudpSocketTransportConnection, + CultNetSchemaCatalog, CultNetSchemaRegistry, CultNetSecret, CultNetServerSecurityOptions, @@ -838,6 +839,66 @@ test("CultNet schema discovery catalog can advertise canonical schemas without i ); }); +test("CultNet schema catalog applies remote descriptors and notifies watchers", () => { + const catalog = new CultNetSchemaCatalog(); + let watchedSchemaId: string | undefined; + const unsubscribe = catalog.watch((descriptor) => { + watchedSchemaId = descriptor.schemaId; + }); + const response = cultNetBuiltinSchemaRegistry.createCatalogResponse({ + schemaVersion: "cultnet.schema_catalog_request.v0", + messageId: "catalog-apply", + includeSchemaJson: true, + kinds: ["document_payload"], + }); + + const applied = catalog.applyResponse(response); + + assert.ok(applied.some((descriptor) => descriptor.documentType === "ghostlight.agent-state")); + assert.equal(watchedSchemaId, applied.at(-1)?.schemaId); + assert.equal(catalog.get(applied[0]!.schemaId)?.schemaJson, undefined); + assert.equal(typeof catalog.get(applied[0]!.schemaId, { includeSchemaJson: true })?.schemaJson, "string"); + assert.deepEqual( + catalog.list({ kinds: ["document_payload"] }).map((descriptor) => descriptor.kind), + catalog.list().map((descriptor) => descriptor.kind), + ); + + unsubscribe(); +}); + +test("CultNet peer can request and sync schema catalogs by message id", async () => { + const { a, b } = createDuplexPair(); + const requester = createTcpFramedCultNetPeer(a, { + runtimeId: "catalog-requester", + wireContract: "cultnet.schema.v0", + }); + const responder = createTcpFramedCultNetPeer(b, { + runtimeId: "catalog-responder", + wireContract: "cultnet.schema.v0", + }); + + responder.on("message", (message) => { + if (message.schemaVersion === "cultnet.schema_catalog_request.v0") { + responder.sendSchemaCatalogResponse(cultNetBuiltinSchemaRegistry.createCatalogResponse(message)); + } + }); + + const synced = new CultNetSchemaCatalog(); + const applied = await requester.syncSchemaCatalog(synced, { + messageId: "peer-schema-catalog", + includeSchemaJson: true, + kinds: ["document_payload"], + timeoutMs: 1_000, + }); + + assert.ok(applied.some((descriptor) => descriptor.documentType === "ghostlight.agent-state")); + assert.equal(synced.get(applied[0]!.schemaId)?.schemaJson, undefined); + assert.equal(typeof synced.get(applied[0]!.schemaId, { includeSchemaJson: true })?.schemaJson, "string"); + + requester.close(); + responder.close(); +}); + test("CultNet schema discovery can round-trip over the legacy wire contract when schemas are requested inline", () => { const registry = new CultNetSchemaRegistry([ {