fix: harden SSE stream parsing

Author: NianJiuZst

src/client/stream.ts

@@ -10,6 +10,41 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
 
   const decoder = new TextDecoder();
   let buffer = '';
+  let event: Partial<ServerSentEvent> = {};
+
+  const processLine = (rawLine: string): ServerSentEvent | undefined => {
+    const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine;
+
+    if (line === '') {
+      const completed = event.data !== undefined
+        ? { data: event.data, event: event.event, id: event.id }
+        : undefined;
+      event = {};
+      return completed;
+    }
+
+    if (line.startsWith(':')) return undefined;
+
+    const colonIndex = line.indexOf(':');
+    if (colonIndex === -1) return undefined;
+
+    const field = line.slice(0, colonIndex);
+    const value = line.slice(colonIndex + 1).trimStart();
+
+    switch (field) {
+      case 'data':
+        event.data = event.data !== undefined ? `${event.data}\n${value}` : value;
+        break;
+      case 'event':
+        event.event = value;
+        break;
+      case 'id':
+        event.id = value;
+        break;
+    }
+
+    return undefined;
+  };
 
   try {
     while (true) {
@@ -21,45 +56,25 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
       const lines = buffer.split('\n');
       buffer = lines.pop() || '';
 
-      let event: Partial<ServerSentEvent> = {};
-
       for (const line of lines) {
-        if (line === '') {
-          if (event.data !== undefined) {
-            yield { data: event.data, event: event.event, id: event.id };
-          }
-          event = {};
-          continue;
+        const completed = processLine(line);
+        if (completed) {
+          yield completed;
         }
+      }
+    }
 
-        if (line.startsWith(':')) continue; // comment
-
-        const colonIndex = line.indexOf(':');
-        if (colonIndex === -1) continue;
-
-        const field = line.slice(0, colonIndex);
-        const value = line.slice(colonIndex + 1).trimStart();
+    buffer += decoder.decode();
 
-        switch (field) {
-          case 'data':
-            event.data = event.data !== undefined ? `${event.data}\n${value}` : value;
-            break;
-          case 'event':
-            event.event = value;
-            break;
-          case 'id':
-            event.id = value;
-            break;
-        }
+    if (buffer.length > 0) {
+      const completed = processLine(buffer);
+      if (completed) {
+        yield completed;
       }
     }
 
-    // Flush remaining
-    if (buffer.trim() && buffer.includes('data:')) {
-      const colonIndex = buffer.indexOf(':');
-      if (colonIndex !== -1) {
-        yield { data: buffer.slice(colonIndex + 1).trimStart() };
-      }
+    if (event.data !== undefined) {
+      yield { data: event.data, event: event.event, id: event.id };
     }
   } finally {
     reader.releaseLock();

test/client/stream.test.ts

@@ -241,6 +241,45 @@ describe('parseSSE', () => {
     expect(events2[1].data).toBe('[DONE]');
   });
 
+  it('preserves event fields split across network chunks', async () => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        const encoder = new TextEncoder();
+        controller.enqueue(encoder.encode('id: 7\nevent: mes'));
+        controller.enqueue(encoder.encode('sage\ndata: hel'));
+        controller.enqueue(encoder.encode('lo\n\n'));
+        controller.close();
+      },
+    });
+
+    const events = await collectEvents(new Response(body));
+
+    expect(events).toEqual([{ id: '7', event: 'message', data: 'hello' }]);
+  });
+
+  it('preserves multi-line data split across network chunks', async () => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        const encoder = new TextEncoder();
+        controller.enqueue(encoder.encode('data: first\n'));
+        controller.enqueue(encoder.encode('data: sec'));
+        controller.enqueue(encoder.encode('ond\n\n'));
+        controller.close();
+      },
+    });
+
+    const events = await collectEvents(new Response(body));
+
+    expect(events).toEqual([{ data: 'first\nsecond' }]);
+  });
+
+  it('handles CRLF line endings', async () => {
+    const body = 'event: message\r\ndata: hello\r\n\r\n';
+    const events = await collectEvents(new Response(body));
+
+    expect(events).toEqual([{ event: 'message', data: 'hello' }]);
+  });
+
   // -------------------------------------------------------------------------
   // Helpers
   // -------------------------------------------------------------------------