import { mappingSuccess, type InstalledBoundaryMapper, } from "../../../src/contracts/boundary-mapper.ts"; import type { ApiOperation } from "../../../src/contracts/api-operations.ts"; import { createRealtimePolicyRegistry, defineEventTypeId, defineExternalEventEffectProfileId, defineRealtimeEndpointId, defineRealtimeKillSwitchId, defineStreamRegistrationId, type RealtimeEventTypeRegistration, type RealtimeLimits, type RealtimePolicyRegistry, type RealtimeRecoveryProfile, type RealtimeStreamRegistration, } from "../../../src/contracts/realtime-streams.ts"; import type { RuntimeSchemaCodec } from "../../../src/contracts/schema-registry.ts"; import { createRealtimeEventCodec, type RealtimeEventCodec, } from "../../../src/adapters/realtime/event-codec.ts"; export const STREAM_ID = defineStreamRegistrationId("REFERENCE_STREAM"); export const EVENT_TYPE = defineEventTypeId("REFERENCE_CHANGED"); export const ENDPOINT_ID = defineRealtimeEndpointId("REFERENCE_ENDPOINT"); export const EFFECT_PROFILE_ID = defineExternalEventEffectProfileId("REFERENCE_INVALIDATE"); export const KILL_SWITCH_ID = defineRealtimeKillSwitchId("REFERENCE_KILL_SWITCH"); export const TEST_LIMITS: RealtimeLimits = Object.freeze({ maxEventBytes: 4_096, maxPayloadDepth: 8, maxPayloadNodes: 128, maxQueueEvents: 8, maxQueueBytes: 32_768, maxDedupeEntries: 16, maxDedupeBytes: 32_768, dedupeTtlMs: 60_000, }); const eventPayloadCodec: RuntimeSchemaCodec = Object.freeze({ schemaId: "ReferenceRealtimePayload", parse(value) { if ( !value || typeof value !== "object" || Array.isArray(value) || Object.keys(value).length !== 1 || typeof (value as Readonly>).value !== "string" ) { return { success: false, issues: [{ path: "value", code: "INVALID_TYPE" }], }; } return { success: true, data: { value: (value as Readonly>).value, }, }; }, }); const checkpointCodec: RuntimeSchemaCodec = Object.freeze({ schemaId: "ReferenceRealtimeCheckpoint", parse: (value) => ({ success: true, data: value }), }); export const TEST_SCHEMA_CODECS = Object.freeze({ ReferenceRealtimePayload: eventPayloadCodec, ReferenceRealtimeCheckpoint: checkpointCodec, }); export const TEST_MAPPER: InstalledBoundaryMapper = Object.freeze({ mapperId: "ReferenceRealtimeMapper", mapperVersion: 1, inputSchemaId: "ReferenceRealtimePayload", outputContractId: "ReferenceRealtimeEvent", owner: "sample-owner", maxOutputItems: 1, map(input) { if ( !input || typeof input !== "object" || typeof (input as Readonly>).value !== "string" ) { return { ok: false, code: "MAPPING_INVARIANT_REJECTED" }; } return mappingSuccess( Object.freeze({ value: (input as Readonly>).value, }), ); }, }); export const TEST_MAPPERS = Object.freeze({ ReferenceRealtimeMapper: TEST_MAPPER, }); const snapshotOperation: ApiOperation = Object.freeze({ method: "GET", path: "/api/reference-snapshot", operationId: "GET_REFERENCE_REALTIME_SNAPSHOT", auth: "external-session", timeoutMs: 10_000, idempotency: "safe", retry: "never", requestSource: "none", requestSchema: "NoRequest", responseSchema: "ReferenceRealtimeCheckpoint", owner: "sample-owner", contractVersion: 2, protocol: "REST", semantics: "QUERY", replayPolicy: "SAFE", idempotencyKeyPolicy: "NONE", mapperId: "ReferenceRealtimeSnapshotMapper", successStatuses: [200], responseMediaTypes: ["application/json"], maxResponseBytes: 16_384, providerId: "PRIMARY_API", authProfileId: "EXTERNAL_SESSION", csrfProfileId: "NONE", pathSchema: "NoRequest", pathParameterNames: [], maxEncodedSearchBytes: 0, }); export const TEST_API_OPERATIONS = Object.freeze({ GET_REFERENCE_REALTIME_SNAPSHOT: snapshotOperation, }); export type TestRegistryOptions = Readonly<{ recovery?: RealtimeRecoveryProfile; delivery?: RealtimeStreamRegistration["delivery"]; stateBearing?: boolean; limits?: RealtimeLimits; streamMutator?: ( stream: RealtimeStreamRegistration, ) => RealtimeStreamRegistration; eventTypeMutator?: ( eventType: RealtimeEventTypeRegistration, ) => RealtimeEventTypeRegistration; }>; export function createTestRealtimeRegistry( options: TestRegistryOptions = {}, ): RealtimePolicyRegistry { const recovery = options.recovery ?? ({ mode: "CURSOR", snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT", checkpointCodecId: "ReferenceRealtimeCheckpoint", barrier: "REPLAY", } as const); const eventType: RealtimeEventTypeRegistration = { id: EVENT_TYPE, owner: "sample-owner", payloadSchemaId: "ReferenceRealtimePayload", mapperId: "ReferenceRealtimeMapper", effectProfileId: EFFECT_PROFILE_ID, stateBearing: options.stateBearing ?? true, }; const stream: RealtimeStreamRegistration = { id: STREAM_ID, protocol: "REALTIME_EVENT_V1", owner: "sample-owner", scope: "ACCOUNT_BOUND", primaryTransport: "SSE", endpointId: ENDPOINT_ID, eventTypeIds: [EVENT_TYPE], delivery: options.delivery ?? "AUTHORITATIVE_DELTA", recovery, fallback: recovery.mode === "SESSION_REBUILD" ? "EXPLICITLY_STALE" : "BOUNDED_POLLING", hiddenPolicy: "CLOSE", limits: options.limits ?? TEST_LIMITS, killSwitchId: KILL_SWITCH_ID, }; return createRealtimePolicyRegistry({ streams: [options.streamMutator?.(stream) ?? stream], eventTypes: [ options.eventTypeMutator?.(eventType) ?? eventType, ], bindings: { schemaCodecs: TEST_SCHEMA_CODECS, mappers: TEST_MAPPERS, apiOperations: TEST_API_OPERATIONS, endpointIds: [ENDPOINT_ID], effectProfileIds: [EFFECT_PROFILE_ID], killSwitchIds: [KILL_SWITCH_ID], rebuildInputIds: ["referenceRealtimeRebuild"], }, }); } export function createTestRealtimeCodec( registry = createTestRealtimeRegistry(), ): RealtimeEventCodec { return createRealtimeEventCodec({ registry, schemaCodecs: TEST_SCHEMA_CODECS, }); } export type EventOverrides = Readonly<{ protocol?: unknown; streamId?: unknown; streamEpoch?: unknown; eventType?: unknown; eventId?: unknown; sequence?: unknown; recoveryMode?: unknown; resumeCursor?: unknown; occurredAt?: unknown; scopeBinding?: unknown; payload?: unknown; }>; export function realtimeEventValue( overrides: EventOverrides = {}, ): Readonly> { return { protocol: "REALTIME_EVENT_V1", streamId: STREAM_ID, streamEpoch: "stream-epoch-0001", eventType: EVENT_TYPE, eventId: "event-00000001", sequence: "1", recoveryMode: "CURSOR", resumeCursor: "cursor-00000001", occurredAt: "2026-07-28T01:02:03.123Z", scopeBinding: "scope-binding-0001", payload: { value: "changed" }, ...overrides, }; } export function realtimeEventJson( overrides: EventOverrides = {}, ): string { return JSON.stringify(realtimeEventValue(overrides)); }