Merge branch 'livekit' into toger5/delayed-event-delegation

src/state/CallViewModel/localMember/LocalMember.ts

@@ -139,7 +139,16 @@ interface Props {
  * We want
  *  - a publisher
  *  -
- * @param param0
+ * @param props The properties required to create the local membership.
+ * @param props.scope The observable scope to use.
+ * @param props.connectionManager The connection manager to get connections from.
+ * @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 local transport to use for publishing.
+ * @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.
  * @returns
  *  - publisher: The handle to create tracks and publish them to the room.
  *  - connected$: the current connection state. Including matrix server and livekit server connection. (only considering the livekit server we are using for our own media publication)
@@ -178,14 +187,18 @@ export const createLocalMembership$ = ({
   // tracks$: Behavior<LocalTrack[]>;
   participant$: Behavior<LocalParticipant | null>;
   connection$: Behavior<Connection | null>;
-  /** Shorthand for homeserverConnected.rtcSession === Status.Reconnecting
-   * Direct translation to the js-sdk membership manager connection `Status`.
+  /**
+   * Tracks the homserver and livekit connected state and based on that computes reconnecting.
    */
   reconnecting$: Behavior<boolean>;
   /** Shorthand for homeserverConnected.rtcSession === Status.Disconnected
    * Direct translation to the js-sdk membership manager connection `Status`.
    */
   disconnected$: Behavior<boolean>;
+  /**
+   * Fully connected
+   */
+  connected$: Behavior<boolean>;
 } => {
   const logger = parentLogger.getChild("[LocalMembership]");
   logger.debug(`Creating local membership..`);
@@ -639,6 +652,7 @@ export const createLocalMembership$ = ({
     localMemberState$,
     participant$,
     reconnecting$,
+    connected$: matrixAndLivekitConnected$,
     disconnected$: scope.behavior(
       homeserverConnected.rtsSession$.pipe(
         map((state) => state === RTCSessionStatus.Disconnected),
@@ -672,9 +686,11 @@ interface EnterRTCSessionOptions {
  *      - Delay events management
  *      - Handles retries (fails only after several attempts)
  *
- * @param rtcSession
- * @param transport
- * @param options
+ * @param rtcSession - The MatrixRTCSession to join.
+ * @param transport - The LivekitTransport to use for this session.
+ * @param options - Options for entering the RTC session.
+ * @param options.encryptMedia - Whether to encrypt media.
+ * @param options.matrixRTCMode - The Matrix RTC mode to use.
  * @throws If the widget could not send ElementWidgetActions.JoinCall action.
  */
 // Exported for unit testing

src/state/CallViewModel/localMember/LocalTransport.test.ts

@@ -5,9 +5,18 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  afterEach,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  type MockedObject,
+  vi,
+} from "vitest";
 import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
-import { BehaviorSubject } from "rxjs";
+import { BehaviorSubject, lastValueFrom } from "rxjs";
+import fetchMock from "fetch-mock";
 
 import { mockConfig, flushPromises, ownMemberMock } from "../../../utils/test";
 import { createLocalTransport$ } from "./LocalTransport";
@@ -18,8 +27,17 @@ import {
   FailToGetOpenIdToken,
 } from "../../../utils/errors";
 import * as openIDSFU from "../../../livekit/openIDSFU";
+import { customLivekitUrl } from "../../../settings/settings";
+import { testJWTToken } from "../../../utils/test-fixtures";
 
 describe("LocalTransport", () => {
+  const openIdResponse: openIDSFU.SFUConfig = {
+    url: "https://lk.example.org",
+    jwt: testJWTToken,
+    livekitAlias: "!example_room_id",
+    livekitIdentity: "@lk_user:ABCDEF",
+  };
+
   let scope: ObservableScope;
   beforeEach(() => (scope = new ObservableScope()));
   afterEach(() => scope.end());
@@ -65,13 +83,15 @@ describe("LocalTransport", () => {
     const errors: Error[] = [];
     const localTransport$ = createLocalTransport$({
       scope,
-      roomId: "!room:example.org",
+      roomId: "!example_room_id",
       useOldestMember$: constant(false),
       memberships$: constant(new Epoch<CallMembership[]>([])),
       client: {
         baseUrl: "https://lk.example.org",
         // Use empty domain to skip .well-known and use config directly
         getDomain: () => "",
+        // eslint-disable-next-line @typescript-eslint/naming-convention
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
         getOpenIdToken: vi.fn(),
         getDeviceId: vi.fn(),
       },
@@ -145,11 +165,13 @@ describe("LocalTransport", () => {
 
     const localTransport$ = createLocalTransport$({
       scope,
-      roomId: "!room:example.org",
+      roomId: "!example_room_id",
       useOldestMember$: constant(true),
       memberships$,
       client: {
         getDomain: () => "",
+        // eslint-disable-next-line @typescript-eslint/naming-convention
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
         getOpenIdToken: vi.fn(),
         getDeviceId: vi.fn(),
         baseUrl: "https://lk.example.org",
@@ -159,14 +181,159 @@ describe("LocalTransport", () => {
       delayId$: constant("delay_id_mock"),
     });
 
-    openIdResolver.resolve?.({ url: "https://lk.example.org", jwt: "jwt" });
+    openIdResolver.resolve?.(openIdResponse);
     expect(localTransport$.value).toBe(null);
     await flushPromises();
     // final
     expect(localTransport$.value).toStrictEqual({
-      livekit_alias: "!room:example.org",
+      livekit_alias: "!example_room_id",
       livekit_service_url: "https://lk.example.org",
       type: "livekit",
     });
   });
+
+  type LocalTransportProps = Parameters<typeof createLocalTransport$>[0];
+
+  describe("transport configuration mechanisms", () => {
+    let localTransportOpts: LocalTransportProps & {
+      client: MockedObject<LocalTransportProps["client"]>;
+    };
+    let openIdResolver: PromiseWithResolvers<openIDSFU.SFUConfig>;
+    beforeEach(() => {
+      mockConfig({});
+      customLivekitUrl.setValue(customLivekitUrl.defaultValue);
+      localTransportOpts = {
+        scope,
+        roomId: "!example_room_id",
+        useOldestMember$: constant(false),
+        memberships$: constant(new Epoch<CallMembership[]>([])),
+        client: {
+          getDomain: vi.fn().mockReturnValue(""),
+          // eslint-disable-next-line @typescript-eslint/naming-convention
+          _unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
+          getOpenIdToken: vi.fn(),
+          getDeviceId: vi.fn(),
+        },
+      };
+      openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
+      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
+        openIdResolver.promise,
+      );
+    });
+
+    afterEach(() => {
+      fetchMock.reset();
+    });
+
+    it("supports getting transport via application config", async () => {
+      mockConfig({
+        livekit: { livekit_service_url: "https://lk.example.org" },
+      });
+      const localTransport$ = createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(localTransport$.value).toBe(null);
+      await flushPromises();
+      expect(localTransport$.value).toStrictEqual({
+        livekit_alias: "!example_room_id",
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      });
+    });
+    it("supports getting transport via user settings", async () => {
+      customLivekitUrl.setValue("https://lk.example.org");
+      const localTransport$ = createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(localTransport$.value).toBe(null);
+      await flushPromises();
+      expect(localTransport$.value).toStrictEqual({
+        livekit_alias: "!example_room_id",
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      });
+    });
+    it("supports getting transport via backend", async () => {
+      localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
+        { type: "livekit", livekit_service_url: "https://lk.example.org" },
+      ]);
+      const localTransport$ = createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(localTransport$.value).toBe(null);
+      await flushPromises();
+      expect(localTransport$.value).toStrictEqual({
+        livekit_alias: "!example_room_id",
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      });
+    });
+    it("fails fast if the openID request fails for backend config", async () => {
+      localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
+        { type: "livekit", livekit_service_url: "https://lk.example.org" },
+      ]);
+      openIdResolver.reject(
+        new FailToGetOpenIdToken(new Error("Test driven error")),
+      );
+      try {
+        await lastValueFrom(createLocalTransport$(localTransportOpts));
+        throw Error("Expected test to throw");
+      } catch (ex) {
+        expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
+      }
+    });
+    it("supports getting transport via well-known", async () => {
+      localTransportOpts.client.getDomain.mockReturnValue("example.org");
+      fetchMock.getOnce("https://example.org/.well-known/matrix/client", {
+        "org.matrix.msc4143.rtc_foci": [
+          { type: "livekit", livekit_service_url: "https://lk.example.org" },
+        ],
+      });
+      const localTransport$ = createLocalTransport$(localTransportOpts);
+      openIdResolver.resolve?.(openIdResponse);
+      expect(localTransport$.value).toBe(null);
+      await flushPromises();
+      expect(localTransport$.value).toStrictEqual({
+        livekit_alias: "!example_room_id",
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      });
+      expect(fetchMock.done()).toEqual(true);
+    });
+    it("fails fast if the openId request fails for the well-known config", async () => {
+      localTransportOpts.client.getDomain.mockReturnValue("example.org");
+      fetchMock.getOnce("https://example.org/.well-known/matrix/client", {
+        "org.matrix.msc4143.rtc_foci": [
+          { type: "livekit", livekit_service_url: "https://lk.example.org" },
+        ],
+      });
+      openIdResolver.reject(
+        new FailToGetOpenIdToken(new Error("Test driven error")),
+      );
+      try {
+        await lastValueFrom(createLocalTransport$(localTransportOpts));
+        throw Error("Expected test to throw");
+      } catch (ex) {
+        expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
+      }
+    });
+    it("throws if no options are available", async () => {
+      const localTransport$ = createLocalTransport$({
+        scope,
+        roomId: "!example_room_id",
+        useOldestMember$: constant(false),
+        memberships$: constant(new Epoch<CallMembership[]>([])),
+        client: {
+          getDomain: () => "",
+          // eslint-disable-next-line @typescript-eslint/naming-convention
+          _unstable_getRTCTransports: async () => Promise.resolve([]),
+          // These won't be called in this error path but satisfy the type
+          getOpenIdToken: vi.fn(),
+          getDeviceId: vi.fn(),
+        },
+      });
+      await flushPromises();
+
+      expect(() => localTransport$.value).toThrow(
+        new MatrixRTCTransportMissingError(""),
+      );
+    });
+  });
 });

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -8,11 +8,11 @@ Please see LICENSE in the repository root for full details.
 import {
   type CallMembership,
   isLivekitTransport,
-  type LivekitTransportConfig,
   type LivekitTransport,
   isLivekitTransportConfig,
+  type Transport,
 } from "matrix-js-sdk/lib/matrixrtc";
-import { type MatrixClient } from "matrix-js-sdk";
+import { MatrixError, type MatrixClient } from "matrix-js-sdk";
 import {
   combineLatest,
   distinctUntilChanged,
@@ -28,7 +28,10 @@ import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/En
 import { type Behavior } from "../../Behavior.ts";
 import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
 import { Config } from "../../../config/Config.ts";
-import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
+import {
+  FailToGetOpenIdToken,
+  MatrixRTCTransportMissingError,
+} from "../../../utils/errors.ts";
 import {
   getSFUConfigWithOpenID,
   type OpenIDClientParts,
@@ -47,7 +50,11 @@ interface Props {
   scope: ObservableScope;
   ownMembershipIdentity: CallMembershipIdentityParts;
   memberships$: Behavior<Epoch<CallMembership[]>>;
-  client: Pick<MatrixClient, "getDomain" | "baseUrl"> & OpenIDClientParts;
+  client: Pick<
+    MatrixClient,
+    "getDomain" | "baseUrl" | "_unstable_getRTCTransports"
+  > &
+    OpenIDClientParts;
   roomId: string;
   useOldestMember$: Behavior<boolean>;
   useOldJwtEndpoint$: Behavior<boolean>;
@@ -141,16 +148,30 @@ export const createLocalTransport$ = ({
 const FOCI_WK_KEY = "org.matrix.msc4143.rtc_foci";
 
 /**
+ * Determine the correct Transport for the current session, including
+ * validating auth against the service to ensure it's correct.
+ * Prefers in order:
  *
- * @param client
- * @param roomId
+
+ * 1. The `urlFromDevSettings` value. If this cannot be validated, the function will throw.
+ * 2. The transports returned via the homeserver.
+ * 3. The transports returned via .well-known.
+ * 4. The transport configured in Element Call's config.
+ *
+ * @param client The authenticated Matrix client for the current user
+ * @param roomId The ID of the room to be connected to.
+ * @param urlFromDevSettings Override URL provided by the user's local config.
  * @param useMatrix2 This implies using the matrix2 jwt endpoint (including delayed event delegation of the jwt token)
  * @param delayId
- * @returns
+ * @returns A fully validated transport config.
  * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
  */
 async function makeTransport(
-  client: Pick<MatrixClient, "getDomain" | "baseUrl"> & OpenIDClientParts,
+  client: Pick<
+    MatrixClient,
+    "getDomain" | "baseUrl" | "_unstable_getRTCTransports"
+  > &
+    OpenIDClientParts,
   membership: CallMembershipIdentityParts,
   roomId: string,
   urlFromDevSettings: string | null,
@@ -159,51 +180,127 @@ async function makeTransport(
 ): Promise<LivekitTransport & { forceOldJwtEndpoint: boolean }> {
   let transport: LivekitTransport | undefined;
   logger.trace("Searching for a preferred transport");
-  //TODO refactor this to use the jwt service returned alias.
-  const livekitAlias = roomId;
+
+  // We will call `getSFUConfigWithOpenID` once per transport here as it's our
+  // only mechanism of valiation. This means we will also ask the
+  // homeserver for a OpenID token a few times. Since OpenID tokens are single
+  // use we don't want to risk any issues by re-using a token.
+  //
+  // If the OpenID request were to fail then it's acceptable for us to fail
+  // this function early, as we assume the homeserver has got some problems.
 
   // DEVTOOL: Highest priority: Load from devtool setting
   if (urlFromDevSettings !== null) {
-    const transportFromStorage: LivekitTransport = {
+    logger.info("Using LiveKit transport from dev tools: ", urlFromDevSettings);
+    // Validate that the SFU is up. Otherwise, we want to fail on this
+    // as we don't permit other SFUs.
+    const config = await getSFUConfigWithOpenID(
+      client,
+      urlFromDevSettings,
+      roomId,
+    );
+    return {
       type: "livekit",
       livekit_service_url: urlFromDevSettings,
-      livekit_alias: livekitAlias,
+      livekit_alias: config.livekitAlias,
     };
-    logger.info(
-      "Using LiveKit transport from dev tools: ",
-      transportFromStorage,
-    );
-    transport = transportFromStorage;
   }
 
-  // WELL_KNOWN: Prioritize the .well-known/matrix/client, if available, over the configured SFU
+  async function getFirstUsableTransport(
+    transports: Transport[],
+  ): Promise<LivekitTransport | null> {
+    for (const potentialTransport of transports) {
+      if (isLivekitTransportConfig(potentialTransport)) {
+        try {
+          const { livekitAlias } = await getSFUConfigWithOpenID(
+            client,
+            potentialTransport.livekit_service_url,
+            roomId,
+          );
+          return {
+            ...potentialTransport,
+            livekit_alias: livekitAlias,
+          };
+        } catch (ex) {
+          if (ex instanceof FailToGetOpenIdToken) {
+            // Explictly throw these
+            throw ex;
+          }
+          logger.debug(
+            `Could not use SFU service "${potentialTransport.livekit_service_url}" as SFU`,
+            ex,
+          );
+        }
+      }
+    }
+    return null;
+  }
+
+  // MSC4143: Attempt to fetch transports from backend.
+  if ("_unstable_getRTCTransports" in client) {
+    try {
+      const selectedTransport = await getFirstUsableTransport(
+        await client._unstable_getRTCTransports(),
+      );
+      if (selectedTransport) {
+        logger.info("Using backend-configured SFU", selectedTransport);
+        return selectedTransport;
+      }
+    } catch (ex) {
+      if (ex instanceof MatrixError && ex.httpStatus === 404) {
+        // Expected, this is an unstable endpoint and it's not required.
+        logger.debug("Backend does not provide any RTC transports", ex);
+      } else if (ex instanceof FailToGetOpenIdToken) {
+        throw ex;
+      } else {
+        // We got an error that wasn't just missing support for the feature, so log it loudly.
+        logger.error(
+          "Unexpected error fetching RTC transports from backend",
+          ex,
+        );
+      }
+    }
+  }
+
+  // Legacy MSC4143 (to be removed) WELL_KNOWN: Prioritize the .well-known/matrix/client, if available.
   const domain = client.getDomain();
-  if (domain && transport === undefined) {
+  if (domain) {
     // we use AutoDiscovery instead of relying on the MatrixClient having already
     // been fully configured and started
     const wellKnownFoci = (await AutoDiscovery.getRawClientConfig(domain))?.[
       FOCI_WK_KEY
     ];
-    if (Array.isArray(wellKnownFoci)) {
-      const wellKnownTransport: LivekitTransportConfig | undefined =
-        wellKnownFoci.find((f) => f && isLivekitTransportConfig(f));
-      if (wellKnownTransport !== undefined) {
-        logger.info("Using LiveKit transport from .well-known: ", transport);
-        transport = { ...wellKnownTransport, livekit_alias: livekitAlias };
-      }
+    const selectedTransport = Array.isArray(wellKnownFoci)
+      ? await getFirstUsableTransport(wellKnownFoci)
+      : null;
+    if (selectedTransport) {
+      logger.info("Using .well-known SFU", selectedTransport);
+      return selectedTransport;
     }
   }
 
   // CONFIG: Least prioritized; Load from config file
   const urlFromConf = Config.get().livekit?.livekit_service_url;
-  if (urlFromConf && transport === undefined) {
-    const transportFromConf: LivekitTransport = {
-      type: "livekit",
-      livekit_service_url: urlFromConf,
-      livekit_alias: livekitAlias,
-    };
-    logger.info("Using LiveKit transport from config: ", transportFromConf);
-    transport = transportFromConf;
+  if (urlFromConf) {
+    try {
+      const { livekitAlias } = await getSFUConfigWithOpenID(
+        client,
+        urlFromConf,
+        roomId,
+      );
+      const selectedTransport: LivekitTransport = {
+        type: "livekit",
+        livekit_service_url: urlFromConf,
+        livekit_alias: livekitAlias,
+      };
+      logger.info("Using config SFU", selectedTransport);
+      return selectedTransport;
+    } catch (ex) {
+      if (ex instanceof FailToGetOpenIdToken) {
+        throw ex;
+      }
+      logger.error("Failed to validate config SFU", ex);
+    }
   }
 
   if (!transport) throw new MatrixRTCTransportMissingError(domain ?? "");

src/state/CallViewModel/localMember/Publisher.ts

@@ -143,7 +143,7 @@ export class Publisher {
     this.logger.debug("createAndSetupTracks called");
     const lkRoom = this.connection.livekitRoom;
     // Observe mute state changes and update LiveKit microphone/camera states accordingly
-    this.observeMuteStates(this.scope);
+    this.observeMuteStates();
 
     // Check if audio and/or video is enabled. We only create tracks if enabled,
     // because it could prompt for permission, and we don't want to do that unnecessarily.
@@ -356,10 +356,9 @@ export class Publisher {
 
   /**
    * Observe changes in the mute states and update the LiveKit room accordingly.
-   * @param scope
    * @private
    */
-  private observeMuteStates(scope: ObservableScope): void {
+  private observeMuteStates(): void {
     const lkRoom = this.connection.livekitRoom;
     this.muteStates.audio.setHandler(async (enable) => {
       try {