diff --git a/src/livekit/MatrixAudioRenderer.test.tsx b/src/livekit/MatrixAudioRenderer.test.tsx index 21183f866..ab252a0e7 100644 --- a/src/livekit/MatrixAudioRenderer.test.tsx +++ b/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( p.identity)} livekitRoom={livekitRoom} - transport={exampleTransport} + transport={exampleTransportLocator} /> , ); @@ -151,7 +151,7 @@ it("should not render or warn for the local participant", () => { , ); @@ -178,7 +178,7 @@ it("should warn only once per unexpected participant", () => { , ); diff --git a/src/livekit/MatrixAudioRenderer.tsx b/src/livekit/MatrixAudioRenderer.tsx index 7f5e8b8fa..4ca1d3bae 100644 --- a/src/livekit/MatrixAudioRenderer.tsx +++ b/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. diff --git a/src/livekit/auth/types.ts b/src/livekit/auth/types.ts index 81982d689..409d88a6b 100644 --- a/src/livekit/auth/types.ts +++ b/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. */ diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index d6791c995..169abb98c 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/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), }), diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index 95920eee7..809dc75a5 100644 --- a/src/state/CallViewModel/CallViewModelTestUtils.ts +++ b/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, diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 437cdd20e..f1c95d1cb 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/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(); + const localTransport$ = new Subject(); 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 ).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, diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 1b1e8aae4..14f853403 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/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; + localTransport$: Observable; 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( diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index 178b5bde4..0b3caa42e 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -46,8 +46,11 @@ describe("LocalTransport", () => { getDomain: () => "example.org", }), ).toStrictEqual({ - livekit_service_url: "https://lk.example.org", - type: "livekit", + transport: { + 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", - type: "livekit", + transport: { + 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", - type: "livekit", + transport: { + 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", - type: "livekit", + transport: { + livekit_service_url: "https://lk.example.org", + type: "livekit", + }, + serverName: "example.org", }); }); diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index f2572bf30..feb8aee6f 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/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, -): Promise { +): Promise { + 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) - throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); + if (transport === null) throw new MatrixRTCTransportMissingError(serverName); - return transport; + return { transport, serverName }; } diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 98a1aed87..f0fc6bfd5 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/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, diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index 575ddfad6..c0fe4fd9a 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/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( diff --git a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts index b2a4833d7..86cc01b51 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts +++ b/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, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts index dba1cb0be..90c0e4c67 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts +++ b/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 = { - type: "livekit", - livekit_service_url: "https://lk.example.org", +const TRANSPORT_1: TransportLocator = { + transport: { + type: "livekit", + livekit_service_url: "https://lk.example.org", + }, + serverName: "example.org", }; -const TRANSPORT_2: UnstableLivekitTransport = { - type: "livekit", - livekit_service_url: "https://lk.sample.com", +const TRANSPORT_2: TransportLocator = { + transport: { + 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) => - areLivekitTransportsEqual(c.transport, transport), - ); + const key = keyFor(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>; - 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 { 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], diff --git a/src/state/CallViewModel/remoteMembers/ConnectionMap.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.ts index 698d9ad97..4adc92b04 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionMap.ts +++ b/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 - * that, its SFU URL. + * Converts a {@link TransportLocator} to a key suitable for identifying it in a + * {@link Map} or {@link Set}. */ -function urlOf(transport: UnstableLivekitTransport): string { - return "livekit_service_url" in transport - ? transport.livekit_service_url - : transport.url; +export function keyFor({ + transport, + serverName, +}: 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( - transports: T[], -): T[] { - return transports.reduce((acc, transport) => { - if (!acc.some((t) => areLivekitTransportsEqual(t, transport))) - acc.push(transport); - return acc; - }, [] as T[]); +/** + * Inverse of {@link keyFor}: turns a key back into a {@link TransportLocator}. + */ +export function fromKey(key: TransportKey): TransportLocator { + const [url, serviceUrl, serverName] = JSON.parse(key); + return { + transport: { + type: "livekit", + ...(url === null ? {} : { url }), + ...(serviceUrl === null ? {} : { livekit_service_url: serviceUrl }), + } as UnstableLivekitTransport, + serverName, + }; +} + +function removeDuplicateTransports( + transports: TransportLocator[], +): TransportLocator[] { + const map = new Map(); + 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; - remoteTransports$: Behavior>; + localTransport: TransportLocator; + remoteTransports$: Behavior>; 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> = + const localAndRemoteTransports$: Behavior> = 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) - ? ("publisher" as const) - : ("subscriber" as const); - yield { keys: [role, urlOf(transport)], data: transport }; + const role = + keyFor(transport) === localTransportKey + ? ("publisher" as const) + : ("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, ); diff --git a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts index 1fdc7e8a3..5d9abd0fe 100644 --- a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts +++ b/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, ); diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts index af98b2342..85f21b478 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts @@ -17,12 +17,7 @@ import { type RemoteMatrixLivekitMember, createRemoteMatrixLivekitMembers$, } from "./MatrixLivekitMembers.ts"; -import { - Epoch, - mapEpoch, - ObservableScope, - trackEpoch, -} from "../../ObservableScope.ts"; +import { Epoch, 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): { memberships$: Observable>; membershipsWithTransport$: Observable< Epoch< - { membership: CallMembership; transport?: UnstableLivekitTransport }[] + { membership: CallMembership; transport: TransportLocator | undefined }[] > >; } { - const memberships$ = m$.pipe(trackEpoch()); - const membershipsWithTransport$ = memberships$.pipe( - mapEpoch((members) => { - return members.map((m) => { - const tr = m.getTransport(); - return { - membership: m, - transport: - tr?.type === "livekit" - ? (tr as UnstableLivekitTransport) - : undefined, - }; - }); - }), + const memberships$ = testScope.behavior(m$.pipe(trackEpoch())); + const { membershipsWithTransport$ } = membershipsAndTransports$( + testScope, + memberships$, ); 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 connectionB = { transport: transportB } as unknown as Connection; + const connectionA = { + 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, []); diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts index 46735bb2f..2f2aa8e0c 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts +++ b/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 { - type CallMembership, - type UnstableLivekitTransport, -} from "matrix-js-sdk/lib/matrixrtc"; +import { type CallMembership } 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>; @@ -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( - t1: T | null, - t2: T | null, -): boolean { - if (t1 && t2) { - return deepCompare(t1, t2); - } - return !t1 && !t2; -} diff --git a/src/state/CallViewModel/remoteMembers/integration.test.ts b/src/state/CallViewModel/remoteMembers/integration.test.ts index 1a965cc02..39737ae6a 100644 --- a/src/state/CallViewModel/remoteMembers/integration.test.ts +++ b/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( - co.transport, - bobMembership.transports[0]! as UnstableLivekitTransport, - ), + deepCompare(co.transport, bobMembership.transports[0]), ), }); 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) => { - expect( - areLivekitTransportsEqual( - connection.transport, - carlMembership.transports[0]! as UnstableLivekitTransport, - ), - ).toBe(true); - return true; - }), + a: expect.toSatisfy((connection) => + deepCompare(connection.transport, carlMembership.transports[0]), + ), }); } 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) => { - expect( - areLivekitTransportsEqual( - connection.transport, - daveMembership.transports[0]! as UnstableLivekitTransport, - ), - ).toBe(true); - return true; - }), + a: expect.toSatisfy((connection) => + deepCompare(connection.transport, daveMembership.transports[0]), + ), }); expectObservable(item.participant.value$).toBe("a", { a: null, diff --git a/src/state/SessionBehaviors.ts b/src/state/SessionBehaviors.ts index 435e77021..6620350cf 100644 --- a/src/state/SessionBehaviors.ts +++ b/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>; + transports$: Behavior>; } => { const membershipsWithTransport$: Behavior< Epoch< - { - membership: CallMembership; - transport: UnstableLivekitTransport | undefined; - }[] + { membership: CallMembership; transport: TransportLocator | 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> = - scope.behavior( - membershipsWithTransport$.pipe( - mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))), - ), - ); + const transports$: Behavior> = scope.behavior( + membershipsWithTransport$.pipe( + mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))), + ), + ); return { membershipsWithTransport$, diff --git a/src/utils/test-viewmodel.ts b/src/utils/test-viewmodel.ts index 9903b99f8..df4f21bb1 100644 --- a/src/utils/test-viewmodel.ts +++ b/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" }), diff --git a/src/utils/test.ts b/src/utils/test.ts index c3efc6b89..e40e56fca 100644 --- a/src/utils/test.ts +++ b/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,