/* Copyright 2023, 2024, 2025 New Vector Ltd. SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ import { observeParticipantEvents } from "@livekit/components-core"; import { ConnectionState, type BaseKeyProvider, type E2EEOptions, ExternalE2EEKeyProvider, type Room as LivekitRoom, type LocalParticipant, ParticipantEvent, type RemoteParticipant, } from "livekit-client"; import E2EEWorker from "livekit-client/e2ee-worker?worker"; import { ClientEvent, type RoomMember, RoomStateEvent, SyncState, type Room as MatrixRoom, type EventTimelineSetHandlerMap, EventType, RoomEvent, } from "matrix-js-sdk"; import { BehaviorSubject, EMPTY, NEVER, type Observable, Subject, combineLatest, concat, concatMap, distinctUntilChanged, endWith, filter, from, fromEvent, ignoreElements, map, merge, of, pairwise, race, scan, skip, skipWhile, startWith, switchAll, switchMap, switchScan, take, takeUntil, throttleTime, timer, } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; import { type CallMembership, isLivekitFocus, type MatrixRTCSession, MatrixRTCSessionEvent, type MatrixRTCSessionEventHandlerMap, MembershipManagerEvent, Status, } from "matrix-js-sdk/lib/matrixrtc"; import { ViewModel } from "./ViewModel"; import { LocalUserMediaViewModel, type MediaViewModel, RemoteUserMediaViewModel, ScreenShareViewModel, type UserMediaViewModel, } from "./MediaViewModel"; import { accumulate, and$, finalizeValue, pauseWhen, } from "../utils/observable"; import { ObservableScope } from "./ObservableScope"; import { duplicateTiles, playReactionsSound, showReactions, showNonMemberTiles, } from "../settings/settings"; import { isFirefox } from "../Platform"; import { setPipEnabled$ } from "../controls"; import { type GridTileViewModel, type SpotlightTileViewModel, } from "./TileViewModel"; import { TileStore } from "./TileStore"; import { gridLikeLayout } from "./GridLikeLayout"; import { spotlightExpandedLayout } from "./SpotlightExpandedLayout"; import { oneOnOneLayout } from "./OneOnOneLayout"; import { pipLayout } from "./PipLayout"; import { type EncryptionSystem } from "../e2ee/sharedKeyManagement"; import { type RaisedHandInfo, type ReactionInfo, type ReactionOption, } from "../reactions"; import { observeSpeaker$ } from "./observeSpeaker"; import { shallowEquals } from "../utils/array"; import { calculateDisplayName, shouldDisambiguate } from "../utils/displayname"; import { type MediaDevices } from "./MediaDevices"; import { type Behavior } from "./Behavior"; import { enterRTCSession, getLivekitAlias, makeFocus, } from "../rtcSessionHelpers"; import { E2eeType } from "../e2ee/e2eeType"; import { MatrixKeyProvider } from "../e2ee/matrixKeyProvider"; import { type ECConnectionState } from "../livekit/useECConnectionState"; import { Connection, PublishConnection } from "./Connection"; import { type MuteStates } from "./MuteStates"; export interface CallViewModelOptions { encryptionSystem: EncryptionSystem; autoLeaveWhenOthersLeft?: boolean; /** * If the call is started in a way where we want it to behave like a telephone usecase * If we sent a notification event, we want the ui to show a ringing state */ waitForCallPickup?: boolean; } // Do not play any sounds if the participant count has exceeded this // number. export const MAX_PARTICIPANT_COUNT_FOR_SOUND = 8; export const THROTTLE_SOUND_EFFECT_MS = 500; // This is the number of participants that we think constitutes a "small" call // on mobile. No spotlight tile should be shown below this threshold. const smallMobileCallThreshold = 3; // How long the footer should be shown for when hovering over or interacting // with the interface const showFooterMs = 4000; export interface GridLayoutMedia { type: "grid"; spotlight?: MediaViewModel[]; grid: UserMediaViewModel[]; } export interface SpotlightLandscapeLayoutMedia { type: "spotlight-landscape"; spotlight: MediaViewModel[]; grid: UserMediaViewModel[]; } export interface SpotlightPortraitLayoutMedia { type: "spotlight-portrait"; spotlight: MediaViewModel[]; grid: UserMediaViewModel[]; } export interface SpotlightExpandedLayoutMedia { type: "spotlight-expanded"; spotlight: MediaViewModel[]; pip?: UserMediaViewModel; } export interface OneOnOneLayoutMedia { type: "one-on-one"; local: UserMediaViewModel; remote: UserMediaViewModel; } export interface PipLayoutMedia { type: "pip"; spotlight: MediaViewModel[]; } export type LayoutMedia = | GridLayoutMedia | SpotlightLandscapeLayoutMedia | SpotlightPortraitLayoutMedia | SpotlightExpandedLayoutMedia | OneOnOneLayoutMedia | PipLayoutMedia; export interface GridLayout { type: "grid"; spotlight?: SpotlightTileViewModel; grid: GridTileViewModel[]; setVisibleTiles: (value: number) => void; } export interface SpotlightLandscapeLayout { type: "spotlight-landscape"; spotlight: SpotlightTileViewModel; grid: GridTileViewModel[]; setVisibleTiles: (value: number) => void; } export interface SpotlightPortraitLayout { type: "spotlight-portrait"; spotlight: SpotlightTileViewModel; grid: GridTileViewModel[]; setVisibleTiles: (value: number) => void; } export interface SpotlightExpandedLayout { type: "spotlight-expanded"; spotlight: SpotlightTileViewModel; pip?: GridTileViewModel; } export interface OneOnOneLayout { type: "one-on-one"; local: GridTileViewModel; remote: GridTileViewModel; } export interface PipLayout { type: "pip"; spotlight: SpotlightTileViewModel; } /** * A layout defining the media tiles present on screen and their visual * arrangement. */ export type Layout = | GridLayout | SpotlightLandscapeLayout | SpotlightPortraitLayout | SpotlightExpandedLayout | OneOnOneLayout | PipLayout; export type GridMode = "grid" | "spotlight"; export type WindowMode = "normal" | "narrow" | "flat" | "pip"; /** * Sorting bins defining the order in which media tiles appear in the layout. */ enum SortingBin { /** * Yourself, when the "always show self" option is on. */ SelfAlwaysShown, /** * Participants that are sharing their screen. */ Presenters, /** * Participants that have been speaking recently. */ Speakers, /** * Participants that have their hand raised. */ HandRaised, /** * Participants with video. */ Video, /** * Participants not sharing any video. */ NoVideo, /** * Yourself, when the "always show self" option is off. */ SelfNotAlwaysShown, } interface LayoutScanState { layout: Layout | null; tiles: TileStore; } class UserMedia { private readonly scope = new ObservableScope(); public readonly vm: UserMediaViewModel; private readonly participant$: BehaviorSubject< LocalParticipant | RemoteParticipant | undefined >; public readonly speaker$: Behavior; public readonly presenter$: Behavior; public constructor( public readonly id: string, member: RoomMember | undefined, participant: LocalParticipant | RemoteParticipant | undefined, encryptionSystem: EncryptionSystem, livekitRoom: LivekitRoom, mediaDevices: MediaDevices, pretendToBeDisconnected$: Behavior, displayname$: Observable, handRaised$: Observable, reaction$: Observable, ) { this.participant$ = new BehaviorSubject(participant); if (participant?.isLocal) { this.vm = new LocalUserMediaViewModel( this.id, member, this.participant$ as Behavior, encryptionSystem, livekitRoom, mediaDevices, this.scope.behavior(displayname$), this.scope.behavior(handRaised$), this.scope.behavior(reaction$), ); } else { this.vm = new RemoteUserMediaViewModel( id, member, this.participant$.asObservable() as Observable< RemoteParticipant | undefined >, encryptionSystem, livekitRoom, pretendToBeDisconnected$, this.scope.behavior(displayname$), this.scope.behavior(handRaised$), this.scope.behavior(reaction$), ); } this.speaker$ = this.scope.behavior(observeSpeaker$(this.vm.speaking$)); this.presenter$ = this.scope.behavior( this.participant$.pipe( switchMap( (p) => (p && observeParticipantEvents( p, ParticipantEvent.TrackPublished, ParticipantEvent.TrackUnpublished, ParticipantEvent.LocalTrackPublished, ParticipantEvent.LocalTrackUnpublished, ).pipe(map((p) => p.isScreenShareEnabled))) ?? of(false), ), ), ); } public updateParticipant( newParticipant: LocalParticipant | RemoteParticipant | undefined, ): void { if (this.participant$.value !== newParticipant) { // Update the BehaviourSubject in the UserMedia. this.participant$.next(newParticipant); } } public destroy(): void { this.scope.end(); this.vm.destroy(); } } class ScreenShare { private readonly scope = new ObservableScope(); public readonly vm: ScreenShareViewModel; private readonly participant$: BehaviorSubject< LocalParticipant | RemoteParticipant >; public constructor( id: string, member: RoomMember | undefined, participant: LocalParticipant | RemoteParticipant, encryptionSystem: EncryptionSystem, livekitRoom: LivekitRoom, pretendToBeDisconnected$: Behavior, displayName$: Observable, ) { this.participant$ = new BehaviorSubject(participant); this.vm = new ScreenShareViewModel( id, member, this.participant$.asObservable(), encryptionSystem, livekitRoom, pretendToBeDisconnected$, this.scope.behavior(displayName$), participant.isLocal, ); } public destroy(): void { this.scope.end(); this.vm.destroy(); } } type MediaItem = UserMedia | ScreenShare; function getRoomMemberFromRtcMember( rtcMember: CallMembership, room: MatrixRoom, ): { id: string; member: RoomMember | undefined } { // WARN! This is not exactly the sender but the user defined in the state key. // This will be available once we change to the new "member as object" format in the MatrixRTC object. let id = rtcMember.sender + ":" + rtcMember.deviceId; if (!rtcMember.sender) { return { id, member: undefined }; } if ( rtcMember.sender === room.client.getUserId() && rtcMember.deviceId === room.client.getDeviceId() ) { id = "local"; } const member = room.getMember(rtcMember.sender) ?? undefined; return { id, member }; } export class CallViewModel extends ViewModel { private readonly livekitAlias = getLivekitAlias(this.matrixRTCSession); private readonly livekitE2EEKeyProvider = getE2eeKeyProvider( this.options.encryptionSystem, this.matrixRTCSession, ); private readonly e2eeLivekitOptions = (): E2EEOptions | undefined => this.livekitE2EEKeyProvider ? { keyProvider: this.livekitE2EEKeyProvider, worker: new E2EEWorker(), } : undefined; private readonly localFocus = makeFocus(this.matrixRTCSession); private readonly localConnection = this.localFocus.then( (focus) => new PublishConnection( focus, this.livekitAlias, this.matrixRTCSession.room.client, this.scope, this.membershipsAndFocusMap$, this.mediaDevices, this.muteStates, this.e2eeLivekitOptions(), ), ); // TODO-MULTI-SFU make sure that we consider the room memberships here as well (so that here we only have valid memberships) // this also makes it possible to use this memberships$ list in all observables based on it. // there should be no other call to: this.matrixRTCSession.memberships! public readonly memberships$ = this.scope.behavior( fromEvent( this.matrixRTCSession, MatrixRTCSessionEvent.MembershipsChanged, ).pipe( startWith(null), map(() => this.matrixRTCSession.memberships), ), ); private readonly membershipsAndFocusMap$ = this.scope.behavior( this.memberships$.pipe( map((memberships) => memberships.flatMap((m) => { const f = this.matrixRTCSession.resolveActiveFocus(m); return f && isLivekitFocus(f) ? [{ membership: m, focus: f }] : []; }), ), ), ); private readonly livekitServiceUrls$ = this.membershipsAndFocusMap$.pipe( map((v) => new Set(v.map(({ focus }) => focus.livekit_service_url))), ); private readonly remoteConnections$ = this.scope.behavior( combineLatest([this.localFocus, this.livekitServiceUrls$]).pipe( accumulate( new Map(), (prev, [localFocus, focusUrls]) => { const stopped = new Map(prev); const next = new Map(); for (const focusUrl of focusUrls) { if (focusUrl !== localFocus.livekit_service_url) { stopped.delete(focusUrl); let nextConnection = prev.get(focusUrl); if (!nextConnection) { logger.log( "SFU remoteConnections$ construct new connection: ", focusUrl, ); nextConnection = new Connection( { livekit_service_url: focusUrl, livekit_alias: this.livekitAlias, type: "livekit", }, this.livekitAlias, this.matrixRTCSession.room.client, this.scope, this.membershipsAndFocusMap$, this.e2eeLivekitOptions(), ); } else { logger.log( "SFU remoteConnections$ use prev connection: ", focusUrl, ); } next.set(focusUrl, nextConnection); } } for (const connection of stopped.values()) connection.stop(); return next; }, ), ), ); private readonly join$ = new Subject(); public join(): void { this.join$.next(); } private readonly leave$ = new Subject(); public leave(): void { this.leave$.next(); } private readonly connectionInstructions$ = this.join$.pipe( switchMap(() => this.remoteConnections$), startWith(new Map()), pairwise(), map(([prev, next]) => { const start = new Set(next.values()); for (const connection of prev.values()) start.delete(connection); const stop = new Set(prev.values()); for (const connection of next.values()) stop.delete(connection); return { start, stop }; }), this.scope.share, ); private readonly startConnection$ = this.connectionInstructions$.pipe( concatMap(({ start }) => start), ); private readonly stopConnection$ = this.connectionInstructions$.pipe( concatMap(({ stop }) => stop), ); public readonly allLivekitRooms$ = this.scope.behavior( combineLatest([ this.remoteConnections$, this.localConnection, this.localFocus, ]).pipe( map(([remoteConnections, localConnection, localFocus]) => Array.from(remoteConnections.entries()) .map(([index, c]) => ({ room: c.livekitRoom, url: index })) .concat([ { room: localConnection.livekitRoom, url: localFocus.livekit_service_url, }, ]), ), startWith([]), ), ); private readonly userId = this.matrixRoom.client.getUserId(); private readonly matrixConnected$ = this.scope.behavior( // To consider ourselves connected to MatrixRTC, we check the following: and$( // The client is connected to the sync loop ( fromEvent(this.matrixRoom.client, ClientEvent.Sync) as Observable< [SyncState] > ).pipe( startWith([this.matrixRoom.client.getSyncState()]), map(([state]) => state === SyncState.Syncing), ), // Room state observed by session says we're connected fromEvent( this.matrixRTCSession, MembershipManagerEvent.StatusChanged, ).pipe( startWith(null), map(() => this.matrixRTCSession.membershipStatus === Status.Connected), ), // Also watch out for warnings that we've likely hit a timeout and our // delayed leave event is being sent (this condition is here because it // provides an earlier warning than the sync loop timeout, and we wouldn't // see the actual leave event until we reconnect to the sync loop) fromEvent( this.matrixRTCSession, MembershipManagerEvent.ProbablyLeft, ).pipe( startWith(null), map(() => this.matrixRTCSession.probablyLeft !== true), ), ), ); private readonly connected$ = this.scope.behavior( and$( this.matrixConnected$, // TODO-MULTI-SFU // this.livekitConnectionState$.pipe( // map((state) => state === ConnectionState.Connected), // ), ), ); /** * Whether we should tell the user that we're reconnecting to the call. */ public readonly reconnecting$ = this.scope.behavior( this.connected$.pipe( // We are reconnecting if we previously had some successful initial // connection but are now disconnected scan( ({ connectedPreviously, reconnecting }, connectedNow) => ({ connectedPreviously: connectedPreviously || connectedNow, reconnecting: connectedPreviously && !connectedNow, }), { connectedPreviously: false, reconnecting: false }, ), map(({ reconnecting }) => reconnecting), ), ); /** * Whether various media/event sources should pretend to be disconnected from * all network input, even if their connection still technically works. */ // We do this when the app is in the 'reconnecting' state, because it might be // that the LiveKit connection is still functional while the homeserver is // down, for example, and we want to avoid making people worry that the app is // in a split-brained state. private readonly pretendToBeDisconnected$ = this.reconnecting$; private readonly participants$ = this.scope.behavior< { participant: LocalParticipant | RemoteParticipant; member: RoomMember; livekitRoom: LivekitRoom; }[] >( from(this.localConnection) .pipe( switchMap((localConnection) => { const memberError = (): never => { throw new Error("No room member for call membership"); }; const localParticipant = { participant: localConnection.livekitRoom.localParticipant, member: this.matrixRoom.getMember(this.userId ?? "") ?? memberError(), livekitRoom: localConnection.livekitRoom, }; return this.remoteConnections$.pipe( switchMap((connections) => combineLatest( [localConnection, ...connections.values()].map((c) => c.publishingParticipants$.pipe( map((ps) => ps.map(({ participant, membership }) => ({ participant, member: getRoomMemberFromRtcMember( membership, this.matrixRoom, )?.member ?? memberError(), livekitRoom: c.livekitRoom, })), ), ), ), ), ), map((remoteParticipants) => [ localParticipant, ...remoteParticipants.flat(1), ]), ); }), ) .pipe(startWith([]), pauseWhen(this.pretendToBeDisconnected$)), ); /** * Displaynames for each member of the call. This will disambiguate * any displaynames that clashes with another member. Only members * joined to the call are considered here. */ // It turns out that doing the disambiguation above is rather expensive on Safari (10x slower // than on Chrome/Firefox). This means it is important that we multicast the result so that we // don't do this work more times than we need to. This is achieved by converting to a behavior: public readonly memberDisplaynames$ = this.scope.behavior( merge( // Handle call membership changes. fromEvent( this.matrixRTCSession, MatrixRTCSessionEvent.MembershipsChanged, ), // Handle room membership changes (and displayname updates) fromEvent(this.matrixRoom, RoomStateEvent.Members), // TODO: do we need: pauseWhen(this.pretendToBeDisconnected$), ).pipe( startWith(null), map(() => { const memberships = this.matrixRTCSession.memberships; const displaynameMap = new Map([ ["local", this.matrixRoom.getMember(this.userId!)!.rawDisplayName], ]); const room = this.matrixRoom; // We only consider RTC members for disambiguation as they are the only visible members. for (const rtcMember of memberships) { const matrixIdentifier = `${rtcMember.sender}:${rtcMember.deviceId}`; const { member } = getRoomMemberFromRtcMember(rtcMember, room); if (!member) { logger.error( "Could not find member for media id:", matrixIdentifier, ); continue; } const disambiguate = shouldDisambiguate(member, memberships, room); displaynameMap.set( matrixIdentifier, calculateDisplayName(member, disambiguate), ); } return displaynameMap; }, ), ); public readonly handsRaised$ = this.scope.behavior( this.handsRaisedSubject$.pipe(pauseWhen(this.pretendToBeDisconnected$)), ); public readonly reactions$ = this.scope.behavior( this.reactionsSubject$.pipe( map((v) => Object.fromEntries( Object.entries(v).map(([a, { reactionOption }]) => [ a, reactionOption, ]), ), ), pauseWhen(this.pretendToBeDisconnected$), ), ); /** * List of MediaItems that we want to display */ private readonly mediaItems$ = this.scope.behavior( combineLatest([ this.participants$, duplicateTiles.value$, this.memberships$, showNonMemberTiles.value$, ]).pipe( scan( ( prevItems, [participants, duplicateTiles, memberships, showNonMemberTiles], ) => { const newItems: Map = new Map( function* (this: CallViewModel): Iterable<[string, MediaItem]> { for (const { participant, member, livekitRoom } of participants) { const matrixId = participant.isLocal ? "local" : participant.identity; for (let i = 0; i < 1 + duplicateTiles; i++) { const mediaId = `${matrixId}:${i}`; let prevMedia = prevItems.get(mediaId); if (prevMedia && prevMedia instanceof UserMedia) { prevMedia.updateParticipant(participant); if (prevMedia.vm.member === undefined) { // We have a previous media created because of the `debugShowNonMember` flag. // In this case we actually replace the media item. // This "hack" never occurs if we do not use the `debugShowNonMember` debugging // option and if we always find a room member for each rtc member (which also // only fails if we have a fundamental problem) prevMedia = undefined; } } yield [ mediaId, // We create UserMedia with or without a participant. // This will be the initial value of a BehaviourSubject. // Once a participant appears we will update the BehaviourSubject. (see above) prevMedia ?? new UserMedia( mediaId, member, participant, this.options.encryptionSystem, livekitRoom, this.mediaDevices, this.pretendToBeDisconnected$, this.memberDisplaynames$.pipe( map((m) => m.get(matrixId) ?? "[👻]"), ), this.handsRaised$.pipe( map((v) => v[matrixId]?.time ?? null), ), this.reactions$.pipe( map((v) => v[matrixId] ?? undefined), ), ), ]; if (participant?.isScreenShareEnabled) { const screenShareId = `${mediaId}:screen-share`; yield [ screenShareId, prevItems.get(screenShareId) ?? new ScreenShare( screenShareId, member, participant, this.options.encryptionSystem, livekitRoom, this.pretendToBeDisconnected$, this.memberDisplaynames$.pipe( map((m) => m.get(matrixId) ?? "[👻]"), ), ), ]; } } } }.bind(this)(), ); // Generate non member items (items without a corresponding MatrixRTC member) // Those items should not be rendered, they are participants in LiveKit that do not have a corresponding // MatrixRTC members. This cannot be any good: // - A malicious user impersonates someone // - Someone injects abusive content // - The user cannot have encryption keys so it makes no sense to participate // We can only trust users that have a MatrixRTC member event. // // This is still available as a debug option. This can be useful // - If one wants to test scalability using the LiveKit CLI. // - If an experimental project does not yet do the MatrixRTC bits. // - If someone wants to debug if the LiveKit connection works but MatrixRTC room state failed to arrive. // TODO-MULTI-SFU // const newNonMemberItems = showNonMemberTiles // ? new Map( // function* ( // this: CallViewModel, // ): Iterable<[string, MediaItem]> { // for (const participant of remoteParticipants) { // for (let i = 0; i < 1 + duplicateTiles; i++) { // const maybeNonMemberParticipantId = // participant.identity + ":" + i; // if (!newItems.has(maybeNonMemberParticipantId)) { // const nonMemberId = maybeNonMemberParticipantId; // yield [ // nonMemberId, // prevItems.get(nonMemberId) ?? // new UserMedia( // nonMemberId, // undefined, // participant, // this.options.encryptionSystem, // localConnection.livekitRoom, // this.mediaDevices, // this.pretendToBeDisconnected$, // this.memberDisplaynames$.pipe( // map( // (m) => // m.get(participant.identity) ?? "[👻]", // ), // ), // of(null), // of(null), // ), // ]; // } // } // } // }.bind(this)(), // ) // : new Map(); // if (newNonMemberItems.size > 0) { // logger.debug("Added NonMember items: ", newNonMemberItems); // } const combinedNew = new Map([ // ...newNonMemberItems.entries(), ...newItems.entries(), ]); for (const [id, t] of prevItems) if (!combinedNew.has(id)) t.destroy(); return combinedNew; }, new Map(), ), map((mediaItems) => [...mediaItems.values()]), finalizeValue((ts) => { for (const t of ts) t.destroy(); }), ), ); /** * List of MediaItems that we want to display, that are of type UserMedia */ private readonly userMedia$ = this.scope.behavior( this.mediaItems$.pipe( map((mediaItems) => mediaItems.filter((m): m is UserMedia => m instanceof UserMedia), ), ), ); public readonly joinSoundEffect$ = this.userMedia$.pipe( pairwise(), filter( ([prev, current]) => current.length <= MAX_PARTICIPANT_COUNT_FOR_SOUND && current.length > prev.length, ), map(() => {}), throttleTime(THROTTLE_SOUND_EFFECT_MS), ); /** * The number of participants currently in the call. * * - Each participant has a corresponding MatrixRTC membership state event * - There can be multiple participants for one Matrix user if they join from * multiple devices. */ public readonly participantCount$ = this.scope.behavior( this.memberships$.pipe(map((ms) => ms.length)), ); private readonly allOthersLeft$ = this.memberships$.pipe( pairwise(), filter( ([prev, current]) => current.every((m) => m.sender === this.userId) && prev.some((m) => m.sender !== this.userId), ), map(() => {}), take(1), ); public readonly autoLeave$ = this.options.autoLeaveWhenOthersLeft ? this.allOthersLeft$ : NEVER; private readonly didSendCallNotification$ = fromEvent( this.matrixRTCSession, MatrixRTCSessionEvent.DidSendCallNotification, ) as Observable< Parameters< MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification] > >; /** * Whenever the RTC session tells us that it intends to ring the remote * participant's devices, this emits an Observable tracking the current state of * that ringing process. */ // This is a behavior since we need to store the latest state for when we subscribe to this after `didSendCallNotification$` // has already emitted but we still need the latest observable with a timeout timer that only gets created on after receiving `notificationEvent`. // A behavior will emit the latest observable with the running timer to new subscribers. // see also: callPickupState$ and in particular the line: `return this.ring$.pipe(mergeAll());` here we otherwise might get an EMPTY observable if // `ring$` would not be a behavior. private readonly ring$: Behavior<"ringing" | "timeout" | "decline" | null> = this.scope.behavior( this.didSendCallNotification$.pipe( filter( ([notificationEvent]) => notificationEvent.notification_type === "ring", ), switchMap(([notificationEvent]) => { const lifetimeMs = notificationEvent?.lifetime ?? 0; return concat( lifetimeMs === 0 ? // If no lifetime, skip the ring state of(null) : // Ring until lifetime ms have passed timer(lifetimeMs).pipe( ignoreElements(), startWith("ringing" as const), ), // The notification lifetime has timed out, meaning ringing has likely // stopped on all receiving clients. of("timeout" as const), // This makes sure we will not drop into the `endWith("decline" as const)` state NEVER, ).pipe( takeUntil( ( fromEvent(this.matrixRoom, RoomEvent.Timeline) as Observable< Parameters > ).pipe( filter( ([event]) => event.getType() === EventType.RTCDecline && event.getRelation()?.rel_type === "m.reference" && event.getRelation()?.event_id === notificationEvent.event_id && event.getSender() !== this.userId, ), ), ), endWith("decline" as const), ); }), ), null, ); /** * Whether some Matrix user other than ourself is joined to the call. */ private readonly someoneElseJoined$ = this.memberships$.pipe( map((ms) => ms.some((m) => m.sender !== this.userId)), ) as Behavior; /** * The current call pickup state of the call. * - "unknown": The client has not yet sent the notification event. We don't know if it will because it first needs to send its own membership. * Then we can conclude if we were the first one to join or not. * This may also be set if we are disconnected. * - "ringing": The call is ringing on other devices in this room (This client should give audiovisual feedback that this is happening). * - "timeout": No-one picked up in the defined time this call should be ringing on others devices. * The call failed. If desired this can be used as a trigger to exit the call. * - "success": Someone else joined. The call is in a normal state. No audiovisual feedback. * - null: EC is configured to never show any waiting for answer state. */ public readonly callPickupState$: Behavior< "unknown" | "ringing" | "timeout" | "decline" | "success" | null > = this.options.waitForCallPickup ? this.scope.behavior< "unknown" | "ringing" | "timeout" | "decline" | "success" >( combineLatest( [this.livekitConnectionState$, this.someoneElseJoined$, this.ring$], (livekitConnectionState, someoneElseJoined, ring) => { if (livekitConnectionState === ConnectionState.Disconnected) { // Do not ring until we're connected. return "unknown" as const; } else if (someoneElseJoined) { return "success" as const; } // Show the ringing state of the most recent ringing attempt. // as long as we have not yet sent an RTC notification event, ring will be null -> callPickupState$ = unknown. return ring ?? ("unknown" as const); }, ), ) : constant(null); public readonly leaveSoundEffect$ = combineLatest([ this.callPickupState$, this.userMedia$, ]).pipe( // Until the call is successful, do not play a leave sound. // If callPickupState$ is null, then we always play the sound as it will not conflict with a decline sound. skipWhile(([c]) => c !== null && c !== "success"), map(([, userMedia]) => userMedia), pairwise(), filter( ([prev, current]) => current.length <= MAX_PARTICIPANT_COUNT_FOR_SOUND && current.length < prev.length, ), map(() => {}), throttleTime(THROTTLE_SOUND_EFFECT_MS), ); /** * List of MediaItems that we want to display, that are of type ScreenShare */ private readonly screenShares$ = this.scope.behavior( this.mediaItems$.pipe( map((mediaItems) => mediaItems.filter((m): m is ScreenShare => m instanceof ScreenShare), ), ), ); private readonly spotlightSpeaker$ = this.scope.behavior( this.userMedia$.pipe( switchMap((mediaItems) => mediaItems.length === 0 ? of([]) : combineLatest( mediaItems.map((m) => m.vm.speaking$.pipe(map((s) => [m, s] as const)), ), ), ), scan<(readonly [UserMedia, boolean])[], UserMedia | undefined, null>( (prev, mediaItems) => { // Only remote users that are still in the call should be sticky const [stickyMedia, stickySpeaking] = (!prev?.vm.local && mediaItems.find(([m]) => m === prev)) || []; // Decide who to spotlight: // If the previous speaker is still speaking, stick with them rather // than switching eagerly to someone else return stickySpeaking ? stickyMedia! : // Otherwise, select any remote user who is speaking (mediaItems.find(([m, s]) => !m.vm.local && s)?.[0] ?? // Otherwise, stick with the person who was last speaking stickyMedia ?? // Otherwise, spotlight an arbitrary remote user mediaItems.find(([m]) => !m.vm.local)?.[0] ?? // Otherwise, spotlight the local user mediaItems.find(([m]) => m.vm.local)?.[0]); }, null, ), map((speaker) => speaker?.vm ?? null), ), ); private readonly grid$ = this.scope.behavior( this.userMedia$.pipe( switchMap((mediaItems) => { const bins = mediaItems.map((m) => combineLatest( [ m.speaker$, m.presenter$, m.vm.videoEnabled$, m.vm.handRaised$, m.vm instanceof LocalUserMediaViewModel ? m.vm.alwaysShow$ : of(false), ], (speaker, presenter, video, handRaised, alwaysShow) => { let bin: SortingBin; if (m.vm.local) bin = alwaysShow ? SortingBin.SelfAlwaysShown : SortingBin.SelfNotAlwaysShown; else if (presenter) bin = SortingBin.Presenters; else if (speaker) bin = SortingBin.Speakers; else if (handRaised) bin = SortingBin.HandRaised; else if (video) bin = SortingBin.Video; else bin = SortingBin.NoVideo; return [m, bin] as const; }, ), ); // Sort the media by bin order and generate a tile for each one return bins.length === 0 ? of([]) : combineLatest(bins, (...bins) => bins.sort(([, bin1], [, bin2]) => bin1 - bin2).map(([m]) => m.vm), ); }), distinctUntilChanged(shallowEquals), ), ); private readonly spotlight$ = this.scope.behavior( this.screenShares$.pipe( switchMap((screenShares) => { if (screenShares.length > 0) { return of(screenShares.map((m) => m.vm)); } return this.spotlightSpeaker$.pipe( map((speaker) => (speaker ? [speaker] : [])), ); }), distinctUntilChanged(shallowEquals), ), ); private readonly pip$ = this.scope.behavior( combineLatest([ this.screenShares$, this.spotlightSpeaker$, this.mediaItems$, ]).pipe( switchMap(([screenShares, spotlight, mediaItems]) => { if (screenShares.length > 0) { return this.spotlightSpeaker$; } if (!spotlight || spotlight.local) { return of(null); } const localUserMedia = mediaItems.find( (m) => m.vm instanceof LocalUserMediaViewModel, ) as UserMedia | undefined; const localUserMediaViewModel = localUserMedia?.vm as | LocalUserMediaViewModel | undefined; if (!localUserMediaViewModel) { return of(null); } return localUserMediaViewModel.alwaysShow$.pipe( map((alwaysShow) => { if (alwaysShow) { return localUserMediaViewModel; } return null; }), ); }), ), ); private readonly hasRemoteScreenShares$: Observable = this.spotlight$.pipe( map((spotlight) => spotlight.some((vm) => !vm.local && vm instanceof ScreenShareViewModel), ), distinctUntilChanged(), ); private readonly pipEnabled$ = this.scope.behavior(setPipEnabled$, false); private readonly naturalWindowMode$ = this.scope.behavior( fromEvent(window, "resize").pipe( startWith(null), map(() => { const height = window.innerHeight; const width = window.innerWidth; if (height <= 400 && width <= 340) return "pip"; // Our layouts for flat windows are better at adapting to a small width // than our layouts for narrow windows are at adapting to a small height, // so we give "flat" precedence here if (height <= 600) return "flat"; if (width <= 600) return "narrow"; return "normal"; }), ), ); /** * The general shape of the window. */ public readonly windowMode$ = this.scope.behavior( this.pipEnabled$.pipe( switchMap((pip) => pip ? of("pip") : this.naturalWindowMode$, ), ), ); private readonly spotlightExpandedToggle$ = new Subject(); public readonly spotlightExpanded$ = this.scope.behavior( this.spotlightExpandedToggle$.pipe( accumulate(false, (expanded) => !expanded), ), ); private readonly gridModeUserSelection$ = new Subject(); /** * The layout mode of the media tile grid. */ public readonly gridMode$ = // If the user hasn't selected spotlight and somebody starts screen sharing, // automatically switch to spotlight mode and reset when screen sharing ends this.scope.behavior( this.gridModeUserSelection$.pipe( startWith(null), switchMap((userSelection) => (userSelection === "spotlight" ? EMPTY : combineLatest([ this.hasRemoteScreenShares$, this.windowMode$, ]).pipe( skip(userSelection === null ? 0 : 1), map( ([hasScreenShares, windowMode]): GridMode => hasScreenShares || windowMode === "flat" ? "spotlight" : "grid", ), ) ).pipe(startWith(userSelection ?? "grid")), ), ), ); public setGridMode(value: GridMode): void { this.gridModeUserSelection$.next(value); } private readonly gridLayoutMedia$: Observable = combineLatest([this.grid$, this.spotlight$], (grid, spotlight) => ({ type: "grid", spotlight: spotlight.some((vm) => vm instanceof ScreenShareViewModel) ? spotlight : undefined, grid, })); private readonly spotlightLandscapeLayoutMedia$: Observable = combineLatest([this.grid$, this.spotlight$], (grid, spotlight) => ({ type: "spotlight-landscape", spotlight, grid, })); private readonly spotlightPortraitLayoutMedia$: Observable = combineLatest([this.grid$, this.spotlight$], (grid, spotlight) => ({ type: "spotlight-portrait", spotlight, grid, })); private readonly spotlightExpandedLayoutMedia$: Observable = combineLatest([this.spotlight$, this.pip$], (spotlight, pip) => ({ type: "spotlight-expanded", spotlight, pip: pip ?? undefined, })); private readonly oneOnOneLayoutMedia$: Observable = this.mediaItems$.pipe( map((mediaItems) => { if (mediaItems.length !== 2) return null; const local = mediaItems.find((vm) => vm.vm.local)?.vm as | LocalUserMediaViewModel | undefined; const remote = mediaItems.find((vm) => !vm.vm.local)?.vm as | RemoteUserMediaViewModel | undefined; // There might not be a remote tile if there are screen shares, or if // only the local user is in the call and they're using the duplicate // tiles option if (!remote || !local) return null; return { type: "one-on-one", local, remote }; }), ); private readonly pipLayoutMedia$: Observable = this.spotlight$.pipe(map((spotlight) => ({ type: "pip", spotlight }))); /** * The media to be used to produce a layout. */ private readonly layoutMedia$ = this.scope.behavior( this.windowMode$.pipe( switchMap((windowMode) => { switch (windowMode) { case "normal": return this.gridMode$.pipe( switchMap((gridMode) => { switch (gridMode) { case "grid": return this.oneOnOneLayoutMedia$.pipe( switchMap((oneOnOne) => oneOnOne === null ? this.gridLayoutMedia$ : of(oneOnOne), ), ); case "spotlight": return this.spotlightExpanded$.pipe( switchMap((expanded) => expanded ? this.spotlightExpandedLayoutMedia$ : this.spotlightLandscapeLayoutMedia$, ), ); } }), ); case "narrow": return this.oneOnOneLayoutMedia$.pipe( switchMap((oneOnOne) => oneOnOne === null ? combineLatest( [this.grid$, this.spotlight$], (grid, spotlight) => grid.length > smallMobileCallThreshold || spotlight.some( (vm) => vm instanceof ScreenShareViewModel, ) ? this.spotlightPortraitLayoutMedia$ : this.gridLayoutMedia$, ).pipe(switchAll()) : // The expanded spotlight layout makes for a better one-on-one // experience in narrow windows this.spotlightExpandedLayoutMedia$, ), ); case "flat": return this.gridMode$.pipe( switchMap((gridMode) => { switch (gridMode) { case "grid": // Yes, grid mode actually gets you a "spotlight" layout in // this window mode. return this.spotlightLandscapeLayoutMedia$; case "spotlight": return this.spotlightExpandedLayoutMedia$; } }), ); case "pip": return this.pipLayoutMedia$; } }), ), ); // There is a cyclical dependency here: the layout algorithms want to know // which tiles are on screen, but to know which tiles are on screen we have to // first render a layout. To deal with this we assume initially that no tiles // are visible, and loop the data back into the layouts with a Subject. private readonly visibleTiles$ = new Subject(); private readonly setVisibleTiles = (value: number): void => this.visibleTiles$.next(value); private readonly layoutInternals$ = this.scope.behavior< LayoutScanState & { layout: Layout } >( combineLatest([ this.layoutMedia$, this.visibleTiles$.pipe(startWith(0), distinctUntilChanged()), ]).pipe( scan< [LayoutMedia, number], LayoutScanState & { layout: Layout }, LayoutScanState >( ({ tiles: prevTiles }, [media, visibleTiles]) => { let layout: Layout; let newTiles: TileStore; switch (media.type) { case "grid": case "spotlight-landscape": case "spotlight-portrait": [layout, newTiles] = gridLikeLayout( media, visibleTiles, this.setVisibleTiles, prevTiles, ); break; case "spotlight-expanded": [layout, newTiles] = spotlightExpandedLayout(media, prevTiles); break; case "one-on-one": [layout, newTiles] = oneOnOneLayout(media, prevTiles); break; case "pip": [layout, newTiles] = pipLayout(media, prevTiles); break; } return { layout, tiles: newTiles }; }, { layout: null, tiles: TileStore.empty() }, ), ), ); /** * The layout of tiles in the call interface. */ public readonly layout$ = this.scope.behavior( this.layoutInternals$.pipe(map(({ layout }) => layout)), ); /** * The current generation of the tile store, exposed for debugging purposes. */ public readonly tileStoreGeneration$ = this.scope.behavior( this.layoutInternals$.pipe(map(({ tiles }) => tiles.generation)), ); public showSpotlightIndicators$ = this.scope.behavior( this.layout$.pipe(map((l) => l.type !== "grid")), ); public showSpeakingIndicators$ = this.scope.behavior( this.layout$.pipe( switchMap((l) => { switch (l.type) { case "spotlight-landscape": case "spotlight-portrait": // If the spotlight is showing the active speaker, we can do without // speaking indicators as they're a redundant visual cue. But if // screen sharing feeds are in the spotlight we still need them. return l.spotlight.media$.pipe( map((models: MediaViewModel[]) => models.some((m) => m instanceof ScreenShareViewModel), ), ); // In expanded spotlight layout, the active speaker is always shown in // the picture-in-picture tile so there is no need for speaking // indicators. And in one-on-one layout there's no question as to who is // speaking. case "spotlight-expanded": case "one-on-one": return of(false); default: return of(true); } }), ), ); public readonly toggleSpotlightExpanded$ = this.scope.behavior< (() => void) | null >( this.windowMode$.pipe( switchMap((mode) => mode === "normal" ? this.layout$.pipe( map( (l) => l.type === "spotlight-landscape" || l.type === "spotlight-expanded", ), ) : of(false), ), distinctUntilChanged(), map((enabled) => enabled ? (): void => this.spotlightExpandedToggle$.next() : null, ), ), ); private readonly screenTap$ = new Subject(); private readonly controlsTap$ = new Subject(); private readonly screenHover$ = new Subject(); private readonly screenUnhover$ = new Subject(); /** * Callback for when the user taps the call view. */ public tapScreen(): void { this.screenTap$.next(); } /** * Callback for when the user taps the call's controls. */ public tapControls(): void { this.controlsTap$.next(); } /** * Callback for when the user hovers over the call view. */ public hoverScreen(): void { this.screenHover$.next(); } /** * Callback for when the user stops hovering over the call view. */ public unhoverScreen(): void { this.screenUnhover$.next(); } public readonly showHeader$ = this.scope.behavior( this.windowMode$.pipe(map((mode) => mode !== "pip" && mode !== "flat")), ); public readonly showFooter$ = this.scope.behavior( this.windowMode$.pipe( switchMap((mode) => { switch (mode) { case "pip": return of(false); case "normal": case "narrow": return of(true); case "flat": // Sadly Firefox has some layering glitches that prevent the footer // from appearing properly. They happen less often if we never hide // the footer. if (isFirefox()) return of(true); // Show/hide the footer in response to interactions return merge( this.screenTap$.pipe(map(() => "tap screen" as const)), this.controlsTap$.pipe(map(() => "tap controls" as const)), this.screenHover$.pipe(map(() => "hover" as const)), ).pipe( switchScan((state, interaction) => { switch (interaction) { case "tap screen": return state ? // Toggle visibility on tap of(false) : // Hide after a timeout timer(showFooterMs).pipe( map(() => false), startWith(true), ); case "tap controls": // The user is interacting with things, so reset the timeout return timer(showFooterMs).pipe( map(() => false), startWith(true), ); case "hover": // Show on hover and hide after a timeout return race( timer(showFooterMs), this.screenUnhover$.pipe(take(1)), ).pipe( map(() => false), startWith(true), ); } }, false), startWith(false), ); } }), ), ); /** * Whether audio is currently being output through the earpiece. */ public readonly earpieceMode$ = this.scope.behavior( combineLatest( [ this.mediaDevices.audioOutput.available$, this.mediaDevices.audioOutput.selected$, ], (available, selected) => selected !== undefined && available.get(selected.id)?.type === "earpiece", ), ); /** * Callback to toggle between the earpiece and the loudspeaker. * * This will be `null` in case the target does not exist in the list * of available audio outputs. */ public readonly audioOutputSwitcher$ = this.scope.behavior<{ targetOutput: "earpiece" | "speaker"; switch: () => void; } | null>( combineLatest( [ this.mediaDevices.audioOutput.available$, this.mediaDevices.audioOutput.selected$, ], (available, selected) => { const selectionType = selected && available.get(selected.id)?.type; // If we are in any output mode other than speaker switch to speaker. const newSelectionType: "earpiece" | "speaker" = selectionType === "speaker" ? "earpiece" : "speaker"; const newSelection = [...available].find( ([, d]) => d.type === newSelectionType, ); if (newSelection === undefined) return null; const [id] = newSelection; return { targetOutput: newSelectionType, switch: (): void => this.mediaDevices.audioOutput.select(id), }; }, ), ); /** * Emits an array of reactions that should be visible on the screen. */ public readonly visibleReactions$ = this.scope.behavior( showReactions.value$.pipe( switchMap((show) => (show ? this.reactions$ : of({}))), scan< Record, { sender: string; emoji: string; startX: number }[] >((acc, latest) => { const newSet: { sender: string; emoji: string; startX: number }[] = []; for (const [sender, reaction] of Object.entries(latest)) { const startX = acc.find((v) => v.sender === sender && v.emoji)?.startX ?? Math.ceil(Math.random() * 80) + 10; newSet.push({ sender, emoji: reaction.emoji, startX }); } return newSet; }, []), ), ); /** * Emits an array of reactions that should be played. */ public readonly audibleReactions$ = playReactionsSound.value$.pipe( switchMap((show) => show ? this.reactions$ : of>({}), ), map((reactions) => Object.values(reactions).map((v) => v.name)), scan( (acc, latest) => { return { playing: latest.filter( (v) => acc.playing.includes(v) || acc.newSounds.includes(v), ), newSounds: latest.filter( (v) => !acc.playing.includes(v) && !acc.newSounds.includes(v), ), }; }, { playing: [], newSounds: [] }, ), map((v) => v.newSounds), ); /** * Emits an event every time a new hand is raised in * the call. */ public readonly newHandRaised$ = this.handsRaised$.pipe( map((v) => Object.keys(v).length), scan( (acc, newValue) => ({ value: newValue, playSounds: newValue > acc.value, }), { value: 0, playSounds: false }, ), filter((v) => v.playSounds), ); /** * Emits an event every time a new screenshare is started in * the call. */ public readonly newScreenShare$ = this.screenShares$.pipe( map((v) => v.length), scan( (acc, newValue) => ({ value: newValue, playSounds: newValue > acc.value, }), { value: 0, playSounds: false }, ), filter((v) => v.playSounds), ); // TODO-REBASE: expose connection state observable public readonly livekitConnectionState$: Observable; public constructor( // A call is permanently tied to a single Matrix room private readonly matrixRTCSession: MatrixRTCSession, private readonly matrixRoom: MatrixRoom, private readonly mediaDevices: MediaDevices, private readonly muteStates: MuteStates, private readonly options: CallViewModelOptions, private readonly handsRaisedSubject$: Observable< Record >, private readonly reactionsSubject$: Observable< Record >, ) { super(); void from(this.localConnection) .pipe(this.scope.bind()) .subscribe( (c) => void c .start() // eslint-disable-next-line no-console .then(() => console.log("successfully started publishing")) // eslint-disable-next-line no-console .catch((e) => console.error("failed to start publishing", e)), ); this.startConnection$ .pipe(this.scope.bind()) .subscribe((c) => void c.start()); this.stopConnection$.pipe(this.scope.bind()).subscribe((c) => c.stop()); combineLatest([this.localFocus, this.join$]) .pipe(this.scope.bind()) .subscribe(([localFocus]) => { void enterRTCSession( this.matrixRTCSession, localFocus, this.options.encryptionSystem.kind !== E2eeType.NONE, true, true, ); }); this.join$.pipe(this.scope.bind()).subscribe(() => { // TODO-MULTI-SFU: this makes no sense what so ever!!! // need to look into this again. // leaveRTCSession( // this.matrixRTCSession, // "user", // TODO-MULTI-SFU ? // // Wait for the sound in widget mode (it's not long) // Promise.resolve(), // TODO-MULTI-SFU // //Promise.all([audioPromise, posthogRequest]), // ).catch((e) => { // logger.error("Error leaving RTC session", e); // }); }); // Pause upstream of all local media tracks when we're disconnected from // MatrixRTC, because it can be an unpleasant surprise for the app to say // 'reconnecting' and yet still be transmitting your media to others. // We use matrixConnected$ rather than reconnecting$ because we want to // pause tracks during the initial joining sequence too until we're sure // that our own media is displayed on screen. void this.localConnection.then((localConnection) => this.matrixConnected$.pipe(this.scope.bind()).subscribe((connected) => { const publications = localConnection.livekitRoom.localParticipant.trackPublications.values(); if (connected) { for (const p of publications) { if (p.track?.isUpstreamPaused === true) { const kind = p.track.kind; logger.log( `Resuming ${kind} track (MatrixRTC connection present)`, ); p.track .resumeUpstream() .catch((e) => logger.error( `Failed to resume ${kind} track after MatrixRTC reconnection`, e, ), ); } } } else { for (const p of publications) { if (p.track?.isUpstreamPaused === false) { const kind = p.track.kind; logger.log( `Pausing ${kind} track (uncertain MatrixRTC connection)`, ); p.track .pauseUpstream() .catch((e) => logger.error( `Failed to pause ${kind} track after entering uncertain MatrixRTC connection`, e, ), ); } } } }), ); // Join automatically this.join(); // TODO-MULTI-SFU: Use this view model for the lobby as well, and only call this once 'join' is clicked? } } // TODO-MULTI-SFU // Setup and update the keyProvider which was create by `createRoom` was a thing before. Now we never update if the E2EEsystem changes // do we need this? function getE2eeKeyProvider( e2eeSystem: EncryptionSystem, rtcSession: MatrixRTCSession, ): BaseKeyProvider | undefined { if (e2eeSystem.kind === E2eeType.NONE) return undefined; if (e2eeSystem.kind === E2eeType.PER_PARTICIPANT) { const keyProvider = new MatrixKeyProvider(); keyProvider.setRTCSession(rtcSession); return keyProvider; } else if (e2eeSystem.kind === E2eeType.SHARED_KEY && e2eeSystem.secret) { const keyProvider = new ExternalE2EEKeyProvider(); keyProvider .setKey(e2eeSystem.secret) .catch((e) => logger.error("Failed to set shared key for E2EE", e)); return keyProvider; } }