Identify transports by server name as well as URLs

A transport with only a LiveKit SFU `url` has to be authenticated through
our own homeserver, which needs to know the name of the server owning the
SFU. Carry that name alongside the transport as a TransportLocator,
derived from the member's user ID for remote transports and from the
client's domain for our own, and key the connection map on the SFU URL,
the service URL and the server name together so that `url`, service URL
and combined transport shapes are deduplicated consistently.

src/livekit/MatrixAudioRenderer.test.tsx

@@ -24,7 +24,7 @@ import { testAudioContext } from "../useAudioContext.test";
 import * as MediaDevicesContext from "../MediaDevicesContext";
 import { LivekitRoomAudioRenderer } from "./MatrixAudioRenderer";
 import {
-  exampleTransport,
+  exampleTransportLocator,
   mockLocalParticipant,
   mockMediaDevices,
   mockRemoteParticipant,
@@ -119,7 +119,7 @@ function renderTestComponent(
       <LivekitRoomAudioRenderer
         validIdentities={participants.map((p) => p.identity)}
         livekitRoom={livekitRoom}
-        transport={exampleTransport}
+        transport={exampleTransportLocator}
       />
     </MediaDevicesProvider>,
   );
@@ -151,7 +151,7 @@ it("should not render or warn for the local participant", () => {
       <LivekitRoomAudioRenderer
         validIdentities={[]}
         livekitRoom={{ remoteParticipants: new Map() } as unknown as Room}
-        transport={exampleTransport}
+        transport={exampleTransportLocator}
       />
     </MediaDevicesProvider>,
   );
@@ -178,7 +178,7 @@ it("should warn only once per unexpected participant", () => {
       <LivekitRoomAudioRenderer
         validIdentities={[]}
         livekitRoom={{ remoteParticipants: new Map() } as unknown as Room}
-        transport={exampleTransport}
+        transport={exampleTransportLocator}
       />
     </MediaDevicesProvider>,
   );

src/livekit/MatrixAudioRenderer.tsx

@@ -19,10 +19,10 @@ import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
 import { useEarpieceAudioConfig } from "../MediaDevicesContext";
 import { useReactiveState } from "../useReactiveState";
 import * as controls from "../controls";
-import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+import { type TransportLocator } from "./auth";
 
 export interface MatrixAudioRendererProps {
-  transport: UnstableLivekitTransport;
+  transport: TransportLocator;
   livekitRoom: LivekitRoom;
   /**
    * The list of participant identities to render audio for.

src/livekit/auth/types.ts

@@ -5,6 +5,17 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
+import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+
+/**
+ * Data identifying a transport and the homeserver to which it belongs, so
+ * consumers know how to reach it.
+ */
+export interface TransportLocator {
+  transport: UnstableLivekitTransport;
+  serverName: string;
+}
+
 /**
  * Configuration and access tokens provided by the SFU on successful authentication.
  */

src/state/CallViewModel/CallViewModel.ts

@@ -165,6 +165,7 @@ import {
   type RingingMediaViewModel,
 } from "../media/RingingMediaViewModel.ts";
 import { type GridTileViewModel } from "../TileViewModel.ts";
+import { type TransportLocator } from "../../livekit/auth";
 
 //TODO
 // Larger rename
@@ -217,7 +218,7 @@ export interface CallViewModelOptions {
    */
   windowSize$: Behavior<{ width: number; height: number }>;
   /** Optional value overriding the local transport, for testing purposes. */
-  localTransport?: UnstableLivekitTransport;
+  localTransport?: TransportLocator;
   /** Optional value overriding the connection factory, for testing purposes. */
   connectionFactory?: ConnectionFactory;
   /** The version & compatibility mode of MatrixRTC that we should use. */
@@ -278,7 +279,7 @@ interface LayoutScanState {
 export type LivekitRoomItem = {
   livekitRoom: LivekitRoom;
   participants: string[];
-  transport: UnstableLivekitTransport;
+  transport: TransportLocator;
 };
 
 /**
@@ -762,7 +763,10 @@ export function createCallViewModel$(
       map((members) =>
         [...Map.groupBy(members, (m) => m.connection)].map(
           ([connection, members]) => ({
-            transport: connection.transport,
+            transport: {
+              transport: connection.transport,
+              serverName: connection.serverName,
+            },
             livekitRoom: connection.livekitRoom,
             participants: members.map((m) => m.participant.identity),
           }),

src/state/CallViewModel/CallViewModelTestUtils.ts

@@ -30,7 +30,7 @@ import {
   type CallViewModelOptions,
 } from "./CallViewModel";
 import {
-  exampleTransport,
+  exampleTransportLocator,
   mockConfig,
   MockConnection,
   mockLivekitRoom,
@@ -208,12 +208,13 @@ export function withCallViewModel(mode: MatrixRTCMode) {
         livekitRoomFactory,
         connectionState$,
         windowSize$,
-        localTransport: exampleTransport,
+        localTransport: exampleTransportLocator,
         connectionFactory: {
           createConnection(
             scope,
             role,
             transport,
+            serverName,
             ownMembershipIdentity,
             logger,
           ) {
@@ -222,6 +223,7 @@ export function withCallViewModel(mode: MatrixRTCMode) {
                 scope,
                 role,
                 transport,
+                serverName,
                 ownMembershipIdentity,
                 client: room.client,
                 roomId: room.roomId,

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

@@ -293,7 +293,7 @@ describe("LocalMembership", () => {
         leave: vi.fn(),
       },
       connectionMap$: constant(new Epoch(new ConnectionMap())),
-      localTransport$: constant(mockTransport),
+      localTransport$: constant(mockTransportLocator),
     });
     const expextedLog =
       "'not connected yet' while updating the call intent (this is expected on startup)";
@@ -308,6 +308,11 @@ describe("LocalMembership", () => {
     livekit_service_url: "a",
   } as UnstableLivekitTransport;
 
+  const mockTransportLocator: livekitAuth.TransportLocator = {
+    transport: mockTransport,
+    serverName: "example.org",
+  };
+
   const connectionTransportAConnected = {
     livekitRoom: mockLivekitRoom({
       localParticipant: {
@@ -320,6 +325,7 @@ describe("LocalMembership", () => {
       livekitState: LivekitConnectionState.Connected,
     }),
     transport: mockTransport,
+    serverName: "example.org",
   } as Connection;
   const connectionTransportAConnecting = {
     ...connectionTransportAConnected,
@@ -360,7 +366,7 @@ describe("LocalMembership", () => {
         ...defaultCreateLocalMemberValues,
         connectionMap$: constant(new Epoch(connectionMap)),
         joinMatrixRTC,
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
         delayId$,
       });
 
@@ -437,7 +443,7 @@ describe("LocalMembership", () => {
       scope,
       ...defaultCreateLocalMemberValues,
       connectionMap$: constant(new Epoch(connectionMap)),
-      localTransport$: constant(mockTransport),
+      localTransport$: constant(mockTransportLocator),
     });
     await flushPromises();
     expect(publisherFactory).toHaveBeenCalledOnce();
@@ -459,7 +465,7 @@ describe("LocalMembership", () => {
   it("tracks livekit state correctly", async () => {
     const scope = new ObservableScope();
     const connectionMap = new ConnectionMap();
-    const localTransport$ = new Subject<UnstableLivekitTransport>();
+    const localTransport$ = new Subject<livekitAuth.TransportLocator>();
 
     const connectionMap$ = new BehaviorSubject(new Epoch(connectionMap));
     const publishers: Publisher[] = [];
@@ -503,7 +509,7 @@ describe("LocalMembership", () => {
     expect(localMembership.localMemberState$.value).toStrictEqual(
       TransportState.Waiting,
     );
-    localTransport$.next(mockTransport);
+    localTransport$.next(mockTransportLocator);
     await flushPromises();
     expect(localMembership.localMemberState$.value).toStrictEqual({
       matrix: RTCMemberStatus.Connected,
@@ -534,7 +540,7 @@ describe("LocalMembership", () => {
     });
 
     (
-      connectionMap2.getConnectionForTransport(mockTransport)!
+      connectionMap2.getConnectionForTransport(mockTransportLocator)!
         .state$ as BehaviorSubject<ConnectionState>
     ).next({
       state: "authenticated",
@@ -641,7 +647,7 @@ describe("LocalMembership", () => {
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
         connectionMap$: constant(new Epoch(connectionMap)),
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
       });
 
       await flushPromises();
@@ -676,7 +682,7 @@ describe("LocalMembership", () => {
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
         connectionMap$: constant(new Epoch(connectionMap)),
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
       });
 
       await flushPromises();
@@ -723,7 +729,7 @@ describe("LocalMembership", () => {
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
         connectionMap$: constant(new Epoch(connectionMap)),
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
       });
 
       await flushPromises();
@@ -768,7 +774,7 @@ describe("LocalMembership", () => {
           rtsSession$: constant(RTCMemberStatus.Connected),
         },
         connectionMap$: constant(new Epoch(connectionMap)),
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
       });
 
       await flushPromises();
@@ -832,7 +838,7 @@ describe("LocalMembership", () => {
         scope,
         ...defaultCreateLocalMemberValues,
         connectionMap$: constant(new Epoch(connectionMap)),
-        localTransport$: constant(mockTransport),
+        localTransport$: constant(mockTransportLocator),
       });
       return { scope, localMembership };
     };
@@ -846,6 +852,7 @@ describe("LocalMembership", () => {
           livekitState: LivekitConnectionState.Connected,
         }),
         transport: mockTransport,
+        serverName: "example.org",
         livekitRoom: mockLivekitRoom({
           localParticipant: mockLocalParticipant({
             isScreenShareEnabled: false,

src/state/CallViewModel/localMember/LocalMember.ts

@@ -78,7 +78,11 @@ import {
   type Connection,
 } from "../remoteMembers/Connection.ts";
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
-import { type ClientOpenIDParts, getSFUConfig } from "../../../livekit/auth";
+import {
+  type ClientOpenIDParts,
+  getSFUConfig,
+  type TransportLocator,
+} from "../../../livekit/auth";
 
 export enum TransportState {
   /** Not even a transport is available to the LocalMembership */
@@ -147,7 +151,7 @@ interface Props {
   homeserverConnected: HomeserverConnected;
   roomId: string;
   ownMembershipIdentity: CallMembershipIdentityParts;
-  localTransport$: Observable<UnstableLivekitTransport>;
+  localTransport$: Observable<TransportLocator>;
   client: ClientOpenIDParts;
   matrixRTCSession: Pick<
     MatrixRTCSession,
@@ -306,7 +310,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 =
           "livekit_service_url" in transport &&
           (await checkDelegationSupport(

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

@@ -46,8 +46,11 @@ describe("LocalTransport", () => {
         getDomain: () => "example.org",
       }),
     ).toStrictEqual({
-      livekit_service_url: "https://lk.example.org",
+      transport: {
-      type: "livekit",
+        livekit_service_url: "https://lk.example.org",
+        type: "livekit",
+      },
+      serverName: "example.org",
     });
   });
 
@@ -74,8 +77,11 @@ describe("LocalTransport", () => {
       });
 
       expect(await getLocalTransport(client)).toStrictEqual({
-        livekit_service_url: "https://lk.example.org",
+        transport: {
-        type: "livekit",
+          livekit_service_url: "https://lk.example.org",
+          type: "livekit",
+        },
+        serverName: "example.org",
       });
     });
 
@@ -83,8 +89,11 @@ describe("LocalTransport", () => {
       customLivekitUrl.setValue("https://lk.example.org");
 
       expect(await getLocalTransport(client)).toStrictEqual({
-        livekit_service_url: "https://lk.example.org",
+        transport: {
-        type: "livekit",
+          livekit_service_url: "https://lk.example.org",
+          type: "livekit",
+        },
+        serverName: "example.org",
       });
     });
 
@@ -94,8 +103,11 @@ describe("LocalTransport", () => {
       ]);
 
       expect(await getLocalTransport(client)).toStrictEqual({
-        livekit_service_url: "https://lk.example.org",
+        transport: {
-        type: "livekit",
+          livekit_service_url: "https://lk.example.org",
+          type: "livekit",
+        },
+        serverName: "example.org",
       });
     });
 

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -13,6 +13,7 @@ import { Config } from "../../../config/Config.ts";
 import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
 import { customLivekitUrl } from "../../../settings/settings.ts";
 import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
+import { type TransportLocator } from "../../../livekit/auth/types.ts";
 
 /**
  * Determines the transport to advertise in our MatrixRTC membership and publish
@@ -20,7 +21,10 @@ import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
  */
 export async function getLocalTransport(
   client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
-): Promise<UnstableLivekitTransport> {
+): Promise<TransportLocator> {
+  const serverName = client.getDomain();
+  if (serverName === null) throw new Error("No server name");
+
   const discovery = new RtcTransportAutoDiscovery({
     client,
     resolvedConfig: Config.get(),
@@ -33,8 +37,7 @@ export async function getLocalTransport(
     ? { type: "livekit", livekit_service_url: customUrl }
     : await discovery.discoverPreferredTransport();
 
-  if (transport === null)
+  if (transport === null) throw new MatrixRTCTransportMissingError(serverName);
-    throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
 
-  return transport;
+  return { transport, serverName };
 }

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

@@ -118,6 +118,7 @@ function setupRemoteConnection(): Connection {
     slotId: SLOT_ID,
     role: "subscriber",
     transport: livekitFocus,
+    serverName: "m.localhost",
     scope: testScope,
     ownMembershipIdentity: ownMemberMock,
     livekitRoomFactory: () => fakeLivekitRoom,
@@ -163,6 +164,7 @@ describe("Start connection states", () => {
       slotId: SLOT_ID,
       role: "subscriber",
       transport: livekitFocus,
+      serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -182,6 +184,7 @@ describe("Start connection states", () => {
       slotId: SLOT_ID,
       role: "subscriber",
       transport: livekitFocus,
+      serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -236,6 +239,7 @@ describe("Start connection states", () => {
       slotId: SLOT_ID,
       role: "subscriber",
       transport: livekitFocus,
+      serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -300,6 +304,7 @@ describe("Start connection states", () => {
       slotId: SLOT_ID,
       role: "subscriber",
       transport: livekitFocus,
+      serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
       livekitRoomFactory: () => fakeLivekitRoom,
@@ -560,6 +565,7 @@ describe("remote track logging", () => {
         slotId: SLOT_ID,
         role: "subscriber",
         transport: livekitFocus,
+        serverName: "m.localhost",
         scope: testScope,
         ownMembershipIdentity: ownMemberMock,
         livekitRoomFactory: () => fakeLivekitRoom,

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -51,6 +51,8 @@ export interface ConnectionOpts {
   role: "publisher" | "subscriber";
   /** The media transport to connect to. */
   transport: UnstableLivekitTransport;
+  /** The name of the homeserver to which the {@link transport} belongs. */
+  serverName: string;
   /** The Matrix client to use for OpenID and SFU config requests. */
   client: ClientOpenIDParts;
   /** The ID of the Matrix room in which the session takes place. */
@@ -99,6 +101,11 @@ export class Connection {
    */
   public readonly transport: UnstableLivekitTransport;
 
+  /**
+   * The name of the homeserver to which the {@link transport} belongs.
+   */
+  public readonly serverName: string;
+
   public readonly livekitRoom: LivekitRoom;
 
   private scope: ObservableScope;
@@ -147,12 +154,13 @@ export class Connection {
       `[Connection ${JSON.stringify(opts.transport)}]`,
     );
     this.logger.debug(`constructor called`);
-    const { role, transport, client, scope } = opts;
+    const { role, transport, serverName, client, scope } = opts;
 
     this.scope = scope;
     this.livekitRoom = opts.livekitRoomFactory();
     this.role = role;
     this.transport = transport;
+    this.serverName = serverName;
     this.client = client;
 
     this.remoteParticipants$ = scope.behavior(

src/state/CallViewModel/remoteMembers/ConnectionFactory.ts

@@ -43,6 +43,7 @@ export interface ConnectionFactory {
     scope: ObservableScope,
     role: "publisher" | "subscriber",
     transport: UnstableLivekitTransport,
+    serverName: string,
     ownMembershipIdentity: CallMembershipIdentityParts,
     logger: Logger,
   ): Connection;
@@ -95,6 +96,7 @@ export class ECConnectionFactory implements ConnectionFactory {
    *
    * @param scope The observable scope (used for clean-up)
    * @param transport The transport to use for this connection.
+   * @param serverName The name of the homeserver to which the {@link transport} belongs.
    * @param ownMembershipIdentity required to connect (using the jwt service) with the SFU.
    * @param logger The logger instance to use for this connection.
    * @param sfuConfig optional config in case we already have a token for this connection.
@@ -104,6 +106,7 @@ export class ECConnectionFactory implements ConnectionFactory {
     scope: ObservableScope,
     role: "publisher" | "subscriber",
     transport: UnstableLivekitTransport,
+    serverName: string,
     ownMembershipIdentity: CallMembershipIdentityParts,
     logger: Logger,
   ): Connection {
@@ -113,6 +116,7 @@ export class ECConnectionFactory implements ConnectionFactory {
         slotId: this.slotId,
         role,
         transport,
+        serverName,
         client: this.client,
         scope: scope,
         livekitRoomFactory: this.livekitRoomFactory,

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

@@ -12,32 +12,41 @@ import { type RemoteParticipant } from "livekit-client";
 import { logger } from "matrix-js-sdk/lib/logger";
 
 import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts";
-import { createConnectionMap$, ConnectionMap } from "./ConnectionMap.ts";
+import {
+  createConnectionMap$,
+  ConnectionMap,
+  keyFor,
+} from "./ConnectionMap.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
 import { type Connection } from "./Connection.ts";
 import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
-import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type Behavior } from "../../Behavior.ts";
+import { type TransportLocator } from "../../../livekit/auth";
 
 // Some test constants
 
-const TRANSPORT_1: UnstableLivekitTransport = {
+const TRANSPORT_1: TransportLocator = {
-  type: "livekit",
+  transport: {
-  livekit_service_url: "https://lk.example.org",
+    type: "livekit",
+    livekit_service_url: "https://lk.example.org",
+  },
+  serverName: "example.org",
 };
 
-const TRANSPORT_2: UnstableLivekitTransport = {
+const TRANSPORT_2: TransportLocator = {
-  type: "livekit",
+  transport: {
-  livekit_service_url: "https://lk.sample.com",
+    type: "livekit",
+    livekit_service_url: "https://lk.sample.com",
+  },
+  serverName: "sample.com",
 };
 
 function findConnection(
-  transport: UnstableLivekitTransport,
+  transport: TransportLocator,
   connections: Connection[],
 ): Connection | undefined {
-  return connections.find((c) =>
+  const key = keyFor(transport);
-    areLivekitTransportsEqual(c.transport, transport),
+  return connections.find((c) => keyFor(c) === key);
-  );
 }
 
 let fakeConnectionFactory: ConnectionFactory;
@@ -57,9 +66,11 @@ beforeEach(() => {
         scope: ObservableScope,
         _role: unknown,
         transport: UnstableLivekitTransport,
+        serverName: string,
       ) => {
         const mockConnection = {
           transport,
+          serverName,
           remoteParticipants$: new BehaviorSubject([]),
         } as unknown as Connection;
         vi.mocked(mockConnection).start = vi.fn();
@@ -210,14 +221,14 @@ describe("ConnectionMap", () => {
   test("warns when a second connection to the same URL is merged", () => {
     const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
     const data = new ConnectionMap(logger);
-    const connection = { transport: TRANSPORT_1 } as unknown as Connection;
+    const connection = TRANSPORT_1 as unknown as Connection;
     const p = (identity: string): RemoteParticipant =>
       ({ identity }) as unknown as RemoteParticipant;
     data.add(connection, [p("a")]);
     data.add({ ...connection } as Connection, [p("b")]);
     expect(warn).toHaveBeenCalledWith(
       expect.stringContaining(
-        "second connection to https://lk.example.org: existing [a], adding [b]",
+        'second connection to [null,"https://lk.example.org","example.org"]: existing [a], adding [b]',
       ),
     );
     expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2);
@@ -228,20 +239,14 @@ describe("connectionMap$", () => {
   // Used in test to control fake connections' remoteParticipants$ streams
   let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
 
-  function keyForTransport(transport: UnstableLivekitTransport): string {
-    return "livekit_service_url" in transport
-      ? transport.livekit_service_url
-      : transport.url;
-  }
-
   beforeEach(() => {
     fakeRemoteParticipantsStreams = new Map();
 
     function getRemoteParticipantsFor(
-      transport: UnstableLivekitTransport,
+      transport: TransportLocator,
     ): Behavior<RemoteParticipant[]> {
       return (
-        fakeRemoteParticipantsStreams.get(keyForTransport(transport)) ??
+        fakeRemoteParticipantsStreams.get(keyFor(transport)) ??
         new BehaviorSubject([])
       );
     }
@@ -254,13 +259,18 @@ describe("connectionMap$", () => {
           scope: ObservableScope,
           _role: unknown,
           transport: UnstableLivekitTransport,
+          serverName: string,
         ) => {
           const fakeRemoteParticipants$ = new BehaviorSubject<
             RemoteParticipant[]
           >([]);
           const mockConnection = {
             transport,
-            remoteParticipants$: getRemoteParticipantsFor(transport),
+            serverName,
+            remoteParticipants$: getRemoteParticipantsFor({
+              transport,
+              serverName,
+            }),
           } as unknown as Connection;
           vi.mocked(mockConnection).start = vi.fn();
           vi.mocked(mockConnection).stop = vi.fn();
@@ -270,7 +280,7 @@ describe("connectionMap$", () => {
           });
 
           fakeRemoteParticipantsStreams.set(
-            keyForTransport(transport),
+            keyFor({ transport, serverName }),
             fakeRemoteParticipants$,
           );
           return mockConnection;
@@ -283,7 +293,7 @@ describe("connectionMap$", () => {
       // Setup the fake participants streams behavior
       // ==============================
       fakeRemoteParticipantsStreams.set(
-        keyForTransport(TRANSPORT_1),
+        keyFor(TRANSPORT_1),
         behavior("oa-b", {
           o: [],
           a: [{ identity: "user1A" } as RemoteParticipant],
@@ -295,7 +305,7 @@ describe("connectionMap$", () => {
       );
 
       fakeRemoteParticipantsStreams.set(
-        keyForTransport(TRANSPORT_2),
+        keyFor(TRANSPORT_2),
         behavior("o-a", {
           o: [],
           a: [{ identity: "user2A" } as RemoteParticipant],

src/state/CallViewModel/remoteMembers/ConnectionMap.ts

@@ -20,44 +20,67 @@ import {
   type ObservableScope,
 } from "../../ObservableScope.ts";
 import { generateItemsWithEpoch } from "../../../utils/observable.ts";
-import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
+import { type TransportLocator } from "../../../livekit/auth/types.ts";
+
+// Branded string type, to enforce that `fromKey` must only be called with
+// values returned by `keyFor`
+export type TransportKey = string & { __transportKey: true };
 
 /**
- * The URL by which a transport is identified: its JWT service URL, or failing
+ * Converts a {@link TransportLocator} to a key suitable for identifying it in a
- * that, its SFU URL.
+ * {@link Map} or {@link Set}.
  */
-function urlOf(transport: UnstableLivekitTransport): string {
+export function keyFor({
-  return "livekit_service_url" in transport
+  transport,
-    ? transport.livekit_service_url
+  serverName,
-    : transport.url;
+}: TransportLocator): TransportKey {
+  // Serialize the fields in a consistent order
+  return JSON.stringify([
+    "url" in transport ? transport.url : null,
+    "livekit_service_url" in transport ? transport.livekit_service_url : null,
+    serverName,
+  ]) as TransportKey;
 }
 
-function removeDuplicateTransports<T extends UnstableLivekitTransport>(
+/**
-  transports: T[],
+ * Inverse of {@link keyFor}: turns a key back into a {@link TransportLocator}.
-): T[] {
+ */
-  return transports.reduce((acc, transport) => {
+export function fromKey(key: TransportKey): TransportLocator {
-    if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
+  const [url, serviceUrl, serverName] = JSON.parse(key);
-      acc.push(transport);
+  return {
-    return acc;
+    transport: {
-  }, [] as T[]);
+      type: "livekit",
+      ...(url === null ? {} : { url }),
+      ...(serviceUrl === null ? {} : { livekit_service_url: serviceUrl }),
+    } as UnstableLivekitTransport,
+    serverName,
+  };
+}
+
+function removeDuplicateTransports(
+  transports: TransportLocator[],
+): TransportLocator[] {
+  const map = new Map<TransportKey, TransportLocator>();
+  for (const transport of transports) map.set(keyFor(transport), transport);
+  return [...map.values()];
 }
 
 export class ConnectionMap {
   private readonly map = new Map<
-    string,
+    TransportKey,
     { connection: Connection; participants: RemoteParticipant[] }
   >();
 
   public constructor(private readonly logger?: Logger) {}
 
   public add(connection: Connection, participants: RemoteParticipant[]): void {
-    const key = this.getKey(connection.transport);
+    const key = keyFor(connection);
     const existing = this.map.get(key);
     if (!existing) {
       this.map.set(key, { connection, participants });
     } else {
-      // Transports are deduplicated by URL upstream, so this should never
+      // Transports are deduplicated by key upstream, so this should never
       // happen; if it does, members may be matched against the wrong room.
       this.logger?.warn(
         `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`,
@@ -66,26 +89,20 @@ export class ConnectionMap {
     }
   }
 
-  private getKey(transport: UnstableLivekitTransport): string {
-    // This is enough as a key because the ConnectionMap is already scoped by room.
-    // We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`.
-    return urlOf(transport);
-  }
-
   public getConnections(): Connection[] {
     return Array.from(this.map.values()).map(({ connection }) => connection);
   }
 
   public getConnectionForTransport(
-    transport: UnstableLivekitTransport,
+    transport: TransportLocator,
   ): Connection | null {
-    return this.map.get(this.getKey(transport))?.connection ?? null;
+    return this.map.get(keyFor(transport))?.connection ?? null;
   }
 
   public getParticipantsForTransport(
-    transport: UnstableLivekitTransport,
+    transport: TransportLocator,
   ): RemoteParticipant[] {
-    const existing = this.map.get(this.getKey(transport));
+    const existing = this.map.get(keyFor(transport));
     if (existing) {
       return existing.participants;
     }
@@ -96,8 +113,8 @@ export class ConnectionMap {
 interface Props {
   scope: ObservableScope;
   connectionFactory: ConnectionFactory;
-  localTransport: UnstableLivekitTransport;
+  localTransport: TransportLocator;
-  remoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
+  remoteTransports$: Behavior<Epoch<TransportLocator[]>>;
 
   logger: Logger;
   ownMembershipIdentity: CallMembershipIdentityParts;
@@ -128,7 +145,7 @@ export function createConnectionMap$({
    * All transports currently represented in the connection map. This list does
    * not include duplicate transports.
    */
-  const localAndRemoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>> =
+  const localAndRemoteTransports$: Behavior<Epoch<TransportLocator[]>> =
     scope.behavior(
       remoteTransports$.pipe(
         mapEpoch((remoteTransports) =>
@@ -140,6 +157,8 @@ export function createConnectionMap$({
       ),
     );
 
+  const localTransportKey = keyFor(localTransport);
+
   /**
    * Connections for each transport in use by one or more session members.
    */
@@ -149,17 +168,25 @@ export function createConnectionMap$({
         "ConnectionMap connections$",
         function* (transports) {
           for (const transport of transports) {
-            const role = areLivekitTransportsEqual(transport, localTransport)
+            const role =
-              ? ("publisher" as const)
+              keyFor(transport) === localTransportKey
-              : ("subscriber" as const);
+                ? ("publisher" as const)
-            yield { keys: [role, urlOf(transport)], data: transport };
+                : ("subscriber" as const);
+            yield {
+              // In order to maintain a static set of map keys, we must
+              // temporarily serialize `transport` into a stable key format
+              keys: [role, keyFor(transport)],
+              data: undefined,
+            };
           }
         },
-        (scope, transport$, role, _url) => {
+        (scope, _data$, role, transportKey) => {
+          const { transport, serverName } = fromKey(transportKey);
           const connection = connectionFactory.createConnection(
             scope,
             role,
-            transport$.value,
+            transport,
+            serverName,
             ownMembershipIdentity,
             logger,
           );

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

@@ -91,6 +91,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
         testScope,
         "subscriber",
         exampleTransport,
+        "example.org",
         ownMemberMock,
         logger,
       );
@@ -141,6 +142,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
         testScope,
         "subscriber",
         exampleTransport,
+        "example.org",
         ownMemberMock,
         logger,
       );
@@ -178,6 +180,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
       testScope,
       "subscriber",
       exampleTransport,
+      "example.org",
       ownMemberMock,
       logger,
     );
@@ -217,6 +220,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
       testScope,
       "subscriber",
       exampleTransport,
+      "example.org",
       ownMemberMock,
       logger,
     );
@@ -257,6 +261,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
       testScope,
       "subscriber",
       exampleTransport,
+      "example.org",
       ownMemberMock,
       logger,
     );

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

@@ -17,12 +17,7 @@ import {
   type RemoteMatrixLivekitMember,
   createRemoteMatrixLivekitMembers$,
 } from "./MatrixLivekitMembers.ts";
-import {
+import { Epoch, ObservableScope, trackEpoch } from "../../ObservableScope.ts";
-  Epoch,
-  mapEpoch,
-  ObservableScope,
-  trackEpoch,
-} from "../../ObservableScope.ts";
 import { ConnectionMap } from "./ConnectionMap.ts";
 import {
   flushPromises,
@@ -32,6 +27,8 @@ import {
 import { type Connection } from "./Connection.ts";
 import { constant } from "../../Behavior.ts";
 import { localRtcMember } from "../../../utils/test-fixtures.ts";
+import { type TransportLocator } from "../../../livekit/auth";
+import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
 
 let testScope: ObservableScope;
 
@@ -112,24 +109,14 @@ function createEpochedMemberships$(m$: Observable<CallMembership[]>): {
   memberships$: Observable<Epoch<CallMembership[]>>;
   membershipsWithTransport$: Observable<
     Epoch<
-      { membership: CallMembership; transport?: UnstableLivekitTransport }[]
+      { membership: CallMembership; transport: TransportLocator | undefined }[]
     >
   >;
 } {
-  const memberships$ = m$.pipe(trackEpoch());
+  const memberships$ = testScope.behavior(m$.pipe(trackEpoch()));
-  const membershipsWithTransport$ = memberships$.pipe(
+  const { membershipsWithTransport$ } = membershipsAndTransports$(
-    mapEpoch((members) => {
+    testScope,
-      return members.map((m) => {
+    memberships$,
-        const tr = m.getTransport();
-        return {
-          membership: m,
-          transport:
-            tr?.type === "livekit"
-              ? (tr as UnstableLivekitTransport)
-              : undefined,
-        };
-      });
-    }),
   );
   return {
     memberships$,
@@ -149,6 +136,7 @@ test("should signal participant on a connection that is publishing", async () =>
 
   const connection = {
     transport: bobMembership.getTransport(),
+    serverName: "example.org",
   } as unknown as Connection;
   const dataWithPublisher = new ConnectionMap();
   dataWithPublisher.add(connection, [
@@ -189,6 +177,7 @@ test("should signal participant on a connection that is not publishing", async (
 
   const connection = {
     transport: bobMembership.getTransport(),
+    serverName: "example.org",
   } as unknown as Connection;
   const dataWithPublisher = new ConnectionMap();
   dataWithPublisher.add(connection, []);
@@ -227,9 +216,11 @@ describe("Publication edge case", () => {
     );
     const connectionA = {
       transport: transportA,
+      serverName: "example.org",
     } as unknown as Connection;
     const connectionB = {
       transport: transportB,
+      serverName: "sample.com",
     } as unknown as Connection;
 
     connectionWithPublisher.add(connectionA, [
@@ -282,8 +273,14 @@ test("bob is publishing in the wrong connection", async () => {
     bobMembership.userId,
     bobMembership.deviceId,
   );
-  const connectionA = { transport: transportA } as unknown as Connection;
+  const connectionA = {
-  const connectionB = { transport: transportB } as unknown as Connection;
+    transport: transportA,
+    serverName: "example.org",
+  } as unknown as Connection;
+  const connectionB = {
+    transport: transportB,
+    serverName: "sample.com",
+  } as unknown as Connection;
 
   // Bob is not publishing on A
   connectionWithPublisher.add(connectionA, []);

src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts

@@ -6,19 +6,16 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
-import {
+import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
-  type CallMembership,
-  type UnstableLivekitTransport,
-} from "matrix-js-sdk/lib/matrixrtc";
 import { combineLatest, filter, map } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
-import { deepCompare } from "matrix-js-sdk/lib/utils";
 
 import { type Behavior } from "../../Behavior";
 import { type ConnectionMap } from "./ConnectionMap";
 import { Epoch, type ObservableScope } from "../../ObservableScope";
 import { type Connection } from "./Connection";
 import { generateItemsWithEpoch } from "../../../utils/observable";
+import { type TransportLocator } from "../../../livekit/auth";
 
 interface LocalTaggedParticipant {
   type: "local";
@@ -62,7 +59,7 @@ interface Props {
   scope: ObservableScope;
   membershipsWithTransport$: Behavior<
     Epoch<
-      { membership: CallMembership; transport?: UnstableLivekitTransport }[]
+      { membership: CallMembership; transport: TransportLocator | undefined }[]
     >
   >;
   connectionMap$: Behavior<Epoch<ConnectionMap>>;
@@ -163,14 +160,3 @@ export function createRemoteMatrixLivekitMembers$({
 }
 
 // TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
-
-// TODO add this to the JS-SDK
-export function areLivekitTransportsEqual<T extends UnstableLivekitTransport>(
-  t1: T | null,
-  t2: T | null,
-): boolean {
-  if (t1 && t2) {
-    return deepCompare(t1, t2);
-  }
-  return !t1 && !t2;
-}

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

@@ -10,7 +10,6 @@ import { BehaviorSubject } from "rxjs";
 import { type Room as LivekitRoom } from "livekit-client";
 import EventEmitter from "events";
 import fetchMock from "fetch-mock";
-import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import { logger } from "matrix-js-sdk/lib/logger";
 
 import {
@@ -21,7 +20,7 @@ import {
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
 import { type ClientOpenIDParts } from "../../../livekit/auth";
 import {
-  exampleTransport,
+  exampleTransportLocator,
   mockMediaDevices,
   mockRtcMembership,
   ownMemberMock,
@@ -29,13 +28,13 @@ import {
 } from "../../../utils/test.ts";
 import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
 import {
-  areLivekitTransportsEqual,
   createRemoteMatrixLivekitMembers$,
   type RemoteMatrixLivekitMember,
 } from "./MatrixLivekitMembers.ts";
 import { createConnectionMap$ } from "./ConnectionMap.ts";
 import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
 import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
+import { deepCompare } from "matrix-js-sdk/lib/utils";
 
 // Test the integration of ConnectionMap and MatrixLivekitMembers
 
@@ -126,7 +125,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
     const connectionMap$ = createConnectionMap$({
       scope: testScope,
       connectionFactory: ecConnectionFactory,
-      localTransport: exampleTransport,
+      localTransport: exampleTransportLocator,
       remoteTransports$: membershipsAndTransports.transports$,
       logger: logger,
       ownMembershipIdentity: ownMemberMock,
@@ -150,10 +149,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
         });
         expectObservable(item.connection$).toBe("a", {
           a: expect.toSatisfy((co) =>
-            areLivekitTransportsEqual(
+            deepCompare(co.transport, bobMembership.transports[0]),
-              co.transport,
-              bobMembership.transports[0]! as UnstableLivekitTransport,
-            ),
           ),
         });
         expectObservable(item.participant.value$).toBe("a", {
@@ -185,15 +181,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
             a: null,
           });
           expectObservable(item.connection$).toBe("a", {
-            a: expect.toSatisfy((connection) => {
+            a: expect.toSatisfy((connection) =>
-              expect(
+              deepCompare(connection.transport, carlMembership.transports[0]),
-                areLivekitTransportsEqual(
+            ),
-                  connection.transport,
-                  carlMembership.transports[0]! as UnstableLivekitTransport,
-                ),
-              ).toBe(true);
-              return true;
-            }),
           });
         }
         return true;
@@ -215,15 +205,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
             a: daveMembership,
           });
           expectObservable(item.connection$).toBe("a", {
-            a: expect.toSatisfy((connection) => {
+            a: expect.toSatisfy((connection) =>
-              expect(
+              deepCompare(connection.transport, daveMembership.transports[0]),
-                areLivekitTransportsEqual(
+            ),
-                  connection.transport,
-                  daveMembership.transports[0]! as UnstableLivekitTransport,
-                ),
-              ).toBe(true);
-              return true;
-            }),
           });
           expectObservable(item.participant.value$).toBe("a", {
             a: null,

src/state/SessionBehaviors.ts

@@ -7,7 +7,6 @@ Please see LICENSE in the repository root for full details.
 
 import {
   type CallMembership,
-  type UnstableLivekitTransport,
   type MatrixRTCSession,
   MatrixRTCSessionEvent,
   isUnstableLivekitTransport,
@@ -21,6 +20,7 @@ import {
   type ObservableScope,
 } from "./ObservableScope";
 import { type Behavior } from "./Behavior";
+import { type TransportLocator } from "../livekit/auth";
 
 /**
  * Tracks the transports used by ourselves, plus all other MatrixRTC session
@@ -32,17 +32,14 @@ export const membershipsAndTransports$ = (
 ): {
   membershipsWithTransport$: Behavior<
     Epoch<
-      { membership: CallMembership; transport?: UnstableLivekitTransport }[]
+      { membership: CallMembership; transport: TransportLocator | undefined }[]
     >
   >;
-  transports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
+  transports$: Behavior<Epoch<TransportLocator[]>>;
 } => {
   const membershipsWithTransport$: Behavior<
     Epoch<
-      {
+      { membership: CallMembership; transport: TransportLocator | undefined }[]
-        membership: CallMembership;
-        transport: UnstableLivekitTransport | undefined;
-      }[]
     >
   > = scope.behavior(
     memberships$.pipe(
@@ -52,7 +49,10 @@ export const membershipsAndTransports$ = (
           return {
             membership,
             transport: isUnstableLivekitTransport(transport)
-              ? transport
+              ? {
+                  transport,
+                  serverName: membership.userId.replace(/^.*?:/, ""),
+                }
               : undefined,
           };
         });
@@ -60,12 +60,11 @@ export const membershipsAndTransports$ = (
     ),
   );
 
-  const transports$: Behavior<Epoch<UnstableLivekitTransport[]>> =
+  const transports$: Behavior<Epoch<TransportLocator[]>> = scope.behavior(
-    scope.behavior(
+    membershipsWithTransport$.pipe(
-      membershipsWithTransport$.pipe(
+      mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
-        mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
+    ),
-      ),
+  );
-    );
 
   return {
     membershipsWithTransport$,

src/utils/test-viewmodel.ts

@@ -69,7 +69,7 @@ export function getBasicRTCSession(
       getUserId: () => localRtcMember.userId,
       getDeviceId: () => localRtcMember.deviceId,
       getSyncState: () => SyncState.Syncing,
-      getDomain: () => null,
+      getDomain: () => "example.com",
       getAccessToken: () => "fake-token",
       sendEvent: vitest.fn().mockResolvedValue({ event_id: "$fake:event" }),
       redactEvent: vitest.fn().mockResolvedValue({ event_id: "$fake:event" }),

src/utils/test.ts

@@ -78,6 +78,7 @@ import {
   type RemoteScreenShareViewModel,
 } from "../state/media/RemoteScreenShareViewModel";
 import { Connection } from "../state/CallViewModel/remoteMembers/Connection";
+import { type TransportLocator } from "../livekit/auth";
 
 export function withFakeTimers(continuation: () => void): void {
   vi.useFakeTimers();
@@ -213,6 +214,11 @@ export const exampleTransport: UnstableLivekitTransport = {
   livekit_service_url: "https://lk.example.org",
 };
 
+export const exampleTransportLocator: TransportLocator = {
+  transport: exampleTransport,
+  serverName: "example.org",
+};
+
 export function mockRtcMembership(
   user: string | RoomMember,
   deviceId: string,