diff --git a/apps/api/src/application/events/event-journal.test.ts b/apps/api/src/application/events/event-journal.test.ts new file mode 100644 index 0000000..31f4732 --- /dev/null +++ b/apps/api/src/application/events/event-journal.test.ts @@ -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); + }); +}); diff --git a/apps/api/src/application/events/event-journal.ts b/apps/api/src/application/events/event-journal.ts new file mode 100644 index 0000000..12cb63e --- /dev/null +++ b/apps/api/src/application/events/event-journal.ts @@ -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; + 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 = { + 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(); + 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 }; + } +} diff --git a/apps/api/src/application/status/fleet-status-coordinator.test.ts b/apps/api/src/application/status/fleet-status-coordinator.test.ts new file mode 100644 index 0000000..5976106 --- /dev/null +++ b/apps/api/src/application/status/fleet-status-coordinator.test.ts @@ -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((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(); + }); +}); diff --git a/apps/api/src/application/status/fleet-status-coordinator.ts b/apps/api/src/application/status/fleet-status-coordinator.ts new file mode 100644 index 0000000..0640254 --- /dev/null +++ b/apps/api/src/application/status/fleet-status-coordinator.ts @@ -0,0 +1,110 @@ +import { StatusPollScheduler, type StatusPollSchedulerOptions } from './status-poll-scheduler.js'; + +export interface FleetStatusSource { + listEnabled(): Promise; +} + +export interface FleetStatusSnapshots { + refreshHealth(instanceId: string): Promise; +} + +export interface FleetStatusScheduler { + add(instanceId: string, intervalMs?: number): void; + remove(instanceId: string): boolean; + start(): void; + stop(): Promise; +} + +export type FleetStatusSchedulerFactory = ( + options: Pick, +) => 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(); + #started = false; + #stopped = false; + #startPromise: Promise | undefined; + #reconcileTail: Promise = Promise.resolve(); + #stopPromise?: Promise; + + 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 { + 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 { + 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 { + 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; + } +} diff --git a/apps/api/src/application/status/status-snapshot-service.test.ts b/apps/api/src/application/status/status-snapshot-service.test.ts new file mode 100644 index 0000000..c3d7566 --- /dev/null +++ b/apps/api/src/application/status/status-snapshot-service.test.ts @@ -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 | undefined; + +const payload = (db: Database.Database) => + JSON.parse(String(row(db)?.payload_json)) as Record; + +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((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>({ 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(); + }); +}); diff --git a/apps/api/src/application/status/status-snapshot-service.ts b/apps/api/src/application/status/status-snapshot-service.ts new file mode 100644 index 0000000..4ec67cc --- /dev/null +++ b/apps/api/src/application/status/status-snapshot-service.ts @@ -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 }; + readonly transport: { get(url: string): Promise }; + 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; + 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(); + #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 { + 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, + ); + } +} diff --git a/apps/api/src/control-plane.test.ts b/apps/api/src/control-plane.test.ts index 31fb78c..780ea57 100644 --- a/apps/api/src/control-plane.test.ts +++ b/apps/api/src/control-plane.test.ts @@ -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'); diff --git a/apps/api/src/control-plane.ts b/apps/api/src/control-plane.ts index dadc983..c2ea4ed 100644 --- a/apps/api/src/control-plane.ts +++ b/apps/api/src/control-plane.ts @@ -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; } export interface ControlPlaneApp extends FastifyInstance { retryPendingSecretCleanup(): Promise; } 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, { diff --git a/apps/api/src/infrastructure/database/database.test.ts b/apps/api/src/infrastructure/database/database.test.ts index 5657841..56a9f10 100644 --- a/apps/api/src/infrastructure/database/database.test.ts +++ b/apps/api/src/infrastructure/database/database.test.ts @@ -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', ]), diff --git a/apps/api/src/infrastructure/database/migrations.ts b/apps/api/src/infrastructure/database/migrations.ts index d4611a6..f82efa6 100644 --- a/apps/api/src/infrastructure/database/migrations.ts +++ b/apps/api/src/infrastructure/database/migrations.ts @@ -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 ( diff --git a/apps/api/src/interface/http/event-routes.test.ts b/apps/api/src/interface/http/event-routes.test.ts new file mode 100644 index 0000000..5722e0c --- /dev/null +++ b/apps/api/src/interface/http/event-routes.test.ts @@ -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[] = []; +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((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((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((resolve, reject) => { + http + .get(`${address}/api/v1/events`, (response) => { + response.destroy(); + resolve(); + }) + .on('error', reject); + }); + await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledOnce()); + }); +}); diff --git a/apps/api/src/interface/http/event-routes.ts b/apps/api/src/interface/http/event-routes.ts new file mode 100644 index 0000000..a658526 --- /dev/null +++ b/apps/api/src/interface/http/event-routes.ts @@ -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 => ({ + 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(); + 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; + }); +}