From 033f79975a93b0802aeec099057d11d7fd04c0bb Mon Sep 17 00:00:00 2001 From: Robin Date: Thu, 8 Oct 2026 23:56:52 +0200 Subject: [PATCH] Account for server name when identifying transports --- src/livekit/auth/csApi.ts | 6 + src/livekit/auth/index.ts | 6 + src/livekit/auth/types.ts | 11 ++ src/settings/DeveloperSettingsTab.tsx | 1 + src/state/CallViewModel/CallViewModel.ts | 3 +- .../CallViewModel/CallViewModelTestUtils.ts | 6 +- .../localMember/LocalMember.test.ts | 29 ++-- .../CallViewModel/localMember/LocalMember.ts | 5 +- .../localMember/LocalTransport.ts | 12 +- .../remoteMembers/Connection.test.ts | 28 ++-- .../CallViewModel/remoteMembers/Connection.ts | 11 +- .../remoteMembers/ConnectionFactory.ts | 4 + .../remoteMembers/ConnectionManager.test.ts | 50 ++++-- .../remoteMembers/ConnectionManager.ts | 155 ++++++++++++++---- .../remoteMembers/ECConnectionFactory.test.ts | 5 + .../MatrixLivekitMembers.test.ts | 29 ++-- .../remoteMembers/MatrixLivekitMembers.ts | 33 +--- .../remoteMembers/integration.test.ts | 10 +- src/state/SessionBehaviors.ts | 22 ++- src/utils/test.ts | 6 + 20 files changed, 289 insertions(+), 143 deletions(-) diff --git a/src/livekit/auth/csApi.ts b/src/livekit/auth/csApi.ts index 99962c4ad..63b9d0cb9 100644 --- a/src/livekit/auth/csApi.ts +++ b/src/livekit/auth/csApi.ts @@ -29,6 +29,10 @@ export interface GetSFUConfigParams { * The WebSocket URL of the SFU for which we wish to get an access token. */ url: string; + /** + * The name of the homeserver to which the SFU belongs. + */ + serverName: string; /** * The ID of the Matrix room in which the session takes place. */ @@ -47,12 +51,14 @@ export async function getSFUConfig({ client, membership, url, + serverName, roomId, slotId, }: GetSFUConfigParams): Promise { const res = await doNetworkOperationWithRetry(async () => client._unstable_getLivekitToken({ url, + server_name: serverName, room_id: roomId, slot_id: slotId, member_id: membership.memberId, diff --git a/src/livekit/auth/index.ts b/src/livekit/auth/index.ts index 01fd07b07..795b494e8 100644 --- a/src/livekit/auth/index.ts +++ b/src/livekit/auth/index.ts @@ -50,6 +50,10 @@ export interface GetSFUConfigParams { * The transport for which we wish to get an access token. */ transport: UnstableLivekitTransport; + /** + * The name of the homeserver to which the {@link transport} belongs. + */ + serverName: string; /** * The ID of the Matrix room in which the session takes place. */ @@ -72,6 +76,7 @@ export async function getSFUConfig({ client, membership, transport, + serverName, roomId, slotId, role, @@ -83,6 +88,7 @@ export async function getSFUConfig({ client, membership, url: transport.url, + serverName, roomId, slotId, }); 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/settings/DeveloperSettingsTab.tsx b/src/settings/DeveloperSettingsTab.tsx index d0153cc61..3ef943ddf 100644 --- a/src/settings/DeveloperSettingsTab.tsx +++ b/src/settings/DeveloperSettingsTab.tsx @@ -526,6 +526,7 @@ export const DeveloperSettingsTab: FC = ({ client, membership: { userId, deviceId, memberId: "" }, transport, + serverName: client.getDomain()!, roomId, slotId: "org.example.fake-slot#room", role: "publisher", diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 709420236..811fa6e9e 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -163,6 +163,7 @@ import { type RingingMediaViewModel, } from "../media/RingingMediaViewModel.ts"; import { type GridTileViewModel } from "../TileViewModel.ts"; +import { type TransportLocator } from "../../livekit/auth"; //TODO // Larger rename @@ -215,7 +216,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. */ diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index 16964ac85..1c1be0127 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 1a11b60e7..c34b998cf 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -305,7 +305,7 @@ describe("LocalMembership", () => { leave: vi.fn(), }, connectionManager: mockConnectionManager, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); const expextedLog = "'not connected yet' while updating the call intent (this is expected on startup)"; @@ -321,7 +321,12 @@ describe("LocalMembership", () => { type: "livekit", url: "https://sfu.example.org", livekit_service_url: "https://jwt.example.org", - } as UnstableLivekitTransport; + }; + + const mockTransportLocator: livekitAuth.TransportLocator = { + transport: mockTransport, + serverName: "example.org", + }; const connectionTransportAConnected = { livekitRoom: mockLivekitRoom({ @@ -375,7 +380,7 @@ describe("LocalMembership", () => { ), }, joinMatrixRTC, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), delayId$, }); @@ -452,7 +457,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); await flushPromises(); expect(publisherFactory).toHaveBeenCalledOnce(); @@ -474,7 +479,7 @@ describe("LocalMembership", () => { it("tracks livekit state correctly", async () => { const scope = new ObservableScope(); const connectionManagerData = new ConnectionManagerData(); - const localTransport$ = new Subject(); + const localTransport$ = new Subject(); const connectionManagerData$ = new BehaviorSubject( new Epoch(connectionManagerData), @@ -522,7 +527,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, @@ -547,7 +552,7 @@ describe("LocalMembership", () => { }); ( - connectionManagerData2.getConnectionForTransport(mockTransport)! + connectionManagerData2.getConnectionForTransport(mockTransportLocator)! .state$ as BehaviorSubject ).next(ConnectionState.LivekitConnected); expect(localMembership.localMemberState$.value).toStrictEqual({ @@ -650,7 +655,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); await flushPromises(); @@ -687,7 +692,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); await flushPromises(); @@ -735,7 +740,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); await flushPromises(); @@ -776,7 +781,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); await flushPromises(); @@ -842,7 +847,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: constant(mockTransport), + localTransport$: constant(mockTransportLocator), }); return { scope, localMembership }; }; diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 5b4fdc94f..0f30d846f 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -82,6 +82,7 @@ import { type ClientDelegationParts, delegateDelayedLeave, type ClientOpenIDParts, + type TransportLocator, } from "../../../livekit/auth"; export enum TransportState { @@ -151,7 +152,7 @@ interface Props { homeserverConnected: HomeserverConnected; roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; - localTransport$: Observable; + localTransport$: Observable; client: ClientDelegationParts & ClientOpenIDParts; matrixRTCSession: Pick< MatrixRTCSession, @@ -310,7 +311,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.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index adfc07efb..82a930721 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -5,7 +5,6 @@ SPDX-License-IdFentifier: 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"; import { type MatrixClient } from "matrix-js-sdk"; import { logger as rootLogger } from "matrix-js-sdk/lib/logger"; @@ -13,6 +12,7 @@ import { Config } from "../../../config/Config.ts"; import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts"; import { customTransport as customTransportSetting } 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 +20,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(), @@ -32,11 +35,10 @@ export async function getLocalTransport( customTransportSetting.value$.value ?? (await discovery.discoverPreferredTransport()); - if (transport === null) - throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); + if (transport === null) throw new MatrixRTCTransportMissingError(serverName); // TODO: Since this module no longer maps auth errors to fatal user-facing // errors (as it doesn't even perform auth), that needs to happen somewhere // else. - return transport; + return { transport, serverName }; } diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 32b3b432e..088ab310d 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -121,11 +121,12 @@ function setupTest(): void { function setupRemoteConnection(): Connection { const opts: ConnectionOpts = { - client: client, + client, roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, @@ -169,11 +170,12 @@ describe("Start connection states", () => { setupTest(); const opts: ConnectionOpts = { - client: client, + client, roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, @@ -188,11 +190,12 @@ describe("Start connection states", () => { vi.useFakeTimers(); const opts: ConnectionOpts = { - client: client, + client, roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, @@ -240,11 +243,12 @@ describe("Start connection states", () => { vi.useFakeTimers(); const opts: ConnectionOpts = { - client: client, + client, roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, @@ -301,11 +305,12 @@ describe("Start connection states", () => { vi.useFakeTimers(); const opts: ConnectionOpts = { - client: client, + client, roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + serverName: "m.localhost", scope: testScope, ownMembershipIdentity: ownMemberMock, livekitRoomFactory: () => fakeLivekitRoom, @@ -545,7 +550,8 @@ describe("remote track logging", () => { roomId: ROOM_ID, slotId: SLOT_ID, role: "subscriber", - transport: transport, + transport, + 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 c3d067f62..84eaf3179 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: ClientGetTokenParts & ClientOpenIDParts; /** The ID of the Matrix room in which the session takes place. */ @@ -114,6 +116,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; @@ -162,12 +169,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( @@ -401,6 +409,7 @@ export class Connection { membership: this.ownMembershipIdentity, role: this.role, transport: this.transport, + serverName: this.serverName, roomId: this.roomId, slotId: this.slotId, logger: this.logger, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts index d4c6ed40d..e5240f096 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts @@ -46,6 +46,7 @@ export interface ConnectionFactory { scope: ObservableScope, role: "publisher" | "subscriber", transport: UnstableLivekitTransport, + serverName: string, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, ): Connection; @@ -98,6 +99,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. @@ -107,6 +109,7 @@ export class ECConnectionFactory implements ConnectionFactory { scope: ObservableScope, role: "publisher" | "subscriber", transport: UnstableLivekitTransport, + serverName: string, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, ): Connection { @@ -116,6 +119,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/ConnectionManager.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts index d0b4c82a1..f4e2a3d64 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts @@ -15,23 +15,30 @@ import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts"; import { createConnectionManager$, ConnectionManagerData, + areUnstableLivekitTransportsEqual, } from "./ConnectionManager.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { type Connection } from "./Connection.ts"; import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts"; -import { areUnstableLivekitTransportsEqual } 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", }; let fakeConnectionFactory: ConnectionFactory; @@ -94,13 +101,19 @@ describe("connections$ stream", () => { ).toHaveBeenCalledTimes(2); const conn1 = connections.find((c) => - areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1), + areUnstableLivekitTransportsEqual( + c.transport, + TRANSPORT_1.transport, + ), ); expect(conn1).toBeDefined(); expect(conn1!.start).toHaveBeenCalled(); const conn2 = connections.find((c) => - areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2), + areUnstableLivekitTransportsEqual( + c.transport, + TRANSPORT_2.transport, + ), ); expect(conn2).toBeDefined(); expect(conn2!.start).toHaveBeenCalled(); @@ -141,12 +154,18 @@ describe("connections$ stream", () => { ).toHaveBeenCalledTimes(2); const conn2 = connections.find((c) => - areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2), + areUnstableLivekitTransportsEqual( + c.transport, + TRANSPORT_2.transport, + ), ); expect(conn2).toBeDefined(); const conn1 = connections.find((c) => - areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1), + areUnstableLivekitTransportsEqual( + c.transport, + TRANSPORT_1.transport, + ), ); expect(conn1).toBeDefined(); expect(conn1!.start).toHaveBeenCalledOnce(); @@ -187,7 +206,10 @@ describe("connections$ stream", () => { expect(connections.length).toBe(1); // The second connection should have been stopped has it is no longer needed. const connection2 = allCreatedConnections.find((c) => - areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2), + areUnstableLivekitTransportsEqual( + c.transport, + TRANSPORT_2.transport, + ), ); expect(connection2).toBeDefined(); expect(connection2!.stop).toHaveBeenCalled(); @@ -279,7 +301,7 @@ describe("connectionManagerData$ stream", () => { // Setup the fake participants streams behavior // ============================== fakeRemoteParticipantsStreams.set( - keyForTransport(TRANSPORT_1), + keyForTransport(TRANSPORT_1.transport), behavior("oa-b", { o: [], a: [{ identity: "user1A" } as RemoteParticipant], @@ -291,7 +313,7 @@ describe("connectionManagerData$ stream", () => { ); fakeRemoteParticipantsStreams.set( - keyForTransport(TRANSPORT_2), + keyForTransport(TRANSPORT_2.transport), behavior("o-a", { o: [], a: [{ identity: "user2A" } as RemoteParticipant], diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts index 189338e95..0c8fa1037 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts @@ -20,8 +20,8 @@ import { type ObservableScope, } from "../../ObservableScope.ts"; import { generateItemsWithEpoch } from "../../../utils/observable.ts"; -import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; +import { type TransportLocator } from "../../../livekit/auth/types.ts"; export class ConnectionManagerData { private readonly store: Map< @@ -32,7 +32,7 @@ export class ConnectionManagerData { public constructor(private readonly logger?: Logger) {} public add(connection: Connection, participants: RemoteParticipant[]): void { - const key = this.getKey(connection.transport); + const key = this.getKey(connection); const existing = this.store.get(key); if (!existing) { this.store.set(key, { connection, participants }); @@ -46,7 +46,7 @@ export class ConnectionManagerData { } } - private getKey(transport: UnstableLivekitTransport): string { + private getKey({ transport, serverName }: TransportLocator): string { // This is enough as a key because the ConnectionManager 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 JSON.stringify([ @@ -54,6 +54,7 @@ export class ConnectionManagerData { "livekit_service_url" in transport ? transport.livekit_service_url : undefined, + serverName, ]); } @@ -62,13 +63,13 @@ export class ConnectionManagerData { } public getConnectionForTransport( - transport: UnstableLivekitTransport, + transport: TransportLocator, ): Connection | null { return this.store.get(this.getKey(transport))?.connection ?? null; } public getParticipantsForTransport( - transport: UnstableLivekitTransport, + transport: TransportLocator, ): RemoteParticipant[] { const key = this.getKey(transport); const existing = this.store.get(key); @@ -79,11 +80,44 @@ export class ConnectionManagerData { } } +/** + * Unpacks an {@link UnstableLivekitTransport} into its constituent `url` and + * `serviceUrl`, discarding the type information that at least one of them must + * be defined. + */ +function unpackTransport(transport: UnstableLivekitTransport): { + url: string | undefined; + serviceUrl: string | undefined; +} { + return { + url: "url" in transport ? transport.url : undefined, + serviceUrl: + "livekit_service_url" in transport + ? transport.livekit_service_url + : undefined, + }; +} + +/** + * Constructs an {@link UnstableLivekitTransport} from its constituent `url` and + * `serviceUrl`. This assumes that at least one of the two are defined. + */ +function repackTransport( + url: string | undefined, + serviceUrl: string | undefined, +): UnstableLivekitTransport { + return { + type: "livekit", + ...(url === undefined ? {} : { url }), + ...(serviceUrl === undefined ? {} : { livekit_service_url: serviceUrl }), + } as UnstableLivekitTransport; +} + interface Props { scope: ObservableScope; connectionFactory: ConnectionFactory; - localTransport: UnstableLivekitTransport; - remoteTransports$: Behavior>; + localTransport: TransportLocator; + remoteTransports$: Behavior>; logger: Logger; ownMembershipIdentity: CallMembershipIdentityParts; @@ -129,11 +163,11 @@ export function createConnectionManager$({ * All transports currently managed by the ConnectionManager. This list does * not include duplicate transports. */ - const localAndRemoteTransports$: Behavior> = + const localAndRemoteTransports$: Behavior> = scope.behavior( remoteTransports$.pipe( mapEpoch((remoteTransports) => - removeDuplicateTransports([localTransport, ...remoteTransports]), + removeDuplicateLocators([localTransport, ...remoteTransports]), ), tap((transports) => logger.debug("localAndRemoteTransports$ = ", transports), @@ -149,30 +183,32 @@ export function createConnectionManager$({ generateItemsWithEpoch( "ConnectionManager connections$", function* (transports) { - for (const transport of transports) { + for (const { transport, serverName } of transports) { const role = - transport === localTransport + serverName === localTransport.serverName && + areUnstableLivekitTransportsEqual( + transport, + localTransport.transport, + ) ? ("publisher" as const) : ("subscriber" as const); - const url = "url" in transport ? transport.url : undefined; - const serviceUrl = - "livekit_service_url" in transport - ? transport.livekit_service_url - : undefined; - yield { keys: [role, url, serviceUrl], data: undefined }; + // In order to maintain a consistent set of map keys, we must + // temporarily downgrade `transport` to a weaker type + const { url, serviceUrl } = unpackTransport(transport); + yield { + keys: [role, url, serviceUrl, serverName], + data: undefined, + }; } }, - (scope, _data$, role, url, serviceUrl) => { + (scope, _data$, role, url, serviceUrl, serverName) => { const connection = connectionFactory.createConnection( scope, role, - { - type: "livekit", - ...(url === undefined ? {} : { url }), - ...(serviceUrl === undefined - ? {} - : { livekit_service_url: serviceUrl }), - } as UnstableLivekitTransport, + // Pack the transport keys back into a stronger + // `UnstableLivekitTransport` type + repackTransport(url, serviceUrl), + serverName, ownMembershipIdentity, logger, ); @@ -230,12 +266,65 @@ export function createConnectionManager$({ return { connectionManagerData$ }; } -function removeDuplicateTransports( - transports: T[], -): T[] { - return transports.reduce((acc, transport) => { - if (!acc.some((t) => areUnstableLivekitTransportsEqual(t, transport))) - acc.push(transport); - return acc; - }, [] as T[]); +// TODO add this to the JS-SDK +export function areUnstableLivekitTransportsEqual< + T extends UnstableLivekitTransport, +>(t1: T | null, t2: T | null): boolean { + if (t1 && t2) { + if ("url" in t1 !== "url" in t2) return false; + if ("livekit_service_url" in t1 !== "livekit_service_url" in t2) + return false; + if ( + "url" in t1 && + t1.url !== (t2 as UnstableLivekitTransport & { url: string }).url + ) + return false; + if ( + "livekit_service_url" in t1 && + t1.livekit_service_url !== + (t2 as UnstableLivekitTransport & { livekit_service_url: string }) + .livekit_service_url + ) + return false; + return true; + } + return !t1 && !t2; +} + +function removeDuplicateLocators( + locators: TransportLocator[], +): TransportLocator[] { + // Set tracking which locators we have added to `deduped` so far + // url -> serviceUrl -> serverNames + const byUrl = new Map< + string | undefined, + Map> + >(); + const deduped: TransportLocator[] = []; + + for (const locator of locators) { + const { transport, serverName } = locator; + const url = "url" in transport ? transport.url : undefined; + const serviceUrl = + "livekit_service_url" in transport + ? transport.livekit_service_url + : undefined; + + let byServiceUrl = byUrl.get(url); + if (byServiceUrl === undefined) { + byServiceUrl = new Map(); + byUrl.set(url, byServiceUrl); + } + let serverNames = byServiceUrl.get(serviceUrl); + if (serverNames === undefined) { + serverNames = new Set(); + byServiceUrl.set(serviceUrl, serverNames); + } + if (!serverNames.has(serverName)) { + serverNames.add(serverName); + deduped.push(locator); + } + } + + return deduped; } diff --git a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts index c5eefa290..cdca44ae4 100644 --- a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts +++ b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts @@ -97,6 +97,7 @@ describe("ECConnectionFactory - Audio inputs options", () => { testScope, "subscriber", exampleTransport, + "example.org", ownMemberMock, logger, ); @@ -147,6 +148,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => { testScope, "subscriber", exampleTransport, + "example.org", ownMemberMock, logger, ); @@ -184,6 +186,7 @@ describe("ECConnectionFactory - Camera quality settings", () => { testScope, "subscriber", exampleTransport, + "example.org", ownMemberMock, logger, ); @@ -223,6 +226,7 @@ describe("ECConnectionFactory - Camera quality settings", () => { testScope, "subscriber", exampleTransport, + "example.org", ownMemberMock, logger, ); @@ -263,6 +267,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 b709d3ff5..9a44aa74d 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts @@ -18,12 +18,7 @@ import { type RemoteMatrixLivekitMember, createRemoteMatrixLivekitMembers$, } from "./MatrixLivekitMembers.ts"; -import { - Epoch, - mapEpoch, - ObservableScope, - trackEpoch, -} from "../../ObservableScope.ts"; +import { Epoch, ObservableScope, trackEpoch } from "../../ObservableScope.ts"; import { ConnectionManagerData } from "./ConnectionManager.ts"; import { flushPromises, @@ -33,6 +28,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; @@ -117,21 +114,15 @@ test("should signal participant not yet connected to livekit", async () => { function createEpochedMemberships$(m$: Observable): { memberships$: Observable>; membershipsWithTransport$: Observable< - Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]> + Epoch< + { 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$, diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts index a85cd8b39..55d1eff22 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts @@ -6,10 +6,7 @@ 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"; @@ -18,6 +15,7 @@ import { type IConnectionManager } from "./ConnectionManager"; 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"; @@ -61,7 +59,7 @@ interface Props { scope: ObservableScope; membershipsWithTransport$: Behavior< Epoch< - { membership: CallMembership; transport?: UnstableLivekitTransport }[] + { membership: CallMembership; transport: TransportLocator | undefined }[] > >; connectionManager: IConnectionManager; @@ -165,28 +163,3 @@ export function createRemoteMatrixLivekitMembers$({ } // TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$) - -// TODO add this to the JS-SDK -export function areUnstableLivekitTransportsEqual< - T extends UnstableLivekitTransport, ->(t1: T | null, t2: T | null): boolean { - if (t1 && t2) { - if ("url" in t1 !== "url" in t2) return false; - if ("livekit_service_url" in t1 !== "livekit_service_url" in t2) - return false; - if ( - "url" in t1 && - t1.url !== (t2 as UnstableLivekitTransport & { url: string }).url - ) - return false; - if ( - "livekit_service_url" in t1 && - t1.livekit_service_url !== - (t2 as UnstableLivekitTransport & { livekit_service_url: string }) - .livekit_service_url - ) - return false; - return true; - } - return !t1 && !t2; -} diff --git a/src/state/CallViewModel/remoteMembers/integration.test.ts b/src/state/CallViewModel/remoteMembers/integration.test.ts index 0ca02d145..5031c8698 100644 --- a/src/state/CallViewModel/remoteMembers/integration.test.ts +++ b/src/state/CallViewModel/remoteMembers/integration.test.ts @@ -24,7 +24,7 @@ import { type ClientOpenIDParts, } from "../../../livekit/auth"; import { - exampleTransport, + exampleTransportLocator, mockMediaDevices, mockRtcMembership, ownMemberMock, @@ -32,11 +32,13 @@ import { } from "../../../utils/test.ts"; import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx"; import { - areUnstableLivekitTransportsEqual, createRemoteMatrixLivekitMembers$, type RemoteMatrixLivekitMember, } from "./MatrixLivekitMembers.ts"; -import { createConnectionManager$ } from "./ConnectionManager.ts"; +import { + areUnstableLivekitTransportsEqual, + createConnectionManager$, +} from "./ConnectionManager.ts"; import { membershipsAndTransports$ } from "../../SessionBehaviors.ts"; import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts"; @@ -129,7 +131,7 @@ test("bob, carl, then bob joining no tracks yet", () => { const connectionManager = createConnectionManager$({ scope: testScope, connectionFactory: ecConnectionFactory, - localTransport: exampleTransport, + localTransport: exampleTransportLocator, remoteTransports$: membershipsAndTransports.transports$, logger: logger, ownMembershipIdentity: ownMemberMock, diff --git a/src/state/SessionBehaviors.ts b/src/state/SessionBehaviors.ts index e227bcb3b..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 @@ -31,16 +31,15 @@ export const membershipsAndTransports$ = ( memberships$: Behavior>, ): { membershipsWithTransport$: Behavior< - Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]> + Epoch< + { 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( @@ -49,14 +48,19 @@ export const membershipsAndTransports$ = ( const transport = membership.getTransport(); return { membership, - transport: isUnstableLivekitTransport(transport) ? transport : undefined, + transport: isUnstableLivekitTransport(transport) + ? { + transport, + serverName: membership.userId.replace(/^.*?:/, ""), + } + : undefined, }; }); }), ), ); - const transports$: Behavior> = scope.behavior( + const transports$: Behavior> = scope.behavior( membershipsWithTransport$.pipe( mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))), ), diff --git a/src/utils/test.ts b/src/utils/test.ts index 29441d2ba..85c919e27 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,