feat: brady bunch (#816)

* brady bunch PRD/tasks

* clean dead daily.co code

* brady bunch prototype (no-mistakes)

* brady bunch prototype (no-mistakes) review

* self-review

* daily poll time match (no-mistakes)

* daily poll self-review (no-mistakes)

* daily poll self-review (no-mistakes)

* daily co doc

* cleanup

* cleanup

* self-review (no-mistakes)

* self-review (no-mistakes)

* self-review

* self-review

* ui typefix

* dupe calls error handling proper

* daily reflector data model doc

* logging style fix

* migration merge

---------

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

server/reflector/views/daily.py

@@ -19,6 +19,7 @@ from reflector.video_platforms.factory import create_platform_client
 from reflector.worker.process import (
     poll_daily_room_presence_task,
     process_multitrack_recording,
+    store_cloud_recording,
 )
 
 router = APIRouter()
@@ -174,46 +175,64 @@ async def _handle_recording_started(event: RecordingStartedEvent):
 async def _handle_recording_ready(event: RecordingReadyEvent):
     room_name = event.payload.room_name
     recording_id = event.payload.recording_id
-    tracks = event.payload.tracks
-
-    if not tracks:
-        logger.warning(
-            "recording.ready-to-download: missing tracks",
-            room_name=room_name,
-            recording_id=recording_id,
-            payload=event.payload,
-        )
-        return
+    recording_type = event.payload.type
 
     logger.info(
         "Recording ready for download",
         room_name=room_name,
         recording_id=recording_id,
-        num_tracks=len(tracks),
+        recording_type=recording_type,
         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"
-        )
+        logger.error("DAILYCO_STORAGE_AWS_BUCKET_NAME not configured")
         return
 
-    track_keys = [t.s3Key for t in tracks if t.type == "audio"]
+    if recording_type == "cloud":
+        await store_cloud_recording(
+            recording_id=recording_id,
+            room_name=room_name,
+            s3_key=event.payload.s3_key,
+            duration=event.payload.duration,
+            start_ts=event.payload.start_ts,
+            source="webhook",
+        )
 
-    logger.info(
-        "Recording webhook queuing processing",
-        recording_id=recording_id,
-        room_name=room_name,
-    )
+    elif recording_type == "raw-tracks":
+        tracks = event.payload.tracks
+        if not tracks:
+            logger.warning(
+                "raw-tracks recording: missing tracks array",
+                room_name=room_name,
+                recording_id=recording_id,
+            )
+            return
 
-    process_multitrack_recording.delay(
-        bucket_name=bucket_name,
-        daily_room_name=room_name,
-        recording_id=recording_id,
-        track_keys=track_keys,
-    )
+        track_keys = [t.s3Key for t in tracks if t.type == "audio"]
+
+        logger.info(
+            "Raw-tracks recording queuing processing",
+            recording_id=recording_id,
+            room_name=room_name,
+            num_tracks=len(track_keys),
+        )
+
+        process_multitrack_recording.delay(
+            bucket_name=bucket_name,
+            daily_room_name=room_name,
+            recording_id=recording_id,
+            track_keys=track_keys,
+            recording_start_ts=event.payload.start_ts,
+        )
+
+    else:
+        logger.warning(
+            "Unknown recording type",
+            recording_type=recording_type,
+            recording_id=recording_id,
+        )
 
 
 async def _handle_recording_error(event: RecordingErrorEvent):

server/reflector/views/meetings.py

@@ -1,16 +1,23 @@
+import json
 from datetime import datetime, timezone
-from typing import Annotated, Optional
+from typing import Annotated, Any, Optional
+from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 import reflector.auth as auth
+from reflector.dailyco_api import RecordingType
+from reflector.dailyco_api.client import DailyApiError
 from reflector.db.meetings import (
     MeetingConsent,
     meeting_consent_controller,
     meetings_controller,
 )
 from reflector.db.rooms import rooms_controller
+from reflector.logger import logger
+from reflector.utils.string import NonEmptyString
+from reflector.video_platforms.factory import create_platform_client
 
 router = APIRouter()
 
@@ -73,3 +80,72 @@ async def meeting_deactivate(
     await meetings_controller.update_meeting(meeting_id, is_active=False)
 
     return {"status": "success", "meeting_id": meeting_id}
+
+
+class StartRecordingRequest(BaseModel):
+    type: RecordingType
+    instanceId: UUID
+
+
+@router.post("/meetings/{meeting_id}/recordings/start")
+async def start_recording(
+    meeting_id: NonEmptyString, body: StartRecordingRequest
+) -> dict[str, Any]:
+    """Start cloud or raw-tracks recording via Daily.co REST API.
+
+    Both cloud and raw-tracks are started via REST API to bypass enable_recording limitation of allowing only 1 recording at a time.
+    Uses different instanceIds for cloud vs raw-tracks (same won't work)
+
+    Note: No authentication required - anonymous users supported. TODO this is a DOS vector
+    """
+    meeting = await meetings_controller.get_by_id(meeting_id)
+    if not meeting:
+        raise HTTPException(status_code=404, detail="Meeting not found")
+
+    log = logger.bind(
+        meeting_id=meeting_id,
+        room_name=meeting.room_name,
+        recording_type=body.type,
+        instance_id=body.instanceId,
+    )
+
+    try:
+        client = create_platform_client("daily")
+        result = await client.start_recording(
+            room_name=meeting.room_name,
+            recording_type=body.type,
+            instance_id=body.instanceId,
+        )
+
+        log.info(f"Started {body.type} recording via REST API")
+
+        return {"status": "ok", "result": result}
+
+    except DailyApiError as e:
+        # Parse Daily.co error response to detect "has an active stream"
+        try:
+            error_body = json.loads(e.response_body)
+            error_info = error_body.get("info", "")
+
+            # "has an active stream" means recording already started by another participant
+            # This is SUCCESS from business logic perspective - return 200
+            if "has an active stream" in error_info:
+                log.info(
+                    f"{body.type} recording already active (started by another participant)"
+                )
+                return {"status": "already_active", "instanceId": str(body.instanceId)}
+        except (json.JSONDecodeError, KeyError):
+            pass  # Fall through to error handling
+
+        # All other Daily.co API errors
+        log.error(f"Failed to start {body.type} recording", error=str(e))
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start recording: {str(e)}"
+        )
+
+    except Exception as e:
+        # Non-Daily.co errors
+        log.error(f"Failed to start {body.type} recording", error=str(e))
+        raise HTTPException(
+            status_code=500, detail=f"Failed to start recording: {str(e)}"
+        )

server/reflector/views/rooms.py

@@ -73,6 +73,8 @@ class Meeting(BaseModel):
     calendar_event_id: str | None = None
     calendar_metadata: dict[str, Any] | None = None
     platform: Platform
+    daily_composed_video_s3_key: str | None = None
+    daily_composed_video_duration: int | None = None
 
 
 class CreateRoom(BaseModel):
@@ -586,7 +588,6 @@ async def rooms_join_meeting(
         )
         token = await client.create_meeting_token(
             meeting.room_name,
-            start_cloud_recording=meeting.recording_type == "cloud",
             enable_recording_ui=enable_recording_ui,
             user_id=user_id,
             is_owner=user_id == room.user_id,