Implement s2 chunk deduping

Author: ericallam

packages/core/src/v3/apiClient/runStream.ts

@@ -259,6 +259,8 @@ export class SSEStreamSubscription implements StreamSubscription {
       // Reset retry count on successful connection
       this.retryCount = 0;
 
+      const seenIds = new Set<string>();
+
       const stream = response.body
         .pipeThrough(new TextDecoderStream())
         .pipeThrough(new EventSourceParserStream())
@@ -287,9 +289,15 @@ export class SSEStreamSubscription implements StreamSubscription {
                   for (const record of data.records) {
                     this.lastEventId = record.seq_num.toString();
 
+                    const parsedBody = safeParseJSON(record.body) as { data: unknown; id: string };
+                    if (seenIds.has(parsedBody.id)) {
+                      continue;
+                    }
+                    seenIds.add(parsedBody.id);
+
                     chunkController.enqueue({
                       id: record.seq_num.toString(),
-                      chunk: safeParseJSON(record.body),
+                      chunk: parsedBody.data,
                       timestamp: record.timestamp,
                     });
                   }

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -1,5 +1,6 @@
 import { S2, AppendRecord, BatchTransform } from "@s2-dev/streamstore";
 import { StreamsWriter } from "./types.js";
+import { nanoid } from "nanoid";
 
 export type StreamsWriterV2Options<T = any> = {
   basin: string;
@@ -140,7 +141,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
                 return;
               }
               // Convert each chunk to JSON string and wrap in AppendRecord
-              controller.enqueue(AppendRecord.make(JSON.stringify(chunk)));
+              controller.enqueue(AppendRecord.make(JSON.stringify({ data: chunk, id: nanoid(7) })));
             },
           })
         )