diff --git a/.gitignore b/.gitignore index 63f0c6c..2dc7733 100644 --- a/.gitignore +++ b/.gitignore @@ -57,4 +57,7 @@ coverage/ Thumbs.db # autobahn reports -reports/ \ No newline at end of file +reports/ + +# Lock files +package-lock.json diff --git a/src/server/server.ts b/src/server/server.ts index 5ca109c..749657b 100644 --- a/src/server/server.ts +++ b/src/server/server.ts @@ -123,8 +123,12 @@ export class RiverEmitter { const event_config = this.eventDefinitions[event_type]; const chunk_size = event_config?.chunkSize ?? 1024; // Default chunk size + // Extract SSE framing fields before processing data + const { id, retry, dataPayload } = this.extractSSEFields(payload); + const prefix = this.buildSSEPrefix(id, retry); + // Extract data from payload for streaming - streaming events must have data property - const data = (payload as any).data; + const data = (dataPayload as any).data; if (data === undefined) { throw new Error(`Stream event ${String(event_type)} requires a 'data' property`); } @@ -140,9 +144,9 @@ export class RiverEmitter { for await (const item of iterable) { chunk.push(item); if (chunk.length >= chunk_size) { - // Create payload with chunked data, preserving other properties - const chunkPayload = { ...payload, data: chunk } as any; - const event_data = `event: ${String( + // Create payload with chunked data, preserving other properties (excluding SSE fields) + const chunkPayload = { ...dataPayload, data: chunk } as any; + const event_data = `${prefix}event: ${String( event_type )}\ndata: ${JSON.stringify(chunkPayload)}\n\n`; writeSuccess = await this.writeChunk(writer, event_data); @@ -152,9 +156,9 @@ export class RiverEmitter { } // Send any remaining items after the loop finishes (if write hasn't failed) if (writeSuccess && chunk.length > 0) { - // Create payload with remaining chunked data, preserving other properties - const finalPayload = { ...payload, data: chunk } as any; - const event_data = `event: ${String( + // Create payload with remaining chunked data, preserving other properties (excluding SSE fields) + const finalPayload = { ...dataPayload, data: chunk } as any; + const event_data = `${prefix}event: ${String( event_type )}\ndata: ${JSON.stringify(finalPayload)}\n\n`; await this.writeChunk(writer, event_data); @@ -169,6 +173,40 @@ export class RiverEmitter { } } + /** + * Strips SSE framing fields (id, retry) from the payload and returns them separately. + */ + private extractSSEFields( + payload: EmitPayload + ): { id?: string; retry?: number; dataPayload: Record } { + const { id, retry, ...dataPayload } = payload as Record & { id?: string; retry?: number }; + return { id, retry, dataPayload }; + } + + /** + * Sanitizes an SSE id field to ensure it does not contain CR or LF characters. + * See: https://html.spec.whatwg.org/multipage/server-sent-events.html + */ + private sanitizeSSEId(id: string): string { + // Remove CR and LF characters which would break SSE framing. + return id.replace(/[\r\n]/g, ''); + } + + /** + * Builds the SSE prefix string for id: and retry: fields. + */ + private buildSSEPrefix(id?: string, retry?: number): string { + let prefix = ''; + if (id !== undefined) { + const safeId = this.sanitizeSSEId(String(id)); + prefix += `id: ${safeId}\n`; + } + if (retry !== undefined) { + prefix += `retry: ${retry}\n`; + } + return prefix; + } + /** * Handles emitting a single event payload (stream: false or undefined). */ @@ -177,9 +215,11 @@ export class RiverEmitter { event_type: K, payload: EmitPayload // Expects the specific payload type for the event ): Promise { + const { id, retry, dataPayload } = this.extractSSEFields(payload); + const prefix = this.buildSSEPrefix(id, retry); // Send the structured payload, correctly JSON stringified. - const event_data = `event: ${String(event_type)}\ndata: ${JSON.stringify( - payload + const event_data = `${prefix}event: ${String(event_type)}\ndata: ${JSON.stringify( + dataPayload )}\n\n`; await this.writeChunk(writer, event_data); } @@ -231,27 +271,31 @@ export class RiverEmitter { * Creates a ReadableStream for a new SSE connection. * The stream outputs Uint8Array chunks representing the SSE message payload. * @param options - Configuration for this specific stream connection. - * @param options.callback - Function executed when the connection starts. Receives an `emit` function scoped to this client and the `clientId`. + * @param options.callback - Function executed when the connection starts. Receives an `emit` function scoped to this client, the `clientId`, and the optional `lastEventId`. * @param options.clientId - Optional custom client ID. If not provided, a random one is generated. * @param options.ondisconnect - Optional callback executed when this client disconnects. * @param options.signal - Optional AbortSignal to link stream lifecycle to an external signal (e.g., HTTP request). + * @param options.lastEventId - Optional Last-Event-ID header value from the client for reconnection support. */ public stream({ callback, clientId: customClientId, ondisconnect, - signal + signal, + lastEventId }: { callback: ( emit: ( event_type: K, payload: EmitPayload ) => Promise, - clientId: string + clientId: string, + lastEventId?: string | null ) => void | Promise; // Allow async setup clientId?: string; ondisconnect?: (clientId: string) => void; signal?: AbortSignal; + lastEventId?: string | null; }): ReadableStream { // Explicitly returns stream of bytes @@ -443,7 +487,7 @@ export class RiverEmitter { // Execute the user's setup callback try { console.log(`RiverEmitter: Client ${clientId} connected.`); - await callback(emit, clientId); + await callback(emit, clientId, lastEventId); // If the callback completing naturally means the stream should end, // you might close the writer here: // await writer.close(); // This would trigger the pipe close/cleanup path. diff --git a/src/types/core.ts b/src/types/core.ts index 468ca8f..1c4a736 100644 --- a/src/types/core.ts +++ b/src/types/core.ts @@ -81,6 +81,7 @@ export type ResponseData = T[K] extends BaseEvent : never; // Type to extract only user-defined properties (excluding stream and chunkSize) +// Also allows optional SSE framing fields (id, retry) that are stripped before data serialization export type EmitPayload = T[K] extends BaseEvent - ? Omit + ? Omit & { id?: string; retry?: number } : never; diff --git a/tests/core/sse-fields.test.ts b/tests/core/sse-fields.test.ts new file mode 100644 index 0000000..84250e3 --- /dev/null +++ b/tests/core/sse-fields.test.ts @@ -0,0 +1,176 @@ +// tests/core/sse-fields.test.ts +import { RiverEvents } from '../../src'; +import { RiverEmitter } from '../../src/server'; +import { describe, it, expect } from 'bun:test'; + +const events = new RiverEvents() + .defineEvent('message', { message: '' as string }) + .defineEvent('data_event', { data: { childId: '' as string } }) + .defineEvent('stream_event', { + stream: true, + data: [] as number[], + chunkSize: 3 + }) + .build(); + +/** Helper to read SSE text from a RiverEmitter stream with a timeout. */ +async function readSSE(stream: ReadableStream, timeoutMs = 2000): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + const chunks: string[] = []; + + const readWithTimeout = () => + Promise.race([ + reader.read(), + new Promise<{ value: undefined; done: true }>((resolve) => + setTimeout(() => resolve({ value: undefined, done: true }), timeoutMs) + ) + ]); + + for (let i = 0; i < 20; i++) { + const { value, done } = await readWithTimeout(); + if (done) break; + if (value) chunks.push(decoder.decode(value)); + } + + reader.cancel(); + return chunks.join(''); +} + +describe('SSE id: and retry: fields', () => { + it('should include id: and retry: fields in single events', async () => { + const emitter = RiverEmitter.init(events); + const stream = emitter.stream({ + callback: async (emit) => { + await emit('data_event', { + data: { childId: 'test123' }, + id: '019abc12-3456-7890', + retry: 3000 + }); + } + }); + + const output = await readSSE(stream); + + expect(output).toContain('id: 019abc12-3456-7890\n'); + expect(output).toContain('retry: 3000\n'); + expect(output).toContain('event: data_event\n'); + + // data should NOT include id or retry + const dataMatch = output.match(/data: (.+)\n/); + expect(dataMatch).not.toBeNull(); + const parsed = JSON.parse(dataMatch![1]); + expect(parsed.id).toBeUndefined(); + expect(parsed.retry).toBeUndefined(); + expect(parsed.data.childId).toBe('test123'); + }); + + it('should omit id: and retry: lines when not provided (backward compat)', async () => { + const emitter = RiverEmitter.init(events); + const stream = emitter.stream({ + callback: async (emit) => { + await emit('data_event', { data: { childId: 'test456' } }); + } + }); + + const output = await readSSE(stream); + + expect(output).not.toContain('id:'); + expect(output).not.toContain('retry:'); + expect(output).toStartWith('event: data_event\n'); + }); + + it('should include only id: when retry is not provided', async () => { + const emitter = RiverEmitter.init(events); + const stream = emitter.stream({ + callback: async (emit) => { + await emit('data_event', { data: { childId: 'test789' }, id: 'evt-001' }); + } + }); + + const output = await readSSE(stream); + + expect(output).toContain('id: evt-001\n'); + expect(output).not.toContain('retry:'); + expect(output).toStartWith('id: evt-001\n'); + }); + + it('should include id: and retry: fields in stream events', async () => { + const emitter = RiverEmitter.init(events); + const stream = emitter.stream({ + callback: async (emit) => { + await emit('stream_event', { + data: [1, 2, 3, 4, 5], + id: 'stream-001', + retry: 5000 + }); + } + }); + + const output = await readSSE(stream); + + expect(output).toContain('id: stream-001\n'); + expect(output).toContain('retry: 5000\n'); + + // Data chunks should not contain id or retry + const dataMatches = [...output.matchAll(/data: (.+)\n/g)]; + expect(dataMatches.length).toBeGreaterThan(0); + for (const match of dataMatches) { + const parsed = JSON.parse(match[1]); + expect(parsed.id).toBeUndefined(); + expect(parsed.retry).toBeUndefined(); + } + }); +}); + +describe('lastEventId support', () => { + it('should pass lastEventId to the callback', async () => { + const emitter = RiverEmitter.init(events); + let receivedLastEventId: string | null | undefined; + + const stream = emitter.stream({ + callback: async (emit, clientId, lastEventId) => { + receivedLastEventId = lastEventId; + await emit('message', { message: 'hello' }); + }, + lastEventId: 'last-evt-42' + }); + + await readSSE(stream); + + expect(receivedLastEventId).toBe('last-evt-42'); + }); + + it('should pass undefined lastEventId when not provided', async () => { + const emitter = RiverEmitter.init(events); + let receivedLastEventId: string | null | undefined = 'SENTINEL'; + + const stream = emitter.stream({ + callback: async (emit, clientId, lastEventId) => { + receivedLastEventId = lastEventId; + await emit('message', { message: 'hello' }); + } + }); + + await readSSE(stream); + + expect(receivedLastEventId).toBeUndefined(); + }); + + it('should pass null lastEventId when explicitly null', async () => { + const emitter = RiverEmitter.init(events); + let receivedLastEventId: string | null | undefined = 'SENTINEL'; + + const stream = emitter.stream({ + callback: async (emit, clientId, lastEventId) => { + receivedLastEventId = lastEventId; + await emit('message', { message: 'hello' }); + }, + lastEventId: null + }); + + await readSSE(stream); + + expect(receivedLastEventId).toBeNull(); + }); +});