Compare commits

..

3 Commits

Author SHA1 Message Date
Igor Loskutov
d5ca2345d4 wip 2026-02-06 18:10:33 -05:00
Igor Loskutov
f6594448e6 regenerate api 2026-02-06 14:02:08 -05:00
Igor Loskutov
08462338de fix: prevent presence race condition during WebRTC handshake
Add /joining, /joined, and /leave endpoints to track user join intent
and trigger presence updates.

Backend:
- Add pending_joins Redis module with 30s TTL
- Add /joining endpoint (before WebRTC handshake)
- Add /joined endpoint (after connection, triggers presence poll)
- Add /leave endpoint (on tab close, triggers presence poll)
- Check for pending joins before deactivating meetings in worker

Frontend:
- Generate unique connectionId per browser tab
- Call /joining before Daily.co join, /joined after connection
- Add beforeunload handler calling /leave via sendBeacon
2026-02-06 12:36:03 -05:00
15 changed files with 2123 additions and 690 deletions

View File

@@ -0,0 +1,17 @@
"""Presence tracking for meetings."""
from reflector.presence.pending_joins import (
PENDING_JOIN_PREFIX,
PENDING_JOIN_TTL,
create_pending_join,
delete_pending_join,
has_pending_joins,
)
__all__ = [
"PENDING_JOIN_PREFIX",
"PENDING_JOIN_TTL",
"create_pending_join",
"delete_pending_join",
"has_pending_joins",
]

View File

@@ -0,0 +1,59 @@
"""Track pending join intents in Redis.
When a user signals intent to join a meeting (before WebRTC handshake completes),
we store a pending join record. This prevents the meeting from being deactivated
while users are still connecting.
"""
import time
from redis.asyncio import Redis
from reflector.logger import logger
PENDING_JOIN_TTL = 30 # seconds
PENDING_JOIN_PREFIX = "pending_join"
# Max keys to scan per Redis SCAN iteration
SCAN_BATCH_SIZE = 100
async def create_pending_join(redis: Redis, meeting_id: str, user_id: str) -> None:
"""Create a pending join record. Called before WebRTC handshake."""
key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
log = logger.bind(meeting_id=meeting_id, user_id=user_id, key=key)
await redis.setex(key, PENDING_JOIN_TTL, str(time.time()))
log.debug("Created pending join")
async def delete_pending_join(redis: Redis, meeting_id: str, user_id: str) -> None:
"""Delete pending join. Called after WebRTC connection established."""
key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
log = logger.bind(meeting_id=meeting_id, user_id=user_id, key=key)
await redis.delete(key)
log.debug("Deleted pending join")
async def has_pending_joins(redis: Redis, meeting_id: str) -> bool:
"""Check if meeting has any pending joins.
Uses Redis SCAN to iterate through all keys matching the pattern.
Properly iterates until cursor returns 0 to ensure all keys are checked.
"""
pattern = f"{PENDING_JOIN_PREFIX}:{meeting_id}:*"
log = logger.bind(meeting_id=meeting_id, pattern=pattern)
cursor = 0
iterations = 0
while True:
cursor, keys = await redis.scan(
cursor=cursor, match=pattern, count=SCAN_BATCH_SIZE
)
iterations += 1
if keys:
log.debug("Found pending joins", count=len(keys), iterations=iterations)
return True
if cursor == 0:
break
log.debug("No pending joins found", iterations=iterations)
return False

View File

@@ -129,10 +129,6 @@ class DailyClient(VideoPlatformClient):
"""Get room presence/session data for a Daily.co room.""" """Get room presence/session data for a Daily.co room."""
return await self._api_client.get_room_presence(room_name) return await self._api_client.get_room_presence(room_name)
async def delete_room(self, room_name: str) -> None:
"""Delete a Daily.co room (idempotent - succeeds even if room doesn't exist)."""
return await self._api_client.delete_room(room_name)
async def get_meeting_participants( async def get_meeting_participants(
self, meeting_id: str self, meeting_id: str
) -> MeetingParticipantsResponse: ) -> MeetingParticipantsResponse:

View File

@@ -1,4 +1,5 @@
import json import json
from datetime import datetime, timezone
from typing import assert_never from typing import assert_never
from fastapi import APIRouter, HTTPException, Request from fastapi import APIRouter, HTTPException, Request
@@ -12,6 +13,9 @@ from reflector.dailyco_api import (
RecordingReadyEvent, RecordingReadyEvent,
RecordingStartedEvent, RecordingStartedEvent,
) )
from reflector.db.daily_participant_sessions import (
daily_participant_sessions_controller,
)
from reflector.db.meetings import meetings_controller from reflector.db.meetings import meetings_controller
from reflector.logger import logger as _logger from reflector.logger import logger as _logger
from reflector.settings import settings from reflector.settings import settings
@@ -141,15 +145,57 @@ async def _handle_participant_joined(event: ParticipantJoinedEvent):
async def _handle_participant_left(event: ParticipantLeftEvent): async def _handle_participant_left(event: ParticipantLeftEvent):
"""Queue poll task for presence reconciliation.""" """Close session directly on webhook and update num_clients.
await _queue_poll_for_room(
event.payload.room_name, The webhook IS the authoritative signal that a participant left.
"participant.left", We close the session immediately rather than polling Daily.co API,
event.payload.user_id, which avoids the race where the API still shows the participant.
event.payload.session_id, A delayed reconciliation poll is queued as a safety net.
duration=event.payload.duration, """
room_name = event.payload.room_name
if not room_name:
logger.warning("participant.left: no room in payload")
return
meeting = await meetings_controller.get_by_room_name(room_name)
if not meeting:
logger.warning("participant.left: meeting not found", room_name=room_name)
return
log = logger.bind(
meeting_id=meeting.id,
room_name=room_name,
session_id=event.payload.session_id,
user_id=event.payload.user_id,
) )
existing = await daily_participant_sessions_controller.get_open_session(
meeting.id, event.payload.session_id
)
if existing:
now = datetime.now(timezone.utc)
await daily_participant_sessions_controller.batch_close_sessions(
[existing.id], left_at=now
)
active = await daily_participant_sessions_controller.get_active_by_meeting(
meeting.id
)
await meetings_controller.update_meeting(meeting.id, num_clients=len(active))
log.info(
"Participant left - session closed",
remaining_clients=len(active),
duration=event.payload.duration,
)
else:
log.info(
"Participant left - no open session found, skipping direct close",
duration=event.payload.duration,
)
# Delayed reconciliation poll as safety net
poll_daily_room_presence_task.apply_async(args=[meeting.id], countdown=5)
async def _handle_recording_started(event: RecordingStartedEvent): async def _handle_recording_started(event: RecordingStartedEvent):
room_name = event.payload.room_name room_name = event.payload.room_name

View File

@@ -1,9 +1,9 @@
import logging import json
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from enum import Enum from enum import Enum
from typing import Annotated, Any, Literal, Optional from typing import Annotated, Any, Literal, Optional
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi_pagination import Page from fastapi_pagination import Page
from fastapi_pagination.ext.databases import apaginate from fastapi_pagination.ext.databases import apaginate
from pydantic import BaseModel from pydantic import BaseModel
@@ -12,19 +12,24 @@ from redis.exceptions import LockError
import reflector.auth as auth import reflector.auth as auth
from reflector.db import get_database from reflector.db import get_database
from reflector.db.calendar_events import calendar_events_controller from reflector.db.calendar_events import calendar_events_controller
from reflector.db.daily_participant_sessions import (
DailyParticipantSession,
daily_participant_sessions_controller,
)
from reflector.db.meetings import meetings_controller from reflector.db.meetings import meetings_controller
from reflector.db.rooms import rooms_controller from reflector.db.rooms import rooms_controller
from reflector.redis_cache import RedisAsyncLock from reflector.logger import logger
from reflector.presence.pending_joins import create_pending_join, delete_pending_join
from reflector.redis_cache import RedisAsyncLock, get_async_redis_client
from reflector.schemas.platform import Platform from reflector.schemas.platform import Platform
from reflector.services.ics_sync import ics_sync_service from reflector.services.ics_sync import ics_sync_service
from reflector.settings import settings from reflector.settings import settings
from reflector.utils.string import NonEmptyString
from reflector.utils.url import add_query_param from reflector.utils.url import add_query_param
from reflector.video_platforms.factory import create_platform_client from reflector.video_platforms.factory import create_platform_client
from reflector.worker.process import poll_daily_room_presence_task from reflector.worker.process import poll_daily_room_presence_task
from reflector.worker.webhook import test_webhook from reflector.worker.webhook import test_webhook
logger = logging.getLogger(__name__)
class Room(BaseModel): class Room(BaseModel):
id: str id: str
@@ -366,53 +371,6 @@ async def rooms_create_meeting(
return meeting return meeting
@router.post("/rooms/{room_name}/meetings/{meeting_id}/joined")
async def rooms_joined_meeting(
room_name: str,
meeting_id: str,
):
"""Trigger presence poll (ideally when user actually joins meeting in Daily iframe)"""
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 meeting.platform == "daily":
poll_daily_room_presence_task.delay(meeting_id)
return {"status": "ok"}
@router.post("/rooms/{room_name}/meetings/{meeting_id}/leave")
async def rooms_leave_meeting(
room_name: str,
meeting_id: str,
delay_seconds: int = 2,
):
"""Trigger presence recheck when user leaves meeting (e.g., tab close/navigation).
Queues presence poll with optional delay to allow Daily.co to detect disconnect.
"""
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 meeting.platform == "daily":
poll_daily_room_presence_task.apply_async(
args=[meeting_id],
countdown=delay_seconds,
)
return {"status": "ok"}
@router.post("/rooms/{room_id}/webhook/test", response_model=WebhookTestResult) @router.post("/rooms/{room_id}/webhook/test", response_model=WebhookTestResult)
async def rooms_test_webhook( async def rooms_test_webhook(
room_id: str, room_id: str,
@@ -645,3 +603,221 @@ async def rooms_join_meeting(
meeting.room_url = add_query_param(meeting.room_url, "t", token) meeting.room_url = add_query_param(meeting.room_url, "t", token)
return meeting return meeting
class JoiningRequest(BaseModel):
"""Request body for /joining endpoint (before WebRTC handshake)."""
connection_id: NonEmptyString
"""Unique identifier for this connection. Generated by client via crypto.randomUUID()."""
class JoiningResponse(BaseModel):
status: Literal["ok"]
class JoinedRequest(BaseModel):
"""Request body for /joined endpoint (after WebRTC connection established)."""
connection_id: NonEmptyString
"""Must match the connection_id sent to /joining."""
session_id: NonEmptyString | None = None
"""Daily.co session_id for direct session creation. Optional for backward compat."""
user_name: str | None = None
"""Display name from Daily.co participant data."""
class JoinedResponse(BaseModel):
status: Literal["ok"]
def _get_pending_join_key(
user: Optional[auth.UserInfo], connection_id: NonEmptyString
) -> str:
"""Get a unique key for pending join tracking.
Uses user ID for authenticated users, connection_id for anonymous users.
This ensures each browser tab has its own unique pending join record.
"""
if user:
return f"{user['sub']}:{connection_id}"
return f"anon:{connection_id}"
@router.post(
"/rooms/{room_name}/meetings/{meeting_id}/joining", response_model=JoiningResponse
)
async def meeting_joining(
room_name: str,
meeting_id: str,
body: JoiningRequest,
user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
) -> JoiningResponse:
"""Signal intent to join meeting. Called before WebRTC handshake starts.
This creates a pending join record that prevents the meeting from being
deactivated while the WebRTC handshake is in progress. The record expires
automatically after 30 seconds if the connection is not established.
"""
log = logger.bind(
room_name=room_name, meeting_id=meeting_id, connection_id=body.connection_id
)
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")
join_key = _get_pending_join_key(user, body.connection_id)
redis = await get_async_redis_client()
try:
await create_pending_join(redis, meeting_id, join_key)
log.debug("Created pending join intent", join_key=join_key)
finally:
await redis.aclose()
return JoiningResponse(status="ok")
@router.post(
"/rooms/{room_name}/meetings/{meeting_id}/joined", response_model=JoinedResponse
)
async def meeting_joined(
room_name: str,
meeting_id: str,
body: JoinedRequest,
user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
) -> JoinedResponse:
"""Signal that WebRTC connection is established.
This clears the pending join record, confirming the user has successfully
connected to the meeting. Safe to call even if meeting was deactivated
during the handshake (idempotent cleanup).
"""
log = logger.bind(
room_name=room_name, meeting_id=meeting_id, connection_id=body.connection_id
)
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")
# Note: We don't check is_active here - the /joined call is a cleanup operation
# and should succeed even if the meeting was deactivated during the handshake
join_key = _get_pending_join_key(user, body.connection_id)
redis = await get_async_redis_client()
try:
await delete_pending_join(redis, meeting_id, join_key)
log.debug("Cleared pending join intent", join_key=join_key)
finally:
await redis.aclose()
# Create session directly when session_id provided (instant presence update)
if body.session_id and meeting.platform == "daily":
session = DailyParticipantSession(
id=f"{meeting.id}:{body.session_id}",
meeting_id=meeting.id,
room_id=room.id,
session_id=body.session_id,
user_id=user["sub"] if user else None,
user_name=body.user_name or "Anonymous",
joined_at=datetime.now(timezone.utc),
)
await daily_participant_sessions_controller.batch_upsert_sessions([session])
active = await daily_participant_sessions_controller.get_active_by_meeting(
meeting.id
)
await meetings_controller.update_meeting(meeting.id, num_clients=len(active))
log.info(
"Session created directly",
session_id=body.session_id,
num_clients=len(active),
)
# Trigger presence poll as reconciliation safety net
if meeting.platform == "daily":
poll_daily_room_presence_task.apply_async(args=[meeting_id], countdown=3)
return JoinedResponse(status="ok")
class LeaveResponse(BaseModel):
status: Literal["ok"]
@router.post(
"/rooms/{room_name}/meetings/{meeting_id}/leave", response_model=LeaveResponse
)
async def meeting_leave(
room_name: str,
meeting_id: str,
request: Request,
user: Annotated[Optional[auth.UserInfo], Depends(auth.current_user_optional)],
) -> LeaveResponse:
"""Trigger presence update when user leaves meeting.
When session_id is provided in the body, closes the session directly
for instant presence update. Falls back to polling when session_id
is not available (e.g., sendBeacon without frame access).
Called on tab close/navigation via sendBeacon().
"""
# Parse session_id from body (sendBeacon may send text/plain or no body)
session_id: str | None = None
try:
body_bytes = await request.body()
if body_bytes:
data = json.loads(body_bytes)
raw = data.get("session_id")
if isinstance(raw, str) and raw.strip():
session_id = raw.strip()
except Exception:
pass
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")
# Close session directly when session_id provided
session_closed = False
if session_id and meeting.platform == "daily":
existing = await daily_participant_sessions_controller.get_open_session(
meeting.id, session_id
)
if existing:
await daily_participant_sessions_controller.batch_close_sessions(
[existing.id], left_at=datetime.now(timezone.utc)
)
active = await daily_participant_sessions_controller.get_active_by_meeting(
meeting.id
)
await meetings_controller.update_meeting(
meeting.id, num_clients=len(active)
)
session_closed = True
# Only queue poll if we couldn't close directly — the poll runs before
# Daily.co API removes the participant, which would undo our correct count
if meeting.platform == "daily" and not session_closed:
poll_daily_room_presence_task.apply_async(args=[meeting_id], countdown=3)
return LeaveResponse(status="ok")

View File

@@ -31,9 +31,10 @@ from reflector.pipelines.main_multitrack_pipeline import (
task_pipeline_multitrack_process, task_pipeline_multitrack_process,
) )
from reflector.pipelines.topic_processing import EmptyPipeline from reflector.pipelines.topic_processing import EmptyPipeline
from reflector.presence.pending_joins import has_pending_joins
from reflector.processors import AudioFileWriterProcessor from reflector.processors import AudioFileWriterProcessor
from reflector.processors.audio_waveform_processor import AudioWaveformProcessor from reflector.processors.audio_waveform_processor import AudioWaveformProcessor
from reflector.redis_cache import RedisAsyncLock from reflector.redis_cache import RedisAsyncLock, get_async_redis_client
from reflector.settings import settings from reflector.settings import settings
from reflector.storage import get_transcripts_storage from reflector.storage import get_transcripts_storage
from reflector.utils.daily import ( from reflector.utils.daily import (
@@ -845,47 +846,15 @@ async def process_meetings():
end_date = end_date.replace(tzinfo=timezone.utc) end_date = end_date.replace(tzinfo=timezone.utc)
client = create_platform_client(meeting.platform) client = create_platform_client(meeting.platform)
has_active_sessions = False room_sessions = await client.get_room_sessions(meeting.room_name)
has_had_sessions = False
if meeting.platform == "daily": has_active_sessions = bool(
try: room_sessions and any(s.ended_at is None for s in room_sessions)
presence = await client.get_room_presence(meeting.room_name) )
has_active_sessions = presence.total_count > 0 has_had_sessions = bool(room_sessions)
logger_.info(
room_sessions = await client.get_room_sessions( f"has_active_sessions={has_active_sessions}, has_had_sessions={has_had_sessions}"
meeting.room_name )
)
has_had_sessions = bool(room_sessions)
logger_.info(
"Daily.co presence check",
has_active_sessions=has_active_sessions,
has_had_sessions=has_had_sessions,
presence_count=presence.total_count,
)
except Exception:
logger_.warning(
"Daily.co presence API failed, falling back to DB sessions",
exc_info=True,
)
room_sessions = await client.get_room_sessions(
meeting.room_name
)
has_active_sessions = bool(
room_sessions
and any(s.ended_at is None for s in room_sessions)
)
has_had_sessions = bool(room_sessions)
else:
room_sessions = await client.get_room_sessions(meeting.room_name)
has_active_sessions = bool(
room_sessions and any(s.ended_at is None for s in room_sessions)
)
has_had_sessions = bool(room_sessions)
logger_.info(
f"has_active_sessions={has_active_sessions}, has_had_sessions={has_had_sessions}"
)
if has_active_sessions: if has_active_sessions:
logger_.debug("Meeting still has active sessions, keep it") logger_.debug("Meeting still has active sessions, keep it")
@@ -901,23 +870,22 @@ async def process_meetings():
logger_.debug("Meeting not yet started, keep it") logger_.debug("Meeting not yet started, keep it")
if should_deactivate: if should_deactivate:
# Check for pending joins before deactivating
# Users might be in the process of connecting via WebRTC
redis = await get_async_redis_client()
try:
if await has_pending_joins(redis, meeting.id):
logger_.info(
"Meeting has pending joins, skipping deactivation"
)
continue
finally:
await redis.aclose()
await meetings_controller.update_meeting( await meetings_controller.update_meeting(
meeting.id, is_active=False meeting.id, is_active=False
) )
logger_.info("Meeting deactivated in database") logger_.info("Meeting is deactivated")
if meeting.platform == "daily":
try:
await client.delete_room(meeting.room_name)
logger_.info(
"Daily.co room deleted", room_name=meeting.room_name
)
except Exception:
logger_.warning(
"Failed to delete Daily.co room",
room_name=meeting.room_name,
exc_info=True,
)
processed_count += 1 processed_count += 1

View File

@@ -1,286 +0,0 @@
"""Unit tests for Daily.co presence-based meeting deactivation logic.
Tests the fix for split room race condition by verifying:
1. Real-time presence checking via Daily.co API
2. Room deletion when meetings deactivate
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch
import pytest
from reflector.dailyco_api.responses import (
RoomPresenceParticipant,
RoomPresenceResponse,
)
from reflector.db.daily_participant_sessions import (
DailyParticipantSession,
daily_participant_sessions_controller,
)
from reflector.db.meetings import meetings_controller
from reflector.db.rooms import rooms_controller
from reflector.video_platforms.daily import DailyClient
@pytest.fixture
async def daily_room_and_meeting():
"""Create test room and meeting for Daily platform."""
room = await rooms_controller.add(
name="test-daily",
user_id="test-user",
platform="daily",
zulip_auto_post=False,
zulip_stream="",
zulip_topic="",
is_locked=False,
room_mode="normal",
recording_type="cloud",
recording_trigger="automatic-2nd-participant",
is_shared=False,
)
current_time = datetime.now(timezone.utc)
end_time = current_time + timedelta(hours=2)
meeting = await meetings_controller.create(
id="test-meeting-id",
room_name="test-daily-20260129120000",
room_url="https://daily.co/test",
host_room_url="https://daily.co/test",
start_date=current_time,
end_date=end_time,
room=room,
)
return room, meeting
@pytest.mark.asyncio
async def test_daily_client_has_delete_room_method():
"""Verify DailyClient has delete_room method for cleanup."""
# Create a mock DailyClient
with patch("reflector.dailyco_api.client.DailyApiClient"):
from reflector.video_platforms.models import VideoPlatformConfig
config = VideoPlatformConfig(api_key="test-key", webhook_secret="test-secret")
client = DailyClient(config)
# Verify delete_room method exists
assert hasattr(client, "delete_room")
assert callable(getattr(client, "delete_room"))
@pytest.mark.asyncio
async def test_get_room_presence_returns_realtime_data(daily_room_and_meeting):
"""Test that get_room_presence returns real-time participant data."""
room, meeting = daily_room_and_meeting
# Mock Daily.co API response
mock_presence = RoomPresenceResponse(
total_count=2,
data=[
RoomPresenceParticipant(
room=meeting.room_name,
id="session-1",
userId="user-1",
userName="User One",
joinTime="2026-01-29T12:00:00.000Z",
duration=120,
),
RoomPresenceParticipant(
room=meeting.room_name,
id="session-2",
userId="user-2",
userName="User Two",
joinTime="2026-01-29T12:05:00.000Z",
duration=60,
),
],
)
with patch("reflector.dailyco_api.client.DailyApiClient") as mock_api:
from reflector.video_platforms.models import VideoPlatformConfig
config = VideoPlatformConfig(api_key="test-key", webhook_secret="test-secret")
client = DailyClient(config)
# Mock the API client method
client._api_client.get_room_presence = AsyncMock(return_value=mock_presence)
# Call get_room_presence
result = await client.get_room_presence(meeting.room_name)
# Verify it calls Daily.co API
client._api_client.get_room_presence.assert_called_once_with(meeting.room_name)
# Verify result contains real-time data
assert result.total_count == 2
assert len(result.data) == 2
assert result.data[0].id == "session-1"
assert result.data[1].id == "session-2"
@pytest.mark.asyncio
async def test_presence_shows_active_even_when_db_stale(daily_room_and_meeting):
"""Test that Daily.co presence API is source of truth, not stale DB sessions."""
room, meeting = daily_room_and_meeting
current_time = datetime.now(timezone.utc)
# Create stale DB session (left_at=NULL but user actually left)
session_id = f"{meeting.id}:stale-user:{int((current_time - timedelta(minutes=5)).timestamp() * 1000)}"
await daily_participant_sessions_controller.upsert_joined(
DailyParticipantSession(
id=session_id,
meeting_id=meeting.id,
room_id=room.id,
session_id="stale-daily-session",
user_name="Stale User",
user_id="stale-user",
joined_at=current_time - timedelta(minutes=5),
left_at=None, # Stale - shows active but user left
)
)
# Verify DB shows active session
db_sessions = await daily_participant_sessions_controller.get_active_by_meeting(
meeting.id
)
assert len(db_sessions) == 1
# But Daily.co API shows room is empty
mock_presence = RoomPresenceResponse(total_count=0, data=[])
with patch("reflector.dailyco_api.client.DailyApiClient"):
from reflector.video_platforms.models import VideoPlatformConfig
config = VideoPlatformConfig(api_key="test-key", webhook_secret="test-secret")
client = DailyClient(config)
client._api_client.get_room_presence = AsyncMock(return_value=mock_presence)
# Get real-time presence
presence = await client.get_room_presence(meeting.room_name)
# Real-time API shows no participants (truth)
assert presence.total_count == 0
assert len(presence.data) == 0
# DB shows 1 participant (stale)
assert len(db_sessions) == 1
# Implementation should trust presence API, not DB
@pytest.mark.asyncio
async def test_meeting_deactivation_logic_with_presence_empty():
"""Test the core deactivation decision logic when presence shows room empty."""
# This tests the logic that will be in process_meetings
# Simulate: DB shows stale active session
has_active_db_sessions = True # DB is stale
# Simulate: Daily.co presence API shows room empty
presence_count = 0 # Real-time truth
# Simulate: Meeting has been used before
has_had_sessions = True
# Decision logic (what process_meetings should do):
# - If presence API available: trust it
# - If presence shows empty AND has_had_sessions: deactivate
if presence_count == 0 and has_had_sessions:
should_deactivate = True
else:
should_deactivate = False
assert should_deactivate is True # Should deactivate despite stale DB
@pytest.mark.asyncio
async def test_meeting_deactivation_logic_with_presence_active():
"""Test that meetings stay active when presence shows participants."""
# Simulate: DB shows no sessions (not yet updated)
has_active_db_sessions = False # DB hasn't caught up
# Simulate: Daily.co presence API shows active participant
presence_count = 1 # Real-time truth
# Decision logic: presence shows activity, keep meeting active
if presence_count > 0:
should_deactivate = False
else:
should_deactivate = True
assert should_deactivate is False # Should stay active
@pytest.mark.asyncio
async def test_delete_room_called_on_deactivation(daily_room_and_meeting):
"""Test that Daily.co room is deleted when meeting deactivates."""
room, meeting = daily_room_and_meeting
with patch("reflector.dailyco_api.client.DailyApiClient"):
from reflector.video_platforms.models import VideoPlatformConfig
config = VideoPlatformConfig(api_key="test-key", webhook_secret="test-secret")
client = DailyClient(config)
# Mock delete_room API call
client._api_client.delete_room = AsyncMock()
# Simulate deactivation - should delete room
await client._api_client.delete_room(meeting.room_name)
# Verify delete was called
client._api_client.delete_room.assert_called_once_with(meeting.room_name)
@pytest.mark.asyncio
async def test_delete_room_idempotent_on_404():
"""Test that room deletion is idempotent (succeeds even if room doesn't exist)."""
from reflector.dailyco_api.client import DailyApiClient
# Create real client to test delete_room logic
client = DailyApiClient(api_key="test-key")
# Mock the HTTP client
mock_http_client = AsyncMock()
mock_response = AsyncMock()
mock_response.status_code = 404 # Room not found
mock_http_client.delete = AsyncMock(return_value=mock_response)
# Mock _get_client to return our mock
async def mock_get_client():
return mock_http_client
client._get_client = mock_get_client
# delete_room should succeed even on 404 (idempotent)
await client.delete_room("nonexistent-room")
# Verify delete was attempted
mock_http_client.delete.assert_called_once()
@pytest.mark.asyncio
async def test_api_failure_fallback_to_db_sessions():
"""Test that system falls back to DB sessions if Daily.co API fails."""
# Simulate: Daily.co API throws exception
api_exception = Exception("API unavailable")
# Simulate: DB shows active session
has_active_db_sessions = True
# Decision logic with fallback:
try:
presence_count = None
raise api_exception # Simulating API failure
except Exception:
# Fallback: use DB sessions (conservative - don't deactivate if unsure)
if has_active_db_sessions:
should_deactivate = False
else:
should_deactivate = True
assert should_deactivate is False # Conservative: keep active on API failure

View File

@@ -0,0 +1,213 @@
"""Tests for direct session close on participant.left webhook.
Verifies that _handle_participant_left:
1. Closes the session directly (authoritative signal)
2. Updates num_clients from remaining active sessions
3. Queues a delayed reconciliation poll as safety net
4. Handles missing session gracefully
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch
import pytest
from reflector.dailyco_api.webhooks import ParticipantLeftEvent, ParticipantLeftPayload
from reflector.db.daily_participant_sessions import DailyParticipantSession
from reflector.db.meetings import Meeting
from reflector.views.daily import _handle_participant_left
@pytest.fixture
def mock_meeting():
return Meeting(
id="meeting-123",
room_id="room-456",
room_name="test-room-20251118120000",
room_url="https://daily.co/test-room-20251118120000",
host_room_url="https://daily.co/test-room-20251118120000?t=host-token",
platform="daily",
num_clients=2,
is_active=True,
start_date=datetime.now(timezone.utc),
end_date=datetime.now(timezone.utc),
)
@pytest.fixture
def participant_left_event():
now = datetime.now(timezone.utc)
return ParticipantLeftEvent(
version="1.0.0",
type="participant.left",
id="evt-left-abc123",
payload=ParticipantLeftPayload(
room_name="test-room-20251118120000",
session_id="session-alice",
user_id="user-alice",
user_name="Alice",
joined_at=int((now - timedelta(minutes=10)).timestamp()),
duration=600,
),
event_ts=int(now.timestamp()),
)
@pytest.fixture
def existing_session():
now = datetime.now(timezone.utc)
return DailyParticipantSession(
id="meeting-123:session-alice",
meeting_id="meeting-123",
room_id="room-456",
session_id="session-alice",
user_id="user-alice",
user_name="Alice",
joined_at=now - timedelta(minutes=10),
left_at=None,
)
@pytest.mark.asyncio
@patch("reflector.views.daily.poll_daily_room_presence_task")
@patch("reflector.views.daily.meetings_controller")
@patch("reflector.views.daily.daily_participant_sessions_controller")
async def test_closes_session_and_updates_num_clients(
mock_sessions_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_meeting,
participant_left_event,
existing_session,
):
"""Webhook directly closes session and updates num_clients from remaining active count."""
mock_meetings_ctrl.get_by_room_name = AsyncMock(return_value=mock_meeting)
mock_sessions_ctrl.get_open_session = AsyncMock(return_value=existing_session)
mock_sessions_ctrl.batch_close_sessions = AsyncMock()
# One remaining active session after close
remaining = DailyParticipantSession(
id="meeting-123:session-bob",
meeting_id="meeting-123",
room_id="room-456",
session_id="session-bob",
user_id="user-bob",
user_name="Bob",
joined_at=datetime.now(timezone.utc),
left_at=None,
)
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(return_value=[remaining])
mock_meetings_ctrl.update_meeting = AsyncMock()
await _handle_participant_left(participant_left_event)
# Session closed
mock_sessions_ctrl.batch_close_sessions.assert_called_once()
closed_ids = mock_sessions_ctrl.batch_close_sessions.call_args.args[0]
assert closed_ids == [existing_session.id]
# num_clients updated to remaining count
mock_meetings_ctrl.update_meeting.assert_called_once_with(
mock_meeting.id, num_clients=1
)
# Delayed reconciliation poll queued
mock_poll_task.apply_async.assert_called_once()
call_kwargs = mock_poll_task.apply_async.call_args.kwargs
assert call_kwargs["countdown"] == 5
assert call_kwargs["args"] == [mock_meeting.id]
@pytest.mark.asyncio
@patch("reflector.views.daily.poll_daily_room_presence_task")
@patch("reflector.views.daily.meetings_controller")
@patch("reflector.views.daily.daily_participant_sessions_controller")
async def test_handles_missing_session(
mock_sessions_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_meeting,
participant_left_event,
):
"""No crash when session not found in DB — still queues reconciliation poll."""
mock_meetings_ctrl.get_by_room_name = AsyncMock(return_value=mock_meeting)
mock_sessions_ctrl.get_open_session = AsyncMock(return_value=None)
await _handle_participant_left(participant_left_event)
# No session close attempted
mock_sessions_ctrl.batch_close_sessions.assert_not_called()
# No num_clients update (no authoritative data without session)
mock_meetings_ctrl.update_meeting.assert_not_called()
# Still queues reconciliation poll
mock_poll_task.apply_async.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.views.daily.poll_daily_room_presence_task")
@patch("reflector.views.daily.meetings_controller")
@patch("reflector.views.daily.daily_participant_sessions_controller")
async def test_updates_num_clients_to_zero_when_last_participant_leaves(
mock_sessions_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_meeting,
participant_left_event,
existing_session,
):
"""num_clients set to 0 when no active sessions remain."""
mock_meetings_ctrl.get_by_room_name = AsyncMock(return_value=mock_meeting)
mock_sessions_ctrl.get_open_session = AsyncMock(return_value=existing_session)
mock_sessions_ctrl.batch_close_sessions = AsyncMock()
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(return_value=[])
mock_meetings_ctrl.update_meeting = AsyncMock()
await _handle_participant_left(participant_left_event)
mock_meetings_ctrl.update_meeting.assert_called_once_with(
mock_meeting.id, num_clients=0
)
@pytest.mark.asyncio
@patch("reflector.views.daily.poll_daily_room_presence_task")
@patch("reflector.views.daily.meetings_controller")
async def test_no_room_name_in_event(
mock_meetings_ctrl,
mock_poll_task,
):
"""No crash when room_name is missing from webhook payload."""
event = ParticipantLeftEvent(
version="1.0.0",
type="participant.left",
id="evt-left-no-room",
payload=ParticipantLeftPayload(
room_name=None,
session_id="session-x",
user_id="user-x",
user_name="X",
joined_at=int(datetime.now(timezone.utc).timestamp()),
duration=0,
),
event_ts=int(datetime.now(timezone.utc).timestamp()),
)
await _handle_participant_left(event)
mock_meetings_ctrl.get_by_room_name.assert_not_called()
mock_poll_task.apply_async.assert_not_called()
@pytest.mark.asyncio
@patch("reflector.views.daily.poll_daily_room_presence_task")
@patch("reflector.views.daily.meetings_controller")
async def test_meeting_not_found(
mock_meetings_ctrl,
mock_poll_task,
participant_left_event,
):
"""No crash when meeting not found for room_name."""
mock_meetings_ctrl.get_by_room_name = AsyncMock(return_value=None)
await _handle_participant_left(participant_left_event)
mock_poll_task.apply_async.assert_not_called()

View File

@@ -0,0 +1,339 @@
"""Tests for direct session management via /joined and /leave endpoints.
Verifies that:
1. /joined with session_id creates session directly, updates num_clients
2. /joined without session_id (backward compat) still works, queues poll
3. /leave with session_id closes session, updates num_clients
4. /leave without session_id falls back to poll
5. Duplicate /joined calls are idempotent (upsert)
6. /leave for already-closed session is a no-op
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from reflector.db.daily_participant_sessions import DailyParticipantSession
from reflector.db.meetings import Meeting
from reflector.views.rooms import (
JoinedRequest,
meeting_joined,
meeting_leave,
)
@pytest.fixture
def mock_room():
room = MagicMock()
room.id = "room-456"
room.name = "test-room"
room.platform = "daily"
return room
@pytest.fixture
def mock_meeting():
return Meeting(
id="meeting-123",
room_id="room-456",
room_name="test-room-20251118120000",
room_url="https://daily.co/test-room",
host_room_url="https://daily.co/test-room?t=host",
platform="daily",
num_clients=0,
is_active=True,
start_date=datetime.now(timezone.utc),
end_date=datetime.now(timezone.utc) + timedelta(hours=8),
)
@pytest.fixture
def mock_redis():
redis = AsyncMock()
redis.aclose = AsyncMock()
return redis
@pytest.fixture
def mock_request_with_session_id():
"""Mock Request with session_id in JSON body."""
request = AsyncMock()
request.body = AsyncMock(return_value=b'{"session_id": "session-abc"}')
return request
@pytest.fixture
def mock_request_empty_body():
"""Mock Request with empty JSON body (old frontend / no frame access)."""
request = AsyncMock()
request.body = AsyncMock(return_value=b"{}")
return request
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.delete_pending_join")
@patch("reflector.views.rooms.get_async_redis_client")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_joined_with_session_id_creates_session(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_redis_client,
mock_delete_pending,
mock_poll_task,
mock_room,
mock_meeting,
mock_redis,
):
"""session_id in /joined -> create session + update num_clients."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
mock_redis_client.return_value = mock_redis
mock_sessions_ctrl.batch_upsert_sessions = AsyncMock()
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(
return_value=[MagicMock()] # 1 active session
)
mock_meetings_ctrl.update_meeting = AsyncMock()
body = JoinedRequest(
connection_id="conn-1",
session_id="session-abc",
user_name="Alice",
)
result = await meeting_joined(
"test-room", "meeting-123", body, user={"sub": "user-1"}
)
assert result.status == "ok"
# Session created via upsert
mock_sessions_ctrl.batch_upsert_sessions.assert_called_once()
sessions = mock_sessions_ctrl.batch_upsert_sessions.call_args.args[0]
assert len(sessions) == 1
assert sessions[0].session_id == "session-abc"
assert sessions[0].meeting_id == "meeting-123"
assert sessions[0].room_id == "room-456"
assert sessions[0].user_name == "Alice"
assert sessions[0].user_id == "user-1"
assert sessions[0].id == "meeting-123:session-abc"
# num_clients updated
mock_meetings_ctrl.update_meeting.assert_called_once_with(
"meeting-123", num_clients=1
)
# Reconciliation poll still queued
mock_poll_task.apply_async.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.delete_pending_join")
@patch("reflector.views.rooms.get_async_redis_client")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_joined_without_session_id_backward_compat(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_redis_client,
mock_delete_pending,
mock_poll_task,
mock_room,
mock_meeting,
mock_redis,
):
"""No session_id in /joined -> no session create, still queues poll."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
mock_redis_client.return_value = mock_redis
body = JoinedRequest(connection_id="conn-1")
result = await meeting_joined(
"test-room", "meeting-123", body, user={"sub": "user-1"}
)
assert result.status == "ok"
mock_sessions_ctrl.batch_upsert_sessions.assert_not_called()
mock_poll_task.apply_async.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.delete_pending_join")
@patch("reflector.views.rooms.get_async_redis_client")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_joined_anonymous_user_sets_null_user_id(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_redis_client,
mock_delete_pending,
mock_poll_task,
mock_room,
mock_meeting,
mock_redis,
):
"""Anonymous user -> session.user_id is None, user_name defaults to 'Anonymous'."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
mock_redis_client.return_value = mock_redis
mock_sessions_ctrl.batch_upsert_sessions = AsyncMock()
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(return_value=[MagicMock()])
mock_meetings_ctrl.update_meeting = AsyncMock()
body = JoinedRequest(connection_id="conn-1", session_id="session-abc")
result = await meeting_joined("test-room", "meeting-123", body, user=None)
assert result.status == "ok"
sessions = mock_sessions_ctrl.batch_upsert_sessions.call_args.args[0]
assert sessions[0].user_id is None
assert sessions[0].user_name == "Anonymous"
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_leave_with_session_id_closes_session(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_room,
mock_meeting,
mock_request_with_session_id,
):
"""session_id in /leave -> close session + update num_clients."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
existing_session = DailyParticipantSession(
id="meeting-123:session-abc",
meeting_id="meeting-123",
room_id="room-456",
session_id="session-abc",
user_id="user-1",
user_name="Alice",
joined_at=datetime.now(timezone.utc) - timedelta(minutes=5),
left_at=None,
)
mock_sessions_ctrl.get_open_session = AsyncMock(return_value=existing_session)
mock_sessions_ctrl.batch_close_sessions = AsyncMock()
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(return_value=[])
mock_meetings_ctrl.update_meeting = AsyncMock()
result = await meeting_leave(
"test-room", "meeting-123", mock_request_with_session_id, user={"sub": "user-1"}
)
assert result.status == "ok"
# Session closed
mock_sessions_ctrl.batch_close_sessions.assert_called_once()
closed_ids = mock_sessions_ctrl.batch_close_sessions.call_args.args[0]
assert closed_ids == ["meeting-123:session-abc"]
# num_clients updated
mock_meetings_ctrl.update_meeting.assert_called_once_with(
"meeting-123", num_clients=0
)
# No poll — direct close is authoritative, poll would race with API latency
mock_poll_task.apply_async.assert_not_called()
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
async def test_leave_without_session_id_falls_back_to_poll(
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_room,
mock_meeting,
mock_request_empty_body,
):
"""No session_id in /leave -> just queues poll as before."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
result = await meeting_leave(
"test-room", "meeting-123", mock_request_empty_body, user=None
)
assert result.status == "ok"
mock_poll_task.apply_async.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.delete_pending_join")
@patch("reflector.views.rooms.get_async_redis_client")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_duplicate_joined_is_idempotent(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_redis_client,
mock_delete_pending,
mock_poll_task,
mock_room,
mock_meeting,
mock_redis,
):
"""Calling /joined twice with same session_id -> upsert both times, no error."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
mock_redis_client.return_value = mock_redis
mock_sessions_ctrl.batch_upsert_sessions = AsyncMock()
mock_sessions_ctrl.get_active_by_meeting = AsyncMock(return_value=[MagicMock()])
mock_meetings_ctrl.update_meeting = AsyncMock()
body = JoinedRequest(
connection_id="conn-1", session_id="session-abc", user_name="Alice"
)
await meeting_joined("test-room", "meeting-123", body, user={"sub": "user-1"})
await meeting_joined("test-room", "meeting-123", body, user={"sub": "user-1"})
assert mock_sessions_ctrl.batch_upsert_sessions.call_count == 2
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.meetings_controller")
@patch("reflector.views.rooms.rooms_controller")
@patch("reflector.views.rooms.daily_participant_sessions_controller")
async def test_leave_already_closed_session_is_noop(
mock_sessions_ctrl,
mock_rooms_ctrl,
mock_meetings_ctrl,
mock_poll_task,
mock_room,
mock_meeting,
mock_request_with_session_id,
):
"""/leave for already-closed session -> no close attempted, just poll."""
mock_rooms_ctrl.get_by_name = AsyncMock(return_value=mock_room)
mock_meetings_ctrl.get_by_id = AsyncMock(return_value=mock_meeting)
mock_sessions_ctrl.get_open_session = AsyncMock(return_value=None)
result = await meeting_leave(
"test-room", "meeting-123", mock_request_with_session_id, user=None
)
assert result.status == "ok"
mock_sessions_ctrl.batch_close_sessions.assert_not_called()
mock_meetings_ctrl.update_meeting.assert_not_called()
mock_poll_task.apply_async.assert_called_once()

View File

@@ -0,0 +1,367 @@
"""Integration tests for /joining and /joined endpoints.
Tests for the join intent tracking to prevent race conditions during
WebRTC handshake when users join meetings.
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch
import pytest
from reflector.db.meetings import Meeting
from reflector.presence.pending_joins import PENDING_JOIN_PREFIX
TEST_CONNECTION_ID = "test-connection-uuid-12345"
@pytest.fixture
def mock_room():
"""Mock room object."""
from reflector.db.rooms import Room
return Room(
id="room-123",
name="test-room",
user_id="owner-user",
created_at=datetime.now(timezone.utc),
zulip_auto_post=False,
zulip_stream="",
zulip_topic="",
is_locked=False,
room_mode="normal",
recording_type="cloud",
recording_trigger="automatic",
is_shared=True,
platform="daily",
skip_consent=False,
)
@pytest.fixture
def mock_meeting():
"""Mock meeting object."""
now = datetime.now(timezone.utc)
return Meeting(
id="meeting-456",
room_id="room-123",
room_name="test-room-20251118120000",
room_url="https://daily.co/test-room-20251118120000",
host_room_url="https://daily.co/test-room-20251118120000?t=host",
platform="daily",
num_clients=0,
is_active=True,
start_date=now,
end_date=now + timedelta(hours=1),
)
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_joining_endpoint_creates_pending_join(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
authenticated_client,
):
"""Test that /joining endpoint creates pending join in Redis."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.setex = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
response = await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 200
assert response.json() == {"status": "ok"}
# Verify Redis setex was called with correct key pattern
mock_redis.setex.assert_called_once()
call_args = mock_redis.setex.call_args[0]
assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
assert TEST_CONNECTION_ID in call_args[0]
@pytest.mark.asyncio
@patch("reflector.views.rooms.poll_daily_room_presence_task")
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_joined_endpoint_deletes_pending_join(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_poll_task,
mock_room,
mock_meeting,
client,
authenticated_client,
):
"""Test that /joined endpoint deletes pending join from Redis."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.delete = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
response = await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joined",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 200
assert response.json() == {"status": "ok"}
# Verify Redis delete was called with correct key pattern
mock_redis.delete.assert_called_once()
call_args = mock_redis.delete.call_args[0]
assert call_args[0].startswith(f"{PENDING_JOIN_PREFIX}:{mock_meeting.id}:")
assert TEST_CONNECTION_ID in call_args[0]
# Verify presence poll was triggered for Daily meetings
mock_poll_task.delay.assert_called_once_with(mock_meeting.id)
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
async def test_joining_endpoint_room_not_found(
mock_get_room,
client,
authenticated_client,
):
"""Test that /joining returns 404 when room not found."""
mock_get_room.return_value = None
response = await client.post(
"/rooms/nonexistent-room/meetings/meeting-123/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 404
assert response.json()["detail"] == "Room not found"
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
async def test_joining_endpoint_meeting_not_found(
mock_get_meeting,
mock_get_room,
mock_room,
client,
authenticated_client,
):
"""Test that /joining returns 404 when meeting not found."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = None
response = await client.post(
f"/rooms/{mock_room.name}/meetings/nonexistent-meeting/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 404
assert response.json()["detail"] == "Meeting not found"
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
async def test_joining_endpoint_meeting_not_active(
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
authenticated_client,
):
"""Test that /joining returns 400 when meeting is not active."""
mock_get_room.return_value = mock_room
inactive_meeting = mock_meeting.model_copy(update={"is_active": False})
mock_get_meeting.return_value = inactive_meeting
response = await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 400
assert response.json()["detail"] == "Meeting is not active"
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_joining_endpoint_anonymous_user(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
):
"""Test that /joining works for anonymous users with unique connection_id."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.setex = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
response = await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
assert response.status_code == 200
assert response.json() == {"status": "ok"}
# Verify Redis setex was called with "anon:" prefix and connection_id
call_args = mock_redis.setex.call_args[0]
assert ":anon:" in call_args[0]
assert TEST_CONNECTION_ID in call_args[0]
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_joining_endpoint_redis_closed_on_success(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
authenticated_client,
):
"""Test that Redis connection is closed after successful operation."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.setex = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
mock_redis.aclose.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_joining_endpoint_redis_closed_on_error(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
authenticated_client,
):
"""Test that Redis connection is closed even when operation fails."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.setex = AsyncMock(side_effect=Exception("Redis error"))
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
with pytest.raises(Exception):
await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": TEST_CONNECTION_ID},
)
mock_redis.aclose.assert_called_once()
@pytest.mark.asyncio
async def test_joining_endpoint_requires_connection_id(
client,
):
"""Test that /joining returns 422 when connection_id is missing."""
response = await client.post(
"/rooms/test-room/meetings/meeting-123/joining",
json={},
)
assert response.status_code == 422 # Validation error
@pytest.mark.asyncio
async def test_joining_endpoint_rejects_empty_connection_id(
client,
):
"""Test that /joining returns 422 when connection_id is empty string."""
response = await client.post(
"/rooms/test-room/meetings/meeting-123/joining",
json={"connection_id": ""},
)
assert response.status_code == 422 # Validation error (NonEmptyString)
@pytest.mark.asyncio
@patch("reflector.views.rooms.rooms_controller.get_by_name")
@patch("reflector.views.rooms.meetings_controller.get_by_id")
@patch("reflector.views.rooms.get_async_redis_client")
async def test_different_connection_ids_create_different_keys(
mock_get_redis,
mock_get_meeting,
mock_get_room,
mock_room,
mock_meeting,
client,
):
"""Test that different connection_ids create different Redis keys."""
mock_get_room.return_value = mock_room
mock_get_meeting.return_value = mock_meeting
mock_redis = AsyncMock()
mock_redis.setex = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
# First connection
await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": "connection-1"},
)
key1 = mock_redis.setex.call_args[0][0]
mock_redis.setex.reset_mock()
# Second connection (different tab)
await client.post(
f"/rooms/{mock_room.name}/meetings/{mock_meeting.id}/joining",
json={"connection_id": "connection-2"},
)
key2 = mock_redis.setex.call_args[0][0]
# Keys should be different
assert key1 != key2
assert "connection-1" in key1
assert "connection-2" in key2

View File

@@ -0,0 +1,153 @@
"""Tests for pending joins Redis helper functions.
TDD tests for tracking join intent to prevent race conditions during
WebRTC handshake when users join meetings.
"""
from unittest.mock import AsyncMock
import pytest
from reflector.presence.pending_joins import (
PENDING_JOIN_PREFIX,
PENDING_JOIN_TTL,
create_pending_join,
delete_pending_join,
has_pending_joins,
)
@pytest.fixture
def mock_redis():
"""Mock async Redis client."""
redis = AsyncMock()
redis.setex = AsyncMock()
redis.delete = AsyncMock()
redis.scan = AsyncMock(return_value=(0, []))
return redis
@pytest.mark.asyncio
async def test_create_pending_join_sets_key_with_ttl(mock_redis):
"""Test that create_pending_join stores key with correct TTL."""
meeting_id = "meeting-123"
user_id = "user-456"
await create_pending_join(mock_redis, meeting_id, user_id)
expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
mock_redis.setex.assert_called_once()
call_args = mock_redis.setex.call_args
assert call_args[0][0] == expected_key
assert call_args[0][1] == PENDING_JOIN_TTL
# Value should be a timestamp string
assert call_args[0][2] is not None
@pytest.mark.asyncio
async def test_delete_pending_join_removes_key(mock_redis):
"""Test that delete_pending_join removes the key."""
meeting_id = "meeting-123"
user_id = "user-456"
await delete_pending_join(mock_redis, meeting_id, user_id)
expected_key = f"{PENDING_JOIN_PREFIX}:{meeting_id}:{user_id}"
mock_redis.delete.assert_called_once_with(expected_key)
@pytest.mark.asyncio
async def test_has_pending_joins_returns_false_when_no_keys(mock_redis):
"""Test has_pending_joins returns False when no matching keys."""
mock_redis.scan.return_value = (0, [])
result = await has_pending_joins(mock_redis, "meeting-123")
assert result is False
mock_redis.scan.assert_called_once()
call_kwargs = mock_redis.scan.call_args.kwargs
assert call_kwargs["match"] == f"{PENDING_JOIN_PREFIX}:meeting-123:*"
@pytest.mark.asyncio
async def test_has_pending_joins_returns_true_when_keys_exist(mock_redis):
"""Test has_pending_joins returns True when matching keys found."""
mock_redis.scan.return_value = (0, [b"pending_join:meeting-123:user-1"])
result = await has_pending_joins(mock_redis, "meeting-123")
assert result is True
@pytest.mark.asyncio
async def test_has_pending_joins_scans_with_correct_pattern(mock_redis):
"""Test has_pending_joins uses correct scan pattern."""
meeting_id = "meeting-abc-def"
mock_redis.scan.return_value = (0, [])
await has_pending_joins(mock_redis, meeting_id)
expected_pattern = f"{PENDING_JOIN_PREFIX}:{meeting_id}:*"
mock_redis.scan.assert_called_once()
call_kwargs = mock_redis.scan.call_args.kwargs
assert call_kwargs["match"] == expected_pattern
assert call_kwargs["count"] == 100
@pytest.mark.asyncio
async def test_multiple_users_pending_joins(mock_redis):
"""Test that multiple users can have pending joins for same meeting."""
meeting_id = "meeting-123"
# Simulate two pending joins
mock_redis.scan.return_value = (
0,
[b"pending_join:meeting-123:user-1", b"pending_join:meeting-123:user-2"],
)
result = await has_pending_joins(mock_redis, meeting_id)
assert result is True
@pytest.mark.asyncio
async def test_pending_join_ttl_value():
"""Test that PENDING_JOIN_TTL has expected value."""
# 30 seconds should be enough for WebRTC handshake but not too long
assert PENDING_JOIN_TTL == 30
@pytest.mark.asyncio
async def test_pending_join_prefix_value():
"""Test that PENDING_JOIN_PREFIX has expected value."""
assert PENDING_JOIN_PREFIX == "pending_join"
@pytest.mark.asyncio
async def test_has_pending_joins_multi_iteration_scan_no_keys(mock_redis):
"""Test has_pending_joins iterates until cursor returns 0."""
# Simulate multi-iteration scan: cursor 100 -> cursor 50 -> cursor 0
mock_redis.scan.side_effect = [
(100, []), # First iteration, no keys, continue
(50, []), # Second iteration, no keys, continue
(0, []), # Third iteration, cursor 0, done
]
result = await has_pending_joins(mock_redis, "meeting-123")
assert result is False
assert mock_redis.scan.call_count == 3
@pytest.mark.asyncio
async def test_has_pending_joins_multi_iteration_finds_key_later(mock_redis):
"""Test has_pending_joins finds key on second iteration."""
# Simulate finding key on second scan iteration
mock_redis.scan.side_effect = [
(100, []), # First iteration, no keys
(0, [b"pending_join:meeting-123:user-1"]), # Second iteration, found key
]
result = await has_pending_joins(mock_redis, "meeting-123")
assert result is True
assert mock_redis.scan.call_count == 2

View File

@@ -0,0 +1,241 @@
"""Tests for process_meetings pending joins check.
Tests that process_meetings correctly skips deactivation when
pending joins exist for a meeting.
"""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch
import pytest
from reflector.db.meetings import Meeting
def _get_process_meetings_fn():
"""Get the underlying async function without Celery/asynctask decorators."""
from reflector.worker import process
fn = process.process_meetings
# Get through both decorator layers (@shared_task and @asynctask)
if hasattr(fn, "__wrapped__"):
fn = fn.__wrapped__
if hasattr(fn, "__wrapped__"):
fn = fn.__wrapped__
return fn
@pytest.fixture
def mock_active_meeting():
"""Mock an active meeting that should be considered for deactivation."""
now = datetime.now(timezone.utc)
return Meeting(
id="meeting-123",
room_id="room-456",
room_name="test-room-20251118120000",
room_url="https://daily.co/test-room-20251118120000",
host_room_url="https://daily.co/test-room-20251118120000?t=host",
platform="daily",
num_clients=0,
is_active=True,
start_date=now - timedelta(hours=1),
end_date=now - timedelta(minutes=30), # Already ended
)
@pytest.mark.asyncio
@patch("reflector.worker.process.meetings_controller.get_all_active")
@patch("reflector.worker.process.RedisAsyncLock")
@patch("reflector.worker.process.create_platform_client")
@patch("reflector.worker.process.get_async_redis_client")
@patch("reflector.worker.process.has_pending_joins")
@patch("reflector.worker.process.meetings_controller.update_meeting")
async def test_process_meetings_skips_deactivation_with_pending_joins(
mock_update_meeting,
mock_has_pending_joins,
mock_get_redis,
mock_create_client,
mock_redis_lock_class,
mock_get_all_active,
mock_active_meeting,
):
"""Test that process_meetings skips deactivation when pending joins exist."""
process_meetings = _get_process_meetings_fn()
mock_get_all_active.return_value = [mock_active_meeting]
# Mock lock acquired
mock_lock_instance = AsyncMock()
mock_lock_instance.acquired = True
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
mock_lock_instance.__aexit__ = AsyncMock()
mock_redis_lock_class.return_value = mock_lock_instance
# Mock platform client - no active sessions, but had sessions (triggers deactivation)
mock_daily_client = AsyncMock()
mock_session = AsyncMock()
mock_session.ended_at = datetime.now(timezone.utc) # Session ended
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
mock_create_client.return_value = mock_daily_client
# Mock Redis client
mock_redis = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
# Mock pending joins exist
mock_has_pending_joins.return_value = True
await process_meetings()
# Verify has_pending_joins was called
mock_has_pending_joins.assert_called_once_with(mock_redis, mock_active_meeting.id)
# Verify meeting was NOT deactivated
mock_update_meeting.assert_not_called()
# Verify Redis was closed
mock_redis.aclose.assert_called_once()
@pytest.mark.asyncio
@patch("reflector.worker.process.meetings_controller.get_all_active")
@patch("reflector.worker.process.RedisAsyncLock")
@patch("reflector.worker.process.create_platform_client")
@patch("reflector.worker.process.get_async_redis_client")
@patch("reflector.worker.process.has_pending_joins")
@patch("reflector.worker.process.meetings_controller.update_meeting")
async def test_process_meetings_deactivates_without_pending_joins(
mock_update_meeting,
mock_has_pending_joins,
mock_get_redis,
mock_create_client,
mock_redis_lock_class,
mock_get_all_active,
mock_active_meeting,
):
"""Test that process_meetings deactivates when no pending joins."""
process_meetings = _get_process_meetings_fn()
mock_get_all_active.return_value = [mock_active_meeting]
# Mock lock acquired
mock_lock_instance = AsyncMock()
mock_lock_instance.acquired = True
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
mock_lock_instance.__aexit__ = AsyncMock()
mock_redis_lock_class.return_value = mock_lock_instance
# Mock platform client - no active sessions, but had sessions
mock_daily_client = AsyncMock()
mock_session = AsyncMock()
mock_session.ended_at = datetime.now(timezone.utc)
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
mock_create_client.return_value = mock_daily_client
# Mock Redis client
mock_redis = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
# Mock no pending joins
mock_has_pending_joins.return_value = False
await process_meetings()
# Verify meeting was deactivated
mock_update_meeting.assert_called_once_with(mock_active_meeting.id, is_active=False)
@pytest.mark.asyncio
@patch("reflector.worker.process.meetings_controller.get_all_active")
@patch("reflector.worker.process.RedisAsyncLock")
@patch("reflector.worker.process.create_platform_client")
async def test_process_meetings_no_check_when_active_sessions(
mock_create_client,
mock_redis_lock_class,
mock_get_all_active,
mock_active_meeting,
):
"""Test that pending joins check is skipped when there are active sessions."""
process_meetings = _get_process_meetings_fn()
mock_get_all_active.return_value = [mock_active_meeting]
# Mock lock acquired
mock_lock_instance = AsyncMock()
mock_lock_instance.acquired = True
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
mock_lock_instance.__aexit__ = AsyncMock()
mock_redis_lock_class.return_value = mock_lock_instance
# Mock platform client - has active session
mock_daily_client = AsyncMock()
mock_session = AsyncMock()
mock_session.ended_at = None # Still active
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
mock_create_client.return_value = mock_daily_client
with (
patch("reflector.worker.process.get_async_redis_client") as mock_get_redis,
patch("reflector.worker.process.has_pending_joins") as mock_has_pending_joins,
patch(
"reflector.worker.process.meetings_controller.update_meeting"
) as mock_update_meeting,
):
await process_meetings()
# Verify pending joins check was NOT called (no need - active sessions exist)
mock_has_pending_joins.assert_not_called()
# Verify meeting was NOT deactivated
mock_update_meeting.assert_not_called()
@pytest.mark.asyncio
@patch("reflector.worker.process.meetings_controller.get_all_active")
@patch("reflector.worker.process.RedisAsyncLock")
@patch("reflector.worker.process.create_platform_client")
@patch("reflector.worker.process.get_async_redis_client")
@patch("reflector.worker.process.has_pending_joins")
@patch("reflector.worker.process.meetings_controller.update_meeting")
async def test_process_meetings_closes_redis_even_on_continue(
mock_update_meeting,
mock_has_pending_joins,
mock_get_redis,
mock_create_client,
mock_redis_lock_class,
mock_get_all_active,
mock_active_meeting,
):
"""Test that Redis connection is always closed, even when skipping deactivation."""
process_meetings = _get_process_meetings_fn()
mock_get_all_active.return_value = [mock_active_meeting]
# Mock lock acquired
mock_lock_instance = AsyncMock()
mock_lock_instance.acquired = True
mock_lock_instance.__aenter__ = AsyncMock(return_value=mock_lock_instance)
mock_lock_instance.__aexit__ = AsyncMock()
mock_redis_lock_class.return_value = mock_lock_instance
# Mock platform client - no active sessions
mock_daily_client = AsyncMock()
mock_session = AsyncMock()
mock_session.ended_at = datetime.now(timezone.utc)
mock_daily_client.get_room_sessions = AsyncMock(return_value=[mock_session])
mock_create_client.return_value = mock_daily_client
# Mock Redis client
mock_redis = AsyncMock()
mock_redis.aclose = AsyncMock()
mock_get_redis.return_value = mock_redis
# Mock pending joins exist (will trigger continue)
mock_has_pending_joins.return_value = True
await process_meetings()
# Verify Redis was closed
mock_redis.aclose.assert_called_once()

View File

@@ -24,24 +24,18 @@ import { useAuth } from "../../lib/AuthProvider";
import { useConsentDialog } from "../../lib/consent"; import { useConsentDialog } from "../../lib/consent";
import { import {
useRoomJoinMeeting, useRoomJoinMeeting,
useRoomJoinedMeeting,
useRoomLeaveMeeting,
useMeetingStartRecording, useMeetingStartRecording,
leaveRoomPostUrl, useMeetingJoining,
LeaveRoomBody, useMeetingJoined,
buildMeetingLeaveUrl,
} from "../../lib/apiHooks"; } from "../../lib/apiHooks";
import { omit } from "remeda"; import { omit } from "remeda";
import { import {
assertExists, assertExists,
assertExistsAndNonEmptyString,
NonEmptyString, NonEmptyString,
parseNonEmptyString, parseNonEmptyString,
} from "../../lib/utils"; } from "../../lib/utils";
import { import { assertMeetingId, DailyRecordingType } from "../../lib/types";
assertMeetingId,
DailyRecordingType,
MeetingId,
} from "../../lib/types";
import { useUuidV5 } from "react-uuid-hook"; import { useUuidV5 } from "react-uuid-hook";
const CONSENT_BUTTON_ID = "recording-consent"; const CONSENT_BUTTON_ID = "recording-consent";
@@ -97,7 +91,7 @@ const useFrame = (
cbs: { cbs: {
onLeftMeeting: () => void; onLeftMeeting: () => void;
onCustomButtonClick: (ev: DailyEventObjectCustomButtonClick) => void; onCustomButtonClick: (ev: DailyEventObjectCustomButtonClick) => void;
onJoinMeeting: () => void; onJoinMeeting: (sessionId: string | null) => void;
}, },
) => { ) => {
const [{ frame, joined }, setState] = useState(USE_FRAME_INIT_STATE); const [{ frame, joined }, setState] = useState(USE_FRAME_INIT_STATE);
@@ -148,7 +142,8 @@ const useFrame = (
console.error("frame is null in joined-meeting callback"); console.error("frame is null in joined-meeting callback");
return; return;
} }
cbs.onJoinMeeting(); const local = frame.participants()?.local;
cbs.onJoinMeeting(local?.session_id ?? null);
}; };
frame.on("joined-meeting", joinCb); frame.on("joined-meeting", joinCb);
return () => { return () => {
@@ -188,58 +183,6 @@ const useFrame = (
] as const; ] as const;
}; };
const leaveDaily = () => {
const frame = DailyIframe.getCallInstance();
frame?.leave();
};
const useDirtyDisconnects = (
meetingId: NonEmptyString,
roomName: NonEmptyString,
) => {
useEffect(() => {
if (!meetingId || !roomName) return;
const handleBeforeUnload = () => {
leaveDaily();
navigator.sendBeacon(
leaveRoomPostUrl(
{
room_name: roomName,
meeting_id: meetingId,
},
{
delay_seconds: 5,
},
),
undefined satisfies LeaveRoomBody,
);
};
window.addEventListener("beforeunload", handleBeforeUnload);
return () => window.removeEventListener("beforeunload", handleBeforeUnload);
}, [meetingId, roomName]);
};
const useDisconnects = (
meetingId: NonEmptyString,
roomName: NonEmptyString,
leaveMutation: ReturnType<typeof useRoomLeaveMeeting>,
) => {
useDirtyDisconnects(meetingId, roomName);
useEffect(() => {
return () => {
leaveDaily();
leaveMutation.mutate({
params: {
path: { meeting_id: meetingId, room_name: roomName },
query: { delay_seconds: 5 },
},
});
};
}, [meetingId, roomName]);
};
export default function DailyRoom({ meeting, room }: DailyRoomProps) { export default function DailyRoom({ meeting, room }: DailyRoomProps) {
const router = useRouter(); const router = useRouter();
const params = useParams(); const params = useParams();
@@ -247,10 +190,15 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
const authLastUserId = auth.lastUserId; const authLastUserId = auth.lastUserId;
const [container, setContainer] = useState<HTMLDivElement | null>(null); const [container, setContainer] = useState<HTMLDivElement | null>(null);
const joinMutation = useRoomJoinMeeting(); const joinMutation = useRoomJoinMeeting();
const joinedMutation = useRoomJoinedMeeting();
const leaveMutation = useRoomLeaveMeeting();
const startRecordingMutation = useMeetingStartRecording(); const startRecordingMutation = useMeetingStartRecording();
const joiningMutation = useMeetingJoining();
const joinedMutation = useMeetingJoined();
const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null); const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null);
const sessionIdRef = useRef<string | null>(null);
// Generate a stable connection ID for this component instance
// Used to track pending joins per browser tab (prevents key collision for anonymous users)
const connectionId = useMemo(() => crypto.randomUUID(), []);
// Generate deterministic instanceIds so all participants use SAME IDs // Generate deterministic instanceIds so all participants use SAME IDs
const cloudInstanceId = parseNonEmptyString(meeting.id); const cloudInstanceId = parseNonEmptyString(meeting.id);
@@ -258,9 +206,7 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
useUuidV5(meeting.id, RAW_TRACKS_NAMESPACE)[0], useUuidV5(meeting.id, RAW_TRACKS_NAMESPACE)[0],
); );
if (typeof params.roomName === "object") const roomName = params?.roomName as string;
throw new Error(`Invalid room name in params. array? ${params.roomName}`);
const roomName = assertExistsAndNonEmptyString(params.roomName);
const { const {
showConsentModal, showConsentModal,
@@ -299,10 +245,34 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
const roomUrl = joinedMeeting?.room_url; const roomUrl = joinedMeeting?.room_url;
const handleLeave = useCallback(() => { const handleLeave = useCallback(() => {
if (meeting?.id && roomName) {
const payload = sessionIdRef.current
? { session_id: sessionIdRef.current }
: {};
navigator.sendBeacon(
buildMeetingLeaveUrl(roomName, meeting.id),
JSON.stringify(payload),
);
}
router.push("/browse"); router.push("/browse");
}, [router]); }, [router, roomName, meeting?.id]);
useDisconnects(meeting.id as MeetingId, roomName, leaveMutation); // Trigger presence recheck on dirty disconnects (tab close, navigation away)
useEffect(() => {
if (!meeting?.id || !roomName) return;
const handleBeforeUnload = () => {
// sendBeacon guarantees delivery even if tab closes mid-request
const url = buildMeetingLeaveUrl(roomName, meeting.id);
const payload = sessionIdRef.current
? { session_id: sessionIdRef.current }
: {};
navigator.sendBeacon(url, JSON.stringify(payload));
};
window.addEventListener("beforeunload", handleBeforeUnload);
return () => window.removeEventListener("beforeunload", handleBeforeUnload);
}, [meeting?.id, roomName]);
const handleCustomButtonClick = useCallback( const handleCustomButtonClick = useCallback(
(ev: DailyEventObjectCustomButtonClick) => { (ev: DailyEventObjectCustomButtonClick) => {
@@ -315,83 +285,106 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
], ],
); );
const handleFrameJoinMeeting = useCallback(() => { const handleFrameJoinMeeting = useCallback(
joinedMutation.mutate({ (sessionId: string | null) => {
params: { sessionIdRef.current = sessionId;
path: {
room_name: roomName, // Signal that WebRTC connection is established
meeting_id: meeting.id, // This clears the pending join intent and creates session record directly
joinedMutation.mutate(
{
params: {
path: {
room_name: roomName,
meeting_id: meeting.id,
},
},
body: {
connection_id: connectionId,
session_id: sessionId,
},
}, },
}, {
}); onError: (error: unknown) => {
// Non-blocking: log but don't fail - this is cleanup, not critical
console.warn("Failed to signal joined:", error);
},
},
);
if (meeting.recording_type === "cloud") { if (meeting.recording_type === "cloud") {
console.log("Starting dual recording via REST API", { console.log("Starting dual recording via REST API", {
cloudInstanceId, cloudInstanceId,
rawTracksInstanceId, rawTracksInstanceId,
}); });
// Start both cloud and raw-tracks via backend REST API (with retry on 404) // Start both cloud and raw-tracks via backend REST API (with retry on 404)
// Daily.co needs time to register call as "hosting" for REST API // Daily.co needs time to register call as "hosting" for REST API
const startRecordingWithRetry = ( const startRecordingWithRetry = (
type: DailyRecordingType, type: DailyRecordingType,
instanceId: NonEmptyString, instanceId: NonEmptyString,
attempt: number = 1, attempt: number = 1,
) => { ) => {
setTimeout(() => { setTimeout(() => {
startRecordingMutation.mutate( startRecordingMutation.mutate(
{ {
params: { params: {
path: { path: {
meeting_id: meeting.id, meeting_id: meeting.id,
},
},
body: {
type,
instanceId,
}, },
}, },
body: { {
type, onError: (error: any) => {
instanceId, const errorText = error?.detail || error?.message || "";
}, const is404NotHosting = errorText.includes(
}, "does not seem to be hosting a call",
{
onError: (error: any) => {
const errorText = error?.detail || error?.message || "";
const is404NotHosting = errorText.includes(
"does not seem to be hosting a call",
);
const isActiveStream = errorText.includes(
"has an active stream",
);
if (is404NotHosting && attempt < RECORDING_START_MAX_RETRIES) {
console.log(
`${type}: Call not hosting yet, retry ${attempt + 1}/${RECORDING_START_MAX_RETRIES} in ${RECORDING_START_DELAY_MS}ms...`,
); );
startRecordingWithRetry(type, instanceId, attempt + 1); const isActiveStream = errorText.includes(
} else if (isActiveStream) { "has an active stream",
console.log(
`${type}: Recording already active (started by another participant)`,
); );
} else {
console.error(`Failed to start ${type} recording:`, error);
}
},
},
);
}, RECORDING_START_DELAY_MS);
};
// Start both recordings if (
startRecordingWithRetry("cloud", cloudInstanceId); is404NotHosting &&
startRecordingWithRetry("raw-tracks", rawTracksInstanceId); attempt < RECORDING_START_MAX_RETRIES
} ) {
}, [ console.log(
joinedMutation, `${type}: Call not hosting yet, retry ${attempt + 1}/${RECORDING_START_MAX_RETRIES} in ${RECORDING_START_DELAY_MS}ms...`,
roomName, );
meeting.id, startRecordingWithRetry(type, instanceId, attempt + 1);
meeting.recording_type, } else if (isActiveStream) {
startRecordingMutation, console.log(
cloudInstanceId, `${type}: Recording already active (started by another participant)`,
rawTracksInstanceId, );
]); } else {
console.error(`Failed to start ${type} recording:`, error);
}
},
},
);
}, RECORDING_START_DELAY_MS);
};
// Start both recordings
startRecordingWithRetry("cloud", cloudInstanceId);
startRecordingWithRetry("raw-tracks", rawTracksInstanceId);
}
},
[
meeting.recording_type,
meeting.id,
roomName,
connectionId,
joinedMutation,
startRecordingMutation,
cloudInstanceId,
rawTracksInstanceId,
],
);
const recordingIconUrl = useMemo( const recordingIconUrl = useMemo(
() => new URL("/recording-icon.svg", window.location.origin), () => new URL("/recording-icon.svg", window.location.origin),
@@ -406,8 +399,28 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
useEffect(() => { useEffect(() => {
if (!frame || !roomUrl) return; if (!frame || !roomUrl) return;
frame
.join({ const joinRoom = async () => {
// Signal intent to join before WebRTC handshake starts
// This prevents race condition where meeting is deactivated during handshake
try {
await joiningMutation.mutateAsync({
params: {
path: {
room_name: roomName,
meeting_id: meeting.id,
},
},
body: {
connection_id: connectionId,
},
});
} catch (error) {
// Non-blocking: log but continue with join
console.warn("Failed to signal joining intent:", error);
}
await frame.join({
url: roomUrl, url: roomUrl,
sendSettings: { sendSettings: {
video: { video: {
@@ -419,9 +432,13 @@ export default function DailyRoom({ meeting, room }: DailyRoomProps) {
}, },
// Note: screenVideo intentionally not configured to preserve full quality for screen shares // Note: screenVideo intentionally not configured to preserve full quality for screen shares
}, },
}) });
.catch(console.error.bind(console, "Failed to join daily room:")); };
}, [frame, roomUrl]);
joinRoom().catch(console.error.bind(console, "Failed to join daily room:"));
// joiningMutation excluded from deps - it's a stable hook reference
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [frame, roomUrl, roomName, meeting.id, connectionId]);
useEffect(() => { useEffect(() => {
setCustomTrayButton( setCustomTrayButton(

View File

@@ -1,13 +1,13 @@
"use client"; "use client";
import { createFinalURL, createQuerySerializer } from "openapi-fetch";
import { $api, API_URL } from "./apiClient"; import { $api, API_URL } from "./apiClient";
import { useError } from "../(errors)/errorContext"; import { useError } from "../(errors)/errorContext";
import { QueryClient, useQueryClient } from "@tanstack/react-query"; import { QueryClient, useQueryClient } from "@tanstack/react-query";
import type { components, operations } from "../reflector-api"; import type { components, paths } from "../reflector-api";
import { useAuth } from "./AuthProvider"; import { useAuth } from "./AuthProvider";
import { MeetingId } from "./types"; import { MeetingId } from "./types";
import { NonEmptyString } from "./utils"; import { NonEmptyString } from "./utils";
import { createFinalURL, createQuerySerializer } from "openapi-fetch";
/* /*
* XXX error types returned from the hooks are not always correct; declared types are ValidationError but real type could be string or any other * XXX error types returned from the hooks are not always correct; declared types are ValidationError but real type could be string or any other
@@ -15,6 +15,10 @@ import { createFinalURL, createQuerySerializer } from "openapi-fetch";
* or, limitation or incorrect usage of .d type generator from json schema * or, limitation or incorrect usage of .d type generator from json schema
* */ * */
/*
if you experience lack of expected endpoints on $api, try to regenerate
*/
export const useAuthReady = () => { export const useAuthReady = () => {
const auth = useAuth(); const auth = useAuth();
@@ -769,6 +773,7 @@ export function useRoomActiveMeetings(roomName: string | null) {
}, },
{ {
enabled: !!roomName, enabled: !!roomName,
refetchInterval: 5000,
}, },
); );
} }
@@ -808,42 +813,45 @@ export function useRoomJoinMeeting() {
); );
} }
export const LEAVE_ROOM_POST_URL_TEMPLATE = // Presence race fix endpoints - signal join intent to prevent race conditions during WebRTC handshake
"/v1/rooms/{room_name}/meetings/{meeting_id}/leave" as const; export function useMeetingJoining() {
return $api.useMutation(
export const leaveRoomPostUrl = ( "post",
path: operations["v1_rooms_leave_meeting"]["parameters"]["path"], "/v1/rooms/{room_name}/meetings/{meeting_id}/joining",
query?: operations["v1_rooms_leave_meeting"]["parameters"]["query"], {},
): string => );
createFinalURL(LEAVE_ROOM_POST_URL_TEMPLATE, {
baseUrl: API_URL,
params: { path, query },
querySerializer: createQuerySerializer(),
});
export type LeaveRoomBody = operations["v1_rooms_leave_meeting"]["requestBody"];
export function useRoomLeaveMeeting() {
return $api.useMutation("post", LEAVE_ROOM_POST_URL_TEMPLATE);
} }
export const JOINED_ROOM_POST_URL_TEMPLATE = export function useMeetingJoined() {
"/v1/rooms/{room_name}/meetings/{meeting_id}/joined" as const; return $api.useMutation(
"post",
"/v1/rooms/{room_name}/meetings/{meeting_id}/joined",
{},
);
}
export const joinedRoomPostUrl = ( export function useMeetingLeave() {
params: operations["v1_rooms_joined_meeting"]["parameters"]["path"], return $api.useMutation(
): string => "post",
createFinalURL(JOINED_ROOM_POST_URL_TEMPLATE, { "/v1/rooms/{room_name}/meetings/{meeting_id}/leave",
{},
);
}
/**
* Build absolute URL for /leave endpoint (for sendBeacon which can't use hooks).
*/
export function buildMeetingLeaveUrl(
roomName: string,
meetingId: string,
): string {
return createFinalURL("/v1/rooms/{room_name}/meetings/{meeting_id}/leave", {
baseUrl: API_URL, baseUrl: API_URL,
params: { path: params }, params: {
querySerializer: () => "", path: { room_name: roomName, meeting_id: meetingId },
},
querySerializer: createQuerySerializer(),
}); });
export type JoinedRoomBody =
operations["v1_rooms_joined_meeting"]["requestBody"];
export function useRoomJoinedMeeting() {
return $api.useMutation("post", JOINED_ROOM_POST_URL_TEMPLATE);
} }
export function useRoomIcsSync() { export function useRoomIcsSync() {

View File

@@ -171,48 +171,6 @@ export interface paths {
patch?: never; patch?: never;
trace?: never; trace?: never;
}; };
"/v1/rooms/{room_name}/meetings/{meeting_id}/joined": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Rooms Joined Meeting
* @description Trigger presence poll (ideally when user actually joins meeting in Daily iframe)
*/
post: operations["v1_rooms_joined_meeting"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/rooms/{room_name}/meetings/{meeting_id}/leave": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Rooms Leave Meeting
* @description Trigger presence recheck when user leaves meeting (e.g., tab close/navigation).
*
* Queues presence poll with optional delay to allow Daily.co to detect disconnect.
*/
post: operations["v1_rooms_leave_meeting"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/rooms/{room_id}/webhook/test": { "/v1/rooms/{room_id}/webhook/test": {
parameters: { parameters: {
query?: never; query?: never;
@@ -355,6 +313,79 @@ export interface paths {
patch?: never; patch?: never;
trace?: never; trace?: never;
}; };
"/v1/rooms/{room_name}/meetings/{meeting_id}/joining": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Meeting Joining
* @description Signal intent to join meeting. Called before WebRTC handshake starts.
*
* This creates a pending join record that prevents the meeting from being
* deactivated while the WebRTC handshake is in progress. The record expires
* automatically after 30 seconds if the connection is not established.
*/
post: operations["v1_meeting_joining"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/rooms/{room_name}/meetings/{meeting_id}/joined": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Meeting Joined
* @description Signal that WebRTC connection is established.
*
* This clears the pending join record, confirming the user has successfully
* connected to the meeting. Safe to call even if meeting was deactivated
* during the handshake (idempotent cleanup).
*/
post: operations["v1_meeting_joined"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/rooms/{room_name}/meetings/{meeting_id}/leave": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Meeting Leave
* @description Trigger presence update when user leaves meeting.
*
* When session_id is provided in the body, closes the session directly
* for instant presence update. Falls back to polling when session_id
* is not available (e.g., sendBeacon without frame access).
* Called on tab close/navigation via sendBeacon().
*/
post: operations["v1_meeting_leave"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/transcripts": { "/v1/transcripts": {
parameters: { parameters: {
query?: never; query?: never;
@@ -1539,6 +1570,56 @@ export interface components {
/** Reason */ /** Reason */
reason?: string | null; reason?: string | null;
}; };
/**
* JoinedRequest
* @description Request body for /joined endpoint (after WebRTC connection established).
*/
JoinedRequest: {
/**
* Connection Id
* @description A non-empty string
*/
connection_id: string;
/** Session Id */
session_id?: string | null;
/** User Name */
user_name?: string | null;
};
/** JoinedResponse */
JoinedResponse: {
/**
* Status
* @constant
*/
status: "ok";
};
/**
* JoiningRequest
* @description Request body for /joining endpoint (before WebRTC handshake).
*/
JoiningRequest: {
/**
* Connection Id
* @description A non-empty string
*/
connection_id: string;
};
/** JoiningResponse */
JoiningResponse: {
/**
* Status
* @constant
*/
status: "ok";
};
/** LeaveResponse */
LeaveResponse: {
/**
* Status
* @constant
*/
status: "ok";
};
/** Meeting */ /** Meeting */
Meeting: { Meeting: {
/** Id */ /** Id */
@@ -2477,72 +2558,6 @@ export interface operations {
}; };
}; };
}; };
v1_rooms_joined_meeting: {
parameters: {
query?: never;
header?: never;
path: {
room_name: string;
meeting_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
v1_rooms_leave_meeting: {
parameters: {
query?: {
delay_seconds?: number;
};
header?: never;
path: {
room_name: string;
meeting_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
v1_rooms_test_webhook: { v1_rooms_test_webhook: {
parameters: { parameters: {
query?: never; query?: never;
@@ -2795,6 +2810,110 @@ export interface operations {
}; };
}; };
}; };
v1_meeting_joining: {
parameters: {
query?: never;
header?: never;
path: {
room_name: string;
meeting_id: string;
};
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["JoiningRequest"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["JoiningResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
v1_meeting_joined: {
parameters: {
query?: never;
header?: never;
path: {
room_name: string;
meeting_id: string;
};
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["JoinedRequest"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["JoinedResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
v1_meeting_leave: {
parameters: {
query?: never;
header?: never;
path: {
room_name: string;
meeting_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["LeaveResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
v1_transcripts_list: { v1_transcripts_list: {
parameters: { parameters: {
query?: { query?: {