chore(main): release 0.8.2 (#595)

* fix: search-logspam

* llm comment

* fix tests

---------

Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>

.github/workflows/db_migrations.yml

@@ -2,6 +2,8 @@ name: Test Database Migrations
 
 on:
   push:
+    branches:
+      - main
     paths:
       - "server/migrations/**"
       - "server/reflector/db/**"
@@ -17,6 +19,9 @@ on:
 jobs:
   test-migrations:
     runs-on: ubuntu-latest
+    concurrency:
+      group: db-ubuntu-latest-${{ github.ref }}
+      cancel-in-progress: true
     services:
       postgres:
         image: postgres:17

.github/workflows/test_server.yml

@@ -5,12 +5,17 @@ on:
     paths:
       - "server/**"
   push:
+    branches:
+      - main
     paths:
       - "server/**"
 
 jobs:
   pytest:
     runs-on: ubuntu-latest
+    concurrency:
+      group: pytest-${{ github.ref }}
+      cancel-in-progress: true
     services:
       redis:
         image: redis:6
@@ -30,6 +35,9 @@ jobs:
 
   docker-amd64:
     runs-on: linux-amd64
+    concurrency:
+      group: docker-amd64-${{ github.ref }}
+      cancel-in-progress: true
     steps:
       - uses: actions/checkout@v4
       - name: Set up Docker Buildx
@@ -45,6 +53,9 @@ jobs:
 
   docker-arm64:
     runs-on: linux-arm64
+    concurrency:
+      group: docker-arm64-${{ github.ref }}
+      cancel-in-progress: true
     steps:
       - uses: actions/checkout@v4
       - name: Set up Docker Buildx

.gitignore

@@ -14,4 +14,6 @@ data/
 www/REFACTOR.md
 www/reload-frontend
 server/test.sqlite
-CLAUDE.local.md
+CLAUDE.local.md
+www/.env.development
+www/.env.production

.gitleaksignore

@@ -0,0 +1 @@
+b9d891d3424f371642cb032ecfd0e2564470a72c:server/tests/test_transcripts_recording_deletion.py:generic-api-key:15

.pre-commit-config.yaml

@@ -27,3 +27,8 @@ repos:
         files: ^server/
       - id: ruff-format
         files: ^server/
+
+  - repo: https://github.com/gitleaks/gitleaks
+    rev: v8.28.0
+    hooks:
+      - id: gitleaks

CHANGELOG.md

@@ -1,5 +1,42 @@
 # Changelog
 
+## [0.8.2](https://github.com/Monadical-SAS/reflector/compare/v0.8.1...v0.8.2) (2025-08-29)
+
+
+### Bug Fixes
+
+* search-logspam ([#593](https://github.com/Monadical-SAS/reflector/issues/593)) ([695d1a9](https://github.com/Monadical-SAS/reflector/commit/695d1a957d4cd862753049f9beed88836cabd5ab))
+
+## [0.8.1](https://github.com/Monadical-SAS/reflector/compare/v0.8.0...v0.8.1) (2025-08-29)
+
+
+### Bug Fixes
+
+* make webhook secret/url allowing null ([#590](https://github.com/Monadical-SAS/reflector/issues/590)) ([84a3812](https://github.com/Monadical-SAS/reflector/commit/84a381220bc606231d08d6f71d4babc818fa3c75))
+
+## [0.8.0](https://github.com/Monadical-SAS/reflector/compare/v0.7.3...v0.8.0) (2025-08-29)
+
+
+### Features
+
+* **cleanup:** add automatic data retention for public instances ([#574](https://github.com/Monadical-SAS/reflector/issues/574)) ([6f0c7c1](https://github.com/Monadical-SAS/reflector/commit/6f0c7c1a5e751713366886c8e764c2009e12ba72))
+* **rooms:** add webhook for transcript completion ([#578](https://github.com/Monadical-SAS/reflector/issues/578)) ([88ed7cf](https://github.com/Monadical-SAS/reflector/commit/88ed7cfa7804794b9b54cad4c3facc8a98cf85fd))
+
+
+### Bug Fixes
+
+* file pipeline status reporting and websocket updates ([#589](https://github.com/Monadical-SAS/reflector/issues/589)) ([9dfd769](https://github.com/Monadical-SAS/reflector/commit/9dfd76996f851cc52be54feea078adbc0816dc57))
+* Igor/evaluation ([#575](https://github.com/Monadical-SAS/reflector/issues/575)) ([124ce03](https://github.com/Monadical-SAS/reflector/commit/124ce03bf86044c18313d27228a25da4bc20c9c5))
+* optimize parakeet transcription batching algorithm ([#577](https://github.com/Monadical-SAS/reflector/issues/577)) ([7030e0f](https://github.com/Monadical-SAS/reflector/commit/7030e0f23649a8cf6c1eb6d5889684a41ce849ec))
+
+## [0.7.3](https://github.com/Monadical-SAS/reflector/compare/v0.7.2...v0.7.3) (2025-08-22)
+
+
+### Bug Fixes
+
+* cleaned repo, and get git-leaks clean ([359280d](https://github.com/Monadical-SAS/reflector/commit/359280dd340433ba4402ed69034094884c825e67))
+* restore previous behavior on live pipeline + audio downscaler ([#561](https://github.com/Monadical-SAS/reflector/issues/561)) ([9265d20](https://github.com/Monadical-SAS/reflector/commit/9265d201b590d23c628c5f19251b70f473859043))
+
 ## [0.7.2](https://github.com/Monadical-SAS/reflector/compare/v0.7.1...v0.7.2) (2025-08-21)
 
 

README.md

@@ -1,43 +1,60 @@
 <div align="center">
+<img width="100" alt="image" src="https://github.com/user-attachments/assets/66fb367b-2c89-4516-9912-f47ac59c6a7f"/>
 
 # Reflector
 
-Reflector Audio Management and Analysis is a cutting-edge web application under development by Monadical. It utilizes AI to record meetings, providing a permanent record with transcripts, translations, and automated summaries.
+Reflector is an AI-powered audio transcription and meeting analysis platform that provides real-time transcription, speaker diarization, translation and summarization for audio content and live meetings. It works 100% with local models (whisper/parakeet, pyannote, seamless-m4t, and your local llm like phi-4).
 
-[![Tests](https://github.com/monadical-sas/reflector/actions/workflows/pytests.yml/badge.svg?branch=main&event=push)](https://github.com/monadical-sas/reflector/actions/workflows/pytests.yml)
+[![Tests](https://github.com/monadical-sas/reflector/actions/workflows/test_server.yml/badge.svg?branch=main&event=push)](https://github.com/monadical-sas/reflector/actions/workflows/test_server.yml)
 [![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](https://opensource.org/licenses/MIT)
 </div>
-
-## Screenshots
+</div>
 <table>
   <tr>
     <td>
-      <a href="https://github.com/user-attachments/assets/3a976930-56c1-47ef-8c76-55d3864309e3">
-        <img width="700" alt="image" src="https://github.com/user-attachments/assets/3a976930-56c1-47ef-8c76-55d3864309e3" />
+      <a href="https://github.com/user-attachments/assets/21f5597c-2930-4899-a154-f7bd61a59e97">
+        <img width="700" alt="image" src="https://github.com/user-attachments/assets/21f5597c-2930-4899-a154-f7bd61a59e97" />
       </a>
     </td>
     <td>
-      <a href="https://github.com/user-attachments/assets/bfe3bde3-08af-4426-a9a1-11ad5cd63b33">
-        <img width="700" alt="image" src="https://github.com/user-attachments/assets/bfe3bde3-08af-4426-a9a1-11ad5cd63b33" />
+      <a href="https://github.com/user-attachments/assets/f6b9399a-5e51-4bae-b807-59128d0a940c">
+        <img width="700" alt="image" src="https://github.com/user-attachments/assets/f6b9399a-5e51-4bae-b807-59128d0a940c" />
       </a>
     </td>
     <td>
-      <a href="https://github.com/user-attachments/assets/7b60c9d0-efe4-474f-a27b-ea13bd0fabdc">
-        <img width="700" alt="image" src="https://github.com/user-attachments/assets/7b60c9d0-efe4-474f-a27b-ea13bd0fabdc" />
+      <a href="https://github.com/user-attachments/assets/a42ce460-c1fd-4489-a995-270516193897">
+        <img width="700" alt="image" src="https://github.com/user-attachments/assets/a42ce460-c1fd-4489-a995-270516193897" />
+      </a>
+    </td>
+    <td>
+      <a href="https://github.com/user-attachments/assets/21929f6d-c309-42fe-9c11-f1299e50fbd4">
+        <img width="700" alt="image" src="https://github.com/user-attachments/assets/21929f6d-c309-42fe-9c11-f1299e50fbd4" />
       </a>
     </td>
   </tr>
 </table>
 
+## What is Reflector?
+
+Reflector is a web application that utilizes local models to process audio content, providing:
+
+- **Real-time Transcription**: Convert speech to text using [Whisper](https://github.com/openai/whisper) (multi-language) or [Parakeet](https://huggingface.co/nvidia/parakeet-tdt-0.6b-v2) (English) models
+- **Speaker Diarization**: Identify and label different speakers using [Pyannote](https://github.com/pyannote/pyannote-audio) 3.1
+- **Live Translation**: Translate audio content in real-time to many languages with [Facebook Seamless-M4T](https://github.com/facebookresearch/seamless_communication)
+- **Topic Detection & Summarization**: Extract key topics and generate concise summaries using LLMs
+- **Meeting Recording**: Create permanent records of meetings with searchable transcripts
+
+Currently we provide [modal.com](https://modal.com/) gpu template to deploy.
+
 ## Background
 
 The project architecture consists of three primary components:
 
-- **Front-End**: NextJS React project hosted on Vercel, located in `www/`.
 - **Back-End**: Python server that offers an API and data persistence, found in `server/`.
-- **GPU implementation**: Providing services such as speech-to-text transcription, topic generation, automated summaries, and translations. Most reliable option is Modal deployment
+- **Front-End**: NextJS React project hosted on Vercel, located in `www/`.
+- **GPU implementation**: Providing services such as speech-to-text transcription, topic generation, automated summaries, and translations.
 
-It also uses authentik for authentication if activated, and Vercel for deployment and configuration of the front-end.
+It also uses authentik for authentication if activated.
 
 ## Contribution Guidelines
 
@@ -72,6 +89,8 @@ Note: We currently do not have instructions for Windows users.
 
 ## Installation
 
+*Note: we're working toward better installation, theses instructions are not accurate for now*
+
 ### Frontend
 
 Start with `cd www`.

server/docs/data_retention.md

@@ -0,0 +1,95 @@
+# Data Retention and Cleanup
+
+## Overview
+
+For public instances of Reflector, a data retention policy is automatically enforced to delete anonymous user data after a configurable period (default: 7 days). This ensures compliance with privacy expectations and prevents unbounded storage growth.
+
+## Configuration
+
+### Environment Variables
+
+- `PUBLIC_MODE` (bool): Must be set to `true` to enable automatic cleanup
+- `PUBLIC_DATA_RETENTION_DAYS` (int): Number of days to retain anonymous data (default: 7)
+
+### What Gets Deleted
+
+When data reaches the retention period, the following items are automatically removed:
+
+1. **Transcripts** from anonymous users (where `user_id` is NULL):
+   - Database records
+   - Local files (audio.wav, audio.mp3, audio.json waveform)
+   - Storage files (cloud storage if configured)
+
+## Automatic Cleanup
+
+### Celery Beat Schedule
+
+When `PUBLIC_MODE=true`, a Celery beat task runs daily at 3 AM to clean up old data:
+
+```python
+# Automatically scheduled when PUBLIC_MODE=true
+"cleanup_old_public_data": {
+    "task": "reflector.worker.cleanup.cleanup_old_public_data",
+    "schedule": crontab(hour=3, minute=0),  # Daily at 3 AM
+}
+```
+
+### Running the Worker
+
+Ensure both Celery worker and beat scheduler are running:
+
+```bash
+# Start Celery worker
+uv run celery -A reflector.worker.app worker --loglevel=info
+
+# Start Celery beat scheduler (in another terminal)
+uv run celery -A reflector.worker.app beat
+```
+
+## Manual Cleanup
+
+For testing or manual intervention, use the cleanup tool:
+
+```bash
+# Delete data older than 7 days (default)
+uv run python -m reflector.tools.cleanup_old_data
+
+# Delete data older than 30 days
+uv run python -m reflector.tools.cleanup_old_data --days 30
+```
+
+Note: The manual tool uses the same implementation as the Celery worker task to ensure consistency.
+
+## Important Notes
+
+1. **User Data Deletion**: Only anonymous data (where `user_id` is NULL) is deleted. Authenticated user data is preserved.
+
+2. **Storage Cleanup**: The system properly cleans up both local files and cloud storage when configured.
+
+3. **Error Handling**: If individual deletions fail, the cleanup continues and logs errors. Failed deletions are reported in the task output.
+
+4. **Public Instance Only**: The automatic cleanup task only runs when `PUBLIC_MODE=true` to prevent accidental data loss in private deployments.
+
+## Testing
+
+Run the cleanup tests:
+
+```bash
+uv run pytest tests/test_cleanup.py -v
+```
+
+## Monitoring
+
+Check Celery logs for cleanup task execution:
+
+```bash
+# Look for cleanup task logs
+grep "cleanup_old_public_data" celery.log
+grep "Starting cleanup of old public data" celery.log
+```
+
+Task statistics are logged after each run:
+- Number of transcripts deleted
+- Number of meetings deleted
+- Number of orphaned recordings deleted
+- Any errors encountered

server/docs/webhook.md

@@ -0,0 +1,212 @@
+# Reflector Webhook Documentation
+
+## Overview
+
+Reflector supports webhook notifications to notify external systems when transcript processing is completed. Webhooks can be configured per room and are triggered automatically after a transcript is successfully processed.
+
+## Configuration
+
+Webhooks are configured at the room level with two fields:
+- `webhook_url`: The HTTPS endpoint to receive webhook notifications
+- `webhook_secret`: Optional secret key for HMAC signature verification (auto-generated if not provided)
+
+## Events
+
+### `transcript.completed`
+
+Triggered when a transcript has been fully processed, including transcription, diarization, summarization, and topic detection.
+
+### `test`
+
+A test event that can be triggered manually to verify webhook configuration.
+
+## Webhook Request Format
+
+### Headers
+
+All webhook requests include the following headers:
+
+| Header | Description | Example |
+|--------|-------------|---------|
+| `Content-Type` | Always `application/json` | `application/json` |
+| `User-Agent` | Identifies Reflector as the source | `Reflector-Webhook/1.0` |
+| `X-Webhook-Event` | The event type | `transcript.completed` or `test` |
+| `X-Webhook-Retry` | Current retry attempt number | `0`, `1`, `2`... |
+| `X-Webhook-Signature` | HMAC signature (if secret configured) | `t=1735306800,v1=abc123...` |
+
+### Signature Verification
+
+If a webhook secret is configured, Reflector includes an HMAC-SHA256 signature in the `X-Webhook-Signature` header to verify the webhook authenticity.
+
+The signature format is: `t={timestamp},v1={signature}`
+
+To verify the signature:
+1. Extract the timestamp and signature from the header
+2. Create the signed payload: `{timestamp}.{request_body}`
+3. Compute HMAC-SHA256 of the signed payload using your webhook secret
+4. Compare the computed signature with the received signature
+
+Example verification (Python):
+```python
+import hmac
+import hashlib
+
+def verify_webhook_signature(payload: bytes, signature_header: str, secret: str) -> bool:
+    # Parse header: "t=1735306800,v1=abc123..."
+    parts = dict(part.split("=") for part in signature_header.split(","))
+    timestamp = parts["t"]
+    received_signature = parts["v1"]
+
+    # Create signed payload
+    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
+
+    # Compute expected signature
+    expected_signature = hmac.new(
+        secret.encode("utf-8"),
+        signed_payload.encode("utf-8"),
+        hashlib.sha256
+    ).hexdigest()
+
+    # Compare signatures
+    return hmac.compare_digest(expected_signature, received_signature)
+```
+
+## Event Payloads
+
+### `transcript.completed` Event
+
+This event includes a convenient URL for accessing the transcript:
+- `frontend_url`: Direct link to view the transcript in the web interface
+
+```json
+{
+  "event": "transcript.completed",
+  "event_id": "transcript.completed-abc-123-def-456",
+  "timestamp": "2025-08-27T12:34:56.789012Z",
+  "transcript": {
+    "id": "abc-123-def-456",
+    "room_id": "room-789",
+    "created_at": "2025-08-27T12:00:00Z",
+    "duration": 1800.5,
+    "title": "Q3 Product Planning Meeting",
+    "short_summary": "Team discussed Q3 product roadmap, prioritizing mobile app features and API improvements.",
+    "long_summary": "The product team met to finalize the Q3 roadmap. Key decisions included...",
+    "webvtt": "WEBVTT\n\n00:00:00.000 --> 00:00:05.000\n<v Speaker 1>Welcome everyone to today's meeting...",
+    "topics": [
+      {
+        "title": "Introduction and Agenda",
+        "summary": "Meeting kickoff with agenda review",
+        "timestamp": 0.0,
+        "duration": 120.0,
+        "webvtt": "WEBVTT\n\n00:00:00.000 --> 00:00:05.000\n<v Speaker 1>Welcome everyone..."
+      },
+      {
+        "title": "Mobile App Features Discussion",
+        "summary": "Team reviewed proposed mobile app features for Q3",
+        "timestamp": 120.0,
+        "duration": 600.0,
+        "webvtt": "WEBVTT\n\n00:02:00.000 --> 00:02:10.000\n<v Speaker 2>Let's talk about the mobile app..."
+      }
+    ],
+    "participants": [
+      {
+        "id": "participant-1",
+        "name": "John Doe",
+        "speaker": "Speaker 1"
+      },
+      {
+        "id": "participant-2",
+        "name": "Jane Smith",
+        "speaker": "Speaker 2"
+      }
+    ],
+    "source_language": "en",
+    "target_language": "en",
+    "status": "completed",
+    "frontend_url": "https://app.reflector.com/transcripts/abc-123-def-456"
+  },
+  "room": {
+    "id": "room-789",
+    "name": "Product Team Room"
+  }
+}
+```
+
+### `test` Event
+
+```json
+{
+  "event": "test",
+  "event_id": "test.2025-08-27T12:34:56.789012Z",
+  "timestamp": "2025-08-27T12:34:56.789012Z",
+  "message": "This is a test webhook from Reflector",
+  "room": {
+    "id": "room-789",
+    "name": "Product Team Room"
+  }
+}
+```
+
+## Retry Policy
+
+Webhooks are delivered with automatic retry logic to handle transient failures. When a webhook delivery fails due to server errors or network issues, Reflector will automatically retry the delivery multiple times over an extended period.
+
+### Retry Mechanism
+
+Reflector implements an exponential backoff strategy for webhook retries:
+
+- **Initial retry delay**: 60 seconds after the first failure
+- **Exponential backoff**: Each subsequent retry waits approximately twice as long as the previous one
+- **Maximum retry interval**: 1 hour (backoff is capped at this duration)
+- **Maximum retry attempts**: 30 attempts total
+- **Total retry duration**: Retries continue for approximately 24 hours
+
+### How Retries Work
+
+When a webhook fails, Reflector will:
+1. Wait 60 seconds, then retry (attempt #1)
+2. If it fails again, wait ~2 minutes, then retry (attempt #2)
+3. Continue doubling the wait time up to a maximum of 1 hour between attempts
+4. Keep retrying at 1-hour intervals until successful or 30 attempts are exhausted
+
+The `X-Webhook-Retry` header indicates the current retry attempt number (0 for the initial attempt, 1 for first retry, etc.), allowing your endpoint to track retry attempts.
+
+### Retry Behavior by HTTP Status Code
+
+| Status Code | Behavior |
+|-------------|----------|
+| 2xx (Success) | No retry, webhook marked as delivered |
+| 4xx (Client Error) | No retry, request is considered permanently failed |
+| 5xx (Server Error) | Automatic retry with exponential backoff |
+| Network/Timeout Error | Automatic retry with exponential backoff |
+
+**Important Notes:**
+- Webhooks timeout after 30 seconds. If your endpoint takes longer to respond, it will be considered a timeout error and retried.
+- During the retry period (~24 hours), you may receive the same webhook multiple times if your endpoint experiences intermittent failures.
+- There is no mechanism to manually retry failed webhooks after the retry period expires.
+
+## Testing Webhooks
+
+You can test your webhook configuration before processing transcripts:
+
+```http
+POST /v1/rooms/{room_id}/webhook/test
+```
+
+Response:
+```json
+{
+  "success": true,
+  "status_code": 200,
+  "message": "Webhook test successful",
+  "response_preview": "OK"
+}
+```
+
+Or in case of failure:
+```json
+{
+  "success": false,
+  "error": "Webhook request timed out (10 seconds)"
+}
+```

server/gpu/modal_deployments/reflector_transcriber_parakeet.py

@@ -3,7 +3,7 @@ import os
 import sys
 import threading
 import uuid
-from typing import Mapping, NewType
+from typing import Generator, Mapping, NamedTuple, NewType, TypedDict
 from urllib.parse import urlparse
 
 import modal
@@ -14,10 +14,7 @@ SAMPLERATE = 16000
 UPLOADS_PATH = "/uploads"
 CACHE_PATH = "/cache"
 VAD_CONFIG = {
-    "max_segment_duration": 30.0,
-    "batch_max_files": 10,
-    "batch_max_duration": 5.0,
-    "min_segment_duration": 0.02,
+    "batch_max_duration": 30.0,
     "silence_padding": 0.5,
     "window_size": 512,
 }
@@ -25,6 +22,37 @@ VAD_CONFIG = {
 ParakeetUniqFilename = NewType("ParakeetUniqFilename", str)
 AudioFileExtension = NewType("AudioFileExtension", str)
 
+
+class TimeSegment(NamedTuple):
+    """Represents a time segment with start and end times."""
+
+    start: float
+    end: float
+
+
+class AudioSegment(NamedTuple):
+    """Represents an audio segment with timing and audio data."""
+
+    start: float
+    end: float
+    audio: any
+
+
+class TranscriptResult(NamedTuple):
+    """Represents a transcription result with text and word timings."""
+
+    text: str
+    words: list["WordTiming"]
+
+
+class WordTiming(TypedDict):
+    """Represents a word with its timing information."""
+
+    word: str
+    start: float
+    end: float
+
+
 app = modal.App("reflector-transcriber-parakeet")
 
 # Volume for caching model weights
@@ -170,12 +198,14 @@ class TranscriberParakeetLive:
                 (output,) = self.model.transcribe([padded_audio], timestamps=True)
 
         text = output.text.strip()
-        words = [
-            {
-                "word": word_info["word"] + " ",
-                "start": round(word_info["start"], 2),
-                "end": round(word_info["end"], 2),
-            }
+        words: list[WordTiming] = [
+            WordTiming(
+                # XXX the space added here is to match the output of whisper
+                # whisper add space to each words, while parakeet don't
+                word=word_info["word"] + " ",
+                start=round(word_info["start"], 2),
+                end=round(word_info["end"], 2),
+            )
             for word_info in output.timestamp["word"]
         ]
 
@@ -211,12 +241,12 @@ class TranscriberParakeetLive:
         for i, (filename, output) in enumerate(zip(filenames, outputs)):
             text = output.text.strip()
 
-            words = [
-                {
-                    "word": word_info["word"] + " ",
-                    "start": round(word_info["start"], 2),
-                    "end": round(word_info["end"], 2),
-                }
+            words: list[WordTiming] = [
+                WordTiming(
+                    word=word_info["word"] + " ",
+                    start=round(word_info["start"], 2),
+                    end=round(word_info["end"], 2),
+                )
                 for word_info in output.timestamp["word"]
             ]
 
@@ -271,7 +301,9 @@ class TranscriberParakeetFile:
             audio_array, sample_rate = librosa.load(file_path, sr=SAMPLERATE, mono=True)
             return audio_array
 
-        def vad_segment_generator(audio_array):
+        def vad_segment_generator(
+            audio_array,
+        ) -> Generator[TimeSegment, None, None]:
             """Generate speech segments using VAD with start/end sample indices"""
             vad_iterator = VADIterator(self.vad_model, sampling_rate=SAMPLERATE)
             window_size = VAD_CONFIG["window_size"]
@@ -297,107 +329,121 @@ class TranscriberParakeetFile:
                     start_time = start / float(SAMPLERATE)
                     end_time = end / float(SAMPLERATE)
 
-                    # Extract the actual audio segment
-                    audio_segment = audio_array[start:end]
-
-                    yield (start_time, end_time, audio_segment)
+                    yield TimeSegment(start_time, end_time)
                     start = None
 
             vad_iterator.reset_states()
 
-        def vad_segment_filter(segments):
-            """Filter VAD segments by duration and chunk large segments"""
-            min_dur = VAD_CONFIG["min_segment_duration"]
-            max_dur = VAD_CONFIG["max_segment_duration"]
+        def batch_speech_segments(
+            segments: Generator[TimeSegment, None, None], max_duration: int
+        ) -> Generator[TimeSegment, None, None]:
+            """
+            Input segments:
+              [0-2] [3-5] [6-8] [10-11] [12-15] [17-19] [20-22]
 
-            for start_time, end_time, audio_segment in segments:
-                segment_duration = end_time - start_time
+                                  ↓ (max_duration=10)
 
-                # Skip very small segments
-                if segment_duration < min_dur:
+              Output batches:
+              [0-8]           [10-19]          [20-22]
+
+            Note: silences are kept for better transcription, previous implementation was
+            passing segments separatly, but the output was less accurate.
+            """
+            batch_start_time = None
+            batch_end_time = None
+
+            for segment in segments:
+                start_time, end_time = segment.start, segment.end
+                if batch_start_time is None or batch_end_time is None:
+                    batch_start_time = start_time
+                    batch_end_time = end_time
                     continue
 
-                # If segment is within max duration, yield as-is
-                if segment_duration <= max_dur:
-                    yield (start_time, end_time, audio_segment)
+                total_duration = end_time - batch_start_time
+
+                if total_duration <= max_duration:
+                    batch_end_time = end_time
                     continue
 
-                # Chunk large segments into smaller pieces
-                chunk_samples = int(max_dur * SAMPLERATE)
-                current_start = start_time
+                yield TimeSegment(batch_start_time, batch_end_time)
+                batch_start_time = start_time
+                batch_end_time = end_time
 
-                for chunk_offset in range(0, len(audio_segment), chunk_samples):
-                    chunk_audio = audio_segment[
-                        chunk_offset : chunk_offset + chunk_samples
-                    ]
-                    if len(chunk_audio) == 0:
-                        break
+            if batch_start_time is None or batch_end_time is None:
+                return
 
-                    chunk_duration = len(chunk_audio) / float(SAMPLERATE)
-                    chunk_end = current_start + chunk_duration
+            yield TimeSegment(batch_start_time, batch_end_time)
 
-                    # Only yield chunks that meet minimum duration
-                    if chunk_duration >= min_dur:
-                        yield (current_start, chunk_end, chunk_audio)
+        def batch_segment_to_audio_segment(
+            segments: Generator[TimeSegment, None, None],
+            audio_array,
+        ) -> Generator[AudioSegment, None, None]:
+            """Extract audio segments and apply padding for Parakeet compatibility.
 
-                    current_start = chunk_end
+            Uses pad_audio to ensure segments are at least 0.5s long, preventing
+            Parakeet crashes. This padding may cause slight timing overlaps between
+            segments, which are corrected by enforce_word_timing_constraints.
+            """
+            for segment in segments:
+                start_time, end_time = segment.start, segment.end
+                start_sample = int(start_time * SAMPLERATE)
+                end_sample = int(end_time * SAMPLERATE)
+                audio_segment = audio_array[start_sample:end_sample]
 
-        def batch_segments(segments, max_files=10, max_duration=5.0):
-            batch = []
-            batch_duration = 0.0
+                padded_segment = pad_audio(audio_segment, SAMPLERATE)
 
-            for start_time, end_time, audio_segment in segments:
-                segment_duration = end_time - start_time
+                yield AudioSegment(start_time, end_time, padded_segment)
 
-                if segment_duration < VAD_CONFIG["silence_padding"]:
-                    silence_samples = int(
-                        (VAD_CONFIG["silence_padding"] - segment_duration) * SAMPLERATE
-                    )
-                    padding = np.zeros(silence_samples, dtype=np.float32)
-                    audio_segment = np.concatenate([audio_segment, padding])
-                    segment_duration = VAD_CONFIG["silence_padding"]
-
-                batch.append((start_time, end_time, audio_segment))
-                batch_duration += segment_duration
-
-                if len(batch) >= max_files or batch_duration >= max_duration:
-                    yield batch
-                    batch = []
-                    batch_duration = 0.0
-
-            if batch:
-                yield batch
-
-        def transcribe_batch(model, audio_segments):
+        def transcribe_batch(model, audio_segments: list) -> list:
             with NoStdStreams():
                 outputs = model.transcribe(audio_segments, timestamps=True)
             return outputs
 
+        def enforce_word_timing_constraints(
+            words: list[WordTiming],
+        ) -> list[WordTiming]:
+            """Enforce that word end times don't exceed the start time of the next word.
+
+            Due to silence padding added in batch_segment_to_audio_segment for better
+            transcription accuracy, word timings from different segments may overlap.
+            This function ensures there are no overlaps by adjusting end times.
+            """
+            if len(words) <= 1:
+                return words
+
+            enforced_words = []
+            for i, word in enumerate(words):
+                enforced_word = word.copy()
+
+                if i < len(words) - 1:
+                    next_start = words[i + 1]["start"]
+                    if enforced_word["end"] > next_start:
+                        enforced_word["end"] = next_start
+
+                enforced_words.append(enforced_word)
+
+            return enforced_words
+
         def emit_results(
-            results,
-            segments_info,
-            batch_index,
-            total_batches,
-        ):
+            results: list,
+            segments_info: list[AudioSegment],
+        ) -> Generator[TranscriptResult, None, None]:
             """Yield transcribed text and word timings from model output, adjusting timestamps to absolute positions."""
-            for i, (output, (start_time, end_time, _)) in enumerate(
-                zip(results, segments_info)
-            ):
+            for i, (output, segment) in enumerate(zip(results, segments_info)):
+                start_time, end_time = segment.start, segment.end
                 text = output.text.strip()
-                words = [
-                    {
-                        "word": word_info["word"] + " ",
-                        "start": round(
+                words: list[WordTiming] = [
+                    WordTiming(
+                        word=word_info["word"] + " ",
+                        start=round(
                             word_info["start"] + start_time + timestamp_offset, 2
                         ),
-                        "end": round(
-                            word_info["end"] + start_time + timestamp_offset, 2
-                        ),
-                    }
+                        end=round(word_info["end"] + start_time + timestamp_offset, 2),
+                    )
                     for word_info in output.timestamp["word"]
                 ]
 
-                yield text, words
+                yield TranscriptResult(text, words)
 
         upload_volume.reload()
 
@@ -407,41 +453,31 @@ class TranscriberParakeetFile:
 
         audio_array = load_and_convert_audio(file_path)
         total_duration = len(audio_array) / float(SAMPLERATE)
-        processed_duration = 0.0
 
-        all_text_parts = []
-        all_words = []
+        all_text_parts: list[str] = []
+        all_words: list[WordTiming] = []
 
         raw_segments = vad_segment_generator(audio_array)
-        filtered_segments = vad_segment_filter(raw_segments)
-        batches = batch_segments(
-            filtered_segments,
-            VAD_CONFIG["batch_max_files"],
+        speech_segments = batch_speech_segments(
+            raw_segments,
             VAD_CONFIG["batch_max_duration"],
         )
+        audio_segments = batch_segment_to_audio_segment(speech_segments, audio_array)
 
-        batch_index = 0
-        total_batches = max(
-            1, int(total_duration / VAD_CONFIG["batch_max_duration"]) + 1
-        )
+        for batch in audio_segments:
+            audio_segment = batch.audio
+            results = transcribe_batch(self.model, [audio_segment])
 
-        for batch in batches:
-            batch_index += 1
-            audio_segments = [seg[2] for seg in batch]
-            results = transcribe_batch(self.model, audio_segments)
-
-            for text, words in emit_results(
+            for result in emit_results(
                 results,
-                batch,
-                batch_index,
-                total_batches,
+                [batch],
             ):
-                if not text:
+                if not result.text:
                     continue
-                all_text_parts.append(text)
-                all_words.extend(words)
+                all_text_parts.append(result.text)
+                all_words.extend(result.words)
 
-            processed_duration += sum(len(seg[2]) / float(SAMPLERATE) for seg in batch)
+        all_words = enforce_word_timing_constraints(all_words)
 
         combined_text = " ".join(all_text_parts)
         return {"text": combined_text, "words": all_words}

server/migrations/versions/0194f65cd6d3_add_webhook_fields_to_rooms.py

@@ -0,0 +1,36 @@
+"""Add webhook fields to rooms
+
+Revision ID: 0194f65cd6d3
+Revises: 5a8907fd1d78
+Create Date: 2025-08-27 09:03:19.610995
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0194f65cd6d3"
+down_revision: Union[str, None] = "5a8907fd1d78"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("webhook_url", sa.String(), nullable=True))
+        batch_op.add_column(sa.Column("webhook_secret", sa.String(), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.drop_column("webhook_secret")
+        batch_op.drop_column("webhook_url")
+
+    # ### end Alembic commands ###

server/migrations/versions/5a8907fd1d78_add_cascade_delete_to_meeting_consent_.py

@@ -0,0 +1,50 @@
+"""add cascade delete to meeting consent foreign key
+
+Revision ID: 5a8907fd1d78
+Revises: 0ab2d7ffaa16
+Create Date: 2025-08-26 17:26:50.945491
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "5a8907fd1d78"
+down_revision: Union[str, None] = "0ab2d7ffaa16"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("meeting_consent", schema=None) as batch_op:
+        batch_op.drop_constraint(
+            batch_op.f("meeting_consent_meeting_id_fkey"), type_="foreignkey"
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("meeting_consent_meeting_id_fkey"),
+            "meeting",
+            ["meeting_id"],
+            ["id"],
+            ondelete="CASCADE",
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("meeting_consent", schema=None) as batch_op:
+        batch_op.drop_constraint(
+            batch_op.f("meeting_consent_meeting_id_fkey"), type_="foreignkey"
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("meeting_consent_meeting_id_fkey"),
+            "meeting",
+            ["meeting_id"],
+            ["id"],
+        )
+
+    # ### end Alembic commands ###

server/migrations/versions/61882a919591_webhook_url_and_secret_null_by_default.py

@@ -0,0 +1,28 @@
+"""webhook url and secret null by default
+
+
+Revision ID: 61882a919591
+Revises: 0194f65cd6d3
+Create Date: 2025-08-29 11:46:36.738091
+
+"""
+
+from typing import Sequence, Union
+
+# revision identifiers, used by Alembic.
+revision: str = "61882a919591"
+down_revision: Union[str, None] = "0194f65cd6d3"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

server/reflector/asynctask.py

@@ -0,0 +1,27 @@
+import asyncio
+import functools
+
+from reflector.db import get_database
+
+
+def asynctask(f):
+    @functools.wraps(f)
+    def wrapper(*args, **kwargs):
+        async def run_with_db():
+            database = get_database()
+            await database.connect()
+            try:
+                return await f(*args, **kwargs)
+            finally:
+                await database.disconnect()
+
+        coro = run_with_db()
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        if loop and loop.is_running():
+            return loop.run_until_complete(coro)
+        return asyncio.run(coro)
+
+    return wrapper

server/reflector/db/meetings.py

@@ -54,7 +54,12 @@ meeting_consent = sa.Table(
     "meeting_consent",
     metadata,
     sa.Column("id", sa.String, primary_key=True),
-    sa.Column("meeting_id", sa.String, sa.ForeignKey("meeting.id"), nullable=False),
+    sa.Column(
+        "meeting_id",
+        sa.String,
+        sa.ForeignKey("meeting.id", ondelete="CASCADE"),
+        nullable=False,
+    ),
     sa.Column("user_id", sa.String),
     sa.Column("consent_given", sa.Boolean, nullable=False),
     sa.Column("consent_timestamp", sa.DateTime(timezone=True), nullable=False),

server/reflector/db/rooms.py

@@ -1,3 +1,4 @@
+import secrets
 from datetime import datetime, timezone
 from sqlite3 import IntegrityError
 from typing import Literal
@@ -40,6 +41,8 @@ rooms = sqlalchemy.Table(
     sqlalchemy.Column(
         "is_shared", sqlalchemy.Boolean, nullable=False, server_default=false()
     ),
+    sqlalchemy.Column("webhook_url", sqlalchemy.String, nullable=True),
+    sqlalchemy.Column("webhook_secret", sqlalchemy.String, nullable=True),
     sqlalchemy.Index("idx_room_is_shared", "is_shared"),
 )
 
@@ -59,6 +62,8 @@ class Room(BaseModel):
         "none", "prompt", "automatic", "automatic-2nd-participant"
     ] = "automatic-2nd-participant"
     is_shared: bool = False
+    webhook_url: str | None = None
+    webhook_secret: str | None = None
 
 
 class RoomController:
@@ -107,10 +112,15 @@ class RoomController:
         recording_type: str,
         recording_trigger: str,
         is_shared: bool,
+        webhook_url: str = "",
+        webhook_secret: str = "",
     ):
         """
         Add a new room
         """
+        if webhook_url and not webhook_secret:
+            webhook_secret = secrets.token_urlsafe(32)
+
         room = Room(
             name=name,
             user_id=user_id,
@@ -122,6 +132,8 @@ class RoomController:
             recording_type=recording_type,
             recording_trigger=recording_trigger,
             is_shared=is_shared,
+            webhook_url=webhook_url,
+            webhook_secret=webhook_secret,
         )
         query = rooms.insert().values(**room.model_dump())
         try:
@@ -134,6 +146,9 @@ class RoomController:
         """
         Update a room fields with key/values in values
         """
+        if values.get("webhook_url") and not values.get("webhook_secret"):
+            values["webhook_secret"] = secrets.token_urlsafe(32)
+
         query = rooms.update().where(rooms.c.id == room.id).values(**values)
         try:
             await get_database().execute(query)

server/reflector/db/search.py

@@ -8,12 +8,14 @@ from typing import Annotated, Any, Dict, Iterator
 
 import sqlalchemy
 import webvtt
+from databases.interfaces import Record as DbRecord
 from fastapi import HTTPException
 from pydantic import (
     BaseModel,
     Field,
     NonNegativeFloat,
     NonNegativeInt,
+    TypeAdapter,
     ValidationError,
     constr,
     field_serializer,
@@ -24,6 +26,7 @@ from reflector.db.rooms import rooms
 from reflector.db.transcripts import SourceKind, transcripts
 from reflector.db.utils import is_postgresql
 from reflector.logger import logger
+from reflector.utils.string import NonEmptyString, try_parse_non_empty_string
 
 DEFAULT_SEARCH_LIMIT = 20
 SNIPPET_CONTEXT_LENGTH = 50  # Characters before/after match to include
@@ -31,12 +34,13 @@ DEFAULT_SNIPPET_MAX_LENGTH = NonNegativeInt(150)
 DEFAULT_MAX_SNIPPETS = NonNegativeInt(3)
 LONG_SUMMARY_MAX_SNIPPETS = 2
 
-SearchQueryBase = constr(min_length=0, strip_whitespace=True)
+SearchQueryBase = constr(min_length=1, strip_whitespace=True)
 SearchLimitBase = Annotated[int, Field(ge=1, le=100)]
 SearchOffsetBase = Annotated[int, Field(ge=0)]
 SearchTotalBase = Annotated[int, Field(ge=0)]
 
 SearchQuery = Annotated[SearchQueryBase, Field(description="Search query text")]
+search_query_adapter = TypeAdapter(SearchQuery)
 SearchLimit = Annotated[SearchLimitBase, Field(description="Results per page")]
 SearchOffset = Annotated[
     SearchOffsetBase, Field(description="Number of results to skip")
@@ -88,7 +92,7 @@ class WebVTTProcessor:
     @staticmethod
     def generate_snippets(
         webvtt_content: WebVTTContent,
-        query: str,
+        query: SearchQuery,
         max_snippets: NonNegativeInt = DEFAULT_MAX_SNIPPETS,
     ) -> list[str]:
         """Generate snippets from WebVTT content."""
@@ -125,7 +129,7 @@ class SnippetCandidate:
 class SearchParameters(BaseModel):
     """Validated search parameters for full-text search."""
 
-    query_text: SearchQuery
+    query_text: SearchQuery | None = None
     limit: SearchLimit = DEFAULT_SEARCH_LIMIT
     offset: SearchOffset = 0
     user_id: str | None = None
@@ -199,15 +203,13 @@ class SnippetGenerator:
             prev_start = start
 
     @staticmethod
-    def count_matches(text: str, query: str) -> NonNegativeInt:
+    def count_matches(text: str, query: SearchQuery) -> NonNegativeInt:
         """Count total number of matches for a query in text."""
         ZERO = NonNegativeInt(0)
         if not text:
             logger.warning("Empty text for search query in count_matches")
             return ZERO
-        if not query:
-            logger.warning("Empty query for search text in count_matches")
-            return ZERO
+        assert query is not None
         return NonNegativeInt(
             sum(1 for _ in SnippetGenerator.find_all_matches(text, query))
         )
@@ -243,13 +245,14 @@ class SnippetGenerator:
     @staticmethod
     def generate(
         text: str,
-        query: str,
+        query: SearchQuery,
         max_length: NonNegativeInt = DEFAULT_SNIPPET_MAX_LENGTH,
         max_snippets: NonNegativeInt = DEFAULT_MAX_SNIPPETS,
     ) -> list[str]:
         """Generate snippets from text."""
-        if not text or not query:
-            logger.warning("Empty text or query for generate_snippets")
+        assert query is not None
+        if not text:
+            logger.warning("Empty text for generate_snippets")
             return []
 
         candidates = (
@@ -270,7 +273,7 @@ class SnippetGenerator:
     @staticmethod
     def from_summary(
         summary: str,
-        query: str,
+        query: SearchQuery,
         max_snippets: NonNegativeInt = LONG_SUMMARY_MAX_SNIPPETS,
     ) -> list[str]:
         """Generate snippets from summary text."""
@@ -278,9 +281,9 @@ class SnippetGenerator:
 
     @staticmethod
     def combine_sources(
-        summary: str | None,
+        summary: NonEmptyString | None,
         webvtt: WebVTTContent | None,
-        query: str,
+        query: SearchQuery,
         max_total: NonNegativeInt = DEFAULT_MAX_SNIPPETS,
     ) -> tuple[list[str], NonNegativeInt]:
         """Combine snippets from multiple sources and return total match count.
@@ -289,6 +292,11 @@ class SnippetGenerator:
 
         snippets can be empty for real in case of e.g. title match
         """
+
+        assert (
+            summary is not None or webvtt is not None
+        ), "At least one source must be present"
+
         webvtt_matches = 0
         summary_matches = 0
 
@@ -355,8 +363,8 @@ class SearchController:
                 else_=rooms.c.name,
             ).label("room_name"),
         ]
-
-        if params.query_text:
+        search_query = None
+        if params.query_text is not None:
             search_query = sqlalchemy.func.websearch_to_tsquery(
                 "english", params.query_text
             )
@@ -373,7 +381,9 @@ class SearchController:
             transcripts.join(rooms, transcripts.c.room_id == rooms.c.id, isouter=True)
         )
 
-        if params.query_text:
+        if params.query_text is not None:
+            # because already initialized based on params.query_text presence above
+            assert search_query is not None
             base_query = base_query.where(
                 transcripts.c.search_vector_en.op("@@")(search_query)
             )
@@ -393,7 +403,7 @@ class SearchController:
                 transcripts.c.source_kind == params.source_kind
             )
 
-        if params.query_text:
+        if params.query_text is not None:
             order_by = sqlalchemy.desc(sqlalchemy.text("rank"))
         else:
             order_by = sqlalchemy.desc(transcripts.c.created_at)
@@ -407,19 +417,29 @@ class SearchController:
         )
         total = await get_database().fetch_val(count_query)
 
-        def _process_result(r) -> SearchResult:
+        def _process_result(r: DbRecord) -> SearchResult:
             r_dict: Dict[str, Any] = dict(r)
+
             webvtt_raw: str | None = r_dict.pop("webvtt", None)
+            webvtt: WebVTTContent | None
             if webvtt_raw:
                 webvtt = WebVTTProcessor.parse(webvtt_raw)
             else:
                 webvtt = None
-            long_summary: str | None = r_dict.pop("long_summary", None)
+
+            long_summary_r: str | None = r_dict.pop("long_summary", None)
+            long_summary: NonEmptyString = try_parse_non_empty_string(long_summary_r)
             room_name: str | None = r_dict.pop("room_name", None)
             db_result = SearchResultDB.model_validate(r_dict)
 
-            snippets, total_match_count = SnippetGenerator.combine_sources(
-                long_summary, webvtt, params.query_text, DEFAULT_MAX_SNIPPETS
+            at_least_one_source = webvtt is not None or long_summary is not None
+            has_query = params.query_text is not None
+            snippets, total_match_count = (
+                SnippetGenerator.combine_sources(
+                    long_summary, webvtt, params.query_text, DEFAULT_MAX_SNIPPETS
+                )
+                if has_query and at_least_one_source
+                else ([], 0)
             )
 
             return SearchResult(

server/reflector/db/transcripts.py

@@ -122,6 +122,15 @@ def generate_transcript_name() -> str:
     return f"Transcript {now.strftime('%Y-%m-%d %H:%M:%S')}"
 
 
+TranscriptStatus = Literal[
+    "idle", "uploaded", "recording", "processing", "error", "ended"
+]
+
+
+class StrValue(BaseModel):
+    value: str
+
+
 class AudioWaveform(BaseModel):
     data: list[float]
 
@@ -185,7 +194,7 @@ class Transcript(BaseModel):
     id: str = Field(default_factory=generate_uuid4)
     user_id: str | None = None
     name: str = Field(default_factory=generate_transcript_name)
-    status: str = "idle"
+    status: TranscriptStatus = "idle"
     duration: float = 0
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     title: str | None = None
@@ -732,5 +741,27 @@ class TranscriptController:
         transcript.delete_participant(participant_id)
         await self.update(transcript, {"participants": transcript.participants_dump()})
 
+    async def set_status(
+        self, transcript_id: str, status: TranscriptStatus
+    ) -> TranscriptEvent | None:
+        """
+        Update the status of a transcript
+
+        Will add an event STATUS + update the status field of transcript
+        """
+        async with self.transaction():
+            transcript = await self.get_by_id(transcript_id)
+            if not transcript:
+                raise Exception(f"Transcript {transcript_id} not found")
+            if transcript.status == status:
+                return
+            resp = await self.append_event(
+                transcript=transcript,
+                event="STATUS",
+                data=StrValue(value=status),
+            )
+            await self.update(transcript, {"status": status})
+        return resp
+
 
 transcripts_controller = TranscriptController()

server/reflector/pipelines/main_file_pipeline.py

@@ -7,18 +7,26 @@ Uses parallel processing for transcription, diarization, and waveform generation
 """
 
 import asyncio
+import uuid
 from pathlib import Path
 
 import av
 import structlog
 from celery import shared_task
 
+from reflector.asynctask import asynctask
+from reflector.db.rooms import rooms_controller
 from reflector.db.transcripts import (
+    SourceKind,
     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,
+    broadcast_to_sockets,
+)
 from reflector.processors import (
     AudioFileWriterProcessor,
     TranscriptFinalSummaryProcessor,
@@ -43,6 +51,7 @@ from reflector.processors.types import (
 )
 from reflector.settings import settings
 from reflector.storage import get_transcripts_storage
+from reflector.worker.webhook import send_transcript_webhook
 
 
 class EmptyPipeline:
@@ -83,12 +92,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 +129,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 +388,34 @@ 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
+
+    # Trigger webhook if this is a room recording with webhook configured
+    if transcript.source_kind == SourceKind.ROOM and transcript.room_id:
+        room = await rooms_controller.get_by_id(transcript.room_id)
+        if room and room.webhook_url:
+            logger.info(
+                "Dispatching webhook task",
+                transcript_id=transcript_id,
+                room_id=room.id,
+                webhook_url=room.webhook_url,
+            )
+            send_transcript_webhook.delay(
+                transcript_id, room.id, event_id=uuid.uuid4().hex
+            )

server/reflector/pipelines/main_live_pipeline.py

@@ -22,7 +22,7 @@ from celery import chord, current_task, group, shared_task
 from pydantic import BaseModel
 from structlog import BoundLogger as Logger
 
-from reflector.db import get_database
+from reflector.asynctask import asynctask
 from reflector.db.meetings import meeting_consent_controller, meetings_controller
 from reflector.db.recordings import recordings_controller
 from reflector.db.rooms import rooms_controller
@@ -32,6 +32,7 @@ from reflector.db.transcripts import (
     TranscriptFinalLongSummary,
     TranscriptFinalShortSummary,
     TranscriptFinalTitle,
+    TranscriptStatus,
     TranscriptText,
     TranscriptTopic,
     TranscriptWaveform,
@@ -69,29 +70,6 @@ from reflector.zulip import (
 )
 
 
-def asynctask(f):
-    @functools.wraps(f)
-    def wrapper(*args, **kwargs):
-        async def run_with_db():
-            database = get_database()
-            await database.connect()
-            try:
-                return await f(*args, **kwargs)
-            finally:
-                await database.disconnect()
-
-        coro = run_with_db()
-        try:
-            loop = asyncio.get_running_loop()
-        except RuntimeError:
-            loop = None
-        if loop and loop.is_running():
-            return loop.run_until_complete(coro)
-        return asyncio.run(coro)
-
-    return wrapper
-
-
 def broadcast_to_sockets(func):
     """
     Decorator to broadcast transcript event to websockets
@@ -188,8 +166,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 +183,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 +206,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):
@@ -794,7 +765,7 @@ def pipeline_post(*, transcript_id: str):
         chain_final_summaries,
     ) | task_pipeline_post_to_zulip.si(transcript_id=transcript_id)
 
-    chain.delay()
+    return chain.delay()
 
 
 @get_transcript

server/reflector/processors/file_transcript_modal.py

@@ -67,6 +67,9 @@ class FileTranscriptModalProcessor(FileTranscriptProcessor):
             for word_info in result.get("words", [])
         ]
 
+        # words come not in order
+        words.sort(key=lambda w: w.start)
+
         return Transcript(words=words)
 
 

server/reflector/settings.py

@@ -1,3 +1,4 @@
+from pydantic.types import PositiveInt
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -90,9 +91,8 @@ class Settings(BaseSettings):
     AUTH_JWT_PUBLIC_KEY: str | None = "authentik.monadical.com_public.pem"
     AUTH_JWT_AUDIENCE: str | None = None
 
-    # API public mode
-    # if set, all anonymous record will be public
     PUBLIC_MODE: bool = False
+    PUBLIC_DATA_RETENTION_DAYS: PositiveInt = 7
 
     # Min transcript length to generate topic + summary
     MIN_TRANSCRIPT_LENGTH: int = 750

server/reflector/tools/cleanup_old_data.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python
+"""
+Manual cleanup tool for old public data.
+Uses the same implementation as the Celery worker task.
+"""
+
+import argparse
+import asyncio
+import sys
+
+import structlog
+
+from reflector.settings import settings
+from reflector.worker.cleanup import _cleanup_old_public_data
+
+logger = structlog.get_logger(__name__)
+
+
+async def cleanup_old_data(days: int = 7):
+    logger.info(
+        "Starting manual cleanup",
+        retention_days=days,
+        public_mode=settings.PUBLIC_MODE,
+    )
+
+    if not settings.PUBLIC_MODE:
+        logger.critical(
+            "WARNING: PUBLIC_MODE is False. "
+            "This tool is intended for public instances only."
+        )
+        raise Exception("Tool intended for public instances only")
+
+    result = await _cleanup_old_public_data(days=days)
+
+    if result:
+        logger.info(
+            "Cleanup completed",
+            transcripts_deleted=result.get("transcripts_deleted", 0),
+            meetings_deleted=result.get("meetings_deleted", 0),
+            recordings_deleted=result.get("recordings_deleted", 0),
+            errors_count=len(result.get("errors", [])),
+        )
+        if result.get("errors"):
+            logger.warning(
+                "Errors encountered during cleanup:", errors=result["errors"][:10]
+            )
+    else:
+        logger.info("Cleanup skipped or completed without results")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Clean up old transcripts and meetings"
+    )
+    parser.add_argument(
+        "--days",
+        type=int,
+        default=7,
+        help="Number of days to keep data (default: 7)",
+    )
+
+    args = parser.parse_args()
+
+    if args.days < 1:
+        logger.error("Days must be at least 1")
+        sys.exit(1)
+
+    asyncio.run(cleanup_old_data(days=args.days))
+
+
+if __name__ == "__main__":
+    main()

server/reflector/tools/process.py

@@ -1,294 +1,204 @@
 """
 Process audio file with diarization support
-===========================================
-
-Extended version of process.py that includes speaker diarization.
-This tool processes audio files locally without requiring the full server infrastructure.
 """
 
+import argparse
 import asyncio
-import tempfile
-import uuid
+import json
+import shutil
+import sys
+import time
 from pathlib import Path
-from typing import List
-
-import av
+from typing import Any, Dict, List, Literal
 
+from reflector.db.transcripts import SourceKind, TranscriptTopic, transcripts_controller
 from reflector.logger import logger
-from reflector.processors import (
-    AudioChunkerAutoProcessor,
-    AudioDownscaleProcessor,
-    AudioFileWriterProcessor,
-    AudioMergeProcessor,
-    AudioTranscriptAutoProcessor,
-    Pipeline,
-    PipelineEvent,
-    TranscriptFinalSummaryProcessor,
-    TranscriptFinalTitleProcessor,
-    TranscriptLinerProcessor,
-    TranscriptTopicDetectorProcessor,
-    TranscriptTranslatorAutoProcessor,
+from reflector.pipelines.main_file_pipeline import (
+    task_pipeline_file_process as task_pipeline_file_process,
 )
-from reflector.processors.base import BroadcastProcessor, Processor
-from reflector.processors.types import (
-    AudioDiarizationInput,
-    TitleSummary,
-    TitleSummaryWithId,
+from reflector.pipelines.main_live_pipeline import pipeline_post as live_pipeline_post
+from reflector.pipelines.main_live_pipeline import (
+    pipeline_process as live_pipeline_process,
 )
 
 
-class TopicCollectorProcessor(Processor):
-    """Collect topics for diarization"""
+def serialize_topics(topics: List[TranscriptTopic]) -> List[Dict[str, Any]]:
+    """Convert TranscriptTopic objects to JSON-serializable dicts"""
+    serialized = []
+    for topic in topics:
+        topic_dict = topic.model_dump()
+        serialized.append(topic_dict)
+    return serialized
 
-    INPUT_TYPE = TitleSummary
-    OUTPUT_TYPE = TitleSummary
 
-    def __init__(self, **kwargs):
-        super().__init__(**kwargs)
-        self.topics: List[TitleSummaryWithId] = []
-        self._topic_id = 0
+def debug_print_speakers(serialized_topics: List[Dict[str, Any]]) -> None:
+    """Print debug info about speakers found in topics"""
+    all_speakers = set()
+    for topic_dict in serialized_topics:
+        for word in topic_dict.get("words", []):
+            all_speakers.add(word.get("speaker", 0))
 
-    async def _push(self, data: TitleSummary):
-        # Convert to TitleSummaryWithId and collect
-        self._topic_id += 1
-        topic_with_id = TitleSummaryWithId(
-            id=str(self._topic_id),
-            title=data.title,
-            summary=data.summary,
-            timestamp=data.timestamp,
-            duration=data.duration,
-            transcript=data.transcript,
+    print(
+        f"Found {len(serialized_topics)} topics with speakers: {all_speakers}",
+        file=sys.stderr,
+    )
+
+
+TranscriptId = str
+
+
+# common interface for every flow: it needs an Entry in db with specific ceremony (file path + status + actual file in file system)
+# ideally we want to get rid of it at some point
+async def prepare_entry(
+    source_path: str,
+    source_language: str,
+    target_language: str,
+) -> TranscriptId:
+    file_path = Path(source_path)
+
+    transcript = await transcripts_controller.add(
+        file_path.name,
+        # note that the real file upload has SourceKind: LIVE for the reason of it's an error
+        source_kind=SourceKind.FILE,
+        source_language=source_language,
+        target_language=target_language,
+        user_id=None,
+    )
+
+    logger.info(
+        f"Created empty transcript {transcript.id} for file {file_path.name} because technically we need an empty transcript before we start transcript"
+    )
+
+    # pipelines expect files as upload.*
+
+    extension = file_path.suffix
+    upload_path = transcript.data_path / f"upload{extension}"
+    upload_path.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(source_path, upload_path)
+    logger.info(f"Copied {source_path} to {upload_path}")
+
+    # pipelines expect entity status "uploaded"
+    await transcripts_controller.update(transcript, {"status": "uploaded"})
+
+    return transcript.id
+
+
+# same reason as prepare_entry
+async def extract_result_from_entry(
+    transcript_id: TranscriptId, output_path: str
+) -> None:
+    post_final_transcript = await transcripts_controller.get_by_id(transcript_id)
+
+    # assert post_final_transcript.status == "ended"
+    # File pipeline doesn't set status to "ended", only live pipeline does https://github.com/Monadical-SAS/reflector/issues/582
+    topics = post_final_transcript.topics
+    if not topics:
+        raise RuntimeError(
+            f"No topics found for transcript {transcript_id} after processing"
         )
-        self.topics.append(topic_with_id)
 
-        # Pass through the original topic
-        await self.emit(data)
+    serialized_topics = serialize_topics(topics)
 
-    def get_topics(self) -> List[TitleSummaryWithId]:
-        return self.topics
+    if output_path:
+        # Write to JSON file
+        with open(output_path, "w") as f:
+            for topic_dict in serialized_topics:
+                json.dump(topic_dict, f)
+                f.write("\n")
+        print(f"Results written to {output_path}", file=sys.stderr)
+    else:
+        # Write to stdout as JSONL
+        for topic_dict in serialized_topics:
+            print(json.dumps(topic_dict))
+
+    debug_print_speakers(serialized_topics)
 
 
-async def process_audio_file(
-    filename,
-    event_callback,
-    only_transcript=False,
-    source_language="en",
-    target_language="en",
-    enable_diarization=True,
-    diarization_backend="pyannote",
+async def process_live_pipeline(
+    transcript_id: TranscriptId,
 ):
-    # Create temp file for audio if diarization is enabled
-    audio_temp_path = None
-    if enable_diarization:
-        audio_temp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
-        audio_temp_path = audio_temp_file.name
-        audio_temp_file.close()
+    """Process transcript_id with transcription and diarization"""
 
-    # Create processor for collecting topics
-    topic_collector = TopicCollectorProcessor()
+    print(f"Processing transcript_id {transcript_id}...", file=sys.stderr)
+    await live_pipeline_process(transcript_id=transcript_id)
+    print(f"Processing complete for transcript {transcript_id}", file=sys.stderr)
 
-    # Build pipeline for audio processing
-    processors = []
+    pre_final_transcript = await transcripts_controller.get_by_id(transcript_id)
 
-    # Add audio file writer at the beginning if diarization is enabled
-    if enable_diarization:
-        processors.append(AudioFileWriterProcessor(audio_temp_path))
+    # assert documented behaviour: after process, the pipeline isn't ended. this is the reason of calling pipeline_post
+    assert pre_final_transcript.status != "ended"
 
-    # Add the rest of the processors
-    processors += [
-        AudioDownscaleProcessor(),
-        AudioChunkerAutoProcessor(),
-        AudioMergeProcessor(),
-        AudioTranscriptAutoProcessor.as_threaded(),
-        TranscriptLinerProcessor(),
-        TranscriptTranslatorAutoProcessor.as_threaded(),
-    ]
+    # at this point, diarization is running but we have no access to it. run diarization in parallel - one will hopefully win after polling
+    result = live_pipeline_post(transcript_id=transcript_id)
 
-    if not only_transcript:
-        processors += [
-            TranscriptTopicDetectorProcessor.as_threaded(),
-            # Collect topics for diarization
-            topic_collector,
-            BroadcastProcessor(
-                processors=[
-                    TranscriptFinalTitleProcessor.as_threaded(),
-                    TranscriptFinalSummaryProcessor.as_threaded(),
-                ],
-            ),
-        ]
-
-    # Create main pipeline
-    pipeline = Pipeline(*processors)
-    pipeline.set_pref("audio:source_language", source_language)
-    pipeline.set_pref("audio:target_language", target_language)
-    pipeline.describe()
-    pipeline.on(event_callback)
-
-    # Start processing audio
-    logger.info(f"Opening {filename}")
-    container = av.open(filename)
-    try:
-        logger.info("Start pushing audio into the pipeline")
-        for frame in container.decode(audio=0):
-            await pipeline.push(frame)
-    finally:
-        logger.info("Flushing the pipeline")
-        await pipeline.flush()
-
-    # Run diarization if enabled and we have topics
-    if enable_diarization and not only_transcript and audio_temp_path:
-        topics = topic_collector.get_topics()
-
-        if topics:
-            logger.info(f"Starting diarization with {len(topics)} topics")
-
-            try:
-                from reflector.processors import AudioDiarizationAutoProcessor
-
-                diarization_processor = AudioDiarizationAutoProcessor(
-                    name=diarization_backend
-                )
-
-                diarization_processor.set_pipeline(pipeline)
-
-                # For Modal backend, we need to upload the file to S3 first
-                if diarization_backend == "modal":
-                    from datetime import datetime
-
-                    from reflector.storage import get_transcripts_storage
-                    from reflector.utils.s3_temp_file import S3TemporaryFile
-
-                    storage = get_transcripts_storage()
-
-                    # Generate a unique filename in evaluation folder
-                    timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
-                    audio_filename = f"evaluation/diarization_temp/{timestamp}_{uuid.uuid4().hex}.wav"
-
-                    # Use context manager for automatic cleanup
-                    async with S3TemporaryFile(storage, audio_filename) as s3_file:
-                        # Read and upload the audio file
-                        with open(audio_temp_path, "rb") as f:
-                            audio_data = f.read()
-
-                        audio_url = await s3_file.upload(audio_data)
-                        logger.info(f"Uploaded audio to S3: {audio_filename}")
-
-                        # Create diarization input with S3 URL
-                        diarization_input = AudioDiarizationInput(
-                            audio_url=audio_url, topics=topics
-                        )
-
-                        # Run diarization
-                        await diarization_processor.push(diarization_input)
-                        await diarization_processor.flush()
-
-                        logger.info("Diarization complete")
-                        # File will be automatically cleaned up when exiting the context
-                else:
-                    # For local backend, use local file path
-                    audio_url = audio_temp_path
-
-                    # Create diarization input
-                    diarization_input = AudioDiarizationInput(
-                        audio_url=audio_url, topics=topics
-                    )
-
-                    # Run diarization
-                    await diarization_processor.push(diarization_input)
-                    await diarization_processor.flush()
-
-                    logger.info("Diarization complete")
-
-            except ImportError as e:
-                logger.error(f"Failed to import diarization dependencies: {e}")
-                logger.error(
-                    "Install with: uv pip install pyannote.audio torch torchaudio"
-                )
-                logger.error(
-                    "And set HF_TOKEN environment variable for pyannote models"
-                )
-                raise SystemExit(1)
-            except Exception as e:
-                logger.error(f"Diarization failed: {e}")
-                raise SystemExit(1)
-        else:
-            logger.warning("Skipping diarization: no topics available")
-
-    # Clean up temp file
-    if audio_temp_path:
-        try:
-            Path(audio_temp_path).unlink()
-        except Exception as e:
-            logger.warning(f"Failed to clean up temp file {audio_temp_path}: {e}")
-
-    logger.info("All done!")
+    # result.ready() blocks even without await; it mutates result also
+    while not result.ready():
+        print(f"Status: {result.state}")
+        time.sleep(2)
 
 
 async def process_file_pipeline(
-    filename: str,
-    event_callback,
-    source_language="en",
-    target_language="en",
-    enable_diarization=True,
-    diarization_backend="modal",
+    transcript_id: TranscriptId,
 ):
     """Process audio/video file using the optimized file pipeline"""
+
+    # task_pipeline_file_process is a Celery task, need to use .delay() for async execution
+    result = task_pipeline_file_process.delay(transcript_id=transcript_id)
+
+    # Wait for the Celery task to complete
+    while not result.ready():
+        print(f"File pipeline status: {result.state}", file=sys.stderr)
+        time.sleep(2)
+
+    logger.info("File pipeline processing complete")
+
+
+async def process(
+    source_path: str,
+    source_language: str,
+    target_language: str,
+    pipeline: Literal["live", "file"],
+    output_path: str = None,
+):
+    from reflector.db import get_database
+
+    database = get_database()
+    # db connect is a part of ceremony
+    await database.connect()
+
     try:
-        from reflector.db import database
-        from reflector.db.transcripts import SourceKind, transcripts_controller
-        from reflector.pipelines.main_file_pipeline import PipelineMainFile
-
-        await database.connect()
-        try:
-            # Create a temporary transcript for processing
-            transcript = await transcripts_controller.add(
-                "",
-                source_kind=SourceKind.FILE,
-                source_language=source_language,
-                target_language=target_language,
-            )
-
-            # Process the file
-            pipeline = PipelineMainFile(transcript_id=transcript.id)
-            await pipeline.process(Path(filename))
-
-            logger.info("File pipeline processing complete")
-
-        finally:
-            await database.disconnect()
-    except ImportError as e:
-        logger.error(f"File pipeline not available: {e}")
-        logger.info("Falling back to stream pipeline")
-        # Fall back to stream pipeline
-        await process_audio_file(
-            filename,
-            event_callback,
-            only_transcript=False,
-            source_language=source_language,
-            target_language=target_language,
-            enable_diarization=enable_diarization,
-            diarization_backend=diarization_backend,
+        transcript_id = await prepare_entry(
+            source_path,
+            source_language,
+            target_language,
         )
 
+        pipeline_handlers = {
+            "live": process_live_pipeline,
+            "file": process_file_pipeline,
+        }
+
+        handler = pipeline_handlers.get(pipeline)
+        if not handler:
+            raise ValueError(f"Unknown pipeline type: {pipeline}")
+
+        await handler(transcript_id)
+
+        await extract_result_from_entry(transcript_id, output_path)
+    finally:
+        await database.disconnect()
+
 
 if __name__ == "__main__":
-    import argparse
-    import os
-
     parser = argparse.ArgumentParser(
-        description="Process audio files with optional speaker diarization"
+        description="Process audio files with speaker diarization"
     )
     parser.add_argument("source", help="Source file (mp3, wav, mp4...)")
     parser.add_argument(
-        "--stream",
-        action="store_true",
-        help="Use streaming pipeline (original frame-based processing)",
-    )
-    parser.add_argument(
-        "--only-transcript",
-        "-t",
-        action="store_true",
-        help="Only generate transcript without topics/summaries",
+        "--pipeline",
+        required=True,
+        choices=["live", "file"],
+        help="Pipeline type to use for processing (live: streaming/incremental, file: batch/parallel)",
     )
     parser.add_argument(
         "--source-language", default="en", help="Source language code (default: en)"
@@ -297,82 +207,14 @@ if __name__ == "__main__":
         "--target-language", default="en", help="Target language code (default: en)"
     )
     parser.add_argument("--output", "-o", help="Output file (output.jsonl)")
-    parser.add_argument(
-        "--enable-diarization",
-        "-d",
-        action="store_true",
-        help="Enable speaker diarization",
-    )
-    parser.add_argument(
-        "--diarization-backend",
-        default="pyannote",
-        choices=["pyannote", "modal"],
-        help="Diarization backend to use (default: pyannote)",
-    )
     args = parser.parse_args()
 
-    if "REDIS_HOST" not in os.environ:
-        os.environ["REDIS_HOST"] = "localhost"
-
-    output_fd = None
-    if args.output:
-        output_fd = open(args.output, "w")
-
-    async def event_callback(event: PipelineEvent):
-        processor = event.processor
-        data = event.data
-
-        # Ignore internal processors
-        if processor in (
-            "AudioDownscaleProcessor",
-            "AudioChunkerAutoProcessor",
-            "AudioMergeProcessor",
-            "AudioFileWriterProcessor",
-            "TopicCollectorProcessor",
-            "BroadcastProcessor",
-        ):
-            return
-
-        # If diarization is enabled, skip the original topic events from the pipeline
-        # The diarization processor will emit the same topics but with speaker info
-        if processor == "TranscriptTopicDetectorProcessor" and args.enable_diarization:
-            return
-
-        # Log all events
-        logger.info(f"Event: {processor} - {type(data).__name__}")
-
-        # Write to output
-        if output_fd:
-            output_fd.write(event.model_dump_json())
-            output_fd.write("\n")
-            output_fd.flush()
-
-    if args.stream:
-        # Use original streaming pipeline
-        asyncio.run(
-            process_audio_file(
-                args.source,
-                event_callback,
-                only_transcript=args.only_transcript,
-                source_language=args.source_language,
-                target_language=args.target_language,
-                enable_diarization=args.enable_diarization,
-                diarization_backend=args.diarization_backend,
-            )
+    asyncio.run(
+        process(
+            args.source,
+            args.source_language,
+            args.target_language,
+            args.pipeline,
+            args.output,
         )
-    else:
-        # Use optimized file pipeline (default)
-        asyncio.run(
-            process_file_pipeline(
-                args.source,
-                event_callback,
-                source_language=args.source_language,
-                target_language=args.target_language,
-                enable_diarization=args.enable_diarization,
-                diarization_backend=args.diarization_backend,
-            )
-        )
-
-    if output_fd:
-        output_fd.close()
-        logger.info(f"Output written to {args.output}")
+    )

server/reflector/tools/process_with_diarization.py

@@ -1,318 +0,0 @@
-"""
-@vibe-generated
-Process audio file with diarization support
-===========================================
-
-Extended version of process.py that includes speaker diarization.
-This tool processes audio files locally without requiring the full server infrastructure.
-"""
-
-import asyncio
-import tempfile
-import uuid
-from pathlib import Path
-from typing import List
-
-import av
-
-from reflector.logger import logger
-from reflector.processors import (
-    AudioChunkerAutoProcessor,
-    AudioDownscaleProcessor,
-    AudioFileWriterProcessor,
-    AudioMergeProcessor,
-    AudioTranscriptAutoProcessor,
-    Pipeline,
-    PipelineEvent,
-    TranscriptFinalSummaryProcessor,
-    TranscriptFinalTitleProcessor,
-    TranscriptLinerProcessor,
-    TranscriptTopicDetectorProcessor,
-    TranscriptTranslatorAutoProcessor,
-)
-from reflector.processors.base import BroadcastProcessor, Processor
-from reflector.processors.types import (
-    AudioDiarizationInput,
-    TitleSummary,
-    TitleSummaryWithId,
-)
-
-
-class TopicCollectorProcessor(Processor):
-    """Collect topics for diarization"""
-
-    INPUT_TYPE = TitleSummary
-    OUTPUT_TYPE = TitleSummary
-
-    def __init__(self, **kwargs):
-        super().__init__(**kwargs)
-        self.topics: List[TitleSummaryWithId] = []
-        self._topic_id = 0
-
-    async def _push(self, data: TitleSummary):
-        # Convert to TitleSummaryWithId and collect
-        self._topic_id += 1
-        topic_with_id = TitleSummaryWithId(
-            id=str(self._topic_id),
-            title=data.title,
-            summary=data.summary,
-            timestamp=data.timestamp,
-            duration=data.duration,
-            transcript=data.transcript,
-        )
-        self.topics.append(topic_with_id)
-
-        # Pass through the original topic
-        await self.emit(data)
-
-    def get_topics(self) -> List[TitleSummaryWithId]:
-        return self.topics
-
-
-async def process_audio_file_with_diarization(
-    filename,
-    event_callback,
-    only_transcript=False,
-    source_language="en",
-    target_language="en",
-    enable_diarization=True,
-    diarization_backend="modal",
-):
-    # Create temp file for audio if diarization is enabled
-    audio_temp_path = None
-    if enable_diarization:
-        audio_temp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
-        audio_temp_path = audio_temp_file.name
-        audio_temp_file.close()
-
-    # Create processor for collecting topics
-    topic_collector = TopicCollectorProcessor()
-
-    # Build pipeline for audio processing
-    processors = []
-
-    # Add audio file writer at the beginning if diarization is enabled
-    if enable_diarization:
-        processors.append(AudioFileWriterProcessor(audio_temp_path))
-
-    # Add the rest of the processors
-    processors += [
-        AudioDownscaleProcessor(),
-        AudioChunkerAutoProcessor(),
-        AudioMergeProcessor(),
-        AudioTranscriptAutoProcessor.as_threaded(),
-    ]
-
-    processors += [
-        TranscriptLinerProcessor(),
-        TranscriptTranslatorAutoProcessor.as_threaded(),
-    ]
-
-    if not only_transcript:
-        processors += [
-            TranscriptTopicDetectorProcessor.as_threaded(),
-            # Collect topics for diarization
-            topic_collector,
-            BroadcastProcessor(
-                processors=[
-                    TranscriptFinalTitleProcessor.as_threaded(),
-                    TranscriptFinalSummaryProcessor.as_threaded(),
-                ],
-            ),
-        ]
-
-    # Create main pipeline
-    pipeline = Pipeline(*processors)
-    pipeline.set_pref("audio:source_language", source_language)
-    pipeline.set_pref("audio:target_language", target_language)
-    pipeline.describe()
-    pipeline.on(event_callback)
-
-    # Start processing audio
-    logger.info(f"Opening {filename}")
-    container = av.open(filename)
-    try:
-        logger.info("Start pushing audio into the pipeline")
-        for frame in container.decode(audio=0):
-            await pipeline.push(frame)
-    finally:
-        logger.info("Flushing the pipeline")
-        await pipeline.flush()
-
-    # Run diarization if enabled and we have topics
-    if enable_diarization and not only_transcript and audio_temp_path:
-        topics = topic_collector.get_topics()
-
-        if topics:
-            logger.info(f"Starting diarization with {len(topics)} topics")
-
-            try:
-                from reflector.processors import AudioDiarizationAutoProcessor
-
-                diarization_processor = AudioDiarizationAutoProcessor(
-                    name=diarization_backend
-                )
-
-                diarization_processor.set_pipeline(pipeline)
-
-                # For Modal backend, we need to upload the file to S3 first
-                if diarization_backend == "modal":
-                    from datetime import datetime, timezone
-
-                    from reflector.storage import get_transcripts_storage
-                    from reflector.utils.s3_temp_file import S3TemporaryFile
-
-                    storage = get_transcripts_storage()
-
-                    # Generate a unique filename in evaluation folder
-                    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
-                    audio_filename = f"evaluation/diarization_temp/{timestamp}_{uuid.uuid4().hex}.wav"
-
-                    # Use context manager for automatic cleanup
-                    async with S3TemporaryFile(storage, audio_filename) as s3_file:
-                        # Read and upload the audio file
-                        with open(audio_temp_path, "rb") as f:
-                            audio_data = f.read()
-
-                        audio_url = await s3_file.upload(audio_data)
-                        logger.info(f"Uploaded audio to S3: {audio_filename}")
-
-                        # Create diarization input with S3 URL
-                        diarization_input = AudioDiarizationInput(
-                            audio_url=audio_url, topics=topics
-                        )
-
-                        # Run diarization
-                        await diarization_processor.push(diarization_input)
-                        await diarization_processor.flush()
-
-                        logger.info("Diarization complete")
-                        # File will be automatically cleaned up when exiting the context
-                else:
-                    # For local backend, use local file path
-                    audio_url = audio_temp_path
-
-                    # Create diarization input
-                    diarization_input = AudioDiarizationInput(
-                        audio_url=audio_url, topics=topics
-                    )
-
-                    # Run diarization
-                    await diarization_processor.push(diarization_input)
-                    await diarization_processor.flush()
-
-                    logger.info("Diarization complete")
-
-            except ImportError as e:
-                logger.error(f"Failed to import diarization dependencies: {e}")
-                logger.error(
-                    "Install with: uv pip install pyannote.audio torch torchaudio"
-                )
-                logger.error(
-                    "And set HF_TOKEN environment variable for pyannote models"
-                )
-                raise SystemExit(1)
-            except Exception as e:
-                logger.error(f"Diarization failed: {e}")
-                raise SystemExit(1)
-        else:
-            logger.warning("Skipping diarization: no topics available")
-
-    # Clean up temp file
-    if audio_temp_path:
-        try:
-            Path(audio_temp_path).unlink()
-        except Exception as e:
-            logger.warning(f"Failed to clean up temp file {audio_temp_path}: {e}")
-
-    logger.info("All done!")
-
-
-if __name__ == "__main__":
-    import argparse
-    import os
-
-    parser = argparse.ArgumentParser(
-        description="Process audio files with optional speaker diarization"
-    )
-    parser.add_argument("source", help="Source file (mp3, wav, mp4...)")
-    parser.add_argument(
-        "--only-transcript",
-        "-t",
-        action="store_true",
-        help="Only generate transcript without topics/summaries",
-    )
-    parser.add_argument(
-        "--source-language", default="en", help="Source language code (default: en)"
-    )
-    parser.add_argument(
-        "--target-language", default="en", help="Target language code (default: en)"
-    )
-    parser.add_argument("--output", "-o", help="Output file (output.jsonl)")
-    parser.add_argument(
-        "--enable-diarization",
-        "-d",
-        action="store_true",
-        help="Enable speaker diarization",
-    )
-    parser.add_argument(
-        "--diarization-backend",
-        default="modal",
-        choices=["modal"],
-        help="Diarization backend to use (default: modal)",
-    )
-    args = parser.parse_args()
-
-    # Set REDIS_HOST to localhost if not provided
-    if "REDIS_HOST" not in os.environ:
-        os.environ["REDIS_HOST"] = "localhost"
-        logger.info("REDIS_HOST not set, defaulting to localhost")
-
-    output_fd = None
-    if args.output:
-        output_fd = open(args.output, "w")
-
-    async def event_callback(event: PipelineEvent):
-        processor = event.processor
-        data = event.data
-
-        # Ignore internal processors
-        if processor in (
-            "AudioDownscaleProcessor",
-            "AudioChunkerAutoProcessor",
-            "AudioMergeProcessor",
-            "AudioFileWriterProcessor",
-            "TopicCollectorProcessor",
-            "BroadcastProcessor",
-        ):
-            return
-
-        # If diarization is enabled, skip the original topic events from the pipeline
-        # The diarization processor will emit the same topics but with speaker info
-        if processor == "TranscriptTopicDetectorProcessor" and args.enable_diarization:
-            return
-
-        # Log all events
-        logger.info(f"Event: {processor} - {type(data).__name__}")
-
-        # Write to output
-        if output_fd:
-            output_fd.write(event.model_dump_json())
-            output_fd.write("\n")
-            output_fd.flush()
-
-    asyncio.run(
-        process_audio_file_with_diarization(
-            args.source,
-            event_callback,
-            only_transcript=args.only_transcript,
-            source_language=args.source_language,
-            target_language=args.target_language,
-            enable_diarization=args.enable_diarization,
-            diarization_backend=args.diarization_backend,
-        )
-    )
-
-    if output_fd:
-        output_fd.close()
-        logger.info(f"Output written to {args.output}")

server/reflector/tools/test_diarization.py

@@ -1,96 +0,0 @@
-#!/usr/bin/env python3
-"""
-@vibe-generated
-Test script for the diarization CLI tool
-=========================================
-
-This script helps test the diarization functionality with sample audio files.
-"""
-
-import asyncio
-import sys
-from pathlib import Path
-
-from reflector.logger import logger
-
-
-async def test_diarization(audio_file: str):
-    """Test the diarization functionality"""
-
-    # Import the processing function
-    from process_with_diarization import process_audio_file_with_diarization
-
-    # Collect events
-    events = []
-
-    async def event_callback(event):
-        events.append({"processor": event.processor, "data": event.data})
-        logger.info(f"Event from {event.processor}")
-
-    # Process the audio file
-    logger.info(f"Processing audio file: {audio_file}")
-
-    try:
-        await process_audio_file_with_diarization(
-            audio_file,
-            event_callback,
-            only_transcript=False,
-            source_language="en",
-            target_language="en",
-            enable_diarization=True,
-            diarization_backend="modal",
-        )
-
-        # Analyze results
-        logger.info(f"Processing complete. Received {len(events)} events")
-
-        # Look for diarization results
-        diarized_topics = []
-        for event in events:
-            if "TitleSummary" in event["processor"]:
-                # Check if words have speaker information
-                if hasattr(event["data"], "transcript") and event["data"].transcript:
-                    words = event["data"].transcript.words
-                    if words and hasattr(words[0], "speaker"):
-                        speakers = set(
-                            w.speaker for w in words if hasattr(w, "speaker")
-                        )
-                        logger.info(
-                            f"Found {len(speakers)} speakers in topic: {event['data'].title}"
-                        )
-                        diarized_topics.append(event["data"])
-
-        if diarized_topics:
-            logger.info(f"Successfully diarized {len(diarized_topics)} topics")
-
-            # Print sample output
-            sample_topic = diarized_topics[0]
-            logger.info("Sample diarized output:")
-            for i, word in enumerate(sample_topic.transcript.words[:10]):
-                logger.info(f"  Word {i}: '{word.text}' - Speaker {word.speaker}")
-        else:
-            logger.warning("No diarization results found in output")
-
-        return events
-
-    except Exception as e:
-        logger.error(f"Error during processing: {e}")
-        raise
-
-
-def main():
-    if len(sys.argv) < 2:
-        print("Usage: python test_diarization.py <audio_file>")
-        sys.exit(1)
-
-    audio_file = sys.argv[1]
-    if not Path(audio_file).exists():
-        print(f"Error: Audio file '{audio_file}' not found")
-        sys.exit(1)
-
-    # Run the test
-    asyncio.run(test_diarization(audio_file))
-
-
-if __name__ == "__main__":
-    main()

server/reflector/utils/string.py

@@ -0,0 +1,20 @@
+from typing import Annotated
+
+from pydantic import Field, TypeAdapter, constr
+
+NonEmptyStringBase = constr(min_length=1, strip_whitespace=False)
+NonEmptyString = Annotated[
+    NonEmptyStringBase,
+    Field(description="A non-empty string", min_length=1),
+]
+non_empty_string_adapter = TypeAdapter(NonEmptyString)
+
+
+def parse_non_empty_string(s: str) -> NonEmptyString:
+    return non_empty_string_adapter.validate_python(s)
+
+
+def try_parse_non_empty_string(s: str) -> NonEmptyString | None:
+    if not s:
+        return None
+    return parse_non_empty_string(s)

server/reflector/views/rooms.py

@@ -15,6 +15,7 @@ from reflector.db.meetings import meetings_controller
 from reflector.db.rooms import rooms_controller
 from reflector.settings import settings
 from reflector.whereby import create_meeting, upload_logo
+from reflector.worker.webhook import test_webhook
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +45,11 @@ class Room(BaseModel):
     is_shared: bool
 
 
+class RoomDetails(Room):
+    webhook_url: str | None
+    webhook_secret: str | None
+
+
 class Meeting(BaseModel):
     id: str
     room_name: str
@@ -64,6 +70,8 @@ class CreateRoom(BaseModel):
     recording_type: str
     recording_trigger: str
     is_shared: bool
+    webhook_url: str
+    webhook_secret: str
 
 
 class UpdateRoom(BaseModel):
@@ -76,16 +84,26 @@ class UpdateRoom(BaseModel):
     recording_type: str
     recording_trigger: str
     is_shared: bool
+    webhook_url: str
+    webhook_secret: str
 
 
 class DeletionStatus(BaseModel):
     status: str
 
 
-@router.get("/rooms", response_model=Page[Room])
+class WebhookTestResult(BaseModel):
+    success: bool
+    message: str = ""
+    error: str = ""
+    status_code: int | None = None
+    response_preview: str | None = None
+
+
+@router.get("/rooms", response_model=Page[RoomDetails])
 async def rooms_list(
     user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
-) -> list[Room]:
+) -> list[RoomDetails]:
     if not user and not settings.PUBLIC_MODE:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -99,6 +117,18 @@ async def rooms_list(
     )
 
 
+@router.get("/rooms/{room_id}", response_model=RoomDetails)
+async def rooms_get(
+    room_id: str,
+    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+):
+    user_id = user["sub"] if user else None
+    room = await rooms_controller.get_by_id_for_http(room_id, user_id=user_id)
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+    return room
+
+
 @router.post("/rooms", response_model=Room)
 async def rooms_create(
     room: CreateRoom,
@@ -117,10 +147,12 @@ async def rooms_create(
         recording_type=room.recording_type,
         recording_trigger=room.recording_trigger,
         is_shared=room.is_shared,
+        webhook_url=room.webhook_url,
+        webhook_secret=room.webhook_secret,
     )
 
 
-@router.patch("/rooms/{room_id}", response_model=Room)
+@router.patch("/rooms/{room_id}", response_model=RoomDetails)
 async def rooms_update(
     room_id: str,
     info: UpdateRoom,
@@ -209,3 +241,24 @@ async def rooms_create_meeting(
         meeting.host_room_url = ""
 
     return meeting
+
+
+@router.post("/rooms/{room_id}/webhook/test", response_model=WebhookTestResult)
+async def rooms_test_webhook(
+    room_id: str,
+    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+):
+    """Test webhook configuration by sending a sample payload."""
+    user_id = user["sub"] if user else None
+
+    room = await rooms_controller.get_by_id(room_id)
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    if user_id and room.user_id != user_id:
+        raise HTTPException(
+            status_code=403, detail="Not authorized to test this room's webhook"
+        )
+
+    result = await test_webhook(room_id)
+    return WebhookTestResult(**result)

server/reflector/views/transcripts.py

@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi_pagination import Page
 from fastapi_pagination.ext.databases import apaginate
 from jose import jwt
-from pydantic import BaseModel, Field, field_serializer
+from pydantic import BaseModel, Field, constr, field_serializer
 
 import reflector.auth as auth
 from reflector.db import get_database
@@ -19,10 +19,10 @@ from reflector.db.search import (
     SearchOffsetBase,
     SearchParameters,
     SearchQuery,
-    SearchQueryBase,
     SearchResult,
     SearchTotal,
     search_controller,
+    search_query_adapter,
 )
 from reflector.db.transcripts import (
     SourceKind,
@@ -114,7 +114,19 @@ class DeletionStatus(BaseModel):
     status: str
 
 
-SearchQueryParam = Annotated[SearchQueryBase, Query(description="Search query text")]
+SearchQueryParamBase = constr(min_length=0, strip_whitespace=True)
+SearchQueryParam = Annotated[
+    SearchQueryParamBase, Query(description="Search query text")
+]
+
+
+# http and api standards accept "q="; we would like to handle it as the absence of query, not as "empty string query"
+def parse_search_query_param(q: SearchQueryParam) -> SearchQuery | None:
+    if q == "":
+        return None
+    return search_query_adapter.validate_python(q)
+
+
 SearchLimitParam = Annotated[SearchLimitBase, Query(description="Results per page")]
 SearchOffsetParam = Annotated[
     SearchOffsetBase, Query(description="Number of results to skip")
@@ -124,7 +136,7 @@ SearchOffsetParam = Annotated[
 class SearchResponse(BaseModel):
     results: list[SearchResult]
     total: SearchTotal
-    query: SearchQuery
+    query: SearchQuery | None = None
     limit: SearchLimit
     offset: SearchOffset
 
@@ -174,7 +186,7 @@ async def transcripts_search(
     user_id = user["sub"] if user else None
 
     search_params = SearchParameters(
-        query_text=q,
+        query_text=parse_search_query_param(q),
         limit=limit,
         offset=offset,
         user_id=user_id,

server/reflector/views/transcripts_process.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 
 import reflector.auth as auth
 from reflector.db.transcripts import transcripts_controller
-from reflector.pipelines.main_live_pipeline import task_pipeline_process
+from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
 
 router = APIRouter()
 
@@ -40,7 +40,7 @@ async def transcript_process(
         return ProcessStatus(status="already running")
 
     # schedule a background task process the file
-    task_pipeline_process.delay(transcript_id=transcript_id)
+    task_pipeline_file_process.delay(transcript_id=transcript_id)
 
     return ProcessStatus(status="ok")
 

server/reflector/views/transcripts_upload.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 
 import reflector.auth as auth
 from reflector.db.transcripts import transcripts_controller
-from reflector.pipelines.main_live_pipeline import task_pipeline_process
+from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
 
 router = APIRouter()
 
@@ -92,6 +92,6 @@ async def transcript_record_upload(
     await transcripts_controller.update(transcript, {"status": "uploaded"})
 
     # launch a background task to process the file
-    task_pipeline_process.delay(transcript_id=transcript_id)
+    task_pipeline_file_process.delay(transcript_id=transcript_id)
 
     return UploadStatus(status="ok")

server/reflector/worker/app.py

@@ -19,6 +19,7 @@ else:
             "reflector.pipelines.main_live_pipeline",
             "reflector.worker.healthcheck",
             "reflector.worker.process",
+            "reflector.worker.cleanup",
         ]
     )
 
@@ -38,6 +39,16 @@ else:
         },
     }
 
+    if settings.PUBLIC_MODE:
+        app.conf.beat_schedule["cleanup_old_public_data"] = {
+            "task": "reflector.worker.cleanup.cleanup_old_public_data_task",
+            "schedule": crontab(hour=3, minute=0),
+        }
+        logger.info(
+            "Public mode cleanup enabled",
+            retention_days=settings.PUBLIC_DATA_RETENTION_DAYS,
+        )
+
     if settings.HEALTHCHECK_URL:
         app.conf.beat_schedule["healthcheck_ping"] = {
             "task": "reflector.worker.healthcheck.healthcheck_ping",

server/reflector/worker/cleanup.py

@@ -0,0 +1,156 @@
+"""
+Main task for cleanup old public data.
+
+Deletes old anonymous transcripts and their associated meetings/recordings.
+Transcripts are the main entry point - any associated data is also removed.
+"""
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+from typing import TypedDict
+
+import structlog
+from celery import shared_task
+from databases import Database
+from pydantic.types import PositiveInt
+
+from reflector.asynctask import asynctask
+from reflector.db import get_database
+from reflector.db.meetings import meetings
+from reflector.db.recordings import recordings
+from reflector.db.transcripts import transcripts, transcripts_controller
+from reflector.settings import settings
+from reflector.storage import get_recordings_storage
+
+logger = structlog.get_logger(__name__)
+
+
+class CleanupStats(TypedDict):
+    """Statistics for cleanup operation."""
+
+    transcripts_deleted: int
+    meetings_deleted: int
+    recordings_deleted: int
+    errors: list[str]
+
+
+async def delete_single_transcript(
+    db: Database, transcript_data: dict, stats: CleanupStats
+):
+    transcript_id = transcript_data["id"]
+    meeting_id = transcript_data["meeting_id"]
+    recording_id = transcript_data["recording_id"]
+
+    try:
+        async with db.transaction(isolation="serializable"):
+            if meeting_id:
+                await db.execute(meetings.delete().where(meetings.c.id == meeting_id))
+                stats["meetings_deleted"] += 1
+                logger.info("Deleted associated meeting", meeting_id=meeting_id)
+
+            if recording_id:
+                recording = await db.fetch_one(
+                    recordings.select().where(recordings.c.id == recording_id)
+                )
+                if recording:
+                    try:
+                        await get_recordings_storage().delete_file(
+                            recording["object_key"]
+                        )
+                    except Exception as storage_error:
+                        logger.warning(
+                            "Failed to delete recording from storage",
+                            recording_id=recording_id,
+                            object_key=recording["object_key"],
+                            error=str(storage_error),
+                        )
+
+                    await db.execute(
+                        recordings.delete().where(recordings.c.id == recording_id)
+                    )
+                    stats["recordings_deleted"] += 1
+                    logger.info(
+                        "Deleted associated recording", recording_id=recording_id
+                    )
+
+            await transcripts_controller.remove_by_id(transcript_id)
+            stats["transcripts_deleted"] += 1
+            logger.info(
+                "Deleted transcript",
+                transcript_id=transcript_id,
+                created_at=transcript_data["created_at"].isoformat(),
+            )
+    except Exception as e:
+        error_msg = f"Failed to delete transcript {transcript_id}: {str(e)}"
+        logger.error(error_msg, exc_info=e)
+        stats["errors"].append(error_msg)
+
+
+async def cleanup_old_transcripts(
+    db: Database, cutoff_date: datetime, stats: CleanupStats
+):
+    """Delete old anonymous transcripts and their associated recordings/meetings."""
+    query = transcripts.select().where(
+        (transcripts.c.created_at < cutoff_date) & (transcripts.c.user_id.is_(None))
+    )
+    old_transcripts = await db.fetch_all(query)
+
+    logger.info(f"Found {len(old_transcripts)} old transcripts to delete")
+
+    for transcript_data in old_transcripts:
+        await delete_single_transcript(db, transcript_data, stats)
+
+
+def log_cleanup_results(stats: CleanupStats):
+    logger.info(
+        "Cleanup completed",
+        transcripts_deleted=stats["transcripts_deleted"],
+        meetings_deleted=stats["meetings_deleted"],
+        recordings_deleted=stats["recordings_deleted"],
+        errors_count=len(stats["errors"]),
+    )
+
+    if stats["errors"]:
+        logger.warning(
+            "Cleanup completed with errors",
+            errors=stats["errors"][:10],
+        )
+
+
+async def cleanup_old_public_data(
+    days: PositiveInt | None = None,
+) -> CleanupStats | None:
+    if days is None:
+        days = settings.PUBLIC_DATA_RETENTION_DAYS
+
+    if not settings.PUBLIC_MODE:
+        logger.info("Skipping cleanup - not a public instance")
+        return None
+
+    cutoff_date = datetime.now(timezone.utc) - timedelta(days=days)
+    logger.info(
+        "Starting cleanup of old public data",
+        cutoff_date=cutoff_date.isoformat(),
+    )
+
+    stats: CleanupStats = {
+        "transcripts_deleted": 0,
+        "meetings_deleted": 0,
+        "recordings_deleted": 0,
+        "errors": [],
+    }
+
+    db = get_database()
+    await cleanup_old_transcripts(db, cutoff_date, stats)
+
+    log_cleanup_results(stats)
+    return stats
+
+
+@shared_task(
+    autoretry_for=(Exception,),
+    retry_kwargs={"max_retries": 3, "countdown": 300},
+)
+@asynctask
+def cleanup_old_public_data_task(days: int | None = None):
+    asyncio.run(cleanup_old_public_data(days=days))

server/reflector/worker/webhook.py

@@ -0,0 +1,258 @@
+"""Webhook task for sending transcript notifications."""
+
+import hashlib
+import hmac
+import json
+import uuid
+from datetime import datetime, timezone
+
+import httpx
+import structlog
+from celery import shared_task
+from celery.utils.log import get_task_logger
+
+from reflector.db.rooms import rooms_controller
+from reflector.db.transcripts import transcripts_controller
+from reflector.pipelines.main_live_pipeline import asynctask
+from reflector.settings import settings
+from reflector.utils.webvtt import topics_to_webvtt
+
+logger = structlog.wrap_logger(get_task_logger(__name__))
+
+
+def generate_webhook_signature(payload: bytes, secret: str, timestamp: str) -> str:
+    """Generate HMAC signature for webhook payload."""
+    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
+    hmac_obj = hmac.new(
+        secret.encode("utf-8"),
+        signed_payload.encode("utf-8"),
+        hashlib.sha256,
+    )
+    return hmac_obj.hexdigest()
+
+
+@shared_task(
+    bind=True,
+    max_retries=30,
+    default_retry_delay=60,
+    retry_backoff=True,
+    retry_backoff_max=3600,  # Max 1 hour between retries
+)
+@asynctask
+async def send_transcript_webhook(
+    self,
+    transcript_id: str,
+    room_id: str,
+    event_id: str,
+):
+    log = logger.bind(
+        transcript_id=transcript_id,
+        room_id=room_id,
+        retry_count=self.request.retries,
+    )
+
+    try:
+        # Fetch transcript and room
+        transcript = await transcripts_controller.get_by_id(transcript_id)
+        if not transcript:
+            log.error("Transcript not found, skipping webhook")
+            return
+
+        room = await rooms_controller.get_by_id(room_id)
+        if not room:
+            log.error("Room not found, skipping webhook")
+            return
+
+        if not room.webhook_url:
+            log.info("No webhook URL configured for room, skipping")
+            return
+
+        # Generate WebVTT content from topics
+        topics_data = []
+
+        if transcript.topics:
+            # Build topics data with diarized content per topic
+            for topic in transcript.topics:
+                topic_webvtt = topics_to_webvtt([topic]) if topic.words else ""
+                topics_data.append(
+                    {
+                        "title": topic.title,
+                        "summary": topic.summary,
+                        "timestamp": topic.timestamp,
+                        "duration": topic.duration,
+                        "webvtt": topic_webvtt,
+                    }
+                )
+
+        # Build webhook payload
+        frontend_url = f"{settings.UI_BASE_URL}/transcripts/{transcript.id}"
+        participants = [
+            {"id": p.id, "name": p.name, "speaker": p.speaker}
+            for p in (transcript.participants or [])
+        ]
+        payload_data = {
+            "event": "transcript.completed",
+            "event_id": event_id,
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "transcript": {
+                "id": transcript.id,
+                "room_id": transcript.room_id,
+                "created_at": transcript.created_at.isoformat(),
+                "duration": transcript.duration,
+                "title": transcript.title,
+                "short_summary": transcript.short_summary,
+                "long_summary": transcript.long_summary,
+                "webvtt": transcript.webvtt,
+                "topics": topics_data,
+                "participants": participants,
+                "source_language": transcript.source_language,
+                "target_language": transcript.target_language,
+                "status": transcript.status,
+                "frontend_url": frontend_url,
+            },
+            "room": {
+                "id": room.id,
+                "name": room.name,
+            },
+        }
+
+        # Convert to JSON
+        payload_json = json.dumps(payload_data, separators=(",", ":"))
+        payload_bytes = payload_json.encode("utf-8")
+
+        # Generate signature if secret is configured
+        headers = {
+            "Content-Type": "application/json",
+            "User-Agent": "Reflector-Webhook/1.0",
+            "X-Webhook-Event": "transcript.completed",
+            "X-Webhook-Retry": str(self.request.retries),
+        }
+
+        if room.webhook_secret:
+            timestamp = str(int(datetime.now(timezone.utc).timestamp()))
+            signature = generate_webhook_signature(
+                payload_bytes, room.webhook_secret, timestamp
+            )
+            headers["X-Webhook-Signature"] = f"t={timestamp},v1={signature}"
+
+        # Send webhook with timeout
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            log.info(
+                "Sending webhook",
+                url=room.webhook_url,
+                payload_size=len(payload_bytes),
+            )
+
+            response = await client.post(
+                room.webhook_url,
+                content=payload_bytes,
+                headers=headers,
+            )
+
+            response.raise_for_status()
+
+            log.info(
+                "Webhook sent successfully",
+                status_code=response.status_code,
+                response_size=len(response.content),
+            )
+
+    except httpx.HTTPStatusError as e:
+        log.error(
+            "Webhook failed with HTTP error",
+            status_code=e.response.status_code,
+            response_text=e.response.text[:500],  # First 500 chars
+        )
+
+        # Don't retry on client errors (4xx)
+        if 400 <= e.response.status_code < 500:
+            log.error("Client error, not retrying")
+            return
+
+        # Retry on server errors (5xx)
+        raise self.retry(exc=e)
+
+    except (httpx.ConnectError, httpx.TimeoutException) as e:
+        # Retry on network errors
+        log.error("Webhook failed with connection error", error=str(e))
+        raise self.retry(exc=e)
+
+    except Exception as e:
+        # Retry on unexpected errors
+        log.exception("Unexpected error in webhook task", error=str(e))
+        raise self.retry(exc=e)
+
+
+async def test_webhook(room_id: str) -> dict:
+    """
+    Test webhook configuration by sending a sample payload.
+    Returns immediately with success/failure status.
+    This is the shared implementation used by both the API endpoint and Celery task.
+    """
+    try:
+        room = await rooms_controller.get_by_id(room_id)
+        if not room:
+            return {"success": False, "error": "Room not found"}
+
+        if not room.webhook_url:
+            return {"success": False, "error": "No webhook URL configured"}
+
+        now = (datetime.now(timezone.utc).isoformat(),)
+        payload_data = {
+            "event": "test",
+            "event_id": uuid.uuid4().hex,
+            "timestamp": now,
+            "message": "This is a test webhook from Reflector",
+            "room": {
+                "id": room.id,
+                "name": room.name,
+            },
+        }
+
+        payload_json = json.dumps(payload_data, separators=(",", ":"))
+        payload_bytes = payload_json.encode("utf-8")
+
+        # Generate headers with signature
+        headers = {
+            "Content-Type": "application/json",
+            "User-Agent": "Reflector-Webhook/1.0",
+            "X-Webhook-Event": "test",
+        }
+
+        if room.webhook_secret:
+            timestamp = str(int(datetime.now(timezone.utc).timestamp()))
+            signature = generate_webhook_signature(
+                payload_bytes, room.webhook_secret, timestamp
+            )
+            headers["X-Webhook-Signature"] = f"t={timestamp},v1={signature}"
+
+        # Send test webhook with short timeout
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.post(
+                room.webhook_url,
+                content=payload_bytes,
+                headers=headers,
+            )
+
+            return {
+                "success": response.is_success,
+                "status_code": response.status_code,
+                "message": f"Webhook test {'successful' if response.is_success else 'failed'}",
+                "response_preview": response.text if response.text else None,
+            }
+
+    except httpx.TimeoutException:
+        return {
+            "success": False,
+            "error": "Webhook request timed out (10 seconds)",
+        }
+    except httpx.ConnectError as e:
+        return {
+            "success": False,
+            "error": f"Could not connect to webhook URL: {str(e)}",
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "error": f"Unexpected error: {str(e)}",
+        }

server/tests/conftest.py

@@ -178,6 +178,63 @@ async def dummy_diarization():
         yield
 
 
+@pytest.fixture
+async def dummy_file_transcript():
+    from reflector.processors.file_transcript import FileTranscriptProcessor
+    from reflector.processors.types import Transcript, Word
+
+    class TestFileTranscriptProcessor(FileTranscriptProcessor):
+        async def _transcript(self, data):
+            return Transcript(
+                text="Hello world. How are you today?",
+                words=[
+                    Word(start=0.0, end=0.5, text="Hello", speaker=0),
+                    Word(start=0.5, end=0.6, text=" ", speaker=0),
+                    Word(start=0.6, end=1.0, text="world", speaker=0),
+                    Word(start=1.0, end=1.1, text=".", speaker=0),
+                    Word(start=1.1, end=1.2, text=" ", speaker=0),
+                    Word(start=1.2, end=1.5, text="How", speaker=0),
+                    Word(start=1.5, end=1.6, text=" ", speaker=0),
+                    Word(start=1.6, end=1.8, text="are", speaker=0),
+                    Word(start=1.8, end=1.9, text=" ", speaker=0),
+                    Word(start=1.9, end=2.1, text="you", speaker=0),
+                    Word(start=2.1, end=2.2, text=" ", speaker=0),
+                    Word(start=2.2, end=2.5, text="today", speaker=0),
+                    Word(start=2.5, end=2.6, text="?", speaker=0),
+                ],
+            )
+
+    with patch(
+        "reflector.processors.file_transcript_auto.FileTranscriptAutoProcessor.__new__"
+    ) as mock_auto:
+        mock_auto.return_value = TestFileTranscriptProcessor()
+        yield
+
+
+@pytest.fixture
+async def dummy_file_diarization():
+    from reflector.processors.file_diarization import (
+        FileDiarizationOutput,
+        FileDiarizationProcessor,
+    )
+    from reflector.processors.types import DiarizationSegment
+
+    class TestFileDiarizationProcessor(FileDiarizationProcessor):
+        async def _diarize(self, data):
+            return FileDiarizationOutput(
+                diarization=[
+                    DiarizationSegment(start=0.0, end=1.1, speaker=0),
+                    DiarizationSegment(start=1.2, end=2.6, speaker=1),
+                ]
+            )
+
+    with patch(
+        "reflector.processors.file_diarization_auto.FileDiarizationAutoProcessor.__new__"
+    ) as mock_auto:
+        mock_auto.return_value = TestFileDiarizationProcessor()
+        yield
+
+
 @pytest.fixture
 async def dummy_transcript_translator():
     from reflector.processors.transcript_translator import TranscriptTranslatorProcessor
@@ -238,9 +295,13 @@ async def dummy_storage():
     with (
         patch("reflector.storage.base.Storage.get_instance") as mock_storage,
         patch("reflector.storage.get_transcripts_storage") as mock_get_transcripts,
+        patch(
+            "reflector.pipelines.main_file_pipeline.get_transcripts_storage"
+        ) as mock_get_transcripts2,
     ):
         mock_storage.return_value = dummy
         mock_get_transcripts.return_value = dummy
+        mock_get_transcripts2.return_value = dummy
         yield
 
 
@@ -260,7 +321,10 @@ def celery_config():
 
 @pytest.fixture(scope="session")
 def celery_includes():
-    return ["reflector.pipelines.main_live_pipeline"]
+    return [
+        "reflector.pipelines.main_live_pipeline",
+        "reflector.pipelines.main_file_pipeline",
+    ]
 
 
 @pytest.fixture
@@ -302,7 +366,7 @@ async def fake_transcript_with_topics(tmpdir, client):
     transcript = await transcripts_controller.get_by_id(tid)
     assert transcript is not None
 
-    await transcripts_controller.update(transcript, {"status": "finished"})
+    await transcripts_controller.update(transcript, {"status": "ended"})
 
     # manually copy a file at the expected location
     audio_filename = transcript.audio_mp3_filename

server/tests/test_cleanup.py

@@ -0,0 +1,287 @@
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from reflector.db.recordings import Recording, recordings_controller
+from reflector.db.transcripts import SourceKind, transcripts_controller
+from reflector.worker.cleanup import cleanup_old_public_data
+
+
+@pytest.mark.asyncio
+async def test_cleanup_old_public_data_skips_when_not_public():
+    """Test that cleanup is skipped when PUBLIC_MODE is False."""
+    with patch("reflector.worker.cleanup.settings") as mock_settings:
+        mock_settings.PUBLIC_MODE = False
+
+        result = await cleanup_old_public_data()
+
+        # Should return early without doing anything
+        assert result is None
+
+
+@pytest.mark.asyncio
+async def test_cleanup_old_public_data_deletes_old_anonymous_transcripts():
+    """Test that old anonymous transcripts are deleted."""
+    # Create old and new anonymous transcripts
+    old_date = datetime.now(timezone.utc) - timedelta(days=8)
+    new_date = datetime.now(timezone.utc) - timedelta(days=2)
+
+    # Create old anonymous transcript (should be deleted)
+    old_transcript = await transcripts_controller.add(
+        name="Old Anonymous Transcript",
+        source_kind=SourceKind.FILE,
+        user_id=None,  # Anonymous
+    )
+    # Manually update created_at to be old
+    from reflector.db import get_database
+    from reflector.db.transcripts import transcripts
+
+    await get_database().execute(
+        transcripts.update()
+        .where(transcripts.c.id == old_transcript.id)
+        .values(created_at=old_date)
+    )
+
+    # Create new anonymous transcript (should NOT be deleted)
+    new_transcript = await transcripts_controller.add(
+        name="New Anonymous Transcript",
+        source_kind=SourceKind.FILE,
+        user_id=None,  # Anonymous
+    )
+
+    # Create old transcript with user (should NOT be deleted)
+    old_user_transcript = await transcripts_controller.add(
+        name="Old User Transcript",
+        source_kind=SourceKind.FILE,
+        user_id="user123",
+    )
+    await get_database().execute(
+        transcripts.update()
+        .where(transcripts.c.id == old_user_transcript.id)
+        .values(created_at=old_date)
+    )
+
+    with patch("reflector.worker.cleanup.settings") as mock_settings:
+        mock_settings.PUBLIC_MODE = True
+        mock_settings.PUBLIC_DATA_RETENTION_DAYS = 7
+
+        # Mock the storage deletion
+        with patch("reflector.db.transcripts.get_transcripts_storage") as mock_storage:
+            mock_storage.return_value.delete_file = AsyncMock()
+
+            result = await cleanup_old_public_data()
+
+    # Check results
+    assert result["transcripts_deleted"] == 1
+    assert result["errors"] == []
+
+    # Verify old anonymous transcript was deleted
+    assert await transcripts_controller.get_by_id(old_transcript.id) is None
+
+    # Verify new anonymous transcript still exists
+    assert await transcripts_controller.get_by_id(new_transcript.id) is not None
+
+    # Verify user transcript still exists
+    assert await transcripts_controller.get_by_id(old_user_transcript.id) is not None
+
+
+@pytest.mark.asyncio
+async def test_cleanup_deletes_associated_meeting_and_recording():
+    """Test that meetings and recordings associated with old transcripts are deleted."""
+    from reflector.db import get_database
+    from reflector.db.meetings import meetings
+    from reflector.db.transcripts import transcripts
+
+    old_date = datetime.now(timezone.utc) - timedelta(days=8)
+
+    # Create a meeting
+    meeting_id = "test-meeting-for-transcript"
+    await get_database().execute(
+        meetings.insert().values(
+            id=meeting_id,
+            room_name="Meeting with Transcript",
+            room_url="https://example.com/meeting",
+            host_room_url="https://example.com/meeting-host",
+            start_date=old_date,
+            end_date=old_date + timedelta(hours=1),
+            user_id=None,
+            room_id=None,
+        )
+    )
+
+    # Create a recording
+    recording = await recordings_controller.create(
+        Recording(
+            bucket_name="test-bucket",
+            object_key="test-recording.mp4",
+            recorded_at=old_date,
+        )
+    )
+
+    # Create an old transcript with both meeting and recording
+    old_transcript = await transcripts_controller.add(
+        name="Old Transcript with Meeting and Recording",
+        source_kind=SourceKind.ROOM,
+        user_id=None,
+        meeting_id=meeting_id,
+        recording_id=recording.id,
+    )
+
+    # Update created_at to be old
+    await get_database().execute(
+        transcripts.update()
+        .where(transcripts.c.id == old_transcript.id)
+        .values(created_at=old_date)
+    )
+
+    with patch("reflector.worker.cleanup.settings") as mock_settings:
+        mock_settings.PUBLIC_MODE = True
+        mock_settings.PUBLIC_DATA_RETENTION_DAYS = 7
+
+        # Mock storage deletion
+        with patch("reflector.db.transcripts.get_transcripts_storage") as mock_storage:
+            mock_storage.return_value.delete_file = AsyncMock()
+            with patch(
+                "reflector.worker.cleanup.get_recordings_storage"
+            ) as mock_rec_storage:
+                mock_rec_storage.return_value.delete_file = AsyncMock()
+
+                result = await cleanup_old_public_data()
+
+    # Check results
+    assert result["transcripts_deleted"] == 1
+    assert result["meetings_deleted"] == 1
+    assert result["recordings_deleted"] == 1
+    assert result["errors"] == []
+
+    # Verify transcript was deleted
+    assert await transcripts_controller.get_by_id(old_transcript.id) is None
+
+    # Verify meeting was deleted
+    query = meetings.select().where(meetings.c.id == meeting_id)
+    meeting_result = await get_database().fetch_one(query)
+    assert meeting_result is None
+
+    # Verify recording was deleted
+    assert await recordings_controller.get_by_id(recording.id) is None
+
+
+@pytest.mark.asyncio
+async def test_cleanup_handles_errors_gracefully():
+    """Test that cleanup continues even when individual deletions fail."""
+    old_date = datetime.now(timezone.utc) - timedelta(days=8)
+
+    # Create multiple old transcripts
+    transcript1 = await transcripts_controller.add(
+        name="Transcript 1",
+        source_kind=SourceKind.FILE,
+        user_id=None,
+    )
+    transcript2 = await transcripts_controller.add(
+        name="Transcript 2",
+        source_kind=SourceKind.FILE,
+        user_id=None,
+    )
+
+    # Update created_at to be old
+    from reflector.db import get_database
+    from reflector.db.transcripts import transcripts
+
+    for t_id in [transcript1.id, transcript2.id]:
+        await get_database().execute(
+            transcripts.update()
+            .where(transcripts.c.id == t_id)
+            .values(created_at=old_date)
+        )
+
+    with patch("reflector.worker.cleanup.settings") as mock_settings:
+        mock_settings.PUBLIC_MODE = True
+        mock_settings.PUBLIC_DATA_RETENTION_DAYS = 7
+
+        # Mock remove_by_id to fail for the first transcript
+        original_remove = transcripts_controller.remove_by_id
+        call_count = 0
+
+        async def mock_remove_by_id(transcript_id, user_id=None):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise Exception("Simulated deletion error")
+            return await original_remove(transcript_id, user_id)
+
+        with patch.object(
+            transcripts_controller, "remove_by_id", side_effect=mock_remove_by_id
+        ):
+            result = await cleanup_old_public_data()
+
+    # Should have one successful deletion and one error
+    assert result["transcripts_deleted"] == 1
+    assert len(result["errors"]) == 1
+    assert "Failed to delete transcript" in result["errors"][0]
+
+
+@pytest.mark.asyncio
+async def test_meeting_consent_cascade_delete():
+    """Test that meeting_consent records are automatically deleted when meeting is deleted."""
+    from reflector.db import get_database
+    from reflector.db.meetings import (
+        meeting_consent,
+        meeting_consent_controller,
+        meetings,
+    )
+
+    # Create a meeting
+    meeting_id = "test-cascade-meeting"
+    await get_database().execute(
+        meetings.insert().values(
+            id=meeting_id,
+            room_name="Test Meeting for CASCADE",
+            room_url="https://example.com/cascade-test",
+            host_room_url="https://example.com/cascade-test-host",
+            start_date=datetime.now(timezone.utc),
+            end_date=datetime.now(timezone.utc) + timedelta(hours=1),
+            user_id="test-user",
+            room_id=None,
+        )
+    )
+
+    # Create consent records for this meeting
+    consent1_id = "consent-1"
+    consent2_id = "consent-2"
+
+    await get_database().execute(
+        meeting_consent.insert().values(
+            id=consent1_id,
+            meeting_id=meeting_id,
+            user_id="user1",
+            consent_given=True,
+            consent_timestamp=datetime.now(timezone.utc),
+        )
+    )
+
+    await get_database().execute(
+        meeting_consent.insert().values(
+            id=consent2_id,
+            meeting_id=meeting_id,
+            user_id="user2",
+            consent_given=False,
+            consent_timestamp=datetime.now(timezone.utc),
+        )
+    )
+
+    # Verify consent records exist
+    consents = await meeting_consent_controller.get_by_meeting_id(meeting_id)
+    assert len(consents) == 2
+
+    # Delete the meeting
+    await get_database().execute(meetings.delete().where(meetings.c.id == meeting_id))
+
+    # Verify meeting is deleted
+    query = meetings.select().where(meetings.c.id == meeting_id)
+    result = await get_database().fetch_one(query)
+    assert result is None
+
+    # Verify consent records are automatically deleted (CASCADE DELETE)
+    consents_after = await meeting_consent_controller.get_by_meeting_id(meeting_id)
+    assert len(consents_after) == 0

server/tests/test_processors_pipeline.py

@@ -1,61 +0,0 @@
-import pytest
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("enable_diarization", [False, True])
-async def test_basic_process(
-    dummy_transcript,
-    dummy_llm,
-    dummy_processors,
-    enable_diarization,
-    dummy_diarization,
-):
-    # goal is to start the server, and send rtc audio to it
-    # validate the events received
-    from pathlib import Path
-
-    from reflector.settings import settings
-    from reflector.tools.process import process_audio_file
-
-    # LLM_BACKEND no longer exists in settings
-    # settings.LLM_BACKEND = "test"
-    settings.TRANSCRIPT_BACKEND = "whisper"
-
-    # event callback
-    marks = {}
-
-    async def event_callback(event):
-        if event.processor not in marks:
-            marks[event.processor] = 0
-        marks[event.processor] += 1
-
-    # invoke the process and capture events
-    path = Path(__file__).parent / "records" / "test_mathieu_hello.wav"
-
-    if enable_diarization:
-        # Test with diarization - may fail if pyannote.audio is not installed
-        try:
-            await process_audio_file(
-                path.as_posix(), event_callback, enable_diarization=True
-            )
-        except SystemExit:
-            pytest.skip("pyannote.audio not installed - skipping diarization test")
-    else:
-        # Test without diarization - should always work
-        await process_audio_file(
-            path.as_posix(), event_callback, enable_diarization=False
-        )
-
-    print(f"Diarization: {enable_diarization}, Marks: {marks}")
-
-    # validate the events
-    # Each processor should be called for each audio segment processed
-    # The final processors (Topic, Title, Summary) should be called once at the end
-    assert marks["TranscriptLinerProcessor"] > 0
-    assert marks["TranscriptTranslatorPassthroughProcessor"] > 0
-    assert marks["TranscriptTopicDetectorProcessor"] == 1
-    assert marks["TranscriptFinalSummaryProcessor"] == 1
-    assert marks["TranscriptFinalTitleProcessor"] == 1
-
-    if enable_diarization:
-        assert marks["TestAudioDiarizationProcessor"] == 1

server/tests/test_search.py

@@ -23,7 +23,7 @@ async def test_search_postgresql_only():
     assert results == []
     assert total == 0
 
-    params_empty = SearchParameters(query_text="")
+    params_empty = SearchParameters(query_text=None)
     results_empty, total_empty = await search_controller.search_transcripts(
         params_empty
     )
@@ -34,7 +34,7 @@ async def test_search_postgresql_only():
 @pytest.mark.asyncio
 async def test_search_with_empty_query():
     """Test that empty query returns all transcripts."""
-    params = SearchParameters(query_text="")
+    params = SearchParameters(query_text=None)
     results, total = await search_controller.search_transcripts(params)
 
     assert isinstance(results, list)

server/tests/test_search_snippets.py

@@ -1,5 +1,7 @@
 """Unit tests for search snippet generation."""
 
+import pytest
+
 from reflector.db.search import (
     SnippetCandidate,
     SnippetGenerator,
@@ -512,11 +514,9 @@ data visualization and data storage"""
         )
         assert webvtt_count == 3
 
-        snippets_empty, count_empty = SnippetGenerator.combine_sources(
-            None, None, "data", max_total=3
-        )
-        assert snippets_empty == []
-        assert count_empty == 0
+        # combine_sources requires at least one source to be present
+        with pytest.raises(AssertionError, match="At least one source must be present"):
+            SnippetGenerator.combine_sources(None, None, "data", max_total=3)
 
     def test_edge_cases(self):
         """Test edge cases for the pure functions."""

server/tests/test_transcripts_audio_download.py

@@ -19,7 +19,7 @@ async def fake_transcript(tmpdir, client):
     transcript = await transcripts_controller.get_by_id(tid)
     assert transcript is not None
 
-    await transcripts_controller.update(transcript, {"status": "finished"})
+    await transcripts_controller.update(transcript, {"status": "ended"})
 
     # manually copy a file at the expected location
     audio_filename = transcript.audio_mp3_filename

server/tests/test_transcripts_process.py

@@ -29,10 +29,10 @@ async def client(app_lifespan):
 @pytest.mark.asyncio
 async def test_transcript_process(
     tmpdir,
-    whisper_transcript,
     dummy_llm,
     dummy_processors,
-    dummy_diarization,
+    dummy_file_transcript,
+    dummy_file_diarization,
     dummy_storage,
     client,
 ):
@@ -56,8 +56,8 @@ async def test_transcript_process(
     assert response.status_code == 200
     assert response.json()["status"] == "ok"
 
-    # wait for processing to finish (max 10 minutes)
-    timeout_seconds = 600  # 10 minutes
+    # wait for processing to finish (max 1 minute)
+    timeout_seconds = 60
     start_time = time.monotonic()
     while (time.monotonic() - start_time) < timeout_seconds:
         # fetch the transcript and check if it is ended
@@ -75,9 +75,10 @@ async def test_transcript_process(
     )
     assert response.status_code == 200
     assert response.json()["status"] == "ok"
+    await asyncio.sleep(2)
 
-    # wait for processing to finish (max 10 minutes)
-    timeout_seconds = 600  # 10 minutes
+    # wait for processing to finish (max 1 minute)
+    timeout_seconds = 60
     start_time = time.monotonic()
     while (time.monotonic() - start_time) < timeout_seconds:
         # fetch the transcript and check if it is ended
@@ -99,4 +100,4 @@ async def test_transcript_process(
     response = await client.get(f"/transcripts/{tid}/topics")
     assert response.status_code == 200
     assert len(response.json()) == 1
-    assert "want to share" in response.json()[0]["transcript"]
+    assert "Hello world. How are you today?" in response.json()[0]["transcript"]

server/tests/test_transcripts_upload.py

@@ -12,7 +12,8 @@ async def test_transcript_upload_file(
     tmpdir,
     dummy_llm,
     dummy_processors,
-    dummy_diarization,
+    dummy_file_transcript,
+    dummy_file_diarization,
     dummy_storage,
     client,
 ):
@@ -36,8 +37,8 @@ async def test_transcript_upload_file(
     assert response.status_code == 200
     assert response.json()["status"] == "ok"
 
-    # wait the processing to finish (max 10 minutes)
-    timeout_seconds = 600  # 10 minutes
+    # wait the processing to finish (max 1 minute)
+    timeout_seconds = 60
     start_time = time.monotonic()
     while (time.monotonic() - start_time) < timeout_seconds:
         # fetch the transcript and check if it is ended
@@ -47,7 +48,7 @@ async def test_transcript_upload_file(
             break
         await asyncio.sleep(1)
     else:
-        pytest.fail(f"Processing timed out after {timeout_seconds} seconds")
+        return pytest.fail(f"Processing timed out after {timeout_seconds} seconds")
 
     # check the transcript is ended
     transcript = resp.json()
@@ -59,4 +60,4 @@ async def test_transcript_upload_file(
     response = await client.get(f"/transcripts/{tid}/topics")
     assert response.status_code == 200
     assert len(response.json()) == 1
-    assert "want to share" in response.json()[0]["transcript"]
+    assert "Hello world. How are you today?" in response.json()[0]["transcript"]

www/app/(app)/rooms/_components/RoomCards.tsx

@@ -12,11 +12,11 @@ import {
   HStack,
 } from "@chakra-ui/react";
 import { LuLink } from "react-icons/lu";
-import { Room } from "../../../api";
+import { RoomDetails } from "../../../api";
 import { RoomActionsMenu } from "./RoomActionsMenu";
 
 interface RoomCardsProps {
-  rooms: Room[];
+  rooms: RoomDetails[];
   linkCopied: string;
   onCopyUrl: (roomName: string) => void;
   onEdit: (roomId: string, roomData: any) => void;

www/app/(app)/rooms/_components/RoomList.tsx

@@ -1,11 +1,11 @@
 import { Box, Heading, Text, VStack } from "@chakra-ui/react";
-import { Room } from "../../../api";
+import { RoomDetails } from "../../../api";
 import { RoomTable } from "./RoomTable";
 import { RoomCards } from "./RoomCards";
 
 interface RoomListProps {
   title: string;
-  rooms: Room[];
+  rooms: RoomDetails[];
   linkCopied: string;
   onCopyUrl: (roomName: string) => void;
   onEdit: (roomId: string, roomData: any) => void;

www/app/(app)/rooms/_components/RoomTable.tsx

@@ -9,11 +9,11 @@ import {
   Spinner,
 } from "@chakra-ui/react";
 import { LuLink } from "react-icons/lu";
-import { Room } from "../../../api";
+import { RoomDetails } from "../../../api";
 import { RoomActionsMenu } from "./RoomActionsMenu";
 
 interface RoomTableProps {
-  rooms: Room[];
+  rooms: RoomDetails[];
   linkCopied: string;
   onCopyUrl: (roomName: string) => void;
   onEdit: (roomId: string, roomData: any) => void;

www/app/(app)/rooms/page.tsx

@@ -11,13 +11,15 @@ import {
   Input,
   Select,
   Spinner,
+  IconButton,
   createListCollection,
   useDisclosure,
 } from "@chakra-ui/react";
 import { useEffect, useState } from "react";
+import { LuEye, LuEyeOff } from "react-icons/lu";
 import useApi from "../../lib/useApi";
 import useRoomList from "./useRoomList";
-import { ApiError, Room } from "../../api";
+import { ApiError, RoomDetails } from "../../api";
 import { RoomList } from "./_components/RoomList";
 import { PaginationPage } from "../browse/_components/Pagination";
 
@@ -55,6 +57,8 @@ const roomInitialState = {
   recordingType: "cloud",
   recordingTrigger: "automatic-2nd-participant",
   isShared: false,
+  webhookUrl: "",
+  webhookSecret: "",
 };
 
 export default function RoomsList() {
@@ -83,6 +87,11 @@ export default function RoomsList() {
   const [topics, setTopics] = useState<Topic[]>([]);
   const [nameError, setNameError] = useState("");
   const [linkCopied, setLinkCopied] = useState("");
+  const [testingWebhook, setTestingWebhook] = useState(false);
+  const [webhookTestResult, setWebhookTestResult] = useState<string | null>(
+    null,
+  );
+  const [showWebhookSecret, setShowWebhookSecret] = useState(false);
   interface Stream {
     stream_id: number;
     name: string;
@@ -155,6 +164,69 @@ export default function RoomsList() {
     }, 2000);
   };
 
+  const handleCloseDialog = () => {
+    setShowWebhookSecret(false);
+    setWebhookTestResult(null);
+    onClose();
+  };
+
+  const handleTestWebhook = async () => {
+    if (!room.webhookUrl || !editRoomId) {
+      setWebhookTestResult("Please enter a webhook URL first");
+      return;
+    }
+
+    setTestingWebhook(true);
+    setWebhookTestResult(null);
+
+    try {
+      const response = await api?.v1RoomsTestWebhook({
+        roomId: editRoomId,
+      });
+
+      if (response?.success) {
+        setWebhookTestResult(
+          `✅ Webhook test successful! Status: ${response.status_code}`,
+        );
+      } else {
+        let errorMsg = `❌ Webhook test failed`;
+        if (response?.status_code) {
+          errorMsg += ` (Status: ${response.status_code})`;
+        }
+        if (response?.error) {
+          errorMsg += `: ${response.error}`;
+        } else if (response?.response_preview) {
+          // Try to parse and extract meaningful error from response
+          // Specific to N8N at the moment, as there is no specification for that
+          // We could just display as is, but decided here to dig a little bit more.
+          try {
+            const preview = JSON.parse(response.response_preview);
+            if (preview.message) {
+              errorMsg += `: ${preview.message}`;
+            }
+          } catch {
+            // If not JSON, just show the preview text (truncated)
+            const previewText = response.response_preview.substring(0, 150);
+            errorMsg += `: ${previewText}`;
+          }
+        } else if (response?.message) {
+          errorMsg += `: ${response.message}`;
+        }
+        setWebhookTestResult(errorMsg);
+      }
+    } catch (error) {
+      console.error("Error testing webhook:", error);
+      setWebhookTestResult("❌ Failed to test webhook. Please check your URL.");
+    } finally {
+      setTestingWebhook(false);
+    }
+
+    // Clear result after 5 seconds
+    setTimeout(() => {
+      setWebhookTestResult(null);
+    }, 5000);
+  };
+
   const handleSaveRoom = async () => {
     try {
       if (RESERVED_PATHS.includes(room.name)) {
@@ -172,6 +244,8 @@ export default function RoomsList() {
         recording_type: room.recordingType,
         recording_trigger: room.recordingTrigger,
         is_shared: room.isShared,
+        webhook_url: room.webhookUrl,
+        webhook_secret: room.webhookSecret,
       };
 
       if (isEditing) {
@@ -190,7 +264,7 @@ export default function RoomsList() {
       setEditRoomId("");
       setNameError("");
       refetch();
-      onClose();
+      handleCloseDialog();
     } catch (err) {
       if (
         err instanceof ApiError &&
@@ -206,18 +280,46 @@ export default function RoomsList() {
     }
   };
 
-  const handleEditRoom = (roomId, roomData) => {
-    setRoom({
-      name: roomData.name,
-      zulipAutoPost: roomData.zulip_auto_post,
-      zulipStream: roomData.zulip_stream,
-      zulipTopic: roomData.zulip_topic,
-      isLocked: roomData.is_locked,
-      roomMode: roomData.room_mode,
-      recordingType: roomData.recording_type,
-      recordingTrigger: roomData.recording_trigger,
-      isShared: roomData.is_shared,
-    });
+  const handleEditRoom = async (roomId, roomData) => {
+    // Reset states
+    setShowWebhookSecret(false);
+    setWebhookTestResult(null);
+
+    // Fetch full room details to get webhook fields
+    try {
+      const detailedRoom = await api?.v1RoomsGet({ roomId });
+      if (detailedRoom) {
+        setRoom({
+          name: detailedRoom.name,
+          zulipAutoPost: detailedRoom.zulip_auto_post,
+          zulipStream: detailedRoom.zulip_stream,
+          zulipTopic: detailedRoom.zulip_topic,
+          isLocked: detailedRoom.is_locked,
+          roomMode: detailedRoom.room_mode,
+          recordingType: detailedRoom.recording_type,
+          recordingTrigger: detailedRoom.recording_trigger,
+          isShared: detailedRoom.is_shared,
+          webhookUrl: detailedRoom.webhook_url || "",
+          webhookSecret: detailedRoom.webhook_secret || "",
+        });
+      }
+    } catch (error) {
+      console.error("Failed to fetch room details, using list data:", error);
+      // Fallback to using the data from the list
+      setRoom({
+        name: roomData.name,
+        zulipAutoPost: roomData.zulip_auto_post,
+        zulipStream: roomData.zulip_stream,
+        zulipTopic: roomData.zulip_topic,
+        isLocked: roomData.is_locked,
+        roomMode: roomData.room_mode,
+        recordingType: roomData.recording_type,
+        recordingTrigger: roomData.recording_trigger,
+        isShared: roomData.is_shared,
+        webhookUrl: roomData.webhook_url || "",
+        webhookSecret: roomData.webhook_secret || "",
+      });
+    }
     setEditRoomId(roomId);
     setIsEditing(true);
     setNameError("");
@@ -250,9 +352,9 @@ export default function RoomsList() {
     });
   };
 
-  const myRooms: Room[] =
+  const myRooms: RoomDetails[] =
     response?.items.filter((roomData) => !roomData.is_shared) || [];
-  const sharedRooms: Room[] =
+  const sharedRooms: RoomDetails[] =
     response?.items.filter((roomData) => roomData.is_shared) || [];
 
   if (loading && !response)
@@ -287,6 +389,8 @@ export default function RoomsList() {
             setIsEditing(false);
             setRoom(roomInitialState);
             setNameError("");
+            setShowWebhookSecret(false);
+            setWebhookTestResult(null);
             onOpen();
           }}
         >
@@ -296,7 +400,7 @@ export default function RoomsList() {
 
       <Dialog.Root
         open={open}
-        onOpenChange={(e) => (e.open ? onOpen() : onClose())}
+        onOpenChange={(e) => (e.open ? onOpen() : handleCloseDialog())}
         size="lg"
       >
         <Dialog.Backdrop />
@@ -533,6 +637,109 @@ export default function RoomsList() {
                   </Select.Positioner>
                 </Select.Root>
               </Field.Root>
+
+              {/* Webhook Configuration Section */}
+              <Field.Root mt={8}>
+                <Field.Label>Webhook URL</Field.Label>
+                <Input
+                  name="webhookUrl"
+                  type="url"
+                  placeholder="https://example.com/webhook"
+                  value={room.webhookUrl}
+                  onChange={handleRoomChange}
+                />
+                <Field.HelperText>
+                  Optional: URL to receive notifications when transcripts are
+                  ready
+                </Field.HelperText>
+              </Field.Root>
+
+              {room.webhookUrl && (
+                <>
+                  <Field.Root mt={4}>
+                    <Field.Label>Webhook Secret</Field.Label>
+                    <Flex gap={2}>
+                      <Input
+                        name="webhookSecret"
+                        type={showWebhookSecret ? "text" : "password"}
+                        value={room.webhookSecret}
+                        onChange={handleRoomChange}
+                        placeholder={
+                          isEditing && room.webhookSecret
+                            ? "••••••••"
+                            : "Leave empty to auto-generate"
+                        }
+                        flex="1"
+                      />
+                      {isEditing && room.webhookSecret && (
+                        <IconButton
+                          size="sm"
+                          variant="ghost"
+                          aria-label={
+                            showWebhookSecret ? "Hide secret" : "Show secret"
+                          }
+                          onClick={() =>
+                            setShowWebhookSecret(!showWebhookSecret)
+                          }
+                        >
+                          {showWebhookSecret ? <LuEyeOff /> : <LuEye />}
+                        </IconButton>
+                      )}
+                    </Flex>
+                    <Field.HelperText>
+                      Used for HMAC signature verification (auto-generated if
+                      left empty)
+                    </Field.HelperText>
+                  </Field.Root>
+
+                  {isEditing && (
+                    <>
+                      <Flex
+                        mt={2}
+                        gap={2}
+                        alignItems="flex-start"
+                        direction="column"
+                      >
+                        <Button
+                          size="sm"
+                          variant="outline"
+                          onClick={handleTestWebhook}
+                          disabled={testingWebhook || !room.webhookUrl}
+                        >
+                          {testingWebhook ? (
+                            <>
+                              <Spinner size="xs" mr={2} />
+                              Testing...
+                            </>
+                          ) : (
+                            "Test Webhook"
+                          )}
+                        </Button>
+                        {webhookTestResult && (
+                          <div
+                            style={{
+                              fontSize: "14px",
+                              wordBreak: "break-word",
+                              maxWidth: "100%",
+                              padding: "8px",
+                              borderRadius: "4px",
+                              backgroundColor: webhookTestResult.startsWith(
+                                "✅",
+                              )
+                                ? "#f0fdf4"
+                                : "#fef2f2",
+                              border: `1px solid ${webhookTestResult.startsWith("✅") ? "#86efac" : "#fca5a5"}`,
+                            }}
+                          >
+                            {webhookTestResult}
+                          </div>
+                        )}
+                      </Flex>
+                    </>
+                  )}
+                </>
+              )}
+
               <Field.Root mt={4}>
                 <Checkbox.Root
                   name="isShared"
@@ -557,7 +764,7 @@ export default function RoomsList() {
               </Field.Root>
             </Dialog.Body>
             <Dialog.Footer>
-              <Button variant="ghost" onClick={onClose}>
+              <Button variant="ghost" onClick={handleCloseDialog}>
                 Cancel
               </Button>
               <Button

www/app/(app)/rooms/useRoomList.tsx

@@ -1,11 +1,11 @@
 import { useEffect, useState } from "react";
 import { useError } from "../../(errors)/errorContext";
 import useApi from "../../lib/useApi";
-import { Page_Room_ } from "../../api";
+import { Page_RoomDetails_ } from "../../api";
 import { PaginationPage } from "../browse/_components/Pagination";
 
 type RoomList = {
-  response: Page_Room_ | null;
+  response: Page_RoomDetails_ | null;
   loading: boolean;
   error: Error | null;
   refetch: () => void;
@@ -13,7 +13,7 @@ type RoomList = {
 
 //always protected
 const useRoomList = (page: PaginationPage): RoomList => {
-  const [response, setResponse] = useState<Page_Room_ | null>(null);
+  const [response, setResponse] = useState<Page_RoomDetails_ | null>(null);
   const [loading, setLoading] = useState<boolean>(true);
   const [error, setErrorState] = useState<Error | null>(null);
   const { setError } = useError();

www/app/api/schemas.gen.ts

@@ -91,6 +91,14 @@ export const $CreateRoom = {
       type: "boolean",
       title: "Is Shared",
     },
+    webhook_url: {
+      type: "string",
+      title: "Webhook Url",
+    },
+    webhook_secret: {
+      type: "string",
+      title: "Webhook Secret",
+    },
   },
   type: "object",
   required: [
@@ -103,6 +111,8 @@ export const $CreateRoom = {
     "recording_type",
     "recording_trigger",
     "is_shared",
+    "webhook_url",
+    "webhook_secret",
   ],
   title: "CreateRoom",
 } as const;
@@ -809,11 +819,11 @@ export const $Page_GetTranscriptMinimal_ = {
   title: "Page[GetTranscriptMinimal]",
 } as const;
 
-export const $Page_Room_ = {
+export const $Page_RoomDetails_ = {
   properties: {
     items: {
       items: {
-        $ref: "#/components/schemas/Room",
+        $ref: "#/components/schemas/RoomDetails",
       },
       type: "array",
       title: "Items",
@@ -869,7 +879,7 @@ export const $Page_Room_ = {
   },
   type: "object",
   required: ["items", "page", "size"],
-  title: "Page[Room]",
+  title: "Page[RoomDetails]",
 } as const;
 
 export const $Participant = {
@@ -969,6 +979,86 @@ export const $Room = {
   title: "Room",
 } as const;
 
+export const $RoomDetails = {
+  properties: {
+    id: {
+      type: "string",
+      title: "Id",
+    },
+    name: {
+      type: "string",
+      title: "Name",
+    },
+    user_id: {
+      type: "string",
+      title: "User Id",
+    },
+    created_at: {
+      type: "string",
+      format: "date-time",
+      title: "Created At",
+    },
+    zulip_auto_post: {
+      type: "boolean",
+      title: "Zulip Auto Post",
+    },
+    zulip_stream: {
+      type: "string",
+      title: "Zulip Stream",
+    },
+    zulip_topic: {
+      type: "string",
+      title: "Zulip Topic",
+    },
+    is_locked: {
+      type: "boolean",
+      title: "Is Locked",
+    },
+    room_mode: {
+      type: "string",
+      title: "Room Mode",
+    },
+    recording_type: {
+      type: "string",
+      title: "Recording Type",
+    },
+    recording_trigger: {
+      type: "string",
+      title: "Recording Trigger",
+    },
+    is_shared: {
+      type: "boolean",
+      title: "Is Shared",
+    },
+    webhook_url: {
+      type: "string",
+      title: "Webhook Url",
+    },
+    webhook_secret: {
+      type: "string",
+      title: "Webhook Secret",
+    },
+  },
+  type: "object",
+  required: [
+    "id",
+    "name",
+    "user_id",
+    "created_at",
+    "zulip_auto_post",
+    "zulip_stream",
+    "zulip_topic",
+    "is_locked",
+    "room_mode",
+    "recording_type",
+    "recording_trigger",
+    "is_shared",
+    "webhook_url",
+    "webhook_secret",
+  ],
+  title: "RoomDetails",
+} as const;
+
 export const $RtcOffer = {
   properties: {
     sdp: {
@@ -1351,6 +1441,14 @@ export const $UpdateRoom = {
       type: "boolean",
       title: "Is Shared",
     },
+    webhook_url: {
+      type: "string",
+      title: "Webhook Url",
+    },
+    webhook_secret: {
+      type: "string",
+      title: "Webhook Secret",
+    },
   },
   type: "object",
   required: [
@@ -1363,6 +1461,8 @@ export const $UpdateRoom = {
     "recording_type",
     "recording_trigger",
     "is_shared",
+    "webhook_url",
+    "webhook_secret",
   ],
   title: "UpdateRoom",
 } as const;
@@ -1541,6 +1641,50 @@ export const $ValidationError = {
   title: "ValidationError",
 } as const;
 
+export const $WebhookTestResult = {
+  properties: {
+    success: {
+      type: "boolean",
+      title: "Success",
+    },
+    message: {
+      type: "string",
+      title: "Message",
+      default: "",
+    },
+    error: {
+      type: "string",
+      title: "Error",
+      default: "",
+    },
+    status_code: {
+      anyOf: [
+        {
+          type: "integer",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Status Code",
+    },
+    response_preview: {
+      anyOf: [
+        {
+          type: "string",
+        },
+        {
+          type: "null",
+        },
+      ],
+      title: "Response Preview",
+    },
+  },
+  type: "object",
+  required: ["success"],
+  title: "WebhookTestResult",
+} as const;
+
 export const $WherebyWebhookEvent = {
   properties: {
     apiVersion: {

www/app/api/services.gen.ts

@@ -10,12 +10,16 @@ import type {
   V1RoomsListResponse,
   V1RoomsCreateData,
   V1RoomsCreateResponse,
+  V1RoomsGetData,
+  V1RoomsGetResponse,
   V1RoomsUpdateData,
   V1RoomsUpdateResponse,
   V1RoomsDeleteData,
   V1RoomsDeleteResponse,
   V1RoomsCreateMeetingData,
   V1RoomsCreateMeetingResponse,
+  V1RoomsTestWebhookData,
+  V1RoomsTestWebhookResponse,
   V1TranscriptsListData,
   V1TranscriptsListResponse,
   V1TranscriptsCreateData,
@@ -118,7 +122,7 @@ export class DefaultService {
    * @param data The data for the request.
    * @param data.page Page number
    * @param data.size Page size
-   * @returns Page_Room_ Successful Response
+   * @returns Page_RoomDetails_ Successful Response
    * @throws ApiError
    */
   public v1RoomsList(
@@ -158,12 +162,34 @@ export class DefaultService {
     });
   }
 
+  /**
+   * Rooms Get
+   * @param data The data for the request.
+   * @param data.roomId
+   * @returns RoomDetails Successful Response
+   * @throws ApiError
+   */
+  public v1RoomsGet(
+    data: V1RoomsGetData,
+  ): CancelablePromise<V1RoomsGetResponse> {
+    return this.httpRequest.request({
+      method: "GET",
+      url: "/v1/rooms/{room_id}",
+      path: {
+        room_id: data.roomId,
+      },
+      errors: {
+        422: "Validation Error",
+      },
+    });
+  }
+
   /**
    * Rooms Update
    * @param data The data for the request.
    * @param data.roomId
    * @param data.requestBody
-   * @returns Room Successful Response
+   * @returns RoomDetails Successful Response
    * @throws ApiError
    */
   public v1RoomsUpdate(
@@ -227,6 +253,29 @@ export class DefaultService {
     });
   }
 
+  /**
+   * Rooms Test Webhook
+   * Test webhook configuration by sending a sample payload.
+   * @param data The data for the request.
+   * @param data.roomId
+   * @returns WebhookTestResult Successful Response
+   * @throws ApiError
+   */
+  public v1RoomsTestWebhook(
+    data: V1RoomsTestWebhookData,
+  ): CancelablePromise<V1RoomsTestWebhookResponse> {
+    return this.httpRequest.request({
+      method: "POST",
+      url: "/v1/rooms/{room_id}/webhook/test",
+      path: {
+        room_id: data.roomId,
+      },
+      errors: {
+        422: "Validation Error",
+      },
+    });
+  }
+
   /**
    * Transcripts List
    * @param data The data for the request.

www/app/api/types.gen.ts

@@ -24,6 +24,8 @@ export type CreateRoom = {
   recording_type: string;
   recording_trigger: string;
   is_shared: boolean;
+  webhook_url: string;
+  webhook_secret: string;
 };
 
 export type CreateTranscript = {
@@ -147,8 +149,8 @@ export type Page_GetTranscriptMinimal_ = {
   pages?: number | null;
 };
 
-export type Page_Room_ = {
-  items: Array<Room>;
+export type Page_RoomDetails_ = {
+  items: Array<RoomDetails>;
   total?: number | null;
   page: number | null;
   size: number | null;
@@ -176,6 +178,23 @@ export type Room = {
   is_shared: boolean;
 };
 
+export type RoomDetails = {
+  id: string;
+  name: string;
+  user_id: string;
+  created_at: string;
+  zulip_auto_post: boolean;
+  zulip_stream: string;
+  zulip_topic: string;
+  is_locked: boolean;
+  room_mode: string;
+  recording_type: string;
+  recording_trigger: string;
+  is_shared: boolean;
+  webhook_url: string;
+  webhook_secret: string;
+};
+
 export type RtcOffer = {
   sdp: string;
   type: string;
@@ -281,6 +300,8 @@ export type UpdateRoom = {
   recording_type: string;
   recording_trigger: string;
   is_shared: boolean;
+  webhook_url: string;
+  webhook_secret: string;
 };
 
 export type UpdateTranscript = {
@@ -307,6 +328,14 @@ export type ValidationError = {
   type: string;
 };
 
+export type WebhookTestResult = {
+  success: boolean;
+  message?: string;
+  error?: string;
+  status_code?: number | null;
+  response_preview?: string | null;
+};
+
 export type WherebyWebhookEvent = {
   apiVersion: string;
   id: string;
@@ -350,7 +379,7 @@ export type V1RoomsListData = {
   size?: number;
 };
 
-export type V1RoomsListResponse = Page_Room_;
+export type V1RoomsListResponse = Page_RoomDetails_;
 
 export type V1RoomsCreateData = {
   requestBody: CreateRoom;
@@ -358,12 +387,18 @@ export type V1RoomsCreateData = {
 
 export type V1RoomsCreateResponse = Room;
 
+export type V1RoomsGetData = {
+  roomId: string;
+};
+
+export type V1RoomsGetResponse = RoomDetails;
+
 export type V1RoomsUpdateData = {
   requestBody: UpdateRoom;
   roomId: string;
 };
 
-export type V1RoomsUpdateResponse = Room;
+export type V1RoomsUpdateResponse = RoomDetails;
 
 export type V1RoomsDeleteData = {
   roomId: string;
@@ -377,6 +412,12 @@ export type V1RoomsCreateMeetingData = {
 
 export type V1RoomsCreateMeetingResponse = Meeting;
 
+export type V1RoomsTestWebhookData = {
+  roomId: string;
+};
+
+export type V1RoomsTestWebhookResponse = WebhookTestResult;
+
 export type V1TranscriptsListData = {
   /**
    * Page number
@@ -613,7 +654,7 @@ export type $OpenApiTs = {
         /**
          * Successful Response
          */
-        200: Page_Room_;
+        200: Page_RoomDetails_;
         /**
          * Validation Error
          */
@@ -635,13 +676,26 @@ export type $OpenApiTs = {
     };
   };
   "/v1/rooms/{room_id}": {
+    get: {
+      req: V1RoomsGetData;
+      res: {
+        /**
+         * Successful Response
+         */
+        200: RoomDetails;
+        /**
+         * Validation Error
+         */
+        422: HTTPValidationError;
+      };
+    };
     patch: {
       req: V1RoomsUpdateData;
       res: {
         /**
          * Successful Response
          */
-        200: Room;
+        200: RoomDetails;
         /**
          * Validation Error
          */
@@ -677,6 +731,21 @@ export type $OpenApiTs = {
       };
     };
   };
+  "/v1/rooms/{room_id}/webhook/test": {
+    post: {
+      req: V1RoomsTestWebhookData;
+      res: {
+        /**
+         * Successful Response
+         */
+        200: WebhookTestResult;
+        /**
+         * Validation Error
+         */
+        422: HTTPValidationError;
+      };
+    };
+  };
   "/v1/transcripts": {
     get: {
       req: V1TranscriptsListData;