From 8507cdd2b62870e7bd11e611416e83c4bfa23057 Mon Sep 17 00:00:00 2001 From: leo <260626284+cameledev@users.noreply.github.com> Date: Tue, 7 Apr 2026 12:11:32 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(backend)=20add=20metadata=20collectio?= =?UTF-8?q?n=20of=20VAD,=20connection=20and=20chat=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce MetadataCollector and MetadataCollectorService classes to centralize the collection and storage of user connections, VAD events, and chat messages. This creates a structured foundation for future speaker assignment logic based on voice activity detection. Add tests for this new feature. --- CHANGELOG.md | 4 + env.d/development/common.dist | 3 + src/agents/exceptions.py | 5 + src/agents/metadata_collector.py | 382 ++++++++++++++++++ ...anscriber.py => multi_user_transcriber.py} | 0 src/agents/pyproject.toml | 6 +- src/backend/core/api/viewsets.py | 13 + .../recording/services/metadata_collector.py | 91 +++++ .../recording/services/recording_events.py | 5 +- src/backend/core/services/livekit_events.py | 14 +- .../rooms/test_api_rooms_start_recording.py | 88 ++++ .../tests/services/test_livekit_events.py | 57 +++ src/backend/meet/settings.py | 10 + 13 files changed, 674 insertions(+), 4 deletions(-) create mode 100644 src/agents/exceptions.py create mode 100644 src/agents/metadata_collector.py rename src/agents/{multi-user-transcriber.py => multi_user_transcriber.py} (100%) create mode 100644 src/backend/core/recording/services/metadata_collector.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a257402e..35fa2cd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ and this project adheres to ## [Unreleased] +### Added + +- ✨(backend) add metadata collection of VAD, connection and chat events + ## [1.14.0] - 2026-04-16 ### Added diff --git a/env.d/development/common.dist b/env.d/development/common.dist index 4dcc335c..12ef54f4 100644 --- a/env.d/development/common.dist +++ b/env.d/development/common.dist @@ -71,6 +71,9 @@ RECORDING_DOWNLOAD_BASE_URL=http://localhost:3000/recording # Telephony ROOM_TELEPHONY_ENABLED=True +# Metadata +METADATA_COLLECTOR_ENABLED=True + FRONTEND_USE_FRENCH_GOV_FOOTER=False FRONTEND_USE_PROCONNECT_BUTTON=False diff --git a/src/agents/exceptions.py b/src/agents/exceptions.py new file mode 100644 index 00000000..7798d686 --- /dev/null +++ b/src/agents/exceptions.py @@ -0,0 +1,5 @@ +"""Storage parsers specific exceptions.""" + + +class MissingConfigError(Exception): + """Raised when a variable is not set in configuration.""" diff --git a/src/agents/metadata_collector.py b/src/agents/metadata_collector.py new file mode 100644 index 00000000..5d4acd08 --- /dev/null +++ b/src/agents/metadata_collector.py @@ -0,0 +1,382 @@ +"""Metadata agent that extracts metadata from active room.""" + +import asyncio +import json +import logging +import os +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from io import BytesIO +from typing import List, Optional + +from dotenv import load_dotenv +from livekit import api, rtc +from livekit.agents import ( + Agent, + AgentServer, + AgentSession, + AutoSubscribe, + JobContext, + JobProcess, + JobRequest, + RoomInputOptions, + RoomIO, + RoomOutputOptions, + WorkerPermissions, + cli, + utils, +) +from livekit.plugins import silero +from minio import Minio +from minio.error import S3Error + +from exceptions import MissingConfigError + +load_dotenv() + +logger = logging.getLogger("metadata-collector") + +AGENT_NAME = os.getenv("METADATA_COLLECTOR_AGENT_NAME", "metadata-collector") + + +def prewarm(proc: JobProcess): + """Preload voice activity detection model.""" + proc.userdata["vad"] = silero.VAD.load() + + +server = AgentServer( + permissions=WorkerPermissions( + can_publish=False, + can_publish_data=False, + can_subscribe=True, + hidden=True, + ), +) +server.setup_fnc = prewarm + + +@dataclass +class MetadataEvent: + """A single timestamped event recorded during a meeting.""" + + participant_id: str + type: str + timestamp: datetime + data: Optional[str] = None + + def serialize(self) -> dict: + """Return a JSON-serializable dictionary representation of the event.""" + data = asdict(self) + data["timestamp"] = self.timestamp.isoformat() + return data + + +class VADAgent(Agent): + """Agent that monitors voice activity for a specific participant.""" + + def __init__(self, participant_identity: str, events: List): + """Initialize with a participant identity and shared events list.""" + super().__init__( + instructions="not-needed", + ) + self.participant_identity = participant_identity + self.events = events + + async def on_enter(self) -> None: + """Initialize VAD monitoring for this participant.""" + + @self.session.on("user_state_changed") + def on_user_state(event): + timestamp = datetime.now(timezone.utc) + + if event.new_state == "speaking": + event = MetadataEvent( + participant_id=self.participant_identity, + type="speech_start", + timestamp=timestamp, + ) + self.events.append(event) + + elif event.old_state == "speaking": + event = MetadataEvent( + participant_id=self.participant_identity, + type="speech_end", + timestamp=timestamp, + ) + self.events.append(event) + + +class MetadataCollector: + """Collect meeting events across all participants in a room. + + Creates one AgentSession per participant to capture VAD events + (speech start/end), and listens for connection, disconnection, + and chat events. Persists all collected events as JSON to S3 + on shutdown. + """ + + def __init__(self, ctx: JobContext, recording_id: str): + """Initialize metadata agent.""" + self.minio_client = Minio( + endpoint=os.getenv("AWS_S3_ENDPOINT_URL"), + access_key=os.getenv("AWS_S3_ACCESS_KEY_ID"), + secret_key=os.getenv("AWS_S3_SECRET_ACCESS_KEY"), + secure=os.getenv("AWS_S3_SECURE_ACCESS", "False").lower() == "true", + ) + + if (bucket_name := os.getenv("AWS_STORAGE_BUCKET_NAME")) is not None: + self.bucket_name = bucket_name + else: + raise MissingConfigError + + self.ctx = ctx + self._sessions: dict[str, AgentSession] = {} + self._tasks: set[asyncio.Task] = set() + + output_folder = os.getenv("AWS_S3_OUTPUT_FOLDER", "metadata") + self.output_filename = f"{output_folder}/{recording_id}-metadata.json" + + # Storage for events + self.events = [] + self.participants = {} + + logger.info("MetadataCollector initialized") + + def start(self): + """Start listening for room-level events.""" + self.ctx.room.on("participant_disconnected", self.on_participant_disconnected) + self.ctx.room.on("participant_name_changed", self.on_participant_name_changed) + + self.ctx.room.register_text_stream_handler("lk.chat", self.handle_chat_stream) + + logger.info("Started listening for participant events") + + async def on_chat_message_received( + self, reader: rtc.TextStreamReader, participant_identity: str + ): + """Read a complete chat message and record it as an event.""" + full_text = await reader.read_all() + logger.info("Received chat message from %s", participant_identity) + + self.events.append( + MetadataEvent( + participant_id=participant_identity, + type="chat_received", + timestamp=datetime.now(timezone.utc), + data=full_text, + ) + ) + + def handle_chat_stream(self, reader, participant_identity): + """Schedule async processing of an incoming chat stream.""" + task = asyncio.create_task( + self.on_chat_message_received(reader, participant_identity) + ) + self._tasks.add(task) + task.add_done_callback(lambda _: self._tasks.remove(task)) + + def save(self): + """Serialize collected events and upload as JSON to S3.""" + logger.info("Persisting metadata…") + + participants = [] + for k, v in self.participants.items(): + participants.append({"participantId": k, "name": v}) + + sorted_events = sorted(self.events, key=lambda e: e.timestamp) + + payload = { + "events": [event.serialize() for event in sorted_events], + "participants": participants, + } + + data = json.dumps(payload, indent=2).encode("utf-8") + stream = BytesIO(data) + + try: + self.minio_client.put_object( + self.bucket_name, + self.output_filename, + stream, + length=len(data), + content_type="application/json", + ) + logger.info( + "Uploaded speaker meeting metadata", + ) + except S3Error: + logger.exception( + "Failed to upload meeting metadata", + ) + + async def aclose(self): + """Close all sessions and cleanup resources.""" + logger.info("Closing all VAD monitoring sessions…") + + await utils.aio.cancel_and_wait(*self._tasks) + + await asyncio.gather( + *[self._close_session(session) for session in self._sessions.values()], + return_exceptions=True, + ) + + self.ctx.room.off("participant_disconnected", self.on_participant_disconnected) + self.ctx.room.off("participant_name_changed", self.on_participant_name_changed) + + logger.info("All VAD sessions closed") + self.save() + + async def on_participant_entrypoint( + self, ctx: JobContext, participant: rtc.RemoteParticipant + ): + """Handle new participant by starting a VAD monitoring session.""" + if participant.identity in self._sessions: + logger.debug("Session already exists for %s", participant.identity) + return + + self.events.append( + MetadataEvent( + participant_id=participant.identity, + type="participant_connected", + timestamp=datetime.now(timezone.utc), + ) + ) + + self.participants[participant.identity] = participant.name + + logger.info("New participant connected: %s", participant.identity) + try: + session = await self._start_session(participant) + self._sessions[participant.identity] = session + except Exception: + logger.exception("Failed to start session for %s", participant.identity) + + def on_participant_disconnected(self, participant: rtc.RemoteParticipant): + """Handle participant disconnection by closing VAD monitoring.""" + self.events.append( + MetadataEvent( + participant_id=participant.identity, + type="participant_disconnected", + timestamp=datetime.now(timezone.utc), + ) + ) + + session = self._sessions.pop(participant.identity, None) + if session is None: + logger.debug("No session found for %s", participant.identity) + return + + logger.info("Participant disconnected: %s", participant.identity) + task = asyncio.create_task(self._close_session(session)) + self._tasks.add(task) + + def on_close_done(_): + self._tasks.discard(task) + logger.info( + "VAD session closed for %s (remaining sessions: %d)", + participant.identity, + len(self._sessions), + ) + + task.add_done_callback(on_close_done) + + def on_participant_name_changed(self, participant: rtc.RemoteParticipant): + """Update stored participant name when it changes.""" + logger.info("Participant's name changed: %s", participant.identity) + self.participants[participant.identity] = participant.name + + async def _start_session(self, participant: rtc.RemoteParticipant) -> AgentSession: + """Create and start VAD monitoring session for participant.""" + if participant.identity in self._sessions: + return self._sessions[participant.identity] + + # Create session with VAD only - no STT, LLM, or TTS + session = AgentSession( + vad=self.ctx.proc.userdata["vad"], + turn_detection="vad", + user_away_timeout=30.0, + ) + + # Set up room IO to receive audio from this specific participant + room_io = RoomIO( + agent_session=session, + room=self.ctx.room, + participant=participant, + input_options=RoomInputOptions( + audio_enabled=True, + text_enabled=False, + ), + output_options=RoomOutputOptions( + audio_enabled=False, + transcription_enabled=False, + ), + ) + + await room_io.start() + await session.start( + agent=VADAgent( + participant_identity=participant.identity, events=self.events + ) + ) + + return session + + async def _close_session(self, session: AgentSession) -> None: + """Close and cleanup VAD monitoring session.""" + try: + await session.drain() + await session.aclose() + except Exception: + logger.exception("Error closing session") + + +async def handle_job_request(job_req: JobRequest) -> None: + """Accept or reject the job request based on agent presence in the room.""" + room_name = job_req.room.name + recording_id = job_req.job.metadata + agent_identity = f"{AGENT_NAME}-{room_name}" + + async with api.LiveKitAPI() as lk: + try: + resp = await lk.room.list_participants( + list=api.ListParticipantsRequest(room=room_name) + ) + already_present = any( + p.kind == rtc.ParticipantKind.PARTICIPANT_KIND_AGENT + and p.identity == agent_identity + for p in resp.participants + ) + if already_present: + logger.info("Agent already in the room '%s' — reject", room_name) + await job_req.reject() + else: + logger.info( + "Accept job for '%s' — identity=%s", room_name, agent_identity + ) + await job_req.accept(identity=agent_identity, metadata=recording_id) + except Exception: + logger.exception("Error treating the job for '%s'", room_name) + await job_req.reject() + + +@server.rtc_session(agent_name=AGENT_NAME, on_request=handle_job_request) +async def entrypoint(ctx: JobContext): + """Initialize and run the metadata collector.""" + logger.info("Starting metadata agent in room: %s", ctx.room.name) + recording_id = ctx.job.metadata + metadata_collector = MetadataCollector(ctx, recording_id) + metadata_collector.start() + + ctx.add_participant_entrypoint(metadata_collector.on_participant_entrypoint) + + await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) + + async def cleanup(): + logger.info("Shutting down metadata collector…") + await metadata_collector.aclose() + + ctx.add_shutdown_callback(cleanup) + + +if __name__ == "__main__": + cli.run_app(server) diff --git a/src/agents/multi-user-transcriber.py b/src/agents/multi_user_transcriber.py similarity index 100% rename from src/agents/multi-user-transcriber.py rename to src/agents/multi_user_transcriber.py diff --git a/src/agents/pyproject.toml b/src/agents/pyproject.toml index 7833ea79..43f053e9 100644 --- a/src/agents/pyproject.toml +++ b/src/agents/pyproject.toml @@ -9,7 +9,8 @@ dependencies = [ "livekit-plugins-silero==1.4.5", "livekit-plugins-kyutai-lasuite==0.0.6", "python-dotenv==1.2.2", - "protobuf==6.33.5" + "protobuf==6.33.5", + "minio==7.2.15" ] [project.optional-dependencies] @@ -17,6 +18,9 @@ dev = [ "ruff==0.15.6", ] +[tool.setuptools] +py-modules = ["multi_user_transcriber", "metadata_collector", "exceptions"] + [build-system] requires = ["setuptools>=61.0"] build-backend = "setuptools.build_meta" diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index 202695de..a2eda72a 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -45,6 +45,10 @@ from core.recording.event.exceptions import ( ) from core.recording.event.notification import notification_service from core.recording.event.parsers import get_parser +from core.recording.services.metadata_collector import ( + MetadataCollectorException, + MetadataCollectorService, +) from core.recording.worker.exceptions import ( RecordingStartError, RecordingStopError, @@ -338,6 +342,15 @@ class RoomViewSet( status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR, ) + if settings.METADATA_COLLECTOR_ENABLED and ( + recording.mode == models.RecordingModeChoices.TRANSCRIPT + or recording.options.get("transcribe", False) + ): + try: + MetadataCollectorService().start(recording) + except MetadataCollectorException: + logger.warning("Failed to start MetadataCollectorService") + return drf_response.Response( {"message": f"Recording successfully started for room {room.slug}"}, status=drf_status.HTTP_201_CREATED, diff --git a/src/backend/core/recording/services/metadata_collector.py b/src/backend/core/recording/services/metadata_collector.py new file mode 100644 index 00000000..e75904fc --- /dev/null +++ b/src/backend/core/recording/services/metadata_collector.py @@ -0,0 +1,91 @@ +"""Meeting metadata collection service.""" + +from logging import getLogger + +from django.conf import settings + +from asgiref.sync import async_to_sync, sync_to_async +from livekit.protocol.agent_dispatch import ( + CreateAgentDispatchRequest, +) + +from core import utils +from core.models import Recording + +logger = getLogger(__name__) + + +class MetadataCollectorException(Exception): + """Generic exception in the metadata collector.""" + + +class MetadataCollectorService: + """Service for dispatching and managing the metadata collector agent.""" + + @async_to_sync + async def start(self, recording: Recording): + """Explicitly dispatch the metadata collector agent to a room.""" + + lkapi = utils.create_livekit_client() + room_id = str(recording.room.id) + + try: + response = await lkapi.agent_dispatch.create_dispatch( + CreateAgentDispatchRequest( + agent_name=settings.METADATA_COLLECTOR_AGENT_NAME, + room=room_id, + metadata=str(recording.id), + ) + ) + except Exception as e: + logger.exception( + "Failed to create metadata collector agent for room %s", room_id + ) + raise MetadataCollectorException( + "Failed to create metadata collector agent" + ) from e + finally: + await lkapi.aclose() + + dispatch_id = getattr(response, "id", None) + + if not dispatch_id: + logger.error("LiveKit response missing dispatch ID for room %s", room_id) + raise MetadataCollectorException( + f"LiveKit did not return a dispatch_id for room {room_id}" + ) + + recording.options["metadata_collector_dispatch_id"] = dispatch_id + await sync_to_async(recording.save)(update_fields=["options"]) + + return dispatch_id + + @async_to_sync + async def stop(self, recording: Recording): + """Stop and delete the agent dispatch associated to the room.""" + + room_id = str(recording.room.id) + dispatch_id = recording.options.get("metadata_collector_dispatch_id") + lkapi = utils.create_livekit_client() + + try: + if not dispatch_id: + logger.warning( + "No metadata collector dispatch ID stored for room %s", room_id + ) + return None + + await lkapi.agent_dispatch.delete_dispatch( + dispatch_id=str(dispatch_id), room_name=room_id + ) + + except Exception as e: + logger.exception( + "Failed to stop metadata collector agent dispatch for room %s", + room_id, + ) + raise MetadataCollectorException( + f"Failed to stop metadata collector agent for room {room_id}" + ) from e + finally: + await lkapi.aclose() diff --git a/src/backend/core/recording/services/recording_events.py b/src/backend/core/recording/services/recording_events.py index 48a682fb..8fac30aa 100644 --- a/src/backend/core/recording/services/recording_events.py +++ b/src/backend/core/recording/services/recording_events.py @@ -7,6 +7,7 @@ from logging import getLogger from livekit import api from core import models, utils +from core.models import Recording logger = getLogger(__name__) @@ -19,7 +20,7 @@ class RecordingEventsService: """Handles recording-related LiveKit webhook events.""" @staticmethod - def handle_update(recording, egress_status): + def handle_update(recording: Recording, egress_status): """Handle egress status updates and sync recording state to room metadata.""" room_name = str(recording.room.id) @@ -40,7 +41,7 @@ class RecordingEventsService: logger.exception("Failed to update room's metadata: %s", e) @staticmethod - def handle_limit_reached(recording): + def handle_limit_reached(recording: Recording): """Stop recording and notify participants when limit is reached.""" recording.status = models.RecordingStatusChoices.STOPPED diff --git a/src/backend/core/services/livekit_events.py b/src/backend/core/services/livekit_events.py index d24c241e..4815a06b 100644 --- a/src/backend/core/services/livekit_events.py +++ b/src/backend/core/services/livekit_events.py @@ -12,6 +12,10 @@ from django.conf import settings from livekit import api from core import models, utils +from core.recording.services.metadata_collector import ( + MetadataCollectorException, + MetadataCollectorService, +) from core.recording.services.recording_events import ( RecordingEventsError, RecordingEventsService, @@ -158,7 +162,7 @@ class LiveKitEventsService: """Handle 'egress_ended' event.""" try: - recording = models.Recording.objects.get( + recording = models.Recording.objects.select_related("room").get( worker_id=data.egress_info.egress_id ) except models.Recording.DoesNotExist as err: @@ -174,6 +178,14 @@ class LiveKitEventsService: except utils.MetadataUpdateException as e: logger.exception("Failed to update room's metadata: %s", e) + if settings.METADATA_COLLECTOR_ENABLED and recording.options.get( + "metadata_collector_dispatch_id" + ): + try: + MetadataCollectorService().stop(recording) + except MetadataCollectorException: + logger.warning("Failed to stop the MetadataCollectorService") + if ( data.egress_info.status == api.EgressStatus.EGRESS_LIMIT_REACHED and recording.status == models.RecordingStatusChoices.ACTIVE diff --git a/src/backend/core/tests/rooms/test_api_rooms_start_recording.py b/src/backend/core/tests/rooms/test_api_rooms_start_recording.py index 36900c4c..07895051 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_start_recording.py +++ b/src/backend/core/tests/rooms/test_api_rooms_start_recording.py @@ -277,6 +277,7 @@ def test_start_recording_options_transcribe_valid_true( ): """Should accept transcribe with any valid pydantic true values.""" settings.RECORDING_ENABLE = True + settings.METADATA_COLLECTOR_ENABLED = False room = RoomFactory() user = UserFactory() room.accesses.create(user=user, role="owner") @@ -487,6 +488,93 @@ def test_start_recording_options_original_mode_omitted( assert recording.options == {} +def test_start_recording_calls_metadata_collector_start( + settings, mock_worker_service_factory, mock_worker_manager +): + """Should call MetadataCollectorService.start when conditions are met.""" + settings.RECORDING_ENABLE = True + settings.METADATA_COLLECTOR_ENABLED = True + + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + + client = APIClient() + client.force_login(user) + + with mock.patch( + "core.api.viewsets.MetadataCollectorService" + ) as mock_collector_class: + mock_collector = mock.Mock() + mock_collector_class.return_value = mock_collector + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + { + "mode": "screen_recording", + "options": {"transcribe": True, "collect_metadata": True}, + }, + format="json", + ) + + assert response.status_code == 201 + + recording = Recording.objects.get(room=room) + mock_collector.start.assert_called_once_with(recording) + + +@pytest.mark.parametrize( + "metadata_enabled,options", + [ + # Metadata collector disabled, regardless of transcribe option + (False, {"transcribe": True}), + (False, {"transcribe": False}), + (False, None), + # Metadata collector enabled, but transcribe is False or missing + (True, {"transcribe": False}), + (True, None), + # Metadata collector enabled, transcribe True, but collect_metadata explicitly False + (True, {"transcribe": True, "collect_metadata": False}), + ], +) +def test_start_recording_does_not_call_metadata_collector_start_when_conditions_not_met( + settings, + mock_worker_service_factory, + mock_worker_manager, + metadata_enabled, + options, +): + """Should not call MetadataCollectorService.start when conditions are not met.""" + settings.RECORDING_ENABLE = True + settings.METADATA_COLLECTOR_ENABLED = metadata_enabled + + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + + client = APIClient() + client.force_login(user) + + payload = {"mode": "screen_recording"} + if options is not None: + payload["options"] = options + + with mock.patch( + "core.api.viewsets.MetadataCollectorService" + ) as mock_collector_class: + mock_collector = mock.Mock() + mock_collector_class.return_value = mock_collector + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + payload, + format="json", + ) + + assert response.status_code == 201 + mock_collector.start.assert_not_called() + + @pytest.mark.parametrize("value", ["invalid_mode", "foo", 123, "SCREEN_RECORDING"]) def test_start_recording_options_original_mode_invalid(settings, value): """Should reject invalid recording mode values for original_mode.""" diff --git a/src/backend/core/tests/services/test_livekit_events.py b/src/backend/core/tests/services/test_livekit_events.py index 54455c47..6ace6d7a 100644 --- a/src/backend/core/tests/services/test_livekit_events.py +++ b/src/backend/core/tests/services/test_livekit_events.py @@ -269,6 +269,63 @@ def test_handle_egress_ended_recording_not_limit_reached( assert recording.status == "stopped" +@mock.patch("core.services.livekit_events.MetadataCollectorService") +@mock.patch("core.utils.update_room_metadata") +def test_handle_egress_ended_calls_metadata_collector_stop_when_conditions_are_met( + mock_update_room_metadata, mock_collector_class, service, settings +): + """Should call MetadataCollectorService.stop when it exists.""" + settings.METADATA_COLLECTOR_ENABLED = True + + recording = RecordingFactory( + worker_id="worker-1", + status="active", + options={"metadata_collector_dispatch_id": "dispatch-123"}, + ) + mock_data = mock.MagicMock() + mock_data.egress_info.egress_id = recording.worker_id + mock_data.egress_info.status = EgressStatus.EGRESS_COMPLETE + + mock_collector = mock.Mock() + mock_collector_class.return_value = mock_collector + + service._handle_egress_ended(mock_data) + + mock_collector.stop.assert_called_once_with(recording) + + +@pytest.mark.parametrize( + "metadata_enabled,options", + [ + (True, {}), + (False, {}), + ], +) +@mock.patch("core.services.livekit_events.MetadataCollectorService") +@mock.patch("core.utils.update_room_metadata") +def test_handle_egress_ended_does_not_call_metadata_collector_stop_when_conditions_not_met( + _, mock_collector_class, metadata_enabled, options, service, settings +): # pylint: disable=too-many-arguments,too-many-positional-arguments + """Should not call MetadataCollectorService.stop when it does not exist.""" + settings.METADATA_COLLECTOR_ENABLED = metadata_enabled + + recording = RecordingFactory( + worker_id="worker-1", + status="active", + options=options, + ) + mock_data = mock.MagicMock() + mock_data.egress_info.egress_id = recording.worker_id + mock_data.egress_info.status = EgressStatus.EGRESS_COMPLETE + + mock_collector = mock.Mock() + mock_collector_class.return_value = mock_collector + + service._handle_egress_ended(mock_data) + + mock_collector.stop.assert_not_called() + + @mock.patch.object(LobbyService, "clear_room_cache") @mock.patch.object(TelephonyService, "delete_dispatch_rule") def test_handle_room_finished_clears_cache_and_deletes_dispatch_rule( diff --git a/src/backend/meet/settings.py b/src/backend/meet/settings.py index 760038bd..a0418ff6 100755 --- a/src/backend/meet/settings.py +++ b/src/backend/meet/settings.py @@ -808,6 +808,16 @@ class Base(Configuration): environ_prefix=None, ) + # Metadata collector settings + METADATA_COLLECTOR_ENABLED = values.BooleanValue( + False, environ_name="METADATA_COLLECTOR_ENABLED", environ_prefix=None + ) + METADATA_COLLECTOR_AGENT_NAME = values.Value( + "metadata-collector", + environ_name="METADATA_COLLECTOR_AGENT_NAME", + environ_prefix=None, + ) + # External Applications APPLICATION_CLIENT_ID_LENGTH = values.PositiveIntegerValue( 40,