From 6d304d48dca52d1225ba2f6a4ef4eeea3c9ba311 Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 6 Oct 2026 17:50:11 +0200 Subject: [PATCH] WIP --- src/state/CallViewModel/CallViewModel.ts | 105 ++++++--------- .../CallViewModel/localMember/LocalMember.ts | 7 +- .../localMember/LocalTransport.ts | 90 ++----------- .../localMember/RtcTransportAutoDiscovery.ts | 4 +- .../CallViewModel/remoteMembers/Connection.ts | 28 ++-- .../remoteMembers/ConnectionFactory.ts | 7 +- .../remoteMembers/ConnectionManager.ts | 122 ++++++------------ src/utils/observable.ts | 22 ++++ 8 files changed, 132 insertions(+), 253 deletions(-) diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 51175566f..9084beb7d 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -56,6 +56,7 @@ import { createToggle$, filterBehavior, generateItems, + mapScoped, pauseWhen, } from "../../utils/observable"; import { @@ -104,17 +105,14 @@ import { type SpotlightPortraitLayoutMedia, } from "../layout-types.ts"; import { ElementCallError, UnknownCallError } from "../../utils/errors.ts"; -import { type Epoch, type ObservableScope } from "../ObservableScope.ts"; +import { Epoch, type ObservableScope } from "../ObservableScope.ts"; import { createHomeserverConnected$ } from "./localMember/HomeserverConnected.ts"; import { createLocalMembership$, enterRTCSession, TransportState, } from "./localMember/LocalMember.ts"; -import { - getLocalTransport, - type LocalTransport, -} from "./localMember/LocalTransport.ts"; +import { getLocalTransport } from "./localMember/LocalTransport.ts"; import { createKeyRotationSuppressed$, createMemberships$, @@ -125,8 +123,9 @@ import { ECConnectionFactory, } from "./remoteMembers/ConnectionFactory.ts"; import { - type ConnectionManagerData, + ConnectionManagerData, createConnectionManager$, + type IConnectionManager, } from "./remoteMembers/ConnectionManager.ts"; import { createRemoteMatrixLivekitMembers$, @@ -215,7 +214,7 @@ export interface CallViewModelOptions { */ windowSize$: Behavior<{ width: number; height: number }>; /** Optional value overriding the local transport, for testing purposes. */ - localTransport?: LocalTransport; + localTransport?: UnstableLivekitTransport; /** Optional value overriding the connection factory, for testing purposes. */ connectionFactory?: ConnectionFactory; /** The version & compatibility mode of MatrixRTC that we should use. */ @@ -575,13 +574,7 @@ export function createCallViewModel$( const localTransport$ = options.localTransport ? constant(options.localTransport) - : from( - getLocalTransport({ - ownMembershipIdentity, - client, - roomId: matrixRoom.roomId, - }), - ); + : from(getLocalTransport(client)); const connectionFactory = options.connectionFactory ?? @@ -595,14 +588,24 @@ export function createCallViewModel$( options.livekitRoomFactory, ); - const connectionManager = createConnectionManager$({ - scope: scope, - connectionFactory: connectionFactory, - localTransport$, - remoteTransports$: membershipsAndTransports.transports$, - logger: logger, - ownMembershipIdentity, - }); + const connectionManager: IConnectionManager = { + connectionManagerData$: scope.behavior( + localTransport$.pipe( + mapScoped("connectionManager$", (scope, localTransport) => + createConnectionManager$({ + scope, + connectionFactory, + localTransport, + remoteTransports$: membershipsAndTransports.transports$, + logger, + ownMembershipIdentity, + }), + ), + switchMap(({ connectionManagerData$ }) => connectionManagerData$), + ), + new Epoch(new ConnectionManagerData()), + ), + }; const remoteMatrixLivekitMembers$: Behavior< Epoch @@ -737,49 +740,27 @@ export function createCallViewModel$( const allConnections$ = scope.behavior( connectionManager.connectionManagerData$.pipe(map((d) => d.value)), ); - const livekitRoomItems$ = scope.behavior( - remoteMatrixLivekitMembers$.pipe( - switchMap((members) => { - const a$ = combineLatest( - members.value.map((member) => - combineLatest([member.connection$, member.participant.value$]).pipe( - map(([connection, participant]) => { - // do not render audio for local participant - if (!connection || !participant || participant.isLocal) - return null; - const livekitRoom = connection.livekitRoom; - const url = connection.transport.livekit_service_url; - - return { - url, - livekitRoom, - participant: participant.identity, - }; - }), + const livekitRoomItems$ = scope.behavior( + allConnections$.pipe( + switchMap((connections) => + combineLatest( + connections.getConnections().map((connection) => + remoteMatrixLivekitMembers$.pipe( + switchMap((members) => + combineLatest(members.value.map((m) => m.participant.value$)), + ), + map((participants) => ({ + url: JSON.stringify(connection.transport), // TODO + livekitRoom: connection.livekitRoom, + participants: participants + .filter((p) => p !== null) + .map((p) => p.identity), + })), ), ), - ); - return a$; - }), - map((members) => - members.reduce((acc, curr) => { - if (!curr) return acc; - - const existing = acc.find((item) => item.url === curr.url); - if (existing) { - existing.participants.push(curr.participant); - } else { - acc.push({ - livekitRoom: curr.livekitRoom, - participants: [curr.participant], - url: curr.url, - }); - } - return acc; - }, []), + ), ), ), - [], ); const handsRaised$ = scope.behavior( @@ -837,7 +818,7 @@ export function createCallViewModel$( connection$.pipe(map((c) => c?.livekitRoom)), ), focusUrl$: scope.behavior( - connection$.pipe(map((c) => c?.transport.livekit_service_url)), + connection$.pipe(map((c) => JSON.stringify(c?.transport))), // TODO ), mediaDevices, pretendToBeDisconnected$: localMembership.reconnecting$, diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index b0f6ccc74..c4907d24b 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -78,7 +78,6 @@ import { type FailedToStartError, } from "../remoteMembers/Connection.ts"; import { type HomeserverConnected } from "./HomeserverConnected.ts"; -import { type LocalTransport } from "./LocalTransport.ts"; import { type ClientDelegationParts, delegateDelayedLeave, @@ -152,7 +151,7 @@ interface Props { homeserverConnected: HomeserverConnected; roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; - localTransport$: Observable; + localTransport$: Observable; client: ClientDelegationParts & ClientOpenIDParts; matrixRTCSession: Pick; /** Whether to hide the screen-sharing button. */ @@ -308,7 +307,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( @@ -331,7 +330,7 @@ export const createLocalMembership$ = ({ connectionManager.connectionManagerData$, localTransport$, ]).pipe( - map(([{ value: connectionData }, { transport }]) => + map(([{ value: connectionData }, transport]) => connectionData.getConnectionForTransport(transport), ), tap((connection) => { diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index 07e2e743f..028ab1703 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -8,62 +8,23 @@ 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"; -import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { Config } from "../../../config/Config.ts"; -import { - FailToGetOpenIdToken, - MatrixRTCTransportMissingError, - NoMatrix2AuthorizationService, -} from "../../../utils/errors.ts"; +import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts"; import { customLivekitUrl } from "../../../settings/settings.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; -import { - getSFUConfig, - type ClientGetTokenParts, - type ClientOpenIDParts, - type SFUConfig, -} from "../../../livekit/auth"; - -interface Props { - ownMembershipIdentity: CallMembershipIdentityParts; - client: Pick & - ClientGetTokenParts & - ClientOpenIDParts; - // Used by the jwt service to create the livekit room and compute the livekit alias. - roomId: string; -} - -export interface LocalTransport { - transport: UnstableLivekitTransport; - sfuConfig: SFUConfig; -} - -export function isLocalTransport( - obj: UnstableLivekitTransport | 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, -}: Props): Promise { - const logger = rootLogger.getChild("[LocalTransport]"); +export async function getLocalTransport( + client: Pick, +): Promise { const discovery = new RtcTransportAutoDiscovery({ client: client, resolvedConfig: Config.get(), - logger: logger, + logger: rootLogger.getChild("[LocalTransport]"), }); const customUrl = customLivekitUrl.value$.value; @@ -75,37 +36,8 @@ export async function getLocalTransport({ if (transport === null) throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); - try { - return { - transport, - sfuConfig: await getSFUConfig({ - client, - membership: ownMembershipIdentity, - transport, - roomId, - role: "publisher", - logger, - }), - }; - } catch (e) { - logger.error( - `Failed to authenticate to transport ${JSON.stringify(transport)}`, - e, - ); - throw mapAuthErrorToUserFriendlyError(e); - } -} - -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)), - ); + // 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; } diff --git a/src/state/CallViewModel/localMember/RtcTransportAutoDiscovery.ts b/src/state/CallViewModel/localMember/RtcTransportAutoDiscovery.ts index f61a1b20b..a5bedcd80 100644 --- a/src/state/CallViewModel/localMember/RtcTransportAutoDiscovery.ts +++ b/src/state/CallViewModel/localMember/RtcTransportAutoDiscovery.ts @@ -45,7 +45,7 @@ export class RtcTransportAutoDiscovery { const backendTransport = await this.tryBackendTransports(); if (backendTransport) { this.logger.info( - `Found backend transport: ${backendTransport.livekit_service_url}`, + `Found backend transport: ${JSON.stringify(backendTransport)}`, ); return backendTransport; } @@ -54,7 +54,7 @@ export class RtcTransportAutoDiscovery { const configTransport = this.tryConfigTransport(); if (configTransport) { this.logger.info( - `Found app config transport: ${configTransport.livekit_service_url}`, + `Found app config transport: ${JSON.stringify(configTransport)}`, ); return configTransport; } diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index 3329dfe5a..4c58560e8 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: UnstableLivekitTransport; /** The Matrix client to use for OpenID and SFU config requests. */ @@ -105,6 +102,8 @@ export class Connection { */ public readonly state$: Behavior = this._state$; + private readonly role: "publisher" | "subscriber"; + /** * The media transport to connect to. */ @@ -142,7 +141,6 @@ export class Connection { private readonly roomId: string; private readonly logger: Logger; private readonly ownMembershipIdentity: CallMembershipIdentityParts; - private readonly existingSFUConfig?: SFUConfig; /** * Creates a new connection to a matrix RTC LiveKit backend. * @@ -152,18 +150,16 @@ export class Connection { */ public constructor(opts: ConnectionOpts, logger: Logger) { this.ownMembershipIdentity = opts.ownMembershipIdentity; - this.existingSFUConfig = opts.existingSFUConfig; this.roomId = opts.roomId; 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; @@ -306,11 +302,7 @@ export class Connection { this.stopped = false; try { this._state$.next(ConnectionState.FetchingConfig); - // We should already have this information after creating the localTransport. - // only call getSFUConfigWithOpenID 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, @@ -394,15 +386,15 @@ export class Connection { } } - protected async getSFUConfigForRemoteConnection(): Promise { + protected async getSFUConfig(): Promise { // This will only be called for sfu's where we do not publish ourselves. // For the local connection we will use the existingJwtTokenData return await getSFUConfig({ client: this.client, membership: this.ownMembershipIdentity, + role: this.role, transport: this.transport, roomId: this.roomId, - role: "subscriber", logger: this.logger, }); } diff --git a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts index 935a83a9d..7da71a4dc 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionFactory.ts @@ -38,17 +38,16 @@ import { import { type ClientGetTokenParts, type ClientOpenIDParts, - type SFUConfig, } from "../../../livekit/auth"; // TODO evaluate if this should be done like the Publisher Factory export interface ConnectionFactory { createConnection( scope: ObservableScope, + role: "publisher" | "subscriber", transport: UnstableLivekitTransport, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, - sfuConfig?: SFUConfig, ): Connection; } @@ -104,15 +103,15 @@ export class ECConnectionFactory implements ConnectionFactory { */ public createConnection( scope: ObservableScope, + role: "publisher" | "subscriber", transport: UnstableLivekitTransport, ownMembershipIdentity: CallMembershipIdentityParts, logger: Logger, - sfuConfig?: SFUConfig, ): Connection { return new Connection( { - existingSFUConfig: sfuConfig, roomId: this.roomId, + role, transport, client: this.client, scope: scope, diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts index 959e792db..189338e95 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts @@ -7,31 +7,21 @@ Please see LICENSE in the repository root for full details. */ import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; -import { - combineLatest, - map, - type Observable, - of, - switchMap, - startWith, - catchError, - NEVER, -} from "rxjs"; +import { combineLatest, map, of, switchMap, tap } from "rxjs"; import { type Logger } from "matrix-js-sdk/lib/logger"; import { type RemoteParticipant } from "livekit-client"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type Behavior } from "../../Behavior.ts"; import { type Connection } from "./Connection.ts"; -import { Epoch, type ObservableScope } from "../../ObservableScope.ts"; +import { + Epoch, + mapEpoch, + type ObservableScope, +} from "../../ObservableScope.ts"; import { generateItemsWithEpoch } from "../../../utils/observable.ts"; import { areUnstableLivekitTransportsEqual } 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< @@ -92,7 +82,7 @@ export class ConnectionManagerData { interface Props { scope: ObservableScope; connectionFactory: ConnectionFactory; - localTransport$: Observable; + localTransport: UnstableLivekitTransport; remoteTransports$: Behavior>; logger: Logger; @@ -127,7 +117,7 @@ export interface IConnectionManager { export function createConnectionManager$({ scope, connectionFactory, - localTransport$, + localTransport, remoteTransports$, logger: parentLogger, ownMembershipIdentity, @@ -135,49 +125,21 @@ export function createConnectionManager$({ const logger = parentLogger.getChild("[ConnectionManager]"); // TODO logger: only construct one logger from the client and make it compatible via a EC specific sing - const localTransportAsArray$ = localTransport$.pipe( - // LocalMember already surfaces local transport errors properly in the UI, - // here we can just swallow them - catchError(() => NEVER), - map((transport) => [transport]), - startWith([]), - ); - /** - * All transports currently managed by the ConnectionManager. - * - * This list does not include duplicate transports. - * - * It is build based on the list of subscribed transports (`transportsSubscriptions$`). - * externally this is modified via `registerTransports()`. + * All transports currently managed by the ConnectionManager. This list does + * not include duplicate transports. */ - const localAndRemoteTransports$: Behavior< - Epoch<(UnstableLivekitTransport | LocalTransport)[]> - > = scope.behavior( - combineLatest([localTransportAsArray$, remoteTransports$]).pipe( - // Combine local and remote transports into one transport array - // and set the forceOldJwtEndpoint property on the local transport - map(([localTransportAsArray, remoteTransports]) => { - const dedupedRemote = removeDuplicateTransports(remoteTransports.value); - const remoteWithoutLocal = dedupedRemote.filter( - (transport) => - !localTransportAsArray.find((l) => - areUnstableLivekitTransportsEqual(l.transport, transport), - ), - ); - logger.debug( - "remoteWithoutLocal", - remoteWithoutLocal, - "localTransportAsArray", - localTransportAsArray, - ); - return new Epoch( - [...localTransportAsArray, ...remoteWithoutLocal], - remoteTransports.epoch, - ); - }), - ), - ); + const localAndRemoteTransports$: Behavior> = + scope.behavior( + remoteTransports$.pipe( + mapEpoch((remoteTransports) => + removeDuplicateTransports([localTransport, ...remoteTransports]), + ), + tap((transports) => + logger.debug("localAndRemoteTransports$ = ", transports), + ), + ), + ); /** * Connections for each transport in use by one or more session members. @@ -188,39 +150,31 @@ export function createConnectionManager$({ "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, - }; - } + const role = + transport === localTransport + ? ("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 }; } }, - (scope, _data$, serviceUrl, sfuConfig) => { + (scope, _data$, role, url, serviceUrl) => { const connection = connectionFactory.createConnection( scope, + role, { type: "livekit", - livekit_service_url: serviceUrl, - }, + ...(url === undefined ? {} : { url }), + ...(serviceUrl === undefined + ? {} + : { livekit_service_url: serviceUrl }), + } as UnstableLivekitTransport, 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/utils/observable.ts b/src/utils/observable.ts index ca51ad196..952811231 100644 --- a/src/utils/observable.ts +++ b/src/utils/observable.ts @@ -218,6 +218,28 @@ export function filterBehavior( ); } +/** + * Maps a changing input value to an item whose lifetime is tied to a certain + * computed key. The item may capture some dynamic data from the input. + */ +export function mapScoped( + name: string, + project: (scope: ObservableScope, a: A) => B, +): OperatorFunction { + return (a$) => + a$.pipe( + generateItemsInternal( + name, + function* (a) { + yield { keys: [a], data: undefined }; + }, + (scope, _data$, a) => project(scope, a), + (items) => items, + ), + map(([item]) => item), + ); +} + function generateItemsInternal< Input, Keys extends [unknown, ...unknown[]],