diff --git a/apps/dashboard/src/features/operations/live-console-model.ts b/apps/dashboard/src/features/operations/live-console-model.ts index cc5a1ff..f1ccc25 100644 --- a/apps/dashboard/src/features/operations/live-console-model.ts +++ b/apps/dashboard/src/features/operations/live-console-model.ts @@ -31,10 +31,19 @@ function boundedAppend( entry: ConsoleEntry, capacity: number, ): readonly ConsoleEntry[] { - if (entries.some(({ id }) => id === entry.id)) return entries; + if (entries.some((candidate) => isSameEntry(candidate, entry))) return entries; return [...entries, entry].slice(-capacity); } +function isSameEntry(left: ConsoleEntry, right: ConsoleEntry): boolean { + return left.id === right.id || ( + left.occurredAt === right.occurredAt && + left.level === right.level && + left.message === right.message && + searchableData(left.data) === searchableData(right.data) + ); +} + export function createLiveConsoleState(capacity = 500): LiveConsoleState { if (!Number.isInteger(capacity) || capacity <= 0) { throw new RangeError('capacity must be a positive integer'); @@ -57,8 +66,8 @@ export function reduceLiveConsoleState( switch (action.type) { case 'append': { if ( - state.entries.some(({ id }) => id === action.entry.id) || - state.pendingEntries.some(({ id }) => id === action.entry.id) + state.entries.some((entry) => isSameEntry(entry, action.entry)) || + state.pendingEntries.some((entry) => isSameEntry(entry, action.entry)) ) return state; return state.isPaused ? { ...state, pendingEntries: boundedAppend(state.pendingEntries, action.entry, state.capacity) } diff --git a/src/web/log-broadcaster.ts b/src/web/log-broadcaster.ts index 034210f..ceeff95 100644 --- a/src/web/log-broadcaster.ts +++ b/src/web/log-broadcaster.ts @@ -22,16 +22,20 @@ export class LogBroadcaster { } push(level: LogEntry['level'], message: string, data?: unknown): void { - const entry: LogEntry = { timestamp: Date.now(), level, message, data }; + const realtimeEvent = this.eventHub.closed + ? undefined + : this.eventHub.publish('log.created', { level, message, data }); + const entry: LogEntry = { + timestamp: realtimeEvent ? Date.parse(realtimeEvent.occurredAt) : Date.now(), + level, + message, + data, + }; this.entries.push(entry); if (this.entries.length > MAX_LOGS) { this.entries = this.entries.slice(-MAX_LOGS); } - if (!this.eventHub.closed) { - this.eventHub.publish('log.created', { level, message, data }); - } - this.fanout(JSON.stringify({ type: 'log', ...entry })); } diff --git a/tests/dashboard/live-console-model.test.ts b/tests/dashboard/live-console-model.test.ts index b2630a1..22087d2 100644 --- a/tests/dashboard/live-console-model.test.ts +++ b/tests/dashboard/live-console-model.test.ts @@ -38,4 +38,19 @@ describe('live console state', () => { expect(state.entries.map(({ id }) => id)).toEqual(['3', '4']); expect(state.pendingEntries).toEqual([]); }); + + it('deduplicates the same log delivered by the initial snapshot and SSE replay', () => { + let state = createLiveConsoleState(); + state = reduceLiveConsoleState(state, { + type: 'append', + entry: entry('initial-1', 'info', 'Database initialized'), + }); + state = reduceLiveConsoleState(state, { + type: 'append', + entry: entry('realtime-1', 'info', 'Database initialized'), + }); + + expect(state.entries).toHaveLength(1); + expect(state.entries[0]?.id).toBe('initial-1'); + }); }); diff --git a/tests/log-broadcaster.test.ts b/tests/log-broadcaster.test.ts index 4d914c5..16225d8 100644 --- a/tests/log-broadcaster.test.ts +++ b/tests/log-broadcaster.test.ts @@ -4,12 +4,21 @@ import { LogBroadcaster } from '../src/web/log-broadcaster.js'; describe('LogBroadcaster realtime bridge', () => { it('publishes each log through the typed event hub', () => { - const hub = new RealtimeEventHub({ capacity: 10 }); + const occurredAt = new Date('2026-08-08T15:00:00.000Z'); + const hub = new RealtimeEventHub({ + capacity: 10, + clock: { now: () => occurredAt }, + }); const received: RealtimeEvent[] = []; + let recent: unknown; hub.subscribe((event) => received.push(event)); const broadcaster = new LogBroadcaster(hub); broadcaster.push('warn', 'Connection is stale', { attempts: 2 }); + broadcaster.recent( + { query: {} } as never, + { json: (value: unknown) => { recent = value; } } as never, + ); expect(received).toEqual([ expect.objectContaining({ @@ -21,5 +30,8 @@ describe('LogBroadcaster realtime bridge', () => { }, }), ]); + expect(recent).toEqual([ + expect.objectContaining({ timestamp: occurredAt.getTime() }), + ]); }); });