diff --git a/sdk/src/RTCParticipation.test.ts b/sdk/src/RTCParticipation.test.ts new file mode 100644 index 000000000..6229068c2 --- /dev/null +++ b/sdk/src/RTCParticipation.test.ts @@ -0,0 +1,397 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { type MatrixClient, type Room, SyncState } from "matrix-js-sdk"; +import { logger } from "matrix-js-sdk/lib/logger"; +import { + MatrixRTCSessionEvent, + MembershipManagerEvent, + type Transport, +} from "matrix-js-sdk/lib/matrixrtc"; +import { BehaviorSubject, EMPTY, Subject } from "rxjs"; +import { describe, expect, it, type Mock, vi } from "vitest"; + +import { + type DataMessage, + type ParticipationState, + type RTCMember, + type RTCParticipationOptions, + type RTCSlot, + type TransportMetadata, +} from "./api"; +import { defaultSessionTimings, MatrixRTCMode } from "./config"; +import { E2eeType } from "./encryption"; +import { MatrixRTCError } from "./errors"; +import { + type DataPacket, + type MediaBackend, + type MediaBackendContext, + MediaConnectionState, + type MediaKey, +} from "./media-backend/api"; +import { type MemberMedia } from "./media-api"; +import { constant } from "./reactive/Behavior"; +import { ObservableScope } from "./reactive/ObservableScope"; +import { createRTCParticipation } from "./RTCParticipation"; +import { transportId } from "./matrixrtc/Transports"; +import { flushPromises, mockEmitter, MockRTCSession } from "./utils/test"; + +const transport: Transport = { + type: "livekit", + livekit_service_url: "https://lk.example.org", +}; +const userId = "@alice:example.org"; +const deviceId = "AAAA"; +const legacyMemberId = `${userId}:${deviceId}`; + +function metadata(raw: Transport): TransportMetadata { + return { + type: raw.type, + id: transportId(raw), + raw, + resolved$: constant(undefined), + }; +} + +function member( + overrides: Partial & Pick, +): RTCMember { + return { + local: false, + rtcBackendIdentity: overrides.memberId, + backendIdentities: [overrides.memberId], + deviceId: "DEV", + transport$: constant(metadata(transport)), + applicationData$: constant({}), + displayName$: constant(overrides.userId), + avatarUrl$: constant(undefined), + ...overrides, + }; +} + +const bob = member({ userId: "@bob:example.org", memberId: "bob-member" }); +const own = member({ userId, deviceId, memberId: legacyMemberId, local: true }); + +/** A backend the test drives: it prepares when told, and reports what the participation asked of it. */ +interface FakeBackend { + backend: MediaBackend; + context: MediaBackendContext; + prepared: PromiseWithResolvers<{ canDelegateDelayedLeave: boolean }>; + connectionState$: BehaviorSubject; + data$: Subject; + setPublishing: Mock<(publish: boolean) => void>; + delegateDelayedLeave: Mock<(delayId: string) => Promise>; + mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>; +} + +function setup( + options: Partial = {}, + slotOptions: { matrixRTCMode?: MatrixRTCMode } = {}, +): { + scope: ObservableScope; + participation: ReturnType["participation"]; + session: MockRTCSession; + mockedSession: ReturnType; + join: ReturnType; + members$: BehaviorSubject; + fake: FakeBackend; + states: ParticipationState[]; +} { + const scope = new ObservableScope(); + const room = { roomId: "!room:example.org" } as Room; + const session = new MockRTCSession(room); + const mockedSession = session.asMockedSession(); + const join = vi.spyOn(session, "join"); + const client = { + ...mockEmitter(), + getUserId: () => userId, + getDeviceId: () => deviceId, + getDomain: () => "example.org", + getSyncState: () => SyncState.Syncing, + _unstable_getRTCTransports: vi.fn().mockResolvedValue([transport]), + } as unknown as MatrixClient; + const members$ = new BehaviorSubject([]); + + const fake: Omit & + Partial> = { + prepared: Promise.withResolvers(), + connectionState$: new BehaviorSubject( + MediaConnectionState.Initialized, + ), + data$: new Subject(), + setPublishing: vi.fn(), + delegateDelayedLeave: vi.fn(async () => Promise.resolve()), + mediaFor: vi.fn(() => ({ + tracks$: constant(null), + encryptionError$: EMPTY, + })), + }; + const backendFactory = ( + backendScope: ObservableScope, + context: MediaBackendContext, + ): MediaBackend => { + const backend: MediaBackend = { + transportType: "livekit", + prepareLocalTransport: async () => fake.prepared.promise, + delegateDelayedLeave: fake.delegateDelayedLeave, + local: { + connectionState$: backendScope.behavior(fake.connectionState$), + setPublishing: fake.setPublishing, + publishError$: constant(null), + publish: vi.fn(), + unpublish: vi.fn(), + tracks$: constant(null), + encryptionError$: EMPTY, + }, + mediaFor: fake.mediaFor, + connections$: constant([]), + setAudioOutputDeviceId: vi.fn(), + sendData: vi.fn(), + data$: fake.data$, + }; + fake.context = context; + fake.backend = backend; + return backend; + }; + + const { participation } = createRTCParticipation( + scope, + { + client, + room, + slot: {} as RTCSlot, + jsSdkSession: mockedSession, + members$: scope.behavior(members$), + transports: { get: metadata, connected$: constant([]) }, + slotOptions: { + matrixRTCMode: MatrixRTCMode.Compatibility, + encryptionSystem: { kind: E2eeType.PER_PARTICIPANT }, + ...slotOptions, + }, + logger, + }, + { publish: [], backend: backendFactory, ...options }, + ); + const states: ParticipationState[] = []; + participation.state$.subscribe((state) => states.push(state)); + return { + scope, + participation, + session, + mockedSession, + join, + members$, + fake: fake as FakeBackend, + states, + }; +} + +describe("createRTCParticipation", () => { + it("joins the session once the transport is prepared, and is connected once the media is", async () => { + const { participation, join, fake, states } = setup(); + expect(states).toEqual([{ kind: "waitingForTransport" }]); + await flushPromises(); + expect(join).not.toHaveBeenCalled(); + + fake.prepared.resolve({ canDelegateDelayedLeave: false }); + await flushPromises(); + expect(join).toHaveBeenCalledWith( + { userId, deviceId, memberId: legacyMemberId }, + [transport], + expect.objectContaining({ + manageMediaKeys: true, + unstableSendStickyEvents: false, + delayedLeaveEventDelayMs: defaultSessionTimings.delayedLeave.delay_ms, + }), + ); + expect(participation.state$.value).toEqual({ kind: "connecting" }); + + fake.connectionState$.next(MediaConnectionState.Connected); + expect(participation.state$.value).toEqual({ kind: "connected" }); + + fake.connectionState$.next(MediaConnectionState.Reconnecting); + expect(participation.state$.value).toEqual({ + kind: "reconnecting", + reason: "media", + }); + }); + + it("publishes only while joined", async () => { + const { participation, session, fake, scope } = setup(); + const ended = vi.fn(); + scope.onEnd(ended); + await flushPromises(); + expect(fake.setPublishing).toHaveBeenLastCalledWith(false); + + fake.prepared.resolve({ canDelegateDelayedLeave: false }); + await flushPromises(); + expect(fake.setPublishing).toHaveBeenLastCalledWith(true); + + participation.leave(); + expect(fake.setPublishing).toHaveBeenLastCalledWith(false); + expect(participation.state$.value).toEqual({ kind: "left" }); + await flushPromises(); + expect(session.leave).toHaveBeenCalled(); + expect(ended).toHaveBeenCalled(); + }); + + it("fails where the transport cannot be prepared", async () => { + const { participation, join, fake } = setup(); + const error = new MatrixRTCError("no token"); + + fake.prepared.reject(error); + await flushPromises(); + + expect(participation.state$.value).toEqual({ kind: "failed", error }); + expect(join).not.toHaveBeenCalled(); + }); + + it("fails where the media connection fails, and stays failed", async () => { + const { participation, fake } = setup(); + fake.prepared.resolve({ canDelegateDelayedLeave: false }); + await flushPromises(); + const error = new MatrixRTCError("SFU gone"); + + fake.connectionState$.next(error); + expect(participation.state$.value).toEqual({ kind: "failed", error }); + + fake.connectionState$.next(MediaConnectionState.Connected); + expect(participation.state$.value).toEqual({ kind: "failed", error }); + }); + + it("joins with the delegated timings where the backend delegates, and hands it each delay id", async () => { + const { session, join, fake } = setup({ + timings: { delegatedDelayedLeave: { delay_ms: 7_200_000 } }, + }); + fake.prepared.resolve({ canDelegateDelayedLeave: true }); + await flushPromises(); + expect(join).toHaveBeenCalledWith( + expect.anything(), + [transport], + expect.objectContaining({ delayedLeaveEventDelayMs: 7_200_000 }), + ); + expect(fake.delegateDelayedLeave).not.toHaveBeenCalled(); + + // Re-emitted by the js-sdk session with the membership manager appended + ( + session as unknown as { emit(event: string, ...args: unknown[]): void } + ).emit(MembershipManagerEvent.DelayIdChanged, "delay-1", {}); + await flushPromises(); + + expect(fake.delegateDelayedLeave).toHaveBeenCalledWith("delay-1"); + }); + + it("names the membership `${userId}:${deviceId}` in compatibility mode and by a uuid in Matrix 2.0", () => { + const compatibility = setup(); + expect(compatibility.fake.context.ownMembershipIdentity).toEqual({ + userId, + deviceId, + memberId: legacyMemberId, + }); + compatibility.scope.end(); + + const matrix2 = setup({}, { matrixRTCMode: MatrixRTCMode.Matrix_2_0 }); + const { memberId } = matrix2.fake.context.ownMembershipIdentity; + expect(memberId).not.toBe(legacyMemberId); + expect(memberId).toMatch(/^[0-9a-f-]{36}$/); + matrix2.scope.end(); + }); + + it("passes the slot and the join options on to the backend", () => { + const { fake, scope } = setup({ + publish: [{ source: "camera" }], + timings: { networkErrorRetryMs: 250 }, + }); + + expect(fake.context).toMatchObject({ + roomId: "!room:example.org", + slotId: "m.call#ROOM", + publish: [{ source: "camera" }], + encryptionSystem: { kind: E2eeType.PER_PARTICIPANT }, + timings: { ...defaultSessionTimings, networkErrorRetryMs: 250 }, + }); + scope.end(); + }); + + it("gives the backend the keys the session holds and the ones that follow", () => { + const { fake, session, mockedSession, scope } = setup(); + const keys: MediaKey[] = []; + fake.context.mediaKeys$.subscribe((key) => keys.push(key)); + expect(mockedSession.reemitEncryptionKeys).toHaveBeenCalledOnce(); + + const key = new Uint8Array([1, 2, 3]); + session.emit( + MatrixRTCSessionEvent.EncryptionKeyChanged, + key, + 2, + { userId: "@bob:example.org", deviceId: "DEV", memberId: "bob-member" }, + "bob-member", + ); + + expect(keys).toEqual([{ participantId: "bob-member", index: 2, key }]); + scope.end(); + }); + + it("builds the remote members on the backend's media and the local member on our own membership", () => { + const { participation, members$, fake, scope } = setup(); + const stale = member({ + userId, + deviceId, + memberId: "old-uuid", + local: true, + }); + + members$.next([own, bob, stale]); + + expect(fake.mediaFor).toHaveBeenCalledTimes(1); + expect(fake.mediaFor).toHaveBeenCalledWith(expect.anything(), bob); + expect(participation.remoteMembers$.value.map((m) => m.memberId)).toEqual([ + "bob-member", + ]); + expect(participation.localMember$.value).toMatchObject({ + local: true, + memberId: legacyMemberId, + }); + scope.end(); + }); + + it("delivers data from a member with its member and drops the rest", () => { + const { participation, members$, fake, scope } = setup(); + members$.next([ + member({ + userId: "@bob:example.org", + memberId: "bob-member", + backendIdentities: ["bob-hashed", "bob-member"], + }), + ]); + const messages: DataMessage[] = []; + participation.data$.subscribe((message) => messages.push(message)); + + fake.data$.next({ senderId: "bob-hashed", topic: "chat", text: "hi" }); + fake.data$.next({ senderId: "stranger", topic: "chat", text: "boo" }); + + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + topic: "chat", + text: "hi", + member: { memberId: "bob-member" }, + }); + scope.end(); + }); + + it("leaves at most once", async () => { + const { participation, session, fake } = setup(); + fake.prepared.resolve({ canDelegateDelayedLeave: false }); + await flushPromises(); + + participation.leave(); + participation.leave(); + await flushPromises(); + + expect(session.leave).toHaveBeenCalledOnce(); + }); +}); diff --git a/sdk/src/media-backend/livekit/LocalMedia.test.ts b/sdk/src/media-backend/livekit/LocalMedia.test.ts new file mode 100644 index 000000000..26ab05bd2 --- /dev/null +++ b/sdk/src/media-backend/livekit/LocalMedia.test.ts @@ -0,0 +1,477 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { + ConnectionState as LivekitConnectionState, + type LocalParticipant, + type LocalTrack, + type LocalTrackPublication, + ParticipantEvent, + type Room as LivekitRoom, + Track, +} from "livekit-client"; +import { logger } from "matrix-js-sdk/lib/logger"; +import { BehaviorSubject } from "rxjs"; +import { describe, expect, it, vi } from "vitest"; + +import { E2eeType } from "../../encryption"; +import { FailToStartLivekitConnection, MatrixRTCError } from "../../errors"; +import { Epoch, type ObservableScope } from "../../reactive/ObservableScope"; +import { flushPromises, mockEmitter, testScope } from "../../utils/test"; +import { MediaConnectionState } from "../api"; +import { type TransportLocator } from "./auth"; +import { type Connection, type ConnectionState } from "./Connection"; +import { ConnectionMap } from "./ConnectionMap"; +import { createLivekitLocalMedia } from "./LocalMedia"; +import { exampleTransport } from "./test"; + +const localTransport: TransportLocator = { + transport: exampleTransport, + serverName: "example.org", +}; +const connected: ConnectionState = { + state: "authenticated", + livekitState: LivekitConnectionState.Connected, +}; + +interface FakeConnection { + connection: Connection; + participant: LocalParticipant; + room: LivekitRoom; + state$: BehaviorSubject; +} + +/** A connection to the local transport with a participant that publishes on demand. */ +function fakeConnection(state: ConnectionState = connected): FakeConnection { + const participant = fakeLocalParticipant(); + const room = { + options: {}, + setE2EEEnabled: vi.fn(), + localParticipant: participant, + remoteParticipants: new Map(), + getActiveDevice: vi.fn(), + switchActiveDevice: vi.fn(async () => Promise.resolve(true)), + ...mockEmitter(), + } as unknown as LivekitRoom; + const state$ = new BehaviorSubject(state); + const connection = { + role: "publisher", + transport: exampleTransport, + serverName: "example.org", + state$, + livekitRoom: room, + } as unknown as Connection; + return { connection, participant, room, state$ }; +} + +function mapOf(...connections: Connection[]): Epoch { + const map = new ConnectionMap(); + for (const connection of connections) map.add(connection, []); + return new Epoch(map); +} + +function setup( + scope: ObservableScope = testScope(), + publish = [{ source: "microphone" as const }], +): { + media: ReturnType; + connectionMap$: BehaviorSubject>; + localTransport$: BehaviorSubject; + audioOutputDeviceId$: BehaviorSubject; +} { + const connectionMap$ = new BehaviorSubject(mapOf()); + const localTransport$ = new BehaviorSubject(null); + const audioOutputDeviceId$ = new BehaviorSubject( + undefined, + ); + const media = createLivekitLocalMedia({ + scope, + connectionMap$: scope.behavior(connectionMap$), + localTransport$: scope.behavior(localTransport$), + publish, + audioOutputDeviceId$, + encryptionSystem: { kind: E2eeType.NONE }, + logger, + }); + return { media, connectionMap$, localTransport$, audioOutputDeviceId$ }; +} + +describe("createLivekitLocalMedia", () => { + it("finds the connection to the local transport once both are known", () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection } = fakeConnection(); + expect(media.connection$.value).toBeNull(); + expect(media.connectionState$.value).toBe(MediaConnectionState.Initialized); + + connectionMap$.next(mapOf(connection)); + expect(media.connection$.value).toBeNull(); + + localTransport$.next(localTransport); + expect(media.connection$.value).toBe(connection); + expect(media.connectionState$.value).toBe(MediaConnectionState.Connected); + }); + + it("follows the connection's state", () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, state$ } = fakeConnection({ state: "authenticating" }); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + expect(media.connectionState$.value).toBe(MediaConnectionState.Preparing); + + const error = new MatrixRTCError("no token"); + state$.next({ state: "error", error }); + expect(media.connectionState$.value).toBe(error); + }); + + it("creates the tracks on the first publish and holds them back until then", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + await flushPromises(); + expect(participant.setMicrophoneEnabled).not.toHaveBeenCalled(); + expect(media.tracks$.value).toEqual([]); + + media.setPublishing(true); + await flushPromises(); + + expect(participant.setMicrophoneEnabled).toHaveBeenCalledOnce(); + expect(media.tracks$.value).toHaveLength(1); + expect(media.tracks$.value![0].source).toBe("microphone"); + expect(track(participant, Track.Source.Microphone).isUpstreamPaused).toBe( + false, + ); + }); + + it("pauses the upstream while the member is not publishing", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + + media.setPublishing(false); + await flushPromises(); + expect(track(participant, Track.Source.Microphone).isUpstreamPaused).toBe( + true, + ); + + media.setPublishing(true); + await flushPromises(); + expect(track(participant, Track.Source.Microphone).isUpstreamPaused).toBe( + false, + ); + }); + + it("resolves a publish made before the connection exists once it is up", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + let published: string | undefined; + const publishing = media + .publish({ source: "camera" }) + .then((t) => (published = t.source)); + await flushPromises(); + expect(published).toBeUndefined(); + + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await publishing; + + expect(published).toBe("camera"); + // Both were requested before the start, so one prompt covers both + expect(participant.enableCameraAndMicrophone).toHaveBeenCalledOnce(); + expect(media.tracks$.value?.map((t) => t.source)).toEqual([ + "microphone", + "camera", + ]); + }); + + it("publishes a request at once on a started publisher", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + + const t = await media.publish({ source: "screenShare" }); + + expect(t.source).toBe("screenShare"); + expect(participant.setScreenShareEnabled).toHaveBeenCalledWith( + true, + expect.objectContaining({ selfBrowserSurface: "include" }), + undefined, + ); + }); + + it("rejects a publish the device refuses, and forgets the request", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + vi.mocked(participant.setCameraEnabled).mockRejectedValueOnce( + new Error("NotAllowedError"), + ); + + const rejection = media.publish({ source: "camera" }); + await expect(rejection).rejects.toBeInstanceOf(MatrixRTCError); + await expect(rejection).rejects.toThrow("NotAllowedError"); + + // A new connection does not try the camera again + const next = fakeConnection(); + connectionMap$.next(mapOf(next.connection)); + await flushPromises(); + expect(next.participant.setCameraEnabled).not.toHaveBeenCalled(); + expect(next.participant.setMicrophoneEnabled).toHaveBeenCalled(); + }); + + it("publishes again on a new connection and stops the old tracks", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const first = fakeConnection(); + connectionMap$.next(mapOf(first.connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + await media.publish({ source: "camera" }); + const firstTracks = [ + track(first.participant, Track.Source.Microphone), + track(first.participant, Track.Source.Camera), + ]; + + const second = fakeConnection(); + connectionMap$.next(mapOf(second.connection)); + await flushPromises(); + + for (const t of firstTracks) + expect(first.participant.unpublishTrack).toHaveBeenCalledWith(t, true); + expect(second.participant.enableCameraAndMicrophone).toHaveBeenCalledOnce(); + expect(media.tracks$.value?.map((t) => t.source)).toEqual([ + "microphone", + "camera", + ]); + expect( + track(second.participant, Track.Source.Camera).isUpstreamPaused, + ).toBe(false); + }); + + it("remembers a mute across connections", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const first = fakeConnection(); + connectionMap$.next(mapOf(first.connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + + await expect(media.tracks$.value![0].setEnabled(false)).resolves.toBe( + false, + ); + expect(first.participant.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + const second = fakeConnection(); + connectionMap$.next(mapOf(second.connection)); + await flushPromises(); + // Muted means not created at all, to spare the permission prompt + expect(second.participant.setMicrophoneEnabled).not.toHaveBeenCalled(); + }); + + it("unpublishes a source, the screen share audio with its video", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + await media.publish({ source: "screenShare" }); + + await media.unpublish("screenShareAudio"); + expect(participant.setScreenShareEnabled).toHaveBeenLastCalledWith(false); + + await media.unpublish("microphone"); + expect(participant.unpublishTrack).toHaveBeenCalledOnce(); + expect(media.tracks$.value).toEqual([]); + }); + + it("keeps the first failure to publish", async () => { + const { media, connectionMap$, localTransport$ } = setup(); + const { connection, participant } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + media.setPublishing(true); + await flushPromises(); + expect(media.publishError$.value).toBeNull(); + + const microphone = track(participant, Track.Source.Microphone); + vi.mocked(microphone.resumeUpstream).mockRejectedValue( + new Error("no upstream"), + ); + media.setPublishing(false); + await flushPromises(); + media.setPublishing(true); + await flushPromises(); + + expect(media.publishError$.value).toBeInstanceOf( + FailToStartLivekitConnection, + ); + expect(media.publishError$.value?.message).toContain("no upstream"); + }); + + it("plays on the device given, on the current room and the rooms to come", async () => { + const { media, connectionMap$, localTransport$, audioOutputDeviceId$ } = + setup(); + await media.setAudioOutputDeviceId("speaker"); + expect(audioOutputDeviceId$.value).toBe("speaker"); + + const { connection, room } = fakeConnection(); + connectionMap$.next(mapOf(connection)); + localTransport$.next(localTransport); + await flushPromises(); + await media.setAudioOutputDeviceId("headset"); + + expect(room.switchActiveDevice).toHaveBeenCalledWith( + "audiooutput", + "headset", + ); + }); +}); + +function track( + participant: LocalParticipant, + source: Track.Source, +): LocalTrack { + const t = participant.getTrackPublication(source)?.track; + if (t === undefined) throw new Error(`No ${source} track`); + return t; +} + +function fakeLocalTrack(source: Track.Source): LocalTrack { + const state = { isMuted: false, isUpstreamPaused: false }; + const t = { + source, + get isMuted() { + return state.isMuted; + }, + get isUpstreamPaused() { + return state.isUpstreamPaused; + }, + attach: vi.fn(), + detach: vi.fn(), + mute: vi.fn(async () => { + state.isMuted = true; + return Promise.resolve(t); + }), + unmute: vi.fn(async () => { + state.isMuted = false; + return Promise.resolve(t); + }), + pauseUpstream: vi.fn(async () => { + state.isUpstreamPaused = true; + return Promise.resolve(); + }), + resumeUpstream: vi.fn(async () => { + state.isUpstreamPaused = false; + return Promise.resolve(); + }), + }; + return t as unknown as LocalTrack; +} + +/** + * The slice of a local participant the publisher and the local tracks use: + * enabling a source publishes a fake track at once, muting an existing one + * mutes it, and the publications map and events are kept as LiveKit would. + */ +function fakeLocalParticipant(): LocalParticipant { + const emitter = mockEmitter(); + const trackPublications = new Map(); + let nextSid = 0; + + const getTrackPublication = ( + source: Track.Source, + ): LocalTrackPublication | undefined => + [...trackPublications.values()].find((p) => p.source === source); + + const publish = (source: Track.Source): LocalTrackPublication => { + const t = fakeLocalTrack(source); + const publication = { + track: t, + source, + kind: + source === Track.Source.Microphone + ? Track.Kind.Audio + : Track.Kind.Video, + trackSid: `TR_${nextSid++}`, + isEncrypted: false, + isSubscribed: true, + get isMuted() { + return t.isMuted; + }, + mute: t.mute, + unmute: t.unmute, + ...mockEmitter(), + } as unknown as LocalTrackPublication; + trackPublications.set(publication.trackSid, publication); + emitter.emit(ParticipantEvent.LocalTrackPublished, publication); + return publication; + }; + const unpublish = (publication: LocalTrackPublication): void => { + trackPublications.delete(publication.trackSid); + emitter.emit(ParticipantEvent.LocalTrackUnpublished, publication); + }; + const setEnabled = async ( + source: Track.Source, + enabled: boolean, + ): Promise => { + const existing = getTrackPublication(source); + if (enabled) { + if (existing) await existing.unmute(); + return existing ?? publish(source); + } + await existing?.mute(); + return existing; + }; + + const participant = { + ...emitter, + identity: "@alice:example.org:DEVICE", + isLocal: true, + isSpeaking: false, + audioLevel: 0, + trackPublications, + getTrackPublication, + setMicrophoneEnabled: vi.fn(async (enabled: boolean) => + setEnabled(Track.Source.Microphone, enabled), + ), + setCameraEnabled: vi.fn(async (enabled: boolean) => + setEnabled(Track.Source.Camera, enabled), + ), + setScreenShareEnabled: vi.fn(async (enabled: boolean) => { + if (enabled) return setEnabled(Track.Source.ScreenShare, true); + const existing = getTrackPublication(Track.Source.ScreenShare); + if (existing) unpublish(existing); + return Promise.resolve(existing); + }), + enableCameraAndMicrophone: vi.fn(async () => { + publish(Track.Source.Microphone); + publish(Track.Source.Camera); + return Promise.resolve(); + }), + unpublishTrack: vi.fn(async (t: LocalTrack) => { + const publication = [...trackPublications.values()].find( + (p) => p.track === t, + ); + if (publication !== undefined) unpublish(publication); + return Promise.resolve(publication); + }), + }; + return participant as unknown as LocalParticipant; +} diff --git a/sdk/src/media-backend/livekit/Publisher.test.ts b/sdk/src/media-backend/livekit/Publisher.test.ts index e0732c150..840ae3ad6 100644 --- a/sdk/src/media-backend/livekit/Publisher.test.ts +++ b/sdk/src/media-backend/livekit/Publisher.test.ts @@ -6,11 +6,21 @@ Please see LICENSE in the repository root for full details. */ import { EventEmitter } from "events"; -import { ConnectionState, type Room as LivekitRoom } from "livekit-client"; +import { + ConnectionState, + LocalParticipant, + type LocalTrack, + type LocalTrackPublication, + ParticipantEvent, + type Room as LivekitRoom, + Track, +} from "livekit-client"; import { logger } from "matrix-js-sdk/lib/logger"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { Publisher } from "./Publisher"; +import { type PublishRequest } from "../../media-api"; +import { flushPromises } from "../../utils/test"; +import { type DesiredMedia, Publisher } from "./Publisher"; describe("Publisher", () => { describe("setAudioOutputDevice", () => { @@ -43,6 +53,261 @@ describe("Publisher", () => { ); }); }); + + // The publisher drives a real LocalParticipant, so that LiveKit's own + // bookkeeping of pending publications is part of what is tested; only the + // device access and the signalling are faked. + describe("publishing", () => { + let participant: LocalParticipant; + let room: LivekitRoom; + let desired: DesiredMedia; + let publisher: Publisher; + // Resolved by default; a test takes hold of it to stretch the track creation + let createTracksLock: Promise; + + const microphone: PublishRequest = { source: "microphone" }; + const camera: PublishRequest = { source: "camera" }; + + function want(...requests: PublishRequest[]): void { + for (const request of requests) + desired.set(request.source, { request, enabled: true }); + } + + function track(source: Track.Source): LocalTrack | undefined { + return participant.getTrackPublication(source)?.track; + } + + beforeEach(() => { + createTracksLock = Promise.resolve(); + participant = fakeLocalParticipant(async () => createTracksLock); + room = { + options: {}, + setE2EEEnabled: () => undefined, + localParticipant: participant, + } as unknown as LivekitRoom; + desired = new Map(); + publisher = new Publisher(room, desired, logger); + }); + + afterEach(async () => { + await publisher.destroy(); + }); + + it("creates no tracks where everything starts muted, to spare the permission prompt", async () => { + want(microphone, camera); + desired.get("microphone")!.enabled = false; + desired.get("camera")!.enabled = false; + + publisher.start(); + await flushPromises(); + + expect(participant.createTracks).not.toHaveBeenCalled(); + }); + + it("asks for the camera and the microphone in one call", async () => { + const enableBoth = vi.spyOn(participant, "enableCameraAndMicrophone"); + want(microphone, camera); + + publisher.start(); + await flushPromises(); + + expect(enableBoth).toHaveBeenCalledOnce(); + expect(participant.createTracks).toHaveBeenCalledOnce(); + expect(participant.createTracks).toHaveBeenCalledWith({ + audio: true, + video: true, + }); + }); + + it.each([ + { enabled: microphone, muted: camera, options: { audio: true } }, + { enabled: camera, muted: microphone, options: { video: true } }, + ])( + "creates only the $enabled.source where the $muted.source is muted", + async ({ enabled, muted, options }) => { + want(enabled, muted); + desired.get(muted.source)!.enabled = false; + + publisher.start(); + await flushPromises(); + + expect(participant.createTracks).toHaveBeenCalledOnce(); + expect(participant.createTracks).toHaveBeenCalledWith(options); + }, + ); + + it("starts again at most once", async () => { + want(microphone); + + publisher.start(); + publisher.start(); + await flushPromises(); + + expect(participant.createTracks).toHaveBeenCalledOnce(); + }); + + it("keeps the upstream paused until the member publishes", async () => { + want(microphone); + + publisher.start(); + await flushPromises(); + + const microphoneTrack = track(Track.Source.Microphone); + expect(microphoneTrack).toBeDefined(); + expect(microphoneTrack!.pauseUpstream).toHaveBeenCalled(); + expect(microphoneTrack!.isUpstreamPaused).toBe(true); + }); + + it("resumes the upstream once the member publishes", async () => { + want(camera); + publisher.start(); + await flushPromises(); + expect(track(Track.Source.Camera)!.isUpstreamPaused).toBe(true); + + await publisher.startPublishing(); + + expect(track(Track.Source.Camera)!.isUpstreamPaused).toBe(false); + }); + + it("pauses the upstream again when the member stops publishing", async () => { + want(camera); + publisher.start(); + await flushPromises(); + await publisher.startPublishing(); + + await publisher.stopPublishing(); + + expect(track(Track.Source.Camera)!.isUpstreamPaused).toBe(true); + }); + + it("leaves a track published while it publishes, paused or not", async () => { + want(microphone); + await publisher.startPublishing(); + + publisher.start(); + await flushPromises(); + + expect(track(Track.Source.Microphone)!.isUpstreamPaused).toBe(false); + }); + + it("mutes and unmutes a track", async () => { + want(microphone); + publisher.start(); + await flushPromises(); + await publisher.startPublishing(); + + await expect(publisher.setEnabled("microphone", false)).resolves.toBe( + false, + ); + expect(track(Track.Source.Microphone)!.isMuted).toBe(true); + + await expect(publisher.setEnabled("microphone", true)).resolves.toBe( + true, + ); + expect(track(Track.Source.Microphone)!.isMuted).toBe(false); + }); + + it("keeps the upstream paused when a track is unmuted before the member publishes", async () => { + want(microphone); + publisher.start(); + await flushPromises(); + await publisher.setEnabled("microphone", false); + + await publisher.setEnabled("microphone", true); + + expect(track(Track.Source.Microphone)!.isUpstreamPaused).toBe(true); + }); + + it("reports the state LiveKit is left in where a mute fails", async () => { + want(camera); + publisher.start(); + await flushPromises(); + vi.spyOn(participant, "setCameraEnabled").mockRejectedValueOnce( + new Error("device gone"), + ); + + await expect(publisher.setEnabled("camera", false)).resolves.toBe(true); + }); + + it("reports a screen share that was never published as muted", async () => { + await expect(publisher.setEnabled("screenShare", true)).resolves.toBe( + false, + ); + }); + + // Creating the tracks takes a while, and the host may mute in the + // meantime: LiveKit's setMicrophoneEnabled(false) does not see a track + // that is still pending, so the publisher has to apply the mute once the + // track is published. + it("mutes a track that was muted while it was being created", async () => { + want(microphone); + const lock = Promise.withResolvers(); + createTracksLock = lock.promise; + + publisher.start(); + await publisher.startPublishing(); + desired.get("microphone")!.enabled = false; + void publisher.setEnabled("microphone", false); + await flushPromises(); + expect(track(Track.Source.Microphone)).toBeUndefined(); + + lock.resolve(); + await flushPromises(); + await flushPromises(); + + const microphoneTrack = track(Track.Source.Microphone); + expect(microphoneTrack).toBeDefined(); + expect(microphoneTrack!.isMuted).toBe(true); + }); + + it("publishes a request on its own once started", async () => { + want(microphone); + publisher.start(); + await flushPromises(); + + const publication = await publisher.publish(camera); + + expect(publication?.source).toBe(Track.Source.Camera); + expect(track(Track.Source.Camera)!.isUpstreamPaused).toBe(true); + }); + + it("unpublishes a source and stops its track", async () => { + want(microphone, camera); + publisher.start(); + await flushPromises(); + const cameraTrack = track(Track.Source.Camera)!; + + await publisher.unpublish("camera"); + + expect(participant.unpublishTrack).toHaveBeenCalledWith( + cameraTrack, + true, + ); + expect(track(Track.Source.Camera)).toBeUndefined(); + expect(track(Track.Source.Microphone)).toBeDefined(); + }); + + it("unpublishes a screen share through LiveKit, so that its audio goes with it", async () => { + await publisher.unpublish("screenShare"); + expect(participant.setScreenShareEnabled).toHaveBeenCalledWith(false); + }); + + it("stops every track on destroy", async () => { + want(microphone, camera); + publisher.start(); + await flushPromises(); + const tracks = [ + track(Track.Source.Microphone), + track(Track.Source.Camera), + ]; + + await publisher.destroy(); + + for (const t of tracks) + expect(participant.unpublishTrack).toHaveBeenCalledWith(t, true); + expect(participant.trackPublications.size).toBe(0); + }); + }); }); /** The slice of a LiveKit room the publisher touches at construction and for the output device. */ @@ -61,3 +326,114 @@ function fakeRoom( } as unknown as LivekitRoom; return { room, switchActiveDevice }; } + +function fakeLocalTrack(source: Track.Source): LocalTrack { + const state = { isMuted: false, isUpstreamPaused: false }; + const track = { + source, + get isMuted() { + return state.isMuted; + }, + get isUpstreamPaused() { + return state.isUpstreamPaused; + }, + stop: vi.fn(), + mute: vi.fn(async () => { + state.isMuted = true; + return Promise.resolve(track); + }), + unmute: vi.fn(async () => { + state.isMuted = false; + return Promise.resolve(track); + }), + pauseUpstream: vi.fn(async () => { + state.isUpstreamPaused = true; + return Promise.resolve(); + }), + resumeUpstream: vi.fn(async () => { + state.isUpstreamPaused = false; + return Promise.resolve(); + }), + }; + return track as unknown as LocalTrack; +} + +/** + * A real LocalParticipant on a fake engine: the track creation, publication + * and unpublication are replaced so that no device or signalling is touched, + * while the enable/disable logic and the pending publications stay LiveKit's. + */ +function fakeLocalParticipant( + createTracksLock: () => Promise, +): LocalParticipant { + const engine = { + client: { sendUpdateLocalMetadata: vi.fn() }, + on: vi.fn().mockReturnThis(), + sendDataPacket: vi.fn(), + }; + const participant = new LocalParticipant( + "local-sid", + "local-identity", + // @ts-expect-error - an engine that answers nothing is enough here + engine, + { + adaptiveStream: true, + dynacast: false, + audioCaptureDefaults: {}, + videoCaptureDefaults: {}, + stopLocalTrackOnUnpublish: true, + reconnectPolicy: "always", + disconnectOnPageLeave: true, + }, + new Map(), + {}, + {}, + {}, + ); + let nextSid = 0; + + vi.spyOn(participant, "createTracks").mockImplementation(async (opts) => { + const tracks: LocalTrack[] = []; + if (opts?.audio) tracks.push(fakeLocalTrack(Track.Source.Microphone)); + if (opts?.video) tracks.push(fakeLocalTrack(Track.Source.Camera)); + await createTracksLock(); + return tracks; + }); + + vi.spyOn(participant, "publishTrack").mockImplementation(async (t) => { + const track = t as LocalTrack; + const publication = { + track, + source: track.source, + trackSid: `TR_${nextSid++}`, + get isMuted() { + return track.isMuted; + }, + mute: track.mute, + unmute: track.unmute, + } as Partial as LocalTrackPublication; + participant.trackPublications.set(publication.trackSid, publication); + participant.emit(ParticipantEvent.LocalTrackPublished, publication); + return Promise.resolve(publication); + }); + + vi.spyOn(participant, "unpublishTrack").mockImplementation(async (t) => { + const publication = [...participant.trackPublications.values()].find( + (p) => p.track === t, + ); + if (publication === undefined) return undefined; + participant.trackPublications.delete(publication.trackSid); + participant.emit(ParticipantEvent.LocalTrackUnpublished, publication); + return Promise.resolve(publication); + }); + + vi.spyOn(participant, "getTrackPublication").mockImplementation((source) => + [...participant.trackPublications.values()].find( + (p) => p.source === source, + ), + ); + + vi.spyOn(participant, "setScreenShareEnabled").mockResolvedValue(undefined); + + return participant; +} diff --git a/sdk/src/utils/test.ts b/sdk/src/utils/test.ts index 5d7c9700f..77c00a1fe 100644 --- a/sdk/src/utils/test.ts +++ b/sdk/src/utils/test.ts @@ -276,6 +276,10 @@ export class MockRTCSession extends TypedEventEmitter< return this.joined; } + public slotId = "m.call#ROOM"; + /** The id of the delayed leave event; undefined until the manager has sent one. */ + public delayId: string | undefined = undefined; + /** The slot event's content; undefined while the room has none. */ public rtcSlot: ReturnType = undefined; public getRtcSlot(): ReturnType {