feat(web): add durable control-plane event updates
This commit is contained in:
chick
@@ -0,0 +1,266 @@
|
||||
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();
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user