Add SSE id: and retry: field support to RiverEmitter

.gitignore

@@ -57,4 +57,7 @@ coverage/
 Thumbs.db
 
 # autobahn reports
-reports/
+reports/
+
+# Lock files
+package-lock.json

src/server/server.ts

@@ -123,8 +123,12 @@ export class RiverEmitter<T extends EventMap> {
     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<T extends EventMap> {
       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<T extends EventMap> {
       }
       // 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<T extends EventMap> {
     }
   }
 
+  /**
+   * Strips SSE framing fields (id, retry) from the payload and returns them separately.
+   */
+  private extractSSEFields<K extends keyof T>(
+    payload: EmitPayload<T, K>
+  ): { id?: string; retry?: number; dataPayload: Record<string, unknown> } {
+    const { id, retry, ...dataPayload } = payload as Record<string, unknown> & { 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<T extends EventMap> {
     event_type: K,
     payload: EmitPayload<T, K> // Expects the specific payload type for the event
   ): Promise<void> {
+    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<T extends EventMap> {
    * 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: <K extends keyof T>(
         event_type: K,
         payload: EmitPayload<T, K>
       ) => Promise<void>,
-      clientId: string
+      clientId: string,
+      lastEventId?: string | null
     ) => void | Promise<void>; // Allow async setup
     clientId?: string;
     ondisconnect?: (clientId: string) => void;
     signal?: AbortSignal;
+    lastEventId?: string | null;
   }): ReadableStream<Uint8Array> {
     // Explicitly returns stream of bytes
 
@@ -443,7 +487,7 @@ export class RiverEmitter<T extends EventMap> {
         // 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.

src/types/core.ts

@@ -81,6 +81,7 @@ export type ResponseData<T, K extends keyof T> = 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 keyof T> = T[K] extends BaseEvent
-  ? Omit<T[K], 'type' | 'stream' | 'chunkSize'>
+  ? Omit<T[K], 'type' | 'stream' | 'chunkSize'> & { id?: string; retry?: number }
   : never;

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<Uint8Array>, timeoutMs = 2000): Promise<string> {
+  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();
+  });
+});