Merge branch 'main' into valere/component_ec_M1

2026-10-16 23:45:50 +00:00 · 2026-09-09 10:23:47 +02:00
parent 3485ce47bc b90ed5bdf7
commit 7821b465a5
20 changed files with 583 additions and 539 deletions
@@ -15,6 +15,7 @@ import {
  MediaDeviceFailure,
 } from "livekit-client";
 import { observeParticipantEvents } from "@livekit/components-core";
+import { type MatrixClient } from "matrix-js-sdk";
 import {
  Status as RTCSessionStatus,
  type LivekitTransport,
@@ -64,7 +65,10 @@ import {
  screenShareCodec,
  parseResolution,
 } from "../../../settings/settings.ts";
-import { MatrixRTCMode } from "../../../config/ConfigOptions.ts";
+import {
+  MatrixRTCMode,
+  type ResolvedDelayedLeaveTimings,
+} from "../../../config/ConfigOptions.ts";
 import { Config } from "../../../config/Config.ts";
 import {
  ConnectionState,
@@ -74,6 +78,8 @@ import {
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
 import { type LocalTransport } from "./LocalTransport.ts";
 import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
+import { or$ } from "../../../utils/observable.ts";
+import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";

 export enum TransportState {
  /** Not even a transport is available to the LocalMembership */
@@ -135,10 +141,15 @@ interface Props {
  muteStates: MuteStates;
  connectionManager: IConnectionManager;
  createPublisherFactory: (connection: Connection) => Publisher;
-  joinMatrixRTC: (transport: LivekitTransportConfig) => void;
+  joinMatrixRTC: (
+    transport: LivekitTransportConfig,
+    delayedLeaveTimings: ResolvedDelayedLeaveTimings,
+  ) => void;
  homeserverConnected: HomeserverConnected;
  roomId: string;
-  localTransport$: Behavior<LocalTransport>;
+  ownMembershipIdentity: CallMembershipIdentityParts;
+  localTransport: LocalTransport;
+  client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
  matrixRTCSession: Pick<
    MatrixRTCSession,
    "updateCallIntent" | "leaveRoomSession"
@@ -147,6 +158,9 @@ interface Props {
  hideScreensharing: boolean;
  /** The application hosting Element Call, to be kept informed of join/leave. */
  hostBridge: HostBridge;
+  baseUrl: string;
+  delayId$: Behavior<string | null>;
+  matrixRTCMode: MatrixRTCMode;
  logger: Logger;
 }

@@ -161,10 +175,12 @@ interface Props {
 * @param props.createPublisherFactory Factory to create a publisher once we have a connection.
 * @param props.joinMatrixRTC Callback to join the matrix RTC session once we have a transport.
 * @param props.homeserverConnected The homeserver connected state.
- * @param props.localTransport$ The transport to advertise in our membership.
+ * @param props.localTransport The transport to advertise in our membership.
 * @param props.logger The logger to use.
 * @param props.muteStates The mute states for video and audio.
 * @param props.matrixRTCSession The matrix RTC session to join.
+ * @param props.baseUrl Base URL of the homeserver.
+ * @param props.delayId$ ID of the delayed leave event to delegate to the SFU.
 * @param props.roomId The room ID used as the call identifier in analytics events.
 * @param props.hideScreensharing Whether to hide the screen-sharing button.
 * @param props.hostBridge The application hosting Element Call.
@@ -178,16 +194,21 @@ interface Props {
 export const createLocalMembership$ = ({
  scope,
  connectionManager,
-  localTransport$,
+  localTransport,
  homeserverConnected,
  createPublisherFactory,
  joinMatrixRTC,
  logger: parentLogger,
  muteStates,
+  client,
  matrixRTCSession,
+  baseUrl,
  roomId,
  hideScreensharing,
  hostBridge,
+  ownMembershipIdentity,
+  delayId$,
+  matrixRTCMode,
 }: Props): {
  /**
   * This request to start audio and video tracks.
@@ -246,25 +267,69 @@ export const createLocalMembership$ = ({
    return of(null);
  };

-  // This is the transport that we will advertise in our membership.
-  const advertisedTransport$ = localTransport$.pipe(
-    switchMap((lt) => lt.advertised$),
+  async function checkDelegationSupport(
+    endpointUrl: string,
+    serviceName: string,
+  ): Promise<boolean> {
+    logger.info(`Checking whether ${serviceName} supports delegation…`);
+    try {
+      // Bluntly hit the endpoint without auth to check for a 404. Unfortunately
+      // we can't wrap this in a retry loop, as many servers don't just disable
+      // delegation support, but in fact are from a time before the endpoint
+      // existed at all, therefore we can hit CORS errors which would just gum
+      // up the retry loop. (May be revisited after Matrix 2.0.)
+      const res = await fetch(endpointUrl, { method: "POST" });
+      if (res.status === 404) {
+        logger.warn(`${serviceName} does not support delegation`);
+        return false;
+      } else {
+        logger.info(`${serviceName} supports delegation`);
+        return true;
+      }
+    } catch (e) {
+      logger.warn(
+        `Failed to determine whether ${serviceName} supports delegation, assuming no support`,
+        e,
+      );
+      return false;
+    }
+  }
+
+  const homeserverSupportsDelegation = checkDelegationSupport(
+    baseUrl +
+      "/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave",
+    "homeserver",
+  );
+
+  // The transport that we will advertise in our membership, paired with info as
+  // to whether delayed event delegation is supported
+  const joinParams$ = localTransport.advertised$.pipe(
    catchError(handleTransportError),
    distinctUntilChanged(areLivekitTransportsEqual),
+    switchMap((transport) => {
+      if (transport === null) return of(null);
+      const transportSupportsDelegation = checkDelegationSupport(
+        transport.livekit_service_url + "/delegate_delayed_leave",
+        `transport ${transport.livekit_service_url}`,
+      );
+      return or$(
+        from(homeserverSupportsDelegation),
+        from(transportSupportsDelegation),
+      ).pipe(
+        map((delegationSupported) => ({ transport, delegationSupported })),
+        startWith(null),
+      );
+    }),
  );

  // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error.
  const activeTransport$ = scope.behavior(
-    localTransport$.pipe(
-      switchMap((lt) => {
-        return combineLatest([lt.active$, lt.advertised$]).pipe(
-          map(([active, advertised]) => {
-            // Our policy is to not publish to another transport if our prefered transport is miss-configured
-            if (advertised == null) return null;
+    combineLatest([localTransport.active$, localTransport.advertised$]).pipe(
+      map(([active, advertised]) => {
+        // Our policy is to not publish to another transport if our prefered transport is miss-configured
+        if (advertised == null) return null;

-            return active?.transport ?? null;
-          }),
-        );
+        return active?.transport ?? null;
      }),
      catchError(handleTransportError),
      distinctUntilChanged(areLivekitTransportsEqual),
@@ -622,16 +687,20 @@ export const createLocalMembership$ = ({

  // Keep matrix rtc session in sync with advertisedTransport$, connectRequested$
  scope.reconcile(
-    scope.behavior(
-      combineLatest([advertisedTransport$, joinAndPublishRequested$]),
-    ),
-    async ([transport, shouldConnect]) => {
-      if (!transport) return;
+    scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])),
+    async ([joinParams, shouldConnect]) => {
+      if (!joinParams) return;
      // if shouldConnect=false we will do the disconnect as the cleanup from the previous reconcile iteration.
      if (!shouldConnect) return;
+      const sessionConfig = Config.get().matrix_rtc_session;

      try {
-        joinMatrixRTC(transport);
+        joinMatrixRTC(
+          joinParams.transport,
+          joinParams.delegationSupported
+            ? sessionConfig.delegated_delayed_leave
+            : sessionConfig.delayed_leave,
+        );
      } catch (error) {
        logger.error("Error entering RTC session", error);
        if (error instanceof Error)
@@ -658,6 +727,34 @@ export const createLocalMembership$ = ({
    ),
  );

+  // Delegate delayed leaves to the SFU
+  scope.reconcile(
+    scope.behavior(combineLatest([joinParams$, delayId$])),
+    async ([joinParams, delayId]) => {
+      if (joinParams?.delegationSupported && delayId !== null) {
+        try {
+          // This will technically cause the service to issue a new JWT token,
+          // but it's safe to discard. We're only interested in triggering
+          // delegation.
+          await getSFUConfigWithOpenID(
+            client,
+            ownMembershipIdentity,
+            joinParams.transport.livekit_service_url,
+            roomId,
+            { matrixRTCMode, delayEndpointBaseUrl: baseUrl, delayId },
+            logger,
+          );
+        } catch (e) {
+          // TODO: Surface this to the user as a service interruption?
+          logger.error(
+            `Failed to delegate leave to ${joinParams.transport.livekit_service_url}`,
+            e,
+          );
+        }
+      }
+    },
+  );
+
  // Pause upstream of all local media tracks when we're disconnected from
  // MatrixRTC, because it can be an unpleasant surprise for the app to say
  // 'reconnecting' and yet still be transmitting your media to others.
@@ -869,6 +966,7 @@ export function observeSharingScreen$(p: Participant): Observable<boolean> {
 interface EnterRTCSessionOptions {
  encryptMedia: boolean;
  matrixRTCMode: MatrixRTCMode;
+  delayedLeaveTimings: ResolvedDelayedLeaveTimings;
  /** Whether and what kind of notification to send when joining. */
  sendNotificationType?: RTCNotificationType;
  /** The kind of call being placed. */
@@ -886,7 +984,8 @@ interface EnterRTCSessionOptions {
 * @param ownMembershipIdentity - Options for entering the RTC session.
 * @param transport - The LivekitTransport to use for this session.
 * @param options - `encryptMedia`: Whether to encrypt media. `matrixRTCMode`: The
- *   Matrix RTC mode to use. `sendNotificationType`: Whether and what kind of
+ *   Matrix RTC mode to use. `delayedLeaveTimings`: The preferred timings for
+ *   delayed leave events. `sendNotificationType`: Whether and what kind of
 *   notification to send on join. `callIntent`: The kind of call being placed.
 * @throws If the host could not be told that we are joining.
 */
@@ -895,14 +994,14 @@ export function enterRTCSession(
  rtcSession: MatrixRTCSession,
  ownMembershipIdentity: CallMembershipIdentityParts,
  transport: LivekitTransportConfig,
-  options: EnterRTCSessionOptions,
-): void {
-  const {
+  {
    encryptMedia,
    matrixRTCMode,
+    delayedLeaveTimings,
    sendNotificationType: notificationType,
    callIntent,
-  } = options;
+  }: EnterRTCSessionOptions,
+): void {
  PosthogAnalytics.instance.eventCallEnded.cacheStartCall(new Date());
  PosthogAnalytics.instance.eventCallStarted.track(rtcSession.room.roomId);

@@ -910,7 +1009,11 @@ export function enterRTCSession(
  // have started tracking by the time calls start getting created.
  // groupCallOTelMembership?.onJoinCall();

-  const { matrix_rtc_session: matrixRtcSessionConfig } = Config.get();
+  const {
+    sync_disconnect_grace_period_ms: gracePeriod,
+    matrix_rtc_session: sessionConfig,
+  } = Config.get();
+  const retryInterval = sessionConfig.network_error_retry_ms;
  const multiSFU =
    matrixRTCMode === MatrixRTCMode.Compatibility ||
    matrixRTCMode === MatrixRTCMode.Matrix_2_0;
@@ -927,16 +1030,11 @@ export function enterRTCSession(
    };
  }

-  // Calculates `maximumNetworkErrorRetryCount`. The connection is failed if EITHER:
-  // - The /sync loop is unresponsive for > `gracePeriod` ms, or
-  // - A delayed leave event is emitted (after `leaveDelay` ms period).
-  // Note: Use leaveDelay >> gracePeriod for delegated leave events.
-  const gracePeriod = Config.get().sync_disconnect_grace_period_ms;
-  const leaveDelay = matrixRtcSessionConfig?.delayed_leave_event_delay_ms;
-  const retryInterval = matrixRtcSessionConfig?.network_error_retry_ms;
-
+  // Set maximumNetworkErrorRetryCount such that we will consider the client
+  // disconnected as soon as either it fails to sync for longer than the grace
+  // period, or it is likely that a delayed leave event has been sent.
  // Math.min is used to account for the respective worst case: /sync not available or leave event emitted.
-  const maxWaitTime = Math.min(gracePeriod, leaveDelay);
+  const maxWaitTime = Math.min(gracePeriod, delayedLeaveTimings.delay_ms);
  const maximumNetworkErrorRetryCount =
    Math.ceil(maxWaitTime / retryInterval) + 1;

@@ -951,18 +1049,14 @@ export function enterRTCSession(
      notificationType,
      callIntent,
      manageMediaKeys: encryptMedia,
-      delayedLeaveEventRestartMs:
-        matrixRtcSessionConfig?.delayed_leave_event_restart_ms,
-      delayedLeaveEventDelayMs:
-        matrixRtcSessionConfig?.delayed_leave_event_delay_ms,
+      delayedLeaveEventRestartMs: delayedLeaveTimings.restart_ms,
+      delayedLeaveEventDelayMs: delayedLeaveTimings.delay_ms,
      delayedLeaveEventRestartLocalTimeoutMs:
-        matrixRtcSessionConfig?.delayed_leave_event_restart_local_timeout_ms,
-      networkErrorRetryMs: matrixRtcSessionConfig?.network_error_retry_ms,
-      makeKeyDelay: matrixRtcSessionConfig?.wait_for_key_rotation_ms,
-      membershipEventExpiryMs:
-        matrixRtcSessionConfig?.membership_event_expiry_ms,
-      keyRotationParticipantLimit:
-        matrixRtcSessionConfig?.key_rotation_participant_limit,
+        delayedLeaveTimings.restart_timeout_ms,
+      networkErrorRetryMs: sessionConfig.network_error_retry_ms,
+      makeKeyDelay: sessionConfig.wait_for_key_rotation_ms,
+      membershipEventExpiryMs: sessionConfig.membership_event_expiry_ms,
+      keyRotationParticipantLimit: sessionConfig.key_rotation_participant_limit,
      unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
      maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount,
    },