chore: initialize from frontend template 4dc033c

This commit is contained in:
DongHyeonka
2026-08-13 18:23:26 +09:00
commit 40107eec84
897 changed files with 234824 additions and 0 deletions
+394
View File
@@ -0,0 +1,394 @@
import {
realtimeFailure,
realtimeSuccess,
type RealtimeResult,
} from "./result.ts";
import {
isCanonicalRealtimeSequence,
isRealtimeOpaqueIdentifier,
isRealtimeResumeCursor,
isRealtimeScopeBinding,
isStrictRealtimeTimestamp,
type RealtimeEventEnvelope,
} from "../../contracts/realtime-events.ts";
import {
REALTIME_EVENT_PROTOCOL,
REALTIME_HARD_LIMITS,
type RealtimeEventTypeRegistration,
type RealtimePolicyRegistry,
type RealtimeStreamRegistration,
} from "../../contracts/realtime-streams.ts";
import {
validateWithRuntimeSchemaRegistry,
type RuntimeSchemaCodec,
} from "../../contracts/schema-registry.ts";
import {
hasDuplicateJsonMembers,
} from "./json-member-scanner.ts";
export type ValidatedRealtimeEventDto = Readonly<{
envelope: RealtimeEventEnvelope;
wireBytes: number;
/**
* Adapter-private semantic identity used only by the bounded conflict
* detector. It must never be logged or projected into diagnostics.
*/
semanticFingerprint: string;
fingerprintBytes: number;
}>;
export type RealtimeEventCodec = Readonly<{
decode(raw: string): RealtimeResult<ValidatedRealtimeEventDto>;
}>;
export type RealtimeEventCodecDependencies = Readonly<{
registry: RealtimePolicyRegistry;
schemaCodecs: Readonly<Record<string, RuntimeSchemaCodec>>;
}>;
const ENVELOPE_KEYS = Object.freeze([
"eventId",
"eventType",
"occurredAt",
"payload",
"protocol",
"recoveryMode",
"resumeCursor",
"scopeBinding",
"sequence",
"streamEpoch",
"streamId",
] as const);
const FORBIDDEN_OBJECT_KEYS = new Set([
"__proto__",
"constructor",
"prototype",
]);
const encoder = new TextEncoder();
const issuedDtos = new WeakSet<object>();
export function createRealtimeEventCodec(
dependencies: RealtimeEventCodecDependencies,
): RealtimeEventCodec {
return Object.freeze({
decode(raw: string): RealtimeResult<ValidatedRealtimeEventDto> {
try {
return decodeUnsafe(raw, dependencies);
} catch {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
},
});
}
export function isValidatedRealtimeEventDto(
value: unknown,
): value is ValidatedRealtimeEventDto {
return (
!!value &&
typeof value === "object" &&
issuedDtos.has(value) &&
Object.isFrozen(value)
);
}
function decodeUnsafe(
raw: string,
dependencies: RealtimeEventCodecDependencies,
): RealtimeResult<ValidatedRealtimeEventDto> {
if (typeof raw !== "string") {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (
raw.length > REALTIME_HARD_LIMITS.maxEventBytes ||
encoder.encode(raw).byteLength > REALTIME_HARD_LIMITS.maxEventBytes
) {
return realtimeFailure("EVENT_TOO_LARGE", "DECODE");
}
const wireBytes = encoder.encode(raw).byteLength;
let input: unknown;
if (
hasDuplicateJsonMembers(raw, {
maxDepth: REALTIME_HARD_LIMITS.maxPayloadDepth + 2,
maxMembers: REALTIME_HARD_LIMITS.maxPayloadNodes + 32,
})
) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
try {
input = JSON.parse(raw);
} catch {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (!hasExactEnvelopeKeys(input)) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (input.protocol !== REALTIME_EVENT_PROTOCOL) {
return realtimeFailure("PROTOCOL_MISMATCH", "DECODE");
}
if (typeof input.streamId !== "string") {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
const stream = dependencies.registry.findStream(input.streamId);
if (!stream) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (wireBytes > stream.limits.maxEventBytes) {
return realtimeFailure("EVENT_TOO_LARGE", "DECODE");
}
if (typeof input.eventType !== "string") {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
const eventType = dependencies.registry.findStreamEventType(
stream.id,
input.eventType,
);
if (!eventType) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (!hasValidEnvelopeFields(input, stream)) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
if (
!withinJsonBudget(
input.payload,
stream.limits.maxPayloadDepth,
stream.limits.maxPayloadNodes,
)
) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
const payload = validateWithRuntimeSchemaRegistry(
eventType.payloadSchemaId,
input.payload,
dependencies.schemaCodecs,
);
if (!payload.success) {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
let payloadSnapshot: unknown;
try {
payloadSnapshot = snapshotJson(
payload.data,
stream.limits.maxPayloadDepth,
stream.limits.maxPayloadNodes,
);
} catch {
return realtimeFailure("MALFORMED_EVENT", "DECODE");
}
const envelope = createEnvelope(
input,
stream,
eventType,
payloadSnapshot,
);
const semanticFingerprint = canonicalJson(envelope);
const fingerprintBytes = encoder.encode(semanticFingerprint).byteLength;
if (fingerprintBytes > stream.limits.maxEventBytes) {
return realtimeFailure("EVENT_TOO_LARGE", "DECODE");
}
const dto = Object.freeze({
envelope,
wireBytes,
semanticFingerprint,
fingerprintBytes,
});
issuedDtos.add(dto);
return realtimeSuccess(dto);
}
function hasValidEnvelopeFields(
input: Readonly<Record<string, unknown>>,
stream: RealtimeStreamRegistration,
): boolean {
if (
!isRealtimeOpaqueIdentifier(input.streamEpoch) ||
!isRealtimeOpaqueIdentifier(input.eventId) ||
!isCanonicalRealtimeSequence(input.sequence) ||
!isStrictRealtimeTimestamp(input.occurredAt) ||
!isRealtimeScopeBinding(input.scopeBinding) ||
input.recoveryMode !== stream.recovery.mode
) {
return false;
}
return stream.recovery.mode === "CURSOR"
? isRealtimeResumeCursor(input.resumeCursor)
: input.resumeCursor === null;
}
function createEnvelope(
input: Readonly<Record<string, unknown>>,
stream: RealtimeStreamRegistration,
eventType: RealtimeEventTypeRegistration,
payload: unknown,
): RealtimeEventEnvelope {
const base = {
protocol: REALTIME_EVENT_PROTOCOL,
streamId: stream.id,
streamEpoch: input.streamEpoch as string,
eventType: eventType.id,
eventId: input.eventId as string,
sequence: input.sequence as string,
occurredAt: input.occurredAt as string,
scopeBinding: input.scopeBinding as string,
payload,
};
return stream.recovery.mode === "CURSOR"
? Object.freeze({
...base,
recoveryMode: "CURSOR" as const,
resumeCursor: input.resumeCursor as string,
})
: Object.freeze({
...base,
recoveryMode: stream.recovery.mode,
resumeCursor: null,
});
}
function hasExactEnvelopeKeys(
value: unknown,
): value is Readonly<Record<string, unknown>> {
if (
!value ||
typeof value !== "object" ||
Array.isArray(value) ||
Object.getPrototypeOf(value) !== Object.prototype
) {
return false;
}
const keys = Object.keys(value).sort();
return (
keys.length === ENVELOPE_KEYS.length &&
keys.every((key, index) => key === ENVELOPE_KEYS[index])
);
}
function withinJsonBudget(
value: unknown,
maxDepth: number,
maxNodes: number,
): boolean {
let nodes = 0;
const visit = (candidate: unknown, depth: number): boolean => {
nodes += 1;
if (nodes > maxNodes || depth > maxDepth) return false;
if (
candidate === null ||
typeof candidate === "string" ||
typeof candidate === "boolean" ||
(typeof candidate === "number" && Number.isFinite(candidate))
) {
return true;
}
if (Array.isArray(candidate)) {
return candidate.every((item) => visit(item, depth + 1));
}
if (
!candidate ||
typeof candidate !== "object" ||
Object.getPrototypeOf(candidate) !== Object.prototype
) {
return false;
}
return Object.entries(candidate).every(
([key, item]) =>
!FORBIDDEN_OBJECT_KEYS.has(key) && visit(item, depth + 1),
);
};
return visit(value, 0);
}
function snapshotJson(
value: unknown,
maxDepth: number,
maxNodes: number,
): unknown {
const seen = new WeakSet<object>();
let nodes = 0;
const visit = (candidate: unknown, depth: number): unknown => {
nodes += 1;
if (nodes > maxNodes || depth > maxDepth) {
throw new TypeError("Realtime payload exceeds its structural budget.");
}
if (
candidate === null ||
typeof candidate === "string" ||
typeof candidate === "boolean" ||
(typeof candidate === "number" && Number.isFinite(candidate))
) {
return candidate;
}
if (!candidate || typeof candidate !== "object") {
throw new TypeError("Realtime payload is not JSON-compatible.");
}
if (seen.has(candidate)) {
throw new TypeError("Realtime payload contains shared object identity.");
}
seen.add(candidate);
if (Array.isArray(candidate)) {
for (let index = 0; index < candidate.length; index += 1) {
if (!Object.hasOwn(candidate, index)) {
throw new TypeError("Realtime payload contains a sparse array.");
}
}
return Object.freeze(
candidate.map((item) => visit(item, depth + 1)),
);
}
if (
Object.getPrototypeOf(candidate) !== Object.prototype &&
Object.getPrototypeOf(candidate) !== null
) {
throw new TypeError("Realtime payload requires plain objects.");
}
const output: Record<string, unknown> = Object.create(null);
const descriptors = Object.getOwnPropertyDescriptors(candidate);
for (const key of Object.keys(descriptors).sort()) {
if (FORBIDDEN_OBJECT_KEYS.has(key)) {
throw new TypeError("Realtime payload contains a forbidden key.");
}
const descriptor = descriptors[key];
if (!descriptor || !("value" in descriptor)) {
throw new TypeError("Realtime payload contains an accessor.");
}
output[key] = visit(descriptor.value, depth + 1);
}
return Object.freeze(output);
};
return visit(value, 0);
}
function canonicalJson(value: unknown): string {
if (
value === null ||
typeof value === "string" ||
typeof value === "boolean" ||
typeof value === "number"
) {
return JSON.stringify(value);
}
if (Array.isArray(value)) {
return `[${value.map(canonicalJson).join(",")}]`;
}
if (!value || typeof value !== "object") {
throw new TypeError("Realtime semantic identity is invalid.");
}
return `{${Object.keys(value)
.sort()
.map(
(key) =>
`${JSON.stringify(key)}:${canonicalJson(
(value as Readonly<Record<string, unknown>>)[key],
)}`,
)
.join(",")}}`;
}
+226
View File
@@ -0,0 +1,226 @@
import type {
RealtimeAcceptDisposition,
RealtimeTransportEventOutcome,
} from "../../application/ports/realtime/event-authority.ts";
import {
REALTIME_TRANSPORT_CONTINUE,
realtimeTransportRecoveryCommitted,
} from "../../application/ports/realtime/event-authority.ts";
import type {
RealtimeResult,
} from "../../application/ports/realtime/shared.ts";
import {
isRealtimeResumeCursor,
} from "../../contracts/realtime-events.ts";
import type {
StreamRegistrationId,
} from "../../contracts/realtime-streams.ts";
import type {
RealtimeEventCodec,
} from "./event-codec.ts";
import type {
RealtimeStreamCoordinator,
} from "./stream-coordinator.ts";
import {
realtimeFailure,
realtimeSuccess,
} from "./result.ts";
export type RealtimeTransportCursor =
| Readonly<{
kind: "SSE_DIRECT_CURSOR";
eventId: string;
}>
| Readonly<{ kind: "SSE_NO_CURSOR" }>
| Readonly<{ kind: "ENCAPSULATED" }>;
export type RealtimeEventConsumer = Readonly<{
consume(
rawEnvelope: string,
cursor: RealtimeTransportCursor,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeAcceptDisposition>>;
consumeEncapsulated(
envelope: Readonly<Record<string, unknown>>,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeAcceptDisposition>>;
consumeForTransport(
rawEnvelope: string,
cursor: RealtimeTransportCursor,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeTransportEventOutcome>>;
consumeEncapsulatedForTransport(
envelope: Readonly<Record<string, unknown>>,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeTransportEventOutcome>>;
}>;
/**
* The single handoff from transport bytes to the common event authority.
* SSE's transport-level `id` is checked here against the validated envelope;
* WebSocket can carry the same envelope without inventing a second cursor.
*/
export function createRealtimeEventConsumer(
dependencies: Readonly<{
codec: RealtimeEventCodec;
coordinator: Pick<RealtimeStreamCoordinator, "accept">;
}>,
): RealtimeEventConsumer {
async function consumeWithStream(
rawEnvelope: string,
cursor: RealtimeTransportCursor,
signal?: AbortSignal,
): Promise<
Readonly<{
streamId: StreamRegistrationId | null;
result: RealtimeResult<RealtimeAcceptDisposition>;
}>
> {
if (signal?.aborted) {
return {
streamId: null,
result: realtimeFailure("ABORTED", "RECEIVE"),
};
}
const decoded = dependencies.codec.decode(rawEnvelope);
if (!decoded.ok) {
return { streamId: null, result: decoded };
}
const envelope = decoded.value.envelope;
if (
(cursor.kind === "SSE_DIRECT_CURSOR" &&
(!isRealtimeResumeCursor(cursor.eventId) ||
envelope.recoveryMode !== "CURSOR" ||
envelope.resumeCursor !== cursor.eventId)) ||
(cursor.kind === "SSE_NO_CURSOR" &&
(envelope.recoveryMode === "CURSOR" ||
envelope.resumeCursor !== null))
) {
return {
streamId: envelope.streamId,
result: realtimeFailure(
"PROTOCOL_MISMATCH",
"RECEIVE",
),
};
}
return {
streamId: envelope.streamId,
result: await dependencies.coordinator.accept(
decoded.value,
signal,
),
};
}
async function consume(
rawEnvelope: string,
cursor: RealtimeTransportCursor,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeAcceptDisposition>> {
return (
await consumeWithStream(rawEnvelope, cursor, signal)
).result;
}
async function consumeEncapsulated(
envelope: Readonly<Record<string, unknown>>,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeAcceptDisposition>> {
const serialized = serializeEnvelope(envelope);
if (!serialized.ok) return serialized;
return await consume(
serialized.value,
{ kind: "ENCAPSULATED" },
signal,
);
}
async function consumeForTransport(
rawEnvelope: string,
cursor: RealtimeTransportCursor,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeTransportEventOutcome>> {
const consumed = await consumeWithStream(
rawEnvelope,
cursor,
signal,
);
return projectTransportOutcome(
consumed.result,
consumed.streamId,
);
}
async function consumeEncapsulatedForTransport(
envelope: Readonly<Record<string, unknown>>,
signal?: AbortSignal,
): Promise<RealtimeResult<RealtimeTransportEventOutcome>> {
const serialized = serializeEnvelope(envelope);
if (!serialized.ok) return serialized;
return await consumeForTransport(
serialized.value,
{ kind: "ENCAPSULATED" },
signal,
);
}
return Object.freeze({
consume,
consumeEncapsulated,
consumeForTransport,
consumeEncapsulatedForTransport,
});
}
function projectTransportOutcome(
accepted: RealtimeResult<RealtimeAcceptDisposition>,
streamId: StreamRegistrationId | null,
): RealtimeResult<RealtimeTransportEventOutcome> {
if (!accepted.ok) return accepted;
if (
accepted.value.outcome === "RECOVERED" ||
accepted.value.outcome === "RECOVERY_BARRIER_REQUIRED"
) {
if (streamId === null) {
return realtimeFailure("PROTOCOL_MISMATCH", "RECEIVE");
}
return realtimeSuccess(
realtimeTransportRecoveryCommitted(
streamId,
accepted.value.resumeState,
),
);
}
if (accepted.value.outcome === "APPLIED") {
return realtimeSuccess(REALTIME_TRANSPORT_CONTINUE);
}
switch (accepted.value.reason) {
case "DUPLICATE_EVENT":
case "STALE_EVENT":
return realtimeSuccess(REALTIME_TRANSPORT_CONTINUE);
case "RECOVERY_IN_PROGRESS":
return realtimeFailure(
"PROTOCOL_MISMATCH",
"RECEIVE",
);
case "CLOSED":
return realtimeFailure("CLOSED", "RECEIVE");
case "SCOPE_FENCED":
return realtimeFailure("SCOPE_FENCED", "RECEIVE");
}
}
function serializeEnvelope(
envelope: Readonly<Record<string, unknown>>,
): RealtimeResult<string> {
let rawEnvelope: string;
try {
rawEnvelope = JSON.stringify(envelope);
} catch {
return realtimeFailure("MALFORMED_EVENT", "RECEIVE");
}
return typeof rawEnvelope === "string"
? realtimeSuccess(rawEnvelope)
: realtimeFailure("MALFORMED_EVENT", "RECEIVE");
}
+58
View File
@@ -0,0 +1,58 @@
export {
createRealtimeEventCodec,
isValidatedRealtimeEventDto,
type RealtimeEventCodec,
type RealtimeEventCodecDependencies,
type ValidatedRealtimeEventDto,
} from "./event-codec.ts";
export {
createRealtimeEventConsumer,
type RealtimeEventConsumer,
type RealtimeTransportCursor,
} from "./event-consumer.ts";
export {
calculateReconnectDelay,
defineReconnectPolicy,
isReconnectAttemptResetEligible,
parseRetryAfterDelay,
REALTIME_RECONNECT_CEILINGS,
reconnectBudgetRemaining,
type ReconnectDelayInput,
type ReconnectPolicy,
} from "./reconnect-policy.ts";
export {
createRealtimeReconnectCoordinator,
type RealtimeCommittedRecovery,
type RealtimeReconnectAttemptContext,
type RealtimeReconnectAttemptSuccess,
type RealtimeReconnectCloseClassification,
type RealtimeReconnectCoordinator,
type RealtimeReconnectCoordinatorDependencies,
type RealtimeReconnectEnvironment,
type RealtimeReconnectOutcome,
type RealtimeReconnectRunInput,
type RealtimeReconnectSession,
type RealtimeRecoveryReconnectDirective,
} from "./reconnect-coordinator.ts";
export {
createLivePollHandoffCoordinator,
LIVE_POLL_HANDOFF_CEILINGS,
type LivePollHandoffCoordinator,
type LivePollHandoffCoordinatorDependencies,
type LivePollHandoffInspection,
type LivePollHandoffLimits,
type LivePollHandoffRecoveryInput,
type LivePollHandoffState,
type LivePollWriterKind,
type LivePollWriterLease,
type LivePollWriteReceipt,
type LiveProbeLease,
} from "./live-poll-handoff-coordinator.ts";
export {
createRealtimeStreamCoordinator,
type RealtimeStreamCoordinator,
type RealtimeStreamCoordinatorDependencies,
} from "./stream-coordinator.ts";
export * from "./polling/index.ts";
export * from "./sse/index.ts";
export * from "./websocket/index.ts";
@@ -0,0 +1,218 @@
type Container =
| {
kind: "OBJECT";
state: "KEY_OR_END" | "COLON" | "VALUE" | "COMMA_OR_END";
keys: Set<string>;
}
| {
kind: "ARRAY";
state: "VALUE_OR_END" | "COMMA_OR_END";
};
/**
* Scans already byte-bounded JSON before `JSON.parse` can apply last-wins
* semantics. Invalid input and scanner budget exhaustion are both rejected.
*/
export function hasDuplicateJsonMembers(
source: string,
limits: Readonly<{
maxDepth: number;
maxMembers: number;
}>,
): boolean {
try {
return scan(source, limits);
} catch {
return true;
}
}
function scan(
source: string,
limits: Readonly<{
maxDepth: number;
maxMembers: number;
}>,
): boolean {
if (
typeof source !== "string" ||
!Number.isSafeInteger(limits.maxDepth) ||
limits.maxDepth < 1 ||
!Number.isSafeInteger(limits.maxMembers) ||
limits.maxMembers < 1
) {
return true;
}
const stack: Container[] = [];
let cursor = skipWhitespace(source, 0);
let rootStarted = false;
let rootComplete = false;
let members = 0;
const consumeValue = (): boolean => {
cursor = skipWhitespace(source, cursor);
const character = source[cursor];
if (character === "{") {
if (stack.length + 1 > limits.maxDepth) return false;
stack.push({
kind: "OBJECT",
state: "KEY_OR_END",
keys: new Set(),
});
cursor += 1;
return true;
}
if (character === "[") {
if (stack.length + 1 > limits.maxDepth) return false;
stack.push({ kind: "ARRAY", state: "VALUE_OR_END" });
cursor += 1;
return true;
}
if (character === "\"") {
const end = jsonStringEnd(source, cursor);
if (end === null) return false;
cursor = end;
return true;
}
const end = primitiveEnd(source, cursor);
if (end === cursor) return false;
cursor = end;
return true;
};
while (!rootComplete) {
if (!rootStarted) {
rootStarted = true;
if (!consumeValue()) return true;
if (stack.length === 0) rootComplete = true;
continue;
}
const container = stack.at(-1);
if (!container) {
rootComplete = true;
continue;
}
cursor = skipWhitespace(source, cursor);
if (container.kind === "ARRAY") {
if (container.state === "VALUE_OR_END") {
if (source[cursor] === "]") {
cursor += 1;
stack.pop();
if (stack.length === 0) rootComplete = true;
continue;
}
container.state = "COMMA_OR_END";
if (!consumeValue()) return true;
continue;
}
if (source[cursor] === ",") {
cursor += 1;
container.state = "VALUE_OR_END";
continue;
}
if (source[cursor] === "]") {
cursor += 1;
stack.pop();
if (stack.length === 0) rootComplete = true;
continue;
}
return true;
}
if (container.state === "KEY_OR_END") {
if (source[cursor] === "}") {
cursor += 1;
stack.pop();
if (stack.length === 0) rootComplete = true;
continue;
}
if (source[cursor] !== "\"") return true;
const end = jsonStringEnd(source, cursor);
if (end === null) return true;
const key = JSON.parse(source.slice(cursor, end)) as unknown;
if (typeof key !== "string" || container.keys.has(key)) {
return true;
}
members += 1;
if (members > limits.maxMembers) return true;
container.keys.add(key);
cursor = end;
container.state = "COLON";
continue;
}
if (container.state === "COLON") {
if (source[cursor] !== ":") return true;
cursor += 1;
container.state = "VALUE";
continue;
}
if (container.state === "VALUE") {
container.state = "COMMA_OR_END";
if (!consumeValue()) return true;
continue;
}
if (source[cursor] === ",") {
cursor += 1;
container.state = "KEY_OR_END";
continue;
}
if (source[cursor] === "}") {
cursor += 1;
stack.pop();
if (stack.length === 0) rootComplete = true;
continue;
}
return true;
}
return skipWhitespace(source, cursor) !== source.length;
}
function jsonStringEnd(source: string, start: number): number | null {
let escaped = false;
for (let cursor = start + 1; cursor < source.length; cursor += 1) {
const character = source[cursor];
if (escaped) {
escaped = false;
} else if (character === "\\") {
escaped = true;
} else if (character === "\"") {
return cursor + 1;
}
}
return null;
}
function primitiveEnd(source: string, start: number): number {
let cursor = start;
while (
cursor < source.length &&
source[cursor] !== "," &&
source[cursor] !== "]" &&
source[cursor] !== "}" &&
!isWhitespace(source[cursor])
) {
cursor += 1;
}
return cursor;
}
function skipWhitespace(source: string, start: number): number {
let cursor = start;
while (cursor < source.length && isWhitespace(source[cursor])) {
cursor += 1;
}
return cursor;
}
function isWhitespace(character: string | undefined): boolean {
return (
character === " " ||
character === "\n" ||
character === "\r" ||
character === "\t"
);
}
@@ -0,0 +1,883 @@
import type { ClockPort } from "../../application/ports/clock-port.ts";
import {
type RealtimeFailure,
type RealtimeOperation,
type RealtimeResult,
} from "../../application/ports/realtime/shared.ts";
import { systemClock } from "../platform/system-clock.ts";
import {
isRealtimeResult,
realtimeFailure,
realtimeSuccess,
} from "./result.ts";
export const LIVE_POLL_HANDOFF_CEILINGS = Object.freeze({
maxQuiescenceTimeoutMs: 30_000,
maxActiveQueueCount: 256,
maxActiveQueueBytes: 4 * 1024 * 1024,
maxProbeBufferedEvents: 256,
maxProbeBufferedBytes: 4 * 1024 * 1024,
maxItemBytes: 64 * 1024,
} as const);
export type LivePollHandoffState =
| "LIVE_ACTIVE"
| "POLL_ACTIVE"
| "LIVE_PROBING"
| "CLOSED";
export type LivePollWriterKind = "LIVE" | "POLL";
export type LivePollHandoffLimits = Readonly<{
quiescenceTimeoutMs: number;
maxActiveQueueCount: number;
maxActiveQueueBytes: number;
maxProbeBufferedEvents: number;
maxProbeBufferedBytes: number;
maxItemBytes: number;
}>;
export type LivePollWriteReceipt = Readonly<{
kind: "APPLIED" | "BUFFERED";
writer: LivePollWriterKind;
generation: number;
}>;
export type LivePollWriterLease<Value> = Readonly<{
writer: LivePollWriterKind;
generation: number;
signal: AbortSignal;
isCurrent(): boolean;
write(
value: Value,
wireBytes: number,
): Promise<RealtimeResult<LivePollWriteReceipt>>;
}>;
export type LiveProbeLease<Value> = LivePollWriterLease<Value> &
Readonly<{
writer: "LIVE";
activate(): Promise<RealtimeResult<LivePollWriterLease<Value>>>;
cancel(): RealtimeResult<LivePollWriterLease<Value>>;
}>;
export type LivePollHandoffInspection = Readonly<{
state: LivePollHandoffState;
activeWriter: LivePollWriterKind | null;
activeGeneration: number | null;
probeGeneration: number | null;
bufferedEvents: number;
bufferedBytes: number;
transitioning: boolean;
}>;
export type LivePollHandoffRecoveryInput = Readonly<{
from: LivePollWriterKind;
to: LivePollWriterKind;
candidateGeneration: number;
signal: AbortSignal;
/**
* Must be checked immediately before committing the checkpoint projection.
*/
isCurrent(): boolean;
}>;
export type LivePollHandoffCoordinator<Value> = Readonly<{
currentWriter(): LivePollWriterLease<Value> | null;
switchToPoll(): Promise<RealtimeResult<LivePollWriterLease<Value>>>;
beginLiveProbe(): RealtimeResult<LiveProbeLease<Value>>;
inspect(): LivePollHandoffInspection;
close(): Promise<RealtimeResult<void>>;
}>;
export type LivePollHandoffCoordinatorDependencies<Value> = Readonly<{
initial: Readonly<{
writer: LivePollWriterKind;
authoritativeCheckpointEstablished: true;
}>;
limits: LivePollHandoffLimits;
apply(input: Readonly<{
writer: LivePollWriterKind;
generation: number;
value: Value;
signal: AbortSignal;
/**
* Must be checked immediately before committing the external effect.
*/
isCurrent(): boolean;
}>): Promise<RealtimeResult<void>>;
establishAuthoritativeCheckpoint(
input: LivePollHandoffRecoveryInput,
): Promise<RealtimeResult<void>>;
clock?: ClockPort;
}>;
type BufferedValue<Value> = Readonly<{
value: Value;
wireBytes: number;
}>;
type InternalWriterLease<Value> = {
readonly writer: LivePollWriterKind;
readonly generation: number;
readonly controller: AbortController;
facade: LivePollWriterLease<Value>;
tail: Promise<void>;
queuedCount: number;
queuedBytes: number;
};
type InternalProbe<Value> = {
readonly lease: InternalWriterLease<Value>;
facade: LiveProbeLease<Value>;
readonly buffer: BufferedValue<Value>[];
bufferedBytes: number;
acceptedEvents: number;
acceptedBytes: number;
};
type QuiescenceOutcome = "QUIESCED" | "TIMER_FAILED" | "TIMED_OUT";
export function createLivePollHandoffCoordinator<Value>(
dependencies: LivePollHandoffCoordinatorDependencies<Value>,
): LivePollHandoffCoordinator<Value> {
if (
!dependencies ||
!dependencies.initial ||
(dependencies.initial.writer !== "LIVE" &&
dependencies.initial.writer !== "POLL") ||
dependencies.initial.authoritativeCheckpointEstablished !== true ||
typeof dependencies.apply !== "function" ||
typeof dependencies.establishAuthoritativeCheckpoint !== "function"
) {
throw new TypeError(
"Invalid live/poll handoff dependencies or initial checkpoint.",
);
}
const limits = validateLimits(dependencies.limits);
const clock = dependencies.clock ?? systemClock;
let state: LivePollHandoffState =
dependencies.initial.writer === "LIVE"
? "LIVE_ACTIVE"
: "POLL_ACTIVE";
let generationCounter = 0;
let lifecycleGeneration = 0;
let transitioning = false;
let active: InternalWriterLease<Value> | null = null;
let probe: InternalProbe<Value> | null = null;
let quiescing: InternalWriterLease<Value> | null = null;
let transitionCandidate: InternalWriterLease<Value> | null = null;
let closePromise: Promise<RealtimeResult<void>> | null = null;
active = createWriterLease(dependencies.initial.writer);
function createWriterLease(
writer: LivePollWriterKind,
): InternalWriterLease<Value> {
const controller = new AbortController();
const generation = ++generationCounter;
const lease: InternalWriterLease<Value> = {
writer,
generation,
controller,
tail: Promise.resolve(),
queuedCount: 0,
queuedBytes: 0,
facade: null as unknown as LivePollWriterLease<Value>,
};
lease.facade = Object.freeze({
writer,
generation,
signal: controller.signal,
isCurrent: () => isActiveLease(lease),
write: async (value: Value, wireBytes: number) =>
await writeFromLease(lease, value, wireBytes),
});
return lease;
}
function createProbe(): InternalProbe<Value> {
const lease = createWriterLease("LIVE");
const selected: InternalProbe<Value> = {
lease,
buffer: [],
bufferedBytes: 0,
acceptedEvents: 0,
acceptedBytes: 0,
facade: null as unknown as LiveProbeLease<Value>,
};
selected.facade = Object.freeze({
...lease.facade,
writer: "LIVE" as const,
activate: async () => await activateProbe(selected),
cancel: () => cancelProbe(selected),
});
return selected;
}
async function writeFromLease(
lease: InternalWriterLease<Value>,
value: Value,
wireBytes: number,
): Promise<RealtimeResult<LivePollWriteReceipt>> {
if (state === "CLOSED") return handoffFailure("CLOSED", "APPLY");
if (probe?.lease === lease && state === "LIVE_PROBING") {
if (!validWireBytes(wireBytes, limits.maxItemBytes)) {
return handoffFailure("EVENT_TOO_LARGE", "APPLY");
}
return bufferProbeValue(probe, value, wireBytes);
}
if (!isActiveLease(lease)) {
return handoffFailure("SCOPE_FENCED", "APPLY");
}
if (!validWireBytes(wireBytes, limits.maxItemBytes)) {
return handoffFailure("EVENT_TOO_LARGE", "APPLY");
}
return await enqueueEffect(lease, value, wireBytes);
}
function bufferProbeValue(
selected: InternalProbe<Value>,
value: Value,
wireBytes: number,
): RealtimeResult<LivePollWriteReceipt> {
if (
selected.acceptedEvents + 1 >
limits.maxProbeBufferedEvents ||
selected.acceptedBytes + wireBytes >
limits.maxProbeBufferedBytes
) {
if (transitioning) {
failClosed();
} else {
selected.lease.controller.abort();
selected.buffer.length = 0;
selected.bufferedBytes = 0;
probe = null;
state = "POLL_ACTIVE";
}
return handoffFailure("QUEUE_OVERFLOW", "APPLY");
}
selected.buffer.push(Object.freeze({ value, wireBytes }));
selected.bufferedBytes += wireBytes;
selected.acceptedEvents += 1;
selected.acceptedBytes += wireBytes;
return realtimeSuccess(
Object.freeze({
kind: "BUFFERED" as const,
writer: "LIVE" as const,
generation: selected.lease.generation,
}),
);
}
function enqueueEffect(
lease: InternalWriterLease<Value>,
value: Value,
wireBytes: number,
): Promise<RealtimeResult<LivePollWriteReceipt>> {
if (
lease.queuedCount + 1 > limits.maxActiveQueueCount ||
lease.queuedBytes + wireBytes > limits.maxActiveQueueBytes
) {
failClosed();
return Promise.resolve(
handoffFailure("QUEUE_OVERFLOW", "APPLY"),
);
}
lease.queuedCount += 1;
lease.queuedBytes += wireBytes;
const result = lease.tail.then(async () => {
try {
if (!isEffectAuthorized(lease)) {
return handoffFailure("SCOPE_FENCED", "APPLY");
}
return await invokeApply(lease, value);
} finally {
lease.queuedCount -= 1;
lease.queuedBytes -= wireBytes;
}
});
lease.tail = result.then(
() => undefined,
() => undefined,
);
return result;
}
async function invokeApply(
lease: InternalWriterLease<Value>,
value: Value,
): Promise<RealtimeResult<LivePollWriteReceipt>> {
try {
const result = await dependencies.apply(
Object.freeze({
writer: lease.writer,
generation: lease.generation,
value,
signal: lease.controller.signal,
isCurrent: () => isEffectAuthorized(lease),
}),
);
if (!isRealtimeResult(result, isUndefined)) {
return handoffFailure(
"MAPPING_CONTRACT_VIOLATION",
"APPLY",
);
}
if (!result.ok) {
return Object.freeze({
ok: false,
error: result.error,
});
}
if (!isEffectAuthorized(lease)) {
return handoffFailure("SCOPE_FENCED", "APPLY");
}
return realtimeSuccess(
Object.freeze({
kind: "APPLIED" as const,
writer: lease.writer,
generation: lease.generation,
}),
);
} catch {
return handoffFailure("PROVIDER_UNAVAILABLE", "APPLY", true);
}
}
async function switchToPoll(): Promise<
RealtimeResult<LivePollWriterLease<Value>>
> {
if (state === "CLOSED") {
return handoffFailure("CLOSED", "RECOVER");
}
if (
state !== "LIVE_ACTIVE" ||
transitioning ||
active?.writer !== "LIVE"
) {
return handoffFailure("PROTOCOL_MISMATCH", "RECOVER");
}
transitioning = true;
const transitionGeneration = ++lifecycleGeneration;
const previous = active;
const candidate = createWriterLease("POLL");
transitionCandidate = candidate;
active = null;
quiescing = previous;
previous.controller.abort();
const quiescence = await awaitQuiescence(previous);
if (!transitionIsCurrent(transitionGeneration, candidate)) {
candidate.controller.abort();
return handoffFailure("CLOSED", "RECOVER");
}
if (quiescence !== "QUIESCED") {
failClosed();
return handoffFailure(
quiescence === "TIMED_OUT"
? "IDLE_TIMEOUT"
: "PROVIDER_UNAVAILABLE",
"RECOVER",
);
}
quiescing = null;
const checkpoint = await establishCheckpoint(
previous.writer,
candidate,
);
if (
!checkpoint.ok ||
!transitionIsCurrent(transitionGeneration, candidate)
) {
failClosed();
return checkpoint.ok
? handoffFailure("CLOSED", "RECOVER")
: checkpoint;
}
active = candidate;
transitionCandidate = null;
state = "POLL_ACTIVE";
transitioning = false;
return realtimeSuccess(candidate.facade);
}
function beginLiveProbe(): RealtimeResult<LiveProbeLease<Value>> {
if (state === "CLOSED") {
return handoffFailure("CLOSED", "SUBSCRIBE");
}
if (
state !== "POLL_ACTIVE" ||
transitioning ||
probe !== null ||
active?.writer !== "POLL"
) {
return handoffFailure("PROTOCOL_MISMATCH", "SUBSCRIBE");
}
const candidate = createProbe();
probe = candidate;
state = "LIVE_PROBING";
return realtimeSuccess(candidate.facade);
}
function cancelProbe(
selected: InternalProbe<Value>,
): RealtimeResult<LivePollWriterLease<Value>> {
if (state === "CLOSED") {
return handoffFailure("CLOSED", "CLOSE");
}
if (
state !== "LIVE_PROBING" ||
transitioning ||
probe !== selected ||
active?.writer !== "POLL"
) {
return handoffFailure("SCOPE_FENCED", "CLOSE");
}
selected.lease.controller.abort();
selected.buffer.length = 0;
selected.bufferedBytes = 0;
probe = null;
state = "POLL_ACTIVE";
return realtimeSuccess(active.facade);
}
async function activateProbe(
selected: InternalProbe<Value>,
): Promise<RealtimeResult<LivePollWriterLease<Value>>> {
if (state === "CLOSED") {
return handoffFailure("CLOSED", "RECOVER");
}
if (
state !== "LIVE_PROBING" ||
transitioning ||
probe !== selected ||
active?.writer !== "POLL"
) {
return handoffFailure("SCOPE_FENCED", "RECOVER");
}
transitioning = true;
const transitionGeneration = ++lifecycleGeneration;
const previous = active;
transitionCandidate = selected.lease;
active = null;
quiescing = previous;
previous.controller.abort();
const quiescence = await awaitQuiescence(previous);
if (!probeTransitionIsCurrent(transitionGeneration, selected)) {
selected.lease.controller.abort();
return handoffFailure("CLOSED", "RECOVER");
}
if (quiescence !== "QUIESCED") {
failClosed();
return handoffFailure(
quiescence === "TIMED_OUT"
? "IDLE_TIMEOUT"
: "PROVIDER_UNAVAILABLE",
"RECOVER",
);
}
quiescing = null;
const checkpoint = await establishCheckpoint(
previous.writer,
selected.lease,
);
if (
!checkpoint.ok ||
!probeTransitionIsCurrent(transitionGeneration, selected)
) {
failClosed();
return checkpoint.ok
? handoffFailure("CLOSED", "RECOVER")
: checkpoint;
}
while (selected.buffer.length > 0) {
if (!probeTransitionIsCurrent(transitionGeneration, selected)) {
return handoffFailure("CLOSED", "RECOVER");
}
const buffered = selected.buffer.shift();
if (!buffered) break;
selected.bufferedBytes -= buffered.wireBytes;
const applied = await enqueueEffect(
selected.lease,
buffered.value,
buffered.wireBytes,
);
if (!applied.ok) {
failClosed();
return Object.freeze({
ok: false,
error: remapFailure(applied.error, "RECOVER"),
});
}
}
if (!probeTransitionIsCurrent(transitionGeneration, selected)) {
return handoffFailure("CLOSED", "RECOVER");
}
active = selected.lease;
transitionCandidate = null;
probe = null;
state = "LIVE_ACTIVE";
transitioning = false;
return realtimeSuccess(selected.lease.facade);
}
async function establishCheckpoint(
from: LivePollWriterKind,
candidate: InternalWriterLease<Value>,
): Promise<RealtimeResult<void>> {
const timer = new AbortController();
let releaseAbortGate = (): void => undefined;
const aborted = new Promise<
Readonly<{ kind: "ABORTED" }>
>((resolve) => {
const onAbort = () => resolve({ kind: "ABORTED" });
candidate.controller.signal.addEventListener(
"abort",
onAbort,
{ once: true },
);
releaseAbortGate = () =>
candidate.controller.signal.removeEventListener(
"abort",
onAbort,
);
if (candidate.controller.signal.aborted) onAbort();
});
const operation = Promise.resolve()
.then(() =>
dependencies.establishAuthoritativeCheckpoint(
Object.freeze({
from,
to: candidate.writer,
candidateGeneration: candidate.generation,
signal: candidate.controller.signal,
isCurrent: () =>
isCheckpointCandidateCurrent(candidate),
}),
),
)
.then(
(value) => ({ kind: "VALUE" as const, value }),
() => ({ kind: "REJECTED" as const }),
);
const timeout = Promise.resolve()
.then(async () => {
await clock.sleep(
limits.quiescenceTimeoutMs,
timer.signal,
);
return { kind: "TIMED_OUT" as const };
})
.catch(() => ({
kind: timer.signal.aborted
? ("CANCELED" as const)
: ("TIMER_FAILED" as const),
}));
const selected = await Promise.race([
operation,
timeout,
aborted,
]);
timer.abort();
releaseAbortGate();
if (selected.kind === "ABORTED") {
return handoffFailure("ABORTED", "RECOVER");
}
if (selected.kind === "TIMED_OUT") {
candidate.controller.abort();
return handoffFailure("IDLE_TIMEOUT", "RECOVER");
}
if (
selected.kind === "TIMER_FAILED" ||
selected.kind === "REJECTED"
) {
candidate.controller.abort();
return handoffFailure(
"PROVIDER_UNAVAILABLE",
"RECOVER",
true,
);
}
if (selected.kind === "CANCELED") {
return handoffFailure("ABORTED", "RECOVER");
}
if (selected.kind !== "VALUE") {
candidate.controller.abort();
return handoffFailure(
"PROVIDER_UNAVAILABLE",
"RECOVER",
true,
);
}
const result = selected.value;
if (!isRealtimeResult(result, isUndefined)) {
candidate.controller.abort();
return handoffFailure(
"MAPPING_CONTRACT_VIOLATION",
"RECOVER",
);
}
if (!result.ok) {
return Object.freeze({
ok: false,
error: remapFailure(result.error, "RECOVER"),
});
}
if (candidate.controller.signal.aborted) {
return handoffFailure("ABORTED", "RECOVER");
}
return realtimeSuccess(undefined);
}
async function awaitQuiescence(
lease: InternalWriterLease<Value>,
): Promise<QuiescenceOutcome> {
const timer = new AbortController();
const settled = lease.tail.then(
() => "QUIESCED" as const,
() => "QUIESCED" as const,
);
const timeout = Promise.resolve()
.then(async () => {
await clock.sleep(
limits.quiescenceTimeoutMs,
timer.signal,
);
return "TIMED_OUT" as const;
})
.catch(() =>
timer.signal.aborted
? ("QUIESCED" as const)
: ("TIMER_FAILED" as const),
);
const outcome = await Promise.race([settled, timeout]);
timer.abort();
return outcome;
}
function close(): Promise<RealtimeResult<void>> {
closePromise ??= performClose();
return closePromise;
}
async function performClose(): Promise<RealtimeResult<void>> {
lifecycleGeneration += 1;
state = "CLOSED";
transitioning = true;
const writers = uniqueLeases([
active,
probe?.lease ?? null,
quiescing,
transitionCandidate,
]);
active = null;
const selectedProbe = probe;
probe = null;
selectedProbe?.buffer.splice(0);
if (selectedProbe) selectedProbe.bufferedBytes = 0;
for (const writer of writers) writer.controller.abort();
const outcomes = await Promise.all(
writers.map(async (writer) => await awaitQuiescence(writer)),
);
quiescing = null;
transitionCandidate = null;
transitioning = false;
if (outcomes.includes("TIMED_OUT")) {
return handoffFailure("IDLE_TIMEOUT", "CLOSE");
}
if (outcomes.includes("TIMER_FAILED")) {
return handoffFailure("PROVIDER_UNAVAILABLE", "CLOSE");
}
return realtimeSuccess(undefined);
}
function inspect(): LivePollHandoffInspection {
return Object.freeze({
state,
activeWriter: active?.writer ?? null,
activeGeneration: active?.generation ?? null,
probeGeneration: probe?.lease.generation ?? null,
bufferedEvents: probe?.buffer.length ?? 0,
bufferedBytes: probe?.bufferedBytes ?? 0,
transitioning,
});
}
function isActiveLease(lease: InternalWriterLease<Value>): boolean {
if (active !== lease || transitioning || state === "CLOSED") {
return false;
}
return (
(lease.writer === "LIVE" && state === "LIVE_ACTIVE") ||
(lease.writer === "POLL" &&
(state === "POLL_ACTIVE" || state === "LIVE_PROBING"))
);
}
function isEffectAuthorized(
lease: InternalWriterLease<Value>,
): boolean {
return (
isActiveLease(lease) ||
(transitioning &&
state === "LIVE_PROBING" &&
probe?.lease === lease &&
!lease.controller.signal.aborted)
);
}
function transitionIsCurrent(
transitionGeneration: number,
candidate: InternalWriterLease<Value>,
): boolean {
return (
state !== "CLOSED" &&
transitioning &&
lifecycleGeneration === transitionGeneration &&
!candidate.controller.signal.aborted
);
}
function probeTransitionIsCurrent(
transitionGeneration: number,
selected: InternalProbe<Value>,
): boolean {
return (
state === "LIVE_PROBING" &&
transitioning &&
lifecycleGeneration === transitionGeneration &&
probe === selected &&
!selected.lease.controller.signal.aborted
);
}
function isCheckpointCandidateCurrent(
candidate: InternalWriterLease<Value>,
): boolean {
return (
state !== "CLOSED" &&
transitioning &&
transitionCandidate === candidate &&
!candidate.controller.signal.aborted
);
}
function failClosed(): void {
lifecycleGeneration += 1;
state = "CLOSED";
transitioning = false;
active?.controller.abort();
probe?.lease.controller.abort();
quiescing?.controller.abort();
transitionCandidate?.controller.abort();
active = null;
if (probe) {
probe.buffer.length = 0;
probe.bufferedBytes = 0;
}
probe = null;
}
return Object.freeze({
currentWriter: () => active?.facade ?? null,
switchToPoll,
beginLiveProbe,
inspect,
close,
});
}
function validateLimits(
limits: LivePollHandoffLimits,
): LivePollHandoffLimits {
if (
!positiveIntegerWithin(
limits.quiescenceTimeoutMs,
LIVE_POLL_HANDOFF_CEILINGS.maxQuiescenceTimeoutMs,
) ||
!positiveIntegerWithin(
limits.maxActiveQueueCount,
LIVE_POLL_HANDOFF_CEILINGS.maxActiveQueueCount,
) ||
!positiveIntegerWithin(
limits.maxActiveQueueBytes,
LIVE_POLL_HANDOFF_CEILINGS.maxActiveQueueBytes,
) ||
!positiveIntegerWithin(
limits.maxProbeBufferedEvents,
LIVE_POLL_HANDOFF_CEILINGS.maxProbeBufferedEvents,
) ||
!positiveIntegerWithin(
limits.maxProbeBufferedBytes,
LIVE_POLL_HANDOFF_CEILINGS.maxProbeBufferedBytes,
) ||
!positiveIntegerWithin(
limits.maxItemBytes,
LIVE_POLL_HANDOFF_CEILINGS.maxItemBytes,
) ||
limits.maxItemBytes > limits.maxProbeBufferedBytes ||
limits.maxItemBytes > limits.maxActiveQueueBytes
) {
throw new TypeError("Invalid live/poll handoff limits.");
}
return Object.freeze({ ...limits });
}
function positiveIntegerWithin(value: number, maximum: number): boolean {
return (
Number.isSafeInteger(value) &&
value > 0 &&
value <= maximum
);
}
function validWireBytes(value: number, maximum: number): boolean {
return positiveIntegerWithin(value, maximum);
}
function isUndefined(value: unknown): value is undefined {
return value === undefined;
}
function handoffFailure(
kind: Parameters<typeof realtimeFailure>[0],
operation: RealtimeOperation,
retryable?: boolean,
): Extract<RealtimeResult<never>, { ok: false }> {
return retryable === undefined
? realtimeFailure(kind, operation)
: realtimeFailure(kind, operation, retryable);
}
function remapFailure(
failure: RealtimeFailure,
operation: RealtimeOperation,
): RealtimeFailure {
return Object.freeze({
kind: failure.kind,
operation,
retryable: failure.retryable,
});
}
function uniqueLeases<Value>(
values: readonly (InternalWriterLease<Value> | null)[],
): InternalWriterLease<Value>[] {
return [
...new Set(
values.filter(
(value): value is InternalWriterLease<Value> =>
value !== null,
),
),
];
}
@@ -0,0 +1,961 @@
import {
assertBoundedPollOperation,
definePollLeasePolicy,
type BoundedPollOperationContract,
type PollLeasePolicy,
} from "../../../application/policies/bounded-polling.ts";
import type { ClockPort } from "../../../application/ports/clock-port.ts";
import {
REALTIME_FAILURE_KINDS,
type RealtimeFailure,
type RealtimeFailureKind,
type RealtimeResult,
} from "../../../application/ports/realtime/shared.ts";
import { systemClock } from "../../platform/system-clock.ts";
import {
realtimeFailure,
realtimeSuccess,
} from "../result.ts";
/**
* Provider-private Retry-After metadata is consumed by this adapter and is
* deliberately stripped before a failure crosses the realtime boundary.
*/
export type BoundedPollAttemptFailure = RealtimeFailure & Readonly<{
retryAfterMs?: number;
}>;
type BoundedPollAttemptSuccess<Value> =
| Readonly<{
kind: "UNCHANGED";
responseBytes: 0;
}>
| Readonly<{
kind: "VALUE";
value: Value;
responseBytes: number;
state?: string;
}>;
export type BoundedPollAttemptResult<Value> =
| Extract<
RealtimeResult<BoundedPollAttemptSuccess<Value>>,
{ ok: true }
>
| Readonly<{ ok: false; error: BoundedPollAttemptFailure }>;
export type BoundedPollResult<Value> =
RealtimeResult<
Readonly<{
kind: "TERMINAL";
attempts: number;
state: string;
value: Value;
}>
>;
export type BoundedPollEnvironment = Readonly<{
visibility(): "HIDDEN" | "VISIBLE";
online(): boolean;
subscribeVisibility?(
listener: (visibility: "HIDDEN" | "VISIBLE") => void,
): () => void;
subscribeOnline?(listener: (online: boolean) => void): () => void;
}>;
export type BoundedPollRunInput<Value> = Readonly<{
signal?: AbortSignal;
onValue?: (
value: Value,
context: Readonly<{ signal: AbortSignal; isCurrent(): boolean }>,
) => void | Promise<void>;
}>;
export type BoundedPollCoordinator<Value> = Readonly<{
run(input?: BoundedPollRunInput<Value>): Promise<
BoundedPollResult<Value>
>;
getState(): "CLOSED" | "DRAINING" | "IDLE" | "RUNNING";
close(): void;
}>;
export type BoundedPollCoordinatorDependencies<Value> = Readonly<{
policy: PollLeasePolicy;
operation: BoundedPollOperationContract;
execute(input: Readonly<{
operationId: string;
attempt: number;
/**
* Hard response-body ceiling that must be enforced before decoding.
*/
maxResponseBytes: number;
signal: AbortSignal;
}>): Promise<BoundedPollAttemptResult<Value>>;
environment: BoundedPollEnvironment;
isCurrent?: () => boolean;
clock?: ClockPort;
random?: () => number;
}>;
const SAFE_STATE = /^[A-Z][A-Z0-9_]{0,63}$/u;
const POLL_RETRYABLE_FAILURE_KINDS = Object.freeze([
"CONNECT_TIMEOUT",
"RATE_LIMITED",
"PROVIDER_UNAVAILABLE",
] as const satisfies readonly RealtimeFailureKind[]);
const POLL_RETRY_AFTER_FAILURE_KINDS = Object.freeze([
"RATE_LIMITED",
"PROVIDER_UNAVAILABLE",
] as const satisfies readonly RealtimeFailureKind[]);
export function createBoundedPollCoordinator<Value>(
dependencies: BoundedPollCoordinatorDependencies<Value>,
): BoundedPollCoordinator<Value> {
const policy = definePollLeasePolicy(dependencies.policy);
assertBoundedPollOperation(policy, dependencies.operation);
const maxResponseBytes = Math.min(
policy.maxResponseBytes,
dependencies.operation.maxResponseBytes,
);
const clock = dependencies.clock ?? systemClock;
const random = dependencies.random ?? Math.random;
const isCurrent = dependencies.isCurrent ?? (() => true);
let state: "CLOSED" | "DRAINING" | "IDLE" | "RUNNING" = "IDLE";
let activeController: AbortController | null = null;
let generation = 0;
let pendingWork = 0;
async function run(
input: BoundedPollRunInput<Value> = {},
): Promise<BoundedPollResult<Value>> {
if (state === "CLOSED") return pollFailure("CLOSED");
if (state !== "IDLE") {
return pollFailure("PROTOCOL_MISMATCH");
}
if (input.signal?.aborted) return pollFailure("ABORTED");
if (safeVisibility(dependencies.environment) !== "VISIBLE") {
return pollFailure("ABORTED");
}
if (safeOnline(dependencies.environment) !== true) {
return pollFailure("OFFLINE", true);
}
if (!safeIsCurrent(isCurrent)) {
return pollFailure("SCOPE_FENCED");
}
state = "RUNNING";
const runGeneration = ++generation;
const controller = new AbortController();
activeController = controller;
let stopKind: RealtimeFailureKind | null = null;
let attempts = 0;
let consecutiveFailures = 0;
let nextDelayMs = policy.minimumIntervalMs;
const startedAtMs = safeNow(clock);
const stop = (kind: RealtimeFailureKind) => {
if (stopKind !== null) return;
stopKind = kind;
controller.abort();
};
const onCallerAbort = () => stop("ABORTED");
input.signal?.addEventListener("abort", onCallerAbort, {
once: true,
});
let unsubscribeVisibility: (() => void) | undefined;
let unsubscribeOnline: (() => void) | undefined;
try {
unsubscribeVisibility =
dependencies.environment.subscribeVisibility?.((visibility) => {
if (visibility !== "VISIBLE") stop("ABORTED");
});
} catch {
stop("ABORTED");
}
try {
unsubscribeOnline =
dependencies.environment.subscribeOnline?.((online) => {
if (!online) stop("OFFLINE");
});
} catch {
stop("OFFLINE");
}
try {
if (startedAtMs === null) {
return pollFailure("PROVIDER_UNAVAILABLE");
}
while (true) {
const lifecycleFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
if (lifecycleFailure) {
return pollFailure(
lifecycleFailure,
lifecycleFailure === "OFFLINE",
);
}
if (attempts >= policy.maxAttempts) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
const nowBeforeSleep = safeNow(clock);
if (
nowBeforeSleep === null ||
nowBeforeSleep < startedAtMs
) {
return pollFailure("PROVIDER_UNAVAILABLE");
}
if (
nowBeforeSleep - startedAtMs + nextDelayMs >=
policy.maxElapsedMs
) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
const cadenceOutcome = await awaitTaskOrAbort(
() => clock.sleep(nextDelayMs, controller.signal),
controller.signal,
);
if (cadenceOutcome.kind !== "VALUE") {
const afterSleepFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
return pollFailure(
afterSleepFailure ??
(cadenceOutcome.kind === "THREW"
? "PROVIDER_UNAVAILABLE"
: "ABORTED"),
afterSleepFailure === "OFFLINE",
);
}
const beforeAttemptFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
if (beforeAttemptFailure) {
return pollFailure(
beforeAttemptFailure,
beforeAttemptFailure === "OFFLINE",
);
}
const attemptStartedAt = safeNow(clock);
if (
attemptStartedAt === null ||
attemptStartedAt < startedAtMs ||
attemptStartedAt - startedAtMs >= policy.maxElapsedMs
) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
attempts += 1;
let result: BoundedPollAttemptResult<Value>;
const attemptOutcome = await awaitWithinLease(
() =>
trackWork(
dependencies.execute({
operationId: policy.operationId,
attempt: attempts,
maxResponseBytes,
signal: controller.signal,
}),
runGeneration,
),
policy.maxElapsedMs - (attemptStartedAt - startedAtMs),
clock,
controller.signal,
() => stop("POLL_BUDGET_EXHAUSTED"),
);
if (attemptOutcome.kind === "VALUE") {
result = attemptOutcome.value;
} else if (attemptOutcome.kind === "THREW") {
result = realtimeFailure(
controller.signal.aborted ? "ABORTED" : "OFFLINE",
"POLL",
!controller.signal.aborted,
);
} else {
if (attemptOutcome.kind === "CLOCK_FAILED") {
controller.abort();
}
const interruptedFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
return pollFailure(
interruptedFailure ??
(attemptOutcome.kind === "CLOCK_FAILED"
? "PROVIDER_UNAVAILABLE"
: "ABORTED"),
interruptedFailure === "OFFLINE",
);
}
const afterAttemptFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
if (afterAttemptFailure) {
return pollFailure(
afterAttemptFailure,
afterAttemptFailure === "OFFLINE",
);
}
const attemptFinishedAt = safeNow(clock);
if (
attemptFinishedAt === null ||
attemptFinishedAt < attemptStartedAt
) {
return pollFailure("PROVIDER_UNAVAILABLE");
}
if (
attemptFinishedAt - startedAtMs >= policy.maxElapsedMs
) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
const parsedResult = parseAttemptResult<Value>(result);
if (!parsedResult) {
return pollFailure("PROTOCOL_MISMATCH");
}
result = parsedResult;
if (!result.ok) {
if (
!result.error.retryable ||
!isPollRetryableFailureKind(result.error.kind) ||
(isPollRetryAfterFailureKind(result.error.kind) &&
result.error.retryAfterMs === undefined)
) {
return pollFailure(result.error.kind);
}
consecutiveFailures += 1;
const failureDelay = retryDelay(
policy,
consecutiveFailures,
isPollRetryAfterFailureKind(result.error.kind)
? result.error.retryAfterMs
: undefined,
random,
);
if (failureDelay === null) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
nextDelayMs = failureDelay;
continue;
}
consecutiveFailures = 0;
if (
result.value.responseBytes > maxResponseBytes
) {
return pollFailure("PROTOCOL_MISMATCH");
}
if (result.value.kind === "UNCHANGED") {
const delay = successDelay(policy, random);
if (delay === null) {
return pollFailure("PROTOCOL_MISMATCH");
}
nextDelayMs = delay;
continue;
}
const valueResult = result.value;
if (input.onValue) {
const beforeApplyAt = safeNow(clock);
if (
beforeApplyAt === null ||
beforeApplyAt < attemptFinishedAt ||
beforeApplyAt - startedAtMs >= policy.maxElapsedMs
) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
const applyOutcome = await awaitWithinLease(
() =>
trackWork(
Promise.resolve(
input.onValue!(valueResult.value, {
signal: controller.signal,
isCurrent: () =>
state === "RUNNING" &&
generation === runGeneration &&
stopKind === null &&
!controller.signal.aborted &&
safeIsCurrent(isCurrent),
}),
),
runGeneration,
),
policy.maxElapsedMs - (beforeApplyAt - startedAtMs),
clock,
controller.signal,
() => stop("POLL_BUDGET_EXHAUSTED"),
);
if (applyOutcome.kind === "THREW") {
return applyFailure();
}
if (applyOutcome.kind !== "VALUE") {
if (applyOutcome.kind === "CLOCK_FAILED") {
controller.abort();
}
const interruptedFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
return pollFailure(
interruptedFailure ??
(applyOutcome.kind === "CLOCK_FAILED"
? "PROVIDER_UNAVAILABLE"
: "ABORTED"),
interruptedFailure === "OFFLINE",
);
}
}
const afterApplyFailure = currentFailure(
stopKind,
state,
runGeneration,
generation,
isCurrent,
dependencies.environment,
);
if (afterApplyFailure) {
return pollFailure(
afterApplyFailure,
afterApplyFailure === "OFFLINE",
);
}
const applyFinishedAt = safeNow(clock);
if (
applyFinishedAt === null ||
applyFinishedAt < attemptFinishedAt
) {
return pollFailure("PROVIDER_UNAVAILABLE");
}
if (applyFinishedAt - startedAtMs >= policy.maxElapsedMs) {
return pollFailure("POLL_BUDGET_EXHAUSTED");
}
if (
valueResult.state &&
policy.terminalStates.includes(valueResult.state)
) {
return realtimeSuccess(
Object.freeze({
kind: "TERMINAL" as const,
attempts,
state: valueResult.state,
value: valueResult.value,
}),
);
}
const delay = successDelay(policy, random);
if (delay === null) {
return pollFailure("PROTOCOL_MISMATCH");
}
nextDelayMs = delay;
}
} finally {
input.signal?.removeEventListener("abort", onCallerAbort);
safelyUnsubscribe(unsubscribeVisibility);
safelyUnsubscribe(unsubscribeOnline);
if (activeController === controller) {
activeController = null;
}
if (generation === runGeneration) {
state = pendingWork === 0 ? "IDLE" : "DRAINING";
}
}
}
function trackWork<WorkValue>(
work: Promise<WorkValue>,
workGeneration: number,
): Promise<WorkValue> {
pendingWork += 1;
void work.then(
() => releaseWork(workGeneration),
() => releaseWork(workGeneration),
);
return work;
}
function releaseWork(workGeneration: number): void {
pendingWork = Math.max(0, pendingWork - 1);
if (
pendingWork === 0 &&
state === "DRAINING" &&
generation === workGeneration
) {
state = "IDLE";
}
}
function close(): void {
if (state === "CLOSED") return;
state = "CLOSED";
generation += 1;
activeController?.abort();
}
return Object.freeze({
run,
getState: () => state,
close,
});
}
function currentFailure(
requested: RealtimeFailureKind | null,
state: "CLOSED" | "DRAINING" | "IDLE" | "RUNNING",
runGeneration: number,
currentGeneration: number,
isCurrent: () => boolean,
environment: BoundedPollEnvironment,
): RealtimeFailureKind | null {
if (requested) return requested;
if (
state === "CLOSED" ||
state === "DRAINING" ||
runGeneration !== currentGeneration
) {
return "CLOSED";
}
if (!safeIsCurrent(isCurrent)) return "SCOPE_FENCED";
if (safeVisibility(environment) !== "VISIBLE") return "ABORTED";
if (safeOnline(environment) !== true) return "OFFLINE";
return null;
}
type TaskOutcome<Value> =
| Readonly<{ kind: "VALUE"; value: Value }>
| Readonly<{ kind: "THREW" }>
| Readonly<{ kind: "ABORTED" }>;
type LeaseTaskOutcome<Value> =
| TaskOutcome<Value>
| Readonly<{ kind: "LEASE_EXPIRED" }>
| Readonly<{ kind: "CLOCK_FAILED" }>;
async function awaitTaskOrAbort<Value>(
task: () => Promise<Value>,
signal: AbortSignal,
): Promise<TaskOutcome<Value>> {
if (signal.aborted) return Object.freeze({ kind: "ABORTED" });
let removeAbortListener: () => void = () => undefined;
const aborted = new Promise<Readonly<{ kind: "ABORTED" }>>(
(resolve) => {
const onAbort = () => resolve(Object.freeze({ kind: "ABORTED" }));
signal.addEventListener("abort", onAbort, { once: true });
removeAbortListener = () =>
signal.removeEventListener("abort", onAbort);
if (signal.aborted) onAbort();
},
);
if (signal.aborted) {
removeAbortListener();
return Object.freeze({ kind: "ABORTED" });
}
let taskPromise: Promise<Value>;
try {
taskPromise = task();
} catch {
removeAbortListener();
return Object.freeze({ kind: "THREW" });
}
const completed = taskPromise.then<
TaskOutcome<Value>,
TaskOutcome<Value>
>(
(value) => Object.freeze({ kind: "VALUE", value }),
() => Object.freeze({ kind: "THREW" }),
);
try {
return await Promise.race([completed, aborted]);
} finally {
removeAbortListener();
}
}
async function awaitWithinLease<Value>(
task: () => Promise<Value>,
remainingMs: number,
clock: ClockPort,
signal: AbortSignal,
onLeaseExpired: () => void,
): Promise<LeaseTaskOutcome<Value>> {
if (!Number.isFinite(remainingMs) || remainingMs <= 0) {
onLeaseExpired();
return Object.freeze({ kind: "LEASE_EXPIRED" });
}
if (signal.aborted) return Object.freeze({ kind: "ABORTED" });
const deadlineController = new AbortController();
let resolveInterruption:
| ((outcome: LeaseTaskOutcome<Value>) => void)
| undefined;
let removeAbortListener: () => void = () => undefined;
let interruptionSettled = false;
const finishInterruption = (
outcome: LeaseTaskOutcome<Value>,
): boolean => {
if (interruptionSettled) return false;
interruptionSettled = true;
resolveInterruption?.(outcome);
return true;
};
const interrupted = new Promise<LeaseTaskOutcome<Value>>((resolve) => {
resolveInterruption = resolve;
const onAbort = () =>
finishInterruption(Object.freeze({ kind: "ABORTED" }));
signal.addEventListener("abort", onAbort, { once: true });
removeAbortListener = () =>
signal.removeEventListener("abort", onAbort);
if (signal.aborted) onAbort();
});
if (signal.aborted) {
removeAbortListener();
return Object.freeze({ kind: "ABORTED" });
}
let deadlineSleep: Promise<void>;
try {
deadlineSleep = clock.sleep(
remainingMs,
deadlineController.signal,
);
} catch {
removeAbortListener();
deadlineController.abort();
return Object.freeze({ kind: "CLOCK_FAILED" });
}
const deadline = deadlineSleep.then(
() => {
if (deadlineController.signal.aborted) return;
if (
finishInterruption(
Object.freeze({ kind: "LEASE_EXPIRED" }),
)
) {
onLeaseExpired();
}
},
() => {
if (!deadlineController.signal.aborted) {
finishInterruption(
Object.freeze({ kind: "CLOCK_FAILED" }),
);
}
},
);
if (signal.aborted) {
removeAbortListener();
deadlineController.abort();
void deadline;
return Object.freeze({ kind: "ABORTED" });
}
let taskPromise: Promise<Value>;
try {
taskPromise = task();
} catch {
taskPromise = Promise.reject(new Error("Task failed."));
}
const completed = taskPromise.then<
LeaseTaskOutcome<Value>,
LeaseTaskOutcome<Value>
>(
(value) => Object.freeze({ kind: "VALUE", value }),
() => Object.freeze({ kind: "THREW" }),
);
try {
const outcome = await Promise.race([completed, interrupted]);
void deadline;
return outcome;
} finally {
removeAbortListener();
deadlineController.abort();
}
}
function parseAttemptResult<Value>(
result: unknown,
): BoundedPollAttemptResult<Value> | null {
const outer = snapshotDataRecord(result, [
["error", "ok"],
["ok", "value"],
]);
if (!outer) return null;
if (outer.ok === false) {
const error = snapshotDataRecord(outer.error, [
["kind", "operation", "retryable"],
["kind", "operation", "retryable", "retryAfterMs"],
]);
if (
!error ||
!REALTIME_FAILURE_KINDS.includes(
error.kind as RealtimeFailureKind,
) ||
error.operation !== "POLL" ||
typeof error.retryable !== "boolean" ||
(Object.hasOwn(error, "retryAfterMs") &&
(!Number.isSafeInteger(error.retryAfterMs) ||
(error.retryAfterMs as number) < 0))
) {
return null;
}
const canonicalError = Object.freeze({
kind: error.kind as RealtimeFailureKind,
operation: "POLL" as const,
retryable: error.retryable,
...(Object.hasOwn(error, "retryAfterMs")
? { retryAfterMs: error.retryAfterMs as number }
: {}),
});
return Object.freeze({
ok: false as const,
error: canonicalError,
});
}
if (outer.ok !== true) return null;
const value = snapshotDataRecord(outer.value, [
["kind", "responseBytes"],
["kind", "responseBytes", "state", "value"],
["kind", "responseBytes", "value"],
]);
if (
!value ||
!Number.isSafeInteger(value.responseBytes) ||
(value.responseBytes as number) < 0
) {
return null;
}
if (value.kind === "UNCHANGED") {
return value.responseBytes === 0
? Object.freeze({
ok: true as const,
value: Object.freeze({
kind: "UNCHANGED" as const,
responseBytes: 0 as const,
}),
})
: null;
}
if (
value.kind !== "VALUE" ||
(Object.hasOwn(value, "state") &&
(typeof value.state !== "string" ||
!SAFE_STATE.test(value.state)))
) {
return null;
}
return Object.freeze({
ok: true as const,
value: Object.freeze({
kind: "VALUE" as const,
value: value.value as Value,
responseBytes: value.responseBytes as number,
...(Object.hasOwn(value, "state")
? { state: value.state as string }
: {}),
}),
});
}
function snapshotDataRecord(
value: unknown,
allowedKeySets: readonly (readonly string[])[],
): Readonly<Record<string, unknown>> | null {
if (!value || typeof value !== "object" || Array.isArray(value)) {
return null;
}
try {
if (Object.getPrototypeOf(value) !== Object.prototype) {
return null;
}
const keys = Reflect.ownKeys(value);
if (keys.some((key) => typeof key !== "string")) return null;
const sortedKeys = (keys as string[]).sort();
if (
!allowedKeySets.some((allowed) => {
const sortedAllowed = [...allowed].sort();
return (
sortedKeys.length === sortedAllowed.length &&
sortedKeys.every(
(key, index) => key === sortedAllowed[index],
)
);
})
) {
return null;
}
const descriptors = Object.getOwnPropertyDescriptors(value);
const snapshot: Record<string, unknown> = {};
for (const key of sortedKeys) {
const descriptor = descriptors[key];
if (!descriptor || !Object.hasOwn(descriptor, "value")) {
return null;
}
snapshot[key] = descriptor.value;
}
return Object.freeze(snapshot);
} catch {
return null;
}
}
function retryDelay(
policy: PollLeasePolicy,
consecutiveFailures: number,
retryAfterMs: number | undefined,
random: () => number,
): number | null {
const sample = safeRandom(random);
if (sample === null) return null;
if (
retryAfterMs !== undefined &&
(!Number.isSafeInteger(retryAfterMs) ||
retryAfterMs < 0 ||
retryAfterMs > policy.maxIntervalMs)
) {
return null;
}
const ceiling = Math.min(
policy.maxIntervalMs,
policy.minimumIntervalMs * 2 ** Math.max(0, consecutiveFailures - 1),
);
return Math.max(
policy.minimumIntervalMs,
Math.floor(ceiling * sample),
retryAfterMs ?? 0,
);
}
function isPollRetryableFailureKind(
kind: RealtimeFailureKind,
): boolean {
return POLL_RETRYABLE_FAILURE_KINDS.includes(
kind as (typeof POLL_RETRYABLE_FAILURE_KINDS)[number],
);
}
function isPollRetryAfterFailureKind(
kind: RealtimeFailureKind,
): boolean {
return POLL_RETRY_AFTER_FAILURE_KINDS.includes(
kind as (typeof POLL_RETRY_AFTER_FAILURE_KINDS)[number],
);
}
function successDelay(
policy: PollLeasePolicy,
random: () => number,
): number | null {
const sample = safeRandom(random);
if (sample === null) return null;
const spread = Math.floor(policy.successIntervalMs * 0.1);
return Math.min(
policy.maxIntervalMs,
Math.max(
policy.minimumIntervalMs,
policy.successIntervalMs -
spread +
Math.floor(2 * spread * sample),
),
);
}
function safeRandom(random: () => number): number | null {
try {
const value = random();
return Number.isFinite(value) && value >= 0 && value < 1
? value
: null;
} catch {
return null;
}
}
function safeNow(clock: ClockPort): number | null {
try {
const value = clock.now();
return Number.isFinite(value) ? value : null;
} catch {
return null;
}
}
function safeIsCurrent(isCurrent: () => boolean): boolean {
try {
return isCurrent() === true;
} catch {
return false;
}
}
function safeVisibility(
environment: BoundedPollEnvironment,
): "HIDDEN" | "VISIBLE" | null {
try {
const value = environment.visibility();
return value === "HIDDEN" || value === "VISIBLE" ? value : null;
} catch {
return null;
}
}
function safeOnline(
environment: BoundedPollEnvironment,
): boolean | null {
try {
const value = environment.online();
return typeof value === "boolean" ? value : null;
} catch {
return null;
}
}
function safelyUnsubscribe(
unsubscribe: (() => void) | undefined,
): void {
try {
unsubscribe?.();
} catch {
// Lifecycle cleanup remains terminal even for a throwing host.
}
}
function pollFailure(
kind: RealtimeFailureKind,
retryable = false,
): BoundedPollResult<never> {
return realtimeFailure(kind, "POLL", retryable);
}
function applyFailure(): BoundedPollResult<never> {
return realtimeFailure("APPLY_FAILED", "APPLY", false);
}
+10
View File
@@ -0,0 +1,10 @@
export {
createBoundedPollCoordinator,
type BoundedPollAttemptFailure,
type BoundedPollAttemptResult,
type BoundedPollCoordinator,
type BoundedPollCoordinatorDependencies,
type BoundedPollEnvironment,
type BoundedPollResult,
type BoundedPollRunInput,
} from "./bounded-poll-coordinator.ts";
File diff suppressed because it is too large Load Diff
+224
View File
@@ -0,0 +1,224 @@
export const REALTIME_RECONNECT_CEILINGS = Object.freeze({
drainTimeoutMs: 2_000,
maxAttempts: 10,
maxDrainTimeoutMs: 30_000,
maxElapsedMs: 5 * 60 * 1_000,
maxDelayMs: 60_000,
maxStableOpenMs: 60_000,
});
export type ReconnectPolicy = Readonly<{
baseDelayMs: number;
maxDelayMs: number;
maxAttempts: number;
maxElapsedMs: number;
stableOpenMs: number;
}>;
const RECONNECT_POLICY_KEYS = Object.freeze([
"baseDelayMs",
"maxDelayMs",
"maxAttempts",
"maxElapsedMs",
"stableOpenMs",
] as const);
export type ReconnectDelayInput = Readonly<{
policy: ReconnectPolicy;
/**
* Zero-based number of the reconnect that is about to be scheduled.
*/
attemptIndex: number;
remainingElapsedMs: number;
random: () => number;
/**
* Relative delay required by Retry-After, SSE retry or another validated
* protocol hint. It is a lower bound, never a value to clamp downward.
*/
serverNotBeforeMs?: number | null;
}>;
export function defineReconnectPolicy(
input: ReconnectPolicy,
): ReconnectPolicy {
const snapshot = snapshotPolicy(input);
if (
!snapshot ||
!positiveInteger(snapshot.baseDelayMs) ||
!positiveInteger(snapshot.maxDelayMs) ||
snapshot.baseDelayMs > snapshot.maxDelayMs ||
snapshot.maxDelayMs >
REALTIME_RECONNECT_CEILINGS.maxDelayMs ||
!positiveInteger(snapshot.maxAttempts) ||
snapshot.maxAttempts >
REALTIME_RECONNECT_CEILINGS.maxAttempts ||
!positiveInteger(snapshot.maxElapsedMs) ||
snapshot.maxElapsedMs >
REALTIME_RECONNECT_CEILINGS.maxElapsedMs ||
!positiveInteger(snapshot.stableOpenMs) ||
snapshot.stableOpenMs >
REALTIME_RECONNECT_CEILINGS.maxStableOpenMs
) {
throw new TypeError("Invalid realtime reconnect policy.");
}
return Object.freeze(snapshot);
}
/**
* Full-jitter exponential backoff with a server-provided not-before floor.
* `null` means the attempt budget cannot safely admit another delay.
*/
export function calculateReconnectDelay(
input: ReconnectDelayInput,
): number | null {
const { policy } = input;
if (
!Number.isSafeInteger(input.attemptIndex) ||
input.attemptIndex < 0 ||
input.attemptIndex >= policy.maxAttempts ||
!Number.isFinite(input.remainingElapsedMs) ||
input.remainingElapsedMs <= 0
) {
return null;
}
let sample: number;
try {
sample = input.random();
} catch {
return null;
}
if (!Number.isFinite(sample) || sample < 0 || sample >= 1) {
return null;
}
const exponentialCeiling = Math.min(
policy.maxDelayMs,
policy.baseDelayMs * 2 ** input.attemptIndex,
);
const localDelay = Math.floor(exponentialCeiling * sample);
const serverNotBeforeMs = input.serverNotBeforeMs ?? 0;
if (
!Number.isSafeInteger(serverNotBeforeMs) ||
serverNotBeforeMs < 0 ||
serverNotBeforeMs > policy.maxDelayMs
) {
return null;
}
const effectiveDelay = Math.max(localDelay, serverNotBeforeMs);
return effectiveDelay >= input.remainingElapsedMs
? null
: effectiveDelay;
}
export function reconnectBudgetRemaining(
policy: ReconnectPolicy,
startedAtMs: number,
nowMs: number,
): number {
if (
!Number.isFinite(startedAtMs) ||
!Number.isFinite(nowMs) ||
nowMs < startedAtMs
) {
return 0;
}
return Math.max(0, policy.maxElapsedMs - (nowMs - startedAtMs));
}
export function isReconnectAttemptResetEligible(input: Readonly<{
policy: ReconnectPolicy;
openedAtMs: number;
nowMs: number;
observedValidHeartbeatOrEvent: boolean;
}>): boolean {
if (input.observedValidHeartbeatOrEvent) return true;
return (
Number.isFinite(input.openedAtMs) &&
Number.isFinite(input.nowMs) &&
input.nowMs - input.openedAtMs >= input.policy.stableOpenMs
);
}
/**
* Parses the HTTP Retry-After delay without applying a runtime ceiling.
* Callers must reject a value that exceeds their remaining/max-delay budget.
*/
export function parseRetryAfterDelay(
value: string | null | undefined,
nowEpochMs: number,
): number | null {
if (
typeof value !== "string" ||
value.length === 0 ||
value.length > 128 ||
!Number.isFinite(nowEpochMs)
) {
return null;
}
const normalized = value.trim();
if (/^\d+$/u.test(normalized)) {
const seconds = Number(normalized);
return Number.isSafeInteger(seconds) &&
seconds <= Math.floor(Number.MAX_SAFE_INTEGER / 1_000)
? seconds * 1_000
: null;
}
if (
!/^(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun), \d{2} (?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec) \d{4} \d{2}:\d{2}:\d{2} GMT$/u.test(
normalized,
)
) {
return null;
}
const timestamp = Date.parse(normalized);
return Number.isFinite(timestamp)
? Math.max(0, timestamp - nowEpochMs)
: null;
}
function positiveInteger(value: number): boolean {
return Number.isSafeInteger(value) && value > 0;
}
function snapshotPolicy(input: unknown): ReconnectPolicy | null {
if (
!input ||
typeof input !== "object" ||
Array.isArray(input)
) {
return null;
}
try {
if (Object.getPrototypeOf(input) !== Object.prototype) {
return null;
}
const ownKeys = Reflect.ownKeys(input);
if (
ownKeys.length !== RECONNECT_POLICY_KEYS.length ||
RECONNECT_POLICY_KEYS.some(
(key) => !ownKeys.includes(key),
)
) {
return null;
}
const descriptors = Object.getOwnPropertyDescriptors(input);
if (
RECONNECT_POLICY_KEYS.some((key) => {
const descriptor = descriptors[key];
return !descriptor || !Object.hasOwn(descriptor, "value");
})
) {
return null;
}
return {
baseDelayMs: descriptors.baseDelayMs!.value as number,
maxDelayMs: descriptors.maxDelayMs!.value as number,
maxAttempts: descriptors.maxAttempts!.value as number,
maxElapsedMs: descriptors.maxElapsedMs!.value as number,
stableOpenMs: descriptors.stableOpenMs!.value as number,
};
} catch {
return null;
}
}
+278
View File
@@ -0,0 +1,278 @@
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<RealtimeFailureKind>([
"OFFLINE",
"CONNECT_TIMEOUT",
"IDLE_TIMEOUT",
"RATE_LIMITED",
"PROVIDER_UNAVAILABLE",
]);
const FAILURE_KINDS = new Set<unknown>(REALTIME_FAILURE_KINDS);
const OPERATIONS = new Set<unknown>(REALTIME_OPERATIONS);
export type RealtimeDataSnapshot = Readonly<{
keys: readonly string[];
values: Readonly<Record<string, unknown>>;
frozen: boolean;
}>;
export function realtimeSuccess<Value>(
value: Value,
): Extract<RealtimeResult<Value>, { ok: true }> {
return Object.freeze({ ok: true, value });
}
export function realtimeFailure(
kind: RealtimeFailureKind,
operation: RealtimeOperation,
retryable = DEFAULT_RETRYABLE.has(kind),
): Extract<RealtimeResult<never>, { 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>(
value: unknown,
isValue: (candidate: unknown) => candidate is Value,
): RealtimeResult<Value> | null {
return parseRealtimeResult(value, isValue, false);
}
export function isRealtimeResult<Value>(
value: unknown,
isValue: (candidate: unknown) => candidate is Value,
): value is RealtimeResult<Value> {
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<string, unknown>;
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>(
value: unknown,
isValue: (candidate: unknown) => candidate is Value,
requireFrozenSource: boolean,
): RealtimeResult<Value> | 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])
);
}
@@ -0,0 +1,738 @@
import type { ClockPort } from "../../../application/ports/clock-port.ts";
import type {
RealtimeFailureKind,
RealtimeOperation,
RealtimeResult,
} from "../../../application/ports/realtime/shared.ts";
import type {
RealtimeTransportEventOutcome,
} from "../../../application/ports/realtime/event-authority.ts";
import {
REALTIME_TRANSPORT_CONTINUE,
} from "../../../application/ports/realtime/event-authority.ts";
import { isRealtimeResumeCursor } from "../../../contracts/realtime-events.ts";
import { systemClock } from "../../platform/system-clock.ts";
import { parseRetryAfterDelay } from "../reconnect-policy.ts";
import {
isRealtimeResult,
isRealtimeTransportEventOutcome,
realtimeFailure,
realtimeSuccess,
} from "../result.ts";
import {
createIncrementalSseParser,
type ParsedSseEvent,
type SseParserItem,
type SseParserLimits,
} from "./sse-parser.ts";
export type SseRecoveryMode =
| "CURSOR"
| "SESSION_REBUILD"
| "SNAPSHOT_ONLY";
export type FetchSseClosedOutcome =
| Readonly<{
kind: "EOF";
incompleteEventDiscarded: boolean;
retryHintMs: number | null;
}>
| Readonly<{ kind: "NO_RECONNECT" }>
| Extract<
RealtimeTransportEventOutcome,
{ kind: "RECOVERY_COMMITTED" }
>;
export type SseInboundEventOutcome =
RealtimeTransportEventOutcome;
export const SSE_CONTINUE: SseInboundEventOutcome =
REALTIME_TRANSPORT_CONTINUE;
export type FetchSseReadInput = Readonly<{
resumeCursor: string | null;
signal?: AbortSignal;
/**
* Runs after the response and stream contract are validated but before any
* event bytes are consumed. A reconnect bridge can hold this gate until the
* exact recovery checkpoint's replay barrier is confirmed.
*/
onOpen?(
signal: AbortSignal,
):
| RealtimeResult<void>
| Promise<RealtimeResult<void>>;
onEvent(
event: ParsedSseEvent,
signal: AbortSignal,
):
| RealtimeResult<SseInboundEventOutcome>
| Promise<RealtimeResult<SseInboundEventOutcome>>;
onComment?: () => void;
onRetryHint?: (retryMs: number) => void;
}>;
export type FetchSseConnection = Readonly<{
read(
input: FetchSseReadInput,
): Promise<RealtimeResult<FetchSseClosedOutcome>>;
close(): void;
}>;
export type FetchSseConnectionDependencies = Readonly<{
endpoint: string;
applicationOrigin: string;
recoveryMode: SseRecoveryMode;
fetcher?: typeof fetch;
clock?: ClockPort;
parserLimits?: Partial<SseParserLimits>;
connectTimeoutMs?: number;
idleTimeoutMs?: number;
maxCursorBytes?: number;
maxRetryAfterMs?: number;
}>;
const DEFAULT_CONNECT_TIMEOUT_MS = 10_000;
const DEFAULT_IDLE_TIMEOUT_MS = 45_000;
const DEFAULT_MAX_CURSOR_BYTES = 1_024;
const DEFAULT_MAX_RETRY_AFTER_MS = 60_000;
const MAX_CONNECT_TIMEOUT_MS = 30_000;
const MAX_IDLE_TIMEOUT_MS = 120_000;
const MAX_CURSOR_BYTES = 1_024;
const READER_CANCEL_TIMEOUT_MS = 2_000;
export function createFetchSseConnection(
dependencies: FetchSseConnectionDependencies,
): FetchSseConnection {
const endpoint = fixedEndpoint(
dependencies.endpoint,
dependencies.applicationOrigin,
);
const fetcher = dependencies.fetcher ?? fetch;
const clock = dependencies.clock ?? systemClock;
const connectTimeoutMs =
dependencies.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS;
const idleTimeoutMs =
dependencies.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS;
const maxCursorBytes =
dependencies.maxCursorBytes ?? DEFAULT_MAX_CURSOR_BYTES;
const maxRetryAfterMs =
dependencies.maxRetryAfterMs ?? DEFAULT_MAX_RETRY_AFTER_MS;
validateDependencies(
dependencies.recoveryMode,
connectTimeoutMs,
idleTimeoutMs,
maxCursorBytes,
maxRetryAfterMs,
);
// Validate immutable parser policy at factory construction, before network
// side effects. A fresh parser is still created for every physical attempt.
createIncrementalSseParser(dependencies.parserLimits);
let closed = false;
let active = false;
let activeController: AbortController | null = null;
let activeReader: ReadableStreamDefaultReader<Uint8Array> | null = null;
async function read(
input: FetchSseReadInput,
): Promise<RealtimeResult<FetchSseClosedOutcome>> {
if (closed) return failed("CLOSED", "CONNECT", false);
if (active) {
return failed("PROTOCOL_MISMATCH", "CONNECT", false);
}
if (
!validResumeCursor(
input.resumeCursor,
dependencies.recoveryMode,
maxCursorBytes,
)
) {
return failed("PROTOCOL_MISMATCH", "CONNECT", false);
}
if (input.signal?.aborted) {
return failed("ABORTED", "CONNECT", false);
}
active = true;
const controller = new AbortController();
activeController = controller;
const onCallerAbort = () => controller.abort();
input.signal?.addEventListener("abort", onCallerAbort, {
once: true,
});
if (input.signal?.aborted) onCallerAbort();
try {
const request = timed(
Promise.resolve().then(() =>
fetcher(endpoint.href, {
method: "GET",
credentials: "same-origin",
redirect: "error",
cache: "no-store",
referrerPolicy: "no-referrer",
headers: {
Accept: "text/event-stream",
...(input.resumeCursor === null
? {}
: { "Last-Event-ID": input.resumeCursor }),
},
signal: controller.signal,
}),
),
connectTimeoutMs,
clock,
controller.signal,
);
const responseResult = await request;
if (responseResult.kind === "CLOCK_FAILED") {
controller.abort();
return failed("PROVIDER_UNAVAILABLE", "CONNECT", true);
}
if (responseResult.kind === "ABORTED") {
return failed("ABORTED", "CONNECT", false);
}
if (responseResult.kind === "TIMEOUT") {
controller.abort();
return failed("CONNECT_TIMEOUT", "CONNECT", true);
}
if (responseResult.kind === "REJECTED") {
return failed(
controller.signal.aborted ? "ABORTED" : "OFFLINE",
"CONNECT",
!controller.signal.aborted,
);
}
const response = responseResult.value;
if (response.redirected) {
return failed("PROTOCOL_MISMATCH", "CONNECT", false);
}
if (response.status === 204) {
return succeeded(Object.freeze({ kind: "NO_RECONNECT" }));
}
if (response.status !== 200) {
const responseObservedAt = readClockNow(clock);
if (responseObservedAt === null) {
controller.abort();
return failed(
"PROVIDER_UNAVAILABLE",
"CONNECT",
true,
);
}
return responseFailure(
response,
responseObservedAt,
maxRetryAfterMs,
input.onRetryHint,
);
}
if (!isEventStreamContentType(response.headers.get("content-type"))) {
return failed("PROTOCOL_MISMATCH", "CONNECT", false);
}
if (!response.body) {
return failed("MALFORMED_EVENT", "RECEIVE", false);
}
const parser = createIncrementalSseParser(
dependencies.parserLimits,
);
const reader = response.body.getReader();
activeReader = reader;
let retryHintMs: number | null = null;
if (input.onOpen) {
let opening: Promise<RealtimeResult<void>>;
try {
opening = Promise.resolve(input.onOpen(controller.signal));
} catch {
await cancelReader(reader, clock, controller);
return failed(
"PROVIDER_UNAVAILABLE",
"CONNECT",
false,
);
}
const opened = await timed(
opening,
connectTimeoutMs,
clock,
controller.signal,
);
if (opened.kind !== "VALUE") {
await cancelReader(reader, clock, controller);
if (opened.kind === "ABORTED") {
return failed("ABORTED", "CONNECT", false);
}
if (opened.kind === "TIMEOUT") {
return failed("CONNECT_TIMEOUT", "CONNECT", true);
}
return failed(
"PROVIDER_UNAVAILABLE",
"CONNECT",
false,
);
}
if (!isRealtimeResult(opened.value, isUndefined)) {
await cancelReader(reader, clock, controller);
return failed(
"PROTOCOL_MISMATCH",
"CONNECT",
false,
);
}
if (!opened.value.ok) {
await cancelReader(reader, clock, controller);
return opened.value;
}
}
async function handleParserItems(
items: readonly SseParserItem[],
): Promise<RealtimeResult<FetchSseClosedOutcome> | null> {
for (const item of items) {
if (item.kind === "COMMENT") {
safelyNotify(input.onComment);
continue;
}
if (item.kind === "RETRY") {
retryHintMs = item.retryMs;
safelyNotify(input.onRetryHint, item.retryMs);
continue;
}
if (
dependencies.recoveryMode === "CURSOR" &&
(!item.hasExplicitId ||
!item.id ||
!isRealtimeResumeCursor(item.id) ||
new TextEncoder().encode(item.id).byteLength >
maxCursorBytes)
) {
await cancelReader(reader, clock, controller);
return failed("PROTOCOL_MISMATCH", "DECODE", false);
}
if (
dependencies.recoveryMode !== "CURSOR" &&
(item.hasExplicitId || item.id !== null)
) {
await cancelReader(reader, clock, controller);
return failed("PROTOCOL_MISMATCH", "DECODE", false);
}
let handler: Promise<
RealtimeResult<SseInboundEventOutcome>
>;
try {
handler = Promise.resolve(
input.onEvent(item, controller.signal),
);
} catch {
await cancelReader(reader, clock, controller);
return failed("APPLY_FAILED", "APPLY", false);
}
const handled = await timed(
handler,
idleTimeoutMs,
clock,
controller.signal,
);
if (handled.kind === "ABORTED") {
await cancelReader(reader, clock, controller);
return failed("ABORTED", "APPLY", false);
}
if (handled.kind === "CLOCK_FAILED") {
await cancelReader(reader, clock, controller);
return failed(
"PROVIDER_UNAVAILABLE",
"APPLY",
false,
);
}
if (
handled.kind === "REJECTED" ||
handled.kind === "TIMEOUT"
) {
await cancelReader(reader, clock, controller);
return failed("APPLY_FAILED", "APPLY", false);
}
if (
handled.kind !== "VALUE" ||
!isRealtimeResult(
handled.value,
isRealtimeTransportEventOutcome,
)
) {
await cancelReader(reader, clock, controller);
return failed("APPLY_FAILED", "APPLY", false);
}
if (!handled.value.ok) {
await cancelReader(reader, clock, controller);
return handled.value;
}
if (
handled.value.value.kind === "RECOVERY_COMMITTED"
) {
await cancelReader(reader, clock, controller);
return succeeded(handled.value.value);
}
if (controller.signal.aborted) {
await cancelReader(reader, clock, controller);
return failed("ABORTED", "APPLY", false);
}
}
return null;
}
while (true) {
const readResult = await timed(
reader.read(),
idleTimeoutMs,
clock,
controller.signal,
);
if (readResult.kind === "ABORTED") {
await cancelReader(reader, clock, controller);
return failed("ABORTED", "RECEIVE", false);
}
if (readResult.kind === "CLOCK_FAILED") {
controller.abort();
await cancelReader(reader, clock, controller);
return failed(
"PROVIDER_UNAVAILABLE",
"RECEIVE",
true,
);
}
if (readResult.kind === "TIMEOUT") {
controller.abort();
await cancelReader(reader, clock, controller);
return failed("IDLE_TIMEOUT", "RECEIVE", true);
}
if (readResult.kind === "REJECTED") {
return failed(
controller.signal.aborted ? "ABORTED" : "OFFLINE",
"RECEIVE",
!controller.signal.aborted,
);
}
if (readResult.value.done) {
const finished = parser.finish();
if (!finished.ok) return finished;
const dispatchFailure = await handleParserItems(
finished.value.items,
);
if (dispatchFailure) return dispatchFailure;
return succeeded(
Object.freeze({
kind: "EOF",
incompleteEventDiscarded:
finished.value.incompleteEventDiscarded,
retryHintMs,
}),
);
}
if (!(readResult.value.value instanceof Uint8Array)) {
await cancelReader(reader, clock, controller);
return failed("MALFORMED_EVENT", "DECODE", false);
}
const parsed = parser.push(readResult.value.value);
if (!parsed.ok) {
await cancelReader(reader, clock, controller);
return parsed;
}
const dispatchFailure = await handleParserItems(parsed.value);
if (dispatchFailure) return dispatchFailure;
}
} catch {
return failed(
controller.signal.aborted ? "ABORTED" : "MALFORMED_EVENT",
activeReader ? "RECEIVE" : "CONNECT",
false,
);
} finally {
input.signal?.removeEventListener("abort", onCallerAbort);
controller.abort();
if (activeReader) {
try {
activeReader.releaseLock();
} catch {
// The terminal outcome is already determined.
}
}
activeReader = null;
activeController = null;
active = false;
}
}
function close(): void {
if (closed) return;
closed = true;
activeController?.abort();
if (activeReader) {
void cancelReader(
activeReader,
clock,
activeController ?? undefined,
);
}
}
return Object.freeze({ read, close });
}
function fixedEndpoint(endpoint: string, applicationOrigin: string): URL {
let parsedEndpoint: URL;
let parsedOrigin: URL;
try {
parsedEndpoint = new URL(endpoint);
parsedOrigin = new URL(applicationOrigin);
} catch {
throw new TypeError("SSE endpoint must be an absolute URL.");
}
if (
parsedEndpoint.protocol !== "https:" ||
parsedEndpoint.origin !== parsedOrigin.origin ||
parsedEndpoint.username ||
parsedEndpoint.password ||
parsedEndpoint.search ||
parsedEndpoint.hash
) {
throw new TypeError("SSE endpoint must be fixed same-origin HTTPS.");
}
return parsedEndpoint;
}
function validateDependencies(
recoveryMode: SseRecoveryMode,
connectTimeoutMs: number,
idleTimeoutMs: number,
maxCursorBytes: number,
maxRetryAfterMs: number,
): void {
if (
!["CURSOR", "SESSION_REBUILD", "SNAPSHOT_ONLY"].includes(
recoveryMode,
) ||
!integerWithin(connectTimeoutMs, 1, MAX_CONNECT_TIMEOUT_MS) ||
!integerWithin(idleTimeoutMs, 1, MAX_IDLE_TIMEOUT_MS) ||
!integerWithin(maxCursorBytes, 1, MAX_CURSOR_BYTES) ||
!integerWithin(maxRetryAfterMs, 1, DEFAULT_MAX_RETRY_AFTER_MS)
) {
throw new TypeError("Invalid fetch SSE connection policy.");
}
}
function validResumeCursor(
cursor: string | null,
recoveryMode: SseRecoveryMode,
maxCursorBytes: number,
): boolean {
if (recoveryMode !== "CURSOR") return cursor === null;
if (cursor === null) return true;
return (
typeof cursor === "string" &&
isRealtimeResumeCursor(cursor) &&
new TextEncoder().encode(cursor).byteLength <= maxCursorBytes
);
}
function isEventStreamContentType(value: string | null): boolean {
if (typeof value !== "string" || value.length > 128) return false;
const parts = value.split(";").map((part) => part.trim().toLowerCase());
if (parts[0] !== "text/event-stream") return false;
if (parts.length === 1) return true;
return (
parts.length === 2 &&
/^(?:charset=utf-8|charset="utf-8")$/u.test(parts[1] ?? "")
);
}
function responseFailure(
response: Response,
nowEpochMs: number,
maxRetryAfterMs: number,
onRetryHint: ((retryMs: number) => void) | undefined,
): RealtimeResult<never> {
const status = response.status;
if (status === 401) {
return failed("AUTH_REQUIRED", "CONNECT", false);
}
if (status === 403) {
return failed("FORBIDDEN", "CONNECT", false);
}
if (status === 409 || status === 410) {
return failed("CURSOR_EXPIRED", "CONNECT", false);
}
const retryAfterMs =
status === 429 || status === 503
? parseRetryAfterDelay(
response.headers.get("retry-after"),
nowEpochMs,
)
: null;
const retryHintAccepted =
retryAfterMs !== null && retryAfterMs <= maxRetryAfterMs;
if (retryHintAccepted) {
safelyNotify(onRetryHint, retryAfterMs);
}
if (status === 429) {
return failed("RATE_LIMITED", "CONNECT", retryHintAccepted);
}
if (status === 503) {
return failed(
"PROVIDER_UNAVAILABLE",
"CONNECT",
retryHintAccepted,
);
}
if (status === 502 || status === 504) {
return failed("PROVIDER_UNAVAILABLE", "CONNECT", true);
}
return failed("PROTOCOL_MISMATCH", "CONNECT", false);
}
type TimedResult<Value> =
| Readonly<{ kind: "VALUE"; value: Value }>
| Readonly<{ kind: "REJECTED" }>
| Readonly<{ kind: "ABORTED" }>
| Readonly<{ kind: "CLOCK_FAILED" }>
| Readonly<{ kind: "TIMEOUT" }>;
async function timed<Value>(
operation: Promise<Value>,
timeoutMs: number,
clock: ClockPort,
signal: AbortSignal,
): Promise<TimedResult<Value>> {
if (signal.aborted) {
return Object.freeze({ kind: "ABORTED" });
}
const timer = new AbortController();
let abortListener: (() => void) | undefined;
const operationResult = operation.then<
TimedResult<Value>,
TimedResult<Value>
>(
(value) => Object.freeze({ kind: "VALUE", value }),
() => Object.freeze({ kind: "REJECTED" }),
);
let timeoutResult: Promise<TimedResult<Value>>;
try {
timeoutResult = clock.sleep(timeoutMs, timer.signal).then<
TimedResult<Value>,
TimedResult<Value>
>(
() => Object.freeze({ kind: "TIMEOUT" }),
() =>
Object.freeze({
kind: timer.signal.aborted
? ("ABORTED" as const)
: ("CLOCK_FAILED" as const),
}),
);
} catch {
return Object.freeze({ kind: "CLOCK_FAILED" });
}
const abortedResult = new Promise<TimedResult<Value>>((resolve) => {
abortListener = () =>
resolve(Object.freeze({ kind: "ABORTED" }));
signal.addEventListener("abort", abortListener, { once: true });
if (signal.aborted) abortListener();
});
const result = await Promise.race([
operationResult,
timeoutResult,
abortedResult,
]);
timer.abort();
if (abortListener) {
signal.removeEventListener("abort", abortListener);
}
return result;
}
function readClockNow(clock: ClockPort): number | null {
try {
const value = clock.now();
return Number.isFinite(value) && value >= 0 ? value : null;
} catch {
return null;
}
}
async function cancelReader(
reader: ReadableStreamDefaultReader<Uint8Array>,
clock: ClockPort,
generation?: AbortController,
): Promise<void> {
generation?.abort();
let cancellation: Promise<void>;
try {
cancellation = Promise.resolve(reader.cancel()).then(
() => undefined,
() => undefined,
);
} catch {
return;
}
const timeout = new AbortController();
let timeoutPromise: Promise<void>;
try {
timeoutPromise = clock
.sleep(READER_CANCEL_TIMEOUT_MS, timeout.signal)
.then(
() => undefined,
() => undefined,
);
} catch {
timeoutPromise = Promise.resolve();
}
await Promise.race([cancellation, timeoutPromise]);
timeout.abort();
}
function safelyNotify(
callback: ((value?: never) => void) | undefined,
): void;
function safelyNotify<Value>(
callback: ((value: Value) => void) | undefined,
value: Value,
): void;
function safelyNotify<Value>(
callback: ((value: Value) => void) | (() => void) | undefined,
value?: Value,
): void {
try {
if (callback) callback(value as Value);
} catch {
// Observation and retry-hint consumers are best effort.
}
}
function succeeded<Value>(value: Value): RealtimeResult<Value> {
return realtimeSuccess(value);
}
function failed(
kind: RealtimeFailureKind,
operation: RealtimeOperation,
retryable?: boolean,
): RealtimeResult<never> {
return realtimeFailure(kind, operation, retryable);
}
function integerWithin(
value: number,
minimum: number,
maximum: number,
): boolean {
return (
Number.isSafeInteger(value) &&
value >= minimum &&
value <= maximum
);
}
function isUndefined(value: unknown): value is undefined {
return value === undefined;
}
+19
View File
@@ -0,0 +1,19 @@
export {
createFetchSseConnection,
SSE_CONTINUE,
type FetchSseClosedOutcome,
type FetchSseConnection,
type FetchSseConnectionDependencies,
type FetchSseReadInput,
type SseInboundEventOutcome,
type SseRecoveryMode,
} from "./fetch-sse-connection.ts";
export {
createIncrementalSseParser,
SSE_PARSER_CEILINGS,
type IncrementalSseParser,
type ParsedSseEvent,
type SseParserFinish,
type SseParserItem,
type SseParserLimits,
} from "./sse-parser.ts";
+346
View File
@@ -0,0 +1,346 @@
import type { RealtimeResult } from "../../../application/ports/realtime/shared.ts";
import {
realtimeFailure,
realtimeSuccess,
} from "../result.ts";
export const SSE_PARSER_CEILINGS = Object.freeze({
maxLineBytes: 64 * 1_024,
maxEventBytes: 64 * 1_024,
maxIncompleteBufferBytes: 128 * 1_024,
maxChunkBytes: 256 * 1_024,
maxItemsPerChunk: 256,
maxRetryMs: 60_000,
});
export type SseParserLimits = Readonly<{
maxLineBytes: number;
maxEventBytes: number;
maxIncompleteBufferBytes: number;
maxChunkBytes: number;
maxItemsPerChunk: number;
maxRetryMs: number;
}>;
export type ParsedSseEvent = Readonly<{
kind: "EVENT";
eventType: string;
data: string;
/**
* Standard SSE last-event-ID state. Consumers that require cursor-after-
* effect must additionally require `hasExplicitId` and commit independently.
*/
id: string | null;
hasExplicitId: boolean;
}>;
export type SseParserItem =
| ParsedSseEvent
| Readonly<{ kind: "COMMENT" }>
| Readonly<{ kind: "RETRY"; retryMs: number }>;
export type SseParserFinish = Readonly<{
items: readonly SseParserItem[];
incompleteEventDiscarded: boolean;
}>;
export type IncrementalSseParser = Readonly<{
push(chunk: Uint8Array): RealtimeResult<readonly SseParserItem[]>;
finish(): RealtimeResult<SseParserFinish>;
}>;
export function createIncrementalSseParser(
limits: Partial<SseParserLimits> = {},
): IncrementalSseParser {
const resolved = resolveLimits(limits);
const decoder = new TextDecoder("utf-8", {
fatal: true,
ignoreBOM: false,
});
const encoder = new TextEncoder();
let state: "OPEN" | "FAILED" | "FINISHED" = "OPEN";
let atStart = true;
let pendingCarriageReturn = false;
let line = "";
let lineBytes = 0;
let blockBytes = 0;
let dataLines: string[] = [];
let eventType = "";
let lastEventId: string | null = null;
let hasExplicitId = false;
function push(
chunk: Uint8Array,
): RealtimeResult<readonly SseParserItem[]> {
if (state !== "OPEN") {
return realtimeFailure("CLOSED", "DECODE");
}
if (!(chunk instanceof Uint8Array)) {
return fail("MALFORMED_EVENT");
}
if (chunk.byteLength > resolved.maxChunkBytes) {
return fail("EVENT_TOO_LARGE");
}
let text: string;
try {
text = decoder.decode(chunk, { stream: true });
} catch {
return fail("MALFORMED_EVENT");
}
return consumeText(text);
}
function finish(): RealtimeResult<SseParserFinish> {
if (state !== "OPEN") {
return realtimeFailure("CLOSED", "DECODE");
}
let tail: string;
try {
tail = decoder.decode();
} catch {
return fail("MALFORMED_EVENT");
}
const consumed = consumeText(tail);
if (!consumed.ok) return consumed;
const items = [...consumed.value];
if (pendingCarriageReturn) {
pendingCarriageReturn = false;
const processed = processLine(1);
if (!processed.ok) return processed;
if (!appendItems(items, processed.value)) {
return fail("QUEUE_OVERFLOW");
}
}
const incompleteEventDiscarded =
lineBytes > 0 ||
blockBytes > 0 ||
dataLines.length > 0 ||
eventType.length > 0 ||
hasExplicitId;
clearBlock();
line = "";
lineBytes = 0;
state = "FINISHED";
return success(
Object.freeze({
items: Object.freeze(items),
incompleteEventDiscarded,
}),
);
}
function consumeText(
text: string,
): RealtimeResult<readonly SseParserItem[]> {
const items: SseParserItem[] = [];
for (const character of text) {
if (atStart) {
atStart = false;
if (character === "\uFEFF") continue;
}
if (pendingCarriageReturn) {
pendingCarriageReturn = false;
const processed = processLine(character === "\n" ? 2 : 1);
if (!processed.ok) return processed;
if (!appendItems(items, processed.value)) {
return fail("QUEUE_OVERFLOW");
}
if (character === "\n") continue;
}
if (character === "\r") {
pendingCarriageReturn = true;
continue;
}
if (character === "\n") {
const processed = processLine(1);
if (!processed.ok) return processed;
if (!appendItems(items, processed.value)) {
return fail("QUEUE_OVERFLOW");
}
continue;
}
line += character;
lineBytes += encoder.encode(character).byteLength;
if (lineBytes > resolved.maxLineBytes) {
return fail("EVENT_TOO_LARGE");
}
if (
lineBytes + blockBytes >
resolved.maxIncompleteBufferBytes
) {
return fail("EVENT_TOO_LARGE");
}
}
return success(Object.freeze(items));
}
function processLine(
terminatorBytes: number,
): RealtimeResult<readonly SseParserItem[]> {
const currentLine = line;
const currentLineBytes = lineBytes;
line = "";
lineBytes = 0;
if (currentLine.length === 0) {
const items: SseParserItem[] = [];
if (dataLines.length > 0) {
items.push(
Object.freeze({
kind: "EVENT",
eventType: eventType.length > 0 ? eventType : "message",
data: dataLines.join("\n"),
id: lastEventId,
hasExplicitId,
}),
);
}
clearBlock();
return success(Object.freeze(items));
}
if (currentLine.startsWith(":")) {
return success(
Object.freeze([
Object.freeze({ kind: "COMMENT" as const }),
]),
);
}
blockBytes += currentLineBytes + terminatorBytes;
if (blockBytes > resolved.maxEventBytes) {
return fail("EVENT_TOO_LARGE");
}
if (blockBytes > resolved.maxIncompleteBufferBytes) {
return fail("EVENT_TOO_LARGE");
}
const separator = currentLine.indexOf(":");
const field =
separator === -1
? currentLine
: currentLine.slice(0, separator);
let value =
separator === -1 ? "" : currentLine.slice(separator + 1);
if (value.startsWith(" ")) value = value.slice(1);
if (field === "data") {
dataLines.push(value);
return success(Object.freeze([]));
}
if (field === "event") {
eventType = value;
return success(Object.freeze([]));
}
if (field === "id") {
if (!value.includes("\0")) {
lastEventId = value;
hasExplicitId = true;
}
return success(Object.freeze([]));
}
if (field === "retry" && /^\d+$/u.test(value)) {
const retryMs = Number(value);
if (
Number.isSafeInteger(retryMs) &&
retryMs <= resolved.maxRetryMs
) {
return success(
Object.freeze([
Object.freeze({ kind: "RETRY" as const, retryMs }),
]),
);
}
}
return success(Object.freeze([]));
}
function clearBlock(): void {
blockBytes = 0;
dataLines = [];
eventType = "";
hasExplicitId = false;
}
function fail(
kind:
| "EVENT_TOO_LARGE"
| "MALFORMED_EVENT"
| "QUEUE_OVERFLOW",
): RealtimeResult<never> {
state = "FAILED";
line = "";
lineBytes = 0;
clearBlock();
return realtimeFailure(kind, "DECODE");
}
function appendItems(
target: SseParserItem[],
additions: readonly SseParserItem[],
): boolean {
if (
target.length + additions.length >
resolved.maxItemsPerChunk
) {
return false;
}
target.push(...additions);
return true;
}
return Object.freeze({ push, finish });
}
function resolveLimits(
input: Partial<SseParserLimits>,
): SseParserLimits {
const limits = {
maxLineBytes:
input.maxLineBytes ?? SSE_PARSER_CEILINGS.maxLineBytes,
maxEventBytes:
input.maxEventBytes ?? SSE_PARSER_CEILINGS.maxEventBytes,
maxIncompleteBufferBytes:
input.maxIncompleteBufferBytes ??
SSE_PARSER_CEILINGS.maxIncompleteBufferBytes,
maxChunkBytes:
input.maxChunkBytes ?? SSE_PARSER_CEILINGS.maxChunkBytes,
maxItemsPerChunk:
input.maxItemsPerChunk ??
SSE_PARSER_CEILINGS.maxItemsPerChunk,
maxRetryMs: input.maxRetryMs ?? SSE_PARSER_CEILINGS.maxRetryMs,
};
if (
!positiveInteger(limits.maxLineBytes) ||
limits.maxLineBytes > SSE_PARSER_CEILINGS.maxLineBytes ||
!positiveInteger(limits.maxEventBytes) ||
limits.maxEventBytes > SSE_PARSER_CEILINGS.maxEventBytes ||
!positiveInteger(limits.maxIncompleteBufferBytes) ||
limits.maxIncompleteBufferBytes >
SSE_PARSER_CEILINGS.maxIncompleteBufferBytes ||
limits.maxIncompleteBufferBytes < limits.maxEventBytes ||
!positiveInteger(limits.maxChunkBytes) ||
limits.maxChunkBytes > SSE_PARSER_CEILINGS.maxChunkBytes ||
limits.maxChunkBytes < limits.maxEventBytes ||
!positiveInteger(limits.maxItemsPerChunk) ||
limits.maxItemsPerChunk >
SSE_PARSER_CEILINGS.maxItemsPerChunk ||
!positiveInteger(limits.maxRetryMs) ||
limits.maxRetryMs > SSE_PARSER_CEILINGS.maxRetryMs
) {
throw new TypeError("Invalid SSE parser limits.");
}
return Object.freeze(limits);
}
function success<Value>(value: Value): RealtimeResult<Value> {
return realtimeSuccess(value);
}
function positiveInteger(value: number): boolean {
return Number.isSafeInteger(value) && value > 0;
}
File diff suppressed because it is too large Load Diff
+43
View File
@@ -0,0 +1,43 @@
export {
createWebSocketConnection,
WEBSOCKET_IMPLEMENTATION_CEILINGS,
type WebSocketClientCeilings,
type WebSocketClosedReceipt,
type WebSocketConnection,
type WebSocketConnectionDependencies,
type WebSocketConnectionObservation,
type WebSocketConnectionSnapshot,
type WebSocketConnectionStatus,
type WebSocketFacade,
type WebSocketInboundEventOutcome,
type WebSocketLocalSendReceipt,
type WebSocketOpenReceipt,
type WebSocketRecoveryRequest,
type WebSocketResumeCheckpoint,
type WebSocketSubscribedReceipt,
type WebSocketSubscriptionRequest,
} from "./websocket-connection.ts";
export {
decodeWebSocketServerFrame,
encodeWebSocketClientFrame,
nextUnsignedSequence,
REALTIME_WEBSOCKET_PROTOCOL,
type WebSocketAdvertisedLimits,
type WebSocketClientCloseFrame,
type WebSocketClientFrame,
type WebSocketCloseCategory,
type WebSocketEventFrame,
type WebSocketHeartbeatAckFrame,
type WebSocketHeartbeatFrame,
type WebSocketProtocolFailure,
type WebSocketProtocolResult,
type WebSocketResetReason,
type WebSocketResetRequiredFrame,
type WebSocketServerCloseFrame,
type WebSocketServerFrame,
type WebSocketSubscribedFrame,
type WebSocketSubscribeFrame,
type WebSocketUnsubscribedFrame,
type WebSocketUnsubscribeFrame,
type WebSocketWelcomeFrame,
} from "./websocket-protocol.ts";
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,518 @@
import {
hasDuplicateJsonMembers,
} from "../json-member-scanner.ts";
export const REALTIME_WEBSOCKET_PROTOCOL = "realtime.v1" as const;
export type WebSocketCloseCategory =
| "NORMAL"
| "RESTART"
| "OVERLOADED"
| "AUTH_REQUIRED"
| "FORBIDDEN"
| "PROTOCOL_MISMATCH"
| "CURSOR_RESET"
| "NETWORK_LOST";
export type WebSocketResetReason =
| "CURSOR_EXPIRED"
| "SEQUENCE_GAP"
| "SERVER_RESET"
| "SCOPE_CHANGED";
export type WebSocketAdvertisedLimits = Readonly<{
maxFrameBytes: number;
maxSubscriptions: number;
maxInboundQueueCount: number;
maxInboundQueueBytes: number;
maxOutboundQueueCount: number;
maxOutboundQueueBytes: number;
maxBufferedAmountBytes: number;
maxEventsPerSecond: number;
}>;
export type WebSocketWelcomeFrame = Readonly<{
type: "WELCOME";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
connectionId: string;
heartbeatMs: number;
heartbeatAckTimeoutMs: number;
limits: WebSocketAdvertisedLimits;
}>;
export type WebSocketSubscribedFrame = Readonly<{
type: "SUBSCRIBED";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
streamEpoch: string;
acceptedCursor: string | null;
nextExpectedSequence: string;
}>;
export type WebSocketUnsubscribedFrame = Readonly<{
type: "UNSUBSCRIBED";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
}>;
export type WebSocketEventFrame = Readonly<{
type: "EVENT";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
envelope: Readonly<Record<string, unknown>>;
}>;
export type WebSocketResetRequiredFrame = Readonly<{
type: "RESET_REQUIRED";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
reason: WebSocketResetReason;
}>;
export type WebSocketHeartbeatAckFrame = Readonly<{
type: "HEARTBEAT_ACK";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
nonce: string;
}>;
export type WebSocketServerCloseFrame = Readonly<{
type: "CLOSE";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
category: WebSocketCloseCategory;
}>;
export type WebSocketServerFrame =
| WebSocketWelcomeFrame
| WebSocketSubscribedFrame
| WebSocketUnsubscribedFrame
| WebSocketEventFrame
| WebSocketResetRequiredFrame
| WebSocketHeartbeatAckFrame
| WebSocketServerCloseFrame;
export type WebSocketSubscribeFrame = Readonly<{
type: "SUBSCRIBE";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
streamId: string;
cursor: string | null;
scopeBinding: string;
}>;
export type WebSocketUnsubscribeFrame = Readonly<{
type: "UNSUBSCRIBE";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
subscriptionId: string;
}>;
export type WebSocketHeartbeatFrame = Readonly<{
type: "HEARTBEAT";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
nonce: string;
}>;
export type WebSocketClientCloseFrame = Readonly<{
type: "CLOSE";
protocol: typeof REALTIME_WEBSOCKET_PROTOCOL;
category: WebSocketCloseCategory;
}>;
export type WebSocketClientFrame =
| WebSocketSubscribeFrame
| WebSocketUnsubscribeFrame
| WebSocketHeartbeatFrame
| WebSocketClientCloseFrame;
export type WebSocketProtocolFailure = Readonly<{
code:
| "BINARY_FRAME"
| "FRAME_TOO_LARGE"
| "MALFORMED_FRAME"
| "PROTOCOL_MISMATCH"
| "UNKNOWN_FRAME";
}>;
export type WebSocketProtocolResult<Value> =
| Readonly<{ ok: true; value: Value; byteLength: number }>
| Readonly<{ ok: false; error: WebSocketProtocolFailure }>;
const IDENTIFIER = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u;
const OPAQUE_VALUE = /^[A-Za-z0-9][A-Za-z0-9._~:+/=-]{0,511}$/u;
const UNSIGNED_DECIMAL = /^(?:0|[1-9][0-9]{0,19})$/u;
const UINT64_MAX = 18_446_744_073_709_551_615n;
const MAX_FRAME_STRUCTURE_DEPTH = 32;
const MAX_FRAME_STRUCTURE_NODES = 4_096;
const CLOSE_CATEGORIES: readonly WebSocketCloseCategory[] = [
"NORMAL",
"RESTART",
"OVERLOADED",
"AUTH_REQUIRED",
"FORBIDDEN",
"PROTOCOL_MISMATCH",
"CURSOR_RESET",
"NETWORK_LOST",
];
const RESET_REASONS: readonly WebSocketResetReason[] = [
"CURSOR_EXPIRED",
"SEQUENCE_GAP",
"SERVER_RESET",
"SCOPE_CHANGED",
];
const LIMIT_KEYS = [
"maxBufferedAmountBytes",
"maxEventsPerSecond",
"maxFrameBytes",
"maxInboundQueueBytes",
"maxInboundQueueCount",
"maxOutboundQueueBytes",
"maxOutboundQueueCount",
"maxSubscriptions",
] as const;
const SERVER_KEYS = Object.freeze({
WELCOME: [
"connectionId",
"heartbeatAckTimeoutMs",
"heartbeatMs",
"limits",
"protocol",
"type",
],
SUBSCRIBED: [
"acceptedCursor",
"nextExpectedSequence",
"protocol",
"streamEpoch",
"subscriptionId",
"type",
],
UNSUBSCRIBED: ["protocol", "subscriptionId", "type"],
EVENT: ["envelope", "protocol", "subscriptionId", "type"],
RESET_REQUIRED: [
"protocol",
"reason",
"subscriptionId",
"type",
],
HEARTBEAT_ACK: ["nonce", "protocol", "type"],
CLOSE: ["category", "protocol", "type"],
} satisfies Record<string, readonly string[]>);
const CLIENT_KEYS = Object.freeze({
SUBSCRIBE: [
"cursor",
"protocol",
"scopeBinding",
"streamId",
"subscriptionId",
"type",
],
UNSUBSCRIBE: ["protocol", "subscriptionId", "type"],
HEARTBEAT: ["nonce", "protocol", "type"],
CLOSE: ["category", "protocol", "type"],
} satisfies Record<string, readonly string[]>);
export function decodeWebSocketServerFrame(
input: unknown,
maxFrameBytes: number,
): WebSocketProtocolResult<WebSocketServerFrame> {
if (typeof input !== "string") {
return protocolFailure("BINARY_FRAME");
}
if (!isPositiveInteger(maxFrameBytes)) {
return protocolFailure("FRAME_TOO_LARGE");
}
const byteLength = utf8ByteLength(input);
if (byteLength > maxFrameBytes) {
return protocolFailure("FRAME_TOO_LARGE");
}
if (
hasDuplicateJsonMembers(input, {
maxDepth: MAX_FRAME_STRUCTURE_DEPTH,
maxMembers: MAX_FRAME_STRUCTURE_NODES,
})
) {
return protocolFailure("MALFORMED_FRAME");
}
let parsed: unknown;
try {
parsed = JSON.parse(input);
} catch {
return protocolFailure("MALFORMED_FRAME");
}
if (!isRecord(parsed) || typeof parsed.type !== "string") {
return protocolFailure("MALFORMED_FRAME");
}
if (parsed.protocol !== REALTIME_WEBSOCKET_PROTOCOL) {
return protocolFailure("PROTOCOL_MISMATCH");
}
const frame = decodeKnownServerFrame(parsed);
if (!frame) {
return protocolFailure(
Object.hasOwn(SERVER_KEYS, parsed.type)
? "MALFORMED_FRAME"
: "UNKNOWN_FRAME",
);
}
try {
if (!freezeBoundedJsonTree(frame)) {
return protocolFailure("MALFORMED_FRAME");
}
return Object.freeze({
ok: true,
value: frame,
byteLength,
});
} catch {
return protocolFailure("MALFORMED_FRAME");
}
}
export function encodeWebSocketClientFrame(
frame: WebSocketClientFrame,
maxFrameBytes: number,
): WebSocketProtocolResult<string> {
if (
!isPositiveInteger(maxFrameBytes) ||
!isRecord(frame) ||
frame.protocol !== REALTIME_WEBSOCKET_PROTOCOL ||
typeof frame.type !== "string"
) {
return protocolFailure("MALFORMED_FRAME");
}
const keys = CLIENT_KEYS[frame.type as keyof typeof CLIENT_KEYS];
if (!keys || !hasExactKeys(frame, keys) || !isValidClientFrame(frame)) {
return protocolFailure(
keys ? "MALFORMED_FRAME" : "UNKNOWN_FRAME",
);
}
let value: string;
try {
value = JSON.stringify(frame);
} catch {
return protocolFailure("MALFORMED_FRAME");
}
const byteLength = utf8ByteLength(value);
if (byteLength > maxFrameBytes) {
return protocolFailure("FRAME_TOO_LARGE");
}
return Object.freeze({ ok: true, value, byteLength });
}
export function nextUnsignedSequence(
sequence: string,
): string | null {
if (!isUnsignedSequence(sequence)) return null;
const value = BigInt(sequence);
return value === UINT64_MAX ? null : String(value + 1n);
}
function decodeKnownServerFrame(
frame: Record<string, unknown>,
): WebSocketServerFrame | null {
switch (frame.type) {
case "WELCOME":
if (
!hasExactKeys(frame, SERVER_KEYS.WELCOME) ||
!isIdentifier(frame.connectionId) ||
!isPositiveInteger(frame.heartbeatMs) ||
!isPositiveInteger(frame.heartbeatAckTimeoutMs) ||
!isAdvertisedLimits(frame.limits)
) {
return null;
}
return frame as WebSocketWelcomeFrame;
case "SUBSCRIBED":
if (
!hasExactKeys(frame, SERVER_KEYS.SUBSCRIBED) ||
!isIdentifier(frame.subscriptionId) ||
!isIdentifier(frame.streamEpoch) ||
!isOptionalOpaque(frame.acceptedCursor) ||
!isUnsignedSequence(frame.nextExpectedSequence)
) {
return null;
}
return frame as WebSocketSubscribedFrame;
case "UNSUBSCRIBED":
if (
!hasExactKeys(frame, SERVER_KEYS.UNSUBSCRIBED) ||
!isIdentifier(frame.subscriptionId)
) {
return null;
}
return frame as WebSocketUnsubscribedFrame;
case "EVENT":
if (
!hasExactKeys(frame, SERVER_KEYS.EVENT) ||
!isIdentifier(frame.subscriptionId) ||
!isRecord(frame.envelope)
) {
return null;
}
return frame as WebSocketEventFrame;
case "RESET_REQUIRED":
if (
!hasExactKeys(frame, SERVER_KEYS.RESET_REQUIRED) ||
!isIdentifier(frame.subscriptionId) ||
!RESET_REASONS.includes(frame.reason as WebSocketResetReason)
) {
return null;
}
return frame as WebSocketResetRequiredFrame;
case "HEARTBEAT_ACK":
if (
!hasExactKeys(frame, SERVER_KEYS.HEARTBEAT_ACK) ||
!isIdentifier(frame.nonce)
) {
return null;
}
return frame as WebSocketHeartbeatAckFrame;
case "CLOSE":
if (
!hasExactKeys(frame, SERVER_KEYS.CLOSE) ||
!CLOSE_CATEGORIES.includes(
frame.category as WebSocketCloseCategory,
)
) {
return null;
}
return frame as WebSocketServerCloseFrame;
default:
return null;
}
}
function isValidClientFrame(
frame: Record<string, unknown>,
): boolean {
switch (frame.type) {
case "SUBSCRIBE":
return (
isIdentifier(frame.subscriptionId) &&
isIdentifier(frame.streamId) &&
isOptionalOpaque(frame.cursor) &&
isOpaque(frame.scopeBinding)
);
case "UNSUBSCRIBE":
return isIdentifier(frame.subscriptionId);
case "HEARTBEAT":
return isIdentifier(frame.nonce);
case "CLOSE":
return CLOSE_CATEGORIES.includes(
frame.category as WebSocketCloseCategory,
);
default:
return false;
}
}
function isAdvertisedLimits(
input: unknown,
): input is WebSocketAdvertisedLimits {
if (!isRecord(input) || !hasExactKeys(input, LIMIT_KEYS)) {
return false;
}
return LIMIT_KEYS.every((key) => isPositiveInteger(input[key]));
}
function isRecord(
input: unknown,
): input is Record<string, unknown> {
return (
typeof input === "object" &&
input !== null &&
!Array.isArray(input) &&
Object.getPrototypeOf(input) === Object.prototype
);
}
function hasExactKeys(
input: Record<string, unknown>,
expected: readonly string[],
): boolean {
const keys = Object.keys(input).sort();
return (
keys.length === expected.length &&
keys.every((key, index) => key === expected[index])
);
}
function isIdentifier(input: unknown): input is string {
return typeof input === "string" && IDENTIFIER.test(input);
}
function isOpaque(input: unknown): input is string {
return typeof input === "string" && OPAQUE_VALUE.test(input);
}
function isOptionalOpaque(input: unknown): input is string | null {
return input === null || isOpaque(input);
}
function isPositiveInteger(input: unknown): input is number {
return Number.isSafeInteger(input) && Number(input) > 0;
}
function isUnsignedSequence(input: unknown): input is string {
if (typeof input !== "string" || !UNSIGNED_DECIMAL.test(input)) {
return false;
}
try {
return BigInt(input) <= UINT64_MAX;
} catch {
return false;
}
}
function utf8ByteLength(input: string): number {
return new TextEncoder().encode(input).byteLength;
}
function protocolFailure(
code: WebSocketProtocolFailure["code"],
): WebSocketProtocolResult<never> {
return Object.freeze({
ok: false,
error: Object.freeze({ code }),
});
}
function freezeBoundedJsonTree(root: object): boolean {
const pending: Array<
Readonly<{
value: object;
depth: number;
freeze: boolean;
}>
> = [{ value: root, depth: 0, freeze: false }];
let discoveredNodes = 1;
while (pending.length > 0) {
const current = pending.pop();
if (!current) return false;
if (current.freeze) {
Object.freeze(current.value);
continue;
}
if (current.depth > MAX_FRAME_STRUCTURE_DEPTH) {
return false;
}
pending.push({ ...current, freeze: true });
for (const child of Object.values(current.value)) {
if (child !== null && typeof child === "object") {
discoveredNodes += 1;
if (discoveredNodes > MAX_FRAME_STRUCTURE_NODES) {
return false;
}
pending.push({
value: child,
depth: current.depth + 1,
freeze: false,
});
}
}
}
return true;
}