diff --git a/docs/cultmesh-provider-session.md b/docs/cultmesh-provider-session.md new file mode 100644 index 00000000..3e45c799 --- /dev/null +++ b/docs/cultmesh-provider-session.md @@ -0,0 +1,61 @@ +# CultMesh Provider Session + +## Authority map + +- Owner: the provider-session broker owns registration, active lease fencing, + accepted publication membership, command retention, receipt acceptance, and + withdrawal. +- Inputs: a physical CultNet session, an explicit registration-authorization + decision, correlated typed operations, the broker clock, provider commands, + and persistence callbacks. The current RUDP adapter does not itself provide + an authenticated claim. +- Outputs: leases, application acceptances, accepted raw documents, pushed + typed commands, persisted receipts, and deterministic removals. +- Derived state: socket sessions, retry schedules, connection health, and + provider projections are observations. They do not authorize publications. +- Forbidden writers: RUDP packet acknowledgements, source addresses, direct raw + document callbacks, renderer-owned catalogs, and arrival-time freshness maps + cannot create accepted provider truth. +- Shared paths: initial connect and reconnect use `provider.register`; initial, + replayed, and live documents use `provider.publication.put`; explicit + withdrawal and lease expiry remove the broker-owned publication set; command + retry ends only when `provider.receipt.put` is accepted. +- Cut line: a receiver must stop accepting unleased provider raw puts before it + claims this protocol as its ingress authority. + +## Wire contract + +Provider lifecycle operations use the reliable ordered CultNet schema channel +and the existing `cultnet.operation_request.v0` / `cultnet.operation_response.v0` +envelopes with service id `gamecult.mesh.provider_session`. Inner payloads are +camel-case MessagePack maps encoded as `messagepack-base64`. + +The v1 operations are: + +| Operation | Request payload | Successful payload | +| --- | --- | --- | +| `provider.register` | `gamecult.mesh.provider_registration.v1` | `gamecult.mesh.provider_lease.v1` | +| `provider.renew` | `gamecult.mesh.provider_lease_renewal.v1` | `gamecult.mesh.provider_lease.v1` | +| `provider.publication.put` | `gamecult.mesh.provider_publication_put.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` | +| `provider.publication.delete` | `gamecult.mesh.provider_publication_delete.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` | +| `provider.receipt.put` | `gamecult.mesh.provider_receipt_put.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` | +| `provider.withdraw` | `gamecult.mesh.provider_withdrawal.v1` | `gamecult.mesh.provider_mutation_acceptance.v1` | + +Broker-to-provider commands are raw typed documents using +`gamecult.mesh.provider_command.v1`. Asset bodies do not travel inside these +operation payloads. + +Responses use `ok`, `conflict`, `expired`, `denied`, or `invalid`. A transport +ACK is never an application acceptance. Renewal issues a new lease id and +immediately fences the previous id. The broker derives the set removed by full +withdrawal; a provider cannot use a supplied list to hide owned publications. + +Public deployment additionally requires a CultNet authentication claim that +authorizes the provider identity. Until that claim is wired and checked, the +TypeScript RUDP adapter is a private-development transport. + +`CultMeshProviderSessionBroker` is an in-memory reference broker and socket +proof. Its command and receipt retention does not survive process restart. +Production receivers must persist that control-plane state atomically with +accepted publication ownership; Odin must not substitute these volatile maps +for its durable broker authority. diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 2409898e..92e3c49d 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -271,7 +271,16 @@ publication, and withdrawal. Provider code owns domain state and idempotent command transactions; the injected transport owns CultNet wire details. ```ts -import { CultMeshProviderSession } from "cultmesh-ts"; +import { + CultMeshProviderRudpTransport, + CultMeshProviderSession, +} from "cultmesh-ts"; + +const odinTransport = new CultMeshProviderRudpTransport({ + endpoint: "rudp://127.0.0.1:17871", + runtimeId: "voidbot-worker-7", + connectionId: 0x43554c54, +}); const session = new CultMeshProviderSession({ identity: { @@ -300,6 +309,19 @@ is persisted before transport publication, remains pending across reconnects, and is marked published only after the current connection accepts it. Receipt store failure degrades command intake without inventing a network outage. +The RUDP provider transport carries lifecycle payloads inside the existing +`cultnet.operation_request.v0` and `cultnet.operation_response.v0` envelopes. +RUDP acknowledgements prove byte delivery only. Registration, renewal, +publication changes, receipts, and withdrawal complete only after the +provider-session broker returns a correlated application response. The broker +owns lease fencing and accepted publication membership; the provider owns its +desired documents and durable receipt outbox. + +`CultMeshProviderRudpTransport` is currently a private-development transport. +Do not advertise it as a public provider boundary until the surrounding +CultNet session authenticates a claim authorizing the four-part provider +identity. Source address and successful contact are not authority. + ## RUDP Helpers Node runtimes can build the shared CultNet reliable-UDP transport from the diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index eca3222d..8c243a77 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -25,6 +25,8 @@ import { type CultNetErrorMessage, type CultNetDocumentBinding, type CultNetMessage, + type CultNetOperationRequestMessage, + type CultNetOperationResponseMessage, type CultNetRawDocumentRecord, type CultNetReconnectPolicy, type CultNetSchemaCatalogOptions, @@ -33,6 +35,9 @@ import { } from "cultnet-ts"; export * from "./provider-session"; +export * from "./provider-session-wire"; +export * from "./provider-rudp-transport"; +export * from "./provider-session-broker"; export interface CultMeshVec2 { readonly x: number; @@ -4095,10 +4100,21 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt getCache?: () => Promise | CultCache; onError?: (error: Error) => void; onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) => void | Promise; + onOperationRequest?: ( + request: CultNetOperationRequestMessage, + session: CultMeshRudpServerSession, + ) => CultNetOperationResponseMessage | Promise; + onSessionClosed?: (session: CultMeshRudpServerSession) => void | Promise; wireContract?: CultNetWireContract; sessionTimeoutMs?: number; } +export interface CultMeshRudpServerSession { + readonly sessionId: string; + readonly remote: { address: string; family: string; port: number }; + send(message: CultNetMessage): void; +} + export interface CultMeshRudpDocumentServer { readonly bind: { host: string; port: number }; start(): Promise; @@ -5542,9 +5558,12 @@ export class CultMesh { const bind = { host, port }; const socket = options.socket ?? createSocket(host.includes(":") ? "udp6" : "udp4"); const sessions = new Map; }>(); + let sessionSequence = 0; const resendPollMs = Math.max(10, options.resendPollMs ?? 25); const sessionTimeoutMs = Math.max(1_000, options.sessionTimeoutMs ?? 30_000); const wireContract = options.wireContract ?? "cultnet.schema.v0"; @@ -5567,7 +5586,9 @@ export class CultMesh { let record = sessions.get(key); if (packet.packetType === "connect") { + if (record) notifySessionClosed(record); record = { + sessionId: `${runtimeId}:${++sessionSequence}`, remote: { address: remote.address, family: remote.family, port: remote.port }, session: new CultNetRudpSession({ connectionId, @@ -5575,6 +5596,7 @@ export class CultMesh { resendDelayMs: options.resendDelayMs, maxPendingReliablePackets: options.maxPendingReliablePackets, }), + work: Promise.resolve(), }; sessions.set(key, record); socket.send(encodeRudpPacket(record.session.acceptConnect(packet, nowMs)), remote.port, remote.address); @@ -5593,12 +5615,13 @@ export class CultMesh { if (frame.channelId !== "schema") { continue; } - handleDocumentServerFrame(record, frame.payload).catch((error) => { - reportError(error); - }); + record.work = record.work + .then(() => handleDocumentServerFrame(record, frame.payload)) + .catch(reportError); } if (result.disconnected) { sessions.delete(key); + notifySessionClosed(record); return; } if (packet.packetType === "accept" || result.delivered.length > 0) { @@ -5611,7 +5634,7 @@ export class CultMesh { socket.on("error", reportError); async function handleDocumentServerFrame( - record: { remote: { address: string; family: string; port: number }; session: CultNetRudpSession }, + record: { sessionId: string; remote: { address: string; family: string; port: number }; session: CultNetRudpSession }, payload: Uint8Array, ): Promise { const message = parseCultNetMessage(decode(payload), wireContract); @@ -5643,6 +5666,16 @@ export class CultMesh { await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote)); } return; + case "cultnet.operation_request.v0": + if (!options.onOperationRequest) { + sendSchemaMessage(record, { + schemaVersion: "cultnet.error.v0", + error: "CultMesh RUDP document server has no operation handler.", + }); + return; + } + sendSchemaMessage(record, await options.onOperationRequest(message, publicSession(record))); + return; default: sendSchemaMessage(record, { schemaVersion: "cultnet.error.v0", @@ -5651,6 +5684,27 @@ export class CultMesh { } } + function publicSession(record: { + sessionId: string; + remote: { address: string; family: string; port: number }; + session: CultNetRudpSession; + }): CultMeshRudpServerSession { + return { + sessionId: record.sessionId, + remote: { ...record.remote }, + send: message => sendSchemaMessage(record, message), + }; + } + + function notifySessionClosed(record: { + sessionId: string; + remote: { address: string; family: string; port: number }; + session: CultNetRudpSession; + }): void { + if (!options.onSessionClosed) return; + void Promise.resolve(options.onSessionClosed(publicSession(record))).catch(reportError); + } + function sendSchemaMessage( record: { remote: { address: string; port: number }; session: CultNetRudpSession }, message: CultNetMessage, @@ -5684,6 +5738,7 @@ export class CultMesh { for (const [key, record] of sessions) { if (record.session.checkTimeout(nowMs, sessionTimeoutMs)) { sessions.delete(key); + notifySessionClosed(record); continue; } for (const packet of record.session.dueResends(nowMs)) { @@ -5698,6 +5753,7 @@ export class CultMesh { clearInterval(resendTimer); resendTimer = undefined; } + for (const record of sessions.values()) notifySessionClosed(record); sessions.clear(); socket.close(); }, diff --git a/packages/cultmesh-ts/src/provider-rudp-transport.ts b/packages/cultmesh-ts/src/provider-rudp-transport.ts new file mode 100644 index 00000000..bb10c6ec --- /dev/null +++ b/packages/cultmesh-ts/src/provider-rudp-transport.ts @@ -0,0 +1,390 @@ +import { randomUUID } from "node:crypto"; +import { createSocket, type Socket } from "node:dgram"; +import { encode } from "@msgpack/msgpack"; +import { + CultNetPeer, + CultNetRudpSocketTransportConnection, + type CultNetMessage, + type CultNetOperationRequestMessage, + type CultNetOperationResponseMessage, + type CultNetRawDocumentRecord, +} from "cultnet-ts"; + +import type { + CultMeshProviderCommand, + CultMeshProviderCommandListener, + CultMeshProviderCommandReceipt, + CultMeshProviderConnection, + CultMeshProviderIdentity, + CultMeshProviderLease, + CultMeshProviderPublication, + CultMeshProviderRegistration, + CultMeshProviderTransport, + CultMeshProviderUnsubscribe, + CultMeshProviderWithdrawal, +} from "./provider-session"; +import { + CULTMESH_PROVIDER_SESSION_SERVICE_ID, + assertMutationAcceptance, + assertProviderLease, + cultMeshProviderSessionOperations, + cultMeshProviderSessionSchemas, + decodeProviderCommandDocument, + decodeProviderSessionPayload, + encodeProviderSessionPayload, + type CultMeshProviderCommandReceiptWire, + type CultMeshProviderLeaseWire, + type CultMeshProviderMutationAcceptanceWire, +} from "./provider-session-wire"; + +export interface CultMeshProviderRudpTransportOptions { + readonly endpoint: string; + readonly runtimeId: string; + readonly connectionId: number; + readonly bindHost?: string; + readonly connectTimeoutMs?: number; + readonly operationTimeoutMs?: number; + readonly maxFragmentBytes?: number; + readonly socketFactory?: () => Socket; +} + +/** + * Private-development provider transport. Public deployment additionally needs + * an authenticated CultNet session whose claims authorize the provider identity. + */ +export class CultMeshProviderRudpTransport implements CultMeshProviderTransport { + readonly #options: CultMeshProviderRudpTransportOptions; + + public constructor(options: CultMeshProviderRudpTransportOptions) { + if (!options.endpoint) throw new Error("CultMesh provider RUDP transport requires endpoint."); + if (!options.runtimeId) throw new Error("CultMesh provider RUDP transport requires runtimeId."); + if (!Number.isInteger(options.connectionId)) throw new Error("CultMesh provider RUDP transport requires integer connectionId."); + this.#options = options; + } + + public async connect( + identity: CultMeshProviderIdentity, + signal: AbortSignal, + ): Promise { + if (signal.aborted) throw abortError(); + const endpoint = parseEndpoint(this.#options.endpoint); + const socket = this.#options.socketFactory?.() ?? createSocket(endpoint.host.includes(":") ? "udp6" : "udp4"); + await bindSocket(socket, this.#options.bindHost ?? (endpoint.host.includes(":") ? "::" : "0.0.0.0"), signal); + const transport = new CultNetRudpSocketTransportConnection({ + mode: "client", + runtimeId: this.#options.runtimeId, + transportId: `${this.#options.runtimeId}.cultmesh-provider-session`, + socket, + remoteHost: endpoint.host, + remotePort: endpoint.port, + connectionId: this.#options.connectionId, + maxFragmentBytes: this.#options.maxFragmentBytes ?? 2_048, + }); + const peer = new CultNetPeer(transport, { wireContract: "cultnet.schema.v0" }); + transport.connect(); + try { + await waitUntilConnected(transport, signal, this.#options.connectTimeoutMs ?? 2_000); + return new RudpProviderConnection(identity, peer, transport, this.#options.operationTimeoutMs ?? 5_000); + } catch (error) { + transport.close(); + throw error; + } + } +} + +class RudpProviderConnection implements CultMeshProviderConnection { + readonly #identity: CultMeshProviderIdentity; + readonly #peer: CultNetPeer; + readonly #transport: CultNetRudpSocketTransportConnection; + readonly #timeoutMs: number; + readonly #listeners = new Set(); + readonly #pending = new Map void; + reject: (error: Error) => void; + timer: NodeJS.Timeout; + }>(); + readonly #publications = new Map(); + readonly #bufferedCommands = new Map(); + #closed = false; + #requestedLeaseDurationMs = 30_000; + + public constructor( + identity: CultMeshProviderIdentity, + peer: CultNetPeer, + transport: CultNetRudpSocketTransportConnection, + timeoutMs: number, + ) { + this.#identity = identity; + this.#peer = peer; + this.#transport = transport; + this.#timeoutMs = timeoutMs; + peer.on("message", message => this.#onMessage(message)); + peer.on("invalidMessage", error => this.#failPending(error)); + peer.on("close", () => this.#failPending(new Error("CultMesh provider RUDP connection closed."))); + peer.on("error", error => this.#failPending(error)); + } + + public async register( + registration: CultMeshProviderRegistration, + signal: AbortSignal, + ): Promise { + this.#requestedLeaseDurationMs = registration.requestedLeaseDurationMs; + const response = await this.#operation( + cultMeshProviderSessionOperations.register, + cultMeshProviderSessionSchemas.registration, + cultMeshProviderSessionSchemas.lease, + { ...registration.identity, requestedLeaseDurationMs: registration.requestedLeaseDurationMs }, + signal, + ); + const wire = decodeProviderSessionPayload(response.payload); + assertProviderLease(wire); + assertIdentity(this.#identity, wire); + return { leaseId: wire.leaseId, expiresAt: new Date(wire.expiresAtUtc) }; + } + + public async renew(lease: CultMeshProviderLease): Promise { + const response = await this.#operation( + cultMeshProviderSessionOperations.renew, + cultMeshProviderSessionSchemas.leaseRenewal, + cultMeshProviderSessionSchemas.lease, + { leaseId: lease.leaseId, requestedLeaseDurationMs: this.#requestedLeaseDurationMs }, + ); + const wire = decodeProviderSessionPayload(response.payload); + assertProviderLease(wire); + assertIdentity(this.#identity, wire); + return { leaseId: wire.leaseId, expiresAt: new Date(wire.expiresAtUtc) }; + } + + public async publish( + publication: CultMeshProviderPublication, + lease: CultMeshProviderLease, + ): Promise { + const document: CultNetRawDocumentRecord = { + schemaId: publication.schemaId, + recordKey: publication.recordKey, + storedAt: new Date().toISOString(), + payloadEncoding: "messagepack", + payload: encode(publication.value), + sourceRuntimeId: this.#identity.serviceInstanceId, + sourceAgentId: this.#identity.providerId, + sourceRole: "provider", + }; + await this.#accepted(cultMeshProviderSessionOperations.publicationPut, cultMeshProviderSessionSchemas.publicationPut, { + leaseId: lease.leaseId, + publicationId: publication.publicationId, + document, + }); + this.#publications.set(publication.publicationId, publication); + } + + public async withdrawPublication(publicationId: string, lease: CultMeshProviderLease): Promise { + const publication = this.#publications.get(publicationId); + if (!publication) throw new Error(`CultMesh provider publication ${publicationId} is not owned by this connection.`); + await this.#accepted(cultMeshProviderSessionOperations.publicationDelete, cultMeshProviderSessionSchemas.publicationDelete, { + leaseId: lease.leaseId, + publicationId, + schemaId: publication.schemaId, + recordKey: publication.recordKey, + }); + this.#publications.delete(publicationId); + } + + public watchCommands(listener: CultMeshProviderCommandListener): CultMeshProviderUnsubscribe { + this.#listeners.add(listener); + for (const [commandId, command] of this.#bufferedCommands) { + this.#bufferedCommands.delete(commandId); + void Promise.resolve(listener(command)).catch(() => undefined); + } + return () => { + this.#listeners.delete(listener); + }; + } + + public async publishReceipt( + receipt: CultMeshProviderCommandReceipt, + lease: CultMeshProviderLease, + ): Promise { + const wire: CultMeshProviderCommandReceiptWire = { + receiptId: receipt.receiptId, + commandId: receipt.commandId, + commandKind: receipt.commandKind, + providerId: receipt.providerId, + serviceInstanceId: receipt.serviceInstanceId, + state: receipt.state, + completedAtUtc: receipt.completedAt.toISOString(), + result: receipt.result, + error: receipt.error, + }; + await this.#accepted(cultMeshProviderSessionOperations.receiptPut, cultMeshProviderSessionSchemas.receiptPut, { + leaseId: lease.leaseId, + receipt: wire, + }); + } + + public async withdraw(withdrawal: CultMeshProviderWithdrawal): Promise { + await this.#accepted(cultMeshProviderSessionOperations.withdraw, cultMeshProviderSessionSchemas.withdrawal, { + leaseId: withdrawal.leaseId, + }); + this.#publications.clear(); + } + + public close(): void { + if (this.#closed) return; + this.#closed = true; + this.#failPending(new Error("CultMesh provider RUDP connection closed.")); + this.#listeners.clear(); + this.#transport.close(); + } + + async #accepted(operation: string, payloadSchema: string, payload: unknown): Promise { + const response = await this.#operation(operation, payloadSchema, cultMeshProviderSessionSchemas.mutationAcceptance, payload); + const acceptance = decodeProviderSessionPayload(response.payload); + assertMutationAcceptance(acceptance); + return acceptance; + } + + async #operation( + operation: string, + payloadSchema: string, + responsePayloadSchema: string, + payload: unknown, + signal?: AbortSignal, + ): Promise { + if (this.#closed) throw new Error("CultMesh provider RUDP connection is closed."); + if (signal?.aborted) throw abortError(); + const messageId = randomUUID(); + const request: CultNetOperationRequestMessage = { + schemaVersion: "cultnet.operation_request.v0", + messageId, + serviceId: CULTMESH_PROVIDER_SESSION_SERVICE_ID, + operation, + payloadSchema, + payloadEncoding: "messagepack-base64", + payload: encodeProviderSessionPayload(payload), + sourceRuntimeId: this.#identity.serviceInstanceId, + }; + return await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.#pending.delete(messageId); + reject(new Error(`CultMesh provider operation ${operation} timed out.`)); + }, this.#timeoutMs); + const abort = (): void => { + clearTimeout(timer); + this.#pending.delete(messageId); + reject(abortError()); + }; + signal?.addEventListener("abort", abort, { once: true }); + this.#pending.set(messageId, { + operation, + payloadSchema: responsePayloadSchema, + timer, + resolve: response => { + signal?.removeEventListener("abort", abort); + if (response.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID || response.operation !== operation) { + reject(new Error(`CultMesh provider response correlation does not match ${operation}.`)); + return; + } + if (response.payloadEncoding !== "messagepack-base64") { + reject(new Error(`CultMesh provider response for ${operation} has incompatible payload encoding.`)); + return; + } + if (response.status !== "ok") { + reject(new Error(`CultMesh provider operation ${operation} was ${response.status}: ${(response.diagnostics ?? []).join("; ")}`)); + return; + } + if (response.payloadSchema !== responsePayloadSchema) { + reject(new Error(`CultMesh provider response for ${operation} has incompatible payload schema.`)); + return; + } + resolve(response); + }, + reject: error => { + signal?.removeEventListener("abort", abort); + reject(error); + }, + }); + this.#peer.send(request); + }); + } + + #onMessage(message: CultNetMessage): void { + if (message.schemaVersion === "cultnet.operation_response.v0") { + const pending = this.#pending.get(message.messageId); + if (!pending) return; + this.#pending.delete(message.messageId); + clearTimeout(pending.timer); + pending.resolve(message); + return; + } + if (message.schemaVersion !== "cultnet.document_put_raw.v0") return; + if (message.document.schemaId !== cultMeshProviderSessionSchemas.command) return; + const wire = decodeProviderCommandDocument(message.document); + if (wire.providerId !== this.#identity.providerId || wire.serviceInstanceId !== this.#identity.serviceInstanceId) return; + const command: CultMeshProviderCommand = { ...wire }; + if (this.#listeners.size === 0) { + this.#bufferedCommands.set(command.commandId, command); + return; + } + for (const listener of [...this.#listeners]) void Promise.resolve(listener(command)).catch(() => undefined); + } + + #failPending(error: Error): void { + for (const [messageId, pending] of this.#pending) { + this.#pending.delete(messageId); + clearTimeout(pending.timer); + pending.reject(error); + } + } +} + +function assertIdentity(identity: CultMeshProviderIdentity, lease: CultMeshProviderLeaseWire): void { + for (const key of ["providerId", "serviceInstanceId", "endpointId", "verseId"] as const) { + if (identity[key] !== lease[key]) throw new Error(`CultMesh provider lease ${key} does not match the registered identity.`); + } +} + +function parseEndpoint(endpoint: string): { host: string; port: number } { + const url = new URL(endpoint); + const port = Number(url.port); + if (url.protocol !== "rudp:" || !url.hostname || !Number.isInteger(port) || port <= 0 || port > 65_535) { + throw new Error(`Invalid CultMesh provider RUDP endpoint ${endpoint}.`); + } + return { host: url.hostname, port }; +} + +function bindSocket(socket: Socket, host: string, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const abort = (): void => { + socket.close(); + reject(abortError()); + }; + signal.addEventListener("abort", abort, { once: true }); + socket.once("error", reject); + socket.bind(0, host, () => { + signal.removeEventListener("abort", abort); + socket.removeListener("error", reject); + resolve(); + }); + }); +} + +async function waitUntilConnected( + transport: CultNetRudpSocketTransportConnection, + signal: AbortSignal, + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!transport.connected) { + if (signal.aborted) throw abortError(); + if (Date.now() >= deadline) throw new Error("CultMesh provider RUDP connection timed out."); + await new Promise(resolve => setTimeout(resolve, 5)); + } +} + +function abortError(): Error { + const error = new Error("CultMesh provider RUDP operation aborted."); + error.name = "AbortError"; + return error; +} diff --git a/packages/cultmesh-ts/src/provider-session-broker.ts b/packages/cultmesh-ts/src/provider-session-broker.ts new file mode 100644 index 00000000..a8ddc3d5 --- /dev/null +++ b/packages/cultmesh-ts/src/provider-session-broker.ts @@ -0,0 +1,499 @@ +import { encode } from "@msgpack/msgpack"; +import type { + CultNetDocumentPutRawMessage, + CultNetOperationRequestMessage, + CultNetOperationResponseMessage, + CultNetRawDocumentRecord, +} from "cultnet-ts"; + +import type { CultMeshRudpServerSession } from "./index"; +import { + CULTMESH_PROVIDER_SESSION_SERVICE_ID, + cultMeshProviderSessionOperations, + cultMeshProviderSessionSchemas, + decodeProviderSessionPayload, + encodeProviderSessionPayload, + type CultMeshProviderCommandReceiptWire, + type CultMeshProviderCommandWire, + type CultMeshProviderLeaseRenewalWire, + type CultMeshProviderLeaseWire, + type CultMeshProviderOperationStatus, + type CultMeshProviderPublicationDeleteWire, + type CultMeshProviderPublicationPutWire, + type CultMeshProviderReceiptPutWire, + type CultMeshProviderRegistrationWire, + type CultMeshProviderWithdrawalWire, +} from "./provider-session-wire"; + +export interface CultMeshProviderBrokerClock { + now(): Date; +} + +export interface CultMeshProviderSessionBrokerOptions { + readonly runtimeId: string; + readonly clock?: CultMeshProviderBrokerClock; + readonly createLeaseId?: () => string; + readonly maximumLeaseDurationMs?: number; + readonly expiryPollMs?: number; + readonly onError?: (error: Error) => void; + readonly authorizeRegistration: ( + identity: CultMeshProviderRegistrationWire, + session: CultMeshRudpServerSession, + ) => boolean | Promise; + readonly acceptPublication: ( + identity: CultMeshProviderRegistrationWire, + publicationId: string, + document: CultNetRawDocumentRecord, + ) => void | Promise; + readonly deletePublications: ( + identity: CultMeshProviderRegistrationWire, + publications: readonly { + publicationId: string; + document: Pick; + }[], + ) => void | Promise; + readonly acceptReceipt: ( + identity: CultMeshProviderRegistrationWire, + receipt: CultMeshProviderCommandReceiptWire, + ) => void | Promise; +} + +interface LeaseRecord { + identity: CultMeshProviderRegistrationWire; + lease: CultMeshProviderLeaseWire; + session?: CultMeshRudpServerSession; + publications: Map; +} + +export class CultMeshProviderSessionBroker { + readonly #options: CultMeshProviderSessionBrokerOptions; + readonly #clock: CultMeshProviderBrokerClock; + readonly #leases = new Map(); + readonly #identityLeases = new Map(); + readonly #sessionLeases = new Map>(); + readonly #commands = new Map(); + readonly #receipts = new Map(); + readonly #publicationOwners = new Map(); + readonly #expiryTimer: NodeJS.Timeout; + #work: Promise = Promise.resolve(); + #leaseSequence = 0; + + public constructor(options: CultMeshProviderSessionBrokerOptions) { + if (!options.runtimeId) throw new Error("CultMesh provider-session broker requires runtimeId."); + if (!options.authorizeRegistration) throw new Error("CultMesh provider-session broker requires an explicit registration authority."); + this.#options = options; + this.#clock = options.clock ?? { now: () => new Date() }; + this.#expiryTimer = setInterval(() => { + void this.expireLeases().catch(error => { + const normalized = error instanceof Error ? error : new Error(String(error)); + if (this.#options.onError) this.#options.onError(normalized); + else console.error(`CultMesh provider-session expiry failed: ${normalized.message}`); + }); + }, Math.max(10, options.expiryPollMs ?? 1_000)); + this.#expiryTimer.unref?.(); + } + + public close(): void { + clearInterval(this.#expiryTimer); + } + + public async handle( + request: CultNetOperationRequestMessage, + session: CultMeshRudpServerSession, + ): Promise { + const work = this.#work.then(() => this.#handle(request, session)); + this.#work = work.then(() => undefined, () => undefined); + return await work; + } + + async #handle( + request: CultNetOperationRequestMessage, + session: CultMeshRudpServerSession, + ): Promise { + if (request.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Unsupported service ${request.serviceId}.`, + ]); + } + if (request.payloadEncoding !== "messagepack-base64") { + return this.#response(request, "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + "Provider-session operation payloads must use messagepack-base64.", + ]); + } + try { + switch (request.operation) { + case cultMeshProviderSessionOperations.register: + return await this.#register(request, session); + case cultMeshProviderSessionOperations.renew: + return await this.#renew(request, session); + case cultMeshProviderSessionOperations.publicationPut: + return await this.#publicationPut(request, session); + case cultMeshProviderSessionOperations.publicationDelete: + return await this.#publicationDelete(request, session); + case cultMeshProviderSessionOperations.receiptPut: + return await this.#receiptPut(request, session); + case cultMeshProviderSessionOperations.withdraw: + return await this.#withdraw(request, session); + default: + return this.#response(request, "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Unsupported provider-session operation ${request.operation}.`, + ]); + } + } catch (error) { + if (error instanceof BrokerInfrastructureError) throw error.cause; + return this.#response(request, error instanceof BrokerStatusError ? error.status : "invalid", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + error instanceof Error ? error.message : String(error), + ]); + } + } + + public sessionClosed(session: CultMeshRudpServerSession): void { + for (const leaseId of this.#sessionLeases.get(session.sessionId) ?? []) { + const record = this.#leases.get(leaseId); + if (record?.session?.sessionId === session.sessionId) record.session = undefined; + } + this.#sessionLeases.delete(session.sessionId); + } + + public enqueueCommand(command: CultMeshProviderCommandWire): void { + requireText(command.commandId, "commandId"); + requireText(command.providerId, "providerId"); + requireText(command.serviceInstanceId, "serviceInstanceId"); + const completed = this.#receipts.get(command.commandId); + if (completed) { + if (completed.commandKind !== command.commandKind || completed.providerId !== command.providerId || completed.serviceInstanceId !== command.serviceInstanceId) { + throw new Error(`Command id ${command.commandId} is already completed for another command transaction.`); + } + return; + } + const existing = this.#commands.get(command.commandId); + if (existing && !wireEquivalent(existing, command)) { + throw new Error(`Command id ${command.commandId} is already retained for another command transaction.`); + } + this.#commands.set(command.commandId, existing ?? command); + for (const record of this.#leases.values()) { + if (!this.#isActive(record)) continue; + if (record.identity.providerId !== command.providerId || record.identity.serviceInstanceId !== command.serviceInstanceId) continue; + this.#sendCommand(record, command); + } + } + + public get activeLeaseCount(): number { + return [...this.#leases.values()].filter(record => this.#isActive(record)).length; + } + + public async expireLeases(): Promise { + const work = this.#work.then(() => this.#expireLeases()); + this.#work = work.then(() => undefined, () => undefined); + await work; + } + + async #expireLeases(): Promise { + const now = this.#clock.now().getTime(); + for (const [leaseId, record] of [...this.#leases]) { + if (Date.parse(record.lease.expiresAtUtc) > now) continue; + await this.#options.deletePublications(record.identity, publicationEntries(record)); + this.#releasePublicationOwners(record); + this.#removeLease(leaseId, record); + } + } + + async #register(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.registration); + const identity = decodeProviderSessionPayload(request.payload); + validateRegistration(identity); + if (!await infrastructure(() => this.#options.authorizeRegistration(identity, session))) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + "The CultNet session does not authorize this provider identity.", + ]); + } + const identityKey = identityTuple(identity); + const previousLeaseId = this.#identityLeases.get(identityKey); + let publications = new Map(); + if (previousLeaseId) { + const previous = this.#leases.get(previousLeaseId); + if (previous) { + publications = previous.publications; + this.#removeLease(previousLeaseId, previous); + } + } + const lease = this.#createLease(identity, identity.requestedLeaseDurationMs); + const record: LeaseRecord = { identity, lease, session, publications }; + this.#leases.set(lease.leaseId, record); + this.#identityLeases.set(identityKey, lease.leaseId); + this.#attachSession(session, lease.leaseId); + this.#replayCommands(record); + return this.#response(request, "ok", cultMeshProviderSessionSchemas.lease, lease); + } + + async #renew(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.leaseRenewal); + const renewal = decodeProviderSessionPayload(request.payload); + const current = this.#requireLease(renewal.leaseId, session); + const lease = this.#createLease(current.identity, renewal.requestedLeaseDurationMs); + const replacement: LeaseRecord = { ...current, lease, session }; + this.#removeLease(renewal.leaseId, current); + this.#leases.set(lease.leaseId, replacement); + this.#identityLeases.set(identityTuple(current.identity), lease.leaseId); + this.#attachSession(session, lease.leaseId); + return this.#response(request, "ok", cultMeshProviderSessionSchemas.lease, lease); + } + + async #publicationPut(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.publicationPut); + const mutation = decodeProviderSessionPayload(request.payload); + const record = this.#requireLease(mutation.leaseId, session); + requireText(mutation.publicationId, "publicationId"); + validateDocument(mutation.document); + const previous = record.publications.get(mutation.publicationId); + if (previous && (previous.schemaId !== mutation.document.schemaId || previous.recordKey !== mutation.document.recordKey)) { + return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Publication ${mutation.publicationId} already owns ${previous.schemaId}:${previous.recordKey}. Delete it before changing its tuple.`, + ]); + } + const tuple = documentTuple(mutation.document); + const owner = this.#publicationOwners.get(tuple); + const ownerIdentity = identityTuple(record.identity); + if (owner && (owner.identityKey !== ownerIdentity || owner.publicationId !== mutation.publicationId)) { + return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Publication tuple ${mutation.document.schemaId}:${mutation.document.recordKey} is already owned.`, + ]); + } + const normalizedDocument: CultNetRawDocumentRecord = { + ...mutation.document, + sourceRuntimeId: record.identity.serviceInstanceId, + sourceAgentId: record.identity.providerId, + sourceRole: "provider", + }; + await infrastructure(() => this.#options.acceptPublication(record.identity, mutation.publicationId, normalizedDocument)); + record.publications.set(mutation.publicationId, normalizedDocument); + this.#publicationOwners.set(tuple, { identityKey: ownerIdentity, publicationId: mutation.publicationId }); + return this.#accepted(request, record.lease.leaseId, { publicationId: mutation.publicationId }); + } + + async #publicationDelete(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.publicationDelete); + const mutation = decodeProviderSessionPayload(request.payload); + const record = this.#requireLease(mutation.leaseId, session); + const owned = record.publications.get(mutation.publicationId); + if (!owned || owned.schemaId !== mutation.schemaId || owned.recordKey !== mutation.recordKey) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Publication ${mutation.publicationId} is not owned by this lease with the supplied schema and key.`, + ]); + } + await infrastructure(() => this.#options.deletePublications(record.identity, [{ publicationId: mutation.publicationId, document: owned }])); + record.publications.delete(mutation.publicationId); + this.#publicationOwners.delete(documentTuple(owned)); + return this.#accepted(request, record.lease.leaseId, { publicationId: mutation.publicationId }); + } + + async #receiptPut(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.receiptPut); + const mutation = decodeProviderSessionPayload(request.payload); + const record = this.#requireLease(mutation.leaseId, session); + const receipt = mutation.receipt; + validateReceipt(receipt); + if (receipt.providerId !== record.identity.providerId || receipt.serviceInstanceId !== record.identity.serviceInstanceId) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, ["Receipt identity does not match its lease."]); + } + const previousReceipt = this.#receipts.get(receipt.commandId); + if (previousReceipt) { + if (!wireEquivalent(previousReceipt, receipt)) { + return this.#response(request, "conflict", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Command ${receipt.commandId} already has a different accepted receipt.`, + ]); + } + return this.#accepted(request, record.lease.leaseId, { commandId: receipt.commandId, receiptId: receipt.receiptId }); + } + const command = this.#commands.get(receipt.commandId); + if (!command || command.commandKind !== receipt.commandKind || command.providerId !== receipt.providerId || command.serviceInstanceId !== receipt.serviceInstanceId) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + `Receipt ${receipt.receiptId} does not match a retained command for this provider generation.`, + ]); + } + await infrastructure(() => this.#options.acceptReceipt(record.identity, receipt)); + this.#receipts.set(receipt.commandId, receipt); + this.#commands.delete(receipt.commandId); + return this.#accepted(request, record.lease.leaseId, { commandId: receipt.commandId, receiptId: receipt.receiptId }); + } + + async #withdraw(request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession): Promise { + expectSchema(request, cultMeshProviderSessionSchemas.withdrawal); + const mutation = decodeProviderSessionPayload(request.payload); + const record = this.#requireLease(mutation.leaseId, session); + await infrastructure(() => this.#options.deletePublications(record.identity, publicationEntries(record))); + this.#releasePublicationOwners(record); + this.#removeLease(mutation.leaseId, record); + return this.#accepted(request, mutation.leaseId); + } + + #requireLease(leaseId: string, session: CultMeshRudpServerSession): LeaseRecord { + requireText(leaseId, "leaseId"); + const record = this.#leases.get(leaseId); + if (!record || !this.#isActive(record)) throw new BrokerStatusError("expired", `Provider lease ${leaseId} is expired or fenced.`); + if (record.session?.sessionId !== session.sessionId) throw new BrokerStatusError("denied", `Provider lease ${leaseId} belongs to another physical session.`); + return record; + } + + #isActive(record: LeaseRecord): boolean { + return Date.parse(record.lease.expiresAtUtc) > this.#clock.now().getTime(); + } + + #createLease(identity: CultMeshProviderRegistrationWire, requestedMs: number): CultMeshProviderLeaseWire { + if (!Number.isFinite(requestedMs) || requestedMs <= 0) throw new Error("requestedLeaseDurationMs must be positive."); + const durationMs = Math.min(requestedMs, this.#options.maximumLeaseDurationMs ?? 120_000); + const now = this.#clock.now(); + return { + providerId: identity.providerId, + serviceInstanceId: identity.serviceInstanceId, + endpointId: identity.endpointId, + verseId: identity.verseId, + leaseId: this.#options.createLeaseId?.() ?? `${this.#options.runtimeId}:lease:${++this.#leaseSequence}`, + validFromUtc: now.toISOString(), + expiresAtUtc: new Date(now.getTime() + durationMs).toISOString(), + }; + } + + #attachSession(session: CultMeshRudpServerSession, leaseId: string): void { + let leases = this.#sessionLeases.get(session.sessionId); + if (!leases) this.#sessionLeases.set(session.sessionId, leases = new Set()); + leases.add(leaseId); + } + + #removeLease(leaseId: string, record: LeaseRecord): void { + this.#leases.delete(leaseId); + if (this.#identityLeases.get(identityTuple(record.identity)) === leaseId) this.#identityLeases.delete(identityTuple(record.identity)); + this.#sessionLeases.get(record.session?.sessionId ?? "")?.delete(leaseId); + } + + #releasePublicationOwners(record: LeaseRecord): void { + const ownerIdentity = identityTuple(record.identity); + for (const [publicationId, document] of record.publications) { + const tuple = documentTuple(document); + const owner = this.#publicationOwners.get(tuple); + if (owner?.identityKey === ownerIdentity && owner.publicationId === publicationId) this.#publicationOwners.delete(tuple); + } + } + + #replayCommands(record: LeaseRecord): void { + for (const command of this.#commands.values()) { + if (command.providerId === record.identity.providerId && command.serviceInstanceId === record.identity.serviceInstanceId) this.#sendCommand(record, command); + } + } + + #sendCommand(record: LeaseRecord, command: CultMeshProviderCommandWire): void { + if (!record.session) return; + const message: CultNetDocumentPutRawMessage = { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: `${this.#options.runtimeId}:command:${command.commandId}`, + document: { + schemaId: cultMeshProviderSessionSchemas.command, + recordKey: `${command.providerId}/${command.serviceInstanceId}/${command.commandId}`, + storedAt: this.#clock.now().toISOString(), + payloadEncoding: "messagepack", + payload: encode(command), + sourceRuntimeId: this.#options.runtimeId, + sourceRole: "provider-session-broker", + }, + }; + record.session.send(message); + } + + #accepted(request: CultNetOperationRequestMessage, leaseId: string, extra: Record = {}): CultNetOperationResponseMessage { + return this.#response(request, "ok", cultMeshProviderSessionSchemas.mutationAcceptance, { + acceptedAtUtc: this.#clock.now().toISOString(), + leaseId, + ...extra, + }); + } + + #response( + request: CultNetOperationRequestMessage, + status: CultMeshProviderOperationStatus, + payloadSchema: string, + payload: unknown, + diagnostics?: string[], + ): CultNetOperationResponseMessage { + const response: CultNetOperationResponseMessage = { + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status, + payloadSchema, + payloadEncoding: "messagepack-base64", + payload: encodeProviderSessionPayload(payload), + sourceRuntimeId: this.#options.runtimeId, + }; + if (diagnostics) response.diagnostics = diagnostics; + return response; + } +} + +class BrokerStatusError extends Error { + public constructor(public readonly status: CultMeshProviderOperationStatus, message: string) { + super(message); + } +} + +class BrokerInfrastructureError extends Error { + public constructor(public readonly cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause)); + } +} + +async function infrastructure(action: () => T | Promise): Promise { + try { + return await action(); + } catch (error) { + throw new BrokerInfrastructureError(error); + } +} + +function expectSchema(request: CultNetOperationRequestMessage, schema: string): void { + if (request.payloadSchema !== schema) throw new Error(`Operation ${request.operation} requires payload schema ${schema}.`); +} + +function validateRegistration(value: CultMeshProviderRegistrationWire): void { + for (const field of ["providerId", "serviceInstanceId", "endpointId", "verseId"] as const) requireText(value[field], field); + if (!Number.isFinite(value.requestedLeaseDurationMs) || value.requestedLeaseDurationMs <= 0) throw new Error("requestedLeaseDurationMs must be positive."); +} + +function validateDocument(document: CultNetRawDocumentRecord): void { + requireText(document?.schemaId, "document.schemaId"); + requireText(document?.recordKey, "document.recordKey"); + if (document.payloadEncoding !== "messagepack" || !(document.payload instanceof Uint8Array)) throw new Error("Provider publications must carry a raw MessagePack document."); +} + +function validateReceipt(receipt: CultMeshProviderCommandReceiptWire): void { + requireText(receipt?.receiptId, "receipt.receiptId"); + requireText(receipt?.commandId, "receipt.commandId"); + requireText(receipt?.commandKind, "receipt.commandKind"); + requireText(receipt?.providerId, "receipt.providerId"); + requireText(receipt?.serviceInstanceId, "receipt.serviceInstanceId"); + if (receipt.state !== "applied" && receipt.state !== "rejected" && receipt.state !== "failed") throw new Error("CultMesh provider-session receipt.state is invalid."); + if (!Number.isFinite(Date.parse(receipt.completedAtUtc))) throw new Error("CultMesh provider-session receipt.completedAtUtc must be RFC3339."); +} + +function identityTuple(identity: CultMeshProviderRegistrationWire): string { + return `${identity.verseId}\u0000${identity.providerId}\u0000${identity.serviceInstanceId}\u0000${identity.endpointId}`; +} + +function documentTuple(document: Pick): string { + return `${document.schemaId}\u0000${document.recordKey}`; +} + +function wireEquivalent(left: unknown, right: unknown): boolean { + return Buffer.from(encode(left)).equals(Buffer.from(encode(right))); +} + +function publicationEntries(record: LeaseRecord): readonly { + publicationId: string; + document: Pick; +}[] { + return [...record.publications].map(([publicationId, document]) => ({ publicationId, document })); +} + +function requireText(value: unknown, name: string): asserts value is string { + if (typeof value !== "string" || value.trim().length === 0) throw new Error(`CultMesh provider-session ${name} must be non-empty.`); +} diff --git a/packages/cultmesh-ts/src/provider-session-wire.ts b/packages/cultmesh-ts/src/provider-session-wire.ts new file mode 100644 index 00000000..5af812c4 --- /dev/null +++ b/packages/cultmesh-ts/src/provider-session-wire.ts @@ -0,0 +1,170 @@ +import { decode, encode } from "@msgpack/msgpack"; +import type { CultNetRawDocumentRecord } from "cultnet-ts"; + +export const CULTMESH_PROVIDER_SESSION_SERVICE_ID = "gamecult.mesh.provider_session"; + +export const cultMeshProviderSessionOperations = { + register: "provider.register", + renew: "provider.renew", + publicationPut: "provider.publication.put", + publicationDelete: "provider.publication.delete", + receiptPut: "provider.receipt.put", + withdraw: "provider.withdraw", +} as const; + +export const cultMeshProviderSessionSchemas = { + registration: "gamecult.mesh.provider_registration.v1", + lease: "gamecult.mesh.provider_lease.v1", + leaseRenewal: "gamecult.mesh.provider_lease_renewal.v1", + publicationPut: "gamecult.mesh.provider_publication_put.v1", + publicationDelete: "gamecult.mesh.provider_publication_delete.v1", + command: "gamecult.mesh.provider_command.v1", + receiptPut: "gamecult.mesh.provider_receipt_put.v1", + withdrawal: "gamecult.mesh.provider_withdrawal.v1", + mutationAcceptance: "gamecult.mesh.provider_mutation_acceptance.v1", +} as const; + +export type CultMeshProviderOperationStatus = + | "ok" + | "conflict" + | "expired" + | "denied" + | "invalid"; + +export interface CultMeshProviderRegistrationWire { + providerId: string; + serviceInstanceId: string; + endpointId: string; + verseId: string; + requestedLeaseDurationMs: number; + authorityLeaseId?: string; +} + +export interface CultMeshProviderLeaseWire { + providerId: string; + serviceInstanceId: string; + endpointId: string; + verseId: string; + leaseId: string; + validFromUtc: string; + expiresAtUtc: string; +} + +export interface CultMeshProviderLeaseRenewalWire { + leaseId: string; + requestedLeaseDurationMs: number; +} + +export interface CultMeshProviderPublicationPutWire { + leaseId: string; + publicationId: string; + document: CultNetRawDocumentRecord; +} + +export interface CultMeshProviderPublicationDeleteWire { + leaseId: string; + publicationId: string; + schemaId: string; + recordKey: string; +} + +export interface CultMeshProviderCommandWire { + commandId: string; + commandKind: string; + providerId: string; + serviceInstanceId: string; + payload: unknown; +} + +export type CultMeshProviderReceiptStateWire = "applied" | "rejected" | "failed"; + +export interface CultMeshProviderCommandReceiptWire { + receiptId: string; + commandId: string; + commandKind: string; + providerId: string; + serviceInstanceId: string; + state: CultMeshProviderReceiptStateWire; + completedAtUtc: string; + result?: unknown; + error?: string; +} + +export interface CultMeshProviderReceiptPutWire { + leaseId: string; + receipt: CultMeshProviderCommandReceiptWire; +} + +export interface CultMeshProviderWithdrawalWire { + leaseId: string; +} + +export interface CultMeshProviderMutationAcceptanceWire { + acceptedAtUtc: string; + leaseId?: string; + publicationId?: string; + commandId?: string; + receiptId?: string; +} + +export function encodeProviderSessionPayload(value: unknown): string { + return Buffer.from(encode(value)).toString("base64"); +} + +export function decodeProviderSessionPayload(payload: string): T { + if (typeof payload !== "string" || payload.length === 0) { + throw new Error("CultMesh provider-session payload must be non-empty MessagePack base64."); + } + return decode(Buffer.from(payload, "base64")) as T; +} + +export function decodeProviderCommandDocument(document: CultNetRawDocumentRecord): CultMeshProviderCommandWire { + if (document.schemaId !== cultMeshProviderSessionSchemas.command) { + throw new Error(`Expected ${cultMeshProviderSessionSchemas.command}, received ${document.schemaId}.`); + } + if (document.payloadEncoding !== "messagepack") { + throw new Error(`Provider command ${document.recordKey} must use MessagePack.`); + } + const command = decode(document.payload) as Partial; + requireText(command.commandId, "commandId"); + requireText(command.commandKind, "commandKind"); + requireText(command.providerId, "providerId"); + requireText(command.serviceInstanceId, "serviceInstanceId"); + return command as CultMeshProviderCommandWire; +} + +export function assertProviderLease(value: unknown): asserts value is CultMeshProviderLeaseWire { + const lease = requireObject(value, "provider lease"); + requireText(lease.providerId, "providerId"); + requireText(lease.serviceInstanceId, "serviceInstanceId"); + requireText(lease.endpointId, "endpointId"); + requireText(lease.verseId, "verseId"); + requireText(lease.leaseId, "leaseId"); + requireTimestamp(lease.validFromUtc, "validFromUtc"); + requireTimestamp(lease.expiresAtUtc, "expiresAtUtc"); +} + +export function assertMutationAcceptance(value: unknown): asserts value is CultMeshProviderMutationAcceptanceWire { + const acceptance = requireObject(value, "provider mutation acceptance"); + requireTimestamp(acceptance.acceptedAtUtc, "acceptedAtUtc"); +} + +function requireObject(value: unknown, name: string): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new Error(`CultMesh ${name} must be a MessagePack map.`); + } + return value as Record; +} + +function requireText(value: unknown, name: string): asserts value is string { + if (typeof value !== "string" || value.trim().length === 0) { + throw new Error(`CultMesh provider-session ${name} must be non-empty.`); + } +} + +function requireTimestamp(value: unknown, name: string): asserts value is string { + requireText(value, name); + if (!Number.isFinite(Date.parse(value))) { + throw new Error(`CultMesh provider-session ${name} must be an RFC3339 timestamp.`); + } +} diff --git a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts new file mode 100644 index 00000000..6856f1cc --- /dev/null +++ b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts @@ -0,0 +1,161 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { decode } from "@msgpack/msgpack"; +import { CultNetDocumentRegistry } from "cultnet-ts"; + +import { + CultMesh, + CultMeshProviderRudpTransport, + CultMeshProviderSessionBroker, + type CultMeshProviderCommand, + type CultMeshProviderIdentity, + type CultMeshProviderPublication, +} from "../src"; + +const identity: CultMeshProviderIdentity = { + providerId: "voidbot.swarm", + serviceInstanceId: "voidbot-worker-7", + endpointId: "voidbot-worker-7.rudp", + verseId: "voidbot.local", +}; + +test("RUDP provider transport requires broker acceptance for the full lifecycle", async () => { + const accepted = new Map(); + const acceptedProvenance = new Map(); + const deleted: string[] = []; + const receipts: string[] = []; + let leaseSequence = 0; + const broker = new CultMeshProviderSessionBroker({ + runtimeId: "odin-test", + createLeaseId: () => `lease-${++leaseSequence}`, + authorizeRegistration: () => true, + acceptPublication: (_identity, publicationId, document) => { + accepted.set(publicationId, decode(document.payload)); + acceptedProvenance.set(publicationId, { + runtime: document.sourceRuntimeId, + agent: document.sourceAgentId, + }); + }, + deletePublications: (_identity, publications) => { + for (const { publicationId } of publications) { + deleted.push(publicationId); + accepted.delete(publicationId); + } + }, + acceptReceipt: (_identity, receipt) => { + receipts.push(receipt.receiptId); + }, + }); + const server = CultMesh.createRudpDocumentServer("odin-test", 0x43554c54, { + bindPort: 0, + documents: new CultNetDocumentRegistry(), + onOperationRequest: (request, session) => broker.handle(request, session), + onSessionClosed: session => broker.sessionClosed(session), + }); + await server.start(); + + const transport = new CultMeshProviderRudpTransport({ + endpoint: `rudp://127.0.0.1:${server.bind.port}`, + runtimeId: identity.serviceInstanceId, + connectionId: 0x43554c54, + }); + const connection = await transport.connect(identity, new AbortController().signal); + const secondIdentity: CultMeshProviderIdentity = { + providerId: "aetheria", + serviceInstanceId: "aetheria-daemon-2", + endpointId: "aetheria-daemon-2.rudp", + verseId: "aetheria.local", + }; + const secondConnection = await new CultMeshProviderRudpTransport({ + endpoint: `rudp://127.0.0.1:${server.bind.port}`, + runtimeId: secondIdentity.serviceInstanceId, + connectionId: 0x43554c54, + }).connect(secondIdentity, new AbortController().signal); + try { + broker.enqueueCommand({ + commandId: "refresh-1", + commandKind: "voidbot.refresh", + providerId: identity.providerId, + serviceInstanceId: identity.serviceInstanceId, + payload: { force: true }, + }); + const lease = await connection.register({ identity, requestedLeaseDurationMs: 10_000 }, new AbortController().signal); + assert.equal(lease.leaseId, "lease-1"); + const secondLease = await secondConnection.register({ + identity: secondIdentity, + requestedLeaseDurationMs: 10_000, + }, new AbortController().signal); + assert.equal(secondLease.leaseId, "lease-2"); + assert.equal(broker.activeLeaseCount, 2, "the broker demultiplexes simultaneous physical provider sessions"); + + const publication: CultMeshProviderPublication = { + publicationId: "voidbot.surface", + documentType: "gamecult.eve.surface_state", + schemaId: "gamecult.eve.surface_state.v1", + recordKey: "voidbot.swarm", + value: { generation: 3 }, + }; + await connection.publish(publication, lease); + assert.deepEqual(accepted.get(publication.publicationId), publication.value); + assert.deepEqual(acceptedProvenance.get(publication.publicationId), { + runtime: identity.serviceInstanceId, + agent: identity.providerId, + }, "the broker derives provenance from the fenced identity"); + const secondPublication: CultMeshProviderPublication = { + publicationId: "aetheria.surface", + documentType: "gamecult.eve.surface_state", + schemaId: "gamecult.eve.surface_state.v1", + recordKey: "aetheria.pilot", + value: { generation: 8 }, + }; + await secondConnection.publish(secondPublication, secondLease); + await assert.rejects(secondConnection.publish({ + ...secondPublication, + publicationId: "aetheria.stolen-tuple", + recordKey: publication.recordKey, + }, secondLease), /already owned/); + await assert.rejects( + secondConnection.publish({ ...secondPublication, value: { generation: 9 } }, lease), + /belongs to another physical session/, + ); + + const commandArrived = deferred(); + connection.watchCommands(command => commandArrived.resolve(command)); + const command = await commandArrived.promise; + assert.equal(command.commandId, "refresh-1"); + await connection.publishReceipt({ + receiptId: "receipt-refresh-1", + commandId: command.commandId, + commandKind: command.commandKind, + providerId: identity.providerId, + serviceInstanceId: identity.serviceInstanceId, + state: "applied", + completedAt: new Date(), + result: { refreshed: true }, + }, lease); + assert.deepEqual(receipts, ["receipt-refresh-1"]); + + const renewed = await connection.renew(lease); + assert.equal(renewed.leaseId, "lease-3"); + await assert.rejects(connection.withdrawPublication(publication.publicationId, lease), /expired|fenced/); + await connection.withdraw({ identity, leaseId: renewed.leaseId, publicationIds: [publication.publicationId] }); + await secondConnection.withdraw({ + identity: secondIdentity, + leaseId: secondLease.leaseId, + publicationIds: [secondPublication.publicationId], + }); + assert.deepEqual(deleted, [publication.publicationId, secondPublication.publicationId]); + assert.equal(accepted.size, 0); + assert.equal(broker.activeLeaseCount, 0); + } finally { + connection.close(); + secondConnection.close(); + server.close(); + broker.close(); + } +}); + +function deferred(): { promise: Promise; resolve(value: T): void } { + let resolve!: (value: T) => void; + return { promise: new Promise(done => resolve = done), resolve }; +} diff --git a/packages/cultmesh-ts/test/provider-session-wire.test.ts b/packages/cultmesh-ts/test/provider-session-wire.test.ts new file mode 100644 index 00000000..bedbfd29 --- /dev/null +++ b/packages/cultmesh-ts/test/provider-session-wire.test.ts @@ -0,0 +1,17 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { encodeProviderSessionPayload } from "../src/provider-session-wire"; + +const REGISTRATION_GOLDEN = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03"; + +test("provider registration matches the shared C# MessagePack golden payload", () => { + assert.equal(encodeProviderSessionPayload({ + providerId: "aetheria", + serviceInstanceId: "aetheria-42", + endpointId: "aetheria-public", + verseId: "public", + requestedLeaseDurationMs: 30_000, + authorityLeaseId: "authority-7", + }), REGISTRATION_GOLDEN); +}); diff --git a/src/GameCult.Mesh/CultMeshProviderSessionWire.cs b/src/GameCult.Mesh/CultMeshProviderSessionWire.cs new file mode 100644 index 00000000..37d991a0 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshProviderSessionWire.cs @@ -0,0 +1,487 @@ +using System; +using System.Globalization; +using GameCult.Networking; +using MessagePack; + +namespace GameCult.Mesh +{ + /// + /// Canonical CultNet service, operation, payload-schema, and status identifiers for + /// the CultMesh provider-session control plane. + /// + public static class CultMeshProviderSessionWireContract + { + /// The CultNet service id. + public const string ServiceId = "gamecult.mesh.provider_session"; + + /// The registration operation. + public const string RegisterOperation = "provider.register"; + /// The lease-renewal operation. + public const string RenewOperation = "provider.renew"; + /// The publication-put operation. + public const string PublicationPutOperation = "provider.publication.put"; + /// The publication-delete operation. + public const string PublicationDeleteOperation = "provider.publication.delete"; + /// The command-receipt-put operation. + public const string ReceiptPutOperation = "provider.receipt.put"; + /// The provider-withdrawal operation. + public const string WithdrawOperation = "provider.withdraw"; + + /// The registration payload schema. + public const string RegistrationSchema = "gamecult.mesh.provider_registration.v1"; + /// The provider-lease payload schema. + public const string LeaseSchema = "gamecult.mesh.provider_lease.v1"; + /// The lease-renewal payload schema. + public const string LeaseRenewalSchema = "gamecult.mesh.provider_lease_renewal.v1"; + /// The publication-put payload schema. + public const string PublicationPutSchema = "gamecult.mesh.provider_publication_put.v1"; + /// The publication-delete payload schema. + public const string PublicationDeleteSchema = "gamecult.mesh.provider_publication_delete.v1"; + /// The broker-to-provider command document schema. + public const string CommandSchema = "gamecult.mesh.provider_command.v1"; + /// The command-receipt-put payload schema. + public const string ReceiptPutSchema = "gamecult.mesh.provider_receipt_put.v1"; + /// The provider-withdrawal payload schema. + public const string WithdrawalSchema = "gamecult.mesh.provider_withdrawal.v1"; + /// The accepted-mutation response payload schema. + public const string MutationAcceptanceSchema = "gamecult.mesh.provider_mutation_acceptance.v1"; + + /// The successful application status. + public const string OkStatus = "ok"; + /// The identity or generation conflict status. + public const string ConflictStatus = "conflict"; + /// The expired-lease status. + public const string ExpiredStatus = "expired"; + /// The authority-denied status. + public const string DeniedStatus = "denied"; + /// The invalid-request status. + public const string InvalidStatus = "invalid"; + /// The CultNet inner-payload encoding. + public const string PayloadEncoding = "messagepack-base64"; + } + + /// Requests a fenced provider lease from the provider-session broker. + [MessagePackObject] + public sealed class CultMeshProviderRegistrationWire + { + /// Gets or sets the stable provider identity. + [Key("providerId")] public string ProviderId { get; set; } = string.Empty; + /// Gets or sets the provider process generation. + [Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty; + /// Gets or sets the advertised endpoint identity. + [Key("endpointId")] public string EndpointId { get; set; } = string.Empty; + /// Gets or sets the Verse identity. + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + /// Gets or sets the requested lease duration in milliseconds. + [Key("requestedLeaseDurationMs")] public int RequestedLeaseDurationMs { get; set; } + /// Gets or sets the optional authority lease authorizing registration. + [Key("authorityLeaseId")] public string? AuthorityLeaseId { get; set; } + } + + /// Identifies the provider generation authorized by a broker lease. + [MessagePackObject] + public sealed class CultMeshProviderLeaseWire + { + /// Gets or sets the stable provider identity. + [Key("providerId")] public string ProviderId { get; set; } = string.Empty; + /// Gets or sets the authorized provider process generation. + [Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty; + /// Gets or sets the authorized endpoint identity. + [Key("endpointId")] public string EndpointId { get; set; } = string.Empty; + /// Gets or sets the authorized Verse identity. + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + /// Gets or sets the unique fencing lease id. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + /// Gets or sets the RFC3339 lease validity start. + [Key("validFromUtc")] public string ValidFromUtc { get; set; } = string.Empty; + /// Gets or sets the RFC3339 lease expiry. + [Key("expiresAtUtc")] public string ExpiresAtUtc { get; set; } = string.Empty; + } + + /// Requests replacement of an active provider lease. + [MessagePackObject] + public sealed class CultMeshProviderLeaseRenewalWire + { + /// Gets or sets the lease being replaced. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + /// Gets or sets the requested replacement duration in milliseconds. + [Key("requestedLeaseDurationMs")] public int RequestedLeaseDurationMs { get; set; } + } + + /// Publishes one typed document under an active provider lease. + [MessagePackObject] + public sealed class CultMeshProviderPublicationPutWire + { + /// Gets or sets the active provider lease. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + /// Gets or sets the provider-owned publication identity. + [Key("publicationId")] public string PublicationId { get; set; } = string.Empty; + /// Gets or sets the exact typed CultNet document. + [Key("document")] public CultNetRawDocumentRecord Document { get; set; } = new CultNetRawDocumentRecord(); + } + + /// Deletes one exact provider-owned publication tuple. + [MessagePackObject] + public sealed class CultMeshProviderPublicationDeleteWire + { + /// Gets or sets the active provider lease. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + /// Gets or sets the provider-owned publication identity. + [Key("publicationId")] public string PublicationId { get; set; } = string.Empty; + /// Gets or sets the exact published schema id. + [Key("schemaId")] public string SchemaId { get; set; } = string.Empty; + /// Gets or sets the exact published record key. + [Key("recordKey")] public string RecordKey { get; set; } = string.Empty; + } + + /// A retained command routed from the broker to one provider generation. + [MessagePackObject] + public sealed class CultMeshProviderCommandWire + { + /// Gets or sets the stable command identity. + [Key("commandId")] public string CommandId { get; set; } = string.Empty; + /// Gets or sets the provider-defined command kind. + [Key("commandKind")] public string CommandKind { get; set; } = string.Empty; + /// Gets or sets the target provider identity. + [Key("providerId")] public string ProviderId { get; set; } = string.Empty; + /// Gets or sets the target provider process generation. + [Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty; + /// Gets or sets the typed command payload value. + [Key("payload")] public object? Payload { get; set; } + } + + /// The exactly-once outcome of one provider command. + [MessagePackObject] + public sealed class CultMeshProviderCommandReceiptWire + { + /// Gets or sets the stable receipt identity. + [Key("receiptId")] public string ReceiptId { get; set; } = string.Empty; + /// Gets or sets the completed command identity. + [Key("commandId")] public string CommandId { get; set; } = string.Empty; + /// Gets or sets the completed command kind. + [Key("commandKind")] public string CommandKind { get; set; } = string.Empty; + /// Gets or sets the provider identity. + [Key("providerId")] public string ProviderId { get; set; } = string.Empty; + /// Gets or sets the provider process generation. + [Key("serviceInstanceId")] public string ServiceInstanceId { get; set; } = string.Empty; + /// Gets or sets applied, rejected, or failed. + [Key("state")] public string State { get; set; } = string.Empty; + /// Gets or sets the RFC3339 completion time. + [Key("completedAtUtc")] public string CompletedAtUtc { get; set; } = string.Empty; + /// Gets or sets the optional typed result value. + [Key("result")] public object? Result { get; set; } + /// Gets or sets the optional bounded failure detail. + [Key("error")] public string? Error { get; set; } + } + + /// Submits a durable command receipt under an active provider lease. + [MessagePackObject] + public sealed class CultMeshProviderReceiptPutWire + { + /// Gets or sets the active provider lease. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + /// Gets or sets the durable command receipt. + [Key("receipt")] public CultMeshProviderCommandReceiptWire Receipt { get; set; } = new CultMeshProviderCommandReceiptWire(); + } + + /// Withdraws the provider generation identified by a lease. + [MessagePackObject] + public sealed class CultMeshProviderWithdrawalWire + { + /// Gets or sets the lease whose owned publications must be withdrawn. + [Key("leaseId")] public string LeaseId { get; set; } = string.Empty; + } + + /// Confirms broker acceptance of one provider-session mutation. + [MessagePackObject] + public sealed class CultMeshProviderMutationAcceptanceWire + { + /// Gets or sets the RFC3339 broker acceptance time. + [Key("acceptedAtUtc")] public string AcceptedAtUtc { get; set; } = string.Empty; + /// Gets or sets the accepted lease identity, when applicable. + [Key("leaseId")] public string? LeaseId { get; set; } + /// Gets or sets the accepted publication identity, when applicable. + [Key("publicationId")] public string? PublicationId { get; set; } + /// Gets or sets the accepted command identity, when applicable. + [Key("commandId")] public string? CommandId { get; set; } + /// Gets or sets the accepted receipt identity, when applicable. + [Key("receiptId")] public string? ReceiptId { get; set; } + } + + /// + /// Encodes provider-session payloads inside the existing CultNet operation envelope. + /// Transport acknowledgement is deliberately not treated as application acceptance. + /// + public static class CultMeshProviderSessionWire + { + private static readonly MessagePackSerializerOptions Options = + MessagePackSerializerOptions.Standard.WithSecurity(MessagePackSecurity.UntrustedData); + + /// Encodes one validated provider-session value as MessagePack base64. + public static string EncodePayload(T value) + { + if (value == null) throw new ArgumentNullException(nameof(value)); + Validate(value); + return Convert.ToBase64String(MessagePackSerializer.Serialize(value, Options)); + } + + /// Decodes and validates one MessagePack-base64 provider-session value. + public static T DecodePayload(string payload) + { + RequireText(payload, nameof(payload)); + byte[] bytes; + try + { + bytes = Convert.FromBase64String(payload); + } + catch (FormatException exception) + { + throw new ArgumentException("Provider-session payload must be MessagePack base64.", nameof(payload), exception); + } + + var value = MessagePackSerializer.Deserialize(bytes, Options); + if (value == null) throw new MessagePackSerializationException("Provider-session payload decoded to null."); + Validate(value); + return value; + } + + /// Decodes a retained provider command carried as a raw CultNet document. + public static CultMeshProviderCommandWire DecodeCommandDocument(CultNetRawDocumentRecord document) + { + if (document == null) throw new ArgumentNullException(nameof(document)); + if (!string.Equals(document.SchemaId, CultMeshProviderSessionWireContract.CommandSchema, StringComparison.Ordinal)) + throw new ArgumentException($"Expected provider command schema '{CultMeshProviderSessionWireContract.CommandSchema}'.", nameof(document)); + if (!string.Equals(document.PayloadEncoding, "messagepack", StringComparison.Ordinal)) + throw new ArgumentException("Provider command document must use MessagePack.", nameof(document)); + var command = MessagePackSerializer.Deserialize(document.Payload, Options); + if (command == null) throw new MessagePackSerializationException("Provider command decoded to null."); + Validate(command); + return command; + } + + /// Creates a canonical provider-session CultNet operation request. + public static CultNetOperationRequestMessage CreateRequest( + string messageId, + string operation, + string payloadSchema, + T payload, + string? sourceRuntimeId = null, + string? targetRuntimeId = null) + { + RequireText(messageId, nameof(messageId)); + ValidateOperation(operation, payloadSchema); + return new CultNetOperationRequestMessage + { + MessageId = messageId, + ServiceId = CultMeshProviderSessionWireContract.ServiceId, + Operation = operation, + PayloadSchema = payloadSchema, + PayloadEncoding = CultMeshProviderSessionWireContract.PayloadEncoding, + Payload = EncodePayload(payload), + SourceRuntimeId = sourceRuntimeId, + TargetRuntimeId = targetRuntimeId + }; + } + + /// Validates an operation request envelope and decodes its typed payload. + public static T DecodeRequest( + CultNetOperationRequestMessage request, + string expectedOperation, + string expectedPayloadSchema) + { + if (request == null) throw new ArgumentNullException(nameof(request)); + RequireEnvelope(request.ServiceId, request.Operation, request.PayloadSchema, request.PayloadEncoding, + expectedOperation, expectedPayloadSchema); + return DecodePayload(request.Payload); + } + + /// Creates a correlated application-level operation response. + public static CultNetOperationResponseMessage CreateResponse( + CultNetOperationRequestMessage request, + string status, + string payloadSchema, + T payload, + string? sourceRuntimeId = null) + { + if (request == null) throw new ArgumentNullException(nameof(request)); + ValidateStatus(status); + return new CultNetOperationResponseMessage + { + MessageId = RequireText(request.MessageId, nameof(request.MessageId)), + ServiceId = CultMeshProviderSessionWireContract.ServiceId, + Operation = RequireText(request.Operation, nameof(request.Operation)), + Status = status, + PayloadSchema = RequireText(payloadSchema, nameof(payloadSchema)), + PayloadEncoding = CultMeshProviderSessionWireContract.PayloadEncoding, + Payload = EncodePayload(payload), + SourceRuntimeId = sourceRuntimeId + }; + } + + /// Validates an operation response envelope and decodes its typed payload. + public static T DecodeResponse( + CultNetOperationResponseMessage response, + string expectedOperation, + string expectedPayloadSchema) + { + if (response == null) throw new ArgumentNullException(nameof(response)); + ValidateStatus(response.Status); + RequireEnvelope(response.ServiceId, response.Operation, response.PayloadSchema, response.PayloadEncoding, + expectedOperation, expectedPayloadSchema); + return DecodePayload(response.Payload); + } + + /// Validates the invariants of a known provider-session wire value. + public static void Validate(T value) + { + switch (value) + { + case CultMeshProviderRegistrationWire registration: + RequireIdentity(registration.ProviderId, registration.ServiceInstanceId, registration.EndpointId, registration.VerseId); + RequirePositive(registration.RequestedLeaseDurationMs, nameof(registration.RequestedLeaseDurationMs)); + RequireOptionalText(registration.AuthorityLeaseId, nameof(registration.AuthorityLeaseId)); + break; + case CultMeshProviderLeaseWire lease: + RequireIdentity(lease.ProviderId, lease.ServiceInstanceId, lease.EndpointId, lease.VerseId); + RequireText(lease.LeaseId, nameof(lease.LeaseId)); + var validFrom = RequireTimestamp(lease.ValidFromUtc, nameof(lease.ValidFromUtc)); + var expiresAt = RequireTimestamp(lease.ExpiresAtUtc, nameof(lease.ExpiresAtUtc)); + if (expiresAt <= validFrom) throw new ArgumentException("Provider lease must expire after validFromUtc."); + break; + case CultMeshProviderLeaseRenewalWire renewal: + RequireText(renewal.LeaseId, nameof(renewal.LeaseId)); + RequirePositive(renewal.RequestedLeaseDurationMs, nameof(renewal.RequestedLeaseDurationMs)); + break; + case CultMeshProviderPublicationPutWire publication: + RequireText(publication.LeaseId, nameof(publication.LeaseId)); + RequireText(publication.PublicationId, nameof(publication.PublicationId)); + if (publication.Document == null) throw new ArgumentException("Provider publication document is required."); + RequireText(publication.Document.SchemaId, nameof(publication.Document.SchemaId)); + RequireText(publication.Document.RecordKey, nameof(publication.Document.RecordKey)); + if (!string.Equals(publication.Document.PayloadEncoding, "messagepack", StringComparison.Ordinal)) + throw new ArgumentException("Provider publication document must use MessagePack."); + break; + case CultMeshProviderPublicationDeleteWire deletion: + RequireText(deletion.LeaseId, nameof(deletion.LeaseId)); + RequireText(deletion.PublicationId, nameof(deletion.PublicationId)); + RequireText(deletion.SchemaId, nameof(deletion.SchemaId)); + RequireText(deletion.RecordKey, nameof(deletion.RecordKey)); + break; + case CultMeshProviderCommandWire command: + RequireText(command.CommandId, nameof(command.CommandId)); + RequireText(command.CommandKind, nameof(command.CommandKind)); + RequireText(command.ProviderId, nameof(command.ProviderId)); + RequireText(command.ServiceInstanceId, nameof(command.ServiceInstanceId)); + break; + case CultMeshProviderCommandReceiptWire receipt: + RequireText(receipt.ReceiptId, nameof(receipt.ReceiptId)); + RequireText(receipt.CommandId, nameof(receipt.CommandId)); + RequireText(receipt.CommandKind, nameof(receipt.CommandKind)); + RequireText(receipt.ProviderId, nameof(receipt.ProviderId)); + RequireText(receipt.ServiceInstanceId, nameof(receipt.ServiceInstanceId)); + ValidateReceiptState(receipt.State); + RequireTimestamp(receipt.CompletedAtUtc, nameof(receipt.CompletedAtUtc)); + RequireOptionalText(receipt.Error, nameof(receipt.Error)); + break; + case CultMeshProviderReceiptPutWire receiptPut: + RequireText(receiptPut.LeaseId, nameof(receiptPut.LeaseId)); + if (receiptPut.Receipt == null) throw new ArgumentException("Provider receipt is required."); + Validate(receiptPut.Receipt); + break; + case CultMeshProviderWithdrawalWire withdrawal: + RequireText(withdrawal.LeaseId, nameof(withdrawal.LeaseId)); + break; + case CultMeshProviderMutationAcceptanceWire acceptance: + RequireTimestamp(acceptance.AcceptedAtUtc, nameof(acceptance.AcceptedAtUtc)); + RequireOptionalText(acceptance.LeaseId, nameof(acceptance.LeaseId)); + RequireOptionalText(acceptance.PublicationId, nameof(acceptance.PublicationId)); + RequireOptionalText(acceptance.CommandId, nameof(acceptance.CommandId)); + RequireOptionalText(acceptance.ReceiptId, nameof(acceptance.ReceiptId)); + break; + } + } + + private static void RequireEnvelope( + string serviceId, + string operation, + string payloadSchema, + string payloadEncoding, + string expectedOperation, + string expectedPayloadSchema) + { + if (!string.Equals(serviceId, CultMeshProviderSessionWireContract.ServiceId, StringComparison.Ordinal)) + throw new ArgumentException($"Expected provider-session service '{CultMeshProviderSessionWireContract.ServiceId}'."); + if (!string.Equals(operation, expectedOperation, StringComparison.Ordinal)) + throw new ArgumentException($"Expected provider-session operation '{expectedOperation}'."); + if (!string.Equals(payloadSchema, expectedPayloadSchema, StringComparison.Ordinal)) + throw new ArgumentException($"Expected provider-session payload schema '{expectedPayloadSchema}'."); + if (!string.Equals(payloadEncoding, CultMeshProviderSessionWireContract.PayloadEncoding, StringComparison.Ordinal)) + throw new ArgumentException("Provider-session operation payload must use MessagePack base64."); + } + + private static void ValidateOperation(string operation, string payloadSchema) + { + var expectedSchema = operation switch + { + CultMeshProviderSessionWireContract.RegisterOperation => CultMeshProviderSessionWireContract.RegistrationSchema, + CultMeshProviderSessionWireContract.RenewOperation => CultMeshProviderSessionWireContract.LeaseRenewalSchema, + CultMeshProviderSessionWireContract.PublicationPutOperation => CultMeshProviderSessionWireContract.PublicationPutSchema, + CultMeshProviderSessionWireContract.PublicationDeleteOperation => CultMeshProviderSessionWireContract.PublicationDeleteSchema, + CultMeshProviderSessionWireContract.ReceiptPutOperation => CultMeshProviderSessionWireContract.ReceiptPutSchema, + CultMeshProviderSessionWireContract.WithdrawOperation => CultMeshProviderSessionWireContract.WithdrawalSchema, + _ => throw new ArgumentException($"Unsupported provider-session operation '{operation}'.", nameof(operation)) + }; + if (!string.Equals(payloadSchema, expectedSchema, StringComparison.Ordinal)) + throw new ArgumentException($"Operation '{operation}' requires payload schema '{expectedSchema}'.", nameof(payloadSchema)); + } + + private static void ValidateStatus(string status) + { + if (status != CultMeshProviderSessionWireContract.OkStatus && + status != CultMeshProviderSessionWireContract.ConflictStatus && + status != CultMeshProviderSessionWireContract.ExpiredStatus && + status != CultMeshProviderSessionWireContract.DeniedStatus && + status != CultMeshProviderSessionWireContract.InvalidStatus) + throw new ArgumentException($"Unsupported provider-session status '{status}'.", nameof(status)); + } + + private static void ValidateReceiptState(string state) + { + if (state != "applied" && state != "rejected" && state != "failed") + throw new ArgumentException($"Unsupported provider receipt state '{state}'.", nameof(state)); + } + + private static void RequireIdentity(string providerId, string serviceInstanceId, string endpointId, string verseId) + { + RequireText(providerId, nameof(providerId)); + RequireText(serviceInstanceId, nameof(serviceInstanceId)); + RequireText(endpointId, nameof(endpointId)); + RequireText(verseId, nameof(verseId)); + } + + private static string RequireText(string value, string name) + { + if (string.IsNullOrWhiteSpace(value)) throw new ArgumentException("Value must be non-empty.", name); + return value; + } + + private static void RequireOptionalText(string? value, string name) + { + if (value != null && string.IsNullOrWhiteSpace(value)) + throw new ArgumentException("Optional value must be null or non-empty.", name); + } + + private static void RequirePositive(int value, string name) + { + if (value <= 0) throw new ArgumentOutOfRangeException(name, "Value must be positive."); + } + + private static DateTimeOffset RequireTimestamp(string value, string name) + { + RequireText(value, name); + if (!DateTimeOffset.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var timestamp)) + throw new ArgumentException("Value must be an RFC3339 timestamp.", name); + return timestamp; + } + } +} diff --git a/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs b/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs new file mode 100644 index 00000000..7e6f89a1 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs @@ -0,0 +1,224 @@ +using System; +using System.Collections.Generic; +using GameCult.Networking; +using MessagePack; +using NUnit.Framework; + +namespace GameCult.Mesh.Tests; + +[TestFixture] +public sealed class CultMeshProviderSessionWireTests +{ + private const string RegistrationGolden = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03"; + + [Test] + public void Registration_RoundTrips_InsideCanonicalCultNetOperationEnvelope() + { + var registration = new CultMeshProviderRegistrationWire + { + ProviderId = "aetheria", + ServiceInstanceId = "aetheria-42", + EndpointId = "aetheria-public", + VerseId = "public", + RequestedLeaseDurationMs = 30_000, + AuthorityLeaseId = "authority-7" + }; + + var request = CultMeshProviderSessionWire.CreateRequest( + "register-1", + CultMeshProviderSessionWireContract.RegisterOperation, + CultMeshProviderSessionWireContract.RegistrationSchema, + registration, + sourceRuntimeId: "aetheria-42", + targetRuntimeId: "odin"); + var envelopeBytes = CultNetSchemaMessageSerialization.Serialize(request); + var decodedEnvelope = (CultNetOperationRequestMessage)CultNetSchemaMessageSerialization.Deserialize(envelopeBytes); + var decoded = CultMeshProviderSessionWire.DecodeRequest( + decodedEnvelope, + CultMeshProviderSessionWireContract.RegisterOperation, + CultMeshProviderSessionWireContract.RegistrationSchema); + + Assert.Multiple(() => + { + Assert.That(CultMeshProviderSessionWire.EncodePayload(registration), Is.EqualTo(RegistrationGolden), + "C# and TypeScript must emit the same canonical registration payload"); + Assert.That(decodedEnvelope.ServiceId, Is.EqualTo("gamecult.mesh.provider_session")); + Assert.That(decodedEnvelope.PayloadEncoding, Is.EqualTo("messagepack-base64")); + Assert.That(decoded.ProviderId, Is.EqualTo("aetheria")); + Assert.That(decoded.ServiceInstanceId, Is.EqualTo("aetheria-42")); + Assert.That(decoded.EndpointId, Is.EqualTo("aetheria-public")); + Assert.That(decoded.VerseId, Is.EqualTo("public")); + Assert.That(decoded.RequestedLeaseDurationMs, Is.EqualTo(30_000)); + Assert.That(decoded.AuthorityLeaseId, Is.EqualTo("authority-7")); + }); + } + + [Test] + public void PublicationPut_RoundTrips_ExactRawDocumentTuple() + { + var publication = new CultMeshProviderPublicationPutWire + { + LeaseId = "lease-1", + PublicationId = "surface:pilot", + Document = new CultNetRawDocumentRecord + { + SchemaId = "gamecult.eve.surface.v1", + RecordKey = "eve:surface:pilot", + StoredAt = "2026-07-14T12:00:00Z", + PayloadEncoding = "messagepack", + Payload = [0x81, 0xA2, 0x69, 0x64, 0x01], + SourceRuntimeId = "aetheria-42" + } + }; + + var encoded = CultMeshProviderSessionWire.EncodePayload(publication); + var decoded = CultMeshProviderSessionWire.DecodePayload(encoded); + + Assert.Multiple(() => + { + Assert.That(decoded.LeaseId, Is.EqualTo("lease-1")); + Assert.That(decoded.PublicationId, Is.EqualTo("surface:pilot")); + Assert.That(decoded.Document.SchemaId, Is.EqualTo("gamecult.eve.surface.v1")); + Assert.That(decoded.Document.RecordKey, Is.EqualTo("eve:surface:pilot")); + Assert.That(decoded.Document.Payload, Is.EqualTo(publication.Document.Payload)); + }); + } + + [Test] + public void CommandAndReceipt_UseCamelCaseMessagePackMaps() + { + var command = new CultMeshProviderCommandWire + { + CommandId = "cmd-7", + CommandKind = "pilot.thrust", + ProviderId = "aetheria", + ServiceInstanceId = "aetheria-42", + Payload = new Dictionary { ["axis"] = 0.75 } + }; + var receiptPut = new CultMeshProviderReceiptPutWire + { + LeaseId = "lease-1", + Receipt = new CultMeshProviderCommandReceiptWire + { + ReceiptId = "receipt-7", + CommandId = command.CommandId, + CommandKind = command.CommandKind, + ProviderId = command.ProviderId, + ServiceInstanceId = command.ServiceInstanceId, + State = "applied", + CompletedAtUtc = "2026-07-14T12:01:02.345Z", + Result = new Dictionary { ["accepted"] = true } + } + }; + + var encodedCommand = CultMeshProviderSessionWire.EncodePayload(command); + var encodedReceipt = CultMeshProviderSessionWire.EncodePayload(receiptPut); + var decodedCommand = CultMeshProviderSessionWire.DecodePayload(encodedCommand); + var decodedReceipt = CultMeshProviderSessionWire.DecodePayload(encodedReceipt); + var decodedCommandDocument = CultMeshProviderSessionWire.DecodeCommandDocument(new CultNetRawDocumentRecord + { + SchemaId = CultMeshProviderSessionWireContract.CommandSchema, + RecordKey = "provider-command:aetheria:aetheria-42:cmd-7", + PayloadEncoding = "messagepack", + Payload = Convert.FromBase64String(encodedCommand) + }); + var commandJson = MessagePackSerializer.ConvertToJson(Convert.FromBase64String(encodedCommand)); + var receiptJson = MessagePackSerializer.ConvertToJson(Convert.FromBase64String(encodedReceipt)); + + Assert.Multiple(() => + { + Assert.That(commandJson, Does.Contain("\"commandId\"")); + Assert.That(commandJson, Does.Contain("\"commandKind\"")); + Assert.That(commandJson, Does.Contain("\"serviceInstanceId\"")); + Assert.That(commandJson, Does.Contain("\"payload\"")); + Assert.That(decodedCommand.CommandKind, Is.EqualTo("pilot.thrust")); + Assert.That(decodedCommand.Payload, Is.Not.Null); + Assert.That(decodedCommandDocument.CommandId, Is.EqualTo("cmd-7")); + Assert.That(receiptJson, Does.Contain("\"completedAtUtc\"")); + Assert.That(receiptJson, Does.Contain("\"receiptId\"")); + Assert.That(receiptJson, Does.Contain("\"result\"")); + Assert.That(decodedReceipt.Receipt.State, Is.EqualTo("applied")); + Assert.That(decodedReceipt.Receipt.Result, Is.Not.Null); + }); + } + + [Test] + public void LeaseResponse_RoundTrips_WithApplicationAcceptance() + { + var request = CultMeshProviderSessionWire.CreateRequest( + "register-1", + CultMeshProviderSessionWireContract.RegisterOperation, + CultMeshProviderSessionWireContract.RegistrationSchema, + new CultMeshProviderRegistrationWire + { + ProviderId = "aetheria", + ServiceInstanceId = "aetheria-42", + EndpointId = "public", + VerseId = "verse", + RequestedLeaseDurationMs = 10_000 + }); + var response = CultMeshProviderSessionWire.CreateResponse( + request, + CultMeshProviderSessionWireContract.OkStatus, + CultMeshProviderSessionWireContract.LeaseSchema, + new CultMeshProviderLeaseWire + { + ProviderId = "aetheria", + ServiceInstanceId = "aetheria-42", + EndpointId = "public", + VerseId = "verse", + LeaseId = "lease-2", + ValidFromUtc = "2026-07-14T12:00:00Z", + ExpiresAtUtc = "2026-07-14T12:00:10Z" + }, + sourceRuntimeId: "odin"); + + var lease = CultMeshProviderSessionWire.DecodeResponse( + response, + CultMeshProviderSessionWireContract.RegisterOperation, + CultMeshProviderSessionWireContract.LeaseSchema); + + Assert.That(response.Status, Is.EqualTo("ok")); + Assert.That(lease.LeaseId, Is.EqualTo("lease-2")); + Assert.That(lease.ExpiresAtUtc, Is.EqualTo("2026-07-14T12:00:10Z")); + } + + [Test] + public void Validation_RejectsSplitAuthorityAndMalformedLifecycleState() + { + Assert.Multiple(() => + { + Assert.That( + () => CultMeshProviderSessionWire.CreateRequest( + "bad-schema", + CultMeshProviderSessionWireContract.RegisterOperation, + CultMeshProviderSessionWireContract.PublicationPutSchema, + new CultMeshProviderRegistrationWire()), + Throws.ArgumentException); + Assert.That( + () => CultMeshProviderSessionWire.EncodePayload(new CultMeshProviderLeaseWire + { + ProviderId = "provider", + ServiceInstanceId = "instance", + EndpointId = "endpoint", + VerseId = "verse", + LeaseId = "lease", + ValidFromUtc = "2026-07-14T12:00:10Z", + ExpiresAtUtc = "2026-07-14T12:00:00Z" + }), + Throws.ArgumentException); + Assert.That( + () => CultMeshProviderSessionWire.EncodePayload(new CultMeshProviderCommandReceiptWire + { + ReceiptId = "receipt", + CommandId = "command", + CommandKind = "kind", + ProviderId = "provider", + ServiceInstanceId = "instance", + State = "maybe", + CompletedAtUtc = "2026-07-14T12:00:00Z" + }), + Throws.ArgumentException); + }); + } +}