import type { RealtimeTransportEventOutcome, } from "../../application/ports/realtime/event-authority.ts"; import type { RealtimeFailure, RealtimeFailureKind, RealtimeOperation, RealtimeResult, } from "../../application/ports/realtime/shared.ts"; import { REALTIME_FAILURE_KINDS, REALTIME_OPERATIONS, } from "../../application/ports/realtime/shared.ts"; import { isCanonicalRealtimeSequence, isRealtimeOpaqueIdentifier, isRealtimeResumeCursor, } from "../../contracts/realtime-events.ts"; export type { RealtimeFailure, RealtimeFailureKind, RealtimeOperation, RealtimeResult, } from "../../application/ports/realtime/shared.ts"; const DEFAULT_RETRYABLE = new Set([ "OFFLINE", "CONNECT_TIMEOUT", "IDLE_TIMEOUT", "RATE_LIMITED", "PROVIDER_UNAVAILABLE", ]); const FAILURE_KINDS = new Set(REALTIME_FAILURE_KINDS); const OPERATIONS = new Set(REALTIME_OPERATIONS); export type RealtimeDataSnapshot = Readonly<{ keys: readonly string[]; values: Readonly>; frozen: boolean; }>; export function realtimeSuccess( value: Value, ): Extract, { ok: true }> { return Object.freeze({ ok: true, value }); } export function realtimeFailure( kind: RealtimeFailureKind, operation: RealtimeOperation, retryable = DEFAULT_RETRYABLE.has(kind), ): Extract, { ok: false }> { return Object.freeze({ ok: false, error: Object.freeze({ kind, operation, retryable, } satisfies RealtimeFailure), }); } export function isRealtimeFailure( value: unknown, ): value is RealtimeFailure { return parseRealtimeFailure(value, true) !== null; } /** * Captures an external result through own data descriptors exactly once and * returns a new canonical value. Callers that need to use the validated fields * must use this returned snapshot rather than reading the source again. */ export function snapshotRealtimeResult( value: unknown, isValue: (candidate: unknown) => candidate is Value, ): RealtimeResult | null { return parseRealtimeResult(value, isValue, false); } export function isRealtimeResult( value: unknown, isValue: (candidate: unknown) => candidate is Value, ): value is RealtimeResult { return parseRealtimeResult(value, isValue, true) !== null; } export function isRealtimeTransportEventOutcome( value: unknown, ): value is RealtimeTransportEventOutcome { const snapshot = captureRealtimeDataSnapshot(value); if (!snapshot || !snapshot.frozen) { return false; } if (snapshot.values.kind === "CONTINUE") { return hasExactSnapshotKeys(snapshot, ["kind"]); } if ( snapshot.values.kind !== "RECOVERY_COMMITTED" || !hasExactSnapshotKeys(snapshot, [ "checkpoint", "kind", "streamId", ]) || typeof snapshot.values.streamId !== "string" ) { return false; } const checkpoint = captureRealtimeDataSnapshot( snapshot.values.checkpoint, ); return ( checkpoint !== null && checkpoint.frozen && hasExactSnapshotKeys(checkpoint, [ "lastAppliedSequence", "recoveryMode", "resumeCursor", "streamEpoch", ]) && isRealtimeOpaqueIdentifier(snapshot.values.streamId) && isRealtimeOpaqueIdentifier(checkpoint.values.streamEpoch) && isCanonicalRealtimeSequence( checkpoint.values.lastAppliedSequence, ) && (checkpoint.values.recoveryMode === "CURSOR" ? isRealtimeResumeCursor(checkpoint.values.resumeCursor) : (checkpoint.values.recoveryMode === "SNAPSHOT_ONLY" || checkpoint.values.recoveryMode === "SESSION_REBUILD") && checkpoint.values.resumeCursor === null) ); } /** * Reads a plain record without invoking property accessors. Symbol keys, * inherited shapes, non-enumerable fields and accessors are rejected. The * returned null-prototype value map is immutable and detached from later * property reads on the source object. */ export function captureRealtimeDataSnapshot( value: unknown, ): RealtimeDataSnapshot | null { try { if ( !value || typeof value !== "object" || Array.isArray(value) ) { return null; } const prototype = Object.getPrototypeOf(value); if ( prototype !== Object.prototype && prototype !== null ) { return null; } const extensible = Object.isExtensible(value); const descriptors = Object.getOwnPropertyDescriptors(value); const ownKeys = Reflect.ownKeys(descriptors); if (ownKeys.some((key) => typeof key !== "string")) { return null; } const keys = (ownKeys as string[]).sort(); const values = Object.create(null) as Record; let frozen = !extensible; for (const key of keys) { const descriptor = descriptors[key]; if ( !descriptor || !Object.hasOwn(descriptor, "value") || descriptor.enumerable !== true ) { return null; } Object.defineProperty(values, key, { configurable: false, enumerable: true, value: descriptor.value, writable: false, }); frozen = frozen && descriptor.configurable === false && descriptor.writable === false; } return Object.freeze({ keys: Object.freeze(keys), values: Object.freeze(values), frozen, }); } catch { return null; } } function parseRealtimeResult( value: unknown, isValue: (candidate: unknown) => candidate is Value, requireFrozenSource: boolean, ): RealtimeResult | null { const snapshot = captureRealtimeDataSnapshot(value); if ( !snapshot || (requireFrozenSource && !snapshot.frozen) ) { return null; } if ( snapshot.values.ok === true && hasExactSnapshotKeys(snapshot, ["ok", "value"]) ) { let accepted: boolean; try { accepted = isValue(snapshot.values.value); } catch { return null; } return accepted ? realtimeSuccess(snapshot.values.value as Value) : null; } if ( snapshot.values.ok !== false || !hasExactSnapshotKeys(snapshot, ["error", "ok"]) ) { return null; } const failure = parseRealtimeFailure( snapshot.values.error, requireFrozenSource, ); return failure ? realtimeFailure( failure.kind, failure.operation, failure.retryable, ) : null; } function parseRealtimeFailure( value: unknown, requireFrozenSource: boolean, ): RealtimeFailure | null { const snapshot = captureRealtimeDataSnapshot(value); if ( !snapshot || (requireFrozenSource && !snapshot.frozen) || !hasExactSnapshotKeys(snapshot, [ "kind", "operation", "retryable", ]) || !FAILURE_KINDS.has(snapshot.values.kind) || !OPERATIONS.has(snapshot.values.operation) || typeof snapshot.values.retryable !== "boolean" ) { return null; } return Object.freeze({ kind: snapshot.values.kind as RealtimeFailureKind, operation: snapshot.values.operation as RealtimeOperation, retryable: snapshot.values.retryable, }); } function hasExactSnapshotKeys( snapshot: RealtimeDataSnapshot, expectedKeys: readonly string[], ): boolean { const expected = [...expectedKeys].sort(); return ( snapshot.keys.length === expected.length && snapshot.keys.every((key, index) => key === expected[index]) ); }