feat: brady bunch

.gitleaksignore

@@ -3,3 +3,4 @@ docs/docs/installation/auth-setup.md:curl-auth-header:250
 docs/docs/installation/daily-setup.md:curl-auth-header:277
 gpu/self_hosted/DEV_SETUP.md:curl-auth-header:74
 gpu/self_hosted/DEV_SETUP.md:curl-auth-header:83
+server/reflector/worker/process.py:generic-api-key:465

server/migrations/versions/1b1e6a6fc465_add_cloud_recording_support.py

@@ -0,0 +1,40 @@
+"""add cloud recording support
+
+Revision ID: 1b1e6a6fc465
+Revises: bd3a729bb379
+Create Date: 2026-01-09 17:17:33.535620
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "1b1e6a6fc465"
+down_revision: Union[str, None] = "bd3a729bb379"
+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.add_column(
+            sa.Column("daily_composed_video_s3_key", sa.String(), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column("daily_composed_video_duration", sa.Integer(), 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.drop_column("daily_composed_video_duration")
+        batch_op.drop_column("daily_composed_video_s3_key")
+
+    # ### end Alembic commands ###

server/reflector/dailyco_api/__init__.py

@@ -3,7 +3,7 @@
 """
 
 # Client
-from .client import DailyApiClient, DailyApiError
+from .client import DailyApiClient, DailyApiError, RecordingType
 
 # Request models
 from .requests import (
@@ -64,6 +64,7 @@
     # Client
     "DailyApiClient",
     "DailyApiError",
+    "RecordingType",
     # Requests
     "CreateRoomRequest",
     "RoomProperties",

server/reflector/dailyco_api/client.py

@@ -7,7 +7,8 @@
 """
 
 from http import HTTPStatus
-from typing import Any
+from typing import Any, Literal
+from uuid import UUID
 
 import httpx
 import structlog
@@ -32,6 +33,8 @@
 
 logger = structlog.get_logger(__name__)
 
+RecordingType = Literal["cloud", "raw-tracks"]
+
 
 class DailyApiError(Exception):
     """Daily.co API error with full request/response context."""
@@ -395,6 +398,38 @@ async def list_recordings(
 
         return [RecordingResponse(**r) for r in data["data"]]
 
+    async def start_recording(
+        self,
+        room_name: NonEmptyString,
+        recording_type: RecordingType,
+        instance_id: UUID,
+    ) -> dict[str, Any]:
+        """Start recording via REST API.
+
+        Reference: https://docs.daily.co/reference/rest-api/rooms/recordings/start
+
+        Args:
+            room_name: Daily.co room name
+            recording_type: Recording type
+            instance_id: UUID for this recording session
+
+        Returns:
+            Recording start confirmation from Daily.co API
+
+        Raises:
+            DailyApiError: If API request fails
+        """
+        client = await self._get_client()
+        response = await client.post(
+            f"{self.base_url}/rooms/{room_name}/recordings/start",
+            headers=self.headers,
+            json={
+                "type": recording_type,
+                "instanceId": str(instance_id),
+            },
+        )
+        return await self._handle_response(response, "start_recording")
+
     # ============================================================================
     # MEETING TOKENS
     # ============================================================================

server/reflector/dailyco_api/instance_id.py

@@ -0,0 +1,37 @@
+"""
+Daily.co recording instanceId generation utilities.
+
+Deterministic instance ID generation for cloud and raw-tracks recordings.
+MUST match frontend logic
+"""
+
+from uuid import UUID, uuid5
+
+from reflector.utils.string import NonEmptyString
+
+# Namespace UUID for UUIDv5 generation of raw-tracks instanceIds
+# DO NOT CHANGE: Breaks instanceId determinism across deployments and frontend/backend matching
+RAW_TRACKS_NAMESPACE = UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890")
+
+
+def generate_cloud_instance_id(meeting_id: NonEmptyString) -> UUID:
+    """
+    Generate instanceId for cloud recording.
+
+    Cloud recordings use meeting ID directly as instanceId.
+    This ensures each meeting has one unique cloud recording.
+    """
+    return UUID(meeting_id)
+
+
+def generate_raw_tracks_instance_id(meeting_id: NonEmptyString) -> UUID:
+    """
+    Generate instanceId for raw-tracks recording.
+
+    Raw-tracks recordings use UUIDv5(meeting_id, namespace) to ensure
+    different instanceId from cloud while remaining deterministic.
+
+    Daily.co requires cloud and raw-tracks to have different instanceIds
+    for concurrent recording.
+    """
+    return uuid5(RAW_TRACKS_NAMESPACE, meeting_id)

server/reflector/dailyco_api/requests.py

@@ -88,13 +88,6 @@ class MeetingTokenProperties(BaseModel):
     is_owner: bool = Field(
         default=False, description="Grant owner privileges to token holder"
     )
-    start_cloud_recording: bool = Field(
-        default=False, description="Automatically start cloud recording on join"
-    )
-    start_cloud_recording_opts: dict | None = Field(
-        default=None,
-        description="Options for startRecording when start_cloud_recording is true (e.g., maxDuration)",
-    )
     enable_recording_ui: bool = Field(
         default=True, description="Show recording controls in UI"
     )

server/reflector/dailyco_api/responses.py

@@ -116,6 +116,7 @@ class RecordingS3Info(BaseModel):
 
     bucket_name: NonEmptyString
     bucket_region: NonEmptyString
+    key: NonEmptyString | None = None
     endpoint: NonEmptyString | None = None
 
 
@@ -132,6 +133,9 @@ class RecordingResponse(BaseModel):
     id: NonEmptyString = Field(description="Recording identifier")
     room_name: NonEmptyString = Field(description="Room where recording occurred")
     start_ts: int = Field(description="Recording start timestamp (Unix epoch seconds)")
+    type: Literal["cloud", "raw-tracks"] | None = Field(
+        None, description="Recording type (may be missing from API)"
+    )
     status: RecordingStatus = Field(
         description="Recording status ('in-progress' or 'finished')"
     )
@@ -145,6 +149,9 @@ class RecordingResponse(BaseModel):
         None, description="Token for sharing recording"
     )
     s3: RecordingS3Info | None = Field(None, description="S3 bucket information")
+    s3key: NonEmptyString | None = Field(
+        None, description="S3 key for cloud recordings (top-level field)"
+    )
     tracks: list[DailyTrack] = Field(
         default_factory=list,
         description="Track list for raw-tracks recordings (always array, never null)",

server/reflector/db/meetings.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, Literal
 
 import sqlalchemy as sa
@@ -9,7 +9,7 @@
 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.utils.string import NonEmptyString, assert_equal
 
 meetings = sa.Table(
     "meeting",
@@ -63,6 +63,9 @@
         nullable=False,
         server_default=assert_equal(WHEREBY_PLATFORM, "whereby"),
     ),
+    # Daily.co composed video (Brady Bunch grid layout) - Daily.co only, not Whereby
+    sa.Column("daily_composed_video_s3_key", sa.String, nullable=True),
+    sa.Column("daily_composed_video_duration", sa.Integer, nullable=True),
     sa.Index("idx_meeting_room_id", "room_id"),
     sa.Index("idx_meeting_calendar_event", "calendar_event_id"),
 )
@@ -110,6 +113,9 @@ class Meeting(BaseModel):
     calendar_event_id: str | None = None
     calendar_metadata: dict[str, Any] | None = None
     platform: Platform = WHEREBY_PLATFORM
+    # Daily.co composed video (Brady Bunch grid) - Daily.co only
+    daily_composed_video_s3_key: str | None = None
+    daily_composed_video_duration: int | None = None
 
 
 class MeetingController:
@@ -171,6 +177,90 @@ async def get_by_room_name(
             return None
         return Meeting(**result)
 
+    async def get_by_room_name_all(self, room_name: str) -> list[Meeting]:
+        """Get all meetings for a room name (not just most recent)."""
+        query = meetings.select().where(meetings.c.room_name == room_name)
+        results = await get_database().fetch_all(query)
+        return [Meeting(**r) for r in results]
+
+    async def get_by_room_name_and_time(
+        self,
+        room_name: NonEmptyString,
+        recording_start: datetime,
+        time_window_hours: int = 168,
+    ) -> Meeting | None:
+        """
+        Get meeting by room name closest to recording timestamp.
+
+        HACK ALERT: Daily.co doesn't return instanceId in recordings API response,
+        and mtgSessionId is separate from our instanceId. Time-based matching is
+        the least-bad workaround.
+
+        This handles edge case of duplicate room_name values in DB (race conditions,
+        double-clicks, etc.) by matching based on temporal proximity.
+
+        Algorithm:
+        1. Find meetings within time_window_hours of recording_start
+        2. Return meeting with start_date closest to recording_start
+        3. If tie, return first by meeting.id (deterministic)
+
+        Args:
+            room_name: Daily.co room name from recording
+            recording_start: Timezone-aware datetime from recording.start_ts
+            time_window_hours: Search window (default 168 = 1 week)
+
+        Returns:
+            Meeting closest to recording timestamp, or None if no matches
+
+        Failure modes:
+        - Multiple meetings in same room within ~5 minutes: picks closest
+        - All meetings outside time window: returns None
+        - Clock skew between Daily.co and DB: 1-week window tolerates this
+
+        Why 1 week window:
+        - Handles webhook failures (recording discovered days later)
+        - Tolerates clock skew
+        - Rejects unrelated meetings from weeks ago
+
+        """
+        # Validate timezone-aware datetime
+        if recording_start.tzinfo is None:
+            raise ValueError(
+                f"recording_start must be timezone-aware, got naive datetime: {recording_start}"
+            )
+
+        window_start = recording_start - timedelta(hours=time_window_hours)
+        window_end = recording_start + timedelta(hours=time_window_hours)
+
+        query = (
+            meetings.select()
+            .where(
+                sa.and_(
+                    meetings.c.room_name == room_name,
+                    meetings.c.start_date >= window_start,
+                    meetings.c.start_date <= window_end,
+                )
+            )
+            .order_by(meetings.c.start_date)
+        )
+
+        results = await get_database().fetch_all(query)
+        if not results:
+            return None
+
+        candidates = [Meeting(**r) for r in results]
+
+        # Find meeting with start_date closest to recording_start
+        closest = min(
+            candidates,
+            key=lambda m: (
+                abs((m.start_date - recording_start).total_seconds()),
+                m.id,  # Tie-breaker: deterministic by UUID
+            ),
+        )
+
+        return closest
+
     async def get_active(self, room: Room, current_time: datetime) -> Meeting | None:
         """
         Get latest active meeting for a room.
@@ -260,6 +350,44 @@ 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 set_cloud_recording_if_missing(
+        self,
+        meeting_id: NonEmptyString,
+        s3_key: NonEmptyString,
+        duration: int,
+    ) -> bool:
+        """
+        Set cloud recording only if not already set.
+
+        Returns True if updated, False if already set.
+        Prevents webhook/polling race condition via atomic WHERE clause.
+        """
+        # Check current value before update to detect actual change
+        meeting_before = await self.get_by_id(meeting_id)
+        if not meeting_before:
+            return False
+
+        was_null = meeting_before.daily_composed_video_s3_key is None
+
+        query = (
+            meetings.update()
+            .where(
+                sa.and_(
+                    meetings.c.id == meeting_id,
+                    meetings.c.daily_composed_video_s3_key.is_(None),
+                )
+            )
+            .values(
+                daily_composed_video_s3_key=s3_key,
+                daily_composed_video_duration=duration,
+            )
+        )
+        await get_database().execute(query)
+
+        # Return True only if value was NULL before (actual update occurred)
+        # If was_null=False, the WHERE clause prevented the update
+        return was_null
+
     async def increment_num_clients(self, meeting_id: str) -> None:
         """Atomically increment participant count."""
         query = (

server/reflector/db/recordings.py

@@ -7,6 +7,7 @@
 
 from reflector.db import get_database, metadata
 from reflector.utils import generate_uuid4
+from reflector.utils.string import NonEmptyString
 
 recordings = sa.Table(
     "recording",
@@ -71,6 +72,19 @@ async def remove_by_id(self, id: str) -> None:
         query = recordings.delete().where(recordings.c.id == id)
         await get_database().execute(query)
 
+    async def set_meeting_id(
+        self,
+        recording_id: NonEmptyString,
+        meeting_id: NonEmptyString,
+    ) -> None:
+        """Link recording to meeting."""
+        query = (
+            recordings.update()
+            .where(recordings.c.id == recording_id)
+            .values(meeting_id=meeting_id)
+        )
+        await get_database().execute(query)
+
     # no check for existence
     async def get_by_ids(self, recording_ids: list[str]) -> list[Recording]:
         if not recording_ids: