From 24b6a632dde26b047c6e7878a8bcc331e4d72893 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 12 Oct 2026 10:22:33 +0200 Subject: [PATCH] Let the publishing Connection authenticate itself getLocalTransport no longer authenticates with the SFU up front and hands the resulting config to the connection. Instead the ConnectionManager tags the connection whose transport equals the local one as the publisher, and every Connection fetches its own SFU config for its role. The local member now only joins the MatrixRTC session once its connection is authenticated, so that the LiveKit room exists before remote members try to get a subscriber token for it. Subscribers authenticating with the OpenID flow lack permission to create the room and would otherwise fail without retrying, causing a cascade of leaves and rejoins at the start of a call. --- src/state/CallViewModel/CallViewModel.ts | 16 +- .../CallViewModel/CallViewModelTestUtils.ts | 10 +- .../localMember/LocalMember.test.ts | 35 ++-- .../CallViewModel/localMember/LocalMember.ts | 63 ++++--- .../localMember/LocalTransport.test.ts | 178 +++--------------- .../localMember/LocalTransport.ts | 126 +------------ .../remoteMembers/Connection.test.ts | 6 + .../CallViewModel/remoteMembers/Connection.ts | 33 ++-- .../remoteMembers/ConnectionFactory.ts | 8 +- .../remoteMembers/ConnectionManager.test.ts | 8 +- .../remoteMembers/ConnectionManager.ts | 57 +++--- .../remoteMembers/ECConnectionFactory.test.ts | 5 + src/utils/test.ts | 8 - 13 files changed, 149 insertions(+), 404 deletions(-) diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 13eb0f8ed..216a1e617 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -112,10 +112,7 @@ import { enterRTCSession, TransportState, } from "./localMember/LocalMember.ts"; -import { - getLocalTransport, - type LocalTransport, -} from "./localMember/LocalTransport.ts"; +import { getLocalTransport } from "./localMember/LocalTransport.ts"; import { createKeyRotationSuppressed$, createMemberships$, @@ -216,7 +213,7 @@ export interface CallViewModelOptions { */ windowSize$: Behavior<{ width: number; height: number }>; /** Optional value overriding the local transport, for testing purposes. */ - localTransport?: LocalTransport; + localTransport?: LivekitTransport; /** Optional value overriding the connection factory, for testing purposes. */ connectionFactory?: ConnectionFactory; /** The version & compatibility mode of MatrixRTC that we should use. */ @@ -565,14 +562,7 @@ export function createCallViewModel$( const localTransport$ = options.localTransport ? constant(options.localTransport) - : from( - getLocalTransport({ - ownMembershipIdentity, - client, - roomId: matrixRoom.roomId, - slotId: matrixRTCSession.slotId, - }), - ); + : from(getLocalTransport(client)); const connectionFactory = options.connectionFactory ?? diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index a38d12f3d..95920eee7 100644 --- a/src/state/CallViewModel/CallViewModelTestUtils.ts +++ b/src/state/CallViewModel/CallViewModelTestUtils.ts @@ -30,7 +30,6 @@ import { type CallViewModelOptions, } from "./CallViewModel"; import { - exampleSfuConfig, exampleTransport, mockConfig, MockConnection, @@ -209,24 +208,21 @@ export function withCallViewModel(mode: MatrixRTCMode) { livekitRoomFactory, connectionState$, windowSize$, - localTransport: { - transport: exampleTransport, - sfuConfig: exampleSfuConfig, - }, + localTransport: exampleTransport, connectionFactory: { createConnection( scope, + role, transport, ownMembershipIdentity, logger, - sfuConfig, ) { return new MockConnection( { scope, + role, transport, ownMembershipIdentity, - existingSFUConfig: sfuConfig, client: room.client, roomId: room.roomId, slotId: "m.call#ROOM", diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 2dd5dfdce..b1e309d58 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -65,7 +65,6 @@ import { import { type Publisher } from "./Publisher"; import { initializeWidget } from "../../../widget"; import { nullHostBridge } from "../../../HostBridge"; -import { type LocalTransport } from "./LocalTransport"; import * as livekitAuth from "../../../livekit/auth"; initializeWidget(); @@ -221,7 +220,10 @@ describe("LocalMembership", () => { processor: undefined, }), logger: logger, - createPublisherFactory: vi.fn(), + createPublisherFactory: vi.fn().mockReturnValue({ + destroy: vi.fn(), + createAndSetupTracks: vi.fn().mockResolvedValue(undefined), + }), joinMatrixRTC: async (): Promise => {}, homeserverConnected: { combined$: constant<[boolean, HomeserverDisconnectReason | null]>([ @@ -254,6 +256,7 @@ describe("LocalMembership", () => { afterEach(async () => { void (await fetchMock.flush()); fetchMock.reset(); + vi.clearAllMocks(); }); it("throws error on missing RTC config error", () => { @@ -322,20 +325,10 @@ describe("LocalMembership", () => { scope.end(); }); - const mockTransportConfig = { + const mockTransport = { livekit_service_url: "a", } as LivekitTransport; - const mockTransport = { - transport: mockTransportConfig, - sfuConfig: { - jwt: "foo", - livekitAlias: "bar", - livekitIdentity: "baz", - url: "bro", - }, - } as LocalTransport; - const connectionTransportAConnected = { livekitRoom: mockLivekitRoom({ localParticipant: { @@ -347,7 +340,7 @@ describe("LocalMembership", () => { state: "authenticated", livekitState: LivekitConnectionState.Connected, }), - transport: mockTransportConfig, + transport: mockTransport, } as Connection; const connectionTransportAConnecting = { ...connectionTransportAConnected, @@ -381,13 +374,13 @@ describe("LocalMembership", () => { if (delegationUrl !== null) fetchMock.post(delegationUrl, () => ({ status: 401, body: {} })); + const connectionManagerData = new ConnectionManagerData(); + connectionManagerData.add(connectionTransportAConnected, []); const localMembership = createLocalMembership$({ scope, ...defaultCreateLocalMemberValues, connectionManager: { - connectionManagerData$: constant( - new Epoch(new ConnectionManagerData()), - ), + connectionManagerData$: constant(new Epoch(connectionManagerData)), }, joinMatrixRTC, localTransport$: constant(mockTransport), @@ -399,7 +392,7 @@ describe("LocalMembership", () => { await flushPromises(); // Joins with timings appropriate for the level of delegation support expect(joinMatrixRTC).toHaveBeenCalledWith( - mockTransportConfig, + mockTransport, delayedLeaveTimings, ); @@ -491,7 +484,7 @@ describe("LocalMembership", () => { it("tracks livekit state correctly", async () => { const scope = new ObservableScope(); const connectionManagerData = new ConnectionManagerData(); - const localTransport$ = new Subject(); + const localTransport$ = new Subject(); const connectionManagerData$ = new BehaviorSubject( new Epoch(connectionManagerData), @@ -570,7 +563,7 @@ describe("LocalMembership", () => { }); ( - connectionManagerData2.getConnectionForTransport(mockTransportConfig)! + connectionManagerData2.getConnectionForTransport(mockTransport)! .state$ as BehaviorSubject ).next({ state: "authenticated", @@ -891,7 +884,7 @@ describe("LocalMembership", () => { state: "authenticated", livekitState: LivekitConnectionState.Connected, }), - transport: mockTransportConfig, + transport: mockTransport, livekitRoom: mockLivekitRoom({ localParticipant: mockLocalParticipant({ isScreenShareEnabled: false, diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 4aad67b08..2fb6e0229 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -78,7 +78,6 @@ import { type Connection, } from "../remoteMembers/Connection.ts"; import { type HomeserverConnected } from "./HomeserverConnected.ts"; -import { type LocalTransport } from "./LocalTransport.ts"; import { type ClientOpenIDParts, getSFUConfig } from "../../../livekit/auth"; export enum TransportState { @@ -148,7 +147,7 @@ interface Props { homeserverConnected: HomeserverConnected; roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; - localTransport$: Observable; + localTransport$: Observable; client: ClientOpenIDParts; matrixRTCSession: Pick< MatrixRTCSession, @@ -307,7 +306,7 @@ export const createLocalMembership$ = ({ // to whether delayed event delegation is supported const joinParams$ = scope.behavior( localTransport$.pipe( - switchMap(async ({ transport }) => { + switchMap(async (transport) => { const transportSupportsDelegation = checkDelegationSupport( transport.livekit_service_url + "/delegate_delayed_leave", `transport ${transport.livekit_service_url}`, @@ -329,7 +328,7 @@ export const createLocalMembership$ = ({ connectionManager.connectionManagerData$, localTransport$, ]).pipe( - map(([{ value: connectionData }, { transport }]) => + map(([{ value: connectionData }, transport]) => connectionData.getConnectionForTransport(transport), ), tap((connection) => { @@ -480,6 +479,10 @@ export const createLocalMembership$ = ({ ), ); + const authenticated$ = scope.behavior( + localConnectionState$.pipe(map(({ state }) => state === "authenticated")), + ); + const mediaState$: Behavior = scope.behavior( combineLatest([ localConnectionState$, @@ -664,34 +667,38 @@ export const createLocalMembership$ = ({ // Join and leave the session as needed scope.reconcile( - scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])), - async ([joinParams, shouldConnect]) => { - if (!joinParams) return; - // if shouldConnect=false we will do the disconnect as the cleanup from the previous reconcile iteration. - if (!shouldConnect) return; + scope.behavior( + combineLatest([joinParams$, joinAndPublishRequested$, authenticated$]), + ), + async ([joinParams, shouldConnect, authenticated]) => { const sessionConfig = Config.get().matrix_rtc_session; - try { - joinMatrixRTC( - joinParams.transport, - joinParams.delegationSupported - ? sessionConfig.delegated_delayed_leave - : sessionConfig.delayed_leave, - ); - } catch (error) { - logger.error("Error entering RTC session", error); - if (error instanceof Error) - setMatrixError(new MembershipManagerError(error)); - } - - return Promise.resolve(async (): Promise => { + // Only join once authenticated, to ensure the LiveKit room exists first. + // Otherwise, subscribers authenticating with the OpenID flow may fail to + // join (they lack permission to create the room) and never retry. + if (joinParams && shouldConnect && authenticated) { try { - // TODO Update matrixRTCSession to allow udpating the transport without leaving the session! - await matrixRTCSession.leave(1000); - } catch (e) { - logger.error("Error leaving RTC session", e); + joinMatrixRTC( + joinParams.transport, + joinParams.delegationSupported + ? sessionConfig.delegated_delayed_leave + : sessionConfig.delayed_leave, + ); + } catch (error) { + logger.error("Error entering RTC session", error); + if (error instanceof Error) + setMatrixError(new MembershipManagerError(error)); } - }); + + // Clean-up callback to leave the session + return Promise.resolve(async (): Promise => { + try { + await matrixRTCSession.leave(1000); + } catch (e) { + logger.error("Error leaving RTC session", e); + } + }); + } }, ); diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index 805142176..178b5bde4 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -16,125 +16,51 @@ import { } from "vitest"; import fetchMock from "fetch-mock"; -import { mockConfig, ownMemberMock } from "../../../utils/test"; +import { mockConfig } from "../../../utils/test"; import { getLocalTransport } from "./LocalTransport"; -import { - MatrixRTCTransportMissingError, - FailToGetOpenIdToken, -} from "../../../utils/errors"; -import * as livekitAuth from "../../../livekit/auth"; +import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { customLivekitUrl } from "../../../settings/settings"; -import { testJWTToken } from "../../../utils/test-fixtures"; +import { type MatrixClient } from "matrix-js-sdk"; describe("LocalTransport", () => { - const openIdResponse: livekitAuth.SFUConfig = { - url: "https://lk.example.org", - jwt: testJWTToken, - livekitAlias: "Akph4alDMhen", - livekitIdentity: "@lk_user:ABCDEF", - }; - beforeEach(() => vi.clearAllMocks()); it("throws if config is missing", async () => { await expect( getLocalTransport({ - roomId: "!room:example.org", - client: { - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: async () => Promise.resolve([]), - getDomain: () => "example.org", - // These won't be called in this error path but satisfy the type - getOpenIdToken: vi.fn(), - }, - ownMembershipIdentity: ownMemberMock, - slotId: "m.call#ROOM", + _unstable_getRTCTransports: async () => Promise.resolve([]), + getDomain: () => "example.org", }), ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org")); }); - it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => { - // Provide a valid config so makeTransportInternal resolves a transport - mockConfig({ - livekit: { livekit_service_url: "https://lk.example.org" }, - }); - vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue( - new FailToGetOpenIdToken(new Error("no openid")), - ); - - await expect( - getLocalTransport({ - roomId: "!example_room_id", - client: { - getDomain: () => "example.org", - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: async () => Promise.resolve([]), - getOpenIdToken: vi.fn(), - }, - ownMembershipIdentity: ownMemberMock, - slotId: "m.call#ROOM", - }), - ).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid"))); - }); - it("returns preferred transport", async () => { - // Use config so transport discovery succeeds, but delay OpenID JWT fetch + // Use config so transport discovery succeeds mockConfig({ livekit: { livekit_service_url: "https://lk.example.org" }, }); - vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue({ - url: "https://lk.example.org", - jwt: "jwt", - livekitAlias: "Akph4alDMhen", - livekitIdentity: ownMemberMock.userId + ":" + ownMemberMock.deviceId, - }); - expect( await getLocalTransport({ - roomId: "!room:example.org", - client: { - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: async () => Promise.resolve([]), - getDomain: () => "example.org", - getOpenIdToken: vi.fn(), - }, - ownMembershipIdentity: ownMemberMock, - slotId: "m.call#ROOM", + _unstable_getRTCTransports: async () => Promise.resolve([]), + getDomain: () => "example.org", }), ).toStrictEqual({ - transport: { - livekit_service_url: "https://lk.example.org", - type: "livekit", - }, - sfuConfig: { - jwt: "jwt", - livekitAlias: "Akph4alDMhen", - livekitIdentity: "@alice:example.org:DEVICE", - url: "https://lk.example.org", - }, + livekit_service_url: "https://lk.example.org", + type: "livekit", }); }); - type LocalTransportProps = Parameters[0]; - describe("transport configuration mechanisms", () => { - let localTransportOpts: LocalTransportProps & { - client: MockedObject; - }; + let client: MockedObject< + Pick + >; beforeEach(() => { mockConfig({}); customLivekitUrl.setValue(customLivekitUrl.defaultValue); - localTransportOpts = { - ownMembershipIdentity: ownMemberMock, - roomId: "!example_room_id", - slotId: "m.call#ROOM", - client: { - getDomain: vi.fn().mockReturnValue("example.org"), - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: vi.fn().mockResolvedValue([]), - getOpenIdToken: vi.fn(), - }, + client = { + getDomain: vi.fn().mockReturnValue("example.org"), + _unstable_getRTCTransports: vi.fn().mockResolvedValue([]), }; }); @@ -146,86 +72,38 @@ describe("LocalTransport", () => { mockConfig({ livekit: { livekit_service_url: "https://lk.example.org" }, }); - vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse); - expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ - transport: { - livekit_service_url: "https://lk.example.org", - type: "livekit", - }, - sfuConfig: { - jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", - livekitAlias: "Akph4alDMhen", - livekitIdentity: "@lk_user:ABCDEF", - url: "https://lk.example.org", - }, + expect(await getLocalTransport(client)).toStrictEqual({ + livekit_service_url: "https://lk.example.org", + type: "livekit", }); }); it("supports getting transport via user settings", async () => { customLivekitUrl.setValue("https://lk.example.org"); - vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse); - expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ - transport: { - livekit_service_url: "https://lk.example.org", - type: "livekit", - }, - sfuConfig: { - jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", - livekitAlias: "Akph4alDMhen", - livekitIdentity: "@lk_user:ABCDEF", - url: "https://lk.example.org", - }, + expect(await getLocalTransport(client)).toStrictEqual({ + livekit_service_url: "https://lk.example.org", + type: "livekit", }); }); it("supports getting transport via backend", async () => { - localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ + client._unstable_getRTCTransports.mockResolvedValue([ { type: "livekit", livekit_service_url: "https://lk.example.org" }, ]); - vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(openIdResponse); - expect(await getLocalTransport(localTransportOpts)).toStrictEqual({ - transport: { - livekit_service_url: "https://lk.example.org", - type: "livekit", - }, - sfuConfig: { - jwt: "e30=.eyJzdWIiOiJAbWU6ZXhhbXBsZS5vcmc6QUJDREVGIiwidmlkZW8iOnsicm9vbSI6IiFleGFtcGxlX3Jvb21faWQifX0=.e30=", - livekitAlias: "Akph4alDMhen", - livekitIdentity: "@lk_user:ABCDEF", - url: "https://lk.example.org", - }, + expect(await getLocalTransport(client)).toStrictEqual({ + livekit_service_url: "https://lk.example.org", + type: "livekit", }); }); - it("fails fast if the openID request fails for backend config", async () => { - localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([ - { type: "livekit", livekit_service_url: "https://lk.example.org" }, - ]); - vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue( - new FailToGetOpenIdToken(new Error("Test driven error")), - ); - - await expect(getLocalTransport(localTransportOpts)).rejects.toThrow( - expect.any(FailToGetOpenIdToken), - ); - }); - it("throws if no options are available", async () => { await expect( getLocalTransport({ - ownMembershipIdentity: ownMemberMock, - roomId: "!example_room_id", - slotId: "m.call#ROOM", - client: { - getDomain: () => "example.org", - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: async () => Promise.resolve([]), - // These won't be called in this error path but satisfy the type - getOpenIdToken: vi.fn(), - }, + getDomain: () => "example.org", + _unstable_getRTCTransports: async () => Promise.resolve([]), }), ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org")); }); diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index 09880c034..891f70ccf 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -7,63 +7,24 @@ Please see LICENSE in the repository root for full details. import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; import { type MatrixClient } from "matrix-js-sdk"; -import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger"; -import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; +import { logger as rootLogger } from "matrix-js-sdk/lib/logger"; import { Config } from "../../../config/Config.ts"; -import { - FailToGetOpenIdToken, - MatrixRTCTransportMissingError, - NoMatrix2AuthorizationService, -} from "../../../utils/errors.ts"; -import { - getSFUConfig, - type SFUConfig, - type ClientOpenIDParts, -} from "../../../livekit/auth"; +import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts"; import { customLivekitUrl } from "../../../settings/settings.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; -interface Props { - ownMembershipIdentity: CallMembershipIdentityParts; - client: Pick & - ClientOpenIDParts; - // Used by the jwt service to create the livekit room and compute the livekit alias. - roomId: string; - slotId: string; -} - -export interface LocalTransport { - transport: LivekitTransport; - sfuConfig: SFUConfig; -} - -export function isLocalTransport( - obj: LivekitTransport | LocalTransport, -): obj is LocalTransport { - return "transport" in obj && "sfuConfig" in obj; -} - /** - * Connects to the JWT service and determines the transport that the local member should use. - * - * @prop useOldJwtEndpoint Whether to set forceOldJwtEndpoint on the returned transport and to use the old JWT endpoint. - * This is used when the connection manager needs to know if it has to use the legacy endpoint which implies a string concatenated rtcBackendIdentity. - * (which is expected for non sticky event based rtc member events) - * @returns The transport to advertise in our MatrixRTC membership and publish media on. - * @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken + * Determines the transport to advertise in our MatrixRTC membership and publish + * media on. */ -export async function getLocalTransport({ - ownMembershipIdentity, - client, - roomId, - slotId, -}: Props): Promise { - const logger = rootLogger.getChild("[LocalTransport]"); +export async function getLocalTransport( + client: Pick, +): Promise { const discovery = new RtcTransportAutoDiscovery({ - client: client, + client, resolvedConfig: Config.get(), - logger: logger, + logger: rootLogger.getChild("[LocalTransport]"), }); const customUrl = customLivekitUrl.value$.value; @@ -75,72 +36,5 @@ export async function getLocalTransport({ if (transport === null) throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); - try { - return await doOpenIdAndJWTFromUrl( - transport, - ownMembershipIdentity, - roomId, - slotId, - client, - logger, - ); - } catch (e) { - logger.error( - `Failed to authenticate to transport ${transport.livekit_service_url}`, - e, - ); - throw mapAuthErrorToUserFriendlyError(e); - } -} - -/** - * Utility to ensure the user can authenticate with the SFU. - * We will call `getSFUConfig` once per transport here as it's our - * only mechanism of validation. This means we will also ask the - * homeserver for a OpenID token a few times. Since OpenID tokens are single - * use we don't want to risk any issues by re-using a token. - * - * @param transport The transport to authenticate with. - * @param membership The identity of the local member. - * @param roomId The room ID to use for the JWT. - * @param slotId The slot ID to use for the JWT. - * @param client The client to use for the OpenID token. - * - * @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService - */ -async function doOpenIdAndJWTFromUrl( - transport: LivekitTransport, - membership: CallMembershipIdentityParts, - roomId: string, - slotId: string, - client: Pick & ClientOpenIDParts, - logger: Logger, -): Promise { - const sfuConfig = await getSFUConfig({ - client, - membership, - serviceUrl: transport.livekit_service_url, - roomId, - slotId, - role: "publisher", - logger, - }); - return { - transport, - sfuConfig, - }; -} - -function mapAuthErrorToUserFriendlyError(e: unknown): Error { - if ( - e instanceof FailToGetOpenIdToken || - e instanceof NoMatrix2AuthorizationService - ) { - // rethrow as is - return e; - } - // Catch others and rethrow as FailToGetOpenIdToken that has user friendly message. - return new FailToGetOpenIdToken( - e instanceof Error ? e : new Error(String(e)), - ); + return transport; } diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 1c0e8ee20..21e79ef97 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -115,6 +115,7 @@ function setupRemoteConnection(): Connection { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -159,6 +160,7 @@ describe("Start connection states", () => { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -177,6 +179,7 @@ describe("Start connection states", () => { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -230,6 +233,7 @@ describe("Start connection states", () => { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -296,6 +300,7 @@ describe("Start connection states", () => { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, @@ -555,6 +560,7 @@ describe("remote track logging", () => { client, roomId: ROOM_ID, slotId: SLOT_ID, + role: "subscriber", transport: livekitFocus, scope: testScope, ownMembershipIdentity: ownMemberMock, diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index da8c34daa..35c2e2c75 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.ts @@ -45,13 +45,10 @@ import { } from "../../../livekit/auth"; export interface ConnectionOpts { - /** - * For the local transport we already do know the jwt token and url. We can reuse it. - * On top the local transport will send additional data to the jwt server to use delayed event delegation. - */ - existingSFUConfig?: SFUConfig; /** The identity parts to use on this connection */ ownMembershipIdentity: CallMembershipIdentityParts; + /** Whether we want to publish or only subscribe on this connection. */ + role: "publisher" | "subscriber"; /** The media transport to connect to. */ transport: LivekitTransport; /** The Matrix client to use for OpenID and SFU config requests. */ @@ -92,6 +89,11 @@ export class Connection { */ public readonly state$: Behavior = this._state$; + /** + * Whether we want to publish or only subscribe on this connection. + */ + public readonly role: "publisher" | "subscriber"; + /** * The media transport to connect to. */ @@ -130,7 +132,6 @@ export class Connection { private readonly slotId: string; private readonly logger: Logger; private readonly ownMembershipIdentity: CallMembershipIdentityParts; - private readonly existingSFUConfig?: SFUConfig; /** * Creates a new connection to a matrix RTC LiveKit backend. * @@ -140,19 +141,17 @@ export class Connection { */ public constructor(opts: ConnectionOpts, logger: Logger) { this.ownMembershipIdentity = opts.ownMembershipIdentity; - this.existingSFUConfig = opts.existingSFUConfig; this.roomId = opts.roomId; this.slotId = opts.slotId; this.logger = logger.getChild( `[Connection ${JSON.stringify(opts.transport)}]`, ); - this.logger.info( - `constructor: roomId: ${this.roomId} withSfuConfig?: ${opts.existingSFUConfig ? JSON.stringify(opts.existingSFUConfig) : "undefined"}`, - ); - const { transport, client, scope } = opts; + this.logger.debug(`constructor called`); + const { role, transport, client, scope } = opts; this.scope = scope; this.livekitRoom = opts.livekitRoomFactory(); + this.role = role; this.transport = transport; this.client = client; @@ -295,11 +294,7 @@ export class Connection { this.stopped = false; try { this._state$.next({ state: "authenticating" }); - // We should already have this information after creating the localTransport. - // only call getSFUConfig for connections where we do not have a token yet. (existingJwtTokenData === undefined) - const { url, jwt, livekitAlias } = - this.existingSFUConfig ?? - (await this.getSFUConfigForRemoteConnection()); + const { url, jwt, livekitAlias } = await this.getSFUConfig(); this.logger.debug( "Starting Connection - jwt: ", jwt, @@ -381,16 +376,14 @@ export class Connection { } } - protected async getSFUConfigForRemoteConnection(): Promise { - // This will only be called for sfu's where we do not publish ourselves. - // For the local connection we will use the existingJwtTokenData + protected async getSFUConfig(): Promise { return await getSFUConfig({ client: this.client, membership: this.ownMembershipIdentity, + role: this.role, serviceUrl: this.transport.livekit_service_url, roomId: this.roomId, slotId: this.slotId, - role: "subscriber", logger: this.logger, }); } diff --git a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts index 4cb1371e1..b062e8d20 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts @@ -35,16 +35,16 @@ import { noiseSuppressionSetting, autoGainControlSetting, } from "../../../settings/settings.ts"; -import { type ClientOpenIDParts, type SFUConfig } from "../../../livekit/auth"; +import { type ClientOpenIDParts } from "../../../livekit/auth"; // TODO evaluate if this should be done like the Publisher Factory export interface ConnectionFactory { createConnection( scope: ObservableScope, + role: "publisher" | "subscriber", transport: LivekitTransport, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, - sfuConfig?: SFUConfig, ): Connection; } @@ -102,16 +102,16 @@ export class ECConnectionFactory implements ConnectionFactory { */ public createConnection( scope: ObservableScope, + role: "publisher" | "subscriber", transport: LivekitTransport, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, - sfuConfig?: SFUConfig, ): Connection { return new Connection( { - existingSFUConfig: sfuConfig, roomId: this.roomId, slotId: this.slotId, + role, transport, client: this.client, scope: scope, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts index e7d91c633..89772fcf4 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts @@ -47,7 +47,7 @@ beforeEach(() => { vi.mocked(fakeConnectionFactory).createConnection = vi .fn() .mockImplementation( - (scope: ObservableScope, transport: LivekitTransport) => { + (scope: ObservableScope, _role: unknown, transport: LivekitTransport) => { const mockConnection = { transport, remoteParticipants$: new BehaviorSubject([]), @@ -245,7 +245,11 @@ describe("connectionManagerData$ stream", () => { vi.mocked(fakeConnectionFactory).createConnection = vi .fn() .mockImplementation( - (scope: ObservableScope, transport: LivekitTransport) => { + ( + scope: ObservableScope, + _role: unknown, + transport: LivekitTransport, + ) => { const fakeRemoteParticipants$ = new BehaviorSubject< RemoteParticipant[] >([]); diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts index 2897cbe0e..3aa53d449 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts @@ -27,11 +27,6 @@ import { Epoch, type ObservableScope } from "../../ObservableScope.ts"; import { generateItemsWithEpoch } from "../../../utils/observable.ts"; import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; -import { - isLocalTransport, - type LocalTransport, -} from "../localMember/LocalTransport.ts"; -import { type SFUConfig } from "../../../livekit/auth"; export class ConnectionManagerData { private readonly store: Map< @@ -87,7 +82,7 @@ export class ConnectionManagerData { interface Props { scope: ObservableScope; connectionFactory: ConnectionFactory; - localTransport$: Observable; + localTransport$: Observable; remoteTransports$: Behavior>; logger: Logger; @@ -147,17 +142,17 @@ export function createConnectionManager$({ * externally this is modified via `registerTransports()`. */ const localAndRemoteTransports$: Behavior< - Epoch<(LivekitTransport | LocalTransport)[]> + Epoch<{ transport: LivekitTransport; role: "publisher" | "subscriber" }[]> > = scope.behavior( combineLatest([localTransportAsArray$, remoteTransports$]).pipe( - // Combine local and remote transports into one transport array - // and set the forceOldJwtEndpoint property on the local transport + // Combine local and remote transports into one transport array, tagged + // with the role we take on each of them map(([localTransportAsArray, remoteTransports]) => { const dedupedRemote = removeDuplicateTransports(remoteTransports.value); const remoteWithoutLocal = dedupedRemote.filter( (transport) => !localTransportAsArray.find((l) => - areLivekitTransportsEqual(l.transport, transport), + areLivekitTransportsEqual(l, transport), ), ); logger.debug( @@ -167,7 +162,16 @@ export function createConnectionManager$({ localTransportAsArray, ); return new Epoch( - [...localTransportAsArray, ...remoteWithoutLocal], + [ + ...localTransportAsArray.map((transport) => ({ + transport, + role: "publisher" as const, + })), + ...remoteWithoutLocal.map((transport) => ({ + transport, + role: "subscriber" as const, + })), + ], remoteTransports.epoch, ); }), @@ -182,40 +186,23 @@ export function createConnectionManager$({ generateItemsWithEpoch( "ConnectionManager connections$", function* (transports) { - for (const transport of transports) { - if (isLocalTransport(transport)) { - // This is the local transport; only the `LocalTransport` - // interface has a `sfuConfig` field. - yield { - keys: [ - transport.transport.livekit_service_url, - transport.sfuConfig, - ], - data: undefined, - }; - } else { - yield { - keys: [ - transport.livekit_service_url, - undefined as SFUConfig | undefined, - ], - data: undefined, - }; - } + for (const { transport, role } of transports) { + yield { + keys: [role, transport.livekit_service_url], + data: undefined, + }; } }, - (scope, _data$, serviceUrl, sfuConfig) => { + (scope, _data$, role, serviceUrl) => { const connection = connectionFactory.createConnection( scope, + role, { type: "livekit", livekit_service_url: serviceUrl, }, ownMembershipIdentity, logger, - // TODO: This whole optional SFUConfig parameter is not particularly elegant. - // I would like it if connections always fetched the SFUConfig by themselves. - sfuConfig, ); // Start the connection immediately // Use connection state to track connection progress diff --git a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts index d954c2e39..1fdc7e8a3 100644 --- a/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts +++ b/src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts @@ -89,6 +89,7 @@ describe("ECConnectionFactory - Audio inputs options", () => { ); ecConnectionFactory.createConnection( testScope, + "subscriber", exampleTransport, ownMemberMock, logger, @@ -138,6 +139,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => { ); ecConnectionFactory.createConnection( testScope, + "subscriber", exampleTransport, ownMemberMock, logger, @@ -174,6 +176,7 @@ describe("ECConnectionFactory - Camera quality settings", () => { ); ecConnectionFactory.createConnection( testScope, + "subscriber", exampleTransport, ownMemberMock, logger, @@ -212,6 +215,7 @@ describe("ECConnectionFactory - Camera quality settings", () => { ); ecConnectionFactory.createConnection( testScope, + "subscriber", exampleTransport, ownMemberMock, logger, @@ -251,6 +255,7 @@ describe("ECConnectionFactory - Camera quality settings", () => { ); ecConnectionFactory.createConnection( testScope, + "subscriber", exampleTransport, ownMemberMock, logger, diff --git a/src/utils/test.ts b/src/utils/test.ts index 9f7f14b80..6949c2171 100644 --- a/src/utils/test.ts +++ b/src/utils/test.ts @@ -78,7 +78,6 @@ import { type RemoteScreenShareViewModel, } from "../state/media/RemoteScreenShareViewModel"; import { Connection } from "../state/CallViewModel/remoteMembers/Connection"; -import { type SFUConfig } from "../livekit/auth"; export function withFakeTimers(continuation: () => void): void { vi.useFakeTimers(); @@ -214,13 +213,6 @@ export const exampleTransport: LivekitTransport = { livekit_service_url: "https://lk.example.org", }; -export const exampleSfuConfig: SFUConfig = { - jwt: "foo", - livekitAlias: "bar", - livekitIdentity: "baz", - url: "bro", -}; - export function mockRtcMembership( user: string | RoomMember, deviceId: string,