Let the publishing Connection authenticate itself

getLocalTransport no longer authenticates with the SFU up front and hands
the resulting config to the connection. Instead the ConnectionManager tags
the connection whose transport equals the local one as the publisher, and
every Connection fetches its own SFU config for its role.

The local member now only joins the MatrixRTC session once its connection
is authenticated, so that the LiveKit room exists before remote members
try to get a subscriber token for it. Subscribers authenticating with the
OpenID flow lack permission to create the room and would otherwise fail
without retrying, causing a cascade of leaves and rejoins at the start of
a call.

src/state/CallViewModel/CallViewModel.ts

@@ -112,10 +112,7 @@ import {
   enterRTCSession,
   TransportState,
 } from "./localMember/LocalMember.ts";
-import {
+import { getLocalTransport } from "./localMember/LocalTransport.ts";
-  getLocalTransport,
-  type LocalTransport,
-} from "./localMember/LocalTransport.ts";
 import {
   createKeyRotationSuppressed$,
   createMemberships$,
@@ -216,7 +213,7 @@ export interface CallViewModelOptions {
    */
   windowSize$: Behavior<{ width: number; height: number }>;
   /** Optional value overriding the local transport, for testing purposes. */
-  localTransport?: LocalTransport;
+  localTransport?: LivekitTransport;
   /** Optional value overriding the connection factory, for testing purposes. */
   connectionFactory?: ConnectionFactory;
   /** The version & compatibility mode of MatrixRTC that we should use. */
@@ -565,14 +562,7 @@ export function createCallViewModel$(
 
   const localTransport$ = options.localTransport
     ? constant(options.localTransport)
-    : from(
+    : from(getLocalTransport(client));
-        getLocalTransport({
-          ownMembershipIdentity,
-          client,
-          roomId: matrixRoom.roomId,
-          slotId: matrixRTCSession.slotId,
-        }),
-      );
 
   const connectionFactory =
     options.connectionFactory ??

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: {
+        localTransport: exampleTransport,
-          transport: exampleTransport,
-          sfuConfig: exampleSfuConfig,
-        },
         connectionFactory: {
           createConnection(
             scope,
+            role,
             transport,
             ownMembershipIdentity,
             logger,
-            sfuConfig,
           ) {
             return new MockConnection(
               {
                 scope,
+                role,
                 transport,
                 ownMembershipIdentity,
-                existingSFUConfig: sfuConfig,
                 client: room.client,
                 roomId: room.roomId,
                 slotId: "m.call#ROOM",

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

@@ -65,7 +65,6 @@ import {
 import { type Publisher } from "./Publisher";
 import { initializeWidget } from "../../../widget";
 import { nullHostBridge } from "../../../HostBridge";
-import { type LocalTransport } from "./LocalTransport";
 import * as livekitAuth from "../../../livekit/auth";
 
 initializeWidget();
@@ -221,7 +220,10 @@ describe("LocalMembership", () => {
       processor: undefined,
     }),
     logger: logger,
-    createPublisherFactory: vi.fn(),
+    createPublisherFactory: vi.fn().mockReturnValue({
+      destroy: vi.fn(),
+      createAndSetupTracks: vi.fn().mockResolvedValue(undefined),
+    }),
     joinMatrixRTC: async (): Promise<void> => {},
     homeserverConnected: {
       combined$: constant<[boolean, HomeserverDisconnectReason | null]>([
@@ -254,6 +256,7 @@ describe("LocalMembership", () => {
   afterEach(async () => {
     void (await fetchMock.flush());
     fetchMock.reset();
+    vi.clearAllMocks();
   });
 
   it("throws error on missing RTC config error", () => {
@@ -322,20 +325,10 @@ describe("LocalMembership", () => {
     scope.end();
   });
 
-  const mockTransportConfig = {
+  const mockTransport = {
     livekit_service_url: "a",
   } as LivekitTransport;
 
-  const mockTransport = {
-    transport: mockTransportConfig,
-    sfuConfig: {
-      jwt: "foo",
-      livekitAlias: "bar",
-      livekitIdentity: "baz",
-      url: "bro",
-    },
-  } as LocalTransport;
-
   const connectionTransportAConnected = {
     livekitRoom: mockLivekitRoom({
       localParticipant: {
@@ -347,7 +340,7 @@ describe("LocalMembership", () => {
       state: "authenticated",
       livekitState: LivekitConnectionState.Connected,
     }),
-    transport: mockTransportConfig,
+    transport: mockTransport,
   } as Connection;
   const connectionTransportAConnecting = {
     ...connectionTransportAConnected,
@@ -381,13 +374,13 @@ describe("LocalMembership", () => {
       if (delegationUrl !== null)
         fetchMock.post(delegationUrl, () => ({ status: 401, body: {} }));
 
+      const connectionManagerData = new ConnectionManagerData();
+      connectionManagerData.add(connectionTransportAConnected, []);
       const localMembership = createLocalMembership$({
         scope,
         ...defaultCreateLocalMemberValues,
         connectionManager: {
-          connectionManagerData$: constant(
+          connectionManagerData$: constant(new Epoch(connectionManagerData)),
-            new Epoch(new ConnectionManagerData()),
-          ),
         },
         joinMatrixRTC,
         localTransport$: constant(mockTransport),
@@ -399,7 +392,7 @@ describe("LocalMembership", () => {
       await flushPromises();
       // Joins with timings appropriate for the level of delegation support
       expect(joinMatrixRTC).toHaveBeenCalledWith(
-        mockTransportConfig,
+        mockTransport,
         delayedLeaveTimings,
       );
 
@@ -491,7 +484,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<LivekitTransport>();
 
     const connectionManagerData$ = new BehaviorSubject(
       new Epoch(connectionManagerData),
@@ -570,7 +563,7 @@ describe("LocalMembership", () => {
     });
 
     (
-      connectionManagerData2.getConnectionForTransport(mockTransportConfig)!
+      connectionManagerData2.getConnectionForTransport(mockTransport)!
         .state$ as BehaviorSubject<ConnectionState>
     ).next({
       state: "authenticated",
@@ -891,7 +884,7 @@ describe("LocalMembership", () => {
           state: "authenticated",
           livekitState: LivekitConnectionState.Connected,
         }),
-        transport: mockTransportConfig,
+        transport: mockTransport,
         livekitRoom: mockLivekitRoom({
           localParticipant: mockLocalParticipant({
             isScreenShareEnabled: false,

src/state/CallViewModel/localMember/LocalMember.ts

@@ -78,7 +78,6 @@ import {
   type Connection,
 } from "../remoteMembers/Connection.ts";
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
-import { type LocalTransport } from "./LocalTransport.ts";
 import { type ClientOpenIDParts, getSFUConfig } from "../../../livekit/auth";
 
 export enum TransportState {
@@ -148,7 +147,7 @@ interface Props {
   homeserverConnected: HomeserverConnected;
   roomId: string;
   ownMembershipIdentity: CallMembershipIdentityParts;
-  localTransport$: Observable<LocalTransport>;
+  localTransport$: Observable<LivekitTransport>;
   client: ClientOpenIDParts;
   matrixRTCSession: Pick<
     MatrixRTCSession,
@@ -307,7 +306,7 @@ export const createLocalMembership$ = ({
   // to whether delayed event delegation is supported
   const joinParams$ = scope.behavior(
     localTransport$.pipe(
-      switchMap(async ({ transport }) => {
+      switchMap(async (transport) => {
         const transportSupportsDelegation = checkDelegationSupport(
           transport.livekit_service_url + "/delegate_delayed_leave",
           `transport ${transport.livekit_service_url}`,
@@ -329,7 +328,7 @@ export const createLocalMembership$ = ({
       connectionManager.connectionManagerData$,
       localTransport$,
     ]).pipe(
-      map(([{ value: connectionData }, { transport }]) =>
+      map(([{ value: connectionData }, transport]) =>
         connectionData.getConnectionForTransport(transport),
       ),
       tap((connection) => {
@@ -480,6 +479,10 @@ export const createLocalMembership$ = ({
     ),
   );
 
+  const authenticated$ = scope.behavior(
+    localConnectionState$.pipe(map(({ state }) => state === "authenticated")),
+  );
+
   const mediaState$: Behavior<LocalMemberMediaState> = scope.behavior(
     combineLatest([
       localConnectionState$,
@@ -664,34 +667,38 @@ export const createLocalMembership$ = ({
 
   // Join and leave the session as needed
   scope.reconcile(
-    scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])),
+    scope.behavior(
-    async ([joinParams, shouldConnect]) => {
+      combineLatest([joinParams$, joinAndPublishRequested$, authenticated$]),
-      if (!joinParams) return;
+    ),
-      // if shouldConnect=false we will do the disconnect as the cleanup from the previous reconcile iteration.
+    async ([joinParams, shouldConnect, authenticated]) => {
-      if (!shouldConnect) return;
       const sessionConfig = Config.get().matrix_rtc_session;
 
-      try {
+      // Only join once authenticated, to ensure the LiveKit room exists first.
-        joinMatrixRTC(
+      // Otherwise, subscribers authenticating with the OpenID flow may fail to
-          joinParams.transport,
+      // join (they lack permission to create the room) and never retry.
-          joinParams.delegationSupported
+      if (joinParams && shouldConnect && authenticated) {
-            ? sessionConfig.delegated_delayed_leave
-            : sessionConfig.delayed_leave,
-        );
-      } catch (error) {
-        logger.error("Error entering RTC session", error);
-        if (error instanceof Error)
-          setMatrixError(new MembershipManagerError(error));
-      }
-
-      return Promise.resolve(async (): Promise<void> => {
         try {
-          // TODO Update matrixRTCSession to allow udpating the transport without leaving the session!
+          joinMatrixRTC(
-          await matrixRTCSession.leave(1000);
+            joinParams.transport,
-        } catch (e) {
+            joinParams.delegationSupported
-          logger.error("Error leaving RTC session", e);
+              ? sessionConfig.delegated_delayed_leave
+              : sessionConfig.delayed_leave,
+          );
+        } catch (error) {
+          logger.error("Error entering RTC session", error);
+          if (error instanceof Error)
+            setMatrixError(new MembershipManagerError(error));
         }
-      });
+
+        // Clean-up callback to leave the session
+        return Promise.resolve(async (): Promise<void> => {
+          try {
+            await matrixRTCSession.leave(1000);
+          } catch (e) {
+            logger.error("Error leaving RTC session", e);
+          }
+        });
+      }
     },
   );
 

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

@@ -16,125 +16,51 @@ import {
 } from "vitest";
 import fetchMock from "fetch-mock";
 
-import { mockConfig, ownMemberMock } from "../../../utils/test";
+import { mockConfig } from "../../../utils/test";
 import { getLocalTransport } from "./LocalTransport";
-import {
+import { MatrixRTCTransportMissingError } from "../../../utils/errors";
-  MatrixRTCTransportMissingError,
-  FailToGetOpenIdToken,
-} from "../../../utils/errors";
-import * as livekitAuth from "../../../livekit/auth";
 import { customLivekitUrl } from "../../../settings/settings";
-import { testJWTToken } from "../../../utils/test-fixtures";
+import { type MatrixClient } from "matrix-js-sdk";
 
 describe("LocalTransport", () => {
-  const openIdResponse: livekitAuth.SFUConfig = {
-    url: "https://lk.example.org",
-    jwt: testJWTToken,
-    livekitAlias: "Akph4alDMhen",
-    livekitIdentity: "@lk_user:ABCDEF",
-  };
-
   beforeEach(() => vi.clearAllMocks());
 
   it("throws if config is missing", async () => {
     await expect(
       getLocalTransport({
-        roomId: "!room:example.org",
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
-        client: {
+        getDomain: () => "example.org",
-          // 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(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        slotId: "m.call#ROOM",
       }),
     ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
   });
 
-  it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
-    // Provide a valid config so makeTransportInternal resolves a transport
-    mockConfig({
-      livekit: { livekit_service_url: "https://lk.example.org" },
-    });
-    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(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        slotId: "m.call#ROOM",
-      }),
-    ).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid")));
-  });
-
   it("returns preferred transport", async () => {
-    // Use config so transport discovery succeeds, but delay OpenID JWT fetch
+    // Use config so transport discovery succeeds
     mockConfig({
       livekit: { livekit_service_url: "https://lk.example.org" },
     });
 
-    vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue({
-      url: "https://lk.example.org",
-      jwt: "jwt",
-      livekitAlias: "Akph4alDMhen",
-      livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId,
-    });
-
     expect(
       await getLocalTransport({
-        roomId: "!room:example.org",
+        _unstable_getRTCTransports: async () => Promise.resolve([]),
-        client: {
+        getDomain: () => "example.org",
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          getDomain: () => "example.org",
-          getOpenIdToken: vi.fn(),
-        },
-        ownMembershipIdentity: ownMemberMock,
-        slotId: "m.call#ROOM",
       }),
     ).toStrictEqual({
-      transport: {
+      livekit_service_url: "https://lk.example.org",
-        livekit_service_url: "https://lk.example.org",
+      type: "livekit",
-        type: "livekit",
-      },
-      sfuConfig: {
-        jwt: "jwt",
-        livekitAlias: "Akph4alDMhen",
-        livekitIdentity: "@alice:example.org:DEVICE",
-        url: "https://lk.example.org",
-      },
     });
   });
 
-  type LocalTransportProps = Parameters<typeof getLocalTransport>[0];
-
   describe("transport configuration mechanisms", () => {
-    let localTransportOpts: LocalTransportProps & {
+    let client: MockedObject<
-      client: MockedObject<LocalTransportProps["client"]>;
+      Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">
-    };
+    >;
     beforeEach(() => {
       mockConfig({});
       customLivekitUrl.setValue(customLivekitUrl.defaultValue);
-      localTransportOpts = {
+      client = {
-        ownMembershipIdentity: ownMemberMock,
+        getDomain: vi.fn().mockReturnValue("example.org"),
-        roomId: "!example_room_id",
+        _unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
-        slotId: "m.call#ROOM",
-        client: {
-          getDomain: vi.fn().mockReturnValue("example.org"),
-          // eslint-disable-next-line @typescript-eslint/naming-convention
-          _unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
-          getOpenIdToken: vi.fn(),
-        },
       };
     });
 
@@ -146,86 +72,38 @@ describe("LocalTransport", () => {
       mockConfig({
         livekit: { livekit_service_url: "https://lk.example.org" },
       });
-      vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
-        transport: {
+        livekit_service_url: "https://lk.example.org",
-          livekit_service_url: "https://lk.example.org",
+        type: "livekit",
-          type: "livekit",
-        },
-        sfuConfig: {
-          jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
-          livekitAlias: "Akph4alDMhen",
-          livekitIdentity: "@lk_user:ABCDEF",
-          url: "https://lk.example.org",
-        },
       });
     });
 
     it("supports getting transport via user settings", async () => {
       customLivekitUrl.setValue("https://lk.example.org");
-      vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
-        transport: {
+        livekit_service_url: "https://lk.example.org",
-          livekit_service_url: "https://lk.example.org",
+        type: "livekit",
-          type: "livekit",
-        },
-        sfuConfig: {
-          jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
-          livekitAlias: "Akph4alDMhen",
-          livekitIdentity: "@lk_user:ABCDEF",
-          url: "https://lk.example.org",
-        },
       });
     });
 
     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(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse);
 
-      expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
+      expect(await getLocalTransport(client)).toStrictEqual({
-        transport: {
+        livekit_service_url: "https://lk.example.org",
-          livekit_service_url: "https://lk.example.org",
+        type: "livekit",
-          type: "livekit",
-        },
-        sfuConfig: {
-          jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=",
-          livekitAlias: "Akph4alDMhen",
-          livekitIdentity: "@lk_user:ABCDEF",
-          url: "https://lk.example.org",
-        },
       });
     });
 
-    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" },
-      ]);
-      vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
-        new FailToGetOpenIdToken(new Error("Test driven error")),
-      );
-
-      await expect(getLocalTransport(localTransportOpts)).rejects.toThrow(
-        expect.any(FailToGetOpenIdToken),
-      );
-    });
-
     it("throws if no options are available", async () => {
       await expect(
         getLocalTransport({
-          ownMembershipIdentity: ownMemberMock,
+          getDomain: () => "example.org",
-          roomId: "!example_room_id",
+          _unstable_getRTCTransports: async () => Promise.resolve([]),
-          slotId: "m.call#ROOM",
-          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(),
-          },
         }),
       ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
     });

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -7,63 +7,24 @@ Please see LICENSE in the repository root for full details.
 
 import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import { type MatrixClient } from "matrix-js-sdk";
-import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
+import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
-import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
 
 import { Config } from "../../../config/Config.ts";
-import {
+import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
-  FailToGetOpenIdToken,
-  MatrixRTCTransportMissingError,
-  NoMatrix2AuthorizationService,
-} from "../../../utils/errors.ts";
-import {
-  getSFUConfig,
-  type SFUConfig,
-  type ClientOpenIDParts,
-} from "../../../livekit/auth";
 import { customLivekitUrl } from "../../../settings/settings.ts";
 import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
 
-interface Props {
-  ownMembershipIdentity: CallMembershipIdentityParts;
-  client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
-    ClientOpenIDParts;
-  // Used by the jwt service to create the livekit room and compute the livekit alias.
-  roomId: string;
-  slotId: string;
-}
-
-export interface LocalTransport {
-  transport: LivekitTransport;
-  sfuConfig: SFUConfig;
-}
-
-export function isLocalTransport(
-  obj: LivekitTransport | LocalTransport,
-): obj is LocalTransport {
-  return "transport" in obj && "sfuConfig" in obj;
-}
-
 /**
- * Connects to the JWT service and determines the transport that the local member should use.
+ * Determines the transport to advertise in our MatrixRTC membership and publish
- *
+ * media on.
- * @prop useOldJwtEndpoint Whether to set forceOldJwtEndpoint on the returned transport and to use the old JWT endpoint.
- * This is used when the connection manager needs to know if it has to use the legacy endpoint which implies a string concatenated rtcBackendIdentity.
- * (which is expected for non sticky event based rtc member events)
- * @returns The transport to advertise in our MatrixRTC membership and publish media on.
- * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
  */
-export async function getLocalTransport({
+export async function getLocalTransport(
-  ownMembershipIdentity,
+  client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
-  client,
+): Promise<LivekitTransport> {
-  roomId,
-  slotId,
-}: Props): Promise<LocalTransport> {
-  const logger = rootLogger.getChild("[LocalTransport]");
   const discovery = new RtcTransportAutoDiscovery({
-    client: client,
+    client,
     resolvedConfig: Config.get(),
-    logger: logger,
+    logger: rootLogger.getChild("[LocalTransport]"),
   });
   const customUrl = customLivekitUrl.value$.value;
 
@@ -75,72 +36,5 @@ export async function getLocalTransport({
   if (transport === null)
     throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
 
-  try {
+  return transport;
-    return await doOpenIdAndJWTFromUrl(
-      transport,
-      ownMembershipIdentity,
-      roomId,
-      slotId,
-      client,
-      logger,
-    );
-  } catch (e) {
-    logger.error(
-      `Failed to authenticate to transport ${transport.livekit_service_url}`,
-      e,
-    );
-    throw mapAuthErrorToUserFriendlyError(e);
-  }
-}
-
-/**
- *  Utility to ensure the user can authenticate with the SFU.
- *  We will call `getSFUConfig` once per transport here as it's our
- *  only mechanism of validation. 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.
- *
- *  @param transport The transport to authenticate with.
- *  @param membership The identity of the local member.
- *  @param roomId The room ID to use for the JWT.
- *  @param slotId The slot ID to use for the JWT.
- *  @param client The client to use for the OpenID token.
- *
- *  @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService
- */
-async function doOpenIdAndJWTFromUrl(
-  transport: LivekitTransport,
-  membership: CallMembershipIdentityParts,
-  roomId: string,
-  slotId: string,
-  client: Pick<MatrixClient, "_unstable_getRTCTransports"> & ClientOpenIDParts,
-  logger: Logger,
-): Promise<LocalTransport> {
-  const sfuConfig = await getSFUConfig({
-    client,
-    membership,
-    serviceUrl: transport.livekit_service_url,
-    roomId,
-    slotId,
-    role: "publisher",
-    logger,
-  });
-  return {
-    transport,
-    sfuConfig,
-  };
-}
-
-function mapAuthErrorToUserFriendlyError(e: unknown): Error {
-  if (
-    e instanceof FailToGetOpenIdToken ||
-    e instanceof NoMatrix2AuthorizationService
-  ) {
-    // rethrow as is
-    return e;
-  }
-  // Catch others and rethrow as FailToGetOpenIdToken that has user friendly message.
-  return new FailToGetOpenIdToken(
-    e instanceof Error ? e : new Error(String(e)),
-  );
 }

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

@@ -115,6 +115,7 @@ function setupRemoteConnection(): Connection {
     client,
     roomId: ROOM_ID,
     slotId: SLOT_ID,
+    role: "subscriber",
     transport: livekitFocus,
     scope: testScope,
     ownMembershipIdentity: ownMemberMock,
@@ -159,6 +160,7 @@ describe("Start connection states", () => {
       client,
       roomId: ROOM_ID,
       slotId: SLOT_ID,
+      role: "subscriber",
       transport: livekitFocus,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -177,6 +179,7 @@ describe("Start connection states", () => {
       client,
       roomId: ROOM_ID,
       slotId: SLOT_ID,
+      role: "subscriber",
       transport: livekitFocus,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -230,6 +233,7 @@ describe("Start connection states", () => {
       client,
       roomId: ROOM_ID,
       slotId: SLOT_ID,
+      role: "subscriber",
       transport: livekitFocus,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -296,6 +300,7 @@ describe("Start connection states", () => {
       client,
       roomId: ROOM_ID,
       slotId: SLOT_ID,
+      role: "subscriber",
       transport: livekitFocus,
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -555,6 +560,7 @@ describe("remote track logging", () => {
         client,
         roomId: ROOM_ID,
         slotId: SLOT_ID,
+        role: "subscriber",
         transport: livekitFocus,
         scope: testScope,
         ownMembershipIdentity: ownMemberMock,

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -45,13 +45,10 @@ import {
 } from "../../../livekit/auth";
 
 export interface ConnectionOpts {
-  /**
-   * For the local transport we already do know the jwt token and url. We can reuse it.
-   * On top the local transport will send additional data to the jwt server to use delayed event delegation.
-   */
-  existingSFUConfig?: SFUConfig;
   /** The identity parts to use on this connection */
   ownMembershipIdentity: CallMembershipIdentityParts;
+  /** Whether we want to publish or only subscribe on this connection. */
+  role: "publisher" | "subscriber";
   /** The media transport to connect to. */
   transport: LivekitTransport;
   /** The Matrix client to use for OpenID and SFU config requests. */
@@ -92,6 +89,11 @@ export class Connection {
    */
   public readonly state$: Behavior<ConnectionState> = this._state$;
 
+  /**
+   * Whether we want to publish or only subscribe on this connection.
+   */
+  public readonly role: "publisher" | "subscriber";
+
   /**
    * The media transport to connect to.
    */
@@ -130,7 +132,6 @@ export class Connection {
   private readonly slotId: string;
   private readonly logger: Logger;
   private readonly ownMembershipIdentity: CallMembershipIdentityParts;
-  private readonly existingSFUConfig?: SFUConfig;
   /**
    * Creates a new connection to a matrix RTC LiveKit backend.
    *
@@ -140,19 +141,17 @@ export class Connection {
    */
   public constructor(opts: ConnectionOpts, logger: Logger) {
     this.ownMembershipIdentity = opts.ownMembershipIdentity;
-    this.existingSFUConfig = opts.existingSFUConfig;
     this.roomId = opts.roomId;
     this.slotId = opts.slotId;
     this.logger = logger.getChild(
       `[Connection ${JSON.stringify(opts.transport)}]`,
     );
-    this.logger.info(
+    this.logger.debug(`constructor called`);
-      `constructor: roomId: ${this.roomId} withSfuConfig?: ${opts.existingSFUConfig ? JSON.stringify(opts.existingSFUConfig) : "undefined"}`,
+    const { role, transport, client, scope } = opts;
-    );
-    const { transport, client, scope } = opts;
 
     this.scope = scope;
     this.livekitRoom = opts.livekitRoomFactory();
+    this.role = role;
     this.transport = transport;
     this.client = client;
 
@@ -295,11 +294,7 @@ export class Connection {
     this.stopped = false;
     try {
       this._state$.next({ state: "authenticating" });
-      // We should already have this information after creating the localTransport.
+      const { url, jwt, livekitAlias } = await this.getSFUConfig();
-      // only call getSFUConfig for connections where we do not have a token yet. (existingJwtTokenData === undefined)
-      const { url, jwt, livekitAlias } =
-        this.existingSFUConfig ??
-        (await this.getSFUConfigForRemoteConnection());
       this.logger.debug(
         "Starting Connection - jwt: ",
         jwt,
@@ -381,16 +376,14 @@ export class Connection {
     }
   }
 
-  protected async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
+  protected async getSFUConfig(): Promise<SFUConfig> {
-    // This will only be called for sfu's where we do not publish ourselves.
-    // For the local connection we will use the existingJwtTokenData
     return await getSFUConfig({
       client: this.client,
       membership: this.ownMembershipIdentity,
+      role: this.role,
       serviceUrl: this.transport.livekit_service_url,
       roomId: this.roomId,
       slotId: this.slotId,
-      role: "subscriber",
       logger: this.logger,
     });
   }

src/state/CallViewModel/remoteMembers/ConnectionFactory.ts

@@ -35,16 +35,16 @@ import {
   noiseSuppressionSetting,
   autoGainControlSetting,
 } from "../../../settings/settings.ts";
-import { type ClientOpenIDParts, type SFUConfig } from "../../../livekit/auth";
+import { type ClientOpenIDParts } from "../../../livekit/auth";
 
 // TODO evaluate if this should be done like the Publisher Factory
 export interface ConnectionFactory {
   createConnection(
     scope: ObservableScope,
+    role: "publisher" | "subscriber",
     transport: LivekitTransport,
     ownMembershipIdentity: CallMembershipIdentityParts,
     logger: Logger,
-    sfuConfig?: SFUConfig,
   ): Connection;
 }
 
@@ -102,16 +102,16 @@ export class ECConnectionFactory implements ConnectionFactory {
    */
   public createConnection(
     scope: ObservableScope,
+    role: "publisher" | "subscriber",
     transport: LivekitTransport,
     ownMembershipIdentity: CallMembershipIdentityParts,
     logger: Logger,
-    sfuConfig?: SFUConfig,
   ): Connection {
     return new Connection(
       {
-        existingSFUConfig: sfuConfig,
         roomId: this.roomId,
         slotId: this.slotId,
+        role,
         transport,
         client: this.client,
         scope: scope,

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

@@ -47,7 +47,7 @@ beforeEach(() => {
   vi.mocked(fakeConnectionFactory).createConnection = vi
     .fn()
     .mockImplementation(
-      (scope: ObservableScope, transport: LivekitTransport) => {
+      (scope: ObservableScope, _role: unknown, transport: LivekitTransport) => {
         const mockConnection = {
           transport,
           remoteParticipants$: new BehaviorSubject([]),
@@ -245,7 +245,11 @@ describe("connectionManagerData$ stream", () => {
     vi.mocked(fakeConnectionFactory).createConnection = vi
       .fn()
       .mockImplementation(
-        (scope: ObservableScope, transport: LivekitTransport) => {
+        (
+          scope: ObservableScope,
+          _role: unknown,
+          transport: LivekitTransport,
+        ) => {
           const fakeRemoteParticipants$ = new BehaviorSubject<
             RemoteParticipant[]
           >([]);

src/state/CallViewModel/remoteMembers/ConnectionManager.ts

@@ -27,11 +27,6 @@ import { Epoch, type ObservableScope } from "../../ObservableScope.ts";
 import { generateItemsWithEpoch } from "../../../utils/observable.ts";
 import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
-import {
-  isLocalTransport,
-  type LocalTransport,
-} from "../localMember/LocalTransport.ts";
-import { type SFUConfig } from "../../../livekit/auth";
 
 export class ConnectionManagerData {
   private readonly store: Map<
@@ -87,7 +82,7 @@ export class ConnectionManagerData {
 interface Props {
   scope: ObservableScope;
   connectionFactory: ConnectionFactory;
-  localTransport$: Observable<LocalTransport>;
+  localTransport$: Observable<LivekitTransport>;
   remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
 
   logger: Logger;
@@ -147,17 +142,17 @@ export function createConnectionManager$({
    * externally this is modified via `registerTransports()`.
    */
   const localAndRemoteTransports$: Behavior<
-    Epoch<(LivekitTransport | LocalTransport)[]>
+    Epoch<{ transport: LivekitTransport; role: "publisher" | "subscriber" }[]>
   > = scope.behavior(
     combineLatest([localTransportAsArray$, remoteTransports$]).pipe(
-      // Combine local and remote transports into one transport array
+      // Combine local and remote transports into one transport array, tagged
-      // and set the forceOldJwtEndpoint property on the local transport
+      // with the role we take on each of them
       map(([localTransportAsArray, remoteTransports]) => {
         const dedupedRemote = removeDuplicateTransports(remoteTransports.value);
         const remoteWithoutLocal = dedupedRemote.filter(
           (transport) =>
             !localTransportAsArray.find((l) =>
-              areLivekitTransportsEqual(l.transport, transport),
+              areLivekitTransportsEqual(l, transport),
             ),
         );
         logger.debug(
@@ -167,7 +162,16 @@ export function createConnectionManager$({
           localTransportAsArray,
         );
         return new Epoch(
-          [...localTransportAsArray, ...remoteWithoutLocal],
+          [
+            ...localTransportAsArray.map((transport) => ({
+              transport,
+              role: "publisher" as const,
+            })),
+            ...remoteWithoutLocal.map((transport) => ({
+              transport,
+              role: "subscriber" as const,
+            })),
+          ],
           remoteTransports.epoch,
         );
       }),
@@ -182,40 +186,23 @@ export function createConnectionManager$({
       generateItemsWithEpoch(
         "ConnectionManager connections$",
         function* (transports) {
-          for (const transport of transports) {
+          for (const { transport, role } of transports) {
-            if (isLocalTransport(transport)) {
+            yield {
-              // This is the local transport; only the `LocalTransport`
+              keys: [role, transport.livekit_service_url],
-              // interface has a `sfuConfig` field.
+              data: undefined,
-              yield {
+            };
-                keys: [
-                  transport.transport.livekit_service_url,
-                  transport.sfuConfig,
-                ],
-                data: undefined,
-              };
-            } else {
-              yield {
-                keys: [
-                  transport.livekit_service_url,
-                  undefined as SFUConfig | undefined,
-                ],
-                data: undefined,
-              };
-            }
           }
         },
-        (scope, _data$, serviceUrl, sfuConfig) => {
+        (scope, _data$, role, serviceUrl) => {
           const connection = connectionFactory.createConnection(
             scope,
+            role,
             {
               type: "livekit",
               livekit_service_url: serviceUrl,
             },
             ownMembershipIdentity,
             logger,
-            // TODO: This whole optional SFUConfig parameter is not particularly elegant.
-            // I would like it if connections always fetched the SFUConfig by themselves.
-            sfuConfig,
           );
           // Start the connection immediately
           // Use connection state to track connection progress

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

@@ -89,6 +89,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
       );
       ecConnectionFactory.createConnection(
         testScope,
+        "subscriber",
         exampleTransport,
         ownMemberMock,
         logger,
@@ -138,6 +139,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
       );
       ecConnectionFactory.createConnection(
         testScope,
+        "subscriber",
         exampleTransport,
         ownMemberMock,
         logger,
@@ -174,6 +176,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,
@@ -212,6 +215,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,
@@ -251,6 +255,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
     );
     ecConnectionFactory.createConnection(
       testScope,
+      "subscriber",
       exampleTransport,
       ownMemberMock,
       logger,

src/utils/test.ts

@@ -78,7 +78,6 @@ import {
   type RemoteScreenShareViewModel,
 } from "../state/media/RemoteScreenShareViewModel";
 import { Connection } from "../state/CallViewModel/remoteMembers/Connection";
-import { type SFUConfig } from "../livekit/auth";
 
 export function withFakeTimers(continuation: () => void): void {
   vi.useFakeTimers();
@@ -214,13 +213,6 @@ export const exampleTransport: LivekitTransport = {
   livekit_service_url: "https://lk.example.org",
 };
 
-export const exampleSfuConfig: SFUConfig = {
-  jwt: "foo",
-  livekitAlias: "bar",
-  livekitIdentity: "baz",
-  url: "bro",
-};
-
 export function mockRtcMembership(
   user: string | RoomMember,
   deviceId: string,