549 lines
16 KiB
TypeScript
549 lines
16 KiB
TypeScript
import type { ApiOperation } from "./api-operations.ts";
|
|
import type { InstalledBoundaryMapper } from "./boundary-mapper.ts";
|
|
import type { RuntimeSchemaCodec } from "./schema-registry.ts";
|
|
|
|
declare const streamRegistrationIdBrand: unique symbol;
|
|
declare const eventTypeIdBrand: unique symbol;
|
|
declare const realtimeEndpointIdBrand: unique symbol;
|
|
declare const externalEventEffectProfileIdBrand: unique symbol;
|
|
declare const killSwitchIdBrand: unique symbol;
|
|
|
|
export type StreamRegistrationId = string &
|
|
Readonly<{ [streamRegistrationIdBrand]: true }>;
|
|
export type EventTypeId = string &
|
|
Readonly<{ [eventTypeIdBrand]: true }>;
|
|
export type RealtimeEndpointId = string &
|
|
Readonly<{ [realtimeEndpointIdBrand]: true }>;
|
|
export type ExternalEventEffectProfileId = string &
|
|
Readonly<{ [externalEventEffectProfileIdBrand]: true }>;
|
|
export type KillSwitchId = string &
|
|
Readonly<{ [killSwitchIdBrand]: true }>;
|
|
|
|
export const REALTIME_EVENT_PROTOCOL = "REALTIME_EVENT_V1" as const;
|
|
|
|
export const REALTIME_HARD_LIMITS = Object.freeze({
|
|
maxEventBytes: 64 * 1024,
|
|
maxPayloadDepth: 16,
|
|
maxPayloadNodes: 4_096,
|
|
maxQueueEvents: 256,
|
|
maxQueueBytes: 4 * 1024 * 1024,
|
|
maxDedupeEntries: 2_048,
|
|
maxDedupeBytes: 4 * 1024 * 1024,
|
|
dedupeTtlMs: 10 * 60 * 1_000,
|
|
});
|
|
|
|
export type RealtimeLimits = Readonly<{
|
|
maxEventBytes: number;
|
|
maxPayloadDepth: number;
|
|
maxPayloadNodes: number;
|
|
maxQueueEvents: number;
|
|
maxQueueBytes: number;
|
|
maxDedupeEntries: number;
|
|
/**
|
|
* Additional implementation memory ceiling for semantic conflict
|
|
* fingerprints. Reaching it has the same recovery meaning as exhausting the
|
|
* count/time dedupe window.
|
|
*/
|
|
maxDedupeBytes: number;
|
|
dedupeTtlMs: number;
|
|
}>;
|
|
|
|
export type RealtimeRecoveryProfile =
|
|
| Readonly<{
|
|
mode: "CURSOR";
|
|
snapshotOperationId: string;
|
|
checkpointCodecId: string;
|
|
barrier: "REPLAY";
|
|
}>
|
|
| Readonly<{
|
|
mode: "SNAPSHOT_ONLY";
|
|
snapshotOperationId: string;
|
|
checkpointCodecId: string;
|
|
barrier: "CONNECT_BUFFER" | "SERVER_HOLD" | "NONE";
|
|
}>
|
|
| Readonly<{
|
|
mode: "SESSION_REBUILD";
|
|
rebuildInputId: string;
|
|
}>;
|
|
|
|
export type RealtimeStreamRegistration = Readonly<{
|
|
id: StreamRegistrationId;
|
|
protocol: typeof REALTIME_EVENT_PROTOCOL;
|
|
owner: string;
|
|
scope: "ORIGIN_SHARED" | "ACCOUNT_BOUND" | "SESSION_BOUND";
|
|
primaryTransport: "SSE" | "WEBSOCKET" | "NONE";
|
|
endpointId: RealtimeEndpointId;
|
|
eventTypeIds: readonly EventTypeId[];
|
|
delivery: "INVALIDATION_HINT" | "AUTHORITATIVE_DELTA" | "EPHEMERAL";
|
|
recovery: RealtimeRecoveryProfile;
|
|
fallback: "BOUNDED_POLLING" | "EXPLICITLY_STALE";
|
|
hiddenPolicy: "CLOSE" | "BOUNDED_GRACE";
|
|
limits: RealtimeLimits;
|
|
killSwitchId: KillSwitchId;
|
|
}>;
|
|
|
|
export type RealtimeEventTypeRegistration = Readonly<{
|
|
id: EventTypeId;
|
|
owner: string;
|
|
payloadSchemaId: string;
|
|
mapperId: string;
|
|
effectProfileId: ExternalEventEffectProfileId;
|
|
stateBearing: boolean;
|
|
}>;
|
|
|
|
export type RealtimePolicyRegistryBindings = Readonly<{
|
|
schemaCodecs: Readonly<Record<string, RuntimeSchemaCodec>>;
|
|
mappers: Readonly<Record<string, InstalledBoundaryMapper>>;
|
|
apiOperations: Readonly<Record<string, ApiOperation>>;
|
|
endpointIds: readonly RealtimeEndpointId[];
|
|
effectProfileIds: readonly ExternalEventEffectProfileId[];
|
|
killSwitchIds: readonly KillSwitchId[];
|
|
rebuildInputIds?: readonly string[];
|
|
}>;
|
|
|
|
export type RealtimePolicyRegistry = Readonly<{
|
|
findStream(id: string): RealtimeStreamRegistration | undefined;
|
|
findEventType(id: string): RealtimeEventTypeRegistration | undefined;
|
|
findStreamEventType(
|
|
streamId: string,
|
|
eventTypeId: string,
|
|
): RealtimeEventTypeRegistration | undefined;
|
|
listStreams(): readonly RealtimeStreamRegistration[];
|
|
listEventTypes(): readonly RealtimeEventTypeRegistration[];
|
|
}>;
|
|
|
|
export type RealtimePolicyRegistryInput = Readonly<{
|
|
streams: readonly RealtimeStreamRegistration[];
|
|
eventTypes: readonly RealtimeEventTypeRegistration[];
|
|
bindings: RealtimePolicyRegistryBindings;
|
|
}>;
|
|
|
|
const REGISTRY_ID = /^[A-Z][A-Z0-9_]{2,79}$/u;
|
|
const OWNED_ID = /^[A-Za-z][A-Za-z0-9._:-]{0,127}$/u;
|
|
const OWNER = /^[a-z0-9][a-z0-9._:-]{0,127}$/u;
|
|
|
|
const STREAM_KEYS = Object.freeze([
|
|
"delivery",
|
|
"endpointId",
|
|
"eventTypeIds",
|
|
"fallback",
|
|
"hiddenPolicy",
|
|
"id",
|
|
"killSwitchId",
|
|
"limits",
|
|
"owner",
|
|
"primaryTransport",
|
|
"protocol",
|
|
"recovery",
|
|
"scope",
|
|
] as const);
|
|
const EVENT_TYPE_KEYS = Object.freeze([
|
|
"effectProfileId",
|
|
"id",
|
|
"mapperId",
|
|
"owner",
|
|
"payloadSchemaId",
|
|
"stateBearing",
|
|
] as const);
|
|
const LIMIT_KEYS = Object.freeze([
|
|
"dedupeTtlMs",
|
|
"maxDedupeBytes",
|
|
"maxDedupeEntries",
|
|
"maxEventBytes",
|
|
"maxPayloadDepth",
|
|
"maxPayloadNodes",
|
|
"maxQueueBytes",
|
|
"maxQueueEvents",
|
|
] as const);
|
|
|
|
export function defineStreamRegistrationId(
|
|
value: string,
|
|
): StreamRegistrationId {
|
|
return defineRegistryId(value, "stream") as StreamRegistrationId;
|
|
}
|
|
|
|
export function defineEventTypeId(value: string): EventTypeId {
|
|
return defineRegistryId(value, "event type") as EventTypeId;
|
|
}
|
|
|
|
export function defineRealtimeEndpointId(
|
|
value: string,
|
|
): RealtimeEndpointId {
|
|
return defineRegistryId(value, "endpoint") as RealtimeEndpointId;
|
|
}
|
|
|
|
export function defineExternalEventEffectProfileId(
|
|
value: string,
|
|
): ExternalEventEffectProfileId {
|
|
return defineRegistryId(
|
|
value,
|
|
"effect profile",
|
|
) as ExternalEventEffectProfileId;
|
|
}
|
|
|
|
export function defineRealtimeKillSwitchId(value: string): KillSwitchId {
|
|
return defineRegistryId(value, "kill switch") as KillSwitchId;
|
|
}
|
|
|
|
/**
|
|
* Builds a composition-time registry and retains no caller-owned registration
|
|
* object or array.
|
|
*/
|
|
export function createRealtimePolicyRegistry(
|
|
input: RealtimePolicyRegistryInput,
|
|
): RealtimePolicyRegistry {
|
|
if (
|
|
!input ||
|
|
typeof input !== "object" ||
|
|
!Array.isArray(input.streams) ||
|
|
input.streams.length < 1 ||
|
|
input.streams.length > 128 ||
|
|
!Array.isArray(input.eventTypes) ||
|
|
input.eventTypes.length < 1 ||
|
|
input.eventTypes.length > 512
|
|
) {
|
|
throw new TypeError("Realtime policy registry is invalid.");
|
|
}
|
|
|
|
const endpointIds = identifierSet(
|
|
input.bindings.endpointIds,
|
|
"endpoint",
|
|
);
|
|
const effectProfileIds = identifierSet(
|
|
input.bindings.effectProfileIds,
|
|
"effect profile",
|
|
);
|
|
const killSwitchIds = identifierSet(
|
|
input.bindings.killSwitchIds,
|
|
"kill switch",
|
|
);
|
|
const rebuildInputIds = ownedIdentifierSet(
|
|
input.bindings.rebuildInputIds ?? [],
|
|
"rebuild input",
|
|
);
|
|
|
|
const eventTypes = new Map<EventTypeId, RealtimeEventTypeRegistration>();
|
|
for (const candidate of input.eventTypes) {
|
|
const registration = snapshotEventType(
|
|
candidate,
|
|
input.bindings,
|
|
effectProfileIds,
|
|
);
|
|
if (eventTypes.has(registration.id)) {
|
|
throw new TypeError("Realtime event type is duplicated.");
|
|
}
|
|
eventTypes.set(registration.id, registration);
|
|
}
|
|
|
|
const streams = new Map<StreamRegistrationId, RealtimeStreamRegistration>();
|
|
const referencedEventTypes = new Set<EventTypeId>();
|
|
for (const candidate of input.streams) {
|
|
const registration = snapshotStream(
|
|
candidate,
|
|
input.bindings,
|
|
eventTypes,
|
|
endpointIds,
|
|
killSwitchIds,
|
|
rebuildInputIds,
|
|
);
|
|
if (streams.has(registration.id)) {
|
|
throw new TypeError("Realtime stream is duplicated.");
|
|
}
|
|
streams.set(registration.id, registration);
|
|
for (const eventTypeId of registration.eventTypeIds) {
|
|
referencedEventTypes.add(eventTypeId);
|
|
}
|
|
}
|
|
|
|
if (
|
|
[...eventTypes.keys()].some(
|
|
(eventTypeId) => !referencedEventTypes.has(eventTypeId),
|
|
)
|
|
) {
|
|
throw new TypeError("Realtime event type is not owned by a stream.");
|
|
}
|
|
|
|
const streamList = Object.freeze([...streams.values()]);
|
|
const eventTypeList = Object.freeze([...eventTypes.values()]);
|
|
|
|
return Object.freeze({
|
|
findStream(id: string) {
|
|
return streams.get(id as StreamRegistrationId);
|
|
},
|
|
findEventType(id: string) {
|
|
return eventTypes.get(id as EventTypeId);
|
|
},
|
|
findStreamEventType(streamId: string, eventTypeId: string) {
|
|
const stream = streams.get(streamId as StreamRegistrationId);
|
|
if (!stream || !stream.eventTypeIds.includes(eventTypeId as EventTypeId)) {
|
|
return undefined;
|
|
}
|
|
return eventTypes.get(eventTypeId as EventTypeId);
|
|
},
|
|
listStreams: () => streamList,
|
|
listEventTypes: () => eventTypeList,
|
|
});
|
|
}
|
|
|
|
function snapshotEventType(
|
|
input: RealtimeEventTypeRegistration,
|
|
bindings: RealtimePolicyRegistryBindings,
|
|
effectProfileIds: ReadonlySet<string>,
|
|
): RealtimeEventTypeRegistration {
|
|
if (
|
|
!hasExactKeys(input, EVENT_TYPE_KEYS) ||
|
|
!REGISTRY_ID.test(input.id) ||
|
|
!OWNER.test(input.owner) ||
|
|
!OWNED_ID.test(input.payloadSchemaId) ||
|
|
!OWNED_ID.test(input.mapperId) ||
|
|
!REGISTRY_ID.test(input.effectProfileId) ||
|
|
typeof input.stateBearing !== "boolean" ||
|
|
bindings.schemaCodecs[input.payloadSchemaId]?.schemaId !==
|
|
input.payloadSchemaId ||
|
|
bindings.mappers[input.mapperId]?.mapperId !== input.mapperId ||
|
|
bindings.mappers[input.mapperId]?.inputSchemaId !==
|
|
input.payloadSchemaId ||
|
|
!effectProfileIds.has(input.effectProfileId)
|
|
) {
|
|
throw new TypeError("Realtime event type registration is invalid.");
|
|
}
|
|
return Object.freeze({ ...input });
|
|
}
|
|
|
|
function snapshotStream(
|
|
input: RealtimeStreamRegistration,
|
|
bindings: RealtimePolicyRegistryBindings,
|
|
eventTypes: ReadonlyMap<EventTypeId, RealtimeEventTypeRegistration>,
|
|
endpointIds: ReadonlySet<string>,
|
|
killSwitchIds: ReadonlySet<string>,
|
|
rebuildInputIds: ReadonlySet<string>,
|
|
): RealtimeStreamRegistration {
|
|
if (
|
|
!hasExactKeys(input, STREAM_KEYS) ||
|
|
!REGISTRY_ID.test(input.id) ||
|
|
input.protocol !== REALTIME_EVENT_PROTOCOL ||
|
|
!OWNER.test(input.owner) ||
|
|
!["ORIGIN_SHARED", "ACCOUNT_BOUND", "SESSION_BOUND"].includes(
|
|
input.scope,
|
|
) ||
|
|
!["SSE", "WEBSOCKET", "NONE"].includes(input.primaryTransport) ||
|
|
!REGISTRY_ID.test(input.endpointId) ||
|
|
!endpointIds.has(input.endpointId) ||
|
|
!Array.isArray(input.eventTypeIds) ||
|
|
input.eventTypeIds.length < 1 ||
|
|
input.eventTypeIds.length > 128 ||
|
|
new Set(input.eventTypeIds).size !== input.eventTypeIds.length ||
|
|
input.eventTypeIds.some(
|
|
(eventTypeId) =>
|
|
!REGISTRY_ID.test(eventTypeId) || !eventTypes.has(eventTypeId),
|
|
) ||
|
|
!["INVALIDATION_HINT", "AUTHORITATIVE_DELTA", "EPHEMERAL"].includes(
|
|
input.delivery,
|
|
) ||
|
|
!["BOUNDED_POLLING", "EXPLICITLY_STALE"].includes(input.fallback) ||
|
|
!["CLOSE", "BOUNDED_GRACE"].includes(input.hiddenPolicy) ||
|
|
!REGISTRY_ID.test(input.killSwitchId) ||
|
|
!killSwitchIds.has(input.killSwitchId)
|
|
) {
|
|
throw new TypeError("Realtime stream registration is invalid.");
|
|
}
|
|
|
|
const limits = snapshotLimits(input.limits);
|
|
const recovery = snapshotRecovery(
|
|
input.recovery,
|
|
bindings,
|
|
rebuildInputIds,
|
|
);
|
|
const selectedEventTypes = input.eventTypeIds.map((eventTypeId) => {
|
|
const selected = eventTypes.get(eventTypeId);
|
|
if (!selected) {
|
|
throw new TypeError("Realtime stream event type is unresolved.");
|
|
}
|
|
return selected;
|
|
});
|
|
const hasStateBearingEvent = selectedEventTypes.some(
|
|
(eventType) => eventType.stateBearing,
|
|
);
|
|
const hasNonStateBearingEvent = selectedEventTypes.some(
|
|
(eventType) => !eventType.stateBearing,
|
|
);
|
|
|
|
if (
|
|
(hasStateBearingEvent && recovery.mode === "SESSION_REBUILD") ||
|
|
(hasStateBearingEvent &&
|
|
recovery.mode === "SNAPSHOT_ONLY" &&
|
|
recovery.barrier === "NONE") ||
|
|
(input.delivery === "EPHEMERAL" && hasStateBearingEvent) ||
|
|
(input.delivery === "AUTHORITATIVE_DELTA" &&
|
|
hasNonStateBearingEvent) ||
|
|
(recovery.mode === "SESSION_REBUILD" &&
|
|
input.delivery !== "EPHEMERAL") ||
|
|
(input.delivery === "EPHEMERAL" &&
|
|
input.fallback === "BOUNDED_POLLING")
|
|
) {
|
|
throw new TypeError("Realtime stream recovery contract is contradictory.");
|
|
}
|
|
|
|
return Object.freeze({
|
|
...input,
|
|
eventTypeIds: Object.freeze([...input.eventTypeIds]),
|
|
recovery,
|
|
limits,
|
|
});
|
|
}
|
|
|
|
function snapshotRecovery(
|
|
input: RealtimeRecoveryProfile,
|
|
bindings: RealtimePolicyRegistryBindings,
|
|
rebuildInputIds: ReadonlySet<string>,
|
|
): RealtimeRecoveryProfile {
|
|
if (!input || typeof input !== "object") {
|
|
throw new TypeError("Realtime recovery profile is invalid.");
|
|
}
|
|
if (input.mode === "CURSOR") {
|
|
if (
|
|
!hasExactKeys(input, [
|
|
"barrier",
|
|
"checkpointCodecId",
|
|
"mode",
|
|
"snapshotOperationId",
|
|
]) ||
|
|
input.barrier !== "REPLAY" ||
|
|
!validSnapshotBindings(input, bindings)
|
|
) {
|
|
throw new TypeError("Realtime cursor recovery profile is invalid.");
|
|
}
|
|
return Object.freeze({ ...input });
|
|
}
|
|
if (input.mode === "SNAPSHOT_ONLY") {
|
|
if (
|
|
!hasExactKeys(input, [
|
|
"barrier",
|
|
"checkpointCodecId",
|
|
"mode",
|
|
"snapshotOperationId",
|
|
]) ||
|
|
!["CONNECT_BUFFER", "SERVER_HOLD", "NONE"].includes(input.barrier) ||
|
|
!validSnapshotBindings(input, bindings)
|
|
) {
|
|
throw new TypeError("Realtime snapshot recovery profile is invalid.");
|
|
}
|
|
return Object.freeze({ ...input });
|
|
}
|
|
if (
|
|
input.mode !== "SESSION_REBUILD" ||
|
|
!hasExactKeys(input, ["mode", "rebuildInputId"]) ||
|
|
!OWNED_ID.test(input.rebuildInputId) ||
|
|
!rebuildInputIds.has(input.rebuildInputId)
|
|
) {
|
|
throw new TypeError("Realtime session rebuild profile is invalid.");
|
|
}
|
|
return Object.freeze({ ...input });
|
|
}
|
|
|
|
function validSnapshotBindings(
|
|
input: Readonly<{
|
|
snapshotOperationId: string;
|
|
checkpointCodecId: string;
|
|
}>,
|
|
bindings: RealtimePolicyRegistryBindings,
|
|
): boolean {
|
|
const operation = bindings.apiOperations[input.snapshotOperationId];
|
|
return (
|
|
OWNED_ID.test(input.snapshotOperationId) &&
|
|
OWNED_ID.test(input.checkpointCodecId) &&
|
|
bindings.schemaCodecs[input.checkpointCodecId]?.schemaId ===
|
|
input.checkpointCodecId &&
|
|
operation?.contractVersion === 2 &&
|
|
operation.protocol === "REST" &&
|
|
operation.semantics === "QUERY" &&
|
|
(operation.replayPolicy === "SAFE" ||
|
|
operation.replayPolicy === "IDEMPOTENT")
|
|
);
|
|
}
|
|
|
|
function snapshotLimits(input: RealtimeLimits): RealtimeLimits {
|
|
if (!hasExactKeys(input, LIMIT_KEYS)) {
|
|
throw new TypeError("Realtime limits are invalid.");
|
|
}
|
|
for (const key of LIMIT_KEYS) {
|
|
const value = input[key];
|
|
if (
|
|
!Number.isSafeInteger(value) ||
|
|
value < 1 ||
|
|
value > REALTIME_HARD_LIMITS[key]
|
|
) {
|
|
throw new TypeError("Realtime limits exceed implementation ceilings.");
|
|
}
|
|
}
|
|
if (
|
|
input.maxQueueBytes < input.maxEventBytes ||
|
|
input.maxDedupeBytes < input.maxEventBytes
|
|
) {
|
|
throw new TypeError("Realtime memory limits cannot hold one event.");
|
|
}
|
|
return Object.freeze({ ...input });
|
|
}
|
|
|
|
function identifierSet(
|
|
values: readonly string[],
|
|
label: string,
|
|
): ReadonlySet<string> {
|
|
if (!Array.isArray(values)) {
|
|
throw new TypeError(`Realtime ${label} bindings are invalid.`);
|
|
}
|
|
const result = new Set<string>();
|
|
for (const value of values) {
|
|
if (!REGISTRY_ID.test(value) || result.has(value)) {
|
|
throw new TypeError(`Realtime ${label} bindings are invalid.`);
|
|
}
|
|
result.add(value);
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function ownedIdentifierSet(
|
|
values: readonly string[],
|
|
label: string,
|
|
): ReadonlySet<string> {
|
|
if (!Array.isArray(values)) {
|
|
throw new TypeError(`Realtime ${label} bindings are invalid.`);
|
|
}
|
|
const result = new Set<string>();
|
|
for (const value of values) {
|
|
if (!OWNED_ID.test(value) || result.has(value)) {
|
|
throw new TypeError(`Realtime ${label} bindings are invalid.`);
|
|
}
|
|
result.add(value);
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function defineRegistryId(value: string, label: string): string {
|
|
if (!REGISTRY_ID.test(value)) {
|
|
throw new TypeError(`Realtime ${label} ID is invalid.`);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function hasExactKeys(
|
|
value: unknown,
|
|
expected: readonly string[],
|
|
): value is Readonly<Record<string, unknown>> {
|
|
if (
|
|
!value ||
|
|
typeof value !== "object" ||
|
|
Array.isArray(value) ||
|
|
(Object.getPrototypeOf(value) !== Object.prototype &&
|
|
Object.getPrototypeOf(value) !== null)
|
|
) {
|
|
return false;
|
|
}
|
|
const keys = Object.keys(value).sort();
|
|
const selected = [...expected].sort();
|
|
return (
|
|
keys.length === selected.length &&
|
|
keys.every((key, index) => key === selected[index])
|
|
);
|
|
}
|