fix: file pipeline status reporting and websocket updates (#589)

* feat: use file pipeline for upload and reprocess action

* fix: make file pipeline correctly report status events

* fix: duplication of transcripts_controller

* fix: tests

* test: fix file upload test

* test: fix reprocess

* fix: also patch from main_file_pipeline

(how patch is done is dependent of file import unfortunately)

server/reflector/pipelines/main_file_pipeline.py

@@ -15,10 +15,15 @@ from celery import shared_task
 
 from reflector.db.transcripts import (
     Transcript,
+    TranscriptStatus,
     transcripts_controller,
 )
 from reflector.logger import logger
-from reflector.pipelines.main_live_pipeline import PipelineMainBase, asynctask
+from reflector.pipelines.main_live_pipeline import (
+    PipelineMainBase,
+    asynctask,
+    broadcast_to_sockets,
+)
 from reflector.processors import (
     AudioFileWriterProcessor,
     TranscriptFinalSummaryProcessor,
@@ -83,12 +88,27 @@ class PipelineMainFile(PipelineMainBase):
                 exc_info=result,
             )
 
+    @broadcast_to_sockets
+    async def set_status(self, transcript_id: str, status: TranscriptStatus):
+        async with self.lock_transaction():
+            return await transcripts_controller.set_status(transcript_id, status)
+
     async def process(self, file_path: Path):
         """Main entry point for file processing"""
         self.logger.info(f"Starting file pipeline for {file_path}")
 
         transcript = await self.get_transcript()
 
+        # Clear transcript as we're going to regenerate everything
+        async with self.transaction():
+            await transcripts_controller.update(
+                transcript,
+                {
+                    "events": [],
+                    "topics": [],
+                },
+            )
+
         # Extract audio and write to transcript location
         audio_path = await self.extract_and_write_audio(file_path, transcript)
 
@@ -105,6 +125,8 @@ class PipelineMainFile(PipelineMainBase):
 
         self.logger.info("File pipeline complete")
 
+        await transcripts_controller.set_status(transcript.id, "ended")
+
     async def extract_and_write_audio(
         self, file_path: Path, transcript: Transcript
     ) -> Path:
@@ -362,14 +384,21 @@ async def task_pipeline_file_process(*, transcript_id: str):
     if not transcript:
         raise Exception(f"Transcript {transcript_id} not found")
 
-    # Find the file to process
-    audio_file = next(transcript.data_path.glob("upload.*"), None)
-    if not audio_file:
-        audio_file = next(transcript.data_path.glob("audio.*"), None)
-
-    if not audio_file:
-        raise Exception("No audio file found to process")
-
-    # Run file pipeline
     pipeline = PipelineMainFile(transcript_id=transcript_id)
-    await pipeline.process(audio_file)
+
+    try:
+        await pipeline.set_status(transcript_id, "processing")
+
+        # Find the file to process
+        audio_file = next(transcript.data_path.glob("upload.*"), None)
+        if not audio_file:
+            audio_file = next(transcript.data_path.glob("audio.*"), None)
+
+        if not audio_file:
+            raise Exception("No audio file found to process")
+
+        await pipeline.process(audio_file)
+
+    except Exception:
+        await pipeline.set_status(transcript_id, "error")
+        raise

server/reflector/pipelines/main_live_pipeline.py

@@ -32,6 +32,7 @@ from reflector.db.transcripts import (
     TranscriptFinalLongSummary,
     TranscriptFinalShortSummary,
     TranscriptFinalTitle,
+    TranscriptStatus,
     TranscriptText,
     TranscriptTopic,
     TranscriptWaveform,
@@ -188,8 +189,15 @@ class PipelineMainBase(PipelineRunner[PipelineMessage], Generic[PipelineMessage]
         ]
 
     @asynccontextmanager
-    async def transaction(self):
+    async def lock_transaction(self):
+        # This lock is to prevent multiple processor starting adding
+        # into event array at the same time
         async with self._lock:
+            yield
+
+    @asynccontextmanager
+    async def transaction(self):
+        async with self.lock_transaction():
             async with transcripts_controller.transaction():
                 yield
 
@@ -198,14 +206,14 @@ class PipelineMainBase(PipelineRunner[PipelineMessage], Generic[PipelineMessage]
         # if it's the first part, update the status of the transcript
         # but do not set the ended status yet.
         if isinstance(self, PipelineMainLive):
-            status_mapping = {
+            status_mapping: dict[str, TranscriptStatus] = {
                 "started": "recording",
                 "push": "recording",
                 "flush": "processing",
                 "error": "error",
             }
         elif isinstance(self, PipelineMainFinalSummaries):
-            status_mapping = {
+            status_mapping: dict[str, TranscriptStatus] = {
                 "push": "processing",
                 "flush": "processing",
                 "error": "error",
@@ -221,22 +229,8 @@ class PipelineMainBase(PipelineRunner[PipelineMessage], Generic[PipelineMessage]
             return
 
         # when the status of the pipeline changes, update the transcript
-        async with self.transaction():
-            transcript = await self.get_transcript()
-            if status == transcript.status:
-                return
-            resp = await transcripts_controller.append_event(
-                transcript=transcript,
-                event="STATUS",
-                data=StrValue(value=status),
-            )
-            await transcripts_controller.update(
-                transcript,
-                {
-                    "status": status,
-                },
-            )
-            return resp
+        async with self._lock:
+            return await transcripts_controller.set_status(self.transcript_id, status)
 
     @broadcast_to_sockets
     async def on_transcript(self, data):