✨(backend) let any authenticated user manage the lobby on trusted rooms

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.

src/backend/core/services/livekit_events.py

@@ -23,6 +23,7 @@ from core.recording.services.recording_events import (
 )
 
 from .lobby import LobbyService
+from .presence import PresenceCache
 from .room_management import (
     RoomManagement,
     RoomManagementException,
@@ -99,6 +100,7 @@ class LiveKitEventsService:
             "egress_ended": self._handle_egress_ended,
             "room_started": self._handle_room_started,
             "room_finished": self._handle_room_finished,
+            "participant_left": self._handle_participant_left,
         }
 
         token_verifier = api.TokenVerifier(
@@ -107,6 +109,7 @@ class LiveKitEventsService:
         )
         self.webhook_receiver = api.WebhookReceiver(token_verifier)
         self.lobby_service = LobbyService()
+        self.presence_cache = PresenceCache()
         self.sip_management = SIPManagement()
         self.recording_events = RecordingEventsService()
 
@@ -285,9 +288,31 @@ class LiveKitEventsService:
                     f"Failed to delete sip dispatch rule for room {room_id}"
                 ) from e
 
+        self.presence_cache.clear_room(room_id)
+
         try:
             self.lobby_service.clear_room_cache(room_id)
         except Exception as e:
             raise ActionFailedError(
                 f"Failed to clear room cache for room {room_id}"
             ) from e
+
+    def _handle_participant_left(self, data):
+        """Handle 'participant_left': invalidate the presence cache.
+
+        Presence entries are created lazily (only for users who administrate
+        the lobby of a trusted room), so for most participants this delete is
+        a no-op DEL on a key that never existed. Eager invalidation shrinks
+        the window during which a departed participant could still act on a
+        trusted room's lobby (cache hit until TTL expiry). It is gated behind
+        `PRESENCE_CLEAR_ON_PARTICIPANT_LEFT` so its production impact can be
+        measured and the behaviour reverted independently of the feature.
+        When disabled, invalidation relies on `room_finished` and the TTL.
+        """
+        if not settings.PRESENCE_CLEAR_ON_PARTICIPANT_LEFT:
+            return
+
+        identity = data.participant.identity
+        if not identity:
+            return
+        self.presence_cache.clear(data.room.name, identity)

src/backend/core/services/participants_management.py

@@ -20,6 +20,7 @@ from livekit.protocol.models import ParticipantInfo
 from core import utils
 
 from .lobby import LobbyService
+from .presence import PresenceCache
 
 logger = getLogger(__name__)
 
@@ -72,7 +73,9 @@ class ParticipantsManagement:
 
     @async_to_sync
     async def remove(self, room_name: str, identity: str):
-        """Remove a participant from a room and clear their lobby cache."""
+        """Remove a participant from a room and clear their lobby/presence cache."""
+
+        PresenceCache().clear(room_name, identity)
 
         try:
             LobbyService().clear_participant_cache(
@@ -156,6 +159,30 @@ class ParticipantsManagement:
         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`.

src/backend/core/services/presence.py

@@ -0,0 +1,50 @@
+"""Presence cache.
+
+Redis-backed memo of "this identity is currently connected to this room".
+
+This module is intentionally a *pure cache store* with no dependency on other
+services, so that `participants_management` (which talks to LiveKit) can
+import it without creating an import cycle. The composition of "check cache,
+fall back to LiveKit" lives in
+`ParticipantsManagement.check_if_in_meeting_cached`.
+
+Only positive answers are stored: a sticky negative would lock out someone
+who joins right after a miss for the whole TTL. The TTL is a safety net in
+case an invalidation webhook is lost.
+"""
+
+from uuid import UUID
+
+from django.conf import settings
+from django.core.cache import cache
+
+
+class PresenceCache:
+    """Store and invalidate (room, identity) presence entries."""
+
+    @staticmethod
+    def _get_cache_key(room_id: UUID | str, identity: str) -> str:
+        """Cache key for a (room, identity) presence entry."""
+        return f"{settings.PRESENCE_KEY_PREFIX}_{room_id!s}_{identity}"
+
+    def is_marked_present(self, room_id: UUID | str, identity: str) -> bool:
+        """Return True if a positive presence entry exists in cache."""
+        return bool(cache.get(self._get_cache_key(room_id, identity)))
+
+    def mark_present(self, room_id: UUID | str, identity: str) -> None:
+        """Record that `identity` is in `room_id`."""
+        cache.set(
+            self._get_cache_key(room_id, identity),
+            True,
+            timeout=settings.PRESENCE_CACHE_TIMEOUT,
+        )
+
+    def clear(self, room_id: UUID | str, identity: str) -> None:
+        """Forget presence for one participant (e.g. on participant_left)."""
+        cache.delete(self._get_cache_key(room_id, identity))
+
+    def clear_room(self, room_id: UUID | str) -> None:
+        """Forget presence for every participant of a room (on room_finished)."""
+        keys = cache.keys(self._get_cache_key(room_id, "*"))
+        if keys:
+            cache.delete_many(keys)