WIP

src/state/CallViewModel/CallViewModel.ts

@@ -39,6 +39,7 @@ import {
   timer,
   takeUntil,
   from,
+  concatWith,
 } from "rxjs";
 import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
 import {
@@ -383,7 +384,7 @@ export interface CallViewModel {
    * key is generated when someone joins or leaves.
    */
   keyRotationSuppressed$: Behavior<boolean>;
-  allConnections$: Behavior<ConnectionManagerData>;
+  allConnections$: Behavior<Connection[]>;
   /** Participants sorted by livekit room so they can be used in the audio rendering */
   livekitRoomItems$: Behavior<LivekitRoomItem[]>;
   /** use the layout instead, this is just for the sdk export. */
@@ -591,6 +592,7 @@ export function createCallViewModel$(
   const connectionManager: IConnectionManager = {
     connectionManagerData$: scope.behavior(
       localTransport$.pipe(
+        concatWith(NEVER), // So the Observable doesn't complete prematurely
         mapScoped("connectionManager$", (scope, localTransport) =>
           createConnectionManager$({
             scope,
@@ -738,13 +740,16 @@ export function createCallViewModel$(
   });
 
   const allConnections$ = scope.behavior(
-    connectionManager.connectionManagerData$.pipe(map((d) => d.value)),
+    connectionManager.connectionManagerData$.pipe(
+      map((d) => d.value.getConnections()),
+    ),
   );
   const livekitRoomItems$ = scope.behavior<LivekitRoomItem[]>(
     allConnections$.pipe(
-      switchMap((connections) =>
-        combineLatest(
-          connections.getConnections().map((connection) =>
+      switchMap((connections) => {
+        if (connections.length === 0) return of([]);
+        return combineLatest(
+          connections.map((connection) =>
             remoteMatrixLivekitMembers$.pipe(
               switchMap((members) =>
                 combineLatest(members.value.map((m) => m.participant.value$)),
@@ -758,8 +763,8 @@ export function createCallViewModel$(
               })),
             ),
           ),
-        ),
-      ),
+        );
+      }),
     ),
   );
 

src/state/CallViewModel/CallViewModelTestUtils.ts

@@ -30,7 +30,6 @@ import {
   type CallViewModelOptions,
 } from "./CallViewModel";
 import {
-  exampleSfuConfig,
   exampleTransport,
   mockConfig,
   MockConnection,
@@ -209,24 +208,21 @@ export function withCallViewModel(mode: MatrixRTCMode) {
         livekitRoomFactory,
         connectionState$,
         windowSize$,
-        localTransport: {
-          transport: exampleTransport,
-          sfuConfig: exampleSfuConfig,
-        },
+        localTransport: exampleTransport,
         connectionFactory: {
           createConnection(
             scope,
+            role,
             transport,
             ownMembershipIdentity,
             logger,
-            sfuConfig,
           ) {
             return new MockConnection(
               {
                 scope,
+                role,
                 transport,
                 ownMembershipIdentity,
-                existingSFUConfig: sfuConfig,
                 client: room.client,
                 roomId: room.roomId,
                 livekitRoomFactory,

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

@@ -58,8 +58,8 @@ import { ConnectionState, type Connection } from "../remoteMembers/Connection";
 import { type Publisher } from "./Publisher";
 import { initializeWidget } from "../../../widget";
 import { nullHostBridge } from "../../../HostBridge";
-import { type LocalTransport } from "./LocalTransport";
-import * as openIDSFU from "../../../livekit/openIDSFU";
+import { type SFUConfig } from "../../../livekit/auth";
+import * as livekitAuth from "../../../livekit/auth";
 
 initializeWidget();
 
@@ -110,6 +110,7 @@ const delegatedTimings: ResolvedDelayedLeaveTimings = {
 };
 
 const mockedClient = {
+  baseUrl: "https://matrix.example.org",
   getDomain: vi.fn().mockReturnValue("example.org"),
   getDeviceId: vi.fn().mockReturnValue("AAAA"),
   getOpenIdToken: vi.fn().mockResolvedValue({
@@ -118,6 +119,7 @@ const mockedClient = {
     matrix_server_name: "localhost",
     expires_in: 10000,
   }),
+  _unstable_delegateDelayedLeave: vi.fn().mockResolvedValue({}),
 };
 
 describe("enterRTCSession", () => {
@@ -315,19 +317,17 @@ describe("LocalMembership", () => {
     scope.end();
   });
 
-  const mockTransportConfig = {
-    livekit_service_url: "a",
+  const mockTransport: UnstableLivekitTransport = {
+    type: "livekit",
+    url: "https://sfu.example.org",
   } as UnstableLivekitTransport;
 
-  const mockTransport = {
-    transport: mockTransportConfig,
-    sfuConfig: {
-      jwt: "foo",
-      livekitAlias: "bar",
-      livekitIdentity: "baz",
-      url: "bro",
-    },
-  } as LocalTransport;
+  const mockSFUConfig: SFUConfig = {
+    jwt: "foo",
+    livekitAlias: "bar",
+    livekitIdentity: "baz",
+    url: "bro",
+  };
 
   const connectionTransportAConnected = {
     livekitRoom: mockLivekitRoom({
@@ -337,7 +337,7 @@ describe("LocalMembership", () => {
       } as unknown as LocalParticipant,
     }),
     state$: constant(ConnectionState.LivekitConnected),
-    transport: mockTransportConfig,
+    transport: mockTransport,
   } as Connection;
   const connectionTransportAConnecting = {
     ...connectionTransportAConnected,
@@ -346,8 +346,8 @@ describe("LocalMembership", () => {
   } as unknown as Connection;
 
   const authCallSpy = vi
-    .spyOn(openIDSFU, "getSFUConfigWithOpenID")
-    .mockImplementation(() => mockedClient.getOpenIdToken());
+    .spyOn(livekitAuth, "getSFUConfig")
+    .mockResolvedValue(mockSFUConfig);
   afterEach(() => authCallSpy.mockClear());
 
   it.each([
@@ -386,7 +386,7 @@ describe("LocalMembership", () => {
       await flushPromises();
       // Joins with timings appropriate for the level of delegation support
       expect(joinMatrixRTC).toHaveBeenCalledWith(
-        mockTransportConfig,
+        mockTransport,
         delayedLeaveTimings,
       );
 
@@ -480,7 +480,7 @@ describe("LocalMembership", () => {
   it("tracks livekit state correctly", async () => {
     const scope = new ObservableScope();
     const connectionManagerData = new ConnectionManagerData();
-    const localTransport$ = new Subject<LocalTransport>();
+    const localTransport$ = new Subject<UnstableLivekitTransport>();
 
     const connectionManagerData$ = new BehaviorSubject(
       new Epoch(connectionManagerData),
@@ -553,7 +553,7 @@ describe("LocalMembership", () => {
     });
 
     (
-      connectionManagerData2.getConnectionForTransport(mockTransportConfig)!
+      connectionManagerData2.getConnectionForTransport(mockTransport)!
         .state$ as BehaviorSubject<ConnectionState>
     ).next(ConnectionState.LivekitConnected);
     expect(localMembership.localMemberState$.value).toStrictEqual({
@@ -858,7 +858,7 @@ describe("LocalMembership", () => {
       const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
       const connection = {
         state$: constant(ConnectionState.LivekitConnected),
-        transport: mockTransportConfig,
+        transport: mockTransport,
         livekitRoom: mockLivekitRoom({
           localParticipant: mockLocalParticipant({
             isScreenShareEnabled: false,

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

@@ -22,13 +22,13 @@ import {
   MatrixRTCTransportMissingError,
   FailToGetOpenIdToken,
 } from "../../../utils/errors";
-import * as openIDSFU from "../../../livekit/openIDSFU";
-import { customLivekitUrl } from "../../../settings/settings";
+import * as livekitAuth from "../../../livekit/auth";
+import { customTransport } from "../../../settings/settings";
 import { testJWTToken } from "../../../utils/test-fixtures";
-import { MatrixRTCMode } from "../../../config/ConfigOptions";
+import { type MatrixClient } from "matrix-js-sdk";
 
 describe("LocalTransport", () => {
-  const openIdResponse: openIDSFU.SFUConfig = {
+  const sfuConfig: livekitAuth.SFUConfig = {
     url: "https://lk.example.org",
     jwt: testJWTToken,
     livekitAlias: "Akph4alDMhen",
@@ -40,42 +40,25 @@ describe("LocalTransport", () => {
   it("throws if config is missing", async () => {
     await expect(
       getLocalTransport({
-        roomId: "!room:example.org",
-        client: {
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getDomain: () => "example.org",
-          // These won't be called in this error path but satisfy the type
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
+        getDomain: () => "example.org",
       }),
     ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
   });
 
-  it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
+  it("threes if SFU config fetch fails", async () => {
     // Provide a valid config so makeTransportInternal resolves a transport
     mockConfig({
       livekit: { livekit_service_url: "https://lk.example.org" },
     });
-    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
+    vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
       new FailToGetOpenIdToken(new Error("no openid")),
     );
 
     await expect(
       getLocalTransport({
-        roomId: "!example_room_id",
-        client: {
-          getDomain: () => "example.org",
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
+        getDomain: () => "example.org",
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
       }),
     ).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid")));
   });
@@ -86,7 +69,7 @@ describe("LocalTransport", () => {
       livekit: { livekit_service_url: "https://lk.example.org" },
     });
 
-    vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({
+    vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue({
       url: "https://lk.example.org",
       jwt: "jwt",
       livekitAlias: "Akph4alDMhen",
@@ -95,16 +78,8 @@ describe("LocalTransport", () => {
 
     expect(
       await getLocalTransport({
-        roomId: "!room:example.org",
-        client: {
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getDomain: () => "example.org",
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        matrixRTCMode: MatrixRTCMode.Compatibility,
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
+        getDomain: () => "example.org",
       }),
     ).toStrictEqual({
       transport: {
@@ -120,26 +95,16 @@ describe("LocalTransport", () => {
     });
   });
 
-  type LocalTransportProps = Parameters<typeof getLocalTransport>[0];
-
   describe("transport configuration mechanisms", () => {
-    let localTransportOpts: LocalTransportProps & {
-      client: MockedObject<LocalTransportProps["client"]>;
-    };
+    let client: MockedObject<
+      Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">
+    >;
     beforeEach(() => {
       mockConfig({});
-      customLivekitUrl.setValue(customLivekitUrl.defaultValue);
-      localTransportOpts = {
-        ownMembershipIdentity: ownMemberMock,
-        roomId: "!example_room_id",
-        matrixRTCMode: MatrixRTCMode.Compatibility,
-        client: {
-          getDomain: vi.fn().mockReturnValue("example.org"),
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
-          getOpenIdToken: vi.fn(),
-          getDeviceId: vi.fn(),
-        },
+      customTransport.setValue(customTransport.defaultValue);
+      client = {
+        getDomain: vi.fn().mockReturnValue("example.org"),
+        _unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
       };
     });
 
@@ -151,11 +116,9 @@ describe("LocalTransport", () => {
       mockConfig({
         livekit: { livekit_service_url: "https://lk.example.org" },
       });
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
+      vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
         transport: {
           livekit_service_url: "https://lk.example.org",
           type: "livekit",
@@ -170,12 +133,13 @@ describe("LocalTransport", () => {
     });
 
     it("supports getting transport via user settings", async () => {
-      customLivekitUrl.setValue("https://lk.example.org");
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
+      customTransport.setValue({
+        type: "livekit",
+        livekit_service_url: "https://lk.example.org",
+      });
+      vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
         transport: {
           livekit_service_url: "https://lk.example.org",
           type: "livekit",
@@ -190,14 +154,12 @@ describe("LocalTransport", () => {
     });
 
     it("supports getting transport via backend", async () => {
-      localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
+      client._unstable_getRTCTransports.mockResolvedValue([
         { type: "livekit", livekit_service_url: "https://lk.example.org" },
       ]);
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
-        openIdResponse,
-      );
+      vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
         transport: {
           livekit_service_url: "https://lk.example.org",
           type: "livekit",
@@ -212,14 +174,14 @@ describe("LocalTransport", () => {
     });
 
     it("fails fast if the openID request fails for backend config", async () => {
-      localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
+      client._unstable_getRTCTransports.mockResolvedValue([
         { type: "livekit", livekit_service_url: "https://lk.example.org" },
       ]);
-      vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
+      vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
         new FailToGetOpenIdToken(new Error("Test driven error")),
       );
 
-      await expect(getLocalTransport(localTransportOpts)).rejects.toThrow(
+      await expect(getLocalTransport(client)).rejects.toThrow(
         expect.any(FailToGetOpenIdToken),
       );
     });
@@ -227,17 +189,8 @@ describe("LocalTransport", () => {
     it("throws if no options are available", async () => {
       await expect(
         getLocalTransport({
-          ownMembershipIdentity: ownMemberMock,
-          roomId: "!example_room_id",
-          matrixRTCMode: MatrixRTCMode.Compatibility,
-          client: {
-            getDomain: () => "example.org",
-            // 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(),
-          },
+          getDomain: () => "example.org",
+          _unstable_getRTCTransports: async () => Promise.resolve([]),
         }),
       ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
     });

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -11,7 +11,7 @@ import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
 
 import { Config } from "../../../config/Config.ts";
 import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
-import { customLivekitUrl } from "../../../settings/settings.ts";
+import { customTransport as customTransportSetting } from "../../../settings/settings.ts";
 import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
 
 /**
@@ -22,16 +22,15 @@ export async function getLocalTransport(
   client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
 ): Promise<UnstableLivekitTransport> {
   const discovery = new RtcTransportAutoDiscovery({
-    client: client,
+    client,
     resolvedConfig: Config.get(),
     logger: rootLogger.getChild("[LocalTransport]"),
   });
-  const customUrl = customLivekitUrl.value$.value;
 
-  // Respect the user's custom URL, if set
-  const transport: UnstableLivekitTransport | null = customUrl
-    ? { type: "livekit", livekit_service_url: customUrl }
-    : await discovery.discoverPreferredTransport();
+  // Respect the user's custom transport, if set
+  const transport =
+    customTransportSetting.value$.value ??
+    (await discovery.discoverPreferredTransport());
 
   if (transport === null)
     throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");

src/state/CallViewModel/remoteMembers/Connection.test.ts

@@ -28,7 +28,7 @@ import {
 } from "livekit-client";
 import fetchMock from "fetch-mock";
 import EventEmitter from "events";
-import { type IOpenIDToken } from "matrix-js-sdk";
+import { MatrixError, type IOpenIDToken } from "matrix-js-sdk";
 import { logger, type Logger } from "matrix-js-sdk/lib/logger";
 import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 
@@ -38,17 +38,20 @@ import {
   type ConnectionOpts,
 } from "./Connection.ts";
 import { ObservableScope } from "../../ObservableScope.ts";
-import { type ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
 import {
   ElementCallError,
   FailToGetOpenIdToken,
 } from "../../../utils/errors.ts";
 import { testJWTToken } from "../../../utils/test-fixtures.ts";
 import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts";
+import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
 
 let testScope: ObservableScope;
 
-let client: MockedObject<ClientOpenIDParts>;
+let client: MockedObject<ClientGetTokenParts & ClientOpenIDParts>;
 
 let fakeLivekitRoom: MockedObject<LivekitRoom>;
 
@@ -57,14 +60,14 @@ let fakeLocalParticipant: MockedObject<LocalParticipant>;
 
 const ROOM_ID = "!roomID:example.org";
 
-const livekitFocus: UnstableLivekitTransport = {
-  livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
+const transport: UnstableLivekitTransport = {
   type: "livekit",
+  url: "wss://matrix-rtc.m.localhost/livekit/sfu",
 };
 
 function setupTest(): void {
   testScope = new ObservableScope();
-  client = vi.mocked<ClientOpenIDParts>({
+  client = vi.mocked<ClientGetTokenParts & ClientOpenIDParts>({
     getOpenIdToken: vi.fn().mockResolvedValue({
       access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
       token_type: "Bearer",
@@ -72,7 +75,8 @@ function setupTest(): void {
       expires_in: 3600,
     }),
     getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
-  } as unknown as ClientOpenIDParts);
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
+  });
 
   localParticipantEventEmiter = new EventEmitter();
 
@@ -118,22 +122,13 @@ function setupRemoteConnection(): Connection {
   const opts: ConnectionOpts = {
     client: client,
     roomId: ROOM_ID,
-    transport: livekitFocus,
+    role: "subscriber",
+    transport: transport,
     scope: testScope,
     ownMembershipIdentity: ownMemberMock,
     livekitRoomFactory: () => fakeLivekitRoom,
   };
 
-  fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
-    return {
-      status: 200,
-      body: {
-        url: "wss://matrix-rtc.m.localhost/livekit/sfu",
-        jwt: testJWTToken,
-      },
-    };
-  });
-
   fakeLivekitRoom.connect.mockImplementation(async (): Promise<void> => {
     const changeEv = RoomEvent.ConnectionStateChanged;
 
@@ -174,7 +169,8 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       roomId: ROOM_ID,
-      transport: livekitFocus,
+      role: "subscriber",
+      transport: transport,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -191,7 +187,8 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       roomId: ROOM_ID,
-      transport: livekitFocus,
+      role: "subscriber",
+      transport: transport,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -228,7 +225,7 @@ describe("Start connection states", () => {
     capturedState = capturedStates.pop();
     if (capturedState instanceof Error) {
       expect(capturedState.message).toEqual("Something went wrong");
-      expect(connection.transport).toEqual(livekitFocus);
+      expect(connection.transport).toEqual(transport);
     } else {
       expect.fail("Expected FailedToStart state but got " + capturedState);
     }
@@ -241,7 +238,8 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       roomId: ROOM_ID,
-      transport: livekitFocus,
+      role: "subscriber",
+      transport: transport,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -256,16 +254,15 @@ describe("Start connection states", () => {
     onTestFinished(() => s.unsubscribe());
 
     const deferredSFU = Promise.withResolvers<void>();
-    // mock the /sfu/get call
-    fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, async () => {
+    client._unstable_getLivekitToken.mockImplementation(async () => {
       await deferredSFU.promise;
-      return {
-        status: 500,
-        body: {
+      throw new MatrixError(
+        {
           errcode: "M_LOOKUP_FAILED",
           error: "Failed to look up user info from homeserver",
         },
-      };
+        500,
+      );
     });
 
     connection.start().catch(() => {
@@ -288,7 +285,7 @@ describe("Start connection states", () => {
       expect(capturedState.cause.message).toContain(
         "Failed to look up user info from homeserver",
       );
-      expect(connection.transport).toEqual(livekitFocus);
+      expect(connection.transport).toEqual(transport);
     } else {
       expect.fail("Expected FailedToStart state but got " + capturedState);
     }
@@ -301,7 +298,8 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       roomId: ROOM_ID,
-      transport: livekitFocus,
+      role: "subscriber",
+      transport: transport,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -316,17 +314,6 @@ describe("Start connection states", () => {
     onTestFinished(() => s.unsubscribe());
 
     const deferredSFU = Promise.withResolvers<void>();
-    // mock the /sfu/get call
-    fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
-      return {
-        status: 200,
-        body: {
-          url: "wss://matrix-rtc.m.localhost/livekit/sfu",
-          jwt: testJWTToken,
-        },
-      };
-    });
-
     fakeLivekitRoom.connect.mockImplementation(async () => {
       await deferredSFU.promise;
       throw new Error("Failed to connect to livekit");
@@ -353,7 +340,7 @@ describe("Start connection states", () => {
       expect(capturedState.cause.message).toContain(
         "Failed to connect to livekit",
       );
-      expect(connection.transport).toEqual(livekitFocus);
+      expect(connection.transport).toEqual(transport);
     } else {
       expect.fail(
         "Expected FailedToStart state but got " + JSON.stringify(capturedState),
@@ -550,7 +537,8 @@ describe("remote track logging", () => {
       {
         client,
         roomId: ROOM_ID,
-        transport: livekitFocus,
+        role: "subscriber",
+        transport: transport,
         scope: testScope,
         ownMembershipIdentity: ownMemberMock,
         livekitRoomFactory: () => fakeLivekitRoom,

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -102,7 +102,10 @@ export class Connection {
    */
   public readonly state$: Behavior<ConnectionState | Error> = this._state$;
 
-  private readonly role: "publisher" | "subscriber";
+  /**
+   * Whether we want to publish or only subscribe on this connection.
+   */
+  public readonly role: "publisher" | "subscriber";
 
   /**
    * The media transport to connect to.

src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import { type RemoteParticipant } from "livekit-client";
 import { logger } from "matrix-js-sdk/lib/logger";
@@ -74,7 +74,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
@@ -115,7 +115,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("abcdef", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1], 1),
@@ -162,7 +162,7 @@ describe("connections$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("abc", {
           a: new Epoch([TRANSPORT_1], 0),
           b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
@@ -226,7 +226,12 @@ describe("connectionManagerData$ stream", () => {
   let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
 
   function keyForTransport(transport: UnstableLivekitTransport): string {
-    return `${transport.livekit_service_url}`;
+    return JSON.stringify([
+      "url" in transport ? transport.url : undefined,
+      "livekit_service_url" in transport
+        ? transport.livekit_service_url
+        : undefined,
+    ]);
   }
 
   beforeEach(() => {
@@ -297,7 +302,7 @@ describe("connectionManagerData$ stream", () => {
       const { connectionManagerData$ } = createConnectionManager$({
         scope: testScope,
         connectionFactory: fakeConnectionFactory,
-        localTransport$: NEVER,
+        localTransport: TRANSPORT_1,
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),

src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts

@@ -14,7 +14,10 @@ import EventEmitter from "events";
 
 import { ObservableScope } from "../../ObservableScope.ts";
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
-import type { ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
+import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
 import {
   exampleTransport,
   mockMediaDevices,
@@ -32,6 +35,7 @@ import {
   cameraBitrate,
   cameraCodec,
 } from "../../../settings/settings.ts";
+import { testJWTToken } from "../../../utils/test-fixtures.ts";
 
 // At the top of your test file, after imports
 vi.mock("livekit-client", async (importOriginal) => {
@@ -51,13 +55,14 @@ vi.mock("livekit-client", async (importOriginal) => {
 });
 
 let testScope: ObservableScope;
-let mockClient: ClientOpenIDParts;
+let mockClient: ClientGetTokenParts & ClientOpenIDParts;
 
 beforeEach(() => {
   testScope = new ObservableScope();
   mockClient = {
     getOpenIdToken: vi.fn().mockReturnValue(""),
     getDeviceId: vi.fn().mockReturnValue("DEV000"),
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
   };
 });
 
@@ -89,6 +94,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
       );
       ecConnectionFactory.createConnection(
         testScope,
+        "subscriber",
         exampleTransport,
         ownMemberMock,
         logger,
@@ -137,6 +143,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
       );
       ecConnectionFactory.createConnection(
         testScope,
+        "subscriber",
         exampleTransport,
         ownMemberMock,
         logger,
@@ -172,6 +179,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,
@@ -209,6 +217,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,
@@ -247,6 +256,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,

src/state/CallViewModel/remoteMembers/integration.test.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { test, vi, expect, beforeEach, afterEach } from "vitest";
-import { BehaviorSubject, NEVER } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { type Room as LivekitRoom } from "livekit-client";
 import EventEmitter from "events";
 import fetchMock from "fetch-mock";
@@ -19,8 +19,12 @@ import {
   trackEpoch,
 } from "../../ObservableScope.ts";
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
-import { type ClientOpenIDParts } from "../../../livekit/openIDSFU.ts";
 import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
+import {
+  exampleTransport,
   mockMediaDevices,
   mockRtcMembership,
   ownMemberMock,
@@ -40,7 +44,7 @@ import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
 
 let testScope: ObservableScope;
 let ecConnectionFactory: ECConnectionFactory;
-let mockClient: ClientOpenIDParts;
+let mockClient: ClientGetTokenParts & ClientOpenIDParts;
 let lkRoomFactory: () => LivekitRoom;
 
 const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
@@ -50,6 +54,7 @@ beforeEach(() => {
   mockClient = {
     getOpenIdToken: vi.fn().mockReturnValue(""),
     getDeviceId: vi.fn().mockReturnValue("DEV000"),
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
   };
 
   lkRoomFactory = vi.fn().mockImplementation(() => {
@@ -123,7 +128,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
     const connectionManager = createConnectionManager$({
       scope: testScope,
       connectionFactory: ecConnectionFactory,
-      localTransport$: NEVER,
+      localTransport: exampleTransport,
       remoteTransports$: membershipsAndTransports.transports$,
       logger: logger,
       ownMembershipIdentity: ownMemberMock,