diff --git a/src/routes/api/events/servers/+server.ts b/src/routes/api/events/servers/+server.ts index c4604af..6d5f74d 100644 --- a/src/routes/api/events/servers/+server.ts +++ b/src/routes/api/events/servers/+server.ts @@ -1,11 +1,14 @@ import { observationReader } from '$lib/server/fleet/status.js'; import { listServers } from '$lib/server/fleet/servers.js'; import { statusEntry } from '$lib/server/pipeline/worker.js'; -import { subscribe } from '$lib/server/pipeline/bus.js'; +import { subscribe, type PipelineEvent } from '$lib/server/pipeline/bus.js'; import { listTelemetry } from '$lib/server/fleet/telemetry.js'; const encoder = new TextEncoder(); const HEARTBEAT_MS = 20_000; +// Bus subscribers receive the same event object. Reuse its complete frame +// across connections without retaining events after delivery. +const frames = new WeakMap(); export async function GET() { let heartbeat: NodeJS.Timeout | null = null; @@ -15,7 +18,12 @@ export async function GET() { start(controller) { unsubscribe = subscribe((event) => { if (event.type === 'server-status' || event.type === 'server-telemetry') { - controller.enqueue(encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`)); + let frame = frames.get(event); + if (!frame) { + frame = encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify({ ...event, serverNow: Date.now() })}\n\n`); + frames.set(event, frame); + } + controller.enqueue(frame); } }); const observation = observationReader(); diff --git a/tests/unit/telemetry.test.ts b/tests/unit/telemetry.test.ts index 915894b..6739557 100644 --- a/tests/unit/telemetry.test.ts +++ b/tests/unit/telemetry.test.ts @@ -90,6 +90,45 @@ it('timestamps initial telemetry and subsequent status and telemetry events with } }); +it('serializes each live server event once across connections and releases cancelled subscribers', async () => { + const { GET } = await import('../../src/routes/api/events/servers/+server.js'); + const { emit } = await import('../../src/lib/server/pipeline/bus.js'); + const readers = await Promise.all([GET(), GET()].map(async response => (await response).body!.getReader())); + const stringify = vi.spyOn(JSON, 'stringify'); + try { + for (const reader of readers) { + await reader.read(); // Initial comment and per-connection snapshots. + await reader.read(); + await reader.read(); + } + for (const event of [ + { type: 'server-status' as const, delta: true, entries: [] }, + { type: 'server-telemetry' as const, entries: [{ ...payload, receivedAt: new Date().toISOString() }] } + ]) { + stringify.mockClear(); + emit(event); + expect(stringify).toHaveBeenCalledTimes(1); + const [first, second] = await Promise.all(readers.map(reader => reader.read())); + expect(first.value).toBe(second.value); + const frame = new TextDecoder().decode(first.value); + expect(frame.startsWith(`event: ${event.type}\ndata: `)).toBe(true); + expect(JSON.parse(frame.split('\n')[1].slice(6))).toEqual({ ...event, serverNow: expect.any(Number) }); + } + await readers[0].cancel(); + stringify.mockClear(); + emit({ type: 'server-status', entries: [] }); + expect(stringify).toHaveBeenCalledTimes(1); + expect((await readers[1].read()).done).toBe(false); + await readers[1].cancel(); + stringify.mockClear(); + emit({ type: 'server-telemetry', entries: [] }); + expect(stringify).not.toHaveBeenCalled(); + } finally { + stringify.mockRestore(); + await Promise.all(readers.map(reader => reader.cancel())); + } +}); + it.each(['round_change', 'heartbeat'] as const)('persists %s stats, broadcasts updates, and leaves RCON safety data untouched', async (event) => { const payload = gameEvent(event, 17); const { subscribe } = await import('../../src/lib/server/pipeline/bus.js');