Fix up append mode writing

PostHog · Feb 21, 2024 · 89dbff9 · 89dbff9
1 parent b5cf116
commit 89dbff9
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/plugin-server/src/main/ingestion-queues/session-recording/services/session-manager-v3.ts b/plugin-server/src/main/ingestion-queues/session-recording/services/session-manager-v3.ts
@@ -420,7 +420,14 @@ export class SessionManagerV3 {
         const writeStream = new PassThrough()
 
         // The uncompressed file which we need for realtime playback
-        pipeline(writeStream, createWriteStream(file)).catch((error) => {
+        pipeline(
+            writeStream,
+            createWriteStream(file, {
+                // Opens in append mode in case it already exists
+                flags: 'a',
+                encoding: 'utf-8',
+            })
+        ).catch((error) => {
             // TODO: If this actually happens we probably want to destroy the buffer as we will be stuck...
             status.error('🧨', '[session-manager] writestream errored', {
                 ...this.context,

diff --git a/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer-v3.ts b/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer-v3.ts
@@ -148,7 +148,7 @@ export class SessionRecordingIngesterV3 {
 
     public async consume(event: IncomingRecordingMessage): Promise<void> {
         const { team_id, session_id } = event
-        const key = `${team_id}-${session_id}`
+        const key = `${team_id}__${session_id}`
 
         const { offset, partition } = event.metadata
         if (this.debugPartition === partition) {