267 lines
9.4 KiB
TypeScript
267 lines
9.4 KiB
TypeScript
import type { EventEnvelope, EventKind } from '@multi-simadmin/contracts';
|
|
|
|
export type EventStreamState = 'connecting' | 'open' | 'reconnecting' | 'resetting' | 'closed';
|
|
export type EventStreamStateListener = (state: EventStreamState, error?: Error) => void;
|
|
export type EventStreamListener = (event: EventEnvelope) => void;
|
|
|
|
export interface SseRecord {
|
|
readonly id?: string;
|
|
readonly data: string;
|
|
}
|
|
|
|
export interface SseParser {
|
|
push(chunk: string): void;
|
|
finish(): void;
|
|
}
|
|
|
|
export interface EventStreamClient {
|
|
subscribe(onEvent: EventStreamListener, onState?: EventStreamStateListener): () => void;
|
|
}
|
|
|
|
export interface EventStreamClientOptions {
|
|
readonly url?: string;
|
|
readonly fetch?: typeof globalThis.fetch;
|
|
readonly initialRetryMs?: number;
|
|
readonly maxRetryMs?: number;
|
|
readonly sleep?: (milliseconds: number, signal: AbortSignal) => Promise<void>;
|
|
readonly onPositionReset?: () => void;
|
|
}
|
|
|
|
const KINDS: readonly EventKind[] = ['instance', 'job', 'item', 'attempt', 'audit'];
|
|
const COMMON_KEYS = ['kind', 'id', 'occurredAt', 'requestId'] as const;
|
|
const KIND_KEYS: Readonly<Record<EventKind, readonly string[]>> = {
|
|
instance: ['instanceId'],
|
|
job: ['jobId'],
|
|
item: ['jobId', 'itemId'],
|
|
attempt: ['jobId', 'attemptId'],
|
|
audit: ['auditEventId'],
|
|
};
|
|
const CONNECTION_ERROR = 'The event stream connection failed.';
|
|
const INVALID_EVENT_ERROR = 'Invalid event envelope.';
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return value !== null && typeof value === 'object' && !Array.isArray(value);
|
|
}
|
|
|
|
function isNonEmptyString(value: unknown): value is string {
|
|
return typeof value === 'string' && value.length > 0;
|
|
}
|
|
|
|
function isIsoTimestamp(value: unknown): value is string {
|
|
if (!isNonEmptyString(value)) return false;
|
|
const match =
|
|
/^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d+)?(Z|[+-]\d{2}:\d{2})$/u.exec(value);
|
|
if (!match) return false;
|
|
const [, year, month, day, hour, minute, second] = match;
|
|
const parts = [year, month, day, hour, minute, second].map(Number);
|
|
if (parts.some((part) => !Number.isInteger(part))) return false;
|
|
const [y, m, d, h, min, s] = parts as [number, number, number, number, number, number];
|
|
if (m < 1 || m > 12 || d < 1 || h > 23 || min > 59 || s > 59) return false;
|
|
const daysInMonth = new Date(Date.UTC(y, m, 0)).getUTCDate();
|
|
if (d > daysInMonth) return false;
|
|
return Number.isFinite(Date.parse(value));
|
|
}
|
|
|
|
/** Validates the exact, frozen EventEnvelope union without retaining unknown fields. */
|
|
export function parseEventEnvelope(value: unknown): EventEnvelope {
|
|
if (
|
|
!isRecord(value) ||
|
|
typeof value.kind !== 'string' ||
|
|
!KINDS.includes(value.kind as EventKind)
|
|
)
|
|
throw new Error(INVALID_EVENT_ERROR);
|
|
|
|
const kind = value.kind as EventKind;
|
|
const kindKeys = KIND_KEYS[kind];
|
|
if (kindKeys === undefined) throw new Error(INVALID_EVENT_ERROR);
|
|
const expected = [...COMMON_KEYS, ...kindKeys];
|
|
const actual = Object.keys(value);
|
|
if (actual.length !== expected.length || actual.some((key) => !expected.includes(key)))
|
|
throw new Error(INVALID_EVENT_ERROR);
|
|
if (
|
|
!isNonEmptyString(value.id) ||
|
|
!isIsoTimestamp(value.occurredAt) ||
|
|
!isNonEmptyString(value.requestId) ||
|
|
kindKeys.some((key) => !isNonEmptyString(value[key]))
|
|
)
|
|
throw new Error(INVALID_EVENT_ERROR);
|
|
|
|
return value as unknown as EventEnvelope;
|
|
}
|
|
|
|
/** Incremental SSE field parser implementing LF, CRLF, CR, comments, and multiline data. */
|
|
export function createSseParser(onRecord: (record: SseRecord) => void): SseParser {
|
|
let buffer = '';
|
|
let data: string[] = [];
|
|
let id: string | undefined;
|
|
|
|
const dispatch = (): void => {
|
|
if (data.length > 0)
|
|
onRecord(id === undefined ? { data: data.join('\n') } : { id, data: data.join('\n') });
|
|
data = [];
|
|
id = undefined;
|
|
};
|
|
const line = (value: string): void => {
|
|
if (value === '') {
|
|
dispatch();
|
|
return;
|
|
}
|
|
if (value.startsWith(':')) return;
|
|
const colon = value.indexOf(':');
|
|
const field = colon < 0 ? value : value.slice(0, colon);
|
|
let fieldValue = colon < 0 ? '' : value.slice(colon + 1);
|
|
if (fieldValue.startsWith(' ')) fieldValue = fieldValue.slice(1);
|
|
if (field === 'data') data.push(fieldValue);
|
|
else if (field === 'id' && !fieldValue.includes('\0')) id = fieldValue;
|
|
};
|
|
const consume = (final: boolean): void => {
|
|
let start = 0;
|
|
for (let index = 0; index < buffer.length; index += 1) {
|
|
const character = buffer[index];
|
|
if (character !== '\r' && character !== '\n') continue;
|
|
if (character === '\r' && index + 1 === buffer.length && !final) break;
|
|
line(buffer.slice(start, index));
|
|
if (character === '\r' && buffer[index + 1] === '\n') index += 1;
|
|
start = index + 1;
|
|
}
|
|
buffer = buffer.slice(start);
|
|
if (final && buffer.length > 0) {
|
|
line(buffer);
|
|
buffer = '';
|
|
}
|
|
};
|
|
return {
|
|
push(chunk) {
|
|
buffer += chunk;
|
|
consume(false);
|
|
},
|
|
finish() {
|
|
consume(true);
|
|
dispatch();
|
|
},
|
|
};
|
|
}
|
|
|
|
function sameOriginUrl(input: string): string {
|
|
const base = globalThis.location?.href ?? 'http://localhost/';
|
|
const url = new URL(input, base);
|
|
const origin = new URL(base).origin;
|
|
if (url.origin !== origin || url.username !== '' || url.password !== '')
|
|
throw new Error('The event stream URL must be same-origin.');
|
|
return input.startsWith('/') ? `${url.pathname}${url.search}` : url.href;
|
|
}
|
|
|
|
function defaultSleep(milliseconds: number, signal: AbortSignal): Promise<void> {
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(resolve, milliseconds);
|
|
signal.addEventListener(
|
|
'abort',
|
|
() => {
|
|
clearTimeout(timer);
|
|
reject(signal.reason);
|
|
},
|
|
{ once: true },
|
|
);
|
|
});
|
|
}
|
|
|
|
function safeConnectionError(): Error {
|
|
return new Error(CONNECTION_ERROR);
|
|
}
|
|
|
|
export function createEventStreamClient(options: EventStreamClientOptions = {}): EventStreamClient {
|
|
const fetcher = options.fetch ?? globalThis.fetch;
|
|
const initialRetryMs = Math.max(1, options.initialRetryMs ?? 1_000);
|
|
const maxRetryMs = Math.max(initialRetryMs, options.maxRetryMs ?? 30_000);
|
|
const sleep = options.sleep ?? defaultSleep;
|
|
|
|
return {
|
|
subscribe(onEvent, onState = () => undefined) {
|
|
const controller = new AbortController();
|
|
let cursor: string | undefined;
|
|
let retryMs = initialRetryMs;
|
|
|
|
const run = async (): Promise<void> => {
|
|
let url: string;
|
|
try {
|
|
url = sameOriginUrl(options.url ?? '/api/v1/events');
|
|
} catch (error) {
|
|
onState('closed', error instanceof Error ? error : safeConnectionError());
|
|
return;
|
|
}
|
|
|
|
onState('connecting');
|
|
while (!controller.signal.aborted) {
|
|
try {
|
|
const headers: Record<string, string> = { Accept: 'text/event-stream' };
|
|
if (cursor !== undefined) headers['Last-Event-ID'] = cursor;
|
|
const response = await fetcher(url, {
|
|
credentials: 'same-origin',
|
|
headers,
|
|
signal: controller.signal,
|
|
});
|
|
if (response.status === 409 && cursor !== undefined) {
|
|
cursor = undefined;
|
|
options.onPositionReset?.();
|
|
onState(
|
|
'resetting',
|
|
new Error('The event stream position expired; refreshing current state.'),
|
|
);
|
|
continue;
|
|
}
|
|
if (!response.ok || response.body === null) throw safeConnectionError();
|
|
const contentType = response.headers.get('content-type')?.toLowerCase() ?? '';
|
|
if (!contentType.startsWith('text/event-stream')) throw safeConnectionError();
|
|
|
|
onState('open');
|
|
retryMs = initialRetryMs;
|
|
const decoder = new TextDecoder();
|
|
let recordFailure: Error | undefined;
|
|
const parser = createSseParser((record) => {
|
|
if (recordFailure !== undefined) return;
|
|
try {
|
|
const event = parseEventEnvelope(JSON.parse(record.data) as unknown);
|
|
onEvent(event);
|
|
if (record.id !== undefined) cursor = record.id;
|
|
} catch {
|
|
recordFailure = new Error(INVALID_EVENT_ERROR);
|
|
}
|
|
});
|
|
const reader = response.body.getReader();
|
|
try {
|
|
while (!controller.signal.aborted) {
|
|
const result = await reader.read();
|
|
if (result.done) break;
|
|
parser.push(decoder.decode(result.value, { stream: true }));
|
|
if (recordFailure !== undefined) throw recordFailure;
|
|
}
|
|
parser.push(decoder.decode());
|
|
parser.finish();
|
|
if (recordFailure !== undefined) throw recordFailure;
|
|
} finally {
|
|
if (controller.signal.aborted) await reader.cancel().catch(() => undefined);
|
|
reader.releaseLock();
|
|
}
|
|
} catch {
|
|
if (controller.signal.aborted) break;
|
|
}
|
|
if (controller.signal.aborted) break;
|
|
// Deliberately replace fetch/parser errors: response bodies and transport details may be sensitive.
|
|
onState('reconnecting', safeConnectionError());
|
|
const delay = retryMs;
|
|
retryMs = Math.min(maxRetryMs, retryMs * 2);
|
|
try {
|
|
await sleep(delay, controller.signal);
|
|
} catch {
|
|
break;
|
|
}
|
|
}
|
|
onState('closed');
|
|
};
|
|
|
|
void run();
|
|
return () => controller.abort();
|
|
},
|
|
};
|
|
}
|