mirror of
https://github.com/suitenumerique/meet.git
synced 2026-08-31 20:58:03 +00:00
lebaudantoine
7369379106
On rooms with the `trusted` access level, any authenticated user connected to the meeting can now manage the lobby. Requested by several organizations, and a step toward generalized lobby management once hubs and groups land (same organization only). Being authenticated is not enough to grant the capability: a `trusted` room means "trusted to join", not "trusted to decide who else joins from outside the call". The new `CanManageLobby` permission therefore also requires the requester to be currently connected to the meeting, verified against LiveKit and failing closed, like `IsPresentInMeeting`. The access level itself is never cached and always read fresh, so an owner switching the room back to `restricted` revokes the capability on the very next request - the one guarantee we did not want to trade for performance. Performance is traded elsewhere: the waiting list is polled by every lobby manager, and on a trusted room that audience grows from a few admins to potentially the whole meeting. Hitting LiveKit once per poll per participant would not survive that fan-out, so presence is memoized in Redis (`PresenceCache`, `PRESENCE_CACHE_TIMEOUT`, 1h). Entries are created lazily because only the minority of participants who actually manage a lobby ever need one, and only positive answers are cached because a sticky negative would lock out someone joining right after a miss for the whole TTL. Eager invalidation on `participant_left`, `room_finished` and admin kick keeps the cache honest; the TTL is the safety net when an event is lost, and its value bounds how long a departed participant could still act. Trade-offs in this v0: * `PRESENCE_CLEAR_ON_PARTICIPANT_LEFT` gates the eager invalidation on `participant_left`: its cost is one Redis DELETE per departure, for every departure, so we want to be able to measure it in production and turn it off independently of the feature. When disabled, invalidation relies on `room_finished` and the TTL only, widening the stale window above. * This can put non-trivial pressure on the cache at scale; the rollout will need to be monitored closely. * The `participant_left` webhook must be enabled in the LiveKit deployment, otherwise eager invalidation silently degrades to the TTL-only behavior.
226 lines
7.0 KiB
Python
226 lines
7.0 KiB
Python
"""Participants management service for LiveKit rooms."""
|
|
|
|
# pylint: disable=too-many-arguments,no-name-in-module,too-many-positional-arguments
|
|
# ruff: noqa:PLR0913
|
|
|
|
import json
|
|
import uuid
|
|
from logging import getLogger
|
|
from typing import Dict, Optional
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from livekit.api import (
|
|
MuteRoomTrackRequest,
|
|
RoomParticipantIdentity,
|
|
TwirpError,
|
|
UpdateParticipantRequest,
|
|
)
|
|
from livekit.protocol.models import ParticipantInfo
|
|
|
|
from core import utils
|
|
|
|
from .lobby import LobbyService
|
|
from .presence import PresenceCache
|
|
|
|
logger = getLogger(__name__)
|
|
|
|
|
|
class ParticipantsManagementException(Exception):
|
|
"""Exception raised when a participant management operations fail."""
|
|
|
|
|
|
class ParticipantNotFoundException(ParticipantsManagementException):
|
|
"""Raised when the target participant does not exist in the room."""
|
|
|
|
|
|
class ParticipantsManagement:
|
|
"""Service for managing participants."""
|
|
|
|
@async_to_sync
|
|
async def mute(self, room_name: str, identity: str, track_sid: str):
|
|
"""Mute a specific audio or video track for a participant in a room."""
|
|
|
|
lkapi = utils.create_livekit_client()
|
|
|
|
try:
|
|
await lkapi.room.mute_published_track(
|
|
MuteRoomTrackRequest(
|
|
room=room_name,
|
|
identity=identity,
|
|
track_sid=track_sid,
|
|
muted=True,
|
|
)
|
|
)
|
|
|
|
except TwirpError as e:
|
|
if e.code == "not_found":
|
|
logger.warning(
|
|
"Participant %s not found in room %s, skipping muting",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantNotFoundException("Participant does not exist") from e
|
|
|
|
logger.exception(
|
|
"Unexpected error muting participant %s for room %s",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantsManagementException("Could not mute participant") from e
|
|
|
|
finally:
|
|
await lkapi.aclose()
|
|
|
|
@async_to_sync
|
|
async def remove(self, room_name: str, identity: str):
|
|
"""Remove a participant from a room and clear their lobby/presence cache."""
|
|
|
|
PresenceCache().clear(room_name, identity)
|
|
|
|
try:
|
|
LobbyService().clear_participant_cache(
|
|
room_id=uuid.UUID(room_name), participant_id=identity
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
logger.warning(
|
|
"participants_management.remove: room_name '%s' is not a UUID; "
|
|
"skipping lobby cache clear",
|
|
room_name,
|
|
exc_info=exc,
|
|
)
|
|
|
|
lkapi = utils.create_livekit_client()
|
|
|
|
try:
|
|
await lkapi.room.remove_participant(
|
|
RoomParticipantIdentity(room=room_name, identity=identity)
|
|
)
|
|
except TwirpError as e:
|
|
if e.code == "not_found":
|
|
logger.warning(
|
|
"Participant %s not found in room %s, skipping removing",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantNotFoundException("Participant does not exist") from e
|
|
|
|
logger.exception(
|
|
"Unexpected error removing participant %s for room %s",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantsManagementException("Could not remove participant") from e
|
|
|
|
finally:
|
|
await lkapi.aclose()
|
|
|
|
@async_to_sync
|
|
async def update( # noqa: PLR0917
|
|
self,
|
|
room_name: str,
|
|
identity: str,
|
|
metadata: Optional[Dict] = None,
|
|
attributes: Optional[Dict] = None,
|
|
permission: Optional[Dict] = None,
|
|
name: Optional[str] = None,
|
|
):
|
|
"""Update participant properties such as metadata, attributes, permissions, or name."""
|
|
|
|
lkapi = utils.create_livekit_client()
|
|
|
|
try:
|
|
await lkapi.room.update_participant(
|
|
UpdateParticipantRequest(
|
|
room=room_name,
|
|
identity=identity,
|
|
metadata=json.dumps(metadata),
|
|
permission=permission,
|
|
attributes=attributes,
|
|
name=name,
|
|
)
|
|
)
|
|
|
|
except TwirpError as e:
|
|
if e.code == "not_found":
|
|
logger.warning(
|
|
"Participant %s not found in room %s, skipping update",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantNotFoundException("Participant does not exist") from e
|
|
|
|
logger.exception(
|
|
"Unexpected error updating participant %s for room %s",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantsManagementException("Could not update participant") from e
|
|
|
|
finally:
|
|
await lkapi.aclose()
|
|
|
|
def check_if_in_meeting_cached(self, room_name: str, identity: str) -> bool:
|
|
"""Cache-first variant of `check_if_in_meeting`.
|
|
|
|
Cache hit -> True without touching LiveKit.
|
|
Cache miss -> ask LiveKit; memoize only positive answers.
|
|
|
|
Raises the same exceptions as `check_if_in_meeting` so callers keep
|
|
failing closed the same way.
|
|
"""
|
|
if not room_name or not identity:
|
|
return False
|
|
|
|
presence_cache = PresenceCache()
|
|
|
|
if presence_cache.is_marked_present(room_name, identity):
|
|
return True
|
|
|
|
present = self.check_if_in_meeting(room_name=room_name, identity=identity)
|
|
|
|
if present:
|
|
presence_cache.mark_present(room_name, identity)
|
|
|
|
return present
|
|
|
|
@async_to_sync
|
|
async def check_if_in_meeting(self, room_name: str, identity: str) -> bool:
|
|
"""Check whether `identity` is currently a participant in `room_name`.
|
|
|
|
Raises ParticipantsManagementException for unexpected LiveKit errors
|
|
so callers can fail closed rather than silently allowing the action.
|
|
"""
|
|
|
|
if not room_name or not identity:
|
|
return False
|
|
|
|
lkapi = utils.create_livekit_client()
|
|
|
|
try:
|
|
participant = await lkapi.room.get_participant(
|
|
RoomParticipantIdentity(
|
|
room=room_name,
|
|
identity=identity,
|
|
)
|
|
)
|
|
except TwirpError as e:
|
|
if e.code == "not_found":
|
|
raise ParticipantNotFoundException("Participant does not exist") from e
|
|
|
|
logger.exception(
|
|
"Unexpected error checking participant %s in room %s",
|
|
identity,
|
|
room_name,
|
|
)
|
|
raise ParticipantsManagementException(
|
|
"Could not verify participant presence"
|
|
) from e
|
|
|
|
finally:
|
|
await lkapi.aclose()
|
|
|
|
return (
|
|
participant is not None
|
|
and participant.state != ParticipantInfo.State.DISCONNECTED
|
|
)
|