chore(main): release 0.17.0 (#717)

* llm instructions

* vibe dailyco

* vibe dailyco

* doc update (vibe)

* dont show recording ui on call

* stub processor (vibe)

* stub processor (vibe) self-review

* stub processor (vibe) self-review

* chore(main): release 0.14.0 (#670)

* Add multitrack pipeline

* Mixdown audio tracks

* Mixdown with pyav filter graph

* Trigger multitrack processing for daily recordings

* apply platform from envs in priority: non-dry

* Use explicit track keys for processing

* Align tracks of a multitrack recording

* Generate waveforms for the mixed audio

* Emit multriack pipeline events

* Fix multitrack pipeline track alignment

* dailico docs

* Enable multitrack reprocessing

* modal temp files uniform names, cleanup. remove llm temporary docs

* docs cleanup

* dont proceed with raw recordings if any of the downloads fail

* dry transcription pipelines

* remove is_miltitrack

* comments

* explicit dailyco room name

* docs

* remove stub data/method

* frontend daily/whereby code self-review (no-mistake)

* frontend daily/whereby code self-review (no-mistakes)

* frontend daily/whereby code self-review (no-mistakes)

* consent cleanup for multitrack (no-mistakes)

* llm fun

* remove extra comments

* fix tests

* merge migrations

* Store participant names

* Get participants by meeting session id

* pop back main branch migration

* s3 paddington (no-mistakes)

* comment

* pr comments

* pr comments

* pr comments

* platform / meeting cleanup

* Use participant names in summary generation

* platform assignment to meeting at controller level

* pr comment

* room playform properly default none

* room playform properly default none

* restore migration lost

* streaming WIP

* extract storage / use common storage / proper env vars for storage

* fix mocks tests

* remove fall back

* streaming for multifile

* cenrtal storage abstraction (no-mistakes)

* remove dead code / vars

* Set participant user id for authenticated users

* whereby recording name parsing fix

* whereby recording name parsing fix

* more file stream

* storage dry + tests

* remove homemade boto3 streaming and use proper boto

* update migration guide

* webhook creation script - print uuid

---------

Co-authored-by: Igor Loskutov <igor.loskutoff@gmail.com>
Co-authored-by: Mathieu Virbel <mat@meltingrocks.com>
Co-authored-by: Sergey Mankovsky <sergey@monadical.com>

.github/workflows/deploy.yml

@@ -1,4 +1,4 @@
-name: Deploy to Amazon ECS
+name: Build container/push to container registry
 
 on: [workflow_dispatch]
 

.github/workflows/docker-frontend.yml

@@ -0,0 +1,57 @@
+name: Build and Push Frontend Docker Image
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - 'www/**'
+      - '.github/workflows/docker-frontend.yml'
+  workflow_dispatch:
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}-frontend
+
+jobs:
+  build-and-push:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=ref,event=branch
+            type=sha,prefix={{branch}}-
+            type=raw,value=latest,enable={{is_default_branch}}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v5
+        with:
+          context: ./www
+          file: ./www/Dockerfile
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+          platforms: linux/amd64,linux/arm64

.github/workflows/test_next_server.yml

@@ -0,0 +1,45 @@
+name: Test Next Server
+
+on:
+  pull_request:
+    paths:
+      - "www/**"
+  push:
+    branches:
+      - main
+    paths:
+      - "www/**"
+
+jobs:
+  test-next-server:
+    runs-on: ubuntu-latest
+
+    defaults:
+      run:
+        working-directory: ./www
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Setup Node.js
+      uses: actions/setup-node@v4
+      with:
+        node-version: '20'
+
+    - name: Install pnpm
+      uses: pnpm/action-setup@v4
+      with:
+        version: 8
+
+    - name: Setup Node.js cache
+      uses: actions/setup-node@v4
+      with:
+        node-version: '20'
+        cache: 'pnpm'
+        cache-dependency-path: './www/pnpm-lock.yaml'
+
+    - name: Install dependencies
+      run: pnpm install
+
+    - name: Run tests
+      run: pnpm test

.gitignore

@@ -17,3 +17,4 @@ server/test.sqlite
 CLAUDE.local.md
 www/.env.development
 www/.env.production
+.playwright-mcp

CHANGELOG.md

@@ -1,5 +1,128 @@
 # Changelog
 
+## [0.17.0](https://github.com/Monadical-SAS/reflector/compare/v0.16.0...v0.17.0) (2025-11-13)
+
+
+### Features
+
+* add API key management UI ([#716](https://github.com/Monadical-SAS/reflector/issues/716)) ([372202b](https://github.com/Monadical-SAS/reflector/commit/372202b0e1a86823900b0aa77be1bfbc2893d8a1))
+* daily.co support as alternative to whereby ([#691](https://github.com/Monadical-SAS/reflector/issues/691)) ([1473fd8](https://github.com/Monadical-SAS/reflector/commit/1473fd82dc472c394cbaa2987212ad662a74bcac))
+
+## [0.16.0](https://github.com/Monadical-SAS/reflector/compare/v0.15.0...v0.16.0) (2025-10-24)
+
+
+### Features
+
+* search date filter ([#710](https://github.com/Monadical-SAS/reflector/issues/710)) ([962c40e](https://github.com/Monadical-SAS/reflector/commit/962c40e2b6428ac42fd10aea926782d7a6f3f902))
+
+## [0.15.0](https://github.com/Monadical-SAS/reflector/compare/v0.14.0...v0.15.0) (2025-10-20)
+
+
+### Features
+
+* api tokens  ([#705](https://github.com/Monadical-SAS/reflector/issues/705)) ([9a258ab](https://github.com/Monadical-SAS/reflector/commit/9a258abc0209b0ac3799532a507ea6a9125d703a))
+
+## [0.14.0](https://github.com/Monadical-SAS/reflector/compare/v0.13.1...v0.14.0) (2025-10-08)
+
+
+### Features
+
+* Add calendar event data to transcript webhook payload ([#689](https://github.com/Monadical-SAS/reflector/issues/689)) ([5f6910e](https://github.com/Monadical-SAS/reflector/commit/5f6910e5131b7f28f86c9ecdcc57fed8412ee3cd))
+* container build for www / github ([#672](https://github.com/Monadical-SAS/reflector/issues/672)) ([969bd84](https://github.com/Monadical-SAS/reflector/commit/969bd84fcc14851d1a101412a0ba115f1b7cde82))
+* docker-compose for production frontend ([#664](https://github.com/Monadical-SAS/reflector/issues/664)) ([5bf64b5](https://github.com/Monadical-SAS/reflector/commit/5bf64b5a41f64535e22849b4bb11734d4dbb4aae))
+
+
+### Bug Fixes
+
+* restore feature boolean logic ([#671](https://github.com/Monadical-SAS/reflector/issues/671)) ([3660884](https://github.com/Monadical-SAS/reflector/commit/36608849ec64e953e3be456172502762e3c33df9))
+* security review ([#656](https://github.com/Monadical-SAS/reflector/issues/656)) ([5d98754](https://github.com/Monadical-SAS/reflector/commit/5d98754305c6c540dd194dda268544f6d88bfaf8))
+* update transcript list on reprocess ([#676](https://github.com/Monadical-SAS/reflector/issues/676)) ([9a71af1](https://github.com/Monadical-SAS/reflector/commit/9a71af145ee9b833078c78d0c684590ab12e9f0e))
+* upgrade nemo toolkit ([#678](https://github.com/Monadical-SAS/reflector/issues/678)) ([eef6dc3](https://github.com/Monadical-SAS/reflector/commit/eef6dc39037329b65804297786d852dddb0557f9))
+
+## [0.13.1](https://github.com/Monadical-SAS/reflector/compare/v0.13.0...v0.13.1) (2025-09-22)
+
+
+### Bug Fixes
+
+* TypeError on not all arguments converted during string formatting in logger ([#667](https://github.com/Monadical-SAS/reflector/issues/667)) ([565a629](https://github.com/Monadical-SAS/reflector/commit/565a62900f5a02fc946b68f9269a42190ed70ab6))
+
+## [0.13.0](https://github.com/Monadical-SAS/reflector/compare/v0.12.1...v0.13.0) (2025-09-19)
+
+
+### Features
+
+* room form edit with enter ([#662](https://github.com/Monadical-SAS/reflector/issues/662)) ([47716f6](https://github.com/Monadical-SAS/reflector/commit/47716f6e5ddee952609d2fa0ffabdfa865286796))
+
+
+### Bug Fixes
+
+* invalid cleanup call ([#660](https://github.com/Monadical-SAS/reflector/issues/660)) ([0abcebf](https://github.com/Monadical-SAS/reflector/commit/0abcebfc9491f87f605f21faa3e53996fafedd9a))
+
+## [0.12.1](https://github.com/Monadical-SAS/reflector/compare/v0.12.0...v0.12.1) (2025-09-17)
+
+
+### Bug Fixes
+
+* production blocked because having existing meeting with room_id null ([#657](https://github.com/Monadical-SAS/reflector/issues/657)) ([870e860](https://github.com/Monadical-SAS/reflector/commit/870e8605171a27155a9cbee215eeccb9a8d6c0a2))
+
+## [0.12.0](https://github.com/Monadical-SAS/reflector/compare/v0.11.0...v0.12.0) (2025-09-17)
+
+
+### Features
+
+* calendar integration ([#608](https://github.com/Monadical-SAS/reflector/issues/608)) ([6f680b5](https://github.com/Monadical-SAS/reflector/commit/6f680b57954c688882c4ed49f40f161c52a00a24))
+* self-hosted gpu api ([#636](https://github.com/Monadical-SAS/reflector/issues/636)) ([ab859d6](https://github.com/Monadical-SAS/reflector/commit/ab859d65a6bded904133a163a081a651b3938d42))
+
+
+### Bug Fixes
+
+* ignore player hotkeys for text inputs ([#646](https://github.com/Monadical-SAS/reflector/issues/646)) ([fa049e8](https://github.com/Monadical-SAS/reflector/commit/fa049e8d068190ce7ea015fd9fcccb8543f54a3f))
+
+## [0.11.0](https://github.com/Monadical-SAS/reflector/compare/v0.10.0...v0.11.0) (2025-09-16)
+
+
+### Features
+
+* remove profanity filter that was there for conference ([#652](https://github.com/Monadical-SAS/reflector/issues/652)) ([b42f7cf](https://github.com/Monadical-SAS/reflector/commit/b42f7cfc606783afcee792590efcc78b507468ab))
+
+
+### Bug Fixes
+
+* zulip and consent handler on the file pipeline ([#645](https://github.com/Monadical-SAS/reflector/issues/645)) ([5f143fe](https://github.com/Monadical-SAS/reflector/commit/5f143fe3640875dcb56c26694254a93189281d17))
+* zulip stream and topic selection in share dialog ([#644](https://github.com/Monadical-SAS/reflector/issues/644)) ([c546e69](https://github.com/Monadical-SAS/reflector/commit/c546e69739e68bb74fbc877eb62609928e5b8de6))
+
+## [0.10.0](https://github.com/Monadical-SAS/reflector/compare/v0.9.0...v0.10.0) (2025-09-11)
+
+
+### Features
+
+* replace nextjs-config with environment variables ([#632](https://github.com/Monadical-SAS/reflector/issues/632)) ([369ecdf](https://github.com/Monadical-SAS/reflector/commit/369ecdff13f3862d926a9c0b87df52c9d94c4dde))
+
+
+### Bug Fixes
+
+* anonymous users transcript permissions ([#621](https://github.com/Monadical-SAS/reflector/issues/621)) ([f81fe99](https://github.com/Monadical-SAS/reflector/commit/f81fe9948a9237b3e0001b2d8ca84f54d76878f9))
+* auth post ([#624](https://github.com/Monadical-SAS/reflector/issues/624)) ([cde99ca](https://github.com/Monadical-SAS/reflector/commit/cde99ca2716f84ba26798f289047732f0448742e))
+* auth post ([#626](https://github.com/Monadical-SAS/reflector/issues/626)) ([3b85ff3](https://github.com/Monadical-SAS/reflector/commit/3b85ff3bdf4fb053b103070646811bc990c0e70a))
+* auth post ([#627](https://github.com/Monadical-SAS/reflector/issues/627)) ([962038e](https://github.com/Monadical-SAS/reflector/commit/962038ee3f2a555dc3c03856be0e4409456e0996))
+* missing follow_redirects=True on modal endpoint ([#630](https://github.com/Monadical-SAS/reflector/issues/630)) ([fc363bd](https://github.com/Monadical-SAS/reflector/commit/fc363bd49b17b075e64f9186e5e0185abc325ea7))
+* sync backend and frontend token refresh logic ([#614](https://github.com/Monadical-SAS/reflector/issues/614)) ([5a5b323](https://github.com/Monadical-SAS/reflector/commit/5a5b3233820df9536da75e87ce6184a983d4713a))
+
+## [0.9.0](https://github.com/Monadical-SAS/reflector/compare/v0.8.2...v0.9.0) (2025-09-06)
+
+
+### Features
+
+* frontend openapi react query ([#606](https://github.com/Monadical-SAS/reflector/issues/606)) ([c4d2825](https://github.com/Monadical-SAS/reflector/commit/c4d2825c81f81ad8835629fbf6ea8c7383f8c31b))
+
+
+### Bug Fixes
+
+* align whisper transcriber api with parakeet ([#602](https://github.com/Monadical-SAS/reflector/issues/602)) ([0663700](https://github.com/Monadical-SAS/reflector/commit/0663700a615a4af69a03c96c410f049e23ec9443))
+* kv use tls explicit ([#610](https://github.com/Monadical-SAS/reflector/issues/610)) ([08d88ec](https://github.com/Monadical-SAS/reflector/commit/08d88ec349f38b0d13e0fa4cb73486c8dfd31836))
+* source kind for file processing ([#601](https://github.com/Monadical-SAS/reflector/issues/601)) ([dc82f8b](https://github.com/Monadical-SAS/reflector/commit/dc82f8bb3bdf3ab3d4088e592a30fd63907319e1))
+* token refresh locking ([#613](https://github.com/Monadical-SAS/reflector/issues/613)) ([7f5a4c9](https://github.com/Monadical-SAS/reflector/commit/7f5a4c9ddc7fd098860c8bdda2ca3b57f63ded2f))
+
 ## [0.8.2](https://github.com/Monadical-SAS/reflector/compare/v0.8.1...v0.8.2) (2025-08-29)
 
 

CLAUDE.md

@@ -66,7 +66,6 @@ pnpm install
 
 # Copy configuration templates
 cp .env_template .env
-cp config-template.ts config.ts
 ```
 
 **Development:**
@@ -152,7 +151,7 @@ All endpoints prefixed `/v1/`:
 
 **Frontend** (`www/.env`):
 - `NEXTAUTH_URL`, `NEXTAUTH_SECRET` - Authentication configuration
-- `NEXT_PUBLIC_REFLECTOR_API_URL` - Backend API endpoint
+- `REFLECTOR_API_URL` - Backend API endpoint
 - `REFLECTOR_DOMAIN_CONFIG` - Feature flags and domain settings
 
 ## Testing Strategy

README.md

@@ -99,11 +99,10 @@ Start with `cd www`.
 
 ```bash
 pnpm install
-cp .env_template .env
-cp config-template.ts config.ts
+cp .env.example .env
 ```
 
-Then, fill in the environment variables in `.env` and the configuration in `config.ts` as needed. If you are unsure on how to proceed, ask in Zulip.
+Then, fill in the environment variables in `.env` as needed. If you are unsure on how to proceed, ask in Zulip.
 
 **Run in development mode**
 
@@ -168,3 +167,41 @@ You can manually process an audio file by calling the process tool:
 ```bash
 uv run python -m reflector.tools.process path/to/audio.wav
 ```
+
+## Build-time env variables
+
+Next.js projects are more used to NEXT_PUBLIC_ prefixed buildtime vars. We don't have those for the reason we need to serve a ccustomizable prebuild docker container.
+
+Instead, all the variables are runtime. Variables needed to the frontend are served to the frontend app at initial render.
+
+It also means there's no static prebuild and no static files to serve for js/html.
+
+## Feature Flags
+
+Reflector uses environment variable-based feature flags to control application functionality. These flags allow you to enable or disable features without code changes.
+
+### Available Feature Flags
+
+| Feature Flag | Environment Variable |
+|-------------|---------------------|
+| `requireLogin` | `FEATURE_REQUIRE_LOGIN` |
+| `privacy` | `FEATURE_PRIVACY` |
+| `browse` | `FEATURE_BROWSE` |
+| `sendToZulip` | `FEATURE_SEND_TO_ZULIP` |
+| `rooms` | `FEATURE_ROOMS` |
+
+### Setting Feature Flags
+
+Feature flags are controlled via environment variables using the pattern `FEATURE_{FEATURE_NAME}` where `{FEATURE_NAME}` is the SCREAMING_SNAKE_CASE version of the feature name.
+
+**Examples:**
+```bash
+# Enable user authentication requirement
+FEATURE_REQUIRE_LOGIN=true
+
+# Disable browse functionality
+FEATURE_BROWSE=false
+
+# Enable Zulip integration
+FEATURE_SEND_TO_ZULIP=true
+```

docker-compose.prod.yml

@@ -0,0 +1,39 @@
+# Production Docker Compose configuration for Frontend
+# Usage: docker compose -f docker-compose.prod.yml up -d
+
+services:
+  web:
+    build:
+      context: ./www
+      dockerfile: Dockerfile
+    image: reflector-frontend:latest
+    environment:
+      - KV_URL=${KV_URL:-redis://redis:6379}
+      - SITE_URL=${SITE_URL}
+      - API_URL=${API_URL}
+      - WEBSOCKET_URL=${WEBSOCKET_URL}
+      - NEXTAUTH_URL=${NEXTAUTH_URL:-http://localhost:3000}
+      - NEXTAUTH_SECRET=${NEXTAUTH_SECRET:-changeme-in-production}
+      - AUTHENTIK_ISSUER=${AUTHENTIK_ISSUER}
+      - AUTHENTIK_CLIENT_ID=${AUTHENTIK_CLIENT_ID}
+      - AUTHENTIK_CLIENT_SECRET=${AUTHENTIK_CLIENT_SECRET}
+      - AUTHENTIK_REFRESH_TOKEN_URL=${AUTHENTIK_REFRESH_TOKEN_URL}
+      - SENTRY_DSN=${SENTRY_DSN}
+      - SENTRY_IGNORE_API_RESOLUTION_ERROR=${SENTRY_IGNORE_API_RESOLUTION_ERROR:-1}
+    depends_on:
+      - redis
+    restart: unless-stopped
+
+  redis:
+    image: redis:7.2-alpine
+    restart: unless-stopped
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 30s
+      timeout: 3s
+      retries: 3
+    volumes:
+      - redis_data:/data
+
+volumes:
+  redis_data:

docker-compose.yml

@@ -6,6 +6,7 @@ services:
       - 1250:1250
     volumes:
       - ./server/:/app/
+      - /app/.venv
     env_file:
       - ./server/.env
     environment:
@@ -16,6 +17,7 @@ services:
       context: server
     volumes:
       - ./server/:/app/
+      - /app/.venv
     env_file:
       - ./server/.env
     environment:
@@ -26,6 +28,7 @@ services:
       context: server
     volumes:
       - ./server/:/app/
+      - /app/.venv
     env_file:
       - ./server/.env
     environment:
@@ -36,7 +39,7 @@ services:
     ports:
       - 6379:6379
   web:
-    image: node:18
+    image: node:22-alpine
     ports:
       - "3000:3000"
     command: sh -c "corepack enable && pnpm install && pnpm dev"
@@ -47,6 +50,8 @@ services:
       - /app/node_modules
     env_file:
       - ./www/.env.local
+    environment:
+      - NODE_ENV=development
 
   postgres:
     image: postgres:17

gpu/modal_deployments/.gitignore

@@ -0,0 +1,33 @@
+# OS / Editor
+.DS_Store
+.vscode/
+.idea/
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Logs
+*.log
+
+# Env and secrets
+.env
+.env.*
+*.env
+*.secret
+
+# Build / dist
+build/
+dist/
+.eggs/
+*.egg-info/
+
+# Coverage / test
+.pytest_cache/
+.coverage*
+htmlcov/
+
+# Modal local state (if any)
+modal_mounts/
+.modal_cache/

gpu/modal_deployments/reflector_transcriber.py

@@ -0,0 +1,608 @@
+import os
+import sys
+import threading
+import uuid
+from typing import Generator, Mapping, NamedTuple, NewType, TypedDict
+from urllib.parse import urlparse
+
+import modal
+
+MODEL_NAME = "large-v2"
+MODEL_COMPUTE_TYPE: str = "float16"
+MODEL_NUM_WORKERS: int = 1
+MINUTES = 60  # seconds
+SAMPLERATE = 16000
+UPLOADS_PATH = "/uploads"
+CACHE_PATH = "/models"
+SUPPORTED_FILE_EXTENSIONS = ["mp3", "mp4", "mpeg", "mpga", "m4a", "wav", "webm"]
+VAD_CONFIG = {
+    "batch_max_duration": 30.0,
+    "silence_padding": 0.5,
+    "window_size": 512,
+}
+
+
+WhisperUniqFilename = NewType("WhisperUniqFilename", str)
+AudioFileExtension = NewType("AudioFileExtension", str)
+
+app = modal.App("reflector-transcriber")
+
+model_cache = modal.Volume.from_name("models", create_if_missing=True)
+upload_volume = modal.Volume.from_name("whisper-uploads", create_if_missing=True)
+
+
+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
+
+
+def download_model():
+    from faster_whisper import download_model
+
+    model_cache.reload()
+
+    download_model(MODEL_NAME, cache_dir=CACHE_PATH)
+
+    model_cache.commit()
+
+
+image = (
+    modal.Image.debian_slim(python_version="3.12")
+    .env(
+        {
+            "HF_HUB_ENABLE_HF_TRANSFER": "1",
+            "LD_LIBRARY_PATH": (
+                "/usr/local/lib/python3.12/site-packages/nvidia/cudnn/lib/:"
+                "/opt/conda/lib/python3.12/site-packages/nvidia/cublas/lib/"
+            ),
+        }
+    )
+    .apt_install("ffmpeg")
+    .pip_install(
+        "huggingface_hub==0.27.1",
+        "hf-transfer==0.1.9",
+        "torch==2.5.1",
+        "faster-whisper==1.1.1",
+        "fastapi==0.115.12",
+        "requests",
+        "librosa==0.10.1",
+        "numpy<2",
+        "silero-vad==5.1.0",
+    )
+    .run_function(download_model, volumes={CACHE_PATH: model_cache})
+)
+
+
+def detect_audio_format(url: str, headers: Mapping[str, str]) -> AudioFileExtension:
+    parsed_url = urlparse(url)
+    url_path = parsed_url.path
+
+    for ext in SUPPORTED_FILE_EXTENSIONS:
+        if url_path.lower().endswith(f".{ext}"):
+            return AudioFileExtension(ext)
+
+    content_type = headers.get("content-type", "").lower()
+    if "audio/mpeg" in content_type or "audio/mp3" in content_type:
+        return AudioFileExtension("mp3")
+    if "audio/wav" in content_type:
+        return AudioFileExtension("wav")
+    if "audio/mp4" in content_type:
+        return AudioFileExtension("mp4")
+
+    raise ValueError(
+        f"Unsupported audio format for URL: {url}. "
+        f"Supported extensions: {', '.join(SUPPORTED_FILE_EXTENSIONS)}"
+    )
+
+
+def download_audio_to_volume(
+    audio_file_url: str,
+) -> tuple[WhisperUniqFilename, AudioFileExtension]:
+    import requests
+    from fastapi import HTTPException
+
+    response = requests.head(audio_file_url, allow_redirects=True)
+    if response.status_code == 404:
+        raise HTTPException(status_code=404, detail="Audio file not found")
+
+    response = requests.get(audio_file_url, allow_redirects=True)
+    response.raise_for_status()
+
+    audio_suffix = detect_audio_format(audio_file_url, response.headers)
+    unique_filename = WhisperUniqFilename(f"{uuid.uuid4()}.{audio_suffix}")
+    file_path = f"{UPLOADS_PATH}/{unique_filename}"
+
+    with open(file_path, "wb") as f:
+        f.write(response.content)
+
+    upload_volume.commit()
+    return unique_filename, audio_suffix
+
+
+def pad_audio(audio_array, sample_rate: int = SAMPLERATE):
+    """Add 0.5s of silence if audio is shorter than the silence_padding window.
+
+    Whisper does not require this strictly, but aligning behavior with Parakeet
+    avoids edge-case crashes on extremely short inputs and makes comparisons easier.
+    """
+    import numpy as np
+
+    audio_duration = len(audio_array) / sample_rate
+    if audio_duration < VAD_CONFIG["silence_padding"]:
+        silence_samples = int(sample_rate * VAD_CONFIG["silence_padding"])
+        silence = np.zeros(silence_samples, dtype=np.float32)
+        return np.concatenate([audio_array, silence])
+    return audio_array
+
+
+@app.cls(
+    gpu="A10G",
+    timeout=5 * MINUTES,
+    scaledown_window=5 * MINUTES,
+    image=image,
+    volumes={CACHE_PATH: model_cache, UPLOADS_PATH: upload_volume},
+)
+@modal.concurrent(max_inputs=10)
+class TranscriberWhisperLive:
+    """Live transcriber class for small audio segments (A10G).
+
+    Mirrors the Parakeet live class API but uses Faster-Whisper under the hood.
+    """
+
+    @modal.enter()
+    def enter(self):
+        import faster_whisper
+        import torch
+
+        self.lock = threading.Lock()
+        self.use_gpu = torch.cuda.is_available()
+        self.device = "cuda" if self.use_gpu else "cpu"
+        self.model = faster_whisper.WhisperModel(
+            MODEL_NAME,
+            device=self.device,
+            compute_type=MODEL_COMPUTE_TYPE,
+            num_workers=MODEL_NUM_WORKERS,
+            download_root=CACHE_PATH,
+            local_files_only=True,
+        )
+        print(f"Model is on device: {self.device}")
+
+    @modal.method()
+    def transcribe_segment(
+        self,
+        filename: str,
+        language: str = "en",
+    ):
+        """Transcribe a single uploaded audio file by filename."""
+        upload_volume.reload()
+
+        file_path = f"{UPLOADS_PATH}/{filename}"
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"File not found: {file_path}")
+
+        with self.lock:
+            with NoStdStreams():
+                segments, _ = self.model.transcribe(
+                    file_path,
+                    language=language,
+                    beam_size=5,
+                    word_timestamps=True,
+                    vad_filter=True,
+                    vad_parameters={"min_silence_duration_ms": 500},
+                )
+
+        segments = list(segments)
+        text = "".join(segment.text for segment in segments).strip()
+        words = [
+            {
+                "word": word.word,
+                "start": round(float(word.start), 2),
+                "end": round(float(word.end), 2),
+            }
+            for segment in segments
+            for word in segment.words
+        ]
+
+        return {"text": text, "words": words}
+
+    @modal.method()
+    def transcribe_batch(
+        self,
+        filenames: list[str],
+        language: str = "en",
+    ):
+        """Transcribe multiple uploaded audio files and return per-file results."""
+        upload_volume.reload()
+
+        results = []
+        for filename in filenames:
+            file_path = f"{UPLOADS_PATH}/{filename}"
+            if not os.path.exists(file_path):
+                raise FileNotFoundError(f"Batch file not found: {file_path}")
+
+            with self.lock:
+                with NoStdStreams():
+                    segments, _ = self.model.transcribe(
+                        file_path,
+                        language=language,
+                        beam_size=5,
+                        word_timestamps=True,
+                        vad_filter=True,
+                        vad_parameters={"min_silence_duration_ms": 500},
+                    )
+
+            segments = list(segments)
+            text = "".join(seg.text for seg in segments).strip()
+            words = [
+                {
+                    "word": w.word,
+                    "start": round(float(w.start), 2),
+                    "end": round(float(w.end), 2),
+                }
+                for seg in segments
+                for w in seg.words
+            ]
+
+            results.append(
+                {
+                    "filename": filename,
+                    "text": text,
+                    "words": words,
+                }
+            )
+
+        return results
+
+
+@app.cls(
+    gpu="L40S",
+    timeout=15 * MINUTES,
+    image=image,
+    volumes={CACHE_PATH: model_cache, UPLOADS_PATH: upload_volume},
+)
+class TranscriberWhisperFile:
+    """File transcriber for larger/longer audio, using VAD-driven batching (L40S)."""
+
+    @modal.enter()
+    def enter(self):
+        import faster_whisper
+        import torch
+        from silero_vad import load_silero_vad
+
+        self.lock = threading.Lock()
+        self.use_gpu = torch.cuda.is_available()
+        self.device = "cuda" if self.use_gpu else "cpu"
+        self.model = faster_whisper.WhisperModel(
+            MODEL_NAME,
+            device=self.device,
+            compute_type=MODEL_COMPUTE_TYPE,
+            num_workers=MODEL_NUM_WORKERS,
+            download_root=CACHE_PATH,
+            local_files_only=True,
+        )
+        self.vad_model = load_silero_vad(onnx=False)
+
+    @modal.method()
+    def transcribe_segment(
+        self, filename: str, timestamp_offset: float = 0.0, language: str = "en"
+    ):
+        import librosa
+        import numpy as np
+        from silero_vad import VADIterator
+
+        def vad_segments(
+            audio_array,
+            sample_rate: int = SAMPLERATE,
+            window_size: int = VAD_CONFIG["window_size"],
+        ) -> Generator[TimeSegment, None, None]:
+            """Generate speech segments as TimeSegment using Silero VAD."""
+            iterator = VADIterator(self.vad_model, sampling_rate=sample_rate)
+            start = None
+            for i in range(0, len(audio_array), window_size):
+                chunk = audio_array[i : i + window_size]
+                if len(chunk) < window_size:
+                    chunk = np.pad(
+                        chunk, (0, window_size - len(chunk)), mode="constant"
+                    )
+                speech = iterator(chunk)
+                if not speech:
+                    continue
+                if "start" in speech:
+                    start = speech["start"]
+                    continue
+                if "end" in speech and start is not None:
+                    end = speech["end"]
+                    yield TimeSegment(
+                        start / float(SAMPLERATE), end / float(SAMPLERATE)
+                    )
+                    start = None
+            iterator.reset_states()
+
+        upload_volume.reload()
+        file_path = f"{UPLOADS_PATH}/{filename}"
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"File not found: {file_path}")
+
+        audio_array, _sr = librosa.load(file_path, sr=SAMPLERATE, mono=True)
+
+        # Batch segments up to ~30s windows by merging contiguous VAD segments
+        merged_batches: list[TimeSegment] = []
+        batch_start = None
+        batch_end = None
+        max_duration = VAD_CONFIG["batch_max_duration"]
+        for segment in vad_segments(audio_array):
+            seg_start, seg_end = segment.start, segment.end
+            if batch_start is None:
+                batch_start, batch_end = seg_start, seg_end
+                continue
+            if seg_end - batch_start <= max_duration:
+                batch_end = seg_end
+            else:
+                merged_batches.append(TimeSegment(batch_start, batch_end))
+                batch_start, batch_end = seg_start, seg_end
+        if batch_start is not None and batch_end is not None:
+            merged_batches.append(TimeSegment(batch_start, batch_end))
+
+        all_text = []
+        all_words = []
+
+        for segment in merged_batches:
+            start_time, end_time = segment.start, segment.end
+            s_idx = int(start_time * SAMPLERATE)
+            e_idx = int(end_time * SAMPLERATE)
+            segment = audio_array[s_idx:e_idx]
+            segment = pad_audio(segment, SAMPLERATE)
+
+            with self.lock:
+                segments, _ = self.model.transcribe(
+                    segment,
+                    language=language,
+                    beam_size=5,
+                    word_timestamps=True,
+                    vad_filter=True,
+                    vad_parameters={"min_silence_duration_ms": 500},
+                )
+
+            segments = list(segments)
+            text = "".join(seg.text for seg in segments).strip()
+            words = [
+                {
+                    "word": w.word,
+                    "start": round(float(w.start) + start_time + timestamp_offset, 2),
+                    "end": round(float(w.end) + start_time + timestamp_offset, 2),
+                }
+                for seg in segments
+                for w in seg.words
+            ]
+            if text:
+                all_text.append(text)
+            all_words.extend(words)
+
+        return {"text": " ".join(all_text), "words": all_words}
+
+
+def detect_audio_format(url: str, headers: dict) -> str:
+    from urllib.parse import urlparse
+
+    from fastapi import HTTPException
+
+    url_path = urlparse(url).path
+    for ext in SUPPORTED_FILE_EXTENSIONS:
+        if url_path.lower().endswith(f".{ext}"):
+            return ext
+
+    content_type = headers.get("content-type", "").lower()
+    if "audio/mpeg" in content_type or "audio/mp3" in content_type:
+        return "mp3"
+    if "audio/wav" in content_type:
+        return "wav"
+    if "audio/mp4" in content_type:
+        return "mp4"
+
+    raise HTTPException(
+        status_code=400,
+        detail=(
+            f"Unsupported audio format for URL. Supported extensions: {', '.join(SUPPORTED_FILE_EXTENSIONS)}"
+        ),
+    )
+
+
+def download_audio_to_volume(audio_file_url: str) -> tuple[str, str]:
+    import requests
+    from fastapi import HTTPException
+
+    response = requests.head(audio_file_url, allow_redirects=True)
+    if response.status_code == 404:
+        raise HTTPException(status_code=404, detail="Audio file not found")
+
+    response = requests.get(audio_file_url, allow_redirects=True)
+    response.raise_for_status()
+
+    audio_suffix = detect_audio_format(audio_file_url, response.headers)
+    unique_filename = f"{uuid.uuid4()}.{audio_suffix}"
+    file_path = f"{UPLOADS_PATH}/{unique_filename}"
+
+    with open(file_path, "wb") as f:
+        f.write(response.content)
+
+    upload_volume.commit()
+    return unique_filename, audio_suffix
+
+
+@app.function(
+    scaledown_window=60,
+    timeout=600,
+    secrets=[
+        modal.Secret.from_name("reflector-gpu"),
+    ],
+    volumes={CACHE_PATH: model_cache, UPLOADS_PATH: upload_volume},
+    image=image,
+)
+@modal.concurrent(max_inputs=40)
+@modal.asgi_app()
+def web():
+    from fastapi import (
+        Body,
+        Depends,
+        FastAPI,
+        Form,
+        HTTPException,
+        UploadFile,
+        status,
+    )
+    from fastapi.security import OAuth2PasswordBearer
+
+    transcriber_live = TranscriberWhisperLive()
+    transcriber_file = TranscriberWhisperFile()
+
+    app = FastAPI()
+
+    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+    def apikey_auth(apikey: str = Depends(oauth2_scheme)):
+        if apikey == os.environ["REFLECTOR_GPU_APIKEY"]:
+            return
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid API key",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    class TranscriptResponse(dict):
+        pass
+
+    @app.post("/v1/audio/transcriptions", dependencies=[Depends(apikey_auth)])
+    def transcribe(
+        file: UploadFile = None,
+        files: list[UploadFile] | None = None,
+        model: str = Form(MODEL_NAME),
+        language: str = Form("en"),
+        batch: bool = Form(False),
+    ):
+        if not file and not files:
+            raise HTTPException(
+                status_code=400, detail="Either 'file' or 'files' parameter is required"
+            )
+        if batch and not files:
+            raise HTTPException(
+                status_code=400, detail="Batch transcription requires 'files'"
+            )
+
+        upload_files = [file] if file else files
+
+        uploaded_filenames: list[str] = []
+        for upload_file in upload_files:
+            audio_suffix = upload_file.filename.split(".")[-1]
+            if audio_suffix not in SUPPORTED_FILE_EXTENSIONS:
+                raise HTTPException(
+                    status_code=400,
+                    detail=(
+                        f"Unsupported audio format. Supported extensions: {', '.join(SUPPORTED_FILE_EXTENSIONS)}"
+                    ),
+                )
+
+            unique_filename = f"{uuid.uuid4()}.{audio_suffix}"
+            file_path = f"{UPLOADS_PATH}/{unique_filename}"
+            with open(file_path, "wb") as f:
+                content = upload_file.file.read()
+                f.write(content)
+            uploaded_filenames.append(unique_filename)
+
+        upload_volume.commit()
+
+        try:
+            if batch and len(upload_files) > 1:
+                func = transcriber_live.transcribe_batch.spawn(
+                    filenames=uploaded_filenames,
+                    language=language,
+                )
+                results = func.get()
+                return {"results": results}
+
+            results = []
+            for filename in uploaded_filenames:
+                func = transcriber_live.transcribe_segment.spawn(
+                    filename=filename,
+                    language=language,
+                )
+                result = func.get()
+                result["filename"] = filename
+                results.append(result)
+
+            return {"results": results} if len(results) > 1 else results[0]
+        finally:
+            for filename in uploaded_filenames:
+                try:
+                    file_path = f"{UPLOADS_PATH}/{filename}"
+                    os.remove(file_path)
+                except Exception:
+                    pass
+            upload_volume.commit()
+
+    @app.post("/v1/audio/transcriptions-from-url", dependencies=[Depends(apikey_auth)])
+    def transcribe_from_url(
+        audio_file_url: str = Body(
+            ..., description="URL of the audio file to transcribe"
+        ),
+        model: str = Body(MODEL_NAME),
+        language: str = Body("en"),
+        timestamp_offset: float = Body(0.0),
+    ):
+        unique_filename, _audio_suffix = download_audio_to_volume(audio_file_url)
+        try:
+            func = transcriber_file.transcribe_segment.spawn(
+                filename=unique_filename,
+                timestamp_offset=timestamp_offset,
+                language=language,
+            )
+            result = func.get()
+            return result
+        finally:
+            try:
+                file_path = f"{UPLOADS_PATH}/{unique_filename}"
+                os.remove(file_path)
+                upload_volume.commit()
+            except Exception:
+                pass
+
+    return app
+
+
+class NoStdStreams:
+    def __init__(self):
+        self.devnull = open(os.devnull, "w")
+
+    def __enter__(self):
+        self._stdout, self._stderr = sys.stdout, sys.stderr
+        self._stdout.flush()
+        self._stderr.flush()
+        sys.stdout, sys.stderr = self.devnull, self.devnull
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        sys.stdout, sys.stderr = self._stdout, self._stderr
+        self.devnull.close()

gpu/modal_deployments/reflector_transcriber_parakeet.py

@@ -77,7 +77,7 @@ image = (
     .pip_install(
         "hf_transfer==0.1.9",
         "huggingface_hub[hf-xet]==0.31.2",
-        "nemo_toolkit[asr]==2.3.0",
+        "nemo_toolkit[asr]==2.5.0",
         "cuda-python==12.8.0",
         "fastapi==0.115.12",
         "numpy<2",

gpu/self_hosted/.env.example

@@ -0,0 +1,2 @@
+REFLECTOR_GPU_APIKEY=
+HF_TOKEN=

gpu/self_hosted/.gitignore

@@ -0,0 +1,38 @@
+cache/
+
+# OS / Editor
+.DS_Store
+.vscode/
+.idea/
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Env and secrets
+.env
+*.env
+*.secret
+HF_TOKEN
+REFLECTOR_GPU_APIKEY
+
+# Virtual env / uv
+.venv/
+venv/
+ENV/
+uv/
+
+# Build / dist
+build/
+dist/
+.eggs/
+*.egg-info/
+
+# Coverage / test
+.pytest_cache/
+.coverage*
+htmlcov/
+
+# Logs
+*.log

gpu/self_hosted/Dockerfile

@@ -0,0 +1,46 @@
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1 \
+    UV_LINK_MODE=copy \
+    UV_NO_CACHE=1
+
+WORKDIR /tmp
+RUN apt-get update \
+ && apt-get install -y \
+    ffmpeg \
+    curl \
+    ca-certificates \
+    gnupg \
+    wget \
+ && apt-get clean
+# Add NVIDIA CUDA repo for Debian 12 (bookworm) and install cuDNN 9 for CUDA 12
+ADD https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb /cuda-keyring.deb
+RUN dpkg -i /cuda-keyring.deb \
+ && rm /cuda-keyring.deb \
+ && apt-get update \
+ && apt-get install -y --no-install-recommends \
+    cuda-cudart-12-6 \
+    libcublas-12-6 \
+    libcudnn9-cuda-12 \
+    libcudnn9-dev-cuda-12 \
+ && apt-get clean \
+ && rm -rf /var/lib/apt/lists/*
+ADD https://astral.sh/uv/install.sh /uv-installer.sh
+RUN sh /uv-installer.sh && rm /uv-installer.sh
+ENV PATH="/root/.local/bin/:$PATH"
+ENV LD_LIBRARY_PATH="/usr/local/cuda/lib64:/usr/lib/x86_64-linux-gnu:$LD_LIBRARY_PATH"
+
+RUN mkdir -p /app
+WORKDIR /app
+COPY pyproject.toml uv.lock /app/
+
+
+COPY ./app /app/app
+COPY ./main.py /app/
+COPY ./runserver.sh /app/
+
+EXPOSE 8000
+
+CMD ["sh", "/app/runserver.sh"]
+
+

gpu/self_hosted/README.md

@@ -0,0 +1,73 @@
+# Self-hosted Model API
+
+Run transcription, translation, and diarization services compatible with Reflector's GPU Model API. Works on CPU or GPU.
+
+Environment variables
+
+- REFLECTOR_GPU_APIKEY: Optional Bearer token. If unset, auth is disabled.
+- HF_TOKEN: Optional. Required for diarization to download pyannote pipelines
+
+Requirements
+
+- FFmpeg must be installed and on PATH (used for URL-based and segmented transcription)
+- Python 3.12+
+- NVIDIA GPU optional. If available, it will be used automatically
+
+Local run
+Set env vars in self_hosted/.env file
+uv sync
+
+uv run uvicorn main:app --host 0.0.0.0 --port 8000
+
+Authentication
+
+- If REFLECTOR_GPU_APIKEY is set, include header: Authorization: Bearer <key>
+
+Endpoints
+
+- POST /v1/audio/transcriptions
+
+  - multipart/form-data
+  - fields: file (single file) OR files[] (multiple files), language, batch (true/false)
+  - response: single { text, words, filename } or { results: [ ... ] }
+
+- POST /v1/audio/transcriptions-from-url
+
+  - application/json
+  - body: { audio_file_url, language, timestamp_offset }
+  - response: { text, words }
+
+- POST /translate
+
+  - text: query parameter
+  - body (application/json): { source_language, target_language }
+  - response: { text: { <src>: original, <tgt>: translated } }
+
+- POST /diarize
+  - query parameters: audio_file_url, timestamp (optional)
+  - requires HF_TOKEN to be set (for pyannote)
+  - response: { diarization: [ { start, end, speaker } ] }
+
+OpenAPI docs
+
+- Visit /docs when the server is running
+
+Docker
+
+- Not yet provided in this directory. A Dockerfile will be added later. For now, use Local run above
+
+Conformance tests
+
+# From this directory
+
+TRANSCRIPT_URL=http://localhost:8000 \
+TRANSCRIPT_API_KEY=dev-key \
+uv run -m pytest -m model_api --no-cov ../../server/tests/test_model_api_transcript.py
+
+TRANSLATION_URL=http://localhost:8000 \
+TRANSLATION_API_KEY=dev-key \
+uv run -m pytest -m model_api --no-cov ../../server/tests/test_model_api_translation.py
+
+DIARIZATION_URL=http://localhost:8000 \
+DIARIZATION_API_KEY=dev-key \
+uv run -m pytest -m model_api --no-cov ../../server/tests/test_model_api_diarization.py

gpu/self_hosted/app/auth.py

@@ -0,0 +1,19 @@
+import os
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+def apikey_auth(apikey: str = Depends(oauth2_scheme)):
+    required_key = os.environ.get("REFLECTOR_GPU_APIKEY")
+    if not required_key:
+        return
+    if apikey == required_key:
+        return
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid API key",
+        headers={"WWW-Authenticate": "Bearer"},
+    )

gpu/self_hosted/app/config.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+
+SUPPORTED_FILE_EXTENSIONS = ["mp3", "mp4", "mpeg", "mpga", "m4a", "wav", "webm"]
+SAMPLE_RATE = 16000
+VAD_CONFIG = {
+    "batch_max_duration": 30.0,
+    "silence_padding": 0.5,
+    "window_size": 512,
+}
+
+# App-level paths
+UPLOADS_PATH = Path("/tmp/whisper-uploads")

gpu/self_hosted/app/factory.py

@@ -0,0 +1,30 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from .routers.diarization import router as diarization_router
+from .routers.transcription import router as transcription_router
+from .routers.translation import router as translation_router
+from .services.transcriber import WhisperService
+from .services.diarizer import PyannoteDiarizationService
+from .utils import ensure_dirs
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    ensure_dirs()
+    whisper_service = WhisperService()
+    whisper_service.load()
+    app.state.whisper = whisper_service
+    diarization_service = PyannoteDiarizationService()
+    diarization_service.load()
+    app.state.diarizer = diarization_service
+    yield
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(lifespan=lifespan)
+    app.include_router(transcription_router)
+    app.include_router(translation_router)
+    app.include_router(diarization_router)
+    return app

gpu/self_hosted/app/routers/diarization.py

@@ -0,0 +1,30 @@
+from typing import List
+
+from fastapi import APIRouter, Depends, Request
+from pydantic import BaseModel
+
+from ..auth import apikey_auth
+from ..services.diarizer import PyannoteDiarizationService
+from ..utils import download_audio_file
+
+router = APIRouter(tags=["diarization"])
+
+
+class DiarizationSegment(BaseModel):
+    start: float
+    end: float
+    speaker: int
+
+
+class DiarizationResponse(BaseModel):
+    diarization: List[DiarizationSegment]
+
+
+@router.post(
+    "/diarize", dependencies=[Depends(apikey_auth)], response_model=DiarizationResponse
+)
+def diarize(request: Request, audio_file_url: str, timestamp: float = 0.0):
+    with download_audio_file(audio_file_url) as (file_path, _ext):
+        file_path = str(file_path)
+        diarizer: PyannoteDiarizationService = request.app.state.diarizer
+        return diarizer.diarize_file(file_path, timestamp=timestamp)

gpu/self_hosted/app/routers/transcription.py

@@ -0,0 +1,109 @@
+import uuid
+from typing import Optional, Union
+
+from fastapi import APIRouter, Body, Depends, Form, HTTPException, Request, UploadFile
+from pydantic import BaseModel
+from pathlib import Path
+from ..auth import apikey_auth
+from ..config import SUPPORTED_FILE_EXTENSIONS, UPLOADS_PATH
+from ..services.transcriber import MODEL_NAME
+from ..utils import cleanup_uploaded_files, download_audio_file
+
+router = APIRouter(prefix="/v1/audio", tags=["transcription"])
+
+
+class WordTiming(BaseModel):
+    word: str
+    start: float
+    end: float
+
+
+class TranscriptResult(BaseModel):
+    text: str
+    words: list[WordTiming]
+    filename: Optional[str] = None
+
+
+class TranscriptBatchResponse(BaseModel):
+    results: list[TranscriptResult]
+
+
+@router.post(
+    "/transcriptions",
+    dependencies=[Depends(apikey_auth)],
+    response_model=Union[TranscriptResult, TranscriptBatchResponse],
+)
+def transcribe(
+    request: Request,
+    file: UploadFile = None,
+    files: list[UploadFile] | None = None,
+    model: str = Form(MODEL_NAME),
+    language: str = Form("en"),
+    batch: bool = Form(False),
+):
+    service = request.app.state.whisper
+    if not file and not files:
+        raise HTTPException(
+            status_code=400, detail="Either 'file' or 'files' parameter is required"
+        )
+    if batch and not files:
+        raise HTTPException(
+            status_code=400, detail="Batch transcription requires 'files'"
+        )
+
+    upload_files = [file] if file else files
+
+    uploaded_paths: list[Path] = []
+    with cleanup_uploaded_files(uploaded_paths):
+        for upload_file in upload_files:
+            audio_suffix = upload_file.filename.split(".")[-1].lower()
+            if audio_suffix not in SUPPORTED_FILE_EXTENSIONS:
+                raise HTTPException(
+                    status_code=400,
+                    detail=(
+                        f"Unsupported audio format. Supported extensions: {', '.join(SUPPORTED_FILE_EXTENSIONS)}"
+                    ),
+                )
+            unique_filename = f"{uuid.uuid4()}.{audio_suffix}"
+            file_path = UPLOADS_PATH / unique_filename
+            with open(file_path, "wb") as f:
+                content = upload_file.file.read()
+                f.write(content)
+            uploaded_paths.append(file_path)
+
+        if batch and len(upload_files) > 1:
+            results = []
+            for path in uploaded_paths:
+                result = service.transcribe_file(str(path), language=language)
+                result["filename"] = path.name
+                results.append(result)
+            return {"results": results}
+
+        results = []
+        for path in uploaded_paths:
+            result = service.transcribe_file(str(path), language=language)
+            result["filename"] = path.name
+            results.append(result)
+
+        return {"results": results} if len(results) > 1 else results[0]
+
+
+@router.post(
+    "/transcriptions-from-url",
+    dependencies=[Depends(apikey_auth)],
+    response_model=TranscriptResult,
+)
+def transcribe_from_url(
+    request: Request,
+    audio_file_url: str = Body(..., description="URL of the audio file to transcribe"),
+    model: str = Body(MODEL_NAME),
+    language: str = Body("en"),
+    timestamp_offset: float = Body(0.0),
+):
+    service = request.app.state.whisper
+    with download_audio_file(audio_file_url) as (file_path, _ext):
+        file_path = str(file_path)
+        result = service.transcribe_vad_url_segment(
+            file_path=file_path, timestamp_offset=timestamp_offset, language=language
+        )
+        return result

gpu/self_hosted/app/routers/translation.py

@@ -0,0 +1,28 @@
+from typing import Dict
+
+from fastapi import APIRouter, Body, Depends
+from pydantic import BaseModel
+
+from ..auth import apikey_auth
+from ..services.translator import TextTranslatorService
+
+router = APIRouter(tags=["translation"])
+
+translator = TextTranslatorService()
+
+
+class TranslationResponse(BaseModel):
+    text: Dict[str, str]
+
+
+@router.post(
+    "/translate",
+    dependencies=[Depends(apikey_auth)],
+    response_model=TranslationResponse,
+)
+def translate(
+    text: str,
+    source_language: str = Body("en"),
+    target_language: str = Body("fr"),
+):
+    return translator.translate(text, source_language, target_language)

gpu/self_hosted/app/services/diarizer.py

@@ -0,0 +1,42 @@
+import os
+import threading
+
+import torch
+import torchaudio
+from pyannote.audio import Pipeline
+
+
+class PyannoteDiarizationService:
+    def __init__(self):
+        self._pipeline = None
+        self._device = "cpu"
+        self._lock = threading.Lock()
+
+    def load(self):
+        self._device = "cuda" if torch.cuda.is_available() else "cpu"
+        self._pipeline = Pipeline.from_pretrained(
+            "pyannote/speaker-diarization-3.1",
+            use_auth_token=os.environ.get("HF_TOKEN"),
+        )
+        self._pipeline.to(torch.device(self._device))
+
+    def diarize_file(self, file_path: str, timestamp: float = 0.0) -> dict:
+        if self._pipeline is None:
+            self.load()
+        waveform, sample_rate = torchaudio.load(file_path)
+        with self._lock:
+            diarization = self._pipeline(
+                {"waveform": waveform, "sample_rate": sample_rate}
+            )
+        words = []
+        for diarization_segment, _, speaker in diarization.itertracks(yield_label=True):
+            words.append(
+                {
+                    "start": round(timestamp + diarization_segment.start, 3),
+                    "end": round(timestamp + diarization_segment.end, 3),
+                    "speaker": int(speaker[-2:])
+                    if speaker and speaker[-2:].isdigit()
+                    else 0,
+                }
+            )
+        return {"diarization": words}

gpu/self_hosted/app/services/transcriber.py

@@ -0,0 +1,208 @@
+import os
+import shutil
+import subprocess
+import threading
+from typing import Generator
+
+import faster_whisper
+import librosa
+import numpy as np
+import torch
+from fastapi import HTTPException
+from silero_vad import VADIterator, load_silero_vad
+
+from ..config import SAMPLE_RATE, VAD_CONFIG
+
+# Whisper configuration (service-local defaults)
+MODEL_NAME = "large-v2"
+# None delegates compute type to runtime: float16 on CUDA, int8 on CPU
+MODEL_COMPUTE_TYPE = None
+MODEL_NUM_WORKERS = 1
+CACHE_PATH = os.path.join(os.path.expanduser("~"), ".cache", "reflector-whisper")
+from ..utils import NoStdStreams
+
+
+class WhisperService:
+    def __init__(self):
+        self.model = None
+        self.device = "cpu"
+        self.lock = threading.Lock()
+
+    def load(self):
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        compute_type = MODEL_COMPUTE_TYPE or (
+            "float16" if self.device == "cuda" else "int8"
+        )
+        self.model = faster_whisper.WhisperModel(
+            MODEL_NAME,
+            device=self.device,
+            compute_type=compute_type,
+            num_workers=MODEL_NUM_WORKERS,
+            download_root=CACHE_PATH,
+        )
+
+    def pad_audio(self, audio_array, sample_rate: int = SAMPLE_RATE):
+        audio_duration = len(audio_array) / sample_rate
+        if audio_duration < VAD_CONFIG["silence_padding"]:
+            silence_samples = int(sample_rate * VAD_CONFIG["silence_padding"])
+            silence = np.zeros(silence_samples, dtype=np.float32)
+            return np.concatenate([audio_array, silence])
+        return audio_array
+
+    def enforce_word_timing_constraints(self, words: list[dict]) -> list[dict]:
+        if len(words) <= 1:
+            return words
+        enforced: list[dict] = []
+        for i, word in enumerate(words):
+            current = dict(word)
+            if i < len(words) - 1:
+                next_start = words[i + 1]["start"]
+                if current["end"] > next_start:
+                    current["end"] = next_start
+            enforced.append(current)
+        return enforced
+
+    def transcribe_file(self, file_path: str, language: str = "en") -> dict:
+        input_for_model: str | "object" = file_path
+        try:
+            audio_array, _sample_rate = librosa.load(
+                file_path, sr=SAMPLE_RATE, mono=True
+            )
+            if len(audio_array) / float(SAMPLE_RATE) < VAD_CONFIG["silence_padding"]:
+                input_for_model = self.pad_audio(audio_array, SAMPLE_RATE)
+        except Exception:
+            pass
+
+        with self.lock:
+            with NoStdStreams():
+                segments, _ = self.model.transcribe(
+                    input_for_model,
+                    language=language,
+                    beam_size=5,
+                    word_timestamps=True,
+                    vad_filter=True,
+                    vad_parameters={"min_silence_duration_ms": 500},
+                )
+
+        segments = list(segments)
+        text = "".join(segment.text for segment in segments).strip()
+        words = [
+            {
+                "word": word.word,
+                "start": round(float(word.start), 2),
+                "end": round(float(word.end), 2),
+            }
+            for segment in segments
+            for word in segment.words
+        ]
+        words = self.enforce_word_timing_constraints(words)
+        return {"text": text, "words": words}
+
+    def transcribe_vad_url_segment(
+        self, file_path: str, timestamp_offset: float = 0.0, language: str = "en"
+    ) -> dict:
+        def load_audio_via_ffmpeg(input_path: str, sample_rate: int) -> np.ndarray:
+            ffmpeg_bin = shutil.which("ffmpeg") or "ffmpeg"
+            cmd = [
+                ffmpeg_bin,
+                "-nostdin",
+                "-threads",
+                "1",
+                "-i",
+                input_path,
+                "-f",
+                "f32le",
+                "-acodec",
+                "pcm_f32le",
+                "-ac",
+                "1",
+                "-ar",
+                str(sample_rate),
+                "pipe:1",
+            ]
+            try:
+                proc = subprocess.run(
+                    cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True
+                )
+            except Exception as e:
+                raise HTTPException(status_code=400, detail=f"ffmpeg failed: {e}")
+            audio = np.frombuffer(proc.stdout, dtype=np.float32)
+            return audio
+
+        def vad_segments(
+            audio_array,
+            sample_rate: int = SAMPLE_RATE,
+            window_size: int = VAD_CONFIG["window_size"],
+        ) -> Generator[tuple[float, float], None, None]:
+            vad_model = load_silero_vad(onnx=False)
+            iterator = VADIterator(vad_model, sampling_rate=sample_rate)
+            start = None
+            for i in range(0, len(audio_array), window_size):
+                chunk = audio_array[i : i + window_size]
+                if len(chunk) < window_size:
+                    chunk = np.pad(
+                        chunk, (0, window_size - len(chunk)), mode="constant"
+                    )
+                speech = iterator(chunk)
+                if not speech:
+                    continue
+                if "start" in speech:
+                    start = speech["start"]
+                    continue
+                if "end" in speech and start is not None:
+                    end = speech["end"]
+                    yield (start / float(SAMPLE_RATE), end / float(SAMPLE_RATE))
+                    start = None
+            iterator.reset_states()
+
+        audio_array = load_audio_via_ffmpeg(file_path, SAMPLE_RATE)
+
+        merged_batches: list[tuple[float, float]] = []
+        batch_start = None
+        batch_end = None
+        max_duration = VAD_CONFIG["batch_max_duration"]
+        for seg_start, seg_end in vad_segments(audio_array):
+            if batch_start is None:
+                batch_start, batch_end = seg_start, seg_end
+                continue
+            if seg_end - batch_start <= max_duration:
+                batch_end = seg_end
+            else:
+                merged_batches.append((batch_start, batch_end))
+                batch_start, batch_end = seg_start, seg_end
+        if batch_start is not None and batch_end is not None:
+            merged_batches.append((batch_start, batch_end))
+
+        all_text = []
+        all_words = []
+        for start_time, end_time in merged_batches:
+            s_idx = int(start_time * SAMPLE_RATE)
+            e_idx = int(end_time * SAMPLE_RATE)
+            segment = audio_array[s_idx:e_idx]
+            segment = self.pad_audio(segment, SAMPLE_RATE)
+            with self.lock:
+                segments, _ = self.model.transcribe(
+                    segment,
+                    language=language,
+                    beam_size=5,
+                    word_timestamps=True,
+                    vad_filter=True,
+                    vad_parameters={"min_silence_duration_ms": 500},
+                )
+            segments = list(segments)
+            text = "".join(seg.text for seg in segments).strip()
+            words = [
+                {
+                    "word": w.word,
+                    "start": round(float(w.start) + start_time + timestamp_offset, 2),
+                    "end": round(float(w.end) + start_time + timestamp_offset, 2),
+                }
+                for seg in segments
+                for w in seg.words
+            ]
+            if text:
+                all_text.append(text)
+            all_words.extend(words)
+
+        all_words = self.enforce_word_timing_constraints(all_words)
+        return {"text": " ".join(all_text), "words": all_words}

gpu/self_hosted/app/services/translator.py

@@ -0,0 +1,44 @@
+import threading
+
+from transformers import MarianMTModel, MarianTokenizer, pipeline
+
+
+class TextTranslatorService:
+    """Simple text-to-text translator using HuggingFace MarianMT models.
+
+    This mirrors the modal translator API shape but uses text translation only.
+    """
+
+    def __init__(self):
+        self._pipeline = None
+        self._lock = threading.Lock()
+
+    def load(self, source_language: str = "en", target_language: str = "fr"):
+        # Pick a default MarianMT model pair if available; fall back to Helsinki-NLP en->fr
+        model_name = self._resolve_model_name(source_language, target_language)
+        tokenizer = MarianTokenizer.from_pretrained(model_name)
+        model = MarianMTModel.from_pretrained(model_name)
+        self._pipeline = pipeline("translation", model=model, tokenizer=tokenizer)
+
+    def _resolve_model_name(self, src: str, tgt: str) -> str:
+        # Minimal mapping; extend as needed
+        pair = (src.lower(), tgt.lower())
+        mapping = {
+            ("en", "fr"): "Helsinki-NLP/opus-mt-en-fr",
+            ("fr", "en"): "Helsinki-NLP/opus-mt-fr-en",
+            ("en", "es"): "Helsinki-NLP/opus-mt-en-es",
+            ("es", "en"): "Helsinki-NLP/opus-mt-es-en",
+            ("en", "de"): "Helsinki-NLP/opus-mt-en-de",
+            ("de", "en"): "Helsinki-NLP/opus-mt-de-en",
+        }
+        return mapping.get(pair, "Helsinki-NLP/opus-mt-en-fr")
+
+    def translate(self, text: str, source_language: str, target_language: str) -> dict:
+        if self._pipeline is None:
+            self.load(source_language, target_language)
+        with self._lock:
+            results = self._pipeline(
+                text, src_lang=source_language, tgt_lang=target_language
+            )
+        translated = results[0]["translation_text"] if results else ""
+        return {"text": {source_language: text, target_language: translated}}

gpu/self_hosted/app/utils.py

@@ -0,0 +1,107 @@
+import logging
+import os
+import sys
+import uuid
+from contextlib import contextmanager
+from typing import Mapping
+from urllib.parse import urlparse
+from pathlib import Path
+
+import requests
+from fastapi import HTTPException
+
+from .config import SUPPORTED_FILE_EXTENSIONS, UPLOADS_PATH
+
+logger = logging.getLogger(__name__)
+
+
+class NoStdStreams:
+    def __init__(self):
+        self.devnull = open(os.devnull, "w")
+
+    def __enter__(self):
+        self._stdout, self._stderr = sys.stdout, sys.stderr
+        self._stdout.flush()
+        self._stderr.flush()
+        sys.stdout, sys.stderr = self.devnull, self.devnull
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        sys.stdout, sys.stderr = self._stdout, self._stderr
+        self.devnull.close()
+
+
+def ensure_dirs():
+    UPLOADS_PATH.mkdir(parents=True, exist_ok=True)
+
+
+def detect_audio_format(url: str, headers: Mapping[str, str]) -> str:
+    url_path = urlparse(url).path
+    for ext in SUPPORTED_FILE_EXTENSIONS:
+        if url_path.lower().endswith(f".{ext}"):
+            return ext
+
+    content_type = headers.get("content-type", "").lower()
+    if "audio/mpeg" in content_type or "audio/mp3" in content_type:
+        return "mp3"
+    if "audio/wav" in content_type:
+        return "wav"
+    if "audio/mp4" in content_type:
+        return "mp4"
+
+    raise HTTPException(
+        status_code=400,
+        detail=(
+            f"Unsupported audio format for URL. Supported extensions: {', '.join(SUPPORTED_FILE_EXTENSIONS)}"
+        ),
+    )
+
+
+def download_audio_to_uploads(audio_file_url: str) -> tuple[Path, str]:
+    response = requests.head(audio_file_url, allow_redirects=True)
+    if response.status_code == 404:
+        raise HTTPException(status_code=404, detail="Audio file not found")
+
+    response = requests.get(audio_file_url, allow_redirects=True)
+    response.raise_for_status()
+
+    audio_suffix = detect_audio_format(audio_file_url, response.headers)
+    unique_filename = f"{uuid.uuid4()}.{audio_suffix}"
+    file_path: Path = UPLOADS_PATH / unique_filename
+
+    with open(file_path, "wb") as f:
+        f.write(response.content)
+
+    return file_path, audio_suffix
+
+
+@contextmanager
+def download_audio_file(audio_file_url: str):
+    """Download an audio file to UPLOADS_PATH and remove it after use.
+
+    Yields (file_path: Path, audio_suffix: str).
+    """
+    file_path, audio_suffix = download_audio_to_uploads(audio_file_url)
+    try:
+        yield file_path, audio_suffix
+    finally:
+        try:
+            file_path.unlink(missing_ok=True)
+        except Exception as e:
+            logger.error("Error deleting temporary file %s: %s", file_path, e)
+
+
+@contextmanager
+def cleanup_uploaded_files(file_paths: list[Path]):
+    """Ensure provided file paths are removed after use.
+
+    The provided list can be populated inside the context; all present entries
+    at exit will be deleted.
+    """
+    try:
+        yield file_paths
+    finally:
+        for path in list(file_paths):
+            try:
+                path.unlink(missing_ok=True)
+            except Exception as e:
+                logger.error("Error deleting temporary file %s: %s", path, e)

gpu/self_hosted/compose.yml

@@ -0,0 +1,10 @@
+services:
+  reflector_gpu:
+    build:
+      context: .
+    ports:
+      - "8000:8000"
+    env_file:
+      - .env
+    volumes:
+      - ./cache:/root/.cache

gpu/self_hosted/main.py

@@ -0,0 +1,3 @@
+from app.factory import create_app
+
+app = create_app()

gpu/self_hosted/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "reflector-gpu"
+version = "0.1.0"
+description = "Self-hosted GPU service for speech transcription, diarization, and translation via FastAPI."
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi[standard]>=0.116.1",
+    "uvicorn[standard]>=0.30.0",
+    "torch>=2.3.0",
+    "faster-whisper>=1.1.0",
+    "librosa==0.10.1",
+    "numpy<2",
+    "silero-vad==5.1.0",
+    "transformers>=4.35.0",
+    "sentencepiece",
+    "pyannote.audio==3.1.0",
+    "torchaudio>=2.3.0",
+]

gpu/self_hosted/runserver.sh

@@ -0,0 +1,17 @@
+#!/bin/sh
+set -e
+
+export PATH="/root/.local/bin:$PATH"
+cd /app
+
+# Install Python dependencies at runtime (first run or when FORCE_SYNC=1)
+if [ ! -d "/app/.venv" ] || [ "$FORCE_SYNC" = "1" ]; then
+  echo "[startup] Installing Python dependencies with uv..."
+  uv sync --compile-bytecode --locked
+else
+  echo "[startup] Using existing virtual environment at /app/.venv"
+fi
+
+exec uv run uvicorn main:app --host 0.0.0.0 --port 8000
+
+

server/README.md

@@ -1,3 +1,29 @@
+## API Key Management
+
+### Finding Your User ID
+
+```bash
+# Get your OAuth sub (user ID) - requires authentication
+curl -H "Authorization: Bearer <your_jwt>" http://localhost:1250/v1/me
+# Returns: {"sub": "your-oauth-sub-here", "email": "...", ...}
+```
+
+### Creating API Keys
+
+```bash
+curl -X POST http://localhost:1250/v1/user/api-keys \
+  -H "Authorization: Bearer <your_jwt>" \
+  -H "Content-Type: application/json" \
+  -d '{"name": "My API Key"}'
+```
+
+### Using API Keys
+
+```bash
+# Use X-API-Key header instead of Authorization
+curl -H "X-API-Key: <your_api_key>" http://localhost:1250/v1/transcripts
+```
+
 ## AWS S3/SQS usage clarification
 
 Whereby.com uploads recordings directly to our S3 bucket when meetings end.

server/docs/gpu/api-transcription.md

@@ -0,0 +1,194 @@
+## Reflector GPU Transcription API (Specification)
+
+This document defines the Reflector GPU transcription API that all implementations must adhere to. Current implementations include NVIDIA Parakeet (NeMo) and Whisper (faster-whisper), both deployed on Modal.com. The API surface and response shapes are OpenAI/Whisper-compatible, so clients can switch implementations by changing only the base URL.
+
+### Base URL and Authentication
+
+- Example base URLs (Modal web endpoints):
+
+  - Parakeet: `https://<account>--reflector-transcriber-parakeet-web.modal.run`
+  - Whisper: `https://<account>--reflector-transcriber-web.modal.run`
+
+- All endpoints are served under `/v1` and require a Bearer token:
+
+```
+Authorization: Bearer <REFLECTOR_GPU_APIKEY>
+```
+
+Note: To switch implementations, deploy the desired variant and point `TRANSCRIPT_URL` to its base URL. The API is identical.
+
+### Supported file types
+
+`mp3, mp4, mpeg, mpga, m4a, wav, webm`
+
+### Models and languages
+
+- Parakeet (NVIDIA NeMo): default `nvidia/parakeet-tdt-0.6b-v2`
+  - Language support: only `en`. Other languages return HTTP 400.
+- Whisper (faster-whisper): default `large-v2` (or deployment-specific)
+  - Language support: multilingual (per Whisper model capabilities).
+
+Note: The `model` parameter is accepted by all implementations for interface parity. Some backends may treat it as informational.
+
+### Endpoints
+
+#### POST /v1/audio/transcriptions
+
+Transcribe one or more uploaded audio files.
+
+Request: multipart/form-data
+
+- `file` (File) — optional. Single file to transcribe.
+- `files` (File[]) — optional. One or more files to transcribe.
+- `model` (string) — optional. Defaults to the implementation-specific model (see above).
+- `language` (string) — optional, defaults to `en`.
+  - Parakeet: only `en` is accepted; other values return HTTP 400
+  - Whisper: model-dependent; typically multilingual
+- `batch` (boolean) — optional, defaults to `false`.
+
+Notes:
+
+- Provide either `file` or `files`, not both. If neither is provided, HTTP 400.
+- `batch` requires `files`; using `batch=true` without `files` returns HTTP 400.
+- Response shape for multiple files is the same regardless of `batch`.
+- Files sent to this endpoint are processed in a single pass (no VAD/chunking). This is intended for short clips (roughly ≤ 30s; depends on GPU memory/model). For longer audio, prefer `/v1/audio/transcriptions-from-url` which supports VAD-based chunking.
+
+Responses
+
+Single file response:
+
+```json
+{
+  "text": "transcribed text",
+  "words": [
+    { "word": "hello", "start": 0.0, "end": 0.5 },
+    { "word": "world", "start": 0.5, "end": 1.0 }
+  ],
+  "filename": "audio.mp3"
+}
+```
+
+Multiple files response:
+
+```json
+{
+  "results": [
+    {"filename": "a1.mp3", "text": "...", "words": [...]},
+    {"filename": "a2.mp3", "text": "...", "words": [...]}]
+}
+```
+
+Notes:
+
+- Word objects always include keys: `word`, `start`, `end`.
+- Some implementations may include a trailing space in `word` to match Whisper tokenization behavior; clients should trim if needed.
+
+Example curl (single file):
+
+```bash
+curl -X POST \
+  -H "Authorization: Bearer $REFLECTOR_GPU_APIKEY" \
+  -F "file=@/path/to/audio.mp3" \
+  -F "language=en" \
+  "$BASE_URL/v1/audio/transcriptions"
+```
+
+Example curl (multiple files, batch):
+
+```bash
+curl -X POST \
+  -H "Authorization: Bearer $REFLECTOR_GPU_APIKEY" \
+  -F "files=@/path/a1.mp3" -F "files=@/path/a2.mp3" \
+  -F "batch=true" -F "language=en" \
+  "$BASE_URL/v1/audio/transcriptions"
+```
+
+#### POST /v1/audio/transcriptions-from-url
+
+Transcribe a single remote audio file by URL.
+
+Request: application/json
+
+Body parameters:
+
+- `audio_file_url` (string) — required. URL of the audio file to transcribe.
+- `model` (string) — optional. Defaults to the implementation-specific model (see above).
+- `language` (string) — optional, defaults to `en`. Parakeet only accepts `en`.
+- `timestamp_offset` (number) — optional, defaults to `0.0`. Added to each word's `start`/`end` in the response.
+
+```json
+{
+  "audio_file_url": "https://example.com/audio.mp3",
+  "model": "nvidia/parakeet-tdt-0.6b-v2",
+  "language": "en",
+  "timestamp_offset": 0.0
+}
+```
+
+Response:
+
+```json
+{
+  "text": "transcribed text",
+  "words": [
+    { "word": "hello", "start": 10.0, "end": 10.5 },
+    { "word": "world", "start": 10.5, "end": 11.0 }
+  ]
+}
+```
+
+Notes:
+
+- `timestamp_offset` is added to each word’s `start`/`end` in the response.
+- Implementations may perform VAD-based chunking and batching for long-form audio; word timings are adjusted accordingly.
+
+Example curl:
+
+```bash
+curl -X POST \
+  -H "Authorization: Bearer $REFLECTOR_GPU_APIKEY" \
+  -H "Content-Type: application/json" \
+  -d '{
+        "audio_file_url": "https://example.com/audio.mp3",
+        "language": "en",
+        "timestamp_offset": 0
+      }' \
+  "$BASE_URL/v1/audio/transcriptions-from-url"
+```
+
+### Error handling
+
+- 400 Bad Request
+  - Parakeet: `language` other than `en`
+  - Missing required parameters (`file`/`files` for upload; `audio_file_url` for URL endpoint)
+  - Unsupported file extension
+- 401 Unauthorized
+  - Missing or invalid Bearer token
+- 404 Not Found
+  - `audio_file_url` does not exist
+
+### Implementation details
+
+- GPUs: A10G for small-file/live, L40S for large-file URL transcription (subject to deployment)
+- VAD chunking and segment batching; word timings adjusted and overlapping ends constrained
+- Pads very short segments (< 0.5s) to avoid model crashes on some backends
+
+### Server configuration (Reflector API)
+
+Set the Reflector server to use the Modal backend and point `TRANSCRIPT_URL` to your chosen deployment:
+
+```
+TRANSCRIPT_BACKEND=modal
+TRANSCRIPT_URL=https://<account>--reflector-transcriber-parakeet-web.modal.run
+TRANSCRIPT_MODAL_API_KEY=<REFLECTOR_GPU_APIKEY>
+```
+
+### Conformance tests
+
+Use the pytest-based conformance tests to validate any new implementation (including self-hosted) against this spec:
+
+```
+TRANSCRIPT_URL=https://<your-deployment-base> \
+TRANSCRIPT_MODAL_API_KEY=your-api-key \
+uv run -m pytest -m model_api --no-cov server/tests/test_model_api_transcript.py
+```

server/docs/video-platforms/README.md

@@ -0,0 +1,234 @@
+# Reflector Architecture: Whereby + Daily.co Recording Storage
+
+## System Overview
+
+```mermaid
+graph TB
+    subgraph "Actors"
+        APP[Our App<br/>Reflector]
+        WHEREBY[Whereby Service<br/>External]
+        DAILY[Daily.co Service<br/>External]
+    end
+
+    subgraph "AWS S3 Buckets"
+        TRANSCRIPT_BUCKET[Transcript Bucket<br/>reflector-transcripts<br/>Output: Processed MP3s]
+        WHEREBY_BUCKET[Whereby Bucket<br/>reflector-whereby-recordings<br/>Input: Raw MP4s]
+        DAILY_BUCKET[Daily.co Bucket<br/>reflector-dailyco-recordings<br/>Input: Raw WebM tracks]
+    end
+
+    subgraph "AWS Infrastructure"
+        SQS[SQS Queue<br/>Whereby notifications]
+    end
+
+    subgraph "Database"
+        DB[(PostgreSQL<br/>Recordings, Transcripts, Meetings)]
+    end
+
+    APP -->|Write processed| TRANSCRIPT_BUCKET
+    APP -->|Read/Delete| WHEREBY_BUCKET
+    APP -->|Read/Delete| DAILY_BUCKET
+    APP -->|Poll| SQS
+    APP -->|Store metadata| DB
+
+    WHEREBY -->|Write recordings| WHEREBY_BUCKET
+    WHEREBY_BUCKET -->|S3 Event| SQS
+    WHEREBY -->|Participant webhooks<br/>room.client.joined/left| APP
+
+    DAILY -->|Write recordings| DAILY_BUCKET
+    DAILY -->|Recording webhook<br/>recording.ready-to-download| APP
+```
+
+**Note on Webhook vs S3 Event for Recording Processing:**
+- **Whereby**: Uses S3 Events → SQS for recording availability (S3 as source of truth, no race conditions)
+- **Daily.co**: Uses webhooks for recording availability (more immediate, built-in reliability)
+- **Both**: Use webhooks for participant tracking (real-time updates)
+
+## Credentials & Permissions
+
+```mermaid
+graph LR
+    subgraph "Master Credentials"
+        MASTER[TRANSCRIPT_STORAGE_AWS_*<br/>Access Key ID + Secret]
+    end
+
+    subgraph "Whereby Upload Credentials"
+        WHEREBY_CREDS[AWS_WHEREBY_ACCESS_KEY_*<br/>Access Key ID + Secret]
+    end
+
+    subgraph "Daily.co Upload Role"
+        DAILY_ROLE[DAILY_STORAGE_AWS_ROLE_ARN<br/>IAM Role ARN]
+    end
+
+    subgraph "Our App Uses"
+        MASTER -->|Read/Write/Delete| TRANSCRIPT_BUCKET[Transcript Bucket]
+        MASTER -->|Read/Delete| WHEREBY_BUCKET[Whereby Bucket]
+        MASTER -->|Read/Delete| DAILY_BUCKET[Daily.co Bucket]
+        MASTER -->|Poll/Delete| SQS[SQS Queue]
+    end
+
+    subgraph "We Give To Services"
+        WHEREBY_CREDS -->|Passed in API call| WHEREBY_SERVICE[Whereby Service]
+        WHEREBY_SERVICE -->|Write Only| WHEREBY_BUCKET
+
+        DAILY_ROLE -->|Passed in API call| DAILY_SERVICE[Daily.co Service]
+        DAILY_SERVICE -->|Assume Role| DAILY_ROLE
+        DAILY_SERVICE -->|Write Only| DAILY_BUCKET
+    end
+```
+
+# Video Platform Recording Integration
+
+This document explains how Reflector receives and identifies multitrack audio recordings from different video platforms.
+
+## Platform Comparison
+
+| Platform | Delivery Method | Track Identification |
+|----------|----------------|---------------------|
+| **Daily.co** | Webhook | Explicit track list in payload |
+| **Whereby** | SQS (S3 notifications) | Single file per notification |
+
+---
+
+## Daily.co (Webhook-based)
+
+Daily.co uses **webhooks** to notify Reflector when recordings are ready.
+
+### How It Works
+
+1. **Daily.co sends webhook** when recording is ready
+   - Event type: `recording.ready-to-download`
+   - Endpoint: `/v1/daily/webhook` (`reflector/views/daily.py:46-102`)
+
+2. **Webhook payload explicitly includes track list**:
+```json
+{
+  "recording_id": "7443ee0a-dab1-40eb-b316-33d6c0d5ff88",
+  "room_name": "daily-20251020193458",
+  "tracks": [
+    {
+      "type": "audio",
+      "s3Key": "monadical/daily-20251020193458/1760988935484-52f7f48b-fbab-431f-9a50-87b9abfc8255-cam-audio-1760988935922",
+      "size": 831843
+    },
+    {
+      "type": "audio",
+      "s3Key": "monadical/daily-20251020193458/1760988935484-a37c35e3-6f8e-4274-a482-e9d0f102a732-cam-audio-1760988943823",
+      "size": 408438
+    },
+    {
+      "type": "video",
+      "s3Key": "monadical/daily-20251020193458/...-video.webm",
+      "size": 30000000
+    }
+  ]
+}
+```
+
+3. **System extracts audio tracks** (`daily.py:211`):
+```python
+track_keys = [t.s3Key for t in tracks if t.type == "audio"]
+```
+
+4. **Triggers multitrack processing** (`daily.py:213-218`):
+```python
+process_multitrack_recording.delay(
+    bucket_name=bucket_name,  # reflector-dailyco-local
+    room_name=room_name,      # daily-20251020193458
+    recording_id=recording_id, # 7443ee0a-dab1-40eb-b316-33d6c0d5ff88
+    track_keys=track_keys      # Only audio s3Keys
+)
+```
+
+### Key Advantage: No Ambiguity
+
+Even though multiple meetings may share the same S3 bucket/folder (`monadical/`), **there's no ambiguity** because:
+- Each webhook payload contains the exact `s3Key` list for that specific `recording_id`
+- No need to scan folders or guess which files belong together
+- Each track's s3Key includes the room timestamp subfolder (e.g., `daily-20251020193458/`)
+
+The room name includes timestamp (`daily-20251020193458`) to keep recordings organized, but **the webhook's explicit track list is what prevents mixing files from different meetings**.
+
+### Track Timeline Extraction
+
+Daily.co provides timing information in two places:
+
+**1. PyAV WebM Metadata (current approach)**:
+```python
+# Read from WebM container stream metadata
+stream.start_time = 8.130s  # Meeting-relative timing
+```
+
+**2. Filename Timestamps (alternative approach, commit 3bae9076)**:
+```
+Filename format: {recording_start_ts}-{uuid}-cam-audio-{track_start_ts}.webm
+Example: 1760988935484-52f7f48b-fbab-431f-9a50-87b9abfc8255-cam-audio-1760988935922.webm
+
+Parse timestamps:
+- recording_start_ts: 1760988935484 (Unix ms)
+- track_start_ts: 1760988935922 (Unix ms)
+- offset: (1760988935922 - 1760988935484) / 1000 = 0.438s
+```
+
+**Time Difference (PyAV vs Filename)**:
+```
+Track 0:
+  Filename offset: 438ms
+  PyAV metadata:   229ms
+  Difference:      209ms
+
+Track 1:
+  Filename offset: 8339ms
+  PyAV metadata:   8130ms
+  Difference:      209ms
+```
+
+**Consistent 209ms delta** suggests network/encoding delay between file upload initiation (filename) and actual audio stream start (metadata).
+
+**Current implementation uses PyAV metadata** because:
+- More accurate (represents when audio actually started)
+- Padding BEFORE transcription produces correct Whisper timestamps automatically
+- No manual offset adjustment needed during transcript merge
+
+### Why Re-encoding During Padding
+
+Padding coincidentally involves re-encoding, which is important for Daily.co + Whisper:
+
+**Problem:** Daily.co skips frames in recordings when microphone is muted or paused
+- WebM containers have gaps where audio frames should be
+- Whisper doesn't understand these gaps and produces incorrect timestamps
+- Example: 5s of audio with 2s muted → file has frames only for 3s, Whisper thinks duration is 3s
+
+**Solution:** Re-encoding via PyAV filter graph (`adelay` + `aresample`)
+- Restores missing frames as silence
+- Produces continuous audio stream without gaps
+- Whisper now sees correct duration and produces accurate timestamps
+
+**Why combined with padding:**
+- Already re-encoding for padding (adding initial silence)
+- More performant to do both operations in single PyAV pipeline
+- Padded values needed for mixdown anyway (creating final MP3)
+
+Implementation: `main_multitrack_pipeline.py:_apply_audio_padding_streaming()`
+
+---
+
+## Whereby (SQS-based)
+
+Whereby uses **AWS SQS** (via S3 notifications) to notify Reflector when files are uploaded.
+
+### How It Works
+
+1. **Whereby uploads recording** to S3
+2. **S3 sends notification** to SQS queue (one notification per file)
+3. **Reflector polls SQS queue** (`worker/process.py:process_messages()`)
+4. **System processes single file** (`worker/process.py:process_recording()`)
+
+### Key Difference from Daily.co
+
+**Whereby (SQS):** System receives S3 notification "file X was created" - only knows about one file at a time, would need to scan folder to find related files
+
+**Daily.co (Webhook):** Daily explicitly tells system which files belong together in the webhook payload
+
+---
+
+

server/docs/webhook.md

@@ -14,7 +14,7 @@ Webhooks are configured at the room level with two fields:
 
 ### `transcript.completed`
 
-Triggered when a transcript has been fully processed, including transcription, diarization, summarization, and topic detection.
+Triggered when a transcript has been fully processed, including transcription, diarization, summarization, topic detection and calendar event integration.
 
 ### `test`
 
@@ -128,6 +128,27 @@ This event includes a convenient URL for accessing the transcript:
   "room": {
     "id": "room-789",
     "name": "Product Team Room"
+  },
+  "calendar_event": {
+    "id": "calendar-event-123",
+    "ics_uid": "event-123",
+    "title": "Q3 Product Planning Meeting",
+    "start_time": "2025-08-27T12:00:00Z",
+    "end_time": "2025-08-27T12:30:00Z",
+    "description": "Team discussed Q3 product roadmap, prioritizing mobile app features and API improvements.",
+    "location": "Conference Room 1",
+    "attendees": [
+      {
+        "id": "participant-1",
+        "name": "John Doe",
+        "speaker": "Speaker 1"
+      },
+      {
+        "id": "participant-2",
+        "name": "Jane Smith",
+        "speaker": "Speaker 2"
+      }
+    ]
   }
 }
 ```

server/env.example

@@ -27,7 +27,7 @@ AUTH_JWT_AUDIENCE=
 #TRANSCRIPT_MODAL_API_KEY=xxxxx
 
 TRANSCRIPT_BACKEND=modal
-TRANSCRIPT_URL=https://monadical-sas--reflector-transcriber-web.modal.run
+TRANSCRIPT_URL=https://monadical-sas--reflector-transcriber-parakeet-web.modal.run
 TRANSCRIPT_MODAL_API_KEY=
 
 ## =======================================================
@@ -71,3 +71,30 @@ DIARIZATION_URL=https://monadical-sas--reflector-diarizer-web.modal.run
 
 ## Sentry DSN configuration
 #SENTRY_DSN=
+
+## =======================================================
+## Video Platform Configuration
+## =======================================================
+
+## Whereby
+#WHEREBY_API_KEY=your-whereby-api-key
+#WHEREBY_WEBHOOK_SECRET=your-whereby-webhook-secret
+#WHEREBY_STORAGE_AWS_ACCESS_KEY_ID=your-aws-key
+#WHEREBY_STORAGE_AWS_SECRET_ACCESS_KEY=your-aws-secret
+#AWS_PROCESS_RECORDING_QUEUE_URL=https://sqs.us-west-2.amazonaws.com/...
+
+## Daily.co
+#DAILY_API_KEY=your-daily-api-key
+#DAILY_WEBHOOK_SECRET=your-daily-webhook-secret
+#DAILY_SUBDOMAIN=your-subdomain
+#DAILY_WEBHOOK_UUID=  # Auto-populated by recreate_daily_webhook.py script
+#DAILYCO_STORAGE_AWS_ROLE_ARN=...  # IAM role ARN for Daily.co S3 access
+#DAILYCO_STORAGE_AWS_BUCKET_NAME=reflector-dailyco
+#DAILYCO_STORAGE_AWS_REGION=us-west-2
+
+## Whereby (optional separate bucket)
+#WHEREBY_STORAGE_AWS_BUCKET_NAME=reflector-whereby
+#WHEREBY_STORAGE_AWS_REGION=us-east-1
+
+## Platform Configuration
+#DEFAULT_VIDEO_PLATFORM=whereby          # Default platform for new rooms

server/gpu/modal_deployments/reflector_transcriber.py

@@ -1,161 +0,0 @@
-import os
-import tempfile
-import threading
-
-import modal
-from pydantic import BaseModel
-
-MODELS_DIR = "/models"
-
-MODEL_NAME = "large-v2"
-MODEL_COMPUTE_TYPE: str = "float16"
-MODEL_NUM_WORKERS: int = 1
-
-MINUTES = 60  # seconds
-
-volume = modal.Volume.from_name("models", create_if_missing=True)
-
-app = modal.App("reflector-transcriber")
-
-
-def download_model():
-    from faster_whisper import download_model
-
-    volume.reload()
-
-    download_model(MODEL_NAME, cache_dir=MODELS_DIR)
-
-    volume.commit()
-
-
-image = (
-    modal.Image.debian_slim(python_version="3.12")
-    .pip_install(
-        "huggingface_hub==0.27.1",
-        "hf-transfer==0.1.9",
-        "torch==2.5.1",
-        "faster-whisper==1.1.1",
-    )
-    .env(
-        {
-            "HF_HUB_ENABLE_HF_TRANSFER": "1",
-            "LD_LIBRARY_PATH": (
-                "/usr/local/lib/python3.12/site-packages/nvidia/cudnn/lib/:"
-                "/opt/conda/lib/python3.12/site-packages/nvidia/cublas/lib/"
-            ),
-        }
-    )
-    .run_function(download_model, volumes={MODELS_DIR: volume})
-)
-
-
-@app.cls(
-    gpu="A10G",
-    timeout=5 * MINUTES,
-    scaledown_window=5 * MINUTES,
-    allow_concurrent_inputs=6,
-    image=image,
-    volumes={MODELS_DIR: volume},
-)
-class Transcriber:
-    @modal.enter()
-    def enter(self):
-        import faster_whisper
-        import torch
-
-        self.lock = threading.Lock()
-        self.use_gpu = torch.cuda.is_available()
-        self.device = "cuda" if self.use_gpu else "cpu"
-        self.model = faster_whisper.WhisperModel(
-            MODEL_NAME,
-            device=self.device,
-            compute_type=MODEL_COMPUTE_TYPE,
-            num_workers=MODEL_NUM_WORKERS,
-            download_root=MODELS_DIR,
-            local_files_only=True,
-        )
-
-    @modal.method()
-    def transcribe_segment(
-        self,
-        audio_data: str,
-        audio_suffix: str,
-        language: str,
-    ):
-        with tempfile.NamedTemporaryFile("wb+", suffix=f".{audio_suffix}") as fp:
-            fp.write(audio_data)
-
-            with self.lock:
-                segments, _ = self.model.transcribe(
-                    fp.name,
-                    language=language,
-                    beam_size=5,
-                    word_timestamps=True,
-                    vad_filter=True,
-                    vad_parameters={"min_silence_duration_ms": 500},
-                )
-
-            segments = list(segments)
-            text = "".join(segment.text for segment in segments)
-            words = [
-                {"word": word.word, "start": word.start, "end": word.end}
-                for segment in segments
-                for word in segment.words
-            ]
-
-            return {"text": text, "words": words}
-
-
-@app.function(
-    scaledown_window=60,
-    timeout=60,
-    allow_concurrent_inputs=40,
-    secrets=[
-        modal.Secret.from_name("reflector-gpu"),
-    ],
-    volumes={MODELS_DIR: volume},
-)
-@modal.asgi_app()
-def web():
-    from fastapi import Body, Depends, FastAPI, HTTPException, UploadFile, status
-    from fastapi.security import OAuth2PasswordBearer
-    from typing_extensions import Annotated
-
-    transcriber = Transcriber()
-
-    app = FastAPI()
-
-    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-
-    supported_file_types = ["mp3", "mp4", "mpeg", "mpga", "m4a", "wav", "webm"]
-
-    def apikey_auth(apikey: str = Depends(oauth2_scheme)):
-        if apikey != os.environ["REFLECTOR_GPU_APIKEY"]:
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Invalid API key",
-                headers={"WWW-Authenticate": "Bearer"},
-            )
-
-    class TranscriptResponse(BaseModel):
-        result: dict
-
-    @app.post("/v1/audio/transcriptions", dependencies=[Depends(apikey_auth)])
-    def transcribe(
-        file: UploadFile,
-        model: str = "whisper-1",
-        language: Annotated[str, Body(...)] = "en",
-    ) -> TranscriptResponse:
-        audio_data = file.file.read()
-        audio_suffix = file.filename.split(".")[-1]
-        assert audio_suffix in supported_file_types
-
-        func = transcriber.transcribe_segment.spawn(
-            audio_data=audio_data,
-            audio_suffix=audio_suffix,
-            language=language,
-        )
-        result = func.get()
-        return result
-
-    return app

server/migrations/versions/0ce521cda2ee_remove_user_id_from_meeting_table.py

@@ -0,0 +1,36 @@
+"""remove user_id from meeting table
+
+Revision ID: 0ce521cda2ee
+Revises: 6dec9fb5b46c
+Create Date: 2025-09-10 12:40:55.688899
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0ce521cda2ee"
+down_revision: Union[str, None] = "6dec9fb5b46c"
+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", schema=None) as batch_op:
+        batch_op.drop_column("user_id")
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("user_id", sa.VARCHAR(), autoincrement=False, nullable=True)
+        )
+
+    # ### end Alembic commands ###

server/migrations/versions/1e49625677e4_add_platform_support.py

@@ -0,0 +1,50 @@
+"""add_platform_support
+
+Revision ID: 1e49625677e4
+Revises: 9e3f7b2a4c8e
+Create Date: 2025-10-08 13:17:29.943612
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "1e49625677e4"
+down_revision: Union[str, None] = "9e3f7b2a4c8e"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add platform field with default 'whereby' for backward compatibility."""
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "platform",
+                sa.String(),
+                nullable=True,
+                server_default=None,
+            )
+        )
+
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "platform",
+                sa.String(),
+                nullable=False,
+                server_default="whereby",
+            )
+        )
+
+
+def downgrade() -> None:
+    """Remove platform field."""
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.drop_column("platform")
+
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.drop_column("platform")

server/migrations/versions/2ae3db106d4e_clean_up_orphaned_room_id_references_in_.py

@@ -0,0 +1,32 @@
+"""clean up orphaned room_id references in meeting table
+
+Revision ID: 2ae3db106d4e
+Revises: def1b5867d4c
+Create Date: 2025-09-11 10:35:15.759967
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "2ae3db106d4e"
+down_revision: Union[str, None] = "def1b5867d4c"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Set room_id to NULL for meetings that reference non-existent rooms
+    op.execute("""
+        UPDATE meeting
+        SET room_id = NULL
+        WHERE room_id IS NOT NULL
+          AND room_id NOT IN (SELECT id FROM room WHERE id IS NOT NULL)
+    """)
+
+
+def downgrade() -> None:
+    # Cannot restore orphaned references - no operation needed
+    pass

server/migrations/versions/6025e9b2bef2_remove_one_active_meeting_per_room_.py

@@ -0,0 +1,53 @@
+"""remove_one_active_meeting_per_room_constraint
+
+Revision ID: 6025e9b2bef2
+Revises: 2ae3db106d4e
+Create Date: 2025-08-18 18:45:44.418392
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "6025e9b2bef2"
+down_revision: Union[str, None] = "2ae3db106d4e"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Remove the unique constraint that prevents multiple active meetings per room
+    # This is needed to support calendar integration with overlapping meetings
+    # Check if index exists before trying to drop it
+    from alembic import context
+
+    if context.get_context().dialect.name == "postgresql":
+        conn = op.get_bind()
+        result = conn.execute(
+            sa.text(
+                "SELECT 1 FROM pg_indexes WHERE indexname = 'idx_one_active_meeting_per_room'"
+            )
+        )
+        if result.fetchone():
+            op.drop_index("idx_one_active_meeting_per_room", table_name="meeting")
+    else:
+        # For SQLite, just try to drop it
+        try:
+            op.drop_index("idx_one_active_meeting_per_room", table_name="meeting")
+        except:
+            pass
+
+
+def downgrade() -> None:
+    # Restore the unique constraint
+    op.create_index(
+        "idx_one_active_meeting_per_room",
+        "meeting",
+        ["room_id"],
+        unique=True,
+        postgresql_where=sa.text("is_active = true"),
+        sqlite_where=sa.text("is_active = 1"),
+    )

server/migrations/versions/6dec9fb5b46c_make_meeting_room_id_required_and_add_.py

@@ -0,0 +1,35 @@
+"""make meeting room_id required and add foreign key
+
+Revision ID: 6dec9fb5b46c
+Revises: 61882a919591
+Create Date: 2025-09-10 10:47:06.006819
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "6dec9fb5b46c"
+down_revision: Union[str, None] = "61882a919591"
+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", schema=None) as batch_op:
+        batch_op.create_foreign_key(
+            None, "room", ["room_id"], ["id"], ondelete="CASCADE"
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.drop_constraint("meeting_room_id_fkey", type_="foreignkey")
+
+    # ### end Alembic commands ###

server/migrations/versions/9e3f7b2a4c8e_add_user_api_keys.py

@@ -0,0 +1,38 @@
+"""add user api keys
+
+Revision ID: 9e3f7b2a4c8e
+Revises: dc035ff72fd5
+Create Date: 2025-10-17 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "9e3f7b2a4c8e"
+down_revision: Union[str, None] = "dc035ff72fd5"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_api_key",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("user_id", sa.String(), nullable=False),
+        sa.Column("key_hash", sa.String(), nullable=False),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+    with op.batch_alter_table("user_api_key", schema=None) as batch_op:
+        batch_op.create_index("idx_user_api_key_hash", ["key_hash"], unique=True)
+        batch_op.create_index("idx_user_api_key_user_id", ["user_id"], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_table("user_api_key")

server/migrations/versions/d4a1c446458c_add_grace_period_fields_to_meeting.py

@@ -0,0 +1,34 @@
+"""add_grace_period_fields_to_meeting
+
+Revision ID: d4a1c446458c
+Revises: 6025e9b2bef2
+Create Date: 2025-08-18 18:50:37.768052
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "d4a1c446458c"
+down_revision: Union[str, None] = "6025e9b2bef2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add fields to track when participants left for grace period logic
+    op.add_column(
+        "meeting", sa.Column("last_participant_left_at", sa.DateTime(timezone=True))
+    )
+    op.add_column(
+        "meeting",
+        sa.Column("grace_period_minutes", sa.Integer, server_default=sa.text("15")),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("meeting", "grace_period_minutes")
+    op.drop_column("meeting", "last_participant_left_at")

server/migrations/versions/d8e204bbf615_add_calendar.py

@@ -0,0 +1,129 @@
+"""add calendar
+
+Revision ID: d8e204bbf615
+Revises: d4a1c446458c
+Create Date: 2025-09-10 19:56:22.295756
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "d8e204bbf615"
+down_revision: Union[str, None] = "d4a1c446458c"
+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! ###
+    op.create_table(
+        "calendar_event",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("room_id", sa.String(), nullable=False),
+        sa.Column("ics_uid", sa.Text(), nullable=False),
+        sa.Column("title", sa.Text(), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("start_time", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("end_time", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("attendees", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+        sa.Column("location", sa.Text(), nullable=True),
+        sa.Column("ics_raw_data", sa.Text(), nullable=True),
+        sa.Column("last_synced", sa.DateTime(timezone=True), nullable=False),
+        sa.Column(
+            "is_deleted", sa.Boolean(), server_default=sa.text("false"), nullable=False
+        ),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["room_id"],
+            ["room.id"],
+            name="fk_calendar_event_room_id",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("room_id", "ics_uid", name="uq_room_calendar_event"),
+    )
+    with op.batch_alter_table("calendar_event", schema=None) as batch_op:
+        batch_op.create_index(
+            "idx_calendar_event_deleted",
+            ["is_deleted"],
+            unique=False,
+            postgresql_where=sa.text("NOT is_deleted"),
+        )
+        batch_op.create_index(
+            "idx_calendar_event_room_start", ["room_id", "start_time"], unique=False
+        )
+
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("calendar_event_id", sa.String(), nullable=True))
+        batch_op.add_column(
+            sa.Column(
+                "calendar_metadata",
+                postgresql.JSONB(astext_type=sa.Text()),
+                nullable=True,
+            )
+        )
+        batch_op.create_index(
+            "idx_meeting_calendar_event", ["calendar_event_id"], unique=False
+        )
+        batch_op.create_foreign_key(
+            "fk_meeting_calendar_event_id",
+            "calendar_event",
+            ["calendar_event_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+
+    with op.batch_alter_table("room", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("ics_url", sa.Text(), nullable=True))
+        batch_op.add_column(
+            sa.Column(
+                "ics_fetch_interval", sa.Integer(), server_default="300", nullable=True
+            )
+        )
+        batch_op.add_column(
+            sa.Column(
+                "ics_enabled",
+                sa.Boolean(),
+                server_default=sa.text("false"),
+                nullable=False,
+            )
+        )
+        batch_op.add_column(
+            sa.Column("ics_last_sync", sa.DateTime(timezone=True), nullable=True)
+        )
+        batch_op.add_column(sa.Column("ics_last_etag", sa.Text(), nullable=True))
+        batch_op.create_index("idx_room_ics_enabled", ["ics_enabled"], unique=False)
+
+    # ### 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_index("idx_room_ics_enabled")
+        batch_op.drop_column("ics_last_etag")
+        batch_op.drop_column("ics_last_sync")
+        batch_op.drop_column("ics_enabled")
+        batch_op.drop_column("ics_fetch_interval")
+        batch_op.drop_column("ics_url")
+
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.drop_constraint("fk_meeting_calendar_event_id", type_="foreignkey")
+        batch_op.drop_index("idx_meeting_calendar_event")
+        batch_op.drop_column("calendar_metadata")
+        batch_op.drop_column("calendar_event_id")
+
+    with op.batch_alter_table("calendar_event", schema=None) as batch_op:
+        batch_op.drop_index("idx_calendar_event_room_start")
+        batch_op.drop_index(
+            "idx_calendar_event_deleted", postgresql_where=sa.text("NOT is_deleted")
+        )
+
+    op.drop_table("calendar_event")
+    # ### end Alembic commands ###

server/migrations/versions/dc035ff72fd5_remove_grace_period_fields.py

@@ -0,0 +1,43 @@
+"""remove_grace_period_fields
+
+Revision ID: dc035ff72fd5
+Revises: d8e204bbf615
+Create Date: 2025-09-11 10:36:45.197588
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "dc035ff72fd5"
+down_revision: Union[str, None] = "d8e204bbf615"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Remove grace period columns from meeting table
+    op.drop_column("meeting", "last_participant_left_at")
+    op.drop_column("meeting", "grace_period_minutes")
+
+
+def downgrade() -> None:
+    # Add back grace period columns to meeting table
+    op.add_column(
+        "meeting",
+        sa.Column(
+            "last_participant_left_at", sa.DateTime(timezone=True), nullable=True
+        ),
+    )
+    op.add_column(
+        "meeting",
+        sa.Column(
+            "grace_period_minutes",
+            sa.Integer(),
+            server_default=sa.text("15"),
+            nullable=True,
+        ),
+    )

server/migrations/versions/def1b5867d4c_make_meeting_room_id_nullable_but_keep_.py

@@ -0,0 +1,34 @@
+"""make meeting room_id nullable but keep foreign key
+
+Revision ID: def1b5867d4c
+Revises: 0ce521cda2ee
+Create Date: 2025-09-11 09:42:18.697264
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "def1b5867d4c"
+down_revision: Union[str, None] = "0ce521cda2ee"
+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", schema=None) as batch_op:
+        batch_op.alter_column("room_id", existing_type=sa.VARCHAR(), nullable=True)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("meeting", schema=None) as batch_op:
+        batch_op.alter_column("room_id", existing_type=sa.VARCHAR(), nullable=False)
+
+    # ### end Alembic commands ###

server/migrations/versions/f8294b31f022_add_track_keys.py

@@ -0,0 +1,28 @@
+"""add_track_keys
+
+Revision ID: f8294b31f022
+Revises: 1e49625677e4
+Create Date: 2025-10-27 18:52:17.589167
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "f8294b31f022"
+down_revision: Union[str, None] = "1e49625677e4"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("recording", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("track_keys", sa.JSON(), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("recording", schema=None) as batch_op:
+        batch_op.drop_column("track_keys")

server/pyproject.toml

@@ -12,7 +12,6 @@ dependencies = [
     "requests>=2.31.0",
     "aiortc>=1.5.0",
     "sortedcontainers>=2.4.0",
-    "loguru>=0.7.0",
     "pydantic-settings>=2.0.2",
     "structlog>=23.1.0",
     "uvicorn[standard]>=0.23.1",
@@ -27,7 +26,6 @@ dependencies = [
     "prometheus-fastapi-instrumentator>=6.1.0",
     "sentencepiece>=0.1.99",
     "protobuf>=4.24.3",
-    "profanityfilter>=2.0.6",
     "celery>=5.3.4",
     "redis>=5.0.1",
     "python-jose[cryptography]>=3.3.0",
@@ -40,6 +38,7 @@ dependencies = [
     "llama-index-llms-openai-like>=0.4.0",
     "pytest-env>=1.1.5",
     "webvtt-py>=0.5.0",
+    "icalendar>=6.0.0",
 ]
 
 [dependency-groups]
@@ -113,13 +112,14 @@ source = ["reflector"]
 [tool.pytest_env]
 ENVIRONMENT = "pytest"
 DATABASE_URL = "postgresql://test_user:test_password@localhost:15432/reflector_test"
+AUTH_BACKEND = "jwt"
 
 [tool.pytest.ini_options]
 addopts = "-ra -q --disable-pytest-warnings --cov --cov-report html -v"
 testpaths = ["tests"]
 asyncio_mode = "auto"
 markers = [
-    "gpu_modal: mark test to run only with GPU Modal endpoints (deselect with '-m \"not gpu_modal\"')",
+    "model_api: tests for the unified model-serving HTTP API (backend- and hardware-agnostic)",
 ]
 
 [tool.ruff.lint]
@@ -131,7 +131,7 @@ select = [
 
 [tool.ruff.lint.per-file-ignores]
 "reflector/processors/summary/summary_builder.py" = ["E501"]
-"gpu/**.py" = ["PLC0415"]
+"gpu/modal_deployments/**.py" = ["PLC0415"]
 "reflector/tools/**.py" = ["PLC0415"]
 "migrations/versions/**.py" = ["PLC0415"]
 "tests/**.py" = ["PLC0415"]

server/reflector/app.py

@@ -12,6 +12,7 @@ from reflector.events import subscribers_shutdown, subscribers_startup
 from reflector.logger import logger
 from reflector.metrics import metrics_init
 from reflector.settings import settings
+from reflector.views.daily import router as daily_router
 from reflector.views.meetings import router as meetings_router
 from reflector.views.rooms import router as rooms_router
 from reflector.views.rtc_offer import router as rtc_offer_router
@@ -26,6 +27,8 @@ from reflector.views.transcripts_upload import router as transcripts_upload_rout
 from reflector.views.transcripts_webrtc import router as transcripts_webrtc_router
 from reflector.views.transcripts_websocket import router as transcripts_websocket_router
 from reflector.views.user import router as user_router
+from reflector.views.user_api_keys import router as user_api_keys_router
+from reflector.views.user_websocket import router as user_ws_router
 from reflector.views.whereby import router as whereby_router
 from reflector.views.zulip import router as zulip_router
 
@@ -65,6 +68,12 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy"}
+
+
 # metrics
 instrumentator = Instrumentator(
     excluded_handlers=["/docs", "/metrics"],
@@ -84,8 +93,11 @@ app.include_router(transcripts_websocket_router, prefix="/v1")
 app.include_router(transcripts_webrtc_router, prefix="/v1")
 app.include_router(transcripts_process_router, prefix="/v1")
 app.include_router(user_router, prefix="/v1")
+app.include_router(user_api_keys_router, prefix="/v1")
+app.include_router(user_ws_router, prefix="/v1")
 app.include_router(zulip_router, prefix="/v1")
 app.include_router(whereby_router, prefix="/v1")
+app.include_router(daily_router, prefix="/v1/daily")
 add_pagination(app)
 
 # prepare celery

server/reflector/auth/auth_jwt.py

@@ -1,14 +1,16 @@
-from typing import Annotated, Optional
+from typing import Annotated, List, Optional
 
 from fastapi import Depends, HTTPException
-from fastapi.security import OAuth2PasswordBearer
+from fastapi.security import APIKeyHeader, OAuth2PasswordBearer
 from jose import JWTError, jwt
 from pydantic import BaseModel
 
+from reflector.db.user_api_keys import user_api_keys_controller
 from reflector.logger import logger
 from reflector.settings import settings
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
+api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
 
 jwt_public_key = open(f"reflector/auth/jwt/keys/{settings.AUTH_JWT_PUBLIC_KEY}").read()
 jwt_algorithm = settings.AUTH_JWT_ALGORITHM
@@ -26,7 +28,7 @@ class JWTException(Exception):
 
 class UserInfo(BaseModel):
     sub: str
-    email: str
+    email: Optional[str] = None
 
     def __getitem__(self, key):
         return getattr(self, key)
@@ -58,33 +60,53 @@ def authenticated(token: Annotated[str, Depends(oauth2_scheme)]):
     return None
 
 
-def current_user(
-    token: Annotated[Optional[str], Depends(oauth2_scheme)],
-    jwtauth: JWTAuth = Depends(),
-):
-    if token is None:
-        raise HTTPException(status_code=401, detail="Not authenticated")
-    try:
-        payload = jwtauth.verify_token(token)
-        sub = payload["sub"]
-        return UserInfo(sub=sub)
-    except JWTError as e:
-        logger.error(f"JWT error: {e}")
-        raise HTTPException(status_code=401, detail="Invalid authentication")
+async def _authenticate_user(
+    jwt_token: Optional[str],
+    api_key: Optional[str],
+    jwtauth: JWTAuth,
+) -> UserInfo | None:
+    user_infos: List[UserInfo] = []
+    if api_key:
+        user_api_key = await user_api_keys_controller.verify_key(api_key)
+        if user_api_key:
+            user_infos.append(UserInfo(sub=user_api_key.user_id, email=None))
 
+    if jwt_token:
+        try:
+            payload = jwtauth.verify_token(jwt_token)
+            sub = payload["sub"]
+            email = payload["email"]
+            user_infos.append(UserInfo(sub=sub, email=email))
+        except JWTError as e:
+            logger.error(f"JWT error: {e}")
+            raise HTTPException(status_code=401, detail="Invalid authentication")
 
-def current_user_optional(
-    token: Annotated[Optional[str], Depends(oauth2_scheme)],
-    jwtauth: JWTAuth = Depends(),
-):
-    # we accept no token, but if one is provided, it must be a valid one.
-    if token is None:
+    if len(user_infos) == 0:
         return None
-    try:
-        payload = jwtauth.verify_token(token)
-        sub = payload["sub"]
-        email = payload["email"]
-        return UserInfo(sub=sub, email=email)
-    except JWTError as e:
-        logger.error(f"JWT error: {e}")
-        raise HTTPException(status_code=401, detail="Invalid authentication")
+
+    if len(set([x.sub for x in user_infos])) > 1:
+        raise JWTException(
+            status_code=401,
+            detail="Invalid authentication: more than one user provided",
+        )
+
+    return user_infos[0]
+
+
+async def current_user(
+    jwt_token: Annotated[Optional[str], Depends(oauth2_scheme)],
+    api_key: Annotated[Optional[str], Depends(api_key_header)],
+    jwtauth: JWTAuth = Depends(),
+):
+    user = await _authenticate_user(jwt_token, api_key, jwtauth)
+    if user is None:
+        raise HTTPException(status_code=401, detail="Not authenticated")
+    return user
+
+
+async def current_user_optional(
+    jwt_token: Annotated[Optional[str], Depends(oauth2_scheme)],
+    api_key: Annotated[Optional[str], Depends(api_key_header)],
+    jwtauth: JWTAuth = Depends(),
+):
+    return await _authenticate_user(jwt_token, api_key, jwtauth)

server/reflector/db/__init__.py

@@ -24,10 +24,12 @@ def get_database() -> databases.Database:
 
 
 # import models
+import reflector.db.calendar_events  # noqa
 import reflector.db.meetings  # noqa
 import reflector.db.recordings  # noqa
 import reflector.db.rooms  # noqa
 import reflector.db.transcripts  # noqa
+import reflector.db.user_api_keys  # noqa
 
 kwargs = {}
 if "postgres" not in settings.DATABASE_URL:

server/reflector/db/calendar_events.py

@@ -0,0 +1,187 @@
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from pydantic import BaseModel, Field
+from sqlalchemy.dialects.postgresql import JSONB
+
+from reflector.db import get_database, metadata
+from reflector.utils import generate_uuid4
+
+calendar_events = sa.Table(
+    "calendar_event",
+    metadata,
+    sa.Column("id", sa.String, primary_key=True),
+    sa.Column(
+        "room_id",
+        sa.String,
+        sa.ForeignKey("room.id", ondelete="CASCADE", name="fk_calendar_event_room_id"),
+        nullable=False,
+    ),
+    sa.Column("ics_uid", sa.Text, nullable=False),
+    sa.Column("title", sa.Text),
+    sa.Column("description", sa.Text),
+    sa.Column("start_time", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("end_time", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("attendees", JSONB),
+    sa.Column("location", sa.Text),
+    sa.Column("ics_raw_data", sa.Text),
+    sa.Column("last_synced", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("is_deleted", sa.Boolean, nullable=False, server_default=sa.false()),
+    sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    sa.UniqueConstraint("room_id", "ics_uid", name="uq_room_calendar_event"),
+    sa.Index("idx_calendar_event_room_start", "room_id", "start_time"),
+    sa.Index(
+        "idx_calendar_event_deleted",
+        "is_deleted",
+        postgresql_where=sa.text("NOT is_deleted"),
+    ),
+)
+
+
+class CalendarEvent(BaseModel):
+    id: str = Field(default_factory=generate_uuid4)
+    room_id: str
+    ics_uid: str
+    title: str | None = None
+    description: str | None = None
+    start_time: datetime
+    end_time: datetime
+    attendees: list[dict[str, Any]] | None = None
+    location: str | None = None
+    ics_raw_data: str | None = None
+    last_synced: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    is_deleted: bool = False
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+
+
+class CalendarEventController:
+    async def get_by_room(
+        self,
+        room_id: str,
+        include_deleted: bool = False,
+        start_after: datetime | None = None,
+        end_before: datetime | None = None,
+    ) -> list[CalendarEvent]:
+        query = calendar_events.select().where(calendar_events.c.room_id == room_id)
+
+        if not include_deleted:
+            query = query.where(calendar_events.c.is_deleted == False)
+
+        if start_after:
+            query = query.where(calendar_events.c.start_time >= start_after)
+
+        if end_before:
+            query = query.where(calendar_events.c.end_time <= end_before)
+
+        query = query.order_by(calendar_events.c.start_time.asc())
+
+        results = await get_database().fetch_all(query)
+        return [CalendarEvent(**result) for result in results]
+
+    async def get_upcoming(
+        self, room_id: str, minutes_ahead: int = 120
+    ) -> list[CalendarEvent]:
+        """Get upcoming events for a room within the specified minutes, including currently happening events."""
+        now = datetime.now(timezone.utc)
+        future_time = now + timedelta(minutes=minutes_ahead)
+
+        query = (
+            calendar_events.select()
+            .where(
+                sa.and_(
+                    calendar_events.c.room_id == room_id,
+                    calendar_events.c.is_deleted == False,
+                    calendar_events.c.start_time <= future_time,
+                    calendar_events.c.end_time >= now,
+                )
+            )
+            .order_by(calendar_events.c.start_time.asc())
+        )
+
+        results = await get_database().fetch_all(query)
+        return [CalendarEvent(**result) for result in results]
+
+    async def get_by_id(self, event_id: str) -> CalendarEvent | None:
+        query = calendar_events.select().where(calendar_events.c.id == event_id)
+        result = await get_database().fetch_one(query)
+        return CalendarEvent(**result) if result else None
+
+    async def get_by_ics_uid(self, room_id: str, ics_uid: str) -> CalendarEvent | None:
+        query = calendar_events.select().where(
+            sa.and_(
+                calendar_events.c.room_id == room_id,
+                calendar_events.c.ics_uid == ics_uid,
+            )
+        )
+        result = await get_database().fetch_one(query)
+        return CalendarEvent(**result) if result else None
+
+    async def upsert(self, event: CalendarEvent) -> CalendarEvent:
+        existing = await self.get_by_ics_uid(event.room_id, event.ics_uid)
+
+        if existing:
+            event.id = existing.id
+            event.created_at = existing.created_at
+            event.updated_at = datetime.now(timezone.utc)
+
+            query = (
+                calendar_events.update()
+                .where(calendar_events.c.id == existing.id)
+                .values(**event.model_dump())
+            )
+        else:
+            query = calendar_events.insert().values(**event.model_dump())
+
+        await get_database().execute(query)
+        return event
+
+    async def soft_delete_missing(
+        self, room_id: str, current_ics_uids: list[str]
+    ) -> int:
+        """Soft delete future events that are no longer in the calendar."""
+        now = datetime.now(timezone.utc)
+
+        select_query = calendar_events.select().where(
+            sa.and_(
+                calendar_events.c.room_id == room_id,
+                calendar_events.c.start_time > now,
+                calendar_events.c.is_deleted == False,
+                calendar_events.c.ics_uid.notin_(current_ics_uids)
+                if current_ics_uids
+                else True,
+            )
+        )
+
+        to_delete = await get_database().fetch_all(select_query)
+        delete_count = len(to_delete)
+
+        if delete_count > 0:
+            update_query = (
+                calendar_events.update()
+                .where(
+                    sa.and_(
+                        calendar_events.c.room_id == room_id,
+                        calendar_events.c.start_time > now,
+                        calendar_events.c.is_deleted == False,
+                        calendar_events.c.ics_uid.notin_(current_ics_uids)
+                        if current_ics_uids
+                        else True,
+                    )
+                )
+                .values(is_deleted=True, updated_at=now)
+            )
+
+            await get_database().execute(update_query)
+
+        return delete_count
+
+    async def delete_by_room(self, room_id: str) -> int:
+        query = calendar_events.delete().where(calendar_events.c.room_id == room_id)
+        result = await get_database().execute(query)
+        return result.rowcount
+
+
+calendar_events_controller = CalendarEventController()

server/reflector/db/meetings.py

@@ -1,13 +1,16 @@
 from datetime import datetime
-from typing import Literal
+from typing import Any, Literal
 
 import sqlalchemy as sa
-from fastapi import HTTPException
 from pydantic import BaseModel, Field
+from sqlalchemy.dialects.postgresql import JSONB
 
 from reflector.db import get_database, metadata
 from reflector.db.rooms import Room
+from reflector.schemas.platform import WHEREBY_PLATFORM, Platform
 from reflector.utils import generate_uuid4
+from reflector.utils.string import assert_equal
+from reflector.video_platforms.factory import get_platform
 
 meetings = sa.Table(
     "meeting",
@@ -18,8 +21,12 @@ meetings = sa.Table(
     sa.Column("host_room_url", sa.String),
     sa.Column("start_date", sa.DateTime(timezone=True)),
     sa.Column("end_date", sa.DateTime(timezone=True)),
-    sa.Column("user_id", sa.String),
-    sa.Column("room_id", sa.String),
+    sa.Column(
+        "room_id",
+        sa.String,
+        sa.ForeignKey("room.id", ondelete="CASCADE"),
+        nullable=True,
+    ),
     sa.Column("is_locked", sa.Boolean, nullable=False, server_default=sa.false()),
     sa.Column("room_mode", sa.String, nullable=False, server_default="normal"),
     sa.Column("recording_type", sa.String, nullable=False, server_default="cloud"),
@@ -41,13 +48,24 @@ meetings = sa.Table(
         nullable=False,
         server_default=sa.true(),
     ),
-    sa.Index("idx_meeting_room_id", "room_id"),
-    sa.Index(
-        "idx_one_active_meeting_per_room",
-        "room_id",
-        unique=True,
-        postgresql_where=sa.text("is_active = true"),
+    sa.Column(
+        "calendar_event_id",
+        sa.String,
+        sa.ForeignKey(
+            "calendar_event.id",
+            ondelete="SET NULL",
+            name="fk_meeting_calendar_event_id",
+        ),
     ),
+    sa.Column("calendar_metadata", JSONB),
+    sa.Column(
+        "platform",
+        sa.String,
+        nullable=False,
+        server_default=assert_equal(WHEREBY_PLATFORM, "whereby"),
+    ),
+    sa.Index("idx_meeting_room_id", "room_id"),
+    sa.Index("idx_meeting_calendar_event", "calendar_event_id"),
 )
 
 meeting_consent = sa.Table(
@@ -81,15 +99,18 @@ class Meeting(BaseModel):
     host_room_url: str
     start_date: datetime
     end_date: datetime
-    user_id: str | None = None
-    room_id: str | None = None
+    room_id: str | None
     is_locked: bool = False
     room_mode: Literal["normal", "group"] = "normal"
     recording_type: Literal["none", "local", "cloud"] = "cloud"
-    recording_trigger: Literal[
+    recording_trigger: Literal[  # whereby-specific
         "none", "prompt", "automatic", "automatic-2nd-participant"
     ] = "automatic-2nd-participant"
     num_clients: int = 0
+    is_active: bool = True
+    calendar_event_id: str | None = None
+    calendar_metadata: dict[str, Any] | None = None
+    platform: Platform = WHEREBY_PLATFORM
 
 
 class MeetingController:
@@ -101,12 +122,10 @@ class MeetingController:
         host_room_url: str,
         start_date: datetime,
         end_date: datetime,
-        user_id: str,
         room: Room,
+        calendar_event_id: str | None = None,
+        calendar_metadata: dict[str, Any] | None = None,
     ):
-        """
-        Create a new meeting
-        """
         meeting = Meeting(
             id=id,
             room_name=room_name,
@@ -114,41 +133,46 @@ class MeetingController:
             host_room_url=host_room_url,
             start_date=start_date,
             end_date=end_date,
-            user_id=user_id,
             room_id=room.id,
             is_locked=room.is_locked,
             room_mode=room.room_mode,
             recording_type=room.recording_type,
             recording_trigger=room.recording_trigger,
+            calendar_event_id=calendar_event_id,
+            calendar_metadata=calendar_metadata,
+            platform=get_platform(room.platform),
         )
         query = meetings.insert().values(**meeting.model_dump())
         await get_database().execute(query)
         return meeting
 
     async def get_all_active(self) -> list[Meeting]:
-        """
-        Get active meetings.
-        """
         query = meetings.select().where(meetings.c.is_active)
-        return await get_database().fetch_all(query)
+        results = await get_database().fetch_all(query)
+        return [Meeting(**result) for result in results]
 
     async def get_by_room_name(
         self,
         room_name: str,
-    ) -> Meeting:
+    ) -> Meeting | None:
         """
         Get a meeting by room name.
+        For backward compatibility, returns the most recent meeting.
         """
-        query = meetings.select().where(meetings.c.room_name == room_name)
+        query = (
+            meetings.select()
+            .where(meetings.c.room_name == room_name)
+            .order_by(meetings.c.end_date.desc())
+        )
         result = await get_database().fetch_one(query)
         if not result:
             return None
-
         return Meeting(**result)
 
-    async def get_active(self, room: Room, current_time: datetime) -> Meeting:
+    async def get_active(self, room: Room, current_time: datetime) -> Meeting | None:
         """
         Get latest active meeting for a room.
+        For backward compatibility, returns the most recent active meeting.
         """
         end_date = getattr(meetings.c, "end_date")
         query = (
@@ -165,40 +189,97 @@ class MeetingController:
         result = await get_database().fetch_one(query)
         if not result:
             return None
-
         return Meeting(**result)
 
-    async def get_by_id(self, meeting_id: str, **kwargs) -> Meeting | None:
+    async def get_all_active_for_room(
+        self, room: Room, current_time: datetime
+    ) -> list[Meeting]:
+        end_date = getattr(meetings.c, "end_date")
+        query = (
+            meetings.select()
+            .where(
+                sa.and_(
+                    meetings.c.room_id == room.id,
+                    meetings.c.end_date > current_time,
+                    meetings.c.is_active,
+                )
+            )
+            .order_by(end_date.desc())
+        )
+        results = await get_database().fetch_all(query)
+        return [Meeting(**result) for result in results]
+
+    async def get_active_by_calendar_event(
+        self, room: Room, calendar_event_id: str, current_time: datetime
+    ) -> Meeting | None:
         """
-        Get a meeting by id
+        Get active meeting for a specific calendar event.
         """
-        query = meetings.select().where(meetings.c.id == meeting_id)
+        query = meetings.select().where(
+            sa.and_(
+                meetings.c.room_id == room.id,
+                meetings.c.calendar_event_id == calendar_event_id,
+                meetings.c.end_date > current_time,
+                meetings.c.is_active,
+            )
+        )
         result = await get_database().fetch_one(query)
         if not result:
             return None
         return Meeting(**result)
 
-    async def get_by_id_for_http(self, meeting_id: str, user_id: str | None) -> Meeting:
-        """
-        Get a meeting by ID for HTTP request.
-
-        If not found, it will raise a 404 error.
-        """
+    async def get_by_id(
+        self, meeting_id: str, room: Room | None = None
+    ) -> Meeting | None:
         query = meetings.select().where(meetings.c.id == meeting_id)
+
+        if room:
+            query = query.where(meetings.c.room_id == room.id)
+
         result = await get_database().fetch_one(query)
         if not result:
-            raise HTTPException(status_code=404, detail="Meeting not found")
+            return None
+        return Meeting(**result)
 
-        meeting = Meeting(**result)
-        if result["user_id"] != user_id:
-            meeting.host_room_url = ""
-
-        return meeting
+    async def get_by_calendar_event(
+        self, calendar_event_id: str, room: Room
+    ) -> Meeting | None:
+        query = meetings.select().where(
+            meetings.c.calendar_event_id == calendar_event_id
+        )
+        if room:
+            query = query.where(meetings.c.room_id == room.id)
+        result = await get_database().fetch_one(query)
+        if not result:
+            return None
+        return Meeting(**result)
 
     async def update_meeting(self, meeting_id: str, **kwargs):
         query = meetings.update().where(meetings.c.id == meeting_id).values(**kwargs)
         await get_database().execute(query)
 
+    async def increment_num_clients(self, meeting_id: str) -> None:
+        """Atomically increment participant count."""
+        query = (
+            meetings.update()
+            .where(meetings.c.id == meeting_id)
+            .values(num_clients=meetings.c.num_clients + 1)
+        )
+        await get_database().execute(query)
+
+    async def decrement_num_clients(self, meeting_id: str) -> None:
+        """Atomically decrement participant count (min 0)."""
+        query = (
+            meetings.update()
+            .where(meetings.c.id == meeting_id)
+            .values(
+                num_clients=sa.case(
+                    (meetings.c.num_clients > 0, meetings.c.num_clients - 1), else_=0
+                )
+            )
+        )
+        await get_database().execute(query)
+
 
 class MeetingConsentController:
     async def get_by_meeting_id(self, meeting_id: str) -> list[MeetingConsent]:
@@ -219,10 +300,9 @@ class MeetingConsentController:
         result = await get_database().fetch_one(query)
         if result is None:
             return None
-        return MeetingConsent(**result) if result else None
+        return MeetingConsent(**result)
 
     async def upsert(self, consent: MeetingConsent) -> MeetingConsent:
-        """Create new consent or update existing one for authenticated users"""
         if consent.user_id:
             # For authenticated users, check if consent already exists
             # not transactional but we're ok with that; the consents ain't deleted anyways

server/reflector/db/recordings.py

@@ -21,6 +21,7 @@ recordings = sa.Table(
         server_default="pending",
     ),
     sa.Column("meeting_id", sa.String),
+    sa.Column("track_keys", sa.JSON, nullable=True),
     sa.Index("idx_recording_meeting_id", "meeting_id"),
 )
 
@@ -28,10 +29,13 @@ recordings = sa.Table(
 class Recording(BaseModel):
     id: str = Field(default_factory=generate_uuid4)
     bucket_name: str
+    # for single-track
     object_key: str
     recorded_at: datetime
     status: Literal["pending", "processing", "completed", "failed"] = "pending"
     meeting_id: str | None = None
+    # for multitrack reprocessing
+    track_keys: list[str] | None = None
 
 
 class RecordingController:

server/reflector/db/rooms.py

@@ -9,6 +9,7 @@ from pydantic import BaseModel, Field
 from sqlalchemy.sql import false, or_
 
 from reflector.db import get_database, metadata
+from reflector.schemas.platform import Platform
 from reflector.utils import generate_uuid4
 
 rooms = sqlalchemy.Table(
@@ -43,7 +44,21 @@ rooms = sqlalchemy.Table(
     ),
     sqlalchemy.Column("webhook_url", sqlalchemy.String, nullable=True),
     sqlalchemy.Column("webhook_secret", sqlalchemy.String, nullable=True),
+    sqlalchemy.Column("ics_url", sqlalchemy.Text),
+    sqlalchemy.Column("ics_fetch_interval", sqlalchemy.Integer, server_default="300"),
+    sqlalchemy.Column(
+        "ics_enabled", sqlalchemy.Boolean, nullable=False, server_default=false()
+    ),
+    sqlalchemy.Column("ics_last_sync", sqlalchemy.DateTime(timezone=True)),
+    sqlalchemy.Column("ics_last_etag", sqlalchemy.Text),
+    sqlalchemy.Column(
+        "platform",
+        sqlalchemy.String,
+        nullable=True,
+        server_default=None,
+    ),
     sqlalchemy.Index("idx_room_is_shared", "is_shared"),
+    sqlalchemy.Index("idx_room_ics_enabled", "ics_enabled"),
 )
 
 
@@ -58,12 +73,18 @@ class Room(BaseModel):
     is_locked: bool = False
     room_mode: Literal["normal", "group"] = "normal"
     recording_type: Literal["none", "local", "cloud"] = "cloud"
-    recording_trigger: Literal[
+    recording_trigger: Literal[  # whereby-specific
         "none", "prompt", "automatic", "automatic-2nd-participant"
     ] = "automatic-2nd-participant"
     is_shared: bool = False
     webhook_url: str | None = None
     webhook_secret: str | None = None
+    ics_url: str | None = None
+    ics_fetch_interval: int = 300
+    ics_enabled: bool = False
+    ics_last_sync: datetime | None = None
+    ics_last_etag: str | None = None
+    platform: Platform | None = None
 
 
 class RoomController:
@@ -114,6 +135,10 @@ class RoomController:
         is_shared: bool,
         webhook_url: str = "",
         webhook_secret: str = "",
+        ics_url: str | None = None,
+        ics_fetch_interval: int = 300,
+        ics_enabled: bool = False,
+        platform: Platform | None = None,
     ):
         """
         Add a new room
@@ -134,6 +159,10 @@ class RoomController:
             is_shared=is_shared,
             webhook_url=webhook_url,
             webhook_secret=webhook_secret,
+            ics_url=ics_url,
+            ics_fetch_interval=ics_fetch_interval,
+            ics_enabled=ics_enabled,
+            platform=platform,
         )
         query = rooms.insert().values(**room.model_dump())
         try:
@@ -198,6 +227,13 @@ class RoomController:
 
         return room
 
+    async def get_ics_enabled(self) -> list[Room]:
+        query = rooms.select().where(
+            rooms.c.ics_enabled == True, rooms.c.ics_url != None
+        )
+        results = await get_database().fetch_all(query)
+        return [Room(**result) for result in results]
+
     async def remove_by_id(
         self,
         room_id: str,

server/reflector/db/search.py

@@ -23,7 +23,7 @@ from pydantic import (
 
 from reflector.db import get_database
 from reflector.db.rooms import rooms
-from reflector.db.transcripts import SourceKind, transcripts
+from reflector.db.transcripts import SourceKind, TranscriptStatus, transcripts
 from reflector.db.utils import is_postgresql
 from reflector.logger import logger
 from reflector.utils.string import NonEmptyString, try_parse_non_empty_string
@@ -135,6 +135,8 @@ class SearchParameters(BaseModel):
     user_id: str | None = None
     room_id: str | None = None
     source_kind: SourceKind | None = None
+    from_datetime: datetime | None = None
+    to_datetime: datetime | None = None
 
 
 class SearchResultDB(BaseModel):
@@ -161,7 +163,7 @@ class SearchResult(BaseModel):
     room_name: str | None = None
     source_kind: SourceKind
     created_at: datetime
-    status: str = Field(..., min_length=1)
+    status: TranscriptStatus = Field(..., min_length=1)
     rank: float = Field(..., ge=0, le=1)
     duration: NonNegativeFloat | None = Field(..., description="Duration in seconds")
     search_snippets: list[str] = Field(
@@ -402,6 +404,14 @@ class SearchController:
             base_query = base_query.where(
                 transcripts.c.source_kind == params.source_kind
             )
+        if params.from_datetime:
+            base_query = base_query.where(
+                transcripts.c.created_at >= params.from_datetime
+            )
+        if params.to_datetime:
+            base_query = base_query.where(
+                transcripts.c.created_at <= params.to_datetime
+            )
 
         if params.query_text is not None:
             order_by = sqlalchemy.desc(sqlalchemy.text("rank"))

server/reflector/db/transcripts.py

@@ -21,7 +21,7 @@ from reflector.db.utils import is_postgresql
 from reflector.logger import logger
 from reflector.processors.types import Word as ProcessorWord
 from reflector.settings import settings
-from reflector.storage import get_recordings_storage, get_transcripts_storage
+from reflector.storage import get_transcripts_storage
 from reflector.utils import generate_uuid4
 from reflector.utils.webvtt import topics_to_webvtt
 
@@ -186,6 +186,7 @@ class TranscriptParticipant(BaseModel):
     id: str = Field(default_factory=generate_uuid4)
     speaker: int | None
     name: str
+    user_id: str | None = None
 
 
 class Transcript(BaseModel):
@@ -623,7 +624,9 @@ class TranscriptController:
                 )
                 if recording:
                     try:
-                        await get_recordings_storage().delete_file(recording.object_key)
+                        await get_transcripts_storage().delete_file(
+                            recording.object_key, bucket=recording.bucket_name
+                        )
                     except Exception as e:
                         logger.warning(
                             "Failed to delete recording object from S3",
@@ -647,6 +650,19 @@ class TranscriptController:
         query = transcripts.delete().where(transcripts.c.recording_id == recording_id)
         await get_database().execute(query)
 
+    @staticmethod
+    def user_can_mutate(transcript: Transcript, user_id: str | None) -> bool:
+        """
+        Returns True if the given user is allowed to modify the transcript.
+
+        Policy:
+        - Anonymous transcripts (user_id is None) cannot be modified via API
+        - Only the owner (matching user_id) can modify their transcript
+        """
+        if transcript.user_id is None:
+            return False
+        return user_id and transcript.user_id == user_id
+
     @asynccontextmanager
     async def transaction(self):
         """
@@ -712,11 +728,13 @@ class TranscriptController:
         """
         Download audio from storage
         """
-        transcript.audio_mp3_filename.write_bytes(
-            await get_transcripts_storage().get_file(
-                transcript.storage_audio_path,
-            )
-        )
+        storage = get_transcripts_storage()
+        try:
+            with open(transcript.audio_mp3_filename, "wb") as f:
+                await storage.stream_to_fileobj(transcript.storage_audio_path, f)
+        except Exception:
+            transcript.audio_mp3_filename.unlink(missing_ok=True)
+            raise
 
     async def upsert_participant(
         self,

server/reflector/db/user_api_keys.py

@@ -0,0 +1,91 @@
+import hmac
+import secrets
+from datetime import datetime, timezone
+from hashlib import sha256
+
+import sqlalchemy
+from pydantic import BaseModel, Field
+
+from reflector.db import get_database, metadata
+from reflector.settings import settings
+from reflector.utils import generate_uuid4
+from reflector.utils.string import NonEmptyString
+
+user_api_keys = sqlalchemy.Table(
+    "user_api_key",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.String, primary_key=True),
+    sqlalchemy.Column("user_id", sqlalchemy.String, nullable=False),
+    sqlalchemy.Column("key_hash", sqlalchemy.String, nullable=False),
+    sqlalchemy.Column("name", sqlalchemy.String, nullable=True),
+    sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True), nullable=False),
+    sqlalchemy.Index("idx_user_api_key_hash", "key_hash", unique=True),
+    sqlalchemy.Index("idx_user_api_key_user_id", "user_id"),
+)
+
+
+class UserApiKey(BaseModel):
+    id: NonEmptyString = Field(default_factory=generate_uuid4)
+    user_id: NonEmptyString
+    key_hash: NonEmptyString
+    name: NonEmptyString | None = None
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+
+
+class UserApiKeyController:
+    @staticmethod
+    def generate_key() -> NonEmptyString:
+        return secrets.token_urlsafe(48)
+
+    @staticmethod
+    def hash_key(key: NonEmptyString) -> str:
+        return hmac.new(
+            settings.SECRET_KEY.encode(), key.encode(), digestmod=sha256
+        ).hexdigest()
+
+    @classmethod
+    async def create_key(
+        cls,
+        user_id: NonEmptyString,
+        name: NonEmptyString | None = None,
+    ) -> tuple[UserApiKey, NonEmptyString]:
+        plaintext = cls.generate_key()
+        api_key = UserApiKey(
+            user_id=user_id,
+            key_hash=cls.hash_key(plaintext),
+            name=name,
+        )
+        query = user_api_keys.insert().values(**api_key.model_dump())
+        await get_database().execute(query)
+        return api_key, plaintext
+
+    @classmethod
+    async def verify_key(cls, plaintext_key: NonEmptyString) -> UserApiKey | None:
+        key_hash = cls.hash_key(plaintext_key)
+        query = user_api_keys.select().where(
+            user_api_keys.c.key_hash == key_hash,
+        )
+        result = await get_database().fetch_one(query)
+        return UserApiKey(**result) if result else None
+
+    @staticmethod
+    async def list_by_user_id(user_id: NonEmptyString) -> list[UserApiKey]:
+        query = (
+            user_api_keys.select()
+            .where(user_api_keys.c.user_id == user_id)
+            .order_by(user_api_keys.c.created_at.desc())
+        )
+        results = await get_database().fetch_all(query)
+        return [UserApiKey(**r) for r in results]
+
+    @staticmethod
+    async def delete_key(key_id: NonEmptyString, user_id: NonEmptyString) -> bool:
+        query = user_api_keys.delete().where(
+            (user_api_keys.c.id == key_id) & (user_api_keys.c.user_id == user_id)
+        )
+        result = await get_database().execute(query)
+        # asyncpg returns None for DELETE, consider it success if no exception
+        return result is None or result > 0
+
+
+user_api_keys_controller = UserApiKeyController()

server/reflector/pipelines/__init__.py

@@ -0,0 +1 @@
+"""Pipeline modules for audio processing."""

server/reflector/pipelines/main_file_pipeline.py

@@ -12,7 +12,7 @@ from pathlib import Path
 
 import av
 import structlog
-from celery import shared_task
+from celery import chain, shared_task
 
 from reflector.asynctask import asynctask
 from reflector.db.rooms import rooms_controller
@@ -23,21 +23,18 @@ from reflector.db.transcripts import (
     transcripts_controller,
 )
 from reflector.logger import logger
+from reflector.pipelines import topic_processing
 from reflector.pipelines.main_live_pipeline import (
     PipelineMainBase,
     broadcast_to_sockets,
+    task_cleanup_consent,
+    task_pipeline_post_to_zulip,
 )
-from reflector.processors import (
-    AudioFileWriterProcessor,
-    TranscriptFinalSummaryProcessor,
-    TranscriptFinalTitleProcessor,
-    TranscriptTopicDetectorProcessor,
-)
+from reflector.pipelines.transcription_helpers import transcribe_file_with_processor
+from reflector.processors import AudioFileWriterProcessor
 from reflector.processors.audio_waveform_processor import AudioWaveformProcessor
 from reflector.processors.file_diarization import FileDiarizationInput
 from reflector.processors.file_diarization_auto import FileDiarizationAutoProcessor
-from reflector.processors.file_transcript import FileTranscriptInput
-from reflector.processors.file_transcript_auto import FileTranscriptAutoProcessor
 from reflector.processors.transcript_diarization_assembler import (
     TranscriptDiarizationAssemblerInput,
     TranscriptDiarizationAssemblerProcessor,
@@ -54,19 +51,6 @@ from reflector.storage import get_transcripts_storage
 from reflector.worker.webhook import send_transcript_webhook
 
 
-class EmptyPipeline:
-    """Empty pipeline for processors that need a pipeline reference"""
-
-    def __init__(self, logger: structlog.BoundLogger):
-        self.logger = logger
-
-    def get_pref(self, k, d=None):
-        return d
-
-    async def emit(self, event):
-        pass
-
-
 class PipelineMainFile(PipelineMainBase):
     """
     Optimized file processing pipeline.
@@ -79,7 +63,7 @@ class PipelineMainFile(PipelineMainBase):
     def __init__(self, transcript_id: str):
         super().__init__(transcript_id=transcript_id)
         self.logger = logger.bind(transcript_id=self.transcript_id)
-        self.empty_pipeline = EmptyPipeline(logger=self.logger)
+        self.empty_pipeline = topic_processing.EmptyPipeline(logger=self.logger)
 
     def _handle_gather_exceptions(self, results: list, operation: str) -> None:
         """Handle exceptions from asyncio.gather with return_exceptions=True"""
@@ -129,7 +113,7 @@ class PipelineMainFile(PipelineMainBase):
 
         self.logger.info("File pipeline complete")
 
-        await transcripts_controller.set_status(transcript.id, "ended")
+        await self.set_status(transcript.id, "ended")
 
     async def extract_and_write_audio(
         self, file_path: Path, transcript: Transcript
@@ -260,24 +244,7 @@ class PipelineMainFile(PipelineMainBase):
 
     async def transcribe_file(self, audio_url: str, language: str) -> TranscriptType:
         """Transcribe complete file"""
-        processor = FileTranscriptAutoProcessor()
-        input_data = FileTranscriptInput(audio_url=audio_url, language=language)
-
-        # Store result for retrieval
-        result: TranscriptType | None = None
-
-        async def capture_result(transcript):
-            nonlocal result
-            result = transcript
-
-        processor.on(capture_result)
-        await processor.push(input_data)
-        await processor.flush()
-
-        if not result:
-            raise ValueError("No transcript captured")
-
-        return result
+        return await transcribe_file_with_processor(audio_url, language)
 
     async def diarize_file(self, audio_url: str) -> list[DiarizationSegment] | None:
         """Get diarization for file"""
@@ -320,63 +287,53 @@ class PipelineMainFile(PipelineMainBase):
     async def detect_topics(
         self, transcript: TranscriptType, target_language: str
     ) -> list[TitleSummary]:
-        """Detect topics from complete transcript"""
-        chunk_size = 300
-        topics: list[TitleSummary] = []
-
-        async def on_topic(topic: TitleSummary):
-            topics.append(topic)
-            return await self.on_topic(topic)
-
-        topic_detector = TranscriptTopicDetectorProcessor(callback=on_topic)
-        topic_detector.set_pipeline(self.empty_pipeline)
-
-        for i in range(0, len(transcript.words), chunk_size):
-            chunk_words = transcript.words[i : i + chunk_size]
-            if not chunk_words:
-                continue
-
-            chunk_transcript = TranscriptType(
-                words=chunk_words, translation=transcript.translation
-            )
-
-            await topic_detector.push(chunk_transcript)
-
-        await topic_detector.flush()
-        return topics
+        return await topic_processing.detect_topics(
+            transcript,
+            target_language,
+            on_topic_callback=self.on_topic,
+            empty_pipeline=self.empty_pipeline,
+        )
 
     async def generate_title(self, topics: list[TitleSummary]):
-        """Generate title from topics"""
-        if not topics:
-            self.logger.warning("No topics for title generation")
-            return
-
-        processor = TranscriptFinalTitleProcessor(callback=self.on_title)
-        processor.set_pipeline(self.empty_pipeline)
-
-        for topic in topics:
-            await processor.push(topic)
-
-        await processor.flush()
+        return await topic_processing.generate_title(
+            topics,
+            on_title_callback=self.on_title,
+            empty_pipeline=self.empty_pipeline,
+            logger=self.logger,
+        )
 
     async def generate_summaries(self, topics: list[TitleSummary]):
-        """Generate long and short summaries from topics"""
-        if not topics:
-            self.logger.warning("No topics for summary generation")
-            return
-
         transcript = await self.get_transcript()
-        processor = TranscriptFinalSummaryProcessor(
-            transcript=transcript,
-            callback=self.on_long_summary,
-            on_short_summary=self.on_short_summary,
+        return await topic_processing.generate_summaries(
+            topics,
+            transcript,
+            on_long_summary_callback=self.on_long_summary,
+            on_short_summary_callback=self.on_short_summary,
+            empty_pipeline=self.empty_pipeline,
+            logger=self.logger,
         )
-        processor.set_pipeline(self.empty_pipeline)
 
-        for topic in topics:
-            await processor.push(topic)
 
-        await processor.flush()
+@shared_task
+@asynctask
+async def task_send_webhook_if_needed(*, transcript_id: str):
+    """Send webhook if this is a room recording with webhook configured"""
+    transcript = await transcripts_controller.get_by_id(transcript_id)
+    if not transcript:
+        return
+
+    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",
+                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
+            )
 
 
 @shared_task
@@ -402,20 +359,19 @@ async def task_pipeline_file_process(*, transcript_id: str):
 
         await pipeline.process(audio_file)
 
-    except Exception:
+    except Exception as e:
+        logger.error(
+            f"File pipeline failed for transcript {transcript_id}: {type(e).__name__}: {str(e)}",
+            exc_info=True,
+            transcript_id=transcript_id,
+        )
         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
-            )
+    # Run post-processing chain: consent cleanup -> zulip -> webhook
+    post_chain = chain(
+        task_cleanup_consent.si(transcript_id=transcript_id),
+        task_pipeline_post_to_zulip.si(transcript_id=transcript_id),
+        task_send_webhook_if_needed.si(transcript_id=transcript_id),
+    )
+    post_chain.delay()

server/reflector/pipelines/main_live_pipeline.py

@@ -17,7 +17,6 @@ from contextlib import asynccontextmanager
 from typing import Generic
 
 import av
-import boto3
 from celery import chord, current_task, group, shared_task
 from pydantic import BaseModel
 from structlog import BoundLogger as Logger
@@ -85,6 +84,20 @@ def broadcast_to_sockets(func):
             message=resp.model_dump(mode="json"),
         )
 
+        transcript = await transcripts_controller.get_by_id(self.transcript_id)
+        if transcript and transcript.user_id:
+            # Emit only relevant events to the user room to avoid noisy updates.
+            # Allowed: STATUS, FINAL_TITLE, DURATION. All are prefixed with TRANSCRIPT_
+            allowed_user_events = {"STATUS", "FINAL_TITLE", "DURATION"}
+            if resp.event in allowed_user_events:
+                await self.ws_manager.send_json(
+                    room_id=f"user:{transcript.user_id}",
+                    message={
+                        "event": f"TRANSCRIPT_{resp.event}",
+                        "data": {"id": self.transcript_id, **resp.data},
+                    },
+                )
+
     return wrapper
 
 
@@ -570,6 +583,7 @@ async def cleanup_consent(transcript: Transcript, logger: Logger):
 
     consent_denied = False
     recording = None
+    meeting = None
     try:
         if transcript.recording_id:
             recording = await recordings_controller.get_by_id(transcript.recording_id)
@@ -580,8 +594,8 @@ async def cleanup_consent(transcript: Transcript, logger: Logger):
                         meeting.id
                     )
     except Exception as e:
-        logger.error(f"Failed to get fetch consent: {e}", exc_info=e)
-        consent_denied = True
+        logger.error(f"Failed to fetch consent: {e}", exc_info=e)
+        raise
 
     if not consent_denied:
         logger.info("Consent approved, keeping all files")
@@ -589,25 +603,24 @@ async def cleanup_consent(transcript: Transcript, logger: Logger):
 
     logger.info("Consent denied, cleaning up all related audio files")
 
-    if recording and recording.bucket_name and recording.object_key:
-        s3_whereby = boto3.client(
-            "s3",
-            aws_access_key_id=settings.AWS_WHEREBY_ACCESS_KEY_ID,
-            aws_secret_access_key=settings.AWS_WHEREBY_ACCESS_KEY_SECRET,
-        )
-        try:
-            s3_whereby.delete_object(
-                Bucket=recording.bucket_name, Key=recording.object_key
-            )
-            logger.info(
-                f"Deleted original Whereby recording: {recording.bucket_name}/{recording.object_key}"
-            )
-        except Exception as e:
-            logger.error(f"Failed to delete Whereby recording: {e}", exc_info=e)
+    deletion_errors = []
+    if recording and recording.bucket_name:
+        keys_to_delete = []
+        if recording.track_keys:
+            keys_to_delete = recording.track_keys
+        elif recording.object_key:
+            keys_to_delete = [recording.object_key]
+
+        master_storage = get_transcripts_storage()
+        for key in keys_to_delete:
+            try:
+                await master_storage.delete_file(key, bucket=recording.bucket_name)
+                logger.info(f"Deleted recording file: {recording.bucket_name}/{key}")
+            except Exception as e:
+                error_msg = f"Failed to delete {key}: {e}"
+                logger.error(error_msg, exc_info=e)
+                deletion_errors.append(error_msg)
 
-    # non-transactional, files marked for deletion not actually deleted is possible
-    await transcripts_controller.update(transcript, {"audio_deleted": True})
-    # 2. Delete processed audio from transcript storage S3 bucket
     if transcript.audio_location == "storage":
         storage = get_transcripts_storage()
         try:
@@ -616,18 +629,28 @@ async def cleanup_consent(transcript: Transcript, logger: Logger):
                 f"Deleted processed audio from storage: {transcript.storage_audio_path}"
             )
         except Exception as e:
-            logger.error(f"Failed to delete processed audio: {e}", exc_info=e)
+            error_msg = f"Failed to delete processed audio: {e}"
+            logger.error(error_msg, exc_info=e)
+            deletion_errors.append(error_msg)
 
-    # 3. Delete local audio files
     try:
         if hasattr(transcript, "audio_mp3_filename") and transcript.audio_mp3_filename:
             transcript.audio_mp3_filename.unlink(missing_ok=True)
         if hasattr(transcript, "audio_wav_filename") and transcript.audio_wav_filename:
             transcript.audio_wav_filename.unlink(missing_ok=True)
     except Exception as e:
-        logger.error(f"Failed to delete local audio files: {e}", exc_info=e)
+        error_msg = f"Failed to delete local audio files: {e}"
+        logger.error(error_msg, exc_info=e)
+        deletion_errors.append(error_msg)
 
-    logger.info("Consent cleanup done")
+    if deletion_errors:
+        logger.warning(
+            f"Consent cleanup completed with {len(deletion_errors)} errors",
+            errors=deletion_errors,
+        )
+    else:
+        await transcripts_controller.update(transcript, {"audio_deleted": True})
+        logger.info("Consent cleanup done - all audio deleted")
 
 
 @get_transcript

server/reflector/pipelines/main_multitrack_pipeline.py

@@ -0,0 +1,694 @@
+import asyncio
+import math
+import tempfile
+from fractions import Fraction
+from pathlib import Path
+
+import av
+from av.audio.resampler import AudioResampler
+from celery import chain, shared_task
+
+from reflector.asynctask import asynctask
+from reflector.db.transcripts import (
+    TranscriptStatus,
+    TranscriptWaveform,
+    transcripts_controller,
+)
+from reflector.logger import logger
+from reflector.pipelines import topic_processing
+from reflector.pipelines.main_file_pipeline import task_send_webhook_if_needed
+from reflector.pipelines.main_live_pipeline import (
+    PipelineMainBase,
+    broadcast_to_sockets,
+    task_cleanup_consent,
+    task_pipeline_post_to_zulip,
+)
+from reflector.pipelines.transcription_helpers import transcribe_file_with_processor
+from reflector.processors import AudioFileWriterProcessor
+from reflector.processors.audio_waveform_processor import AudioWaveformProcessor
+from reflector.processors.types import TitleSummary
+from reflector.processors.types import Transcript as TranscriptType
+from reflector.storage import Storage, get_transcripts_storage
+from reflector.utils.string import NonEmptyString
+
+# Audio encoding constants
+OPUS_STANDARD_SAMPLE_RATE = 48000
+OPUS_DEFAULT_BIT_RATE = 128000
+
+# Storage operation constants
+PRESIGNED_URL_EXPIRATION_SECONDS = 7200  # 2 hours
+
+
+class PipelineMainMultitrack(PipelineMainBase):
+    def __init__(self, transcript_id: str):
+        super().__init__(transcript_id=transcript_id)
+        self.logger = logger.bind(transcript_id=self.transcript_id)
+        self.empty_pipeline = topic_processing.EmptyPipeline(logger=self.logger)
+
+    async def pad_track_for_transcription(
+        self,
+        track_url: NonEmptyString,
+        track_idx: int,
+        storage: Storage,
+    ) -> NonEmptyString:
+        """
+        Pad a single track with silence based on stream metadata start_time.
+        Downloads from S3 presigned URL, processes via PyAV using tempfile, uploads to S3.
+        Returns presigned URL of padded track (or original URL if no padding needed).
+
+        Memory usage:
+        - Pattern: fixed_overhead(2-5MB) for PyAV codec/filters
+        - PyAV streams input efficiently (no full download, verified)
+        - Output written to tempfile (disk-based, not memory)
+        - Upload streams from file handle (boto3 chunks, typically 5-10MB)
+
+        Daily.co raw-tracks timing - Two approaches:
+
+            CURRENT APPROACH (PyAV metadata):
+            The WebM stream.start_time field encodes MEETING-RELATIVE timing:
+            - t=0: When Daily.co recording started (first participant joined)
+            - start_time=8.13s: This participant's track began 8.13s after recording started
+            - Purpose: Enables track alignment without external manifest files
+
+            This is NOT:
+            - Stream-internal offset (first packet timestamp relative to stream start)
+            - Absolute/wall-clock time
+            - Recording duration
+
+            ALTERNATIVE APPROACH (filename parsing):
+            Daily.co filenames contain Unix timestamps (milliseconds):
+            Format: {recording_start_ts}-{participant_id}-cam-audio-{track_start_ts}.webm
+            Example: 1760988935484-52f7f48b-fbab-431f-9a50-87b9abfc8255-cam-audio-1760988935922.webm
+
+            Can calculate offset: (track_start_ts - recording_start_ts) / 1000
+            - Track 0: (1760988935922 - 1760988935484) / 1000 = 0.438s
+            - Track 1: (1760988943823 - 1760988935484) / 1000 = 8.339s
+
+            TIME DIFFERENCE: PyAV metadata vs filename timestamps differ by ~209ms:
+            - Track 0: filename=438ms, metadata=229ms (diff: 209ms)
+            - Track 1: filename=8339ms, metadata=8130ms (diff: 209ms)
+
+            Consistent delta suggests network/encoding delay. PyAV metadata is ground truth
+            (represents when audio stream actually started vs when file upload initiated).
+
+            Example with 2 participants:
+                Track A: start_time=0.2s → Joined 200ms after recording began
+                Track B: start_time=8.1s → Joined 8.1 seconds later
+
+                After padding:
+                    Track A: [0.2s silence] + [speech...]
+                    Track B: [8.1s silence] + [speech...]
+
+                Whisper transcription timestamps are now synchronized:
+                    Track A word at 5.0s → happened at meeting t=5.0s
+                    Track B word at 10.0s → happened at meeting t=10.0s
+
+                Merging just sorts by timestamp - no offset calculation needed.
+
+        Padding coincidentally involves re-encoding. It's important when we work with Daily.co + Whisper.
+        This is because Daily.co returns recordings with skipped frames e.g. when microphone muted.
+        Daily.co doesn't understand those frames and ignores them, causing timestamp issues in transcription.
+        Re-encoding restores those frames. We do padding and re-encoding together just because it's convenient and more performant:
+        we need padded values for mix mp3 anyways
+        """
+
+        transcript = await self.get_transcript()
+
+        try:
+            # PyAV streams input from S3 URL efficiently (2-5MB fixed overhead for codec/filters)
+            with av.open(track_url) as in_container:
+                start_time_seconds = self._extract_stream_start_time_from_container(
+                    in_container, track_idx
+                )
+
+                if start_time_seconds <= 0:
+                    self.logger.info(
+                        f"Track {track_idx} requires no padding (start_time={start_time_seconds}s)",
+                        track_idx=track_idx,
+                    )
+                    return track_url
+
+                # Use tempfile instead of BytesIO for better memory efficiency
+                # Reduces peak memory usage during encoding/upload
+                with tempfile.NamedTemporaryFile(
+                    suffix=".webm", delete=False
+                ) as temp_file:
+                    temp_path = temp_file.name
+
+                try:
+                    self._apply_audio_padding_to_file(
+                        in_container, temp_path, start_time_seconds, track_idx
+                    )
+
+                    storage_path = (
+                        f"file_pipeline/{transcript.id}/tracks/padded_{track_idx}.webm"
+                    )
+
+                    # Upload using file handle for streaming
+                    with open(temp_path, "rb") as padded_file:
+                        await storage.put_file(storage_path, padded_file)
+                finally:
+                    # Clean up temp file
+                    Path(temp_path).unlink(missing_ok=True)
+
+                padded_url = await storage.get_file_url(
+                    storage_path,
+                    operation="get_object",
+                    expires_in=PRESIGNED_URL_EXPIRATION_SECONDS,
+                )
+
+                self.logger.info(
+                    f"Successfully padded track {track_idx}",
+                    track_idx=track_idx,
+                    start_time_seconds=start_time_seconds,
+                    padded_url=padded_url,
+                )
+
+                return padded_url
+
+        except Exception as e:
+            self.logger.error(
+                f"Failed to process track {track_idx}",
+                track_idx=track_idx,
+                url=track_url,
+                error=str(e),
+                exc_info=True,
+            )
+            raise Exception(
+                f"Track {track_idx} padding failed - transcript would have incorrect timestamps"
+            ) from e
+
+    def _extract_stream_start_time_from_container(
+        self, container, track_idx: int
+    ) -> float:
+        """
+        Extract meeting-relative start time from WebM stream metadata.
+        Uses PyAV to read stream.start_time from WebM container.
+        More accurate than filename timestamps by ~209ms due to network/encoding delays.
+        """
+        start_time_seconds = 0.0
+        try:
+            audio_streams = [s for s in container.streams if s.type == "audio"]
+            stream = audio_streams[0] if audio_streams else container.streams[0]
+
+            # 1) Try stream-level start_time (most reliable for Daily.co tracks)
+            if stream.start_time is not None and stream.time_base is not None:
+                start_time_seconds = float(stream.start_time * stream.time_base)
+
+            # 2) Fallback to container-level start_time (in av.time_base units)
+            if (start_time_seconds <= 0) and (container.start_time is not None):
+                start_time_seconds = float(container.start_time * av.time_base)
+
+            # 3) Fallback to first packet DTS in stream.time_base
+            if start_time_seconds <= 0:
+                for packet in container.demux(stream):
+                    if packet.dts is not None:
+                        start_time_seconds = float(packet.dts * stream.time_base)
+                        break
+        except Exception as e:
+            self.logger.warning(
+                "PyAV metadata read failed; assuming 0 start_time",
+                track_idx=track_idx,
+                error=str(e),
+            )
+            start_time_seconds = 0.0
+
+        self.logger.info(
+            f"Track {track_idx} stream metadata: start_time={start_time_seconds:.3f}s",
+            track_idx=track_idx,
+        )
+        return start_time_seconds
+
+    def _apply_audio_padding_to_file(
+        self,
+        in_container,
+        output_path: str,
+        start_time_seconds: float,
+        track_idx: int,
+    ) -> None:
+        """Apply silence padding to audio track using PyAV filter graph, writing to file"""
+        delay_ms = math.floor(start_time_seconds * 1000)
+
+        self.logger.info(
+            f"Padding track {track_idx} with {delay_ms}ms delay using PyAV",
+            track_idx=track_idx,
+            delay_ms=delay_ms,
+        )
+
+        try:
+            with av.open(output_path, "w", format="webm") as out_container:
+                in_stream = next(
+                    (s for s in in_container.streams if s.type == "audio"), None
+                )
+                if in_stream is None:
+                    raise Exception("No audio stream in input")
+
+                out_stream = out_container.add_stream(
+                    "libopus", rate=OPUS_STANDARD_SAMPLE_RATE
+                )
+                out_stream.bit_rate = OPUS_DEFAULT_BIT_RATE
+                graph = av.filter.Graph()
+
+                abuf_args = (
+                    f"time_base=1/{OPUS_STANDARD_SAMPLE_RATE}:"
+                    f"sample_rate={OPUS_STANDARD_SAMPLE_RATE}:"
+                    f"sample_fmt=s16:"
+                    f"channel_layout=stereo"
+                )
+                src = graph.add("abuffer", args=abuf_args, name="src")
+                aresample_f = graph.add("aresample", args="async=1", name="ares")
+                # adelay requires one delay value per channel separated by '|'
+                delays_arg = f"{delay_ms}|{delay_ms}"
+                adelay_f = graph.add(
+                    "adelay", args=f"delays={delays_arg}:all=1", name="delay"
+                )
+                sink = graph.add("abuffersink", name="sink")
+
+                src.link_to(aresample_f)
+                aresample_f.link_to(adelay_f)
+                adelay_f.link_to(sink)
+                graph.configure()
+
+                resampler = AudioResampler(
+                    format="s16", layout="stereo", rate=OPUS_STANDARD_SAMPLE_RATE
+                )
+                # Decode -> resample -> push through graph -> encode Opus
+                for frame in in_container.decode(in_stream):
+                    out_frames = resampler.resample(frame) or []
+                    for rframe in out_frames:
+                        rframe.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                        rframe.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                        src.push(rframe)
+
+                        while True:
+                            try:
+                                f_out = sink.pull()
+                            except Exception:
+                                break
+                            f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                            f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                            for packet in out_stream.encode(f_out):
+                                out_container.mux(packet)
+
+                src.push(None)
+                while True:
+                    try:
+                        f_out = sink.pull()
+                    except Exception:
+                        break
+                    f_out.sample_rate = OPUS_STANDARD_SAMPLE_RATE
+                    f_out.time_base = Fraction(1, OPUS_STANDARD_SAMPLE_RATE)
+                    for packet in out_stream.encode(f_out):
+                        out_container.mux(packet)
+
+                for packet in out_stream.encode(None):
+                    out_container.mux(packet)
+        except Exception as e:
+            self.logger.error(
+                "PyAV padding failed for track",
+                track_idx=track_idx,
+                delay_ms=delay_ms,
+                error=str(e),
+                exc_info=True,
+            )
+            raise
+
+    async def mixdown_tracks(
+        self,
+        track_urls: list[str],
+        writer: AudioFileWriterProcessor,
+        offsets_seconds: list[float] | None = None,
+    ) -> None:
+        """Multi-track mixdown using PyAV filter graph (amix), reading from S3 presigned URLs"""
+
+        target_sample_rate: int | None = None
+        for url in track_urls:
+            if not url:
+                continue
+            container = None
+            try:
+                container = av.open(url)
+                for frame in container.decode(audio=0):
+                    target_sample_rate = frame.sample_rate
+                    break
+            except Exception:
+                continue
+            finally:
+                if container is not None:
+                    container.close()
+            if target_sample_rate:
+                break
+
+        if not target_sample_rate:
+            self.logger.error("Mixdown failed - no decodable audio frames found")
+            raise Exception("Mixdown failed: No decodable audio frames in any track")
+        # Build PyAV filter graph:
+        # N abuffer (s32/stereo)
+        #   -> optional adelay per input (for alignment)
+        #   -> amix (s32)
+        #   -> aformat(s16)
+        #   -> sink
+        graph = av.filter.Graph()
+        inputs = []
+        valid_track_urls = [url for url in track_urls if url]
+        input_offsets_seconds = None
+        if offsets_seconds is not None:
+            input_offsets_seconds = [
+                offsets_seconds[i] for i, url in enumerate(track_urls) if url
+            ]
+        for idx, url in enumerate(valid_track_urls):
+            args = (
+                f"time_base=1/{target_sample_rate}:"
+                f"sample_rate={target_sample_rate}:"
+                f"sample_fmt=s32:"
+                f"channel_layout=stereo"
+            )
+            in_ctx = graph.add("abuffer", args=args, name=f"in{idx}")
+            inputs.append(in_ctx)
+
+        if not inputs:
+            self.logger.error("Mixdown failed - no valid inputs for graph")
+            raise Exception("Mixdown failed: No valid inputs for filter graph")
+
+        mixer = graph.add("amix", args=f"inputs={len(inputs)}:normalize=0", name="mix")
+
+        fmt = graph.add(
+            "aformat",
+            args=(
+                f"sample_fmts=s32:channel_layouts=stereo:sample_rates={target_sample_rate}"
+            ),
+            name="fmt",
+        )
+
+        sink = graph.add("abuffersink", name="out")
+
+        # Optional per-input delay before mixing
+        delays_ms: list[int] = []
+        if input_offsets_seconds is not None:
+            base = min(input_offsets_seconds) if input_offsets_seconds else 0.0
+            delays_ms = [
+                max(0, int(round((o - base) * 1000))) for o in input_offsets_seconds
+            ]
+        else:
+            delays_ms = [0 for _ in inputs]
+
+        for idx, in_ctx in enumerate(inputs):
+            delay_ms = delays_ms[idx] if idx < len(delays_ms) else 0
+            if delay_ms > 0:
+                # adelay requires one value per channel; use same for stereo
+                adelay = graph.add(
+                    "adelay",
+                    args=f"delays={delay_ms}|{delay_ms}:all=1",
+                    name=f"delay{idx}",
+                )
+                in_ctx.link_to(adelay)
+                adelay.link_to(mixer, 0, idx)
+            else:
+                in_ctx.link_to(mixer, 0, idx)
+        mixer.link_to(fmt)
+        fmt.link_to(sink)
+        graph.configure()
+
+        containers = []
+        try:
+            # Open all containers with cleanup guaranteed
+            for i, url in enumerate(valid_track_urls):
+                try:
+                    c = av.open(url)
+                    containers.append(c)
+                except Exception as e:
+                    self.logger.warning(
+                        "Mixdown: failed to open container from URL",
+                        input=i,
+                        url=url,
+                        error=str(e),
+                    )
+
+            if not containers:
+                self.logger.error("Mixdown failed - no valid containers opened")
+                raise Exception("Mixdown failed: Could not open any track containers")
+
+            decoders = [c.decode(audio=0) for c in containers]
+            active = [True] * len(decoders)
+            resamplers = [
+                AudioResampler(format="s32", layout="stereo", rate=target_sample_rate)
+                for _ in decoders
+            ]
+
+            while any(active):
+                for i, (dec, is_active) in enumerate(zip(decoders, active)):
+                    if not is_active:
+                        continue
+                    try:
+                        frame = next(dec)
+                    except StopIteration:
+                        active[i] = False
+                        continue
+
+                    if frame.sample_rate != target_sample_rate:
+                        continue
+                    out_frames = resamplers[i].resample(frame) or []
+                    for rf in out_frames:
+                        rf.sample_rate = target_sample_rate
+                        rf.time_base = Fraction(1, target_sample_rate)
+                        inputs[i].push(rf)
+
+                    while True:
+                        try:
+                            mixed = sink.pull()
+                        except Exception:
+                            break
+                        mixed.sample_rate = target_sample_rate
+                        mixed.time_base = Fraction(1, target_sample_rate)
+                        await writer.push(mixed)
+
+            for in_ctx in inputs:
+                in_ctx.push(None)
+            while True:
+                try:
+                    mixed = sink.pull()
+                except Exception:
+                    break
+                mixed.sample_rate = target_sample_rate
+                mixed.time_base = Fraction(1, target_sample_rate)
+                await writer.push(mixed)
+        finally:
+            # Cleanup all containers, even if processing failed
+            for c in containers:
+                if c is not None:
+                    try:
+                        c.close()
+                    except Exception:
+                        pass  # Best effort cleanup
+
+    @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 on_waveform(self, data):
+        async with self.transaction():
+            waveform = TranscriptWaveform(waveform=data)
+            transcript = await self.get_transcript()
+            return await transcripts_controller.append_event(
+                transcript=transcript, event="WAVEFORM", data=waveform
+            )
+
+    async def process(self, bucket_name: str, track_keys: list[str]):
+        transcript = await self.get_transcript()
+        async with self.transaction():
+            await transcripts_controller.update(
+                transcript,
+                {
+                    "events": [],
+                    "topics": [],
+                },
+            )
+
+        source_storage = get_transcripts_storage()
+        transcript_storage = source_storage
+
+        track_urls: list[str] = []
+        for key in track_keys:
+            url = await source_storage.get_file_url(
+                key,
+                operation="get_object",
+                expires_in=PRESIGNED_URL_EXPIRATION_SECONDS,
+                bucket=bucket_name,
+            )
+            track_urls.append(url)
+            self.logger.info(
+                f"Generated presigned URL for track from {bucket_name}",
+                key=key,
+            )
+
+        created_padded_files = set()
+        padded_track_urls: list[str] = []
+        for idx, url in enumerate(track_urls):
+            padded_url = await self.pad_track_for_transcription(
+                url, idx, transcript_storage
+            )
+            padded_track_urls.append(padded_url)
+            if padded_url != url:
+                storage_path = f"file_pipeline/{transcript.id}/tracks/padded_{idx}.webm"
+                created_padded_files.add(storage_path)
+            self.logger.info(f"Track {idx} processed, padded URL: {padded_url}")
+
+        transcript.data_path.mkdir(parents=True, exist_ok=True)
+
+        mp3_writer = AudioFileWriterProcessor(
+            path=str(transcript.audio_mp3_filename),
+            on_duration=self.on_duration,
+        )
+        await self.mixdown_tracks(padded_track_urls, mp3_writer, offsets_seconds=None)
+        await mp3_writer.flush()
+
+        if not transcript.audio_mp3_filename.exists():
+            raise Exception(
+                "Mixdown failed - no MP3 file generated. Cannot proceed without playable audio."
+            )
+
+        storage_path = f"{transcript.id}/audio.mp3"
+        # Use file handle streaming to avoid loading entire MP3 into memory
+        mp3_size = transcript.audio_mp3_filename.stat().st_size
+        with open(transcript.audio_mp3_filename, "rb") as mp3_file:
+            await transcript_storage.put_file(storage_path, mp3_file)
+        mp3_url = await transcript_storage.get_file_url(storage_path)
+
+        await transcripts_controller.update(transcript, {"audio_location": "storage"})
+
+        self.logger.info(
+            f"Uploaded mixed audio to storage",
+            storage_path=storage_path,
+            size=mp3_size,
+            url=mp3_url,
+        )
+
+        self.logger.info("Generating waveform from mixed audio")
+        waveform_processor = AudioWaveformProcessor(
+            audio_path=transcript.audio_mp3_filename,
+            waveform_path=transcript.audio_waveform_filename,
+            on_waveform=self.on_waveform,
+        )
+        waveform_processor.set_pipeline(self.empty_pipeline)
+        await waveform_processor.flush()
+        self.logger.info("Waveform generated successfully")
+
+        speaker_transcripts: list[TranscriptType] = []
+        for idx, padded_url in enumerate(padded_track_urls):
+            if not padded_url:
+                continue
+
+            t = await self.transcribe_file(padded_url, transcript.source_language)
+
+            if not t.words:
+                continue
+
+            for w in t.words:
+                w.speaker = idx
+
+            speaker_transcripts.append(t)
+            self.logger.info(
+                f"Track {idx} transcribed successfully with {len(t.words)} words",
+                track_idx=idx,
+            )
+
+        valid_track_count = len([url for url in padded_track_urls if url])
+        if valid_track_count > 0 and len(speaker_transcripts) != valid_track_count:
+            raise Exception(
+                f"Only {len(speaker_transcripts)}/{valid_track_count} tracks transcribed successfully. "
+                f"All tracks must succeed to avoid incomplete transcripts."
+            )
+
+        if not speaker_transcripts:
+            raise Exception("No valid track transcriptions")
+
+        self.logger.info(f"Cleaning up {len(created_padded_files)} temporary S3 files")
+        cleanup_tasks = []
+        for storage_path in created_padded_files:
+            cleanup_tasks.append(transcript_storage.delete_file(storage_path))
+
+        if cleanup_tasks:
+            cleanup_results = await asyncio.gather(
+                *cleanup_tasks, return_exceptions=True
+            )
+            for storage_path, result in zip(created_padded_files, cleanup_results):
+                if isinstance(result, Exception):
+                    self.logger.warning(
+                        "Failed to cleanup temporary padded track",
+                        storage_path=storage_path,
+                        error=str(result),
+                    )
+
+        merged_words = []
+        for t in speaker_transcripts:
+            merged_words.extend(t.words)
+        merged_words.sort(
+            key=lambda w: w.start if hasattr(w, "start") and w.start is not None else 0
+        )
+
+        merged_transcript = TranscriptType(words=merged_words, translation=None)
+
+        await self.on_transcript(merged_transcript)
+
+        topics = await self.detect_topics(merged_transcript, transcript.target_language)
+        await asyncio.gather(
+            self.generate_title(topics),
+            self.generate_summaries(topics),
+            return_exceptions=False,
+        )
+
+        await self.set_status(transcript.id, "ended")
+
+    async def transcribe_file(self, audio_url: str, language: str) -> TranscriptType:
+        return await transcribe_file_with_processor(audio_url, language)
+
+    async def detect_topics(
+        self, transcript: TranscriptType, target_language: str
+    ) -> list[TitleSummary]:
+        return await topic_processing.detect_topics(
+            transcript,
+            target_language,
+            on_topic_callback=self.on_topic,
+            empty_pipeline=self.empty_pipeline,
+        )
+
+    async def generate_title(self, topics: list[TitleSummary]):
+        return await topic_processing.generate_title(
+            topics,
+            on_title_callback=self.on_title,
+            empty_pipeline=self.empty_pipeline,
+            logger=self.logger,
+        )
+
+    async def generate_summaries(self, topics: list[TitleSummary]):
+        transcript = await self.get_transcript()
+        return await topic_processing.generate_summaries(
+            topics,
+            transcript,
+            on_long_summary_callback=self.on_long_summary,
+            on_short_summary_callback=self.on_short_summary,
+            empty_pipeline=self.empty_pipeline,
+            logger=self.logger,
+        )
+
+
+@shared_task
+@asynctask
+async def task_pipeline_multitrack_process(
+    *, transcript_id: str, bucket_name: str, track_keys: list[str]
+):
+    pipeline = PipelineMainMultitrack(transcript_id=transcript_id)
+    try:
+        await pipeline.set_status(transcript_id, "processing")
+        await pipeline.process(bucket_name, track_keys)
+    except Exception:
+        await pipeline.set_status(transcript_id, "error")
+        raise
+
+    post_chain = chain(
+        task_cleanup_consent.si(transcript_id=transcript_id),
+        task_pipeline_post_to_zulip.si(transcript_id=transcript_id),
+        task_send_webhook_if_needed.si(transcript_id=transcript_id),
+    )
+    post_chain.delay()

server/reflector/pipelines/topic_processing.py

@@ -0,0 +1,109 @@
+"""
+Topic processing utilities
+==========================
+
+Shared topic detection, title generation, and summarization logic
+used across file and multitrack pipelines.
+"""
+
+from typing import Callable
+
+import structlog
+
+from reflector.db.transcripts import Transcript
+from reflector.processors import (
+    TranscriptFinalSummaryProcessor,
+    TranscriptFinalTitleProcessor,
+    TranscriptTopicDetectorProcessor,
+)
+from reflector.processors.types import TitleSummary
+from reflector.processors.types import Transcript as TranscriptType
+
+
+class EmptyPipeline:
+    def __init__(self, logger: structlog.BoundLogger):
+        self.logger = logger
+
+    def get_pref(self, k, d=None):
+        return d
+
+    async def emit(self, event):
+        pass
+
+
+async def detect_topics(
+    transcript: TranscriptType,
+    target_language: str,
+    *,
+    on_topic_callback: Callable,
+    empty_pipeline: EmptyPipeline,
+) -> list[TitleSummary]:
+    chunk_size = 300
+    topics: list[TitleSummary] = []
+
+    async def on_topic(topic: TitleSummary):
+        topics.append(topic)
+        return await on_topic_callback(topic)
+
+    topic_detector = TranscriptTopicDetectorProcessor(callback=on_topic)
+    topic_detector.set_pipeline(empty_pipeline)
+
+    for i in range(0, len(transcript.words), chunk_size):
+        chunk_words = transcript.words[i : i + chunk_size]
+        if not chunk_words:
+            continue
+
+        chunk_transcript = TranscriptType(
+            words=chunk_words, translation=transcript.translation
+        )
+
+        await topic_detector.push(chunk_transcript)
+
+    await topic_detector.flush()
+    return topics
+
+
+async def generate_title(
+    topics: list[TitleSummary],
+    *,
+    on_title_callback: Callable,
+    empty_pipeline: EmptyPipeline,
+    logger: structlog.BoundLogger,
+):
+    if not topics:
+        logger.warning("No topics for title generation")
+        return
+
+    processor = TranscriptFinalTitleProcessor(callback=on_title_callback)
+    processor.set_pipeline(empty_pipeline)
+
+    for topic in topics:
+        await processor.push(topic)
+
+    await processor.flush()
+
+
+async def generate_summaries(
+    topics: list[TitleSummary],
+    transcript: Transcript,
+    *,
+    on_long_summary_callback: Callable,
+    on_short_summary_callback: Callable,
+    empty_pipeline: EmptyPipeline,
+    logger: structlog.BoundLogger,
+):
+    if not topics:
+        logger.warning("No topics for summary generation")
+        return
+
+    processor = TranscriptFinalSummaryProcessor(
+        transcript=transcript,
+        callback=on_long_summary_callback,
+        on_short_summary=on_short_summary_callback,
+    )
+    processor.set_pipeline(empty_pipeline)
+
+    for topic in topics:
+        await processor.push(topic)
+
+    await processor.flush()

server/reflector/pipelines/transcription_helpers.py

@@ -0,0 +1,34 @@
+from reflector.processors.file_transcript import FileTranscriptInput
+from reflector.processors.file_transcript_auto import FileTranscriptAutoProcessor
+from reflector.processors.types import Transcript as TranscriptType
+
+
+async def transcribe_file_with_processor(
+    audio_url: str,
+    language: str,
+    processor_name: str | None = None,
+) -> TranscriptType:
+    processor = (
+        FileTranscriptAutoProcessor(name=processor_name)
+        if processor_name
+        else FileTranscriptAutoProcessor()
+    )
+    input_data = FileTranscriptInput(audio_url=audio_url, language=language)
+
+    result: TranscriptType | None = None
+
+    async def capture_result(transcript):
+        nonlocal result
+        result = transcript
+
+    processor.on(capture_result)
+    await processor.push(input_data)
+    await processor.flush()
+
+    if not result:
+        processor_label = processor_name or "default"
+        raise ValueError(
+            f"No transcript captured from {processor_label} processor for audio: {audio_url}"
+        )
+
+    return result

server/reflector/processors/file_diarization_modal.py

@@ -47,6 +47,7 @@ class FileDiarizationModalProcessor(FileDiarizationProcessor):
                     "audio_file_url": data.audio_url,
                     "timestamp": 0,
                 },
+                follow_redirects=True,
             )
             response.raise_for_status()
             diarization_data = response.json()["diarization"]

server/reflector/processors/file_transcript_modal.py

@@ -54,7 +54,18 @@ class FileTranscriptModalProcessor(FileTranscriptProcessor):
                     "language": data.language,
                     "batch": True,
                 },
+                follow_redirects=True,
             )
+
+            if response.status_code != 200:
+                error_body = response.text
+                self.logger.error(
+                    "Modal API error",
+                    audio_url=data.audio_url,
+                    status_code=response.status_code,
+                    error_body=error_body,
+                )
+
             response.raise_for_status()
             result = response.json()
 

server/reflector/processors/summary/summary_builder.py

@@ -165,6 +165,7 @@ class SummaryBuilder:
         self.llm: LLM = llm
         self.model_name: str = llm.model_name
         self.logger = logger or structlog.get_logger()
+        self.participant_instructions: str | None = None
         if filename:
             self.read_transcript_from_file(filename)
 
@@ -191,14 +192,61 @@ class SummaryBuilder:
         self, prompt: str, output_cls: Type[T], tone_name: str | None = None
     ) -> T:
         """Generic function to get structured output from LLM for non-function-calling models."""
+        # Add participant instructions to the prompt if available
+        enhanced_prompt = self._enhance_prompt_with_participants(prompt)
         return await self.llm.get_structured_response(
-            prompt, [self.transcript], output_cls, tone_name=tone_name
+            enhanced_prompt, [self.transcript], output_cls, tone_name=tone_name
         )
 
+    async def _get_response(
+        self, prompt: str, texts: list[str], tone_name: str | None = None
+    ) -> str:
+        """Get text response with automatic participant instructions injection."""
+        enhanced_prompt = self._enhance_prompt_with_participants(prompt)
+        return await self.llm.get_response(enhanced_prompt, texts, tone_name=tone_name)
+
+    def _enhance_prompt_with_participants(self, prompt: str) -> str:
+        """Add participant instructions to any prompt if participants are known."""
+        if self.participant_instructions:
+            self.logger.debug("Adding participant instructions to prompt")
+            return f"{prompt}\n\n{self.participant_instructions}"
+        return prompt
+
     # ----------------------------------------------------------------------------
     # Participants
     # ----------------------------------------------------------------------------
 
+    def set_known_participants(self, participants: list[str]) -> None:
+        """
+        Set known participants directly without LLM identification.
+        This is used when participants are already identified and stored.
+        They are appended at the end of the transcript, providing more context for the assistant.
+        """
+        if not participants:
+            self.logger.warning("No participants provided")
+            return
+
+        self.logger.info(
+            "Using known participants",
+            participants=participants,
+        )
+
+        participants_md = self.format_list_md(participants)
+        self.transcript += f"\n\n# Participants\n\n{participants_md}"
+
+        # Set instructions that will be automatically added to all prompts
+        participants_list = ", ".join(participants)
+        self.participant_instructions = dedent(
+            f"""
+            # IMPORTANT: Participant Names
+            The following participants are identified in this conversation: {participants_list}
+
+            You MUST use these specific participant names when referring to people in your response.
+            Do NOT use generic terms like "a participant", "someone", "attendee", "Speaker 1", "Speaker 2", etc.
+            Always refer to people by their actual names (e.g., "John suggested..." not "A participant suggested...").
+            """
+        ).strip()
+
     async def identify_participants(self) -> None:
         """
         From a transcript, try to identify the participants using TreeSummarize with structured output.
@@ -232,6 +280,19 @@ class SummaryBuilder:
             if unique_participants:
                 participants_md = self.format_list_md(unique_participants)
                 self.transcript += f"\n\n# Participants\n\n{participants_md}"
+
+                # Set instructions that will be automatically added to all prompts
+                participants_list = ", ".join(unique_participants)
+                self.participant_instructions = dedent(
+                    f"""
+                    # IMPORTANT: Participant Names
+                    The following participants are identified in this conversation: {participants_list}
+
+                    You MUST use these specific participant names when referring to people in your response.
+                    Do NOT use generic terms like "a participant", "someone", "attendee", "Speaker 1", "Speaker 2", etc.
+                    Always refer to people by their actual names (e.g., "John suggested..." not "A participant suggested...").
+                    """
+                ).strip()
             else:
                 self.logger.warning("No participants identified in the transcript")
 
@@ -318,13 +379,13 @@ class SummaryBuilder:
         for subject in self.subjects:
             detailed_prompt = DETAILED_SUBJECT_PROMPT_TEMPLATE.format(subject=subject)
 
-            detailed_response = await self.llm.get_response(
+            detailed_response = await self._get_response(
                 detailed_prompt, [self.transcript], tone_name="Topic assistant"
             )
 
             paragraph_prompt = PARAGRAPH_SUMMARY_PROMPT
 
-            paragraph_response = await self.llm.get_response(
+            paragraph_response = await self._get_response(
                 paragraph_prompt, [str(detailed_response)], tone_name="Topic summarizer"
             )
 
@@ -345,7 +406,7 @@ class SummaryBuilder:
 
         recap_prompt = RECAP_PROMPT
 
-        recap_response = await self.llm.get_response(
+        recap_response = await self._get_response(
             recap_prompt, [summaries_text], tone_name="Recap summarizer"
         )
 

server/reflector/processors/transcript_final_summary.py

@@ -26,7 +26,25 @@ class TranscriptFinalSummaryProcessor(Processor):
     async def get_summary_builder(self, text) -> SummaryBuilder:
         builder = SummaryBuilder(self.llm, logger=self.logger)
         builder.set_transcript(text)
-        await builder.identify_participants()
+
+        # Use known participants if available, otherwise identify them
+        if self.transcript and self.transcript.participants:
+            # Extract participant names from the stored participants
+            participant_names = [p.name for p in self.transcript.participants if p.name]
+            if participant_names:
+                self.logger.info(
+                    f"Using {len(participant_names)} known participants from transcript"
+                )
+                builder.set_known_participants(participant_names)
+            else:
+                self.logger.info(
+                    "Participants field exists but is empty, identifying participants"
+                )
+                await builder.identify_participants()
+        else:
+            self.logger.info("No participants stored, identifying participants")
+            await builder.identify_participants()
+
         await builder.generate_summary()
         return builder
 
@@ -49,18 +67,30 @@ class TranscriptFinalSummaryProcessor(Processor):
         speakermap = {}
         if self.transcript:
             speakermap = {
-                participant["speaker"]: participant["name"]
-                for participant in self.transcript.participants
+                p.speaker: p.name
+                for p in (self.transcript.participants or [])
+                if p.speaker is not None and p.name
             }
+            self.logger.info(
+                f"Built speaker map with {len(speakermap)} participants",
+                speakermap=speakermap,
+            )
 
         # build the transcript as a single string
-        # XXX: unsure if the participants name as replaced directly in speaker ?
+        # Replace speaker IDs with actual participant names if available
         text_transcript = []
+        unique_speakers = set()
         for topic in self.chunks:
             for segment in topic.transcript.as_segments():
                 name = speakermap.get(segment.speaker, f"Speaker {segment.speaker}")
+                unique_speakers.add((segment.speaker, name))
                 text_transcript.append(f"{name}: {segment.text}")
 
+        self.logger.info(
+            f"Built transcript with {len(unique_speakers)} unique speakers",
+            speakers=list(unique_speakers),
+        )
+
         text_transcript = "\n".join(text_transcript)
 
         last_chunk = self.chunks[-1]

server/reflector/processors/transcript_topic_detector.py

@@ -1,6 +1,6 @@
 from textwrap import dedent
 
-from pydantic import BaseModel, Field
+from pydantic import AliasChoices, BaseModel, Field
 
 from reflector.llm import LLM
 from reflector.processors.base import Processor
@@ -34,8 +34,14 @@ TOPIC_PROMPT = dedent(
 class TopicResponse(BaseModel):
     """Structured response for topic detection"""
 
-    title: str = Field(description="A descriptive title for the topic being discussed")
-    summary: str = Field(description="A concise 1-2 sentence summary of the discussion")
+    title: str = Field(
+        description="A descriptive title for the topic being discussed",
+        validation_alias=AliasChoices("title", "Title"),
+    )
+    summary: str = Field(
+        description="A concise 1-2 sentence summary of the discussion",
+        validation_alias=AliasChoices("summary", "Summary"),
+    )
 
 
 class TranscriptTopicDetectorProcessor(Processor):

server/reflector/processors/types.py

@@ -4,11 +4,8 @@ import tempfile
 from pathlib import Path
 from typing import Annotated, TypedDict
 
-from profanityfilter import ProfanityFilter
 from pydantic import BaseModel, Field, PrivateAttr
 
-from reflector.redis_cache import redis_cache
-
 
 class DiarizationSegment(TypedDict):
     """Type definition for diarization segment containing speaker information"""
@@ -20,9 +17,6 @@ class DiarizationSegment(TypedDict):
 
 PUNC_RE = re.compile(r"[.;:?!…]")
 
-profanity_filter = ProfanityFilter()
-profanity_filter.set_censor("*")
-
 
 class AudioFile(BaseModel):
     name: str
@@ -124,21 +118,11 @@ def words_to_segments(words: list[Word]) -> list[TranscriptSegment]:
 
 class Transcript(BaseModel):
     translation: str | None = None
-    words: list[Word] = None
-
-    @property
-    def raw_text(self):
-        # Uncensored text
-        return "".join([word.text for word in self.words])
-
-    @redis_cache(prefix="profanity", duration=3600 * 24 * 7)
-    def _get_censored_text(self, text: str):
-        return profanity_filter.censor(text).strip()
+    words: list[Word] = []
 
     @property
     def text(self):
-        # Censored text
-        return self._get_censored_text(self.raw_text)
+        return "".join([word.text for word in self.words])
 
     @property
     def human_timestamp(self):
@@ -170,12 +154,6 @@ class Transcript(BaseModel):
             word.start += offset
             word.end += offset
 
-    def clone(self):
-        words = [
-            Word(text=word.text, start=word.start, end=word.end) for word in self.words
-        ]
-        return Transcript(text=self.text, translation=self.translation, words=words)
-
     def as_segments(self) -> list[TranscriptSegment]:
         return words_to_segments(self.words)
 

server/reflector/redis_cache.py

@@ -1,10 +1,17 @@
+import asyncio
 import functools
 import json
+from typing import Optional
 
 import redis
+import redis.asyncio as redis_async
+import structlog
+from redis.exceptions import LockError
 
 from reflector.settings import settings
 
+logger = structlog.get_logger(__name__)
+
 redis_clients = {}
 
 
@@ -21,6 +28,12 @@ def get_redis_client(db=0):
     return redis_clients[db]
 
 
+async def get_async_redis_client(db: int = 0):
+    return await redis_async.from_url(
+        f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/{db}"
+    )
+
+
 def redis_cache(prefix="cache", duration=3600, db=settings.REDIS_CACHE_DB, argidx=1):
     """
     Cache the result of a function in Redis.
@@ -49,3 +62,87 @@ def redis_cache(prefix="cache", duration=3600, db=settings.REDIS_CACHE_DB, argid
         return wrapper
 
     return decorator
+
+
+class RedisAsyncLock:
+    def __init__(
+        self,
+        key: str,
+        timeout: int = 120,
+        extend_interval: int = 30,
+        skip_if_locked: bool = False,
+        blocking: bool = True,
+        blocking_timeout: Optional[float] = None,
+    ):
+        self.key = f"async_lock:{key}"
+        self.timeout = timeout
+        self.extend_interval = extend_interval
+        self.skip_if_locked = skip_if_locked
+        self.blocking = blocking
+        self.blocking_timeout = blocking_timeout
+        self._lock = None
+        self._redis = None
+        self._extend_task = None
+        self._acquired = False
+
+    async def _extend_lock_periodically(self):
+        while True:
+            try:
+                await asyncio.sleep(self.extend_interval)
+                if self._lock:
+                    await self._lock.extend(self.timeout, replace_ttl=True)
+                    logger.debug("Extended lock", key=self.key)
+            except LockError:
+                logger.warning("Failed to extend lock", key=self.key)
+                break
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error("Error extending lock", key=self.key, error=str(e))
+                break
+
+    async def __aenter__(self):
+        self._redis = await get_async_redis_client()
+        self._lock = self._redis.lock(
+            self.key,
+            timeout=self.timeout,
+            blocking=self.blocking,
+            blocking_timeout=self.blocking_timeout,
+        )
+
+        self._acquired = await self._lock.acquire()
+
+        if not self._acquired:
+            if self.skip_if_locked:
+                logger.warning(
+                    "Lock already acquired by another process, skipping", key=self.key
+                )
+                return self
+            else:
+                raise LockError(f"Failed to acquire lock: {self.key}")
+
+        self._extend_task = asyncio.create_task(self._extend_lock_periodically())
+        logger.info("Acquired lock", key=self.key)
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        if self._extend_task:
+            self._extend_task.cancel()
+            try:
+                await self._extend_task
+            except asyncio.CancelledError:
+                pass
+
+        if self._acquired and self._lock:
+            try:
+                await self._lock.release()
+                logger.info("Released lock", key=self.key)
+            except LockError:
+                logger.debug("Lock already released or expired", key=self.key)
+
+        if self._redis:
+            await self._redis.aclose()
+
+    @property
+    def acquired(self) -> bool:
+        return self._acquired

server/reflector/schemas/platform.py

@@ -0,0 +1,5 @@
+from typing import Literal
+
+Platform = Literal["whereby", "daily"]
+WHEREBY_PLATFORM: Platform = "whereby"
+DAILY_PLATFORM: Platform = "daily"

server/reflector/services/ics_sync.py

@@ -0,0 +1,408 @@
+"""
+ICS Calendar Synchronization Service
+
+This module provides services for fetching, parsing, and synchronizing ICS (iCalendar)
+calendar feeds with room booking data in the database.
+
+Key Components:
+- ICSFetchService: Handles HTTP fetching and parsing of ICS calendar data
+- ICSSyncService: Manages the synchronization process between ICS feeds and database
+
+Example Usage:
+    # Sync a room's calendar
+    room = Room(id="room1", name="conference-room", ics_url="https://cal.example.com/room.ics")
+    result = await ics_sync_service.sync_room_calendar(room)
+
+    # Result structure:
+    {
+        "status": "success",  # success|unchanged|error|skipped
+        "hash": "abc123...",  # MD5 hash of ICS content
+        "events_found": 5,    # Events matching this room
+        "total_events": 12,   # Total events in calendar within time window
+        "events_created": 2,  # New events added to database
+        "events_updated": 3,  # Existing events modified
+        "events_deleted": 1   # Events soft-deleted (no longer in calendar)
+    }
+
+Event Matching:
+    Events are matched to rooms by checking if the room's full URL appears in the
+    event's LOCATION or DESCRIPTION fields. Only events within a 25-hour window
+    (1 hour ago to 24 hours from now) are processed.
+
+Input: ICS calendar URL (e.g., "https://calendar.google.com/calendar/ical/...")
+Output: EventData objects with structured calendar information:
+    {
+        "ics_uid": "event123@google.com",
+        "title": "Team Meeting",
+        "description": "Weekly sync meeting",
+        "location": "https://meet.company.com/conference-room",
+        "start_time": datetime(2024, 1, 15, 14, 0, tzinfo=UTC),
+        "end_time": datetime(2024, 1, 15, 15, 0, tzinfo=UTC),
+        "attendees": [
+            {"email": "user@company.com", "name": "John Doe", "role": "ORGANIZER"},
+            {"email": "attendee@company.com", "name": "Jane Smith", "status": "ACCEPTED"}
+        ],
+        "ics_raw_data": "BEGIN:VEVENT\nUID:event123@google.com\n..."
+    }
+"""
+
+import hashlib
+from datetime import date, datetime, timedelta, timezone
+from enum import Enum
+from typing import TypedDict
+
+import httpx
+import pytz
+import structlog
+from icalendar import Calendar, Event
+
+from reflector.db.calendar_events import CalendarEvent, calendar_events_controller
+from reflector.db.rooms import Room, rooms_controller
+from reflector.redis_cache import RedisAsyncLock
+from reflector.settings import settings
+
+logger = structlog.get_logger()
+
+EVENT_WINDOW_DELTA_START = timedelta(hours=-1)
+EVENT_WINDOW_DELTA_END = timedelta(hours=24)
+
+
+class SyncStatus(str, Enum):
+    SUCCESS = "success"
+    UNCHANGED = "unchanged"
+    ERROR = "error"
+    SKIPPED = "skipped"
+
+
+class AttendeeData(TypedDict, total=False):
+    email: str | None
+    name: str | None
+    status: str | None
+    role: str | None
+
+
+class EventData(TypedDict):
+    ics_uid: str
+    title: str | None
+    description: str | None
+    location: str | None
+    start_time: datetime
+    end_time: datetime
+    attendees: list[AttendeeData]
+    ics_raw_data: str
+
+
+class SyncStats(TypedDict):
+    events_created: int
+    events_updated: int
+    events_deleted: int
+
+
+class SyncResultBase(TypedDict):
+    status: SyncStatus
+
+
+class SyncResult(SyncResultBase, total=False):
+    hash: str | None
+    events_found: int
+    total_events: int
+    events_created: int
+    events_updated: int
+    events_deleted: int
+    error: str | None
+    reason: str | None
+
+
+class ICSFetchService:
+    def __init__(self):
+        self.client = httpx.AsyncClient(
+            timeout=30.0, headers={"User-Agent": "Reflector/1.0"}
+        )
+
+    async def fetch_ics(self, url: str) -> str:
+        response = await self.client.get(url)
+        response.raise_for_status()
+
+        return response.text
+
+    def parse_ics(self, ics_content: str) -> Calendar:
+        return Calendar.from_ical(ics_content)
+
+    def extract_room_events(
+        self, calendar: Calendar, room_name: str, room_url: str
+    ) -> tuple[list[EventData], int]:
+        events = []
+        total_events = 0
+        now = datetime.now(timezone.utc)
+        window_start = now + EVENT_WINDOW_DELTA_START
+        window_end = now + EVENT_WINDOW_DELTA_END
+
+        for component in calendar.walk():
+            if component.name != "VEVENT":
+                continue
+
+            status = component.get("STATUS", "").upper()
+            if status == "CANCELLED":
+                continue
+
+            # Count total non-cancelled events in the time window
+            event_data = self._parse_event(component)
+            if event_data and window_start <= event_data["start_time"] <= window_end:
+                total_events += 1
+
+                # Check if event matches this room
+                if self._event_matches_room(component, room_name, room_url):
+                    events.append(event_data)
+
+        return events, total_events
+
+    def _event_matches_room(self, event: Event, room_name: str, room_url: str) -> bool:
+        location = str(event.get("LOCATION", ""))
+        description = str(event.get("DESCRIPTION", ""))
+
+        # Only match full room URL
+        # XXX leaved here as a patterns, to later be extended with tinyurl or such too
+        patterns = [
+            room_url,
+        ]
+
+        # Check location and description for patterns
+        text_to_check = f"{location} {description}".lower()
+        for pattern in patterns:
+            if pattern.lower() in text_to_check:
+                return True
+
+        return False
+
+    def _parse_event(self, event: Event) -> EventData | None:
+        uid = str(event.get("UID", ""))
+        summary = str(event.get("SUMMARY", ""))
+        description = str(event.get("DESCRIPTION", ""))
+        location = str(event.get("LOCATION", ""))
+        dtstart = event.get("DTSTART")
+        dtend = event.get("DTEND")
+
+        if not dtstart:
+            return None
+
+        # Convert fields
+        start_time = self._normalize_datetime(
+            dtstart.dt if hasattr(dtstart, "dt") else dtstart
+        )
+        end_time = (
+            self._normalize_datetime(dtend.dt if hasattr(dtend, "dt") else dtend)
+            if dtend
+            else start_time + timedelta(hours=1)
+        )
+        attendees = self._parse_attendees(event)
+
+        # Get raw event data for storage
+        raw_data = event.to_ical().decode("utf-8")
+
+        return {
+            "ics_uid": uid,
+            "title": summary,
+            "description": description,
+            "location": location,
+            "start_time": start_time,
+            "end_time": end_time,
+            "attendees": attendees,
+            "ics_raw_data": raw_data,
+        }
+
+    def _normalize_datetime(self, dt) -> datetime:
+        # Ensure datetime is with timezone, if not, assume UTC
+        if isinstance(dt, date) and not isinstance(dt, datetime):
+            dt = datetime.combine(dt, datetime.min.time())
+            dt = pytz.UTC.localize(dt)
+        elif isinstance(dt, datetime):
+            if dt.tzinfo is None:
+                dt = pytz.UTC.localize(dt)
+            else:
+                dt = dt.astimezone(pytz.UTC)
+
+        return dt
+
+    def _parse_attendees(self, event: Event) -> list[AttendeeData]:
+        # Extracts attendee information from both ATTENDEE and ORGANIZER properties.
+        # Handles malformed comma-separated email addresses in single ATTENDEE fields
+        # by splitting them into separate attendee entries. Returns a list of attendee
+        # data including email, name, status, and role information.
+        final_attendees = []
+
+        attendees = event.get("ATTENDEE", [])
+        if not isinstance(attendees, list):
+            attendees = [attendees]
+        for att in attendees:
+            email_str = str(att).replace("mailto:", "") if att else None
+
+            # Handle malformed comma-separated email addresses in a single ATTENDEE field
+            if email_str and "," in email_str:
+                # Split comma-separated emails and create separate attendee entries
+                email_parts = [email.strip() for email in email_str.split(",")]
+                for email in email_parts:
+                    if email and "@" in email:
+                        clean_email = email.replace("MAILTO:", "").replace(
+                            "mailto:", ""
+                        )
+                        att_data: AttendeeData = {
+                            "email": clean_email,
+                            "name": att.params.get("CN")
+                            if hasattr(att, "params") and email == email_parts[0]
+                            else None,
+                            "status": att.params.get("PARTSTAT")
+                            if hasattr(att, "params") and email == email_parts[0]
+                            else None,
+                            "role": att.params.get("ROLE")
+                            if hasattr(att, "params") and email == email_parts[0]
+                            else None,
+                        }
+                        final_attendees.append(att_data)
+            else:
+                # Normal single attendee
+                att_data: AttendeeData = {
+                    "email": email_str,
+                    "name": att.params.get("CN") if hasattr(att, "params") else None,
+                    "status": att.params.get("PARTSTAT")
+                    if hasattr(att, "params")
+                    else None,
+                    "role": att.params.get("ROLE") if hasattr(att, "params") else None,
+                }
+                final_attendees.append(att_data)
+
+        # Add organizer
+        organizer = event.get("ORGANIZER")
+        if organizer:
+            org_email = (
+                str(organizer).replace("mailto:", "").replace("MAILTO:", "")
+                if organizer
+                else None
+            )
+            org_data: AttendeeData = {
+                "email": org_email,
+                "name": organizer.params.get("CN")
+                if hasattr(organizer, "params")
+                else None,
+                "role": "ORGANIZER",
+            }
+            final_attendees.append(org_data)
+
+        return final_attendees
+
+
+class ICSSyncService:
+    def __init__(self):
+        self.fetch_service = ICSFetchService()
+
+    async def sync_room_calendar(self, room: Room) -> SyncResult:
+        async with RedisAsyncLock(
+            f"ics_sync_room:{room.id}", skip_if_locked=True
+        ) as lock:
+            if not lock.acquired:
+                logger.warning("ICS sync already in progress for room", room_id=room.id)
+                return {
+                    "status": SyncStatus.SKIPPED,
+                    "reason": "Sync already in progress",
+                }
+
+            return await self._sync_room_calendar(room)
+
+    async def _sync_room_calendar(self, room: Room) -> SyncResult:
+        if not room.ics_enabled or not room.ics_url:
+            return {"status": SyncStatus.SKIPPED, "reason": "ICS not configured"}
+
+        try:
+            if not self._should_sync(room):
+                return {"status": SyncStatus.SKIPPED, "reason": "Not time to sync yet"}
+
+            ics_content = await self.fetch_service.fetch_ics(room.ics_url)
+            calendar = self.fetch_service.parse_ics(ics_content)
+
+            content_hash = hashlib.md5(ics_content.encode()).hexdigest()
+            if room.ics_last_etag == content_hash:
+                logger.info("No changes in ICS for room", room_id=room.id)
+                room_url = f"{settings.UI_BASE_URL}/{room.name}"
+                events, total_events = self.fetch_service.extract_room_events(
+                    calendar, room.name, room_url
+                )
+                return {
+                    "status": SyncStatus.UNCHANGED,
+                    "hash": content_hash,
+                    "events_found": len(events),
+                    "total_events": total_events,
+                    "events_created": 0,
+                    "events_updated": 0,
+                    "events_deleted": 0,
+                }
+
+            # Extract matching events
+            room_url = f"{settings.UI_BASE_URL}/{room.name}"
+            events, total_events = self.fetch_service.extract_room_events(
+                calendar, room.name, room_url
+            )
+            sync_result = await self._sync_events_to_database(room.id, events)
+
+            # Update room sync metadata
+            await rooms_controller.update(
+                room,
+                {
+                    "ics_last_sync": datetime.now(timezone.utc),
+                    "ics_last_etag": content_hash,
+                },
+                mutate=False,
+            )
+
+            return {
+                "status": SyncStatus.SUCCESS,
+                "hash": content_hash,
+                "events_found": len(events),
+                "total_events": total_events,
+                **sync_result,
+            }
+
+        except Exception as e:
+            logger.error("Failed to sync ICS for room", room_id=room.id, error=str(e))
+            return {"status": SyncStatus.ERROR, "error": str(e)}
+
+    def _should_sync(self, room: Room) -> bool:
+        if not room.ics_last_sync:
+            return True
+
+        time_since_sync = datetime.now(timezone.utc) - room.ics_last_sync
+        return time_since_sync.total_seconds() >= room.ics_fetch_interval
+
+    async def _sync_events_to_database(
+        self, room_id: str, events: list[EventData]
+    ) -> SyncStats:
+        created = 0
+        updated = 0
+
+        current_ics_uids = []
+
+        for event_data in events:
+            calendar_event = CalendarEvent(room_id=room_id, **event_data)
+            existing = await calendar_events_controller.get_by_ics_uid(
+                room_id, event_data["ics_uid"]
+            )
+
+            if existing:
+                updated += 1
+            else:
+                created += 1
+
+            await calendar_events_controller.upsert(calendar_event)
+            current_ics_uids.append(event_data["ics_uid"])
+
+        # Soft delete events that are no longer in calendar
+        deleted = await calendar_events_controller.soft_delete_missing(
+            room_id, current_ics_uids
+        )
+
+        return {
+            "events_created": created,
+            "events_updated": updated,
+            "events_deleted": deleted,
+        }
+
+
+ics_sync_service = ICSSyncService()

server/reflector/settings.py

@@ -1,6 +1,9 @@
 from pydantic.types import PositiveInt
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from reflector.schemas.platform import WHEREBY_PLATFORM, Platform
+from reflector.utils.string import NonEmptyString
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
@@ -45,14 +48,17 @@ class Settings(BaseSettings):
     TRANSCRIPT_STORAGE_AWS_ACCESS_KEY_ID: str | None = None
     TRANSCRIPT_STORAGE_AWS_SECRET_ACCESS_KEY: str | None = None
 
-    # Recording storage
-    RECORDING_STORAGE_BACKEND: str | None = None
+    # Platform-specific recording storage (follows {PREFIX}_STORAGE_AWS_{CREDENTIAL} pattern)
+    # Whereby storage configuration
+    WHEREBY_STORAGE_AWS_BUCKET_NAME: str | None = None
+    WHEREBY_STORAGE_AWS_REGION: str | None = None
+    WHEREBY_STORAGE_AWS_ACCESS_KEY_ID: str | None = None
+    WHEREBY_STORAGE_AWS_SECRET_ACCESS_KEY: str | None = None
 
-    # Recording storage configuration for AWS
-    RECORDING_STORAGE_AWS_BUCKET_NAME: str = "recording-bucket"
-    RECORDING_STORAGE_AWS_REGION: str = "us-east-1"
-    RECORDING_STORAGE_AWS_ACCESS_KEY_ID: str | None = None
-    RECORDING_STORAGE_AWS_SECRET_ACCESS_KEY: str | None = None
+    # Daily.co storage configuration
+    DAILYCO_STORAGE_AWS_BUCKET_NAME: str | None = None
+    DAILYCO_STORAGE_AWS_REGION: str | None = None
+    DAILYCO_STORAGE_AWS_ROLE_ARN: str | None = None
 
     # Translate into the target language
     TRANSLATION_BACKEND: str = "passthrough"
@@ -120,13 +126,22 @@ class Settings(BaseSettings):
 
     # Whereby integration
     WHEREBY_API_URL: str = "https://api.whereby.dev/v1"
-    WHEREBY_API_KEY: str | None = None
+    WHEREBY_API_KEY: NonEmptyString | None = None
     WHEREBY_WEBHOOK_SECRET: str | None = None
-    AWS_WHEREBY_ACCESS_KEY_ID: str | None = None
-    AWS_WHEREBY_ACCESS_KEY_SECRET: str | None = None
     AWS_PROCESS_RECORDING_QUEUE_URL: str | None = None
     SQS_POLLING_TIMEOUT_SECONDS: int = 60
 
+    # Daily.co integration
+    DAILY_API_KEY: str | None = None
+    DAILY_WEBHOOK_SECRET: str | None = None
+    DAILY_SUBDOMAIN: str | None = None
+    DAILY_WEBHOOK_UUID: str | None = (
+        None  # Webhook UUID for this environment. Not used by production code
+    )
+
+    # Platform Configuration
+    DEFAULT_VIDEO_PLATFORM: Platform = WHEREBY_PLATFORM
+
     # Zulip integration
     ZULIP_REALM: str | None = None
     ZULIP_API_KEY: str | None = None

server/reflector/storage/__init__.py

@@ -3,6 +3,13 @@ from reflector.settings import settings
 
 
 def get_transcripts_storage() -> Storage:
+    """
+    Get storage for processed transcript files (master credentials).
+
+    Also use this for ALL our file operations with bucket override:
+        master = get_transcripts_storage()
+        master.delete_file(key, bucket=recording.bucket_name)
+    """
     assert settings.TRANSCRIPT_STORAGE_BACKEND
     return Storage.get_instance(
         name=settings.TRANSCRIPT_STORAGE_BACKEND,
@@ -10,8 +17,53 @@ def get_transcripts_storage() -> Storage:
     )
 
 
-def get_recordings_storage() -> Storage:
+def get_whereby_storage() -> Storage:
+    """
+    Get storage config for Whereby (for passing to Whereby API).
+
+    Usage:
+        whereby_storage = get_whereby_storage()
+        key_id, secret = whereby_storage.key_credentials
+        whereby_api.create_meeting(
+            bucket=whereby_storage.bucket_name,
+            access_key_id=key_id,
+            secret=secret,
+        )
+
+    Do NOT use for our file operations - use get_transcripts_storage() instead.
+    """
+    if not settings.WHEREBY_STORAGE_AWS_BUCKET_NAME:
+        raise ValueError(
+            "WHEREBY_STORAGE_AWS_BUCKET_NAME required for Whereby with AWS storage"
+        )
+
     return Storage.get_instance(
-        name=settings.RECORDING_STORAGE_BACKEND,
-        settings_prefix="RECORDING_STORAGE_",
+        name="aws",
+        settings_prefix="WHEREBY_STORAGE_",
+    )
+
+
+def get_dailyco_storage() -> Storage:
+    """
+    Get storage config for Daily.co (for passing to Daily API).
+
+    Usage:
+        daily_storage = get_dailyco_storage()
+        daily_api.create_meeting(
+            bucket=daily_storage.bucket_name,
+            region=daily_storage.region,
+            role_arn=daily_storage.role_credential,
+        )
+
+    Do NOT use for our file operations - use get_transcripts_storage() instead.
+    """
+    # Fail fast if platform-specific config missing
+    if not settings.DAILYCO_STORAGE_AWS_BUCKET_NAME:
+        raise ValueError(
+            "DAILYCO_STORAGE_AWS_BUCKET_NAME required for Daily.co with AWS storage"
+        )
+
+    return Storage.get_instance(
+        name="aws",
+        settings_prefix="DAILYCO_STORAGE_",
     )

server/reflector/storage/base.py

@@ -1,10 +1,23 @@
 import importlib
+from typing import BinaryIO, Union
 
 from pydantic import BaseModel
 
 from reflector.settings import settings
 
 
+class StorageError(Exception):
+    """Base exception for storage operations."""
+
+    pass
+
+
+class StoragePermissionError(StorageError):
+    """Exception raised when storage operation fails due to permission issues."""
+
+    pass
+
+
 class FileResult(BaseModel):
     filename: str
     url: str
@@ -36,26 +49,113 @@ class Storage:
 
         return cls._registry[name](**config)
 
-    async def put_file(self, filename: str, data: bytes) -> FileResult:
-        return await self._put_file(filename, data)
-
-    async def _put_file(self, filename: str, data: bytes) -> FileResult:
+    # Credential properties for API passthrough
+    @property
+    def bucket_name(self) -> str:
+        """Default bucket name for this storage instance."""
         raise NotImplementedError
 
-    async def delete_file(self, filename: str):
-        return await self._delete_file(filename)
-
-    async def _delete_file(self, filename: str):
+    @property
+    def region(self) -> str:
+        """AWS region for this storage instance."""
         raise NotImplementedError
 
-    async def get_file_url(self, filename: str) -> str:
-        return await self._get_file_url(filename)
+    @property
+    def access_key_id(self) -> str | None:
+        """AWS access key ID (None for role-based auth). Prefer key_credentials property."""
+        return None
 
-    async def _get_file_url(self, filename: str) -> str:
+    @property
+    def secret_access_key(self) -> str | None:
+        """AWS secret access key (None for role-based auth). Prefer key_credentials property."""
+        return None
+
+    @property
+    def role_arn(self) -> str | None:
+        """AWS IAM role ARN for role-based auth (None for key-based auth). Prefer role_credential property."""
+        return None
+
+    @property
+    def key_credentials(self) -> tuple[str, str]:
+        """
+        Get (access_key_id, secret_access_key) for key-based auth.
+        Raises ValueError if storage uses IAM role instead.
+        """
         raise NotImplementedError
 
-    async def get_file(self, filename: str):
-        return await self._get_file(filename)
-
-    async def _get_file(self, filename: str):
+    @property
+    def role_credential(self) -> str:
+        """
+        Get IAM role ARN for role-based auth.
+        Raises ValueError if storage uses access keys instead.
+        """
+        raise NotImplementedError
+
+    async def put_file(
+        self, filename: str, data: Union[bytes, BinaryIO], *, bucket: str | None = None
+    ) -> FileResult:
+        """Upload data. bucket: override instance default if provided."""
+        return await self._put_file(filename, data, bucket=bucket)
+
+    async def _put_file(
+        self, filename: str, data: Union[bytes, BinaryIO], *, bucket: str | None = None
+    ) -> FileResult:
+        raise NotImplementedError
+
+    async def delete_file(self, filename: str, *, bucket: str | None = None):
+        """Delete file. bucket: override instance default if provided."""
+        return await self._delete_file(filename, bucket=bucket)
+
+    async def _delete_file(self, filename: str, *, bucket: str | None = None):
+        raise NotImplementedError
+
+    async def get_file_url(
+        self,
+        filename: str,
+        operation: str = "get_object",
+        expires_in: int = 3600,
+        *,
+        bucket: str | None = None,
+    ) -> str:
+        """Generate presigned URL. bucket: override instance default if provided."""
+        return await self._get_file_url(filename, operation, expires_in, bucket=bucket)
+
+    async def _get_file_url(
+        self,
+        filename: str,
+        operation: str = "get_object",
+        expires_in: int = 3600,
+        *,
+        bucket: str | None = None,
+    ) -> str:
+        raise NotImplementedError
+
+    async def get_file(self, filename: str, *, bucket: str | None = None):
+        """Download file. bucket: override instance default if provided."""
+        return await self._get_file(filename, bucket=bucket)
+
+    async def _get_file(self, filename: str, *, bucket: str | None = None):
+        raise NotImplementedError
+
+    async def list_objects(
+        self, prefix: str = "", *, bucket: str | None = None
+    ) -> list[str]:
+        """List object keys. bucket: override instance default if provided."""
+        return await self._list_objects(prefix, bucket=bucket)
+
+    async def _list_objects(
+        self, prefix: str = "", *, bucket: str | None = None
+    ) -> list[str]:
+        raise NotImplementedError
+
+    async def stream_to_fileobj(
+        self, filename: str, fileobj: BinaryIO, *, bucket: str | None = None
+    ):
+        """Stream file directly to file object without loading into memory.
+        bucket: override instance default if provided."""
+        return await self._stream_to_fileobj(filename, fileobj, bucket=bucket)
+
+    async def _stream_to_fileobj(
+        self, filename: str, fileobj: BinaryIO, *, bucket: str | None = None
+    ):
         raise NotImplementedError

server/reflector/storage/storage_aws.py

@@ -1,79 +1,236 @@
+from functools import wraps
+from typing import BinaryIO, Union
+
 import aioboto3
+from botocore.config import Config
+from botocore.exceptions import ClientError
 
 from reflector.logger import logger
-from reflector.storage.base import FileResult, Storage
+from reflector.storage.base import FileResult, Storage, StoragePermissionError
+
+
+def handle_s3_client_errors(operation_name: str):
+    """Decorator to handle S3 ClientError with bucket-aware messaging.
+
+    Args:
+        operation_name: Human-readable operation name for error messages (e.g., "upload", "delete")
+    """
+
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(self, *args, **kwargs):
+            bucket = kwargs.get("bucket")
+            try:
+                return await func(self, *args, **kwargs)
+            except ClientError as e:
+                error_code = e.response.get("Error", {}).get("Code")
+                if error_code in ("AccessDenied", "NoSuchBucket"):
+                    actual_bucket = bucket or self._bucket_name
+                    bucket_context = (
+                        f"overridden bucket '{actual_bucket}'"
+                        if bucket
+                        else f"default bucket '{actual_bucket}'"
+                    )
+                    raise StoragePermissionError(
+                        f"S3 {operation_name} failed for {bucket_context}: {error_code}. "
+                        f"Check TRANSCRIPT_STORAGE_AWS_* credentials have permission."
+                    ) from e
+                raise
+
+        return wrapper
+
+    return decorator
 
 
 class AwsStorage(Storage):
+    """AWS S3 storage with bucket override for multi-platform recording architecture.
+    Master credentials access all buckets via optional bucket parameter in operations."""
+
     def __init__(
         self,
-        aws_access_key_id: str,
-        aws_secret_access_key: str,
         aws_bucket_name: str,
         aws_region: str,
+        aws_access_key_id: str | None = None,
+        aws_secret_access_key: str | None = None,
+        aws_role_arn: str | None = None,
     ):
-        if not aws_access_key_id:
-            raise ValueError("Storage `aws_storage` require `aws_access_key_id`")
-        if not aws_secret_access_key:
-            raise ValueError("Storage `aws_storage` require `aws_secret_access_key`")
         if not aws_bucket_name:
             raise ValueError("Storage `aws_storage` require `aws_bucket_name`")
         if not aws_region:
             raise ValueError("Storage `aws_storage` require `aws_region`")
+        if not aws_access_key_id and not aws_role_arn:
+            raise ValueError(
+                "Storage `aws_storage` require either `aws_access_key_id` or `aws_role_arn`"
+            )
+        if aws_role_arn and (aws_access_key_id or aws_secret_access_key):
+            raise ValueError(
+                "Storage `aws_storage` cannot use both `aws_role_arn` and access keys"
+            )
 
         super().__init__()
-        self.aws_bucket_name = aws_bucket_name
+        self._bucket_name = aws_bucket_name
+        self._region = aws_region
+        self._access_key_id = aws_access_key_id
+        self._secret_access_key = aws_secret_access_key
+        self._role_arn = aws_role_arn
+
         self.aws_folder = ""
         if "/" in aws_bucket_name:
-            self.aws_bucket_name, self.aws_folder = aws_bucket_name.split("/", 1)
+            self._bucket_name, self.aws_folder = aws_bucket_name.split("/", 1)
+        self.boto_config = Config(retries={"max_attempts": 3, "mode": "adaptive"})
         self.session = aioboto3.Session(
             aws_access_key_id=aws_access_key_id,
             aws_secret_access_key=aws_secret_access_key,
             region_name=aws_region,
         )
-        self.base_url = f"https://{aws_bucket_name}.s3.amazonaws.com/"
+        self.base_url = f"https://{self._bucket_name}.s3.amazonaws.com/"
 
-    async def _put_file(self, filename: str, data: bytes) -> FileResult:
-        bucket = self.aws_bucket_name
-        folder = self.aws_folder
-        logger.info(f"Uploading {filename} to S3 {bucket}/{folder}")
-        s3filename = f"{folder}/{filename}" if folder else filename
-        async with self.session.client("s3") as client:
-            await client.put_object(
-                Bucket=bucket,
-                Key=s3filename,
-                Body=data,
+    # Implement credential properties
+    @property
+    def bucket_name(self) -> str:
+        return self._bucket_name
+
+    @property
+    def region(self) -> str:
+        return self._region
+
+    @property
+    def access_key_id(self) -> str | None:
+        return self._access_key_id
+
+    @property
+    def secret_access_key(self) -> str | None:
+        return self._secret_access_key
+
+    @property
+    def role_arn(self) -> str | None:
+        return self._role_arn
+
+    @property
+    def key_credentials(self) -> tuple[str, str]:
+        """Get (access_key_id, secret_access_key) for key-based auth."""
+        if self._role_arn:
+            raise ValueError(
+                "Storage uses IAM role authentication. "
+                "Use role_credential property instead of key_credentials."
             )
+        if not self._access_key_id or not self._secret_access_key:
+            raise ValueError("Storage access key credentials not configured")
+        return (self._access_key_id, self._secret_access_key)
 
-    async def _get_file_url(self, filename: str) -> FileResult:
-        bucket = self.aws_bucket_name
+    @property
+    def role_credential(self) -> str:
+        """Get IAM role ARN for role-based auth."""
+        if self._access_key_id or self._secret_access_key:
+            raise ValueError(
+                "Storage uses access key authentication. "
+                "Use key_credentials property instead of role_credential."
+            )
+        if not self._role_arn:
+            raise ValueError("Storage IAM role ARN not configured")
+        return self._role_arn
+
+    @handle_s3_client_errors("upload")
+    async def _put_file(
+        self, filename: str, data: Union[bytes, BinaryIO], *, bucket: str | None = None
+    ) -> FileResult:
+        actual_bucket = bucket or self._bucket_name
         folder = self.aws_folder
         s3filename = f"{folder}/{filename}" if folder else filename
-        async with self.session.client("s3") as client:
+        logger.info(f"Uploading {filename} to S3 {actual_bucket}/{folder}")
+
+        async with self.session.client("s3", config=self.boto_config) as client:
+            if isinstance(data, bytes):
+                await client.put_object(Bucket=actual_bucket, Key=s3filename, Body=data)
+            else:
+                # boto3 reads file-like object in chunks
+                # avoids creating extra memory copy vs bytes.getvalue() approach
+                await client.upload_fileobj(data, Bucket=actual_bucket, Key=s3filename)
+
+        url = await self._get_file_url(filename, bucket=bucket)
+        return FileResult(filename=filename, url=url)
+
+    @handle_s3_client_errors("presign")
+    async def _get_file_url(
+        self,
+        filename: str,
+        operation: str = "get_object",
+        expires_in: int = 3600,
+        *,
+        bucket: str | None = None,
+    ) -> str:
+        actual_bucket = bucket or self._bucket_name
+        folder = self.aws_folder
+        s3filename = f"{folder}/{filename}" if folder else filename
+        async with self.session.client("s3", config=self.boto_config) as client:
             presigned_url = await client.generate_presigned_url(
-                "get_object",
-                Params={"Bucket": bucket, "Key": s3filename},
-                ExpiresIn=3600,
+                operation,
+                Params={"Bucket": actual_bucket, "Key": s3filename},
+                ExpiresIn=expires_in,
             )
 
             return presigned_url
 
-    async def _delete_file(self, filename: str):
-        bucket = self.aws_bucket_name
+    @handle_s3_client_errors("delete")
+    async def _delete_file(self, filename: str, *, bucket: str | None = None):
+        actual_bucket = bucket or self._bucket_name
         folder = self.aws_folder
-        logger.info(f"Deleting {filename} from S3 {bucket}/{folder}")
+        logger.info(f"Deleting {filename} from S3 {actual_bucket}/{folder}")
         s3filename = f"{folder}/{filename}" if folder else filename
-        async with self.session.client("s3") as client:
-            await client.delete_object(Bucket=bucket, Key=s3filename)
+        async with self.session.client("s3", config=self.boto_config) as client:
+            await client.delete_object(Bucket=actual_bucket, Key=s3filename)
 
-    async def _get_file(self, filename: str):
-        bucket = self.aws_bucket_name
+    @handle_s3_client_errors("download")
+    async def _get_file(self, filename: str, *, bucket: str | None = None):
+        actual_bucket = bucket or self._bucket_name
         folder = self.aws_folder
-        logger.info(f"Downloading {filename} from S3 {bucket}/{folder}")
+        logger.info(f"Downloading {filename} from S3 {actual_bucket}/{folder}")
         s3filename = f"{folder}/{filename}" if folder else filename
-        async with self.session.client("s3") as client:
-            response = await client.get_object(Bucket=bucket, Key=s3filename)
+        async with self.session.client("s3", config=self.boto_config) as client:
+            response = await client.get_object(Bucket=actual_bucket, Key=s3filename)
             return await response["Body"].read()
 
+    @handle_s3_client_errors("list_objects")
+    async def _list_objects(
+        self, prefix: str = "", *, bucket: str | None = None
+    ) -> list[str]:
+        actual_bucket = bucket or self._bucket_name
+        folder = self.aws_folder
+        # Combine folder and prefix
+        s3prefix = f"{folder}/{prefix}" if folder else prefix
+        logger.info(f"Listing objects from S3 {actual_bucket} with prefix '{s3prefix}'")
+
+        keys = []
+        async with self.session.client("s3", config=self.boto_config) as client:
+            paginator = client.get_paginator("list_objects_v2")
+            async for page in paginator.paginate(Bucket=actual_bucket, Prefix=s3prefix):
+                if "Contents" in page:
+                    for obj in page["Contents"]:
+                        # Strip folder prefix from keys if present
+                        key = obj["Key"]
+                        if folder:
+                            if key.startswith(f"{folder}/"):
+                                key = key[len(folder) + 1 :]
+                            elif key == folder:
+                                # Skip folder marker itself
+                                continue
+                        keys.append(key)
+
+        return keys
+
+    @handle_s3_client_errors("stream")
+    async def _stream_to_fileobj(
+        self, filename: str, fileobj: BinaryIO, *, bucket: str | None = None
+    ):
+        """Stream file from S3 directly to file object without loading into memory."""
+        actual_bucket = bucket or self._bucket_name
+        folder = self.aws_folder
+        logger.info(f"Streaming {filename} from S3 {actual_bucket}/{folder}")
+        s3filename = f"{folder}/{filename}" if folder else filename
+        async with self.session.client("s3", config=self.boto_config) as client:
+            await client.download_fileobj(
+                Bucket=actual_bucket, Key=s3filename, Fileobj=fileobj
+            )
+
 
 Storage.register("aws", AwsStorage)

server/reflector/utils/daily.py

@@ -0,0 +1,26 @@
+from reflector.utils.string import NonEmptyString
+
+DailyRoomName = str
+
+
+def extract_base_room_name(daily_room_name: DailyRoomName) -> NonEmptyString:
+    """
+    Extract base room name from Daily.co timestamped room name.
+
+    Daily.co creates rooms with timestamp suffix: {base_name}-YYYYMMDDHHMMSS
+    This function removes the timestamp to get the original room name.
+
+    Examples:
+        "daily-20251020193458" → "daily"
+        "daily-2-20251020193458" → "daily-2"
+        "my-room-name-20251020193458" → "my-room-name"
+
+    Args:
+        daily_room_name: Full Daily.co room name with optional timestamp
+
+    Returns:
+        Base room name without timestamp suffix
+    """
+    base_name = daily_room_name.rsplit("-", 1)[0]
+    assert base_name, f"Extracted base name is empty from: {daily_room_name}"
+    return base_name

server/reflector/utils/datetime.py

@@ -0,0 +1,9 @@
+from datetime import datetime, timezone
+
+
+def parse_datetime_with_timezone(iso_string: str) -> datetime:
+    """Parse ISO datetime string and ensure timezone awareness (defaults to UTC if naive)."""
+    dt = datetime.fromisoformat(iso_string)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt

server/reflector/utils/string.py

@@ -1,4 +1,4 @@
-from typing import Annotated
+from typing import Annotated, TypeVar
 
 from pydantic import Field, TypeAdapter, constr
 
@@ -10,11 +10,23 @@ NonEmptyString = Annotated[
 non_empty_string_adapter = TypeAdapter(NonEmptyString)
 
 
-def parse_non_empty_string(s: str) -> NonEmptyString:
-    return non_empty_string_adapter.validate_python(s)
+def parse_non_empty_string(s: str, error: str | None = None) -> NonEmptyString:
+    try:
+        return non_empty_string_adapter.validate_python(s)
+    except Exception as e:
+        raise ValueError(f"{e}: {error}" if error else e) from e
 
 
 def try_parse_non_empty_string(s: str) -> NonEmptyString | None:
     if not s:
         return None
     return parse_non_empty_string(s)
+
+
+T = TypeVar("T", bound=str)
+
+
+def assert_equal[T](s1: T, s2: T) -> T:
+    if s1 != s2:
+        raise ValueError(f"assert_equal: {s1} != {s2}")
+    return s1

server/reflector/utils/url.py

@@ -0,0 +1,37 @@
+"""URL manipulation utilities."""
+
+from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
+
+
+def add_query_param(url: str, key: str, value: str) -> str:
+    """
+    Add or update a query parameter in a URL.
+
+    Properly handles URLs with or without existing query parameters,
+    preserving fragments and encoding special characters.
+
+    Args:
+        url: The URL to modify
+        key: The query parameter name
+        value: The query parameter value
+
+    Returns:
+        The URL with the query parameter added or updated
+
+    Examples:
+        >>> add_query_param("https://example.com/room", "t", "token123")
+        'https://example.com/room?t=token123'
+
+        >>> add_query_param("https://example.com/room?existing=param", "t", "token123")
+        'https://example.com/room?existing=param&t=token123'
+    """
+    parsed = urlparse(url)
+
+    query_params = parse_qs(parsed.query, keep_blank_values=True)
+
+    query_params[key] = [value]
+
+    new_query = urlencode(query_params, doseq=True)
+
+    new_parsed = parsed._replace(query=new_query)
+    return urlunparse(new_parsed)

server/reflector/video_platforms/__init__.py

@@ -0,0 +1,11 @@
+from .base import VideoPlatformClient
+from .models import MeetingData, VideoPlatformConfig
+from .registry import get_platform_client, register_platform
+
+__all__ = [
+    "VideoPlatformClient",
+    "VideoPlatformConfig",
+    "MeetingData",
+    "get_platform_client",
+    "register_platform",
+]

server/reflector/video_platforms/base.py

@@ -0,0 +1,54 @@
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+from ..schemas.platform import Platform
+from ..utils.string import NonEmptyString
+from .models import MeetingData, VideoPlatformConfig
+
+if TYPE_CHECKING:
+    from reflector.db.rooms import Room
+
+# separator doesn't guarantee there's no more "ROOM_PREFIX_SEPARATOR" strings in room name
+ROOM_PREFIX_SEPARATOR = "-"
+
+
+class VideoPlatformClient(ABC):
+    PLATFORM_NAME: Platform
+
+    def __init__(self, config: VideoPlatformConfig):
+        self.config = config
+
+    @abstractmethod
+    async def create_meeting(
+        self, room_name_prefix: NonEmptyString, end_date: datetime, room: "Room"
+    ) -> MeetingData:
+        pass
+
+    @abstractmethod
+    async def get_room_sessions(self, room_name: str) -> List[Any] | None:
+        pass
+
+    @abstractmethod
+    async def delete_room(self, room_name: str) -> bool:
+        pass
+
+    @abstractmethod
+    async def upload_logo(self, room_name: str, logo_path: str) -> bool:
+        pass
+
+    @abstractmethod
+    def verify_webhook_signature(
+        self, body: bytes, signature: str, timestamp: Optional[str] = None
+    ) -> bool:
+        pass
+
+    def format_recording_config(self, room: "Room") -> Dict[str, Any]:
+        if room.recording_type == "cloud" and self.config.s3_bucket:
+            return {
+                "type": room.recording_type,
+                "bucket": self.config.s3_bucket,
+                "region": self.config.s3_region,
+                "trigger": room.recording_trigger,
+            }
+        return {"type": room.recording_type}

server/reflector/video_platforms/daily.py

@@ -0,0 +1,198 @@
+import base64
+import hmac
+from datetime import datetime
+from hashlib import sha256
+from http import HTTPStatus
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from reflector.db.rooms import Room
+from reflector.logger import logger
+from reflector.storage import get_dailyco_storage
+
+from ..schemas.platform import Platform
+from ..utils.daily import DailyRoomName
+from ..utils.string import NonEmptyString
+from .base import ROOM_PREFIX_SEPARATOR, VideoPlatformClient
+from .models import MeetingData, RecordingType, VideoPlatformConfig
+
+
+class DailyClient(VideoPlatformClient):
+    PLATFORM_NAME: Platform = "daily"
+    TIMEOUT = 10
+    BASE_URL = "https://api.daily.co/v1"
+    TIMESTAMP_FORMAT = "%Y%m%d%H%M%S"
+    RECORDING_NONE: RecordingType = "none"
+    RECORDING_CLOUD: RecordingType = "cloud"
+
+    def __init__(self, config: VideoPlatformConfig):
+        super().__init__(config)
+        self.headers = {
+            "Authorization": f"Bearer {config.api_key}",
+            "Content-Type": "application/json",
+        }
+
+    async def create_meeting(
+        self, room_name_prefix: NonEmptyString, end_date: datetime, room: Room
+    ) -> MeetingData:
+        """
+        Daily.co rooms vs meetings:
+        - We create a NEW Daily.co room for each Reflector meeting
+        - Daily.co meeting/session starts automatically when first participant joins
+        - Room auto-deletes after exp time
+        - Meeting.room_name stores the timestamped Daily.co room name
+        """
+        timestamp = datetime.now().strftime(self.TIMESTAMP_FORMAT)
+        room_name = f"{room_name_prefix}{ROOM_PREFIX_SEPARATOR}{timestamp}"
+
+        data = {
+            "name": room_name,
+            "privacy": "private" if room.is_locked else "public",
+            "properties": {
+                "enable_recording": "raw-tracks"
+                if room.recording_type != self.RECORDING_NONE
+                else False,
+                "enable_chat": True,
+                "enable_screenshare": True,
+                "start_video_off": False,
+                "start_audio_off": False,
+                "exp": int(end_date.timestamp()),
+            },
+        }
+
+        # Get storage config for passing to Daily API
+        daily_storage = get_dailyco_storage()
+        assert daily_storage.bucket_name, "S3 bucket must be configured"
+        data["properties"]["recordings_bucket"] = {
+            "bucket_name": daily_storage.bucket_name,
+            "bucket_region": daily_storage.region,
+            "assume_role_arn": daily_storage.role_credential,
+            "allow_api_access": True,
+        }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.BASE_URL}/rooms",
+                headers=self.headers,
+                json=data,
+                timeout=self.TIMEOUT,
+            )
+            if response.status_code >= 400:
+                logger.error(
+                    "Daily.co API error",
+                    status_code=response.status_code,
+                    response_body=response.text,
+                    request_data=data,
+                )
+            response.raise_for_status()
+            result = response.json()
+
+        room_url = result["url"]
+
+        return MeetingData(
+            meeting_id=result["id"],
+            room_name=result["name"],
+            room_url=room_url,
+            host_room_url=room_url,
+            platform=self.PLATFORM_NAME,
+            extra_data=result,
+        )
+
+    async def get_room_sessions(self, room_name: str) -> List[Any] | None:
+        # no such api
+        return None
+
+    async def get_room_presence(self, room_name: str) -> Dict[str, Any]:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{self.BASE_URL}/rooms/{room_name}/presence",
+                headers=self.headers,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            return response.json()
+
+    async def get_meeting_participants(self, meeting_id: str) -> Dict[str, Any]:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{self.BASE_URL}/meetings/{meeting_id}/participants",
+                headers=self.headers,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            return response.json()
+
+    async def get_recording(self, recording_id: str) -> Dict[str, Any]:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{self.BASE_URL}/recordings/{recording_id}",
+                headers=self.headers,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            return response.json()
+
+    async def delete_room(self, room_name: str) -> bool:
+        async with httpx.AsyncClient() as client:
+            response = await client.delete(
+                f"{self.BASE_URL}/rooms/{room_name}",
+                headers=self.headers,
+                timeout=self.TIMEOUT,
+            )
+            return response.status_code in (HTTPStatus.OK, HTTPStatus.NOT_FOUND)
+
+    async def upload_logo(self, room_name: str, logo_path: str) -> bool:
+        return True
+
+    def verify_webhook_signature(
+        self, body: bytes, signature: str, timestamp: Optional[str] = None
+    ) -> bool:
+        """Verify Daily.co webhook signature.
+
+        Daily.co uses:
+        - X-Webhook-Signature header
+        - X-Webhook-Timestamp header
+        - Signature format: HMAC-SHA256(base64_decode(secret), timestamp + '.' + body)
+        - Result is base64 encoded
+        """
+        if not signature or not timestamp:
+            return False
+
+        try:
+            secret_bytes = base64.b64decode(self.config.webhook_secret)
+
+            signed_content = timestamp.encode() + b"." + body
+
+            expected = hmac.new(secret_bytes, signed_content, sha256).digest()
+            expected_b64 = base64.b64encode(expected).decode()
+
+            return hmac.compare_digest(expected_b64, signature)
+        except Exception as e:
+            logger.error("Daily.co webhook signature verification failed", exc_info=e)
+            return False
+
+    async def create_meeting_token(
+        self,
+        room_name: DailyRoomName,
+        enable_recording: bool,
+        user_id: Optional[str] = None,
+    ) -> str:
+        data = {"properties": {"room_name": room_name}}
+
+        if enable_recording:
+            data["properties"]["start_cloud_recording"] = True
+            data["properties"]["enable_recording_ui"] = False
+
+        if user_id:
+            data["properties"]["user_id"] = user_id
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.BASE_URL}/meeting-tokens",
+                headers=self.headers,
+                json=data,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            return response.json()["token"]

server/reflector/video_platforms/factory.py

@@ -0,0 +1,62 @@
+from typing import Optional
+
+from reflector.settings import settings
+from reflector.storage import get_dailyco_storage, get_whereby_storage
+
+from ..schemas.platform import WHEREBY_PLATFORM, Platform
+from .base import VideoPlatformClient, VideoPlatformConfig
+from .registry import get_platform_client
+
+
+def get_platform_config(platform: Platform) -> VideoPlatformConfig:
+    if platform == WHEREBY_PLATFORM:
+        if not settings.WHEREBY_API_KEY:
+            raise ValueError(
+                "WHEREBY_API_KEY is required when platform='whereby'. "
+                "Set WHEREBY_API_KEY environment variable."
+            )
+        whereby_storage = get_whereby_storage()
+        key_id, secret = whereby_storage.key_credentials
+        return VideoPlatformConfig(
+            api_key=settings.WHEREBY_API_KEY,
+            webhook_secret=settings.WHEREBY_WEBHOOK_SECRET or "",
+            api_url=settings.WHEREBY_API_URL,
+            s3_bucket=whereby_storage.bucket_name,
+            s3_region=whereby_storage.region,
+            aws_access_key_id=key_id,
+            aws_access_key_secret=secret,
+        )
+    elif platform == "daily":
+        if not settings.DAILY_API_KEY:
+            raise ValueError(
+                "DAILY_API_KEY is required when platform='daily'. "
+                "Set DAILY_API_KEY environment variable."
+            )
+        if not settings.DAILY_SUBDOMAIN:
+            raise ValueError(
+                "DAILY_SUBDOMAIN is required when platform='daily'. "
+                "Set DAILY_SUBDOMAIN environment variable."
+            )
+        daily_storage = get_dailyco_storage()
+        return VideoPlatformConfig(
+            api_key=settings.DAILY_API_KEY,
+            webhook_secret=settings.DAILY_WEBHOOK_SECRET or "",
+            subdomain=settings.DAILY_SUBDOMAIN,
+            s3_bucket=daily_storage.bucket_name,
+            s3_region=daily_storage.region,
+            aws_role_arn=daily_storage.role_credential,
+        )
+    else:
+        raise ValueError(f"Unknown platform: {platform}")
+
+
+def create_platform_client(platform: Platform) -> VideoPlatformClient:
+    config = get_platform_config(platform)
+    return get_platform_client(platform, config)
+
+
+def get_platform(room_platform: Optional[Platform] = None) -> Platform:
+    if room_platform:
+        return room_platform
+
+    return settings.DEFAULT_VIDEO_PLATFORM

server/reflector/video_platforms/models.py

@@ -0,0 +1,40 @@
+from typing import Any, Dict, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from reflector.schemas.platform import WHEREBY_PLATFORM, Platform
+
+RecordingType = Literal["none", "local", "cloud"]
+
+
+class MeetingData(BaseModel):
+    platform: Platform
+    meeting_id: str = Field(description="Platform-specific meeting identifier")
+    room_url: str = Field(description="URL for participants to join")
+    host_room_url: str = Field(description="URL for hosts (may be same as room_url)")
+    room_name: str = Field(description="Human-readable room name")
+    extra_data: Dict[str, Any] = Field(default_factory=dict)
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "platform": WHEREBY_PLATFORM,
+                "meeting_id": "12345678",
+                "room_url": "https://subdomain.whereby.com/room-20251008120000",
+                "host_room_url": "https://subdomain.whereby.com/room-20251008120000?roomKey=abc123",
+                "room_name": "room-20251008120000",
+            }
+        }
+
+
+class VideoPlatformConfig(BaseModel):
+    api_key: str
+    webhook_secret: str
+    api_url: Optional[str] = None
+    subdomain: Optional[str] = None  # Whereby/Daily subdomain
+    s3_bucket: Optional[str] = None
+    s3_region: Optional[str] = None
+    # Whereby uses access keys, Daily uses IAM role
+    aws_access_key_id: Optional[str] = None
+    aws_access_key_secret: Optional[str] = None
+    aws_role_arn: Optional[str] = None

server/reflector/video_platforms/registry.py

@@ -0,0 +1,35 @@
+from typing import Dict, Type
+
+from ..schemas.platform import DAILY_PLATFORM, WHEREBY_PLATFORM, Platform
+from .base import VideoPlatformClient, VideoPlatformConfig
+
+_PLATFORMS: Dict[Platform, Type[VideoPlatformClient]] = {}
+
+
+def register_platform(name: Platform, client_class: Type[VideoPlatformClient]):
+    _PLATFORMS[name] = client_class
+
+
+def get_platform_client(
+    platform: Platform, config: VideoPlatformConfig
+) -> VideoPlatformClient:
+    if platform not in _PLATFORMS:
+        raise ValueError(f"Unknown video platform: {platform}")
+
+    client_class = _PLATFORMS[platform]
+    return client_class(config)
+
+
+def get_available_platforms() -> list[Platform]:
+    return list(_PLATFORMS.keys())
+
+
+def _register_builtin_platforms():
+    from .daily import DailyClient  # noqa: PLC0415
+    from .whereby import WherebyClient  # noqa: PLC0415
+
+    register_platform(WHEREBY_PLATFORM, WherebyClient)
+    register_platform(DAILY_PLATFORM, DailyClient)
+
+
+_register_builtin_platforms()

server/reflector/video_platforms/whereby.py

@@ -0,0 +1,141 @@
+import hmac
+import json
+import re
+import time
+from datetime import datetime
+from hashlib import sha256
+from typing import Any, Dict, Optional
+
+import httpx
+
+from reflector.db.rooms import Room
+from reflector.storage import get_whereby_storage
+
+from ..schemas.platform import WHEREBY_PLATFORM, Platform
+from ..utils.string import NonEmptyString
+from .base import (
+    MeetingData,
+    VideoPlatformClient,
+    VideoPlatformConfig,
+)
+from .whereby_utils import whereby_room_name_prefix
+
+
+class WherebyClient(VideoPlatformClient):
+    PLATFORM_NAME: Platform = WHEREBY_PLATFORM
+    TIMEOUT = 10  # seconds
+    MAX_ELAPSED_TIME = 60 * 1000  # 1 minute in milliseconds
+
+    def __init__(self, config: VideoPlatformConfig):
+        super().__init__(config)
+        self.headers = {
+            "Content-Type": "application/json; charset=utf-8",
+            "Authorization": f"Bearer {config.api_key}",
+        }
+
+    async def create_meeting(
+        self, room_name_prefix: NonEmptyString, end_date: datetime, room: Room
+    ) -> MeetingData:
+        data = {
+            "isLocked": room.is_locked,
+            "roomNamePrefix": whereby_room_name_prefix(room_name_prefix),
+            "roomNamePattern": "uuid",
+            "roomMode": room.room_mode,
+            "endDate": end_date.isoformat(),
+            "fields": ["hostRoomUrl"],
+        }
+
+        if room.recording_type == "cloud":
+            # Get storage config for passing credentials to Whereby API
+            whereby_storage = get_whereby_storage()
+            key_id, secret = whereby_storage.key_credentials
+            data["recording"] = {
+                "type": room.recording_type,
+                "destination": {
+                    "provider": "s3",
+                    "bucket": whereby_storage.bucket_name,
+                    "accessKeyId": key_id,
+                    "accessKeySecret": secret,
+                    "fileFormat": "mp4",
+                },
+                "startTrigger": room.recording_trigger,
+            }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.config.api_url}/meetings",
+                headers=self.headers,
+                json=data,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            result = response.json()
+
+        return MeetingData(
+            meeting_id=result["meetingId"],
+            room_name=result["roomName"],
+            room_url=result["roomUrl"],
+            host_room_url=result["hostRoomUrl"],
+            platform=self.PLATFORM_NAME,
+            extra_data=result,
+        )
+
+    async def get_room_sessions(self, room_name: str) -> Dict[str, Any]:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"{self.config.api_url}/insights/room-sessions?roomName={room_name}",
+                headers=self.headers,
+                timeout=self.TIMEOUT,
+            )
+            response.raise_for_status()
+            return response.json().get("results", [])
+
+    async def delete_room(self, room_name: str) -> bool:
+        return True
+
+    async def upload_logo(self, room_name: str, logo_path: str) -> bool:
+        async with httpx.AsyncClient() as client:
+            with open(logo_path, "rb") as f:
+                response = await client.put(
+                    f"{self.config.api_url}/rooms/{room_name}/theme/logo",
+                    headers={
+                        "Authorization": f"Bearer {self.config.api_key}",
+                    },
+                    timeout=self.TIMEOUT,
+                    files={"image": f},
+                )
+                response.raise_for_status()
+        return True
+
+    def verify_webhook_signature(
+        self, body: bytes, signature: str, timestamp: Optional[str] = None
+    ) -> bool:
+        if not signature:
+            return False
+
+        matches = re.match(r"t=(.*),v1=(.*)", signature)
+        if not matches:
+            return False
+
+        ts, sig = matches.groups()
+
+        current_time = int(time.time() * 1000)
+        diff_time = current_time - int(ts) * 1000
+        if diff_time >= self.MAX_ELAPSED_TIME:
+            return False
+
+        body_dict = json.loads(body)
+        signed_payload = f"{ts}.{json.dumps(body_dict, separators=(',', ':'))}"
+        hmac_obj = hmac.new(
+            self.config.webhook_secret.encode("utf-8"),
+            signed_payload.encode("utf-8"),
+            sha256,
+        )
+        expected_signature = hmac_obj.hexdigest()
+
+        try:
+            return hmac.compare_digest(
+                expected_signature.encode("utf-8"), sig.encode("utf-8")
+            )
+        except Exception:
+            return False

server/reflector/video_platforms/whereby_utils.py

@@ -0,0 +1,38 @@
+import re
+from datetime import datetime
+
+from reflector.utils.datetime import parse_datetime_with_timezone
+from reflector.utils.string import NonEmptyString, parse_non_empty_string
+from reflector.video_platforms.base import ROOM_PREFIX_SEPARATOR
+
+
+def parse_whereby_recording_filename(
+    object_key: NonEmptyString,
+) -> (NonEmptyString, datetime):
+    filename = parse_non_empty_string(object_key.rsplit(".", 1)[0])
+    timestamp_pattern = r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z)"
+    match = re.search(timestamp_pattern, filename)
+    if not match:
+        raise ValueError(f"No ISO timestamp found in filename: {object_key}")
+    timestamp_str = match.group(1)
+    timestamp_start = match.start(1)
+    room_name_part = filename[:timestamp_start]
+    if room_name_part.endswith(ROOM_PREFIX_SEPARATOR):
+        room_name_part = room_name_part[: -len(ROOM_PREFIX_SEPARATOR)]
+    else:
+        raise ValueError(
+            f"room name {room_name_part} doesnt have {ROOM_PREFIX_SEPARATOR} at the end of filename: {object_key}"
+        )
+
+    return parse_non_empty_string(room_name_part), parse_datetime_with_timezone(
+        timestamp_str
+    )
+
+
+def whereby_room_name_prefix(room_name_prefix: NonEmptyString) -> NonEmptyString:
+    return room_name_prefix + ROOM_PREFIX_SEPARATOR
+
+
+# room name comes with "/" from whereby api but lacks "/" e.g. in recording filenames
+def room_name_to_whereby_api_room_name(room_name: NonEmptyString) -> NonEmptyString:
+    return f"/{room_name}"

server/reflector/views/daily.py

@@ -0,0 +1,233 @@
+import json
+from typing import Any, Dict, Literal
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+
+from reflector.db.meetings import meetings_controller
+from reflector.logger import logger as _logger
+from reflector.settings import settings
+from reflector.utils.daily import DailyRoomName
+from reflector.video_platforms.factory import create_platform_client
+from reflector.worker.process import process_multitrack_recording
+
+router = APIRouter()
+
+logger = _logger.bind(platform="daily")
+
+
+class DailyTrack(BaseModel):
+    type: Literal["audio", "video"]
+    s3Key: str
+    size: int
+
+
+class DailyWebhookEvent(BaseModel):
+    version: str
+    type: str
+    id: str
+    payload: Dict[str, Any]
+    event_ts: float
+
+
+def _extract_room_name(event: DailyWebhookEvent) -> DailyRoomName | None:
+    """Extract room name from Daily event payload.
+
+    Daily.co API inconsistency:
+    - participant.* events use "room" field
+    - recording.* events use "room_name" field
+    """
+    return event.payload.get("room_name") or event.payload.get("room")
+
+
+@router.post("/webhook")
+async def webhook(request: Request):
+    """Handle Daily webhook events.
+
+    Daily.co circuit-breaker: After 3+ failed responses (4xx/5xx), webhook
+    state→FAILED, stops sending events. Reset: scripts/recreate_daily_webhook.py
+    """
+    body = await request.body()
+    signature = request.headers.get("X-Webhook-Signature", "")
+    timestamp = request.headers.get("X-Webhook-Timestamp", "")
+
+    client = create_platform_client("daily")
+
+    # TEMPORARY: Bypass signature check for testing
+    # TODO: Remove this after testing is complete
+    BYPASS_FOR_TESTING = True
+    if not BYPASS_FOR_TESTING:
+        if not client.verify_webhook_signature(body, signature, timestamp):
+            logger.warning(
+                "Invalid webhook signature",
+                signature=signature,
+                timestamp=timestamp,
+                has_body=bool(body),
+            )
+            raise HTTPException(status_code=401, detail="Invalid webhook signature")
+
+    try:
+        body_json = json.loads(body)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=422, detail="Invalid JSON")
+
+    if body_json.get("test") == "test":
+        logger.info("Received Daily webhook test event")
+        return {"status": "ok"}
+
+    # Parse as actual event
+    try:
+        event = DailyWebhookEvent(**body_json)
+    except Exception as e:
+        logger.error("Failed to parse webhook event", error=str(e), body=body.decode())
+        raise HTTPException(status_code=422, detail="Invalid event format")
+
+    # Handle participant events
+    if event.type == "participant.joined":
+        await _handle_participant_joined(event)
+    elif event.type == "participant.left":
+        await _handle_participant_left(event)
+    elif event.type == "recording.started":
+        await _handle_recording_started(event)
+    elif event.type == "recording.ready-to-download":
+        await _handle_recording_ready(event)
+    elif event.type == "recording.error":
+        await _handle_recording_error(event)
+    else:
+        logger.warning(
+            "Unhandled Daily webhook event type",
+            event_type=event.type,
+            payload=event.payload,
+        )
+
+    return {"status": "ok"}
+
+
+async def _handle_participant_joined(event: DailyWebhookEvent):
+    daily_room_name = _extract_room_name(event)
+    if not daily_room_name:
+        logger.warning("participant.joined: no room in payload", payload=event.payload)
+        return
+
+    meeting = await meetings_controller.get_by_room_name(daily_room_name)
+    if meeting:
+        await meetings_controller.increment_num_clients(meeting.id)
+        logger.info(
+            "Participant joined",
+            meeting_id=meeting.id,
+            room_name=daily_room_name,
+            recording_type=meeting.recording_type,
+            recording_trigger=meeting.recording_trigger,
+        )
+    else:
+        logger.warning(
+            "participant.joined: meeting not found", room_name=daily_room_name
+        )
+
+
+async def _handle_participant_left(event: DailyWebhookEvent):
+    room_name = _extract_room_name(event)
+    if not room_name:
+        return
+
+    meeting = await meetings_controller.get_by_room_name(room_name)
+    if meeting:
+        await meetings_controller.decrement_num_clients(meeting.id)
+
+
+async def _handle_recording_started(event: DailyWebhookEvent):
+    room_name = _extract_room_name(event)
+    if not room_name:
+        logger.warning(
+            "recording.started: no room_name in payload", payload=event.payload
+        )
+        return
+
+    meeting = await meetings_controller.get_by_room_name(room_name)
+    if meeting:
+        logger.info(
+            "Recording started",
+            meeting_id=meeting.id,
+            room_name=room_name,
+            recording_id=event.payload.get("recording_id"),
+            platform="daily",
+        )
+    else:
+        logger.warning("recording.started: meeting not found", room_name=room_name)
+
+
+async def _handle_recording_ready(event: DailyWebhookEvent):
+    """Handle recording ready for download event.
+
+    Daily.co webhook payload for raw-tracks recordings:
+    {
+      "recording_id": "...",
+      "room_name": "test2-20251009192341",
+      "tracks": [
+        {"type": "audio", "s3Key": "monadical/test2-.../uuid-cam-audio-123.webm", "size": 400000},
+        {"type": "video", "s3Key": "monadical/test2-.../uuid-cam-video-456.webm", "size": 30000000}
+      ]
+    }
+    """
+    room_name = _extract_room_name(event)
+    recording_id = event.payload.get("recording_id")
+    tracks_raw = event.payload.get("tracks", [])
+
+    if not room_name or not tracks_raw:
+        logger.warning(
+            "recording.ready-to-download: missing room_name or tracks",
+            room_name=room_name,
+            has_tracks=bool(tracks_raw),
+            payload=event.payload,
+        )
+        return
+
+    try:
+        tracks = [DailyTrack(**t) for t in tracks_raw]
+    except Exception as e:
+        logger.error(
+            "recording.ready-to-download: invalid tracks structure",
+            error=str(e),
+            tracks=tracks_raw,
+        )
+        return
+
+    logger.info(
+        "Recording ready for download",
+        room_name=room_name,
+        recording_id=recording_id,
+        num_tracks=len(tracks),
+        platform="daily",
+    )
+
+    bucket_name = settings.DAILYCO_STORAGE_AWS_BUCKET_NAME
+    if not bucket_name:
+        logger.error(
+            "DAILYCO_STORAGE_AWS_BUCKET_NAME not configured; cannot process Daily recording"
+        )
+        return
+
+    track_keys = [t.s3Key for t in tracks if t.type == "audio"]
+
+    process_multitrack_recording.delay(
+        bucket_name=bucket_name,
+        daily_room_name=room_name,
+        recording_id=recording_id,
+        track_keys=track_keys,
+    )
+
+
+async def _handle_recording_error(event: DailyWebhookEvent):
+    room_name = _extract_room_name(event)
+    error = event.payload.get("error", "Unknown error")
+
+    if room_name:
+        meeting = await meetings_controller.get_by_room_name(room_name)
+        if meeting:
+            logger.error(
+                "Recording error",
+                meeting_id=meeting.id,
+                room_name=room_name,
+                error=error,
+                platform="daily",
+            )

server/reflector/views/meetings.py

@@ -10,6 +10,7 @@ from reflector.db.meetings import (
     meeting_consent_controller,
     meetings_controller,
 )
+from reflector.db.rooms import rooms_controller
 
 router = APIRouter()
 
@@ -41,3 +42,34 @@ async def meeting_audio_consent(
     updated_consent = await meeting_consent_controller.upsert(consent)
 
     return {"status": "success", "consent_id": updated_consent.id}
+
+
+@router.patch("/meetings/{meeting_id}/deactivate")
+async def meeting_deactivate(
+    meeting_id: str,
+    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user)],
+):
+    user_id = user["sub"] if user else None
+    if not user_id:
+        raise HTTPException(status_code=401, detail="Authentication required")
+
+    meeting = await meetings_controller.get_by_id(meeting_id)
+    if not meeting:
+        raise HTTPException(status_code=404, detail="Meeting not found")
+
+    if not meeting.is_active:
+        return {"status": "success", "meeting_id": meeting_id}
+
+    # Only room owner or meeting creator can deactivate
+    room = await rooms_controller.get_by_id(meeting.room_id)
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    if user_id != room.user_id and user_id != meeting.user_id:
+        raise HTTPException(
+            status_code=403, detail="Only the room owner can deactivate meetings"
+        )
+
+    await meetings_controller.update_meeting(meeting_id, is_active=False)
+
+    return {"status": "success", "meeting_id": meeting_id}

server/reflector/views/rooms.py

@@ -1,34 +1,32 @@
 import logging
-import sqlite3
 from datetime import datetime, timedelta, timezone
-from typing import Annotated, Literal, Optional
+from enum import Enum
+from typing import Annotated, Any, Literal, Optional
 
-import asyncpg.exceptions
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi_pagination import Page
 from fastapi_pagination.ext.databases import apaginate
 from pydantic import BaseModel
+from redis.exceptions import LockError
 
 import reflector.auth as auth
 from reflector.db import get_database
+from reflector.db.calendar_events import calendar_events_controller
 from reflector.db.meetings import meetings_controller
 from reflector.db.rooms import rooms_controller
+from reflector.redis_cache import RedisAsyncLock
+from reflector.schemas.platform import Platform
+from reflector.services.ics_sync import ics_sync_service
 from reflector.settings import settings
-from reflector.whereby import create_meeting, upload_logo
+from reflector.utils.url import add_query_param
+from reflector.video_platforms.factory import (
+    create_platform_client,
+    get_platform,
+)
 from reflector.worker.webhook import test_webhook
 
 logger = logging.getLogger(__name__)
 
-router = APIRouter()
-
-
-def parse_datetime_with_timezone(iso_string: str) -> datetime:
-    """Parse ISO datetime string and ensure timezone awareness (defaults to UTC if naive)."""
-    dt = datetime.fromisoformat(iso_string)
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt
-
 
 class Room(BaseModel):
     id: str
@@ -43,6 +41,12 @@ class Room(BaseModel):
     recording_type: str
     recording_trigger: str
     is_shared: bool
+    ics_url: Optional[str] = None
+    ics_fetch_interval: int = 300
+    ics_enabled: bool = False
+    ics_last_sync: Optional[datetime] = None
+    ics_last_etag: Optional[str] = None
+    platform: Platform
 
 
 class RoomDetails(Room):
@@ -54,10 +58,23 @@ class Meeting(BaseModel):
     id: str
     room_name: str
     room_url: str
+    # TODO it's not always present, | None
     host_room_url: str
     start_date: datetime
     end_date: datetime
+    user_id: str | None = None
+    room_id: str | None = None
+    is_locked: bool = False
+    room_mode: Literal["normal", "group"] = "normal"
     recording_type: Literal["none", "local", "cloud"] = "cloud"
+    recording_trigger: Literal[
+        "none", "prompt", "automatic", "automatic-2nd-participant"
+    ] = "automatic-2nd-participant"
+    num_clients: int = 0
+    is_active: bool = True
+    calendar_event_id: str | None = None
+    calendar_metadata: dict[str, Any] | None = None
+    platform: Platform
 
 
 class CreateRoom(BaseModel):
@@ -72,20 +89,32 @@ class CreateRoom(BaseModel):
     is_shared: bool
     webhook_url: str
     webhook_secret: str
+    ics_url: Optional[str] = None
+    ics_fetch_interval: int = 300
+    ics_enabled: bool = False
+    platform: Optional[Platform] = None
 
 
 class UpdateRoom(BaseModel):
-    name: str
-    zulip_auto_post: bool
-    zulip_stream: str
-    zulip_topic: str
-    is_locked: bool
-    room_mode: str
-    recording_type: str
-    recording_trigger: str
-    is_shared: bool
-    webhook_url: str
-    webhook_secret: str
+    name: Optional[str] = None
+    zulip_auto_post: Optional[bool] = None
+    zulip_stream: Optional[str] = None
+    zulip_topic: Optional[str] = None
+    is_locked: Optional[bool] = None
+    room_mode: Optional[str] = None
+    recording_type: Optional[str] = None
+    recording_trigger: Optional[str] = None
+    is_shared: Optional[bool] = None
+    webhook_url: Optional[str] = None
+    webhook_secret: Optional[str] = None
+    ics_url: Optional[str] = None
+    ics_fetch_interval: Optional[int] = None
+    ics_enabled: Optional[bool] = None
+    platform: Optional[Platform] = None
+
+
+class CreateRoomMeeting(BaseModel):
+    allow_duplicated: Optional[bool] = False
 
 
 class DeletionStatus(BaseModel):
@@ -100,6 +129,51 @@ class WebhookTestResult(BaseModel):
     response_preview: str | None = None
 
 
+class ICSStatus(BaseModel):
+    status: Literal["enabled", "disabled"]
+    last_sync: Optional[datetime] = None
+    next_sync: Optional[datetime] = None
+    last_etag: Optional[str] = None
+    events_count: int = 0
+
+
+class SyncStatus(str, Enum):
+    success = "success"
+    unchanged = "unchanged"
+    error = "error"
+    skipped = "skipped"
+
+
+class ICSSyncResult(BaseModel):
+    status: SyncStatus
+    hash: Optional[str] = None
+    events_found: int = 0
+    total_events: int = 0
+    events_created: int = 0
+    events_updated: int = 0
+    events_deleted: int = 0
+    error: Optional[str] = None
+    reason: Optional[str] = None
+
+
+class CalendarEventResponse(BaseModel):
+    id: str
+    room_id: str
+    ics_uid: str
+    title: Optional[str] = None
+    description: Optional[str] = None
+    start_time: datetime
+    end_time: datetime
+    attendees: Optional[list[dict]] = None
+    location: Optional[str] = None
+    last_synced: datetime
+    created_at: datetime
+    updated_at: datetime
+
+
+router = APIRouter()
+
+
 @router.get("/rooms", response_model=Page[RoomDetails])
 async def rooms_list(
     user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
@@ -109,13 +183,18 @@ async def rooms_list(
 
     user_id = user["sub"] if user else None
 
-    return await apaginate(
+    paginated = await apaginate(
         get_database(),
         await rooms_controller.get_all(
             user_id=user_id, order_by="-created_at", return_query=True
         ),
     )
 
+    for room in paginated.items:
+        room.platform = get_platform(room.platform)
+
+    return paginated
+
 
 @router.get("/rooms/{room_id}", response_model=RoomDetails)
 async def rooms_get(
@@ -126,15 +205,41 @@ async def rooms_get(
     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")
+    if not room.is_shared and (user_id is None or room.user_id != user_id):
+        raise HTTPException(status_code=403, detail="Room access denied")
+    room.platform = get_platform(room.platform)
     return room
 
 
+@router.get("/rooms/name/{room_name}", response_model=RoomDetails)
+async def rooms_get_by_name(
+    room_name: 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_name(room_name)
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    room_dict = room.__dict__.copy()
+    if user_id == room.user_id:
+        room_dict["webhook_url"] = getattr(room, "webhook_url", None)
+        room_dict["webhook_secret"] = getattr(room, "webhook_secret", None)
+    else:
+        room_dict["webhook_url"] = None
+        room_dict["webhook_secret"] = None
+
+    room_dict["platform"] = get_platform(room.platform)
+
+    return RoomDetails(**room_dict)
+
+
 @router.post("/rooms", response_model=Room)
 async def rooms_create(
     room: CreateRoom,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
 
     return await rooms_controller.add(
         name=room.name,
@@ -149,6 +254,10 @@ async def rooms_create(
         is_shared=room.is_shared,
         webhook_url=room.webhook_url,
         webhook_secret=room.webhook_secret,
+        ics_url=room.ics_url,
+        ics_fetch_interval=room.ics_fetch_interval,
+        ics_enabled=room.ics_enabled,
+        platform=room.platform,
     )
 
 
@@ -156,26 +265,31 @@ async def rooms_create(
 async def rooms_update(
     room_id: str,
     info: UpdateRoom,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     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")
+    if room.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
     values = info.dict(exclude_unset=True)
     await rooms_controller.update(room, values)
+    room.platform = get_platform(room.platform)
     return room
 
 
 @router.delete("/rooms/{room_id}", response_model=DeletionStatus)
 async def rooms_delete(
     room_id: str,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
-    room = await rooms_controller.get_by_id(room_id, user_id=user_id)
+    user_id = user["sub"]
+    room = await rooms_controller.get_by_id(room_id)
     if not room:
         raise HTTPException(status_code=404, detail="Room not found")
+    if room.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
     await rooms_controller.remove_by_id(room.id, user_id=user_id)
     return DeletionStatus(status="ok")
 
@@ -183,6 +297,7 @@ async def rooms_delete(
 @router.post("/rooms/{room_name}/meeting", response_model=Meeting)
 async def rooms_create_meeting(
     room_name: str,
+    info: CreateRoomMeeting,
     user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
 ):
     user_id = user["sub"] if user else None
@@ -190,52 +305,59 @@ async def rooms_create_meeting(
     if not room:
         raise HTTPException(status_code=404, detail="Room not found")
 
-    current_time = datetime.now(timezone.utc)
-    meeting = await meetings_controller.get_active(room=room, current_time=current_time)
+    try:
+        async with RedisAsyncLock(
+            f"create_meeting:{room_name}",
+            timeout=30,
+            extend_interval=10,
+            blocking_timeout=5.0,
+        ) as lock:
+            current_time = datetime.now(timezone.utc)
 
-    if meeting is None:
-        end_date = current_time + timedelta(hours=8)
+            meeting = None
+            if not info.allow_duplicated:
+                meeting = await meetings_controller.get_active(
+                    room=room, current_time=current_time
+                )
 
-        whereby_meeting = await create_meeting("", end_date=end_date, room=room)
-        await upload_logo(whereby_meeting["roomName"], "./images/logo.png")
-
-        # Now try to save to database
-        try:
-            meeting = await meetings_controller.create(
-                id=whereby_meeting["meetingId"],
-                room_name=whereby_meeting["roomName"],
-                room_url=whereby_meeting["roomUrl"],
-                host_room_url=whereby_meeting["hostRoomUrl"],
-                start_date=parse_datetime_with_timezone(whereby_meeting["startDate"]),
-                end_date=parse_datetime_with_timezone(whereby_meeting["endDate"]),
-                user_id=user_id,
-                room=room,
-            )
-        except (asyncpg.exceptions.UniqueViolationError, sqlite3.IntegrityError):
-            # Another request already created a meeting for this room
-            # Log this race condition occurrence
-            logger.info(
-                "Race condition detected for room %s - fetching existing meeting",
-                room.name,
-            )
-            logger.warning(
-                "Whereby meeting %s was created but not used (resource leak) for room %s",
-                whereby_meeting["meetingId"],
-                room.name,
-            )
-
-            # Fetch the meeting that was created by the other request
-            meeting = await meetings_controller.get_active(
-                room=room, current_time=current_time
-            )
             if meeting is None:
-                # Edge case: meeting was created but expired/deleted between checks
-                logger.error(
-                    "Meeting disappeared after race condition for room %s", room.name
+                end_date = current_time + timedelta(hours=8)
+
+                platform = get_platform(room.platform)
+                client = create_platform_client(platform)
+
+                meeting_data = await client.create_meeting(
+                    room.name, end_date=end_date, room=room
                 )
-                raise HTTPException(
-                    status_code=503, detail="Unable to join meeting - please try again"
+
+                await client.upload_logo(meeting_data.room_name, "./images/logo.png")
+
+                meeting = await meetings_controller.create(
+                    id=meeting_data.meeting_id,
+                    room_name=meeting_data.room_name,
+                    room_url=meeting_data.room_url,
+                    host_room_url=meeting_data.host_room_url,
+                    start_date=current_time,
+                    end_date=end_date,
+                    room=room,
                 )
+    except LockError:
+        logger.warning("Failed to acquire lock for room %s within timeout", room_name)
+        raise HTTPException(
+            status_code=503, detail="Meeting creation in progress, please try again"
+        )
+
+    if meeting.platform == "daily" and room.recording_trigger != "none":
+        client = create_platform_client(meeting.platform)
+        token = await client.create_meeting_token(
+            meeting.room_name,
+            enable_recording=True,
+            user_id=user_id,
+        )
+        meeting = meeting.model_copy()
+        meeting.room_url = add_query_param(meeting.room_url, "t", token)
+        if meeting.host_room_url:
+            meeting.host_room_url = add_query_param(meeting.host_room_url, "t", token)
 
     if user_id != room.user_id:
         meeting.host_room_url = ""
@@ -246,19 +368,210 @@ async def rooms_create_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)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
     """Test webhook configuration by sending a sample payload."""
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
 
     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:
+    if 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)
+
+
+@router.post("/rooms/{room_name}/ics/sync", response_model=ICSSyncResult)
+async def rooms_sync_ics(
+    room_name: 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_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    if user_id != room.user_id:
+        raise HTTPException(
+            status_code=403, detail="Only room owner can trigger ICS sync"
+        )
+
+    if not room.ics_enabled or not room.ics_url:
+        raise HTTPException(status_code=400, detail="ICS not configured for this room")
+
+    result = await ics_sync_service.sync_room_calendar(room)
+
+    if result["status"] == "error":
+        raise HTTPException(
+            status_code=500, detail=result.get("error", "Unknown error")
+        )
+
+    return ICSSyncResult(**result)
+
+
+@router.get("/rooms/{room_name}/ics/status", response_model=ICSStatus)
+async def rooms_ics_status(
+    room_name: 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_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    if user_id != room.user_id:
+        raise HTTPException(
+            status_code=403, detail="Only room owner can view ICS status"
+        )
+
+    next_sync = None
+    if room.ics_enabled and room.ics_last_sync:
+        next_sync = room.ics_last_sync + timedelta(seconds=room.ics_fetch_interval)
+
+    events = await calendar_events_controller.get_by_room(
+        room.id, include_deleted=False
+    )
+
+    return ICSStatus(
+        status="enabled" if room.ics_enabled else "disabled",
+        last_sync=room.ics_last_sync,
+        next_sync=next_sync,
+        last_etag=room.ics_last_etag,
+        events_count=len(events),
+    )
+
+
+@router.get("/rooms/{room_name}/meetings", response_model=list[CalendarEventResponse])
+async def rooms_list_meetings(
+    room_name: 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_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    events = await calendar_events_controller.get_by_room(
+        room.id, include_deleted=False
+    )
+
+    if user_id != room.user_id:
+        for event in events:
+            event.description = None
+            event.attendees = None
+
+    return events
+
+
+@router.get(
+    "/rooms/{room_name}/meetings/upcoming", response_model=list[CalendarEventResponse]
+)
+async def rooms_list_upcoming_meetings(
+    room_name: str,
+    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    minutes_ahead: int = 120,
+):
+    user_id = user["sub"] if user else None
+    room = await rooms_controller.get_by_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    events = await calendar_events_controller.get_upcoming(
+        room.id, minutes_ahead=minutes_ahead
+    )
+
+    if user_id != room.user_id:
+        for event in events:
+            event.description = None
+            event.attendees = None
+
+    return events
+
+
+@router.get("/rooms/{room_name}/meetings/active", response_model=list[Meeting])
+async def rooms_list_active_meetings(
+    room_name: 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_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    current_time = datetime.now(timezone.utc)
+    meetings = await meetings_controller.get_all_active_for_room(
+        room=room, current_time=current_time
+    )
+
+    effective_platform = get_platform(room.platform)
+    for meeting in meetings:
+        meeting.platform = effective_platform
+
+    if user_id != room.user_id:
+        for meeting in meetings:
+            meeting.host_room_url = ""
+
+    return meetings
+
+
+@router.get("/rooms/{room_name}/meetings/{meeting_id}", response_model=Meeting)
+async def rooms_get_meeting(
+    room_name: str,
+    meeting_id: str,
+    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+):
+    """Get a single meeting by ID within a specific room."""
+    user_id = user["sub"] if user else None
+
+    room = await rooms_controller.get_by_name(room_name)
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    meeting = await meetings_controller.get_by_id(meeting_id, room=room)
+    if not meeting:
+        raise HTTPException(status_code=404, detail="Meeting not found")
+
+    if user_id != room.user_id and not room.is_shared:
+        meeting.host_room_url = ""
+
+    return meeting
+
+
+@router.post("/rooms/{room_name}/meetings/{meeting_id}/join", response_model=Meeting)
+async def rooms_join_meeting(
+    room_name: str,
+    meeting_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_name(room_name)
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    meeting = await meetings_controller.get_by_id(meeting_id, room=room)
+
+    if not meeting:
+        raise HTTPException(status_code=404, detail="Meeting not found")
+
+    if not meeting.is_active:
+        raise HTTPException(status_code=400, detail="Meeting is not active")
+
+    current_time = datetime.now(timezone.utc)
+    if meeting.end_date <= current_time:
+        raise HTTPException(status_code=400, detail="Meeting has ended")
+
+    if user_id != room.user_id:
+        meeting.host_room_url = ""
+
+    return meeting

server/reflector/views/transcripts.py

@@ -5,12 +5,10 @@ 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, constr, field_serializer
+from pydantic import AwareDatetime, BaseModel, Field, constr, field_serializer
 
 import reflector.auth as auth
 from reflector.db import get_database
-from reflector.db.meetings import meetings_controller
-from reflector.db.rooms import rooms_controller
 from reflector.db.search import (
     DEFAULT_SEARCH_LIMIT,
     SearchLimit,
@@ -27,12 +25,14 @@ from reflector.db.search import (
 from reflector.db.transcripts import (
     SourceKind,
     TranscriptParticipant,
+    TranscriptStatus,
     TranscriptTopic,
     transcripts_controller,
 )
 from reflector.processors.types import Transcript as ProcessorTranscript
 from reflector.processors.types import Word
 from reflector.settings import settings
+from reflector.ws_manager import get_ws_manager
 from reflector.zulip import (
     InvalidMessageError,
     get_zulip_message,
@@ -63,7 +63,7 @@ class GetTranscriptMinimal(BaseModel):
     id: str
     user_id: str | None
     name: str
-    status: str
+    status: TranscriptStatus
     locked: bool
     duration: float
     title: str | None
@@ -133,6 +133,21 @@ SearchOffsetParam = Annotated[
     SearchOffsetBase, Query(description="Number of results to skip")
 ]
 
+SearchFromDatetimeParam = Annotated[
+    AwareDatetime | None,
+    Query(
+        alias="from",
+        description="Filter transcripts created on or after this datetime (ISO 8601 with timezone)",
+    ),
+]
+SearchToDatetimeParam = Annotated[
+    AwareDatetime | None,
+    Query(
+        alias="to",
+        description="Filter transcripts created on or before this datetime (ISO 8601 with timezone)",
+    ),
+]
+
 
 class SearchResponse(BaseModel):
     results: list[SearchResult]
@@ -174,18 +189,23 @@ async def transcripts_search(
     offset: SearchOffsetParam = 0,
     room_id: Optional[str] = None,
     source_kind: Optional[SourceKind] = None,
+    from_datetime: SearchFromDatetimeParam = None,
+    to_datetime: SearchToDatetimeParam = None,
     user: Annotated[
         Optional[auth.UserInfo], Depends(auth.current_user_optional)
     ] = None,
 ):
-    """
-    Full-text search across transcript titles and content.
-    """
+    """Full-text search across transcript titles and content."""
     if not user and not settings.PUBLIC_MODE:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
     user_id = user["sub"] if user else None
 
+    if from_datetime and to_datetime and from_datetime > to_datetime:
+        raise HTTPException(
+            status_code=400, detail="'from' must be less than or equal to 'to'"
+        )
+
     search_params = SearchParameters(
         query_text=parse_search_query_param(q),
         limit=limit,
@@ -193,6 +213,8 @@ async def transcripts_search(
         user_id=user_id,
         room_id=room_id,
         source_kind=source_kind,
+        from_datetime=from_datetime,
+        to_datetime=to_datetime,
     )
 
     results, total = await search_controller.search_transcripts(search_params)
@@ -212,7 +234,7 @@ async def transcripts_create(
     user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
 ):
     user_id = user["sub"] if user else None
-    return await transcripts_controller.add(
+    transcript = await transcripts_controller.add(
         info.name,
         source_kind=info.source_kind or SourceKind.LIVE,
         source_language=info.source_language,
@@ -220,6 +242,14 @@ async def transcripts_create(
         user_id=user_id,
     )
 
+    if user_id:
+        await get_ws_manager().send_json(
+            room_id=f"user:{user_id}",
+            message={"event": "TRANSCRIPT_CREATED", "data": {"id": transcript.id}},
+        )
+
+    return transcript
+
 
 # ==============================================================
 # Single transcript
@@ -343,14 +373,14 @@ async def transcript_get(
 async def transcript_update(
     transcript_id: str,
     info: UpdateTranscript,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
-    if not transcript:
-        raise HTTPException(status_code=404, detail="Transcript not found")
+    if not transcripts_controller.user_can_mutate(transcript, user_id):
+        raise HTTPException(status_code=403, detail="Not authorized")
     values = info.dict(exclude_unset=True)
     updated_transcript = await transcripts_controller.update(transcript, values)
     return updated_transcript
@@ -359,20 +389,20 @@ async def transcript_update(
 @router.delete("/transcripts/{transcript_id}", response_model=DeletionStatus)
 async def transcript_delete(
     transcript_id: str,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id(transcript_id)
     if not transcript:
         raise HTTPException(status_code=404, detail="Transcript not found")
-
-    if transcript.meeting_id:
-        meeting = await meetings_controller.get_by_id(transcript.meeting_id)
-        room = await rooms_controller.get_by_id(meeting.room_id)
-        if room.is_shared:
-            user_id = None
+    if not transcripts_controller.user_can_mutate(transcript, user_id):
+        raise HTTPException(status_code=403, detail="Not authorized")
 
     await transcripts_controller.remove_by_id(transcript.id, user_id=user_id)
+    await get_ws_manager().send_json(
+        room_id=f"user:{user_id}",
+        message={"event": "TRANSCRIPT_DELETED", "data": {"id": transcript.id}},
+    )
     return DeletionStatus(status="ok")
 
 
@@ -444,15 +474,16 @@ async def transcript_post_to_zulip(
     stream: str,
     topic: str,
     include_topics: bool,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ):
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
     if not transcript:
         raise HTTPException(status_code=404, detail="Transcript not found")
-
+    if not transcripts_controller.user_can_mutate(transcript, user_id):
+        raise HTTPException(status_code=403, detail="Not authorized")
     content = get_zulip_message(transcript, include_topics)
 
     message_updated = False

server/reflector/views/transcripts_participants.py

@@ -56,12 +56,14 @@ async def transcript_get_participants(
 async def transcript_add_participant(
     transcript_id: str,
     participant: CreateParticipant,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ) -> Participant:
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
+    if transcript.user_id is not None and transcript.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
 
     # ensure the speaker is unique
     if participant.speaker is not None and transcript.participants is not None:
@@ -101,12 +103,14 @@ async def transcript_update_participant(
     transcript_id: str,
     participant_id: str,
     participant: UpdateParticipant,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ) -> Participant:
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
+    if transcript.user_id is not None and transcript.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
 
     # ensure the speaker is unique
     for p in transcript.participants:
@@ -138,11 +142,13 @@ async def transcript_update_participant(
 async def transcript_delete_participant(
     transcript_id: str,
     participant_id: str,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ) -> DeletionStatus:
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
+    if transcript.user_id is not None and transcript.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
     await transcripts_controller.delete_participant(transcript, participant_id)
     return DeletionStatus(status="ok")

server/reflector/views/transcripts_process.py

@@ -5,8 +5,12 @@ from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
 import reflector.auth as auth
+from reflector.db.recordings import recordings_controller
 from reflector.db.transcripts import transcripts_controller
 from reflector.pipelines.main_file_pipeline import task_pipeline_file_process
+from reflector.pipelines.main_multitrack_pipeline import (
+    task_pipeline_multitrack_process,
+)
 
 router = APIRouter()
 
@@ -33,14 +37,35 @@ async def transcript_process(
             status_code=400, detail="Recording is not ready for processing"
         )
 
+    # avoid duplicate scheduling for either pipeline
     if task_is_scheduled_or_active(
         "reflector.pipelines.main_file_pipeline.task_pipeline_file_process",
         transcript_id=transcript_id,
+    ) or task_is_scheduled_or_active(
+        "reflector.pipelines.main_multitrack_pipeline.task_pipeline_multitrack_process",
+        transcript_id=transcript_id,
     ):
         return ProcessStatus(status="already running")
 
-    # schedule a background task process the file
-    task_pipeline_file_process.delay(transcript_id=transcript_id)
+    # Determine processing mode strictly from DB to avoid S3 scans
+    bucket_name = None
+    track_keys: list[str] = []
+
+    if transcript.recording_id:
+        recording = await recordings_controller.get_by_id(transcript.recording_id)
+        if recording:
+            bucket_name = recording.bucket_name
+            track_keys = list(getattr(recording, "track_keys", []) or [])
+
+    if bucket_name:
+        task_pipeline_multitrack_process.delay(
+            transcript_id=transcript_id,
+            bucket_name=bucket_name,
+            track_keys=track_keys,
+        )
+    else:
+        # Default single-file pipeline
+        task_pipeline_file_process.delay(transcript_id=transcript_id)
 
     return ProcessStatus(status="ok")
 

server/reflector/views/transcripts_speaker.py

@@ -35,12 +35,14 @@ class SpeakerMerge(BaseModel):
 async def transcript_assign_speaker(
     transcript_id: str,
     assignment: SpeakerAssignment,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ) -> SpeakerAssignmentStatus:
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
+    if transcript.user_id is not None and transcript.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
 
     if not transcript:
         raise HTTPException(status_code=404, detail="Transcript not found")
@@ -113,12 +115,14 @@ async def transcript_assign_speaker(
 async def transcript_merge_speaker(
     transcript_id: str,
     merge: SpeakerMerge,
-    user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
+    user: Annotated[auth.UserInfo, Depends(auth.current_user)],
 ) -> SpeakerAssignmentStatus:
-    user_id = user["sub"] if user else None
+    user_id = user["sub"]
     transcript = await transcripts_controller.get_by_id_for_http(
         transcript_id, user_id=user_id
     )
+    if transcript.user_id is not None and transcript.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Not authorized")
 
     if not transcript:
         raise HTTPException(status_code=404, detail="Transcript not found")