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,