feat(api): add fleet snapshots and durable event stream
This commit is contained in:
chick
@@ -0,0 +1,164 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { EventJournal, formatServerSentEvent, type EventEnvelope } from './event-journal.js';
|
||||
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
|
||||
|
||||
const at = '2026-07-17T12:34:56.789Z';
|
||||
|
||||
function event(kind: EventEnvelope['kind'], id = `${kind}-1`): EventEnvelope {
|
||||
const common = {
|
||||
kind,
|
||||
id,
|
||||
occurredAt: at,
|
||||
requestId: 'request-1',
|
||||
};
|
||||
switch (kind) {
|
||||
case 'instance':
|
||||
return { ...common, kind, instanceId: 'instance-1' };
|
||||
case 'job':
|
||||
return { ...common, kind, jobId: 'job-1' };
|
||||
case 'item':
|
||||
return { ...common, kind, jobId: 'job-1', itemId: 'item-1' };
|
||||
case 'attempt':
|
||||
return { ...common, kind, jobId: 'job-1', attemptId: 'attempt-1' };
|
||||
case 'audit':
|
||||
return { ...common, kind, auditEventId: 'audit-event-1' };
|
||||
}
|
||||
}
|
||||
|
||||
function journal(database = new Database(':memory:')): EventJournal {
|
||||
migrateDatabase(database);
|
||||
return new EventJournal(database);
|
||||
}
|
||||
|
||||
describe('EventJournal', () => {
|
||||
it('requires its forward-migrated schema instead of creating it implicitly', () => {
|
||||
const database = new Database(':memory:');
|
||||
|
||||
expect(() => new EventJournal(database)).toThrow(
|
||||
/event_journal.*schema|schema.*event_journal/i,
|
||||
);
|
||||
expect(
|
||||
database
|
||||
.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='event_journal'")
|
||||
.get(),
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it('durably appends the exact five envelope variants with stable integer sequence', () => {
|
||||
const database = new Database(':memory:');
|
||||
migrateDatabase(database);
|
||||
const first = new EventJournal(database);
|
||||
const envelopes = (['instance', 'job', 'item', 'attempt', 'audit'] as const).map((kind) =>
|
||||
event(kind),
|
||||
);
|
||||
|
||||
expect(envelopes.map((envelope) => first.append(envelope).sequence)).toEqual([1, 2, 3, 4, 5]);
|
||||
const reopened = new EventJournal(database);
|
||||
expect(reopened.readAfter()).toEqual({
|
||||
status: 'current',
|
||||
events: [],
|
||||
latestEventId: 'audit-1',
|
||||
});
|
||||
expect(reopened.readAfter('instance-1')).toEqual({
|
||||
status: 'found',
|
||||
events: envelopes.slice(1).map((envelope, index) => ({ sequence: index + 2, envelope })),
|
||||
latestEventId: 'audit-1',
|
||||
});
|
||||
});
|
||||
|
||||
it('enforces unique public IDs without consuming a sequence on failure', () => {
|
||||
const subject = journal();
|
||||
subject.append(event('job', 'same'));
|
||||
|
||||
expect(() => subject.append(event('audit', 'same'))).toThrow(/unique|already exists/i);
|
||||
expect(subject.append(event('audit', 'next')).sequence).toBe(2);
|
||||
});
|
||||
|
||||
it('reports current, found (strictly after), and unavailable cursors', () => {
|
||||
const subject = journal();
|
||||
subject.append(event('instance', 'one'));
|
||||
subject.append(event('job', 'two'));
|
||||
|
||||
expect(subject.readAfter()).toEqual({ status: 'current', events: [], latestEventId: 'two' });
|
||||
expect(subject.readAfter('two')).toEqual({ status: 'found', events: [], latestEventId: 'two' });
|
||||
expect(subject.readAfter('one')).toEqual({
|
||||
status: 'found',
|
||||
events: [{ sequence: 2, envelope: event('job', 'two') }],
|
||||
latestEventId: 'two',
|
||||
});
|
||||
expect(subject.readAfter('missing')).toEqual({
|
||||
status: 'unavailable',
|
||||
events: [],
|
||||
latestEventId: 'two',
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
[{ ...event('job'), kind: 'other' }, /kind/i],
|
||||
[{ ...event('job'), type: 'job' }, /additional|field/i],
|
||||
[{ ...event('job'), extra: true }, /additional|field/i],
|
||||
[{ ...event('job'), occurredAt: 'Friday' }, /occurredAt|date-time/i],
|
||||
[{ ...event('job'), id: 'bad\nvalue' }, /id|control/i],
|
||||
[{ ...event('job'), requestId: 'bad\rvalue' }, /requestId|control/i],
|
||||
[{ ...event('job'), jobId: 'bad\0value' }, /jobId|control/i],
|
||||
[{ kind: 'job', id: 'job-1', occurredAt: at, requestId: 'request-1' }, /jobId|missing/i],
|
||||
[{ ...event('item'), attemptId: 'attempt-1' }, /additional|field/i],
|
||||
[{ ...event('job'), id: '' }, /id/i],
|
||||
])('rejects invalid envelopes before persistence: %j', (invalid, message) => {
|
||||
const subject = journal();
|
||||
expect(() => subject.append(invalid as EventEnvelope)).toThrow(message);
|
||||
expect(subject.readAfter()).toEqual({
|
||||
status: 'current',
|
||||
events: [],
|
||||
latestEventId: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it('validates persisted envelope_json rather than trusting database contents', () => {
|
||||
const database = new Database(':memory:');
|
||||
migrateDatabase(database);
|
||||
database
|
||||
.prepare('INSERT INTO event_journal (public_id, envelope_json) VALUES (?, ?)')
|
||||
.run('corrupt', '{"kind":"job"}');
|
||||
|
||||
expect(() => new EventJournal(database).readAfter('corrupt')).toThrow(
|
||||
/persisted.*envelope|invalid/i,
|
||||
);
|
||||
});
|
||||
|
||||
it('notifies subscribers only after durable append and supports idempotent unsubscribe', () => {
|
||||
const subject = journal();
|
||||
const listener = vi.fn();
|
||||
const unsubscribe = subject.subscribe(listener);
|
||||
const appended = subject.append(event('attempt'));
|
||||
|
||||
expect(listener).toHaveBeenCalledOnce();
|
||||
expect(listener).toHaveBeenCalledWith(appended);
|
||||
unsubscribe();
|
||||
unsubscribe();
|
||||
subject.append(event('audit'));
|
||||
expect(listener).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
|
||||
describe('formatServerSentEvent', () => {
|
||||
it('formats exactly id, compact JSON data, and a blank line without an event field', () => {
|
||||
const envelope = event('item', 'public-7');
|
||||
|
||||
expect(formatServerSentEvent(envelope)).toBe(
|
||||
`id: public-7\ndata: ${JSON.stringify(envelope)}\n\n`,
|
||||
);
|
||||
expect(formatServerSentEvent(envelope)).not.toContain('\nevent:');
|
||||
});
|
||||
|
||||
it('applies the same strict envelope validation at the wire boundary', () => {
|
||||
expect(() =>
|
||||
formatServerSentEvent({ ...event('audit'), requestId: 'inject\ndata: x' }),
|
||||
).toThrow(/requestId|control/i);
|
||||
expect(() =>
|
||||
formatServerSentEvent({ ...event('audit'), unexpected: 1 } as unknown as EventEnvelope),
|
||||
).toThrow(/additional|field/i);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,264 @@
|
||||
import type Database from 'better-sqlite3';
|
||||
|
||||
const EVENT_KINDS = ['instance', 'job', 'item', 'attempt', 'audit'] as const;
|
||||
const COMMON_EVENT_FIELDS = ['kind', 'id', 'occurredAt', 'requestId'] as const;
|
||||
const CORRELATION_FIELDS = ['instanceId', 'jobId', 'itemId', 'attemptId', 'auditEventId'] as const;
|
||||
const FORBIDDEN_ID_CHARACTERS = /[\r\n\0]/u;
|
||||
const DATE_TIME =
|
||||
/^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.(\d+))?(Z|[+-]\d{2}:\d{2})$/u;
|
||||
|
||||
export type EventKind = (typeof EVENT_KINDS)[number];
|
||||
|
||||
interface EventEnvelopeFields {
|
||||
readonly id: string;
|
||||
readonly occurredAt: string;
|
||||
readonly requestId: string;
|
||||
}
|
||||
|
||||
export type EventEnvelope =
|
||||
| (EventEnvelopeFields & { readonly kind: 'instance'; readonly instanceId: string })
|
||||
| (EventEnvelopeFields & { readonly kind: 'job'; readonly jobId: string })
|
||||
| (EventEnvelopeFields & {
|
||||
readonly kind: 'item';
|
||||
readonly jobId: string;
|
||||
readonly itemId: string;
|
||||
})
|
||||
| (EventEnvelopeFields & {
|
||||
readonly kind: 'attempt';
|
||||
readonly jobId: string;
|
||||
readonly attemptId: string;
|
||||
})
|
||||
| (EventEnvelopeFields & { readonly kind: 'audit'; readonly auditEventId: string });
|
||||
|
||||
export interface JournalEvent {
|
||||
readonly sequence: number;
|
||||
readonly envelope: EventEnvelope;
|
||||
}
|
||||
|
||||
export type EventJournalRead =
|
||||
| {
|
||||
readonly status: 'current';
|
||||
readonly events: readonly [];
|
||||
readonly latestEventId: string | undefined;
|
||||
}
|
||||
| {
|
||||
readonly status: 'found';
|
||||
readonly events: readonly JournalEvent[];
|
||||
readonly latestEventId: string | undefined;
|
||||
}
|
||||
| {
|
||||
readonly status: 'unavailable';
|
||||
readonly events: readonly [];
|
||||
readonly latestEventId: string | undefined;
|
||||
};
|
||||
|
||||
export type EventJournalSubscriber = (event: JournalEvent) => void;
|
||||
|
||||
interface JournalRow {
|
||||
sequence: number;
|
||||
public_id: string;
|
||||
envelope_json: string;
|
||||
}
|
||||
|
||||
function validateIdentifier(value: unknown, field: string): asserts value is string {
|
||||
if (typeof value !== 'string' || value.length === 0) {
|
||||
throw new TypeError(`${field} must be a non-empty string`);
|
||||
}
|
||||
if (FORBIDDEN_ID_CHARACTERS.test(value)) {
|
||||
throw new TypeError(`${field} contains a forbidden CR, LF, or NUL control character`);
|
||||
}
|
||||
}
|
||||
|
||||
function isValidDateTime(value: unknown): value is string {
|
||||
if (typeof value !== 'string') return false;
|
||||
const match = DATE_TIME.exec(value);
|
||||
if (!match) return false;
|
||||
const zone = match[8]!;
|
||||
const year = Number(match[1]);
|
||||
const month = Number(match[2]);
|
||||
const day = Number(match[3]);
|
||||
const hour = Number(match[4]);
|
||||
const minute = Number(match[5]);
|
||||
const second = Number(match[6]);
|
||||
if (month < 1 || month > 12 || hour > 23 || minute > 59 || second > 59) return false;
|
||||
const daysInMonth = new Date(Date.UTC(year, month, 0)).getUTCDate();
|
||||
if (day < 1 || day > daysInMonth) return false;
|
||||
if (zone !== 'Z') {
|
||||
const zoneHour = Number(zone.slice(1, 3));
|
||||
const zoneMinute = Number(zone.slice(4, 6));
|
||||
if (zoneHour > 23 || zoneMinute > 59) return false;
|
||||
}
|
||||
return Number.isFinite(Date.parse(value));
|
||||
}
|
||||
|
||||
export function validateEventEnvelope(value: unknown): asserts value is EventEnvelope {
|
||||
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
|
||||
throw new TypeError('event envelope must be an object');
|
||||
}
|
||||
const record = value as Record<string, unknown>;
|
||||
if (!(EVENT_KINDS as readonly unknown[]).includes(record.kind)) {
|
||||
throw new TypeError('event envelope kind must be instance, job, item, attempt, or audit');
|
||||
}
|
||||
const variantFields: Record<EventKind, readonly string[]> = {
|
||||
instance: ['instanceId'],
|
||||
job: ['jobId'],
|
||||
item: ['jobId', 'itemId'],
|
||||
attempt: ['jobId', 'attemptId'],
|
||||
audit: ['auditEventId'],
|
||||
};
|
||||
const expected: readonly string[] = [
|
||||
...COMMON_EVENT_FIELDS,
|
||||
...variantFields[record.kind as EventKind],
|
||||
];
|
||||
const keys = Object.keys(record);
|
||||
const additional = keys.filter((key) => !expected.includes(key));
|
||||
const missing = expected.filter((key) => !Object.prototype.hasOwnProperty.call(record, key));
|
||||
if (additional.length > 0) {
|
||||
throw new TypeError(`event envelope has additional field: ${additional.join(', ')}`);
|
||||
}
|
||||
if (missing.length > 0) {
|
||||
throw new TypeError(`event envelope is missing field: ${missing.join(', ')}`);
|
||||
}
|
||||
validateIdentifier(record.id, 'id');
|
||||
validateIdentifier(record.requestId, 'requestId');
|
||||
for (const field of CORRELATION_FIELDS) {
|
||||
if (Object.prototype.hasOwnProperty.call(record, field))
|
||||
validateIdentifier(record[field], field);
|
||||
}
|
||||
if (!isValidDateTime(record.occurredAt)) {
|
||||
throw new TypeError('occurredAt must be a valid RFC 3339 date-time');
|
||||
}
|
||||
}
|
||||
|
||||
export function formatServerSentEvent(envelope: EventEnvelope): string {
|
||||
validateEventEnvelope(envelope);
|
||||
return `id: ${envelope.id}\ndata: ${JSON.stringify(envelope)}\n\n`;
|
||||
}
|
||||
|
||||
export interface EventJournalOptions {
|
||||
readonly retentionMs?: number;
|
||||
readonly maxReplayEvents?: number;
|
||||
}
|
||||
|
||||
export class EventJournal {
|
||||
readonly #database: Database.Database;
|
||||
readonly #subscribers = new Set<EventJournalSubscriber>();
|
||||
readonly #retentionMs: number;
|
||||
readonly #maxReplayEvents: number;
|
||||
|
||||
constructor(database: Database.Database, options: EventJournalOptions = {}) {
|
||||
if (
|
||||
options.retentionMs !== undefined &&
|
||||
(!Number.isSafeInteger(options.retentionMs) || options.retentionMs <= 0)
|
||||
) {
|
||||
throw new RangeError('retentionMs must be a positive safe integer');
|
||||
}
|
||||
if (
|
||||
options.maxReplayEvents !== undefined &&
|
||||
(!Number.isSafeInteger(options.maxReplayEvents) || options.maxReplayEvents <= 0)
|
||||
) {
|
||||
throw new RangeError('maxReplayEvents must be a positive safe integer');
|
||||
}
|
||||
this.#database = database;
|
||||
this.#retentionMs = options.retentionMs ?? 86_400_000;
|
||||
this.#maxReplayEvents = options.maxReplayEvents ?? 1_000;
|
||||
const columns = database.pragma('table_info(event_journal)') as Array<{ name: string }>;
|
||||
const names = new Set(columns.map(({ name }) => name));
|
||||
if (
|
||||
columns.length === 0 ||
|
||||
!names.has('sequence') ||
|
||||
!names.has('public_id') ||
|
||||
!names.has('envelope_json')
|
||||
) {
|
||||
throw new Error(
|
||||
'event_journal schema is missing or incompatible; apply its forward migration',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
append(envelope: EventEnvelope): JournalEvent {
|
||||
validateEventEnvelope(envelope);
|
||||
const serialized = JSON.stringify(envelope);
|
||||
const result = this.#database
|
||||
.prepare('INSERT INTO event_journal (public_id, envelope_json) VALUES (?, ?)')
|
||||
.run(envelope.id, serialized);
|
||||
const sequence = Number(result.lastInsertRowid);
|
||||
if (!Number.isSafeInteger(sequence))
|
||||
throw new Error('event journal sequence is not a safe integer');
|
||||
this.#prune();
|
||||
const appended = { sequence, envelope };
|
||||
for (const subscriber of [...this.#subscribers]) subscriber(appended);
|
||||
return appended;
|
||||
}
|
||||
|
||||
readAfter(lastEventId?: string): EventJournalRead {
|
||||
const latest = this.#database
|
||||
.prepare('SELECT public_id FROM event_journal ORDER BY sequence DESC LIMIT 1')
|
||||
.get() as { public_id: string } | undefined;
|
||||
const latestEventId = latest?.public_id;
|
||||
if (lastEventId === undefined) return { status: 'current', events: [], latestEventId };
|
||||
validateIdentifier(lastEventId, 'lastEventId');
|
||||
|
||||
const cursor = this.#database
|
||||
.prepare('SELECT sequence, public_id, envelope_json FROM event_journal WHERE public_id = ?')
|
||||
.get(lastEventId) as JournalRow | undefined;
|
||||
if (!cursor) return { status: 'unavailable', events: [], latestEventId };
|
||||
this.#decodeRow(cursor);
|
||||
|
||||
const oldest = this.#database
|
||||
.prepare('SELECT sequence FROM event_journal ORDER BY sequence ASC LIMIT 1')
|
||||
.get() as { sequence: number } | undefined;
|
||||
if (oldest && cursor.sequence < oldest.sequence) {
|
||||
return { status: 'unavailable', events: [], latestEventId };
|
||||
}
|
||||
const rows = this.#database
|
||||
.prepare(
|
||||
'SELECT sequence, public_id, envelope_json FROM event_journal WHERE sequence > ? ORDER BY sequence ASC LIMIT ?',
|
||||
)
|
||||
.all(cursor.sequence, this.#maxReplayEvents + 1) as JournalRow[];
|
||||
if (rows.length > this.#maxReplayEvents) {
|
||||
return { status: 'unavailable', events: [], latestEventId };
|
||||
}
|
||||
return {
|
||||
status: 'found',
|
||||
events: rows.map((row) => this.#decodeRow(row)),
|
||||
latestEventId,
|
||||
};
|
||||
}
|
||||
|
||||
subscribe(subscriber: EventJournalSubscriber): () => void {
|
||||
this.#subscribers.add(subscriber);
|
||||
let subscribed = true;
|
||||
return () => {
|
||||
if (!subscribed) return;
|
||||
subscribed = false;
|
||||
this.#subscribers.delete(subscriber);
|
||||
};
|
||||
}
|
||||
|
||||
#prune(): void {
|
||||
const cutoff = new Date(Date.now() - this.#retentionMs).toISOString();
|
||||
this.#database
|
||||
.prepare("DELETE FROM event_journal WHERE json_extract(envelope_json, '$.occurredAt') < ?")
|
||||
.run(cutoff);
|
||||
}
|
||||
|
||||
#decodeRow(row: JournalRow): JournalEvent {
|
||||
let envelope: unknown;
|
||||
try {
|
||||
envelope = JSON.parse(row.envelope_json);
|
||||
validateEventEnvelope(envelope);
|
||||
if (envelope.id !== row.public_id)
|
||||
throw new TypeError('public ID does not match envelope ID');
|
||||
} catch (error) {
|
||||
const detail = error instanceof Error ? `: ${error.message}` : '';
|
||||
throw new Error(`invalid persisted event envelope at sequence ${row.sequence}${detail}`, {
|
||||
cause: error,
|
||||
});
|
||||
}
|
||||
if (!Number.isSafeInteger(row.sequence) || row.sequence < 1) {
|
||||
throw new Error(`invalid persisted event journal sequence: ${row.sequence}`);
|
||||
}
|
||||
return { sequence: row.sequence, envelope };
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { FleetStatusCoordinator } from './fleet-status-coordinator.js';
|
||||
import { StatusPollScheduler, type StatusPollSchedulerOptions } from './status-poll-scheduler.js';
|
||||
|
||||
const flush = async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
};
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe('FleetStatusCoordinator', () => {
|
||||
it('starts by syncing enabled IDs and delegates scheduled polls to refreshHealth', async () => {
|
||||
const refreshHealth = vi.fn(async (instanceId: string) => {
|
||||
void instanceId;
|
||||
});
|
||||
const coordinator = new FleetStatusCoordinator({
|
||||
source: { listEnabled: vi.fn(async () => ['alpha', 'beta']) },
|
||||
snapshots: { refreshHealth },
|
||||
intervalMs: 100,
|
||||
scheduler: (options) => new StatusPollScheduler({ ...options, concurrency: 1 }),
|
||||
});
|
||||
|
||||
await coordinator.start();
|
||||
await flush();
|
||||
|
||||
expect(refreshHealth.mock.calls.map(([id]) => id)).toEqual(['alpha', 'beta']);
|
||||
await coordinator.stop();
|
||||
});
|
||||
|
||||
it('reconciles additions and removals without re-registering unchanged IDs', async () => {
|
||||
let enabled: readonly string[] = ['alpha', 'beta'];
|
||||
let task!: StatusPollSchedulerOptions['task'];
|
||||
const add = vi.fn();
|
||||
const remove = vi.fn(() => true);
|
||||
const start = vi.fn();
|
||||
const stop = vi.fn(async () => undefined);
|
||||
const scheduler = { add, remove, start, stop };
|
||||
const refreshHealth = vi.fn(async (instanceId: string) => {
|
||||
void instanceId;
|
||||
});
|
||||
const coordinator = new FleetStatusCoordinator({
|
||||
source: { listEnabled: async () => enabled },
|
||||
snapshots: { refreshHealth },
|
||||
intervalMs: 250,
|
||||
scheduler: (options) => {
|
||||
task = options.task;
|
||||
return scheduler;
|
||||
},
|
||||
});
|
||||
|
||||
await coordinator.start();
|
||||
expect(add.mock.calls).toEqual([
|
||||
['alpha', 250],
|
||||
['beta', 250],
|
||||
]);
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
|
||||
enabled = ['beta', 'gamma'];
|
||||
await coordinator.reconcile();
|
||||
expect(remove).toHaveBeenCalledWith('alpha');
|
||||
expect(add).toHaveBeenLastCalledWith('gamma', 250);
|
||||
expect(add).toHaveBeenCalledTimes(3);
|
||||
|
||||
await task('gamma', new AbortController().signal);
|
||||
expect(refreshHealth).toHaveBeenCalledWith('gamma');
|
||||
await coordinator.stop();
|
||||
});
|
||||
|
||||
it('makes concurrent start calls idempotent', async () => {
|
||||
let release!: (ids: readonly string[]) => void;
|
||||
const listEnabled = vi.fn(
|
||||
() =>
|
||||
new Promise<readonly string[]>((resolve) => {
|
||||
release = resolve;
|
||||
}),
|
||||
);
|
||||
const scheduler = {
|
||||
add: vi.fn(),
|
||||
remove: vi.fn(() => true),
|
||||
start: vi.fn(),
|
||||
stop: vi.fn(async () => undefined),
|
||||
};
|
||||
const coordinator = new FleetStatusCoordinator({
|
||||
source: { listEnabled },
|
||||
snapshots: { refreshHealth: vi.fn(async () => undefined) },
|
||||
intervalMs: 50,
|
||||
scheduler: () => scheduler,
|
||||
});
|
||||
|
||||
const first = coordinator.start();
|
||||
const second = coordinator.start();
|
||||
await flush();
|
||||
release(['alpha']);
|
||||
await Promise.all([first, second]);
|
||||
|
||||
expect(listEnabled).toHaveBeenCalledTimes(1);
|
||||
expect(scheduler.add).toHaveBeenCalledTimes(1);
|
||||
expect(scheduler.start).toHaveBeenCalledTimes(1);
|
||||
await coordinator.stop();
|
||||
expect(scheduler.remove).toHaveBeenCalledWith('alpha');
|
||||
});
|
||||
|
||||
it('stop removes every key and prevents future polling', async () => {
|
||||
vi.useFakeTimers();
|
||||
const refreshHealth = vi.fn(async () => undefined);
|
||||
const coordinator = new FleetStatusCoordinator({
|
||||
source: { listEnabled: async () => ['alpha', 'beta'] },
|
||||
snapshots: { refreshHealth },
|
||||
intervalMs: 10,
|
||||
scheduler: (options) => new StatusPollScheduler({ ...options, concurrency: 2 }),
|
||||
});
|
||||
|
||||
await coordinator.start();
|
||||
await flush();
|
||||
expect(refreshHealth).toHaveBeenCalledTimes(2);
|
||||
|
||||
await coordinator.stop();
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect(refreshHealth).toHaveBeenCalledTimes(2);
|
||||
await coordinator.reconcile();
|
||||
expect(refreshHealth).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('isolates one failed refresh so another fleet member is still polled', async () => {
|
||||
const refreshHealth = vi.fn(async (id: string) => {
|
||||
if (id === 'bad') throw new Error('unreachable');
|
||||
});
|
||||
const coordinator = new FleetStatusCoordinator({
|
||||
source: { listEnabled: async () => ['bad', 'good'] },
|
||||
snapshots: { refreshHealth },
|
||||
intervalMs: 1_000,
|
||||
scheduler: (options) => new StatusPollScheduler({ ...options, concurrency: 1 }),
|
||||
});
|
||||
|
||||
await coordinator.start();
|
||||
await flush();
|
||||
|
||||
expect(refreshHealth.mock.calls.map(([id]) => id)).toEqual(['bad', 'good']);
|
||||
await coordinator.stop();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,110 @@
|
||||
import { StatusPollScheduler, type StatusPollSchedulerOptions } from './status-poll-scheduler.js';
|
||||
|
||||
export interface FleetStatusSource {
|
||||
listEnabled(): Promise<readonly string[]>;
|
||||
}
|
||||
|
||||
export interface FleetStatusSnapshots {
|
||||
refreshHealth(instanceId: string): Promise<unknown>;
|
||||
}
|
||||
|
||||
export interface FleetStatusScheduler {
|
||||
add(instanceId: string, intervalMs?: number): void;
|
||||
remove(instanceId: string): boolean;
|
||||
start(): void;
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
export type FleetStatusSchedulerFactory = (
|
||||
options: Pick<StatusPollSchedulerOptions, 'task' | 'intervalMs'>,
|
||||
) => FleetStatusScheduler;
|
||||
|
||||
export interface FleetStatusCoordinatorOptions {
|
||||
readonly source: FleetStatusSource;
|
||||
readonly snapshots: FleetStatusSnapshots;
|
||||
readonly intervalMs: number;
|
||||
/** Primarily an injection seam for testing; the default uses the fleet scheduler. */
|
||||
readonly scheduler?: FleetStatusSchedulerFactory;
|
||||
}
|
||||
|
||||
/** Keeps the polling scheduler's registrations aligned with the enabled fleet. */
|
||||
export class FleetStatusCoordinator {
|
||||
readonly #source: FleetStatusSource;
|
||||
readonly #intervalMs: number;
|
||||
readonly #scheduler: FleetStatusScheduler;
|
||||
readonly #registered = new Set<string>();
|
||||
#started = false;
|
||||
#stopped = false;
|
||||
#startPromise: Promise<void> | undefined;
|
||||
#reconcileTail: Promise<void> = Promise.resolve();
|
||||
#stopPromise?: Promise<void>;
|
||||
|
||||
constructor(options: FleetStatusCoordinatorOptions) {
|
||||
if (!Number.isFinite(options.intervalMs) || options.intervalMs <= 0) {
|
||||
throw new RangeError('intervalMs must be greater than zero');
|
||||
}
|
||||
this.#source = options.source;
|
||||
this.#intervalMs = options.intervalMs;
|
||||
const createScheduler =
|
||||
options.scheduler ??
|
||||
((schedulerOptions) => new StatusPollScheduler({ ...schedulerOptions, concurrency: 4 }));
|
||||
this.#scheduler = createScheduler({
|
||||
intervalMs: options.intervalMs,
|
||||
task: async (instanceId) => {
|
||||
await options.snapshots.refreshHealth(instanceId);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
start(): Promise<void> {
|
||||
if (this.#started || this.#stopped) return this.#startPromise ?? Promise.resolve();
|
||||
if (this.#startPromise) return this.#startPromise;
|
||||
|
||||
this.#startPromise = (async () => {
|
||||
await this.reconcile();
|
||||
if (this.#stopped) return;
|
||||
this.#scheduler.start();
|
||||
this.#started = true;
|
||||
})().catch((error: unknown) => {
|
||||
this.#startPromise = undefined;
|
||||
throw error;
|
||||
});
|
||||
return this.#startPromise;
|
||||
}
|
||||
|
||||
reconcile(): Promise<void> {
|
||||
if (this.#stopped) return Promise.resolve();
|
||||
const reconciliation = this.#reconcileTail.then(async () => {
|
||||
if (this.#stopped) return;
|
||||
const enabled = new Set(await this.#source.listEnabled());
|
||||
if (this.#stopped) return;
|
||||
|
||||
for (const instanceId of this.#registered) {
|
||||
if (!enabled.has(instanceId)) {
|
||||
this.#scheduler.remove(instanceId);
|
||||
this.#registered.delete(instanceId);
|
||||
}
|
||||
}
|
||||
for (const instanceId of enabled) {
|
||||
if (!this.#registered.has(instanceId)) {
|
||||
this.#scheduler.add(instanceId, this.#intervalMs);
|
||||
this.#registered.add(instanceId);
|
||||
}
|
||||
}
|
||||
});
|
||||
this.#reconcileTail = reconciliation.catch(() => undefined);
|
||||
return reconciliation;
|
||||
}
|
||||
|
||||
stop(): Promise<void> {
|
||||
if (this.#stopPromise) return this.#stopPromise;
|
||||
this.#stopped = true;
|
||||
this.#stopPromise = (async () => {
|
||||
await this.#reconcileTail;
|
||||
for (const instanceId of this.#registered) this.#scheduler.remove(instanceId);
|
||||
this.#registered.clear();
|
||||
await this.#scheduler.stop();
|
||||
})();
|
||||
return this.#stopPromise;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,282 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
|
||||
import { UpstreamError } from '../../infrastructure/transport/upstream-error.js';
|
||||
import {
|
||||
StatusSnapshotError,
|
||||
StatusSnapshotService,
|
||||
type HealthTransportResponse,
|
||||
} from './status-snapshot-service.js';
|
||||
|
||||
const databases: Database.Database[] = [];
|
||||
afterEach(() => {
|
||||
for (const db of databases.splice(0)) db.close();
|
||||
});
|
||||
|
||||
const instant = (value: string) => () => new Date(value);
|
||||
|
||||
function fixture(
|
||||
options: { response?: HealthTransportResponse; failure?: unknown; now?: () => Date } = {},
|
||||
) {
|
||||
const db = new Database(':memory:');
|
||||
db.pragma('foreign_keys=ON');
|
||||
migrateDatabase(db);
|
||||
databases.push(db);
|
||||
const created = '2026-07-17T10:00:00.000Z';
|
||||
db.prepare(
|
||||
'INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)',
|
||||
).run('instance-1', 'LAN', 'http://192.168.1.20:3000', 'none', 1, 1, created, created);
|
||||
const instances = {
|
||||
get: vi.fn(async (id: string) =>
|
||||
id === 'instance-1' ? { id, origin: 'http://192.168.1.20:3000' } : undefined,
|
||||
),
|
||||
};
|
||||
const transport = {
|
||||
get: vi.fn(async (url: string) => {
|
||||
expect(url).toBe('http://192.168.1.20:3000/api/health');
|
||||
if (options.failure !== undefined) throw options.failure;
|
||||
return (
|
||||
options.response ?? {
|
||||
status: 200,
|
||||
body: '{"status":"ok","version":"1.2.3","platform":"darwin","secret":"no"}',
|
||||
}
|
||||
);
|
||||
}),
|
||||
};
|
||||
const service = new StatusSnapshotService({
|
||||
db,
|
||||
instances,
|
||||
transport,
|
||||
ttlMs: 60_000,
|
||||
maxStaleMs: 300_000,
|
||||
now: options.now ?? instant('2026-07-17T12:00:00.000Z'),
|
||||
idFactory: () => 'snapshot-id',
|
||||
});
|
||||
return { db, instances, transport, service };
|
||||
}
|
||||
|
||||
const row = (db: Database.Database) =>
|
||||
db
|
||||
.prepare(
|
||||
'SELECT id,instance_id,category,state,payload_json,observed_at,expires_at,created_at FROM status_snapshots',
|
||||
)
|
||||
.get() as Record<string, unknown> | undefined;
|
||||
|
||||
const payload = (db: Database.Database) =>
|
||||
JSON.parse(String(row(db)?.payload_json)) as Record<string, unknown>;
|
||||
|
||||
const seed = (db: Database.Database, overrides: { observedAt?: string; payload?: string } = {}) => {
|
||||
const observedAt = overrides.observedAt ?? '2026-07-17T11:58:00.000Z';
|
||||
db.prepare(
|
||||
'INSERT INTO status_snapshots (id,instance_id,category,state,payload_json,observed_at,expires_at,created_at) VALUES (?,?,?,?,?,?,?,?)',
|
||||
).run(
|
||||
'original-id',
|
||||
'instance-1',
|
||||
'health',
|
||||
'fresh',
|
||||
overrides.payload ?? '{"status":"old","version":"1.0.0"}',
|
||||
observedAt,
|
||||
'2026-07-17T11:59:00.000Z',
|
||||
'2026-07-17T11:58:00.000Z',
|
||||
);
|
||||
};
|
||||
|
||||
describe('StatusSnapshotService health snapshots', () => {
|
||||
it('persists only a bounded allowlisted fresh envelope', async () => {
|
||||
const { db, service } = fixture();
|
||||
await expect(service.refreshHealth('instance-1')).resolves.toEqual({
|
||||
instanceId: 'instance-1',
|
||||
category: 'health',
|
||||
state: 'fresh',
|
||||
payload: { status: 'ok', version: '1.2.3', platform: 'darwin' },
|
||||
observedAt: '2026-07-17T12:00:00.000Z',
|
||||
expiresAt: '2026-07-17T12:01:00.000Z',
|
||||
});
|
||||
expect(payload(db)).toEqual({
|
||||
schemaVersion: 1,
|
||||
data: { status: 'ok', version: '1.2.3', platform: 'darwin' },
|
||||
errorCode: null,
|
||||
httpStatus: 200,
|
||||
fetchedAt: '2026-07-17T12:00:00.000Z',
|
||||
durationMs: 0,
|
||||
freshness: 'fresh',
|
||||
supported: true,
|
||||
dataFetchedAt: '2026-07-17T12:00:00.000Z',
|
||||
});
|
||||
expect(JSON.stringify(row(db))).not.toContain('secret');
|
||||
});
|
||||
|
||||
it.each([
|
||||
[401, 'AUTH_REQUIRED'],
|
||||
[403, 'AUTH_REQUIRED'],
|
||||
[500, 'UPSTREAM_DEGRADED'],
|
||||
[503, 'UPSTREAM_DEGRADED'],
|
||||
] as const)('retains recent safe data as stale for HTTP %i', async (status, errorCode) => {
|
||||
const { db, service } = fixture({ response: { status, body: 'password=never-store' } });
|
||||
seed(db);
|
||||
await expect(service.refreshHealth('instance-1')).resolves.toMatchObject({
|
||||
state: 'stale',
|
||||
payload: { status: 'old', version: '1.0.0' },
|
||||
expiresAt: '2026-07-17T12:03:00.000Z',
|
||||
});
|
||||
expect(payload(db)).toEqual({
|
||||
schemaVersion: 1,
|
||||
data: { status: 'old', version: '1.0.0' },
|
||||
errorCode,
|
||||
httpStatus: status,
|
||||
fetchedAt: '2026-07-17T12:00:00.000Z',
|
||||
durationMs: 0,
|
||||
freshness: 'stale',
|
||||
supported: true,
|
||||
dataFetchedAt: '2026-07-17T11:58:00.000Z',
|
||||
});
|
||||
expect(row(db)).toMatchObject({ id: 'original-id', created_at: '2026-07-17T11:58:00.000Z' });
|
||||
expect(JSON.stringify(row(db))).not.toContain('password');
|
||||
});
|
||||
|
||||
it('marks 404 unsupported without retaining upstream content', async () => {
|
||||
const { db, service } = fixture({ response: { status: 404, body: '{"token":"never"}' } });
|
||||
seed(db);
|
||||
await expect(service.refreshHealth('instance-1')).resolves.toMatchObject({
|
||||
state: 'unknown',
|
||||
payload: {},
|
||||
expiresAt: null,
|
||||
});
|
||||
expect(payload(db)).toEqual({
|
||||
schemaVersion: 1,
|
||||
data: {},
|
||||
errorCode: 'UNSUPPORTED',
|
||||
httpStatus: 404,
|
||||
fetchedAt: '2026-07-17T12:00:00.000Z',
|
||||
durationMs: 0,
|
||||
freshness: 'unknown',
|
||||
supported: false,
|
||||
dataFetchedAt: null,
|
||||
});
|
||||
expect(JSON.stringify(row(db))).not.toContain('token');
|
||||
});
|
||||
|
||||
it.each([
|
||||
[new UpstreamError('UPSTREAM_TIMEOUT'), 'UPSTREAM_TIMEOUT'],
|
||||
[new UpstreamError('UPSTREAM_RESPONSE_TOO_LARGE'), 'RESPONSE_TOO_LARGE'],
|
||||
[new Error('private origin password'), 'UPSTREAM_UNAVAILABLE'],
|
||||
] as const)('persists stable typed transport classification', async (failure, errorCode) => {
|
||||
const { db, service } = fixture({ failure });
|
||||
await service.refreshHealth('instance-1');
|
||||
expect(payload(db)).toEqual({
|
||||
schemaVersion: 1,
|
||||
data: {},
|
||||
errorCode,
|
||||
httpStatus: null,
|
||||
fetchedAt: '2026-07-17T12:00:00.000Z',
|
||||
durationMs: 0,
|
||||
freshness: 'unknown',
|
||||
supported: null,
|
||||
dataFetchedAt: null,
|
||||
});
|
||||
if (!(failure instanceof UpstreamError)) {
|
||||
expect(JSON.stringify(row(db))).not.toContain(failure.message);
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ status: 200, body: 'not-json' },
|
||||
{ status: 200, body: '{"status":{"secret":"x"}}' },
|
||||
{ status: 200, body: `{"version":"${'x'.repeat(129)}"}` },
|
||||
{ status: 700, body: '{"status":"ok"}' },
|
||||
])('treats invalid responses as unknown and never stores raw content', async (response) => {
|
||||
const { db, service } = fixture({ response });
|
||||
await expect(service.refreshHealth('instance-1')).resolves.toMatchObject({
|
||||
state: 'unknown',
|
||||
payload: {},
|
||||
});
|
||||
expect(payload(db)).toMatchObject({
|
||||
data: {},
|
||||
freshness: 'unknown',
|
||||
errorCode: 'INVALID_RESPONSE',
|
||||
});
|
||||
expect(JSON.stringify(row(db))).not.toContain('secret');
|
||||
});
|
||||
|
||||
it('does not retain data beyond max-stale', async () => {
|
||||
const { db, service } = fixture({ response: { status: 503, body: 'private' } });
|
||||
seed(db, { observedAt: '2026-07-17T11:54:59.999Z' });
|
||||
await service.refreshHealth('instance-1');
|
||||
expect(row(db)).toMatchObject({
|
||||
id: 'original-id',
|
||||
state: 'unknown',
|
||||
expires_at: null,
|
||||
created_at: '2026-07-17T11:58:00.000Z',
|
||||
});
|
||||
expect(payload(db)).toMatchObject({ data: {}, freshness: 'unknown' });
|
||||
});
|
||||
|
||||
it('does not extend max-stale when repeated failures update observedAt', async () => {
|
||||
const first = fixture({ response: { status: 503, body: 'private' } });
|
||||
seed(first.db, {
|
||||
payload: JSON.stringify({
|
||||
schemaVersion: 1,
|
||||
data: { status: 'old' },
|
||||
errorCode: 'UPSTREAM_DEGRADED',
|
||||
httpStatus: 503,
|
||||
fetchedAt: '2026-07-17T11:59:30.000Z',
|
||||
durationMs: 0,
|
||||
freshness: 'stale',
|
||||
supported: true,
|
||||
dataFetchedAt: '2026-07-17T11:54:59.999Z',
|
||||
}),
|
||||
});
|
||||
await first.service.refreshHealth('instance-1');
|
||||
expect(payload(first.db)).toMatchObject({ data: {}, freshness: 'unknown' });
|
||||
});
|
||||
|
||||
it('uses generation and observed-at fences for latest-wins', async () => {
|
||||
const { db, service, transport } = fixture();
|
||||
let resolveOld!: (value: HealthTransportResponse) => void;
|
||||
const oldResponse = new Promise<HealthTransportResponse>((resolve) => {
|
||||
resolveOld = resolve;
|
||||
});
|
||||
transport.get
|
||||
.mockImplementationOnce(async () => oldResponse)
|
||||
.mockResolvedValueOnce({ status: 200, body: '{"status":"new"}' });
|
||||
const old = service.refreshHealth('instance-1');
|
||||
await vi.waitFor(() => expect(transport.get).toHaveBeenCalledTimes(1));
|
||||
await service.refreshHealth('instance-1');
|
||||
resolveOld({ status: 200, body: '{"status":"old"}' });
|
||||
await old;
|
||||
expect(payload(db)).toMatchObject({ data: { status: 'new' } });
|
||||
|
||||
db.prepare(
|
||||
"UPDATE status_snapshots SET observed_at=?,payload_json=? WHERE instance_id=? AND category='health'",
|
||||
).run('2099-01-01T00:00:00.000Z', '{"status":"future"}', 'instance-1');
|
||||
await service.refreshHealth('instance-1');
|
||||
expect(row(db)).toMatchObject({
|
||||
observed_at: '2099-01-01T00:00:00.000Z',
|
||||
payload_json: '{"status":"future"}',
|
||||
});
|
||||
});
|
||||
|
||||
it('does no network/write for missing instances and no write after in-flight deletion', async () => {
|
||||
const first = fixture();
|
||||
await expect(first.service.refreshHealth('missing')).rejects.toEqual(
|
||||
expect.objectContaining<Partial<StatusSnapshotError>>({ code: 'INSTANCE_NOT_FOUND' }),
|
||||
);
|
||||
expect(first.transport.get).not.toHaveBeenCalled();
|
||||
expect(row(first.db)).toBeUndefined();
|
||||
|
||||
let resolve!: (value: HealthTransportResponse) => void;
|
||||
first.transport.get.mockImplementationOnce(
|
||||
async () =>
|
||||
new Promise((done) => {
|
||||
resolve = done;
|
||||
}),
|
||||
);
|
||||
const running = first.service.refreshHealth('instance-1');
|
||||
await vi.waitFor(() => expect(first.transport.get).toHaveBeenCalledTimes(1));
|
||||
first.db.prepare('DELETE FROM instances WHERE id=?').run('instance-1');
|
||||
resolve({ status: 200, body: '{"status":"ok"}' });
|
||||
await running;
|
||||
expect(row(first.db)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,380 @@
|
||||
import type Database from 'better-sqlite3';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { UpstreamError } from '../../infrastructure/transport/upstream-error.js';
|
||||
|
||||
export interface HealthTransportResponse {
|
||||
readonly status: number;
|
||||
readonly body: string;
|
||||
}
|
||||
|
||||
export type HealthSnapshotErrorCode =
|
||||
| 'AUTH_REQUIRED'
|
||||
| 'UNSUPPORTED'
|
||||
| 'UPSTREAM_DEGRADED'
|
||||
| 'UPSTREAM_TIMEOUT'
|
||||
| 'UPSTREAM_UNAVAILABLE'
|
||||
| 'RESPONSE_TOO_LARGE'
|
||||
| 'INVALID_RESPONSE';
|
||||
|
||||
export interface HealthSnapshotInstance {
|
||||
readonly id: string;
|
||||
readonly origin: string;
|
||||
}
|
||||
|
||||
export interface HealthEnvelope {
|
||||
readonly status?: string;
|
||||
readonly version?: string;
|
||||
readonly platform?: string;
|
||||
}
|
||||
|
||||
export type HealthSnapshotState = 'fresh' | 'stale' | 'unknown';
|
||||
|
||||
export interface HealthSnapshot {
|
||||
readonly instanceId: string;
|
||||
readonly category: 'health';
|
||||
readonly state: HealthSnapshotState;
|
||||
readonly payload: HealthEnvelope;
|
||||
readonly observedAt: string;
|
||||
readonly expiresAt: string | null;
|
||||
}
|
||||
|
||||
export interface StatusSnapshotServiceOptions {
|
||||
readonly db: Database.Database;
|
||||
readonly instances: { get(instanceId: string): Promise<HealthSnapshotInstance | undefined> };
|
||||
readonly transport: { get(url: string): Promise<HealthTransportResponse> };
|
||||
readonly ttlMs: number;
|
||||
readonly maxStaleMs: number;
|
||||
readonly now?: () => Date;
|
||||
readonly idFactory?: () => string;
|
||||
}
|
||||
|
||||
export class StatusSnapshotError extends Error {
|
||||
constructor(readonly code: 'INSTANCE_NOT_FOUND') {
|
||||
super(code);
|
||||
this.name = 'StatusSnapshotError';
|
||||
}
|
||||
}
|
||||
|
||||
interface SnapshotRow {
|
||||
payload_json: string;
|
||||
observed_at: string;
|
||||
}
|
||||
|
||||
type SnapshotErrorCode = HealthSnapshotErrorCode;
|
||||
|
||||
interface PersistedHealthEnvelope {
|
||||
readonly schemaVersion: 1;
|
||||
readonly data: HealthEnvelope;
|
||||
readonly errorCode: SnapshotErrorCode | null;
|
||||
readonly httpStatus: number | null;
|
||||
readonly fetchedAt: string;
|
||||
readonly durationMs: number;
|
||||
readonly freshness: HealthSnapshotState;
|
||||
readonly supported: boolean | null;
|
||||
readonly dataFetchedAt: string | null;
|
||||
}
|
||||
|
||||
interface ClassifiedSnapshot {
|
||||
readonly snapshot: HealthSnapshot;
|
||||
readonly persisted: PersistedHealthEnvelope;
|
||||
}
|
||||
|
||||
const MAX_FIELD_LENGTH = 128;
|
||||
const SAFE_FIELD = /^[^\x00-\x1f\x7f]*$/;
|
||||
|
||||
function parseEnvelope(body: string): HealthEnvelope | undefined {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(body);
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) return undefined;
|
||||
const source = parsed as Record<string, unknown>;
|
||||
const envelope: { status?: string; version?: string; platform?: string } = {};
|
||||
for (const key of ['status', 'version', 'platform'] as const) {
|
||||
const value = source[key];
|
||||
if (value === undefined) continue;
|
||||
if (
|
||||
typeof value !== 'string' ||
|
||||
value.length === 0 ||
|
||||
value.length > MAX_FIELD_LENGTH ||
|
||||
!SAFE_FIELD.test(value)
|
||||
)
|
||||
return undefined;
|
||||
envelope[key] = value;
|
||||
}
|
||||
return Object.freeze(envelope);
|
||||
}
|
||||
|
||||
function safePreviousPayload(payloadJson: string): HealthEnvelope | undefined {
|
||||
let candidate = payloadJson;
|
||||
try {
|
||||
const parsed = JSON.parse(payloadJson) as { data?: unknown };
|
||||
if (parsed && typeof parsed === 'object' && 'data' in parsed)
|
||||
candidate = JSON.stringify(parsed.data);
|
||||
} catch {
|
||||
// parseEnvelope below handles malformed legacy payloads.
|
||||
}
|
||||
const envelope = parseEnvelope(candidate);
|
||||
return envelope && Object.keys(envelope).length > 0 ? envelope : undefined;
|
||||
}
|
||||
|
||||
function previousDataFetchedAt(previous: SnapshotRow): string {
|
||||
try {
|
||||
const parsed = JSON.parse(previous.payload_json) as { dataFetchedAt?: unknown };
|
||||
if (
|
||||
typeof parsed.dataFetchedAt === 'string' &&
|
||||
Number.isFinite(Date.parse(parsed.dataFetchedAt))
|
||||
) {
|
||||
return parsed.dataFetchedAt;
|
||||
}
|
||||
} catch {
|
||||
// Legacy rows use observed_at as their data fetch time.
|
||||
}
|
||||
return previous.observed_at;
|
||||
}
|
||||
|
||||
function responseErrorCode(response: HealthTransportResponse | undefined): SnapshotErrorCode {
|
||||
if (
|
||||
!response ||
|
||||
!Number.isSafeInteger(response.status) ||
|
||||
response.status < 100 ||
|
||||
response.status > 599
|
||||
) {
|
||||
return 'INVALID_RESPONSE';
|
||||
}
|
||||
if (response.status === 401 || response.status === 403) return 'AUTH_REQUIRED';
|
||||
if (response.status === 404) return 'UNSUPPORTED';
|
||||
if (response.status >= 500) return 'UPSTREAM_DEGRADED';
|
||||
return 'INVALID_RESPONSE';
|
||||
}
|
||||
|
||||
function transportErrorCode(error: unknown): SnapshotErrorCode {
|
||||
if (!(error instanceof UpstreamError)) return 'UPSTREAM_UNAVAILABLE';
|
||||
if (error.code === 'UPSTREAM_TIMEOUT') return 'UPSTREAM_TIMEOUT';
|
||||
if (error.code === 'UPSTREAM_RESPONSE_TOO_LARGE') return 'RESPONSE_TOO_LARGE';
|
||||
return 'UPSTREAM_UNAVAILABLE';
|
||||
}
|
||||
|
||||
const healthUrl = (origin: string): string => `${origin.replace(/\/$/, '')}/api/health`;
|
||||
|
||||
export class StatusSnapshotService {
|
||||
readonly #db: Database.Database;
|
||||
readonly #instances: StatusSnapshotServiceOptions['instances'];
|
||||
readonly #transport: StatusSnapshotServiceOptions['transport'];
|
||||
readonly #ttlMs: number;
|
||||
readonly #maxStaleMs: number;
|
||||
readonly #now: () => Date;
|
||||
readonly #id: () => string;
|
||||
readonly #latestGeneration = new Map<string, number>();
|
||||
#nextGeneration = 0;
|
||||
|
||||
constructor(options: StatusSnapshotServiceOptions) {
|
||||
if (!Number.isFinite(options.ttlMs) || options.ttlMs <= 0) {
|
||||
throw new RangeError('ttlMs must be greater than zero');
|
||||
}
|
||||
if (!Number.isFinite(options.maxStaleMs) || options.maxStaleMs < options.ttlMs) {
|
||||
throw new RangeError('maxStaleMs must be at least ttlMs');
|
||||
}
|
||||
this.#db = options.db;
|
||||
this.#instances = options.instances;
|
||||
this.#transport = options.transport;
|
||||
this.#ttlMs = options.ttlMs;
|
||||
this.#maxStaleMs = options.maxStaleMs;
|
||||
this.#now = options.now ?? (() => new Date());
|
||||
this.#id = options.idFactory ?? randomUUID;
|
||||
}
|
||||
|
||||
async refreshHealth(instanceId: string): Promise<HealthSnapshot> {
|
||||
const instance = await this.#instances.get(instanceId);
|
||||
if (!instance) throw new StatusSnapshotError('INSTANCE_NOT_FOUND');
|
||||
|
||||
const generation = ++this.#nextGeneration;
|
||||
this.#latestGeneration.set(instanceId, generation);
|
||||
|
||||
const started = this.#now();
|
||||
let response: HealthTransportResponse | undefined;
|
||||
let transportError: unknown;
|
||||
try {
|
||||
response = await this.#transport.get(healthUrl(instance.origin));
|
||||
} catch (error) {
|
||||
transportError = error;
|
||||
}
|
||||
const completed = this.#now();
|
||||
const observedAt = completed.toISOString();
|
||||
const previous = this.#previous(instanceId);
|
||||
const classified = this.#classify(
|
||||
instanceId,
|
||||
response,
|
||||
transportError,
|
||||
previous,
|
||||
completed,
|
||||
observedAt,
|
||||
Math.max(0, Math.min(Number.MAX_SAFE_INTEGER, completed.getTime() - started.getTime())),
|
||||
);
|
||||
const { snapshot } = classified;
|
||||
|
||||
if (this.#latestGeneration.get(instanceId) === generation) {
|
||||
this.#persist(snapshot, classified.persisted);
|
||||
this.#latestGeneration.delete(instanceId);
|
||||
}
|
||||
return Object.freeze({ ...snapshot, payload: Object.freeze({ ...snapshot.payload }) });
|
||||
}
|
||||
|
||||
#previous(instanceId: string): SnapshotRow | undefined {
|
||||
return this.#db
|
||||
.prepare(
|
||||
"SELECT payload_json,observed_at FROM status_snapshots WHERE instance_id=? AND category='health'",
|
||||
)
|
||||
.get(instanceId) as SnapshotRow | undefined;
|
||||
}
|
||||
|
||||
#classify(
|
||||
instanceId: string,
|
||||
response: HealthTransportResponse | undefined,
|
||||
transportError: unknown,
|
||||
previous: SnapshotRow | undefined,
|
||||
completed: Date,
|
||||
observedAt: string,
|
||||
durationMs: number,
|
||||
): ClassifiedSnapshot {
|
||||
if (
|
||||
response &&
|
||||
Number.isSafeInteger(response.status) &&
|
||||
response.status >= 200 &&
|
||||
response.status < 300 &&
|
||||
typeof response.body === 'string'
|
||||
) {
|
||||
const payload = parseEnvelope(response.body);
|
||||
if (payload) {
|
||||
const snapshot: HealthSnapshot = {
|
||||
instanceId,
|
||||
category: 'health',
|
||||
state: 'fresh',
|
||||
payload,
|
||||
observedAt,
|
||||
expiresAt: new Date(completed.getTime() + this.#ttlMs).toISOString(),
|
||||
};
|
||||
return {
|
||||
snapshot,
|
||||
persisted: this.#envelope(snapshot, null, response.status, durationMs, true, observedAt),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (response?.status === 404) {
|
||||
const snapshot = this.#unknown(instanceId, observedAt);
|
||||
return {
|
||||
snapshot,
|
||||
persisted: this.#envelope(snapshot, 'UNSUPPORTED', 404, durationMs, false, null),
|
||||
};
|
||||
}
|
||||
|
||||
const errorCode: SnapshotErrorCode = transportError
|
||||
? transportErrorCode(transportError)
|
||||
: responseErrorCode(response);
|
||||
const httpStatus =
|
||||
response &&
|
||||
Number.isSafeInteger(response.status) &&
|
||||
response.status >= 100 &&
|
||||
response.status <= 599
|
||||
? response.status
|
||||
: null;
|
||||
const previousPayload = previous && safePreviousPayload(previous.payload_json);
|
||||
const previousTime = previous ? Date.parse(previousDataFetchedAt(previous)) : Number.NaN;
|
||||
if (
|
||||
previousPayload &&
|
||||
Number.isFinite(previousTime) &&
|
||||
completed.getTime() - previousTime <= this.#maxStaleMs
|
||||
) {
|
||||
const snapshot: HealthSnapshot = {
|
||||
instanceId,
|
||||
category: 'health',
|
||||
state: 'stale',
|
||||
payload: previousPayload,
|
||||
observedAt,
|
||||
expiresAt: new Date(previousTime + this.#maxStaleMs).toISOString(),
|
||||
};
|
||||
return {
|
||||
snapshot,
|
||||
persisted: this.#envelope(
|
||||
snapshot,
|
||||
errorCode,
|
||||
httpStatus,
|
||||
durationMs,
|
||||
httpStatus === null ? null : true,
|
||||
previousDataFetchedAt(previous),
|
||||
),
|
||||
};
|
||||
}
|
||||
const snapshot = this.#unknown(instanceId, observedAt);
|
||||
return {
|
||||
snapshot,
|
||||
persisted: this.#envelope(
|
||||
snapshot,
|
||||
errorCode,
|
||||
httpStatus,
|
||||
durationMs,
|
||||
httpStatus === null ? null : true,
|
||||
null,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
#unknown(instanceId: string, observedAt: string): HealthSnapshot {
|
||||
return {
|
||||
instanceId,
|
||||
category: 'health',
|
||||
state: 'unknown',
|
||||
payload: {},
|
||||
observedAt,
|
||||
expiresAt: null,
|
||||
};
|
||||
}
|
||||
|
||||
#envelope(
|
||||
snapshot: HealthSnapshot,
|
||||
errorCode: SnapshotErrorCode | null,
|
||||
httpStatus: number | null,
|
||||
durationMs: number,
|
||||
supported: boolean | null,
|
||||
dataFetchedAt: string | null,
|
||||
): PersistedHealthEnvelope {
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
data: snapshot.payload,
|
||||
errorCode,
|
||||
httpStatus,
|
||||
fetchedAt: snapshot.observedAt,
|
||||
durationMs,
|
||||
freshness: snapshot.state,
|
||||
supported,
|
||||
dataFetchedAt,
|
||||
};
|
||||
}
|
||||
|
||||
#persist(snapshot: HealthSnapshot, persisted: PersistedHealthEnvelope): void {
|
||||
this.#db
|
||||
.prepare(
|
||||
`INSERT INTO status_snapshots
|
||||
(id,instance_id,category,state,payload_json,observed_at,expires_at,created_at)
|
||||
SELECT ?,id,'health',?,?,?,?,? FROM instances WHERE id=?
|
||||
ON CONFLICT(instance_id,category) DO UPDATE SET
|
||||
state=excluded.state,
|
||||
payload_json=excluded.payload_json,
|
||||
observed_at=excluded.observed_at,
|
||||
expires_at=excluded.expires_at
|
||||
WHERE excluded.observed_at > status_snapshots.observed_at`,
|
||||
)
|
||||
.run(
|
||||
this.#id(),
|
||||
snapshot.state,
|
||||
JSON.stringify(persisted),
|
||||
snapshot.observedAt,
|
||||
snapshot.expiresAt,
|
||||
snapshot.observedAt,
|
||||
snapshot.instanceId,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -23,6 +23,63 @@ class Store implements SecretStore {
|
||||
}
|
||||
}
|
||||
describe('buildControlPlaneApp', () => {
|
||||
it('registers the durable event route fail-closed when no authentication dependency is supplied', async () => {
|
||||
const db = new Database(':memory:');
|
||||
db.pragma('foreign_keys=ON');
|
||||
migrateDatabase(db);
|
||||
dbs.push(db);
|
||||
const app = buildControlPlaneApp({
|
||||
db,
|
||||
store: new Store(),
|
||||
upstream: {
|
||||
get: async () => ({ status: 200, headers: {}, body: '' }),
|
||||
request: async () => ({ status: 200, headers: {}, body: '' }),
|
||||
postNetworkRegisterAuto: async () => ({ status: 200 }),
|
||||
},
|
||||
});
|
||||
|
||||
const response = await app.inject({ method: 'GET', url: '/api/v1/events' });
|
||||
|
||||
expect(response.statusCode).toBe(401);
|
||||
expect(response.headers['content-type']).toContain('application/problem+json');
|
||||
expect(response.json()).toMatchObject({ status: 401, code: 'UNAUTHORIZED' });
|
||||
await app.close();
|
||||
});
|
||||
|
||||
it('passes an explicit event-stream authenticator to durable cursor replay', async () => {
|
||||
const db = new Database(':memory:');
|
||||
db.pragma('foreign_keys=ON');
|
||||
migrateDatabase(db);
|
||||
dbs.push(db);
|
||||
const app = buildControlPlaneApp({
|
||||
db,
|
||||
store: new Store(),
|
||||
upstream: {
|
||||
get: async () => ({ status: 200, headers: {}, body: '' }),
|
||||
request: async () => ({ status: 200, headers: {}, body: '' }),
|
||||
postNetworkRegisterAuto: async () => ({ status: 200 }),
|
||||
},
|
||||
authenticateEventStream: (request) => request.headers.authorization === 'Bearer allowed',
|
||||
});
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/v1/events',
|
||||
headers: {
|
||||
authorization: 'Bearer allowed',
|
||||
'last-event-id': 'unknown-cursor',
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(409);
|
||||
expect(response.headers['content-type']).toContain('application/problem+json');
|
||||
expect(response.json()).toMatchObject({
|
||||
status: 409,
|
||||
code: 'EVENT_POSITION_UNAVAILABLE',
|
||||
});
|
||||
await app.close();
|
||||
});
|
||||
|
||||
it('assembles instance CRUD, connection checks, login and logout without listening or reading external config', async () => {
|
||||
const db = new Database(':memory:');
|
||||
db.pragma('foreign_keys=ON');
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
import type Database from 'better-sqlite3';
|
||||
import { EventJournal } from './application/events/event-journal.js';
|
||||
import { registerOperationRoutes } from './interface/http/operation-routes.js';
|
||||
import { operationCatalogRegistry } from './application/operations/operation-catalog-data.js';
|
||||
import {
|
||||
SecureOperationExecution,
|
||||
secureOperationRegistry,
|
||||
} from './application/operations/secure-operation-execution.js';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { FastifyInstance, FastifyRequest } from 'fastify';
|
||||
import { buildApp, type BuildAppOptions } from './app.js';
|
||||
import {
|
||||
ConnectionProbe,
|
||||
@@ -22,6 +23,7 @@ import { InstanceService } from './application/instances/instance-service.js';
|
||||
import { DeleteInstanceOperation } from './application/operations/delete-instance-operation.js';
|
||||
import type { SecretStore } from './infrastructure/secrets/secret-store.js';
|
||||
import { registerInstanceRoutes } from './interface/http/instance-routes.js';
|
||||
import { registerEventRoutes } from './interface/http/event-routes.js';
|
||||
|
||||
export interface SafeControlPlaneUpstream extends ConnectionTransport {
|
||||
request: UpstreamSessionClientOptions['request'];
|
||||
@@ -32,12 +34,14 @@ export interface ControlPlaneOptions {
|
||||
readonly store: SecretStore;
|
||||
readonly upstream: SafeControlPlaneUpstream;
|
||||
readonly now?: () => Date;
|
||||
readonly authenticateEventStream?: (request: FastifyRequest) => boolean;
|
||||
readonly app?: Omit<BuildAppOptions, 'registerRoutes'>;
|
||||
}
|
||||
export interface ControlPlaneApp extends FastifyInstance {
|
||||
retryPendingSecretCleanup(): Promise<unknown>;
|
||||
}
|
||||
export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlaneApp {
|
||||
const eventJournal = new EventJournal(options.db);
|
||||
const instances = options.now
|
||||
? new InstanceService({ db: options.db, store: options.store, now: options.now })
|
||||
: new InstanceService({ db: options.db, store: options.store });
|
||||
@@ -82,6 +86,12 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
|
||||
registerDeletionPreparationRoute: false,
|
||||
});
|
||||
registerOperationRoutes(app, operationCatalogRegistry, secureExecution, deletion);
|
||||
registerEventRoutes(app, {
|
||||
journal: eventJournal,
|
||||
...(options.authenticateEventStream
|
||||
? { authenticate: options.authenticateEventStream }
|
||||
: {}),
|
||||
});
|
||||
},
|
||||
});
|
||||
Object.assign(app, {
|
||||
|
||||
@@ -62,6 +62,7 @@ describe('database migrations', () => {
|
||||
'app_settings',
|
||||
'audit_events',
|
||||
'capabilities',
|
||||
'event_journal',
|
||||
'instance_tags',
|
||||
'instances',
|
||||
'job_attempts',
|
||||
@@ -117,6 +118,7 @@ describe('database migrations', () => {
|
||||
expect(indexes).toEqual(
|
||||
expect.arrayContaining([
|
||||
'idx_audit_events_created_at',
|
||||
'idx_event_journal_sequence',
|
||||
'idx_jobs_status_created_at',
|
||||
'idx_status_snapshots_instance_observed_at',
|
||||
]),
|
||||
|
||||
@@ -270,6 +270,18 @@ export const MIGRATIONS: readonly Migration[] = [
|
||||
'CREATE INDEX idx_operation_preparations_status_expires_at ON operation_preparations(status, expires_at)',
|
||||
],
|
||||
},
|
||||
{
|
||||
id: 6,
|
||||
name: 'durable-event-journal',
|
||||
statements: [
|
||||
`CREATE TABLE event_journal (
|
||||
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
public_id TEXT NOT NULL UNIQUE,
|
||||
envelope_json TEXT NOT NULL
|
||||
)`,
|
||||
'CREATE INDEX idx_event_journal_sequence ON event_journal(sequence)',
|
||||
],
|
||||
},
|
||||
];
|
||||
|
||||
const createMigrationsTable = `CREATE TABLE schema_migrations (
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import http from 'node:http';
|
||||
|
||||
import Database from 'better-sqlite3';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { buildApp } from '../../app.js';
|
||||
import { EventJournal, type EventEnvelope } from '../../application/events/event-journal.js';
|
||||
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
|
||||
import { registerEventRoutes } from './event-routes.js';
|
||||
|
||||
const databases: Database.Database[] = [];
|
||||
const apps: ReturnType<typeof buildApp>[] = [];
|
||||
const at = '2026-07-17T12:34:56.789Z';
|
||||
|
||||
const event = (id: string): EventEnvelope => ({
|
||||
kind: 'job',
|
||||
id,
|
||||
occurredAt: at,
|
||||
requestId: `request-${id}`,
|
||||
jobId: 'job-1',
|
||||
});
|
||||
|
||||
function fixture(authenticate: ((request: unknown) => boolean) | null = () => true) {
|
||||
const database = new Database(':memory:');
|
||||
migrateDatabase(database);
|
||||
databases.push(database);
|
||||
const journal = new EventJournal(database);
|
||||
const app = buildApp({
|
||||
registerRoutes: (scope) =>
|
||||
registerEventRoutes(scope, {
|
||||
journal,
|
||||
...(authenticate === null ? {} : { authenticate }),
|
||||
}),
|
||||
});
|
||||
apps.push(app);
|
||||
return { app, journal };
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(apps.splice(0).map(async (app) => app.close()));
|
||||
for (const database of databases.splice(0)) database.close();
|
||||
});
|
||||
|
||||
describe('events HTTP route', () => {
|
||||
it('requires the route-local authentication dependency and does not silently expose events', async () => {
|
||||
for (const authenticate of [null, () => false]) {
|
||||
const { app } = fixture(authenticate);
|
||||
const response = await app.inject({ method: 'GET', url: '/api/v1/events' });
|
||||
|
||||
expect(response.statusCode).toBe(401);
|
||||
expect(response.headers['content-type']).toContain('application/problem+json');
|
||||
expect(response.headers['x-request-id']).toBeTruthy();
|
||||
expect(response.json()).toMatchObject({
|
||||
type: 'about:blank',
|
||||
title: 'Unauthorized',
|
||||
status: 401,
|
||||
code: 'UNAUTHORIZED',
|
||||
requestId: response.headers['x-request-id'],
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it.each(['', '\0bad'])('rejects a malformed Last-Event-ID value (%j)', async (lastEventId) => {
|
||||
const { app } = fixture();
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/v1/events',
|
||||
headers: { 'last-event-id': lastEventId },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(400);
|
||||
expect(response.headers['content-type']).toContain('application/problem+json');
|
||||
expect(response.json()).toMatchObject({ status: 400, code: 'VALIDATION_FAILED' });
|
||||
expect(response.json().requestId).toBe(response.headers['x-request-id']);
|
||||
});
|
||||
|
||||
it('returns the frozen conflict Problem Details for an unavailable cursor', async () => {
|
||||
const { app, journal } = fixture();
|
||||
journal.append(event('known'));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/v1/events',
|
||||
headers: { 'last-event-id': 'expired-or-unknown' },
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(409);
|
||||
expect(response.headers['content-type']).toContain('application/problem+json');
|
||||
expect(response.headers['x-request-id']).toBeTruthy();
|
||||
expect(response.json()).toEqual({
|
||||
type: 'about:blank',
|
||||
title: 'Conflict',
|
||||
status: 409,
|
||||
code: 'EVENT_POSITION_UNAVAILABLE',
|
||||
detail: 'The requested event position is no longer available.',
|
||||
requestId: response.headers['x-request-id'],
|
||||
});
|
||||
});
|
||||
|
||||
it('replays strictly after the cursor, then streams live events without an invented event field', async () => {
|
||||
const { app, journal } = fixture();
|
||||
journal.append(event('one'));
|
||||
journal.append(event('two'));
|
||||
const address = await app.listen({ host: '127.0.0.1', port: 0 });
|
||||
|
||||
const chunks: string[] = [];
|
||||
let response!: http.IncomingMessage;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const request = http.get(
|
||||
`${address}/api/v1/events`,
|
||||
{ headers: { 'Last-Event-ID': 'one' } },
|
||||
(incoming) => {
|
||||
response = incoming;
|
||||
incoming.setEncoding('utf8');
|
||||
incoming.on('data', (chunk: string) => {
|
||||
chunks.push(chunk);
|
||||
if (chunks.join('').includes('id: three\n')) resolve();
|
||||
});
|
||||
incoming.on('error', reject);
|
||||
journal.append(event('three'));
|
||||
},
|
||||
);
|
||||
request.on('error', reject);
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.headers['cache-control']).toBe('no-cache');
|
||||
expect(response.headers['x-request-id']).toBeTruthy();
|
||||
const body = chunks.join('');
|
||||
expect(body).toBe(
|
||||
`id: two\ndata: ${JSON.stringify(event('two'))}\n\n` +
|
||||
`id: three\ndata: ${JSON.stringify(event('three'))}\n\n`,
|
||||
);
|
||||
expect(body).not.toContain('\nevent:');
|
||||
response.destroy();
|
||||
});
|
||||
|
||||
it('subscribes before replay so an event arriving during the read is neither lost nor duplicated', async () => {
|
||||
const { app, journal } = fixture();
|
||||
journal.append(event('one'));
|
||||
const originalRead = journal.readAfter.bind(journal);
|
||||
vi.spyOn(journal, 'readAfter').mockImplementation((cursor) => {
|
||||
journal.append(event('during-read'));
|
||||
return originalRead(cursor);
|
||||
});
|
||||
const address = await app.listen({ host: '127.0.0.1', port: 0 });
|
||||
|
||||
const body = await new Promise<string>((resolve, reject) => {
|
||||
http
|
||||
.get(`${address}/api/v1/events`, { headers: { 'Last-Event-ID': 'one' } }, (response) => {
|
||||
response.setEncoding('utf8');
|
||||
response.once('data', (chunk: string) => {
|
||||
response.destroy();
|
||||
resolve(chunk);
|
||||
});
|
||||
response.on('error', reject);
|
||||
})
|
||||
.on('error', reject);
|
||||
});
|
||||
|
||||
const frame = `id: during-read\ndata: ${JSON.stringify(event('during-read'))}\n\n`;
|
||||
expect(body).toBe(frame);
|
||||
});
|
||||
|
||||
it('unsubscribes when the client disconnects', async () => {
|
||||
const { app, journal } = fixture();
|
||||
const unsubscribe = vi.fn();
|
||||
const originalSubscribe = journal.subscribe.bind(journal);
|
||||
vi.spyOn(journal, 'subscribe').mockImplementation((subscriber) => {
|
||||
const originalUnsubscribe = originalSubscribe(subscriber);
|
||||
return () => {
|
||||
originalUnsubscribe();
|
||||
unsubscribe();
|
||||
};
|
||||
});
|
||||
const address = await app.listen({ host: '127.0.0.1', port: 0 });
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
http
|
||||
.get(`${address}/api/v1/events`, (response) => {
|
||||
response.destroy();
|
||||
resolve();
|
||||
})
|
||||
.on('error', reject);
|
||||
});
|
||||
await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledOnce());
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,162 @@
|
||||
import type { ServerResponse } from 'node:http';
|
||||
|
||||
import type { FastifyInstance, FastifyRequest } from 'fastify';
|
||||
|
||||
import {
|
||||
EventJournal,
|
||||
formatServerSentEvent,
|
||||
type JournalEvent,
|
||||
} from '../../application/events/event-journal.js';
|
||||
|
||||
const FORBIDDEN_CURSOR_CHARACTERS = /[\r\n\0]/u;
|
||||
|
||||
export interface EventRoutesOptions {
|
||||
readonly journal: EventJournal;
|
||||
/**
|
||||
* Route-local until the control plane supplies its cookie authentication hook.
|
||||
* Deliberately fail closed when omitted rather than exposing the event stream.
|
||||
*/
|
||||
readonly authenticate?: (request: FastifyRequest) => boolean;
|
||||
}
|
||||
|
||||
const problem = (
|
||||
request: FastifyRequest,
|
||||
status: 400 | 401 | 409,
|
||||
code: string,
|
||||
detail: string,
|
||||
): Record<string, unknown> => ({
|
||||
type: 'about:blank',
|
||||
title: status === 400 ? 'Bad Request' : status === 401 ? 'Unauthorized' : 'Conflict',
|
||||
status,
|
||||
code,
|
||||
detail,
|
||||
requestId: request.id,
|
||||
});
|
||||
|
||||
function cursor(request: FastifyRequest): string | undefined {
|
||||
const value = request.headers['last-event-id'];
|
||||
if (value === undefined) return undefined;
|
||||
if (typeof value !== 'string' || value.length === 0 || FORBIDDEN_CURSOR_CHARACTERS.test(value)) {
|
||||
throw new TypeError('Last-Event-ID must be a non-empty event identifier');
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function writeEvent(response: ServerResponse, event: JournalEvent, cleanup: () => void): void {
|
||||
if (response.destroyed || response.writableEnded) {
|
||||
cleanup();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
response.write(formatServerSentEvent(event.envelope), (error) => {
|
||||
if (error) cleanup();
|
||||
});
|
||||
} catch {
|
||||
cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
export function registerEventRoutes(app: FastifyInstance, options: EventRoutesOptions): void {
|
||||
app.get('/api/v1/events', async (request, reply) => {
|
||||
if (!options.authenticate?.(request)) {
|
||||
return reply
|
||||
.code(401)
|
||||
.type('application/problem+json')
|
||||
.send(
|
||||
problem(
|
||||
request,
|
||||
401,
|
||||
'UNAUTHORIZED',
|
||||
'Authentication is required to access the event stream.',
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
let lastEventId: string | undefined;
|
||||
try {
|
||||
lastEventId = cursor(request);
|
||||
} catch {
|
||||
return reply
|
||||
.code(400)
|
||||
.type('application/problem+json')
|
||||
.send(
|
||||
problem(
|
||||
request,
|
||||
400,
|
||||
'VALIDATION_FAILED',
|
||||
'Last-Event-ID must be a non-empty valid event identifier.',
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
// Subscribe first. Events appended while the durable replay query runs are buffered;
|
||||
// sequence de-duplication below closes the replay/subscription race.
|
||||
const buffered: JournalEvent[] = [];
|
||||
const stream: { response?: ServerResponse } = {};
|
||||
let live = false;
|
||||
let cleaned = false;
|
||||
let unsubscribe = () => {};
|
||||
const cleanup = () => {
|
||||
if (cleaned) return;
|
||||
cleaned = true;
|
||||
unsubscribe();
|
||||
const response = stream.response;
|
||||
if (response && !response.destroyed && !response.writableEnded) response.destroy();
|
||||
};
|
||||
unsubscribe = options.journal.subscribe((event) => {
|
||||
const response = stream.response;
|
||||
if (!live || !response) {
|
||||
buffered.push(event);
|
||||
return;
|
||||
}
|
||||
writeEvent(response, event, cleanup);
|
||||
});
|
||||
|
||||
let replay;
|
||||
try {
|
||||
replay = options.journal.readAfter(lastEventId);
|
||||
} catch (error) {
|
||||
cleanup();
|
||||
throw error;
|
||||
}
|
||||
if (replay.status === 'unavailable') {
|
||||
cleanup();
|
||||
return reply
|
||||
.code(409)
|
||||
.type('application/problem+json')
|
||||
.send(
|
||||
problem(
|
||||
request,
|
||||
409,
|
||||
'EVENT_POSITION_UNAVAILABLE',
|
||||
'The requested event position is no longer available.',
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
reply.hijack();
|
||||
const response = reply.raw;
|
||||
stream.response = response;
|
||||
response.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream; charset=utf-8',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
'X-Request-Id': request.id,
|
||||
});
|
||||
response.flushHeaders();
|
||||
response.on('close', cleanup);
|
||||
response.on('error', cleanup);
|
||||
|
||||
const replayedSequences = new Set<number>();
|
||||
for (const event of replay.events) {
|
||||
replayedSequences.add(event.sequence);
|
||||
writeEvent(response, event, cleanup);
|
||||
}
|
||||
for (const event of buffered) {
|
||||
if (!replayedSequences.has(event.sequence)) writeEvent(response, event, cleanup);
|
||||
}
|
||||
buffered.length = 0;
|
||||
live = true;
|
||||
return reply;
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user