renames and restructure

- session -> matrixrtc (folder)
 - move MatrixRtcClient to top level of sdk
 - RtcTransport -> Transport
 - backend -> media-backend
This commit is contained in:
Timo K.
2026-10-08 13:53:57 +02:00
parent d0a2ff12f1
commit 5b13e1da81
51 changed files with 96 additions and 85 deletions
+190
View File
@@ -0,0 +1,190 @@
/*
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.
*/
/**
* The seam between the session and whatever carries its media. The session
* speaks MatrixRTC: memberships, transports as they appear in them, the
* delayed leave. A backend speaks one media protocol and hands back the
* media of each member. `backend/livekit/` is the one backend today.
*/
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Observable } from "rxjs";
import {
type LocalAudioMediaTrack,
type LocalMediaInputs,
type LocalMemberMedia,
type LocalVideoMediaTrack,
type MemberMedia,
type PublishRequest,
type ResolvedTransport,
type RTCMembership,
} from "../api";
import { type SessionTimings } from "../config";
import { type EncryptionSystem } from "../encryption";
import { type Behavior } from "../reactive/Behavior";
import { type ObservableScope } from "../reactive/ObservableScope";
export enum MediaConnectionState {
/** No connection attempt yet, or no connection at all. */
Initialized = "initialized",
/** Fetching whatever the connection needs first; the SFU token for LiveKit. */
Preparing = "preparing",
Connecting = "connecting",
Connected = "connected",
Reconnecting = "reconnecting",
Disconnected = "disconnected",
Stopped = "stopped",
}
/** One per-participant media key, as the session's key exchange delivers it. */
export interface MediaKey {
/** The `rtcBackendIdentity` of the member the key belongs to. */
participantId: string;
index: number;
key: Uint8Array<ArrayBuffer>;
}
export interface MediaBackendContext {
roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
localMedia: LocalMediaInputs;
encryptionSystem: EncryptionSystem;
/**
* Every media key the session knows: the current ones replayed on
* subscribe, new ones as they arrive. Empty for a shared key, which is in
* `encryptionSystem`.
*/
mediaKeys$: Observable<MediaKey>;
timings: SessionTimings;
logger: Logger;
}
/**
* The type of the `backend` option: the session calls it once with its scope
* and the context. A backend's own factory keeps the shape of every other
* factory here, `createLivekitBackend(scope, context, options)`, and a host
* closes over its options. The Matrix client is not in the context: a backend
* that needs it takes a `Pick` of exactly what it calls in its own options.
*/
export type MediaBackendFactory = (
scope: ObservableScope,
context: MediaBackendContext,
) => MediaBackend;
/** What preparing the local transport established; what the MatrixRTC join needs from the media side. */
export interface TransportCapabilities {
/**
* Whether the backend can restart the delayed leave on the client's behalf,
* so that a client that vanishes is removed by the backend rather than by
* the timeout. Decides which delayed leave timings the join uses, and
* whether `delegateDelayedLeave` is ever called.
*/
canDelegateDelayedLeave: boolean;
}
/** One connection a backend holds, for diagnostics. */
export interface BackendConnection {
/** The transport as it appears in memberships; what the registry keys on. */
transport: Transport;
state: MediaConnectionState | Error;
/**
* What the backend fetched to connect; undefined until it has, and again
* after the connection stops. One object per fetch, so that a change of
* state does not read as a change of this.
*/
resolved: ResolvedTransport | undefined;
}
export interface DataPacket {
/** The backend identity of the sender. */
senderId: string;
topic: string;
text: string;
}
export interface MediaBackend {
/** The membership transport type this backend serves: "livekit". */
readonly transportType: string;
/**
* Makes `transport` usable as the one the local member publishes on. The
* session found it; the backend does whatever has to succeed before it is
* advertised in the membership. Rejects with the error the session reports
* as fatal. Called once per session.
*/
prepareLocalTransport(transport: Transport): Promise<TransportCapabilities>;
/**
* Hands the delayed leave event to the backend, which restarts it from
* then on. Called each time the membership manager reports a new delay id,
* only when `canDelegateDelayedLeave` is true. A rejection is logged by
* the session, not fatal.
*/
delegateDelayedLeave(delayId: string): Promise<void>;
readonly local: LocalMediaBackend;
/**
* The media of one remote member, null while nothing has arrived for it on
* its transport. Built in `scope`, which the caller ends with the member.
* This is also how the backend learns which remote transports exist: it
* follows the membership's transport for as long as `scope` lives, and
* connects to every transport some member is on.
*/
mediaFor$(
scope: ObservableScope,
membership$: Behavior<RTCMembership>,
): Behavior<MemberMedia | null>;
/** Every connection the backend holds. For debugging and devtool purposes.*/
readonly connections$: Behavior<BackendConnection[]>;
/**
* A text channel beside the media: sent on the local transport, received on every one.
* This could well be just another track kind. Not needed for element call. just for the sdk target.
* Keeping it on the top level for visibility. Should be tackled in a future refactor.
*/
sendData(topic: string, text: string): Promise<void>;
readonly data$: Observable<DataPacket>;
}
export interface LocalMediaBackend {
/**
* State of the connection the local member publishes on. `Initialized`
* while there is none; an `Error` is the connection's failure.
*/
readonly connectionState$: Behavior<MediaConnectionState | Error>;
/** The local member's own media; null until the connection carries a local participant. */
readonly media$: Behavior<LocalMemberMedia | null>;
/**
* Whether the local tracks reach the transport. The session sets it to
* "joined and the homeserver is reachable". Remembered across connections:
* the tracks are created on the first `true`, and their upstream is paused
* while `false`, so the local preview stays live while nothing is heard.
*/
setPublishing(publish: boolean): void;
/** The first failure to publish, kept until the session ends. Not fatal. */
readonly publishError$: Behavior<Error | null>;
/**
* Publishes a source and resolves with its track once it is in `media$`'s
* `tracks$`. Remembered across connections, so a request before the
* transport is up is applied once it is, and republished after a
* reconnection. Rejects where the device could not be used.
*/
publish(
request: PublishRequest,
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack>;
/** Removes one of our tracks by id; a screen share takes its audio with it. */
unpublish(id: string): Promise<void>;
}
@@ -0,0 +1,734 @@
/*
Copyright 2025 Element Creations Ltd.
Copyright 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 {
afterEach,
beforeEach,
describe,
expect,
it,
type MockedObject,
onTestFinished,
vi,
} from "vitest";
import {
type LocalParticipant,
type RemoteParticipant,
type RemoteTrack,
type RemoteTrackPublication,
type Room as LivekitRoom,
RoomEvent,
Track,
ConnectionState as LivekitConnectionState,
} from "livekit-client";
import fetchMock from "fetch-mock";
import EventEmitter from "events";
import { type IOpenIDToken } from "matrix-js-sdk";
import { logger, type Logger } from "matrix-js-sdk/lib/logger";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
Connection,
ConnectionState,
type ConnectionOpts,
} from "./Connection.ts";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type OpenIDClientParts } from "./openIDSFU";
import { MatrixRTCError, FailToGetOpenIdToken } from "../../errors";
import { testJWTToken } from "../../utils/test-fixtures";
import { ownMemberMock } from "../../utils/test";
import { mockRemoteParticipant } from "./test";
let testScope: ObservableScope;
let client: MockedObject<OpenIDClientParts>;
let fakeLivekitRoom: MockedObject<LivekitRoom>;
let localParticipantEventEmiter: EventEmitter;
let fakeLocalParticipant: MockedObject<LocalParticipant>;
const ROOM_ID = "!roomID:example.org";
const livekitFocus: LivekitTransport = {
livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
type: "livekit",
};
function setupTest(): void {
testScope = new ObservableScope();
client = vi.mocked<OpenIDClientParts>({
getOpenIdToken: vi.fn().mockResolvedValue({
access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
token_type: "Bearer",
matrix_server_name: "example.org",
expires_in: 3600,
}),
getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
} as unknown as OpenIDClientParts);
localParticipantEventEmiter = new EventEmitter();
fakeLocalParticipant = vi.mocked<LocalParticipant>({
identity: "@me:example.org",
isMicrophoneEnabled: vi.fn().mockReturnValue(true),
getTrackPublication: vi.fn().mockReturnValue(undefined),
on: localParticipantEventEmiter.on.bind(localParticipantEventEmiter),
off: localParticipantEventEmiter.off.bind(localParticipantEventEmiter),
addListener: localParticipantEventEmiter.addListener.bind(
localParticipantEventEmiter,
),
removeListener: localParticipantEventEmiter.removeListener.bind(
localParticipantEventEmiter,
),
removeAllListeners: localParticipantEventEmiter.removeAllListeners.bind(
localParticipantEventEmiter,
),
} as unknown as LocalParticipant);
const fakeRoomEventEmitter = new EventEmitter();
fakeLivekitRoom = vi.mocked<LivekitRoom>({
connect: vi.fn(),
disconnect: vi.fn(),
remoteParticipants: new Map(),
localParticipant: fakeLocalParticipant,
state: LivekitConnectionState.Disconnected,
on: fakeRoomEventEmitter.on.bind(fakeRoomEventEmitter),
off: fakeRoomEventEmitter.off.bind(fakeRoomEventEmitter),
addListener: fakeRoomEventEmitter.addListener.bind(fakeRoomEventEmitter),
removeListener:
fakeRoomEventEmitter.removeListener.bind(fakeRoomEventEmitter),
removeAllListeners:
fakeRoomEventEmitter.removeAllListeners.bind(fakeRoomEventEmitter),
setE2EEEnabled: vi.fn().mockResolvedValue(undefined),
emit: (eventName: string | symbol, ...args: unknown[]) => {
fakeRoomEventEmitter.emit(eventName, ...args);
},
} as unknown as LivekitRoom);
}
function setupRemoteConnection(): Connection {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
return {
status: 200,
body: {
url: "wss://matrix-rtc.m.localhost/livekit/sfu",
jwt: testJWTToken,
},
};
});
fakeLivekitRoom.connect.mockImplementation(async (): Promise<void> => {
const changeEv = RoomEvent.ConnectionStateChanged;
fakeLivekitRoom.state = LivekitConnectionState.Connecting;
fakeLivekitRoom.emit(changeEv, fakeLivekitRoom.state);
fakeLivekitRoom.state = LivekitConnectionState.Connected;
fakeLivekitRoom.emit(changeEv, fakeLivekitRoom.state);
return Promise.resolve();
});
return new Connection(opts, logger);
}
afterEach(() => {
vi.useRealTimers();
vi.clearAllMocks();
fetchMock.reset();
});
describe("Start connection states", () => {
beforeEach(() => {
fetchMock.post(
`https://matrix-rtc.example.org/livekit/jwt/get_token`,
() => {
return {
// Return a non-retryable error, if not, the retry logic will
// wait and fail the test with a timeout.
status: 404,
};
},
);
});
it("start in initialized state", () => {
setupTest();
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
expect(connection.state$.getValue()).toEqual("Initialized");
});
it("fail to getOpenId token then error state", async () => {
setupTest();
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
capturedStates.push(value);
});
onTestFinished(() => s.unsubscribe());
const deferred = Promise.withResolvers<IOpenIDToken>();
client.getOpenIdToken.mockImplementation(
async (): Promise<IOpenIDToken> => {
return await deferred.promise;
},
);
connection.start().catch(() => {
// expected to throw
});
let capturedState = capturedStates.pop();
expect(capturedState).toBeDefined();
expect(capturedState!).toEqual("FetchingConfig");
deferred.reject(new FailToGetOpenIdToken(new Error("Failed to get token")));
await vi.runAllTimersAsync();
capturedState = capturedStates.pop();
if (capturedState instanceof Error) {
expect(capturedState).toBeInstanceOf(FailToGetOpenIdToken);
expect(connection.transport).toEqual(livekitFocus);
} else {
expect.fail("Expected FailedToStart state but got " + capturedState);
}
});
it("fail to get JWT token and error state", async () => {
setupTest();
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
capturedStates.push(value);
});
onTestFinished(() => s.unsubscribe());
const deferredSFU = Promise.withResolvers<void>();
// mock the /sfu/get call
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, async () => {
await deferredSFU.promise;
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
connection.start().catch(() => {
// expected to throw
});
let capturedState = capturedStates.pop();
expect(capturedState).toBeDefined();
expect(capturedState).toEqual(ConnectionState.FetchingConfig);
deferredSFU.resolve();
await vi.runAllTimersAsync();
capturedState = capturedStates.pop();
if (
capturedState instanceof MatrixRTCError &&
capturedState.cause instanceof Error
) {
expect(capturedState.cause.message).toContain(
"Failed to look up user info from homeserver",
);
expect(connection.transport).toEqual(livekitFocus);
} else {
expect.fail("Expected FailedToStart state but got " + capturedState);
}
});
it("fail to connect to livekit error state", async () => {
setupTest();
vi.useFakeTimers();
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
};
const connection = new Connection(opts, logger);
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
capturedStates.push(value);
});
onTestFinished(() => s.unsubscribe());
const deferredSFU = Promise.withResolvers<void>();
// mock the /sfu/get call
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, () => {
return {
status: 200,
body: {
url: "wss://matrix-rtc.m.localhost/livekit/sfu",
jwt: testJWTToken,
},
};
});
fakeLivekitRoom.connect.mockImplementation(async () => {
await deferredSFU.promise;
throw new Error("Failed to connect to livekit");
});
connection.start().catch(() => {
// expected to throw
});
let capturedState = capturedStates.pop();
expect(capturedState).toBeDefined();
expect(capturedState).toEqual(ConnectionState.FetchingConfig);
deferredSFU.resolve();
await vi.runAllTimersAsync();
capturedState = capturedStates.pop();
if (
capturedState instanceof MatrixRTCError &&
capturedState.cause instanceof Error
) {
expect(capturedState.cause.message).toContain(
"Failed to connect to livekit",
);
expect(connection.transport).toEqual(livekitFocus);
} else {
expect.fail(
"Expected FailedToStart state but got " + JSON.stringify(capturedState),
);
}
});
it("connection states happy path", async () => {
vi.useFakeTimers();
setupTest();
const connection = setupRemoteConnection();
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
capturedStates.push(value);
});
onTestFinished(() => s.unsubscribe());
await connection.start();
await vi.runAllTimersAsync();
const initialState = capturedStates.shift();
expect(initialState).toEqual(ConnectionState.Initialized);
const fetchingState = capturedStates.shift();
expect(fetchingState).toEqual(ConnectionState.FetchingConfig);
const disconnectedState = capturedStates.shift();
expect(disconnectedState).toEqual(ConnectionState.LivekitDisconnected);
const connectingState = capturedStates.shift();
expect(connectingState).toEqual(ConnectionState.LivekitConnecting);
const connectedState = capturedStates.shift();
expect(connectedState).toEqual(ConnectionState.LivekitConnected);
});
it("stopping while connecting does not report an error", async () => {
setupTest();
const connection = setupRemoteConnection();
const capturedStates: (ConnectionState | Error)[] = [];
const s = connection.state$.subscribe((value) => {
capturedStates.push(value);
});
onTestFinished(() => s.unsubscribe());
// livekit-client rejects a pending connect() when disconnect() is called.
const pendingConnect = Promise.withResolvers<void>();
fakeLivekitRoom.connect.mockReturnValue(pendingConnect.promise);
fakeLivekitRoom.disconnect.mockResolvedValue(undefined);
const started = connection.start();
await vi.waitFor(() => expect(fakeLivekitRoom.connect).toHaveBeenCalled());
const stopping = connection.stop();
pendingConnect.reject(new Error("Client initiated disconnect"));
await stopping;
// start() resolves rather than rejecting (it is not awaited by the
// ConnectionManager, so a rejection would be unhandled).
await expect(started).resolves.toBeUndefined();
expect(capturedStates.at(-1)).toEqual(ConnectionState.Stopped);
expect(capturedStates.some((st) => st instanceof Error)).toBe(false);
});
it("shutting down the scope should stop the connection", async () => {
setupTest();
vi.useFakeTimers();
const connection = setupRemoteConnection();
await connection.start();
const stopSpy = vi.spyOn(connection, "stop");
testScope.end();
expect(stopSpy).toHaveBeenCalled();
expect(fakeLivekitRoom.disconnect).toHaveBeenCalled();
});
});
describe("remote participants", () => {
it("emits the list of remote participants", () => {
setupTest();
const connection = setupRemoteConnection();
const observedParticipants: RemoteParticipant[][] = [];
const s = connection.remoteParticipants$.subscribe((participants) => {
observedParticipants.push(participants);
});
onTestFinished(() => s.unsubscribe());
// The remoteParticipants$ observable is derived from the current members of the
// livekitRoom and the rtc membership in order to publish the members that are publishing
// on this connection.
let participants: RemoteParticipant[] = [
mockRemoteParticipant({ identity: "@alice:example.org:DEV000" }),
mockRemoteParticipant({ identity: "@bob:example.org:DEV111" }),
mockRemoteParticipant({ identity: "@carol:example.org:DEV222" }),
// Mock Dan to have no published tracks. We want him to still show show up
// in the participants list.
mockRemoteParticipant({
identity: "@dan:example.org:DEV333",
getTrackPublication: () => undefined,
getTrackPublications: () => [],
}),
];
// Let's simulate 3 members on the livekitRoom
vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockImplementation(
() => new Map(participants.map((p) => [p.identity, p])),
);
participants.forEach((p) =>
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, p),
);
// At this point there should be ~~no~~ publishers
// We do have publisher now, since we do not filter for publishers anymore (to also have participants with only data tracks)
// The filtering we do is just based on the matrixRTC member events.
expect(observedParticipants.pop()!.length).toEqual(4);
participants = [
mockRemoteParticipant({ identity: "@alice:example.org:DEV000" }),
mockRemoteParticipant({ identity: "@bob:example.org:DEV111" }),
mockRemoteParticipant({ identity: "@carol:example.org:DEV222" }),
mockRemoteParticipant({ identity: "@dan:example.org:DEV333" }),
];
participants.forEach((p) =>
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, p),
);
// At this point there should be no publishers
expect(observedParticipants.pop()!.length).toEqual(4);
});
it("should be scoped to parent scope", (): void => {
setupTest();
const connection = setupRemoteConnection();
let observedParticipants: RemoteParticipant[][] = [];
const s = connection.remoteParticipants$.subscribe((participants) => {
observedParticipants.push(participants);
});
onTestFinished(() => s.unsubscribe());
let participants: RemoteParticipant[] = [
mockRemoteParticipant({ identity: "@bob:example.org:DEV111" }),
];
// Let's simulate 3 members on the livekitRoom
vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockImplementation(
() => new Map(participants.map((p) => [p.identity, p])),
);
for (const participant of participants) {
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, participant);
}
// We should have bob as a participant now
const ps = observedParticipants.pop();
expect(ps?.length).toEqual(1);
expect(ps?.[0]?.identity).toEqual("@bob:example.org:DEV111");
// end the parent scope
testScope.end();
observedParticipants = [];
// SHOULD NOT emit any more participants as the scope is ended
participants = participants.filter(
(p) => p.identity !== "@bob:example.org:DEV111",
);
fakeLivekitRoom.emit(
RoomEvent.ParticipantDisconnected,
mockRemoteParticipant({ identity: "@bob:example.org:DEV111" }),
);
expect(observedParticipants.length).toEqual(0);
});
});
describe("remote track logging", () => {
it("logs remote participant and track events on the connection logger", () => {
setupTest();
const info = vi.fn();
const warn = vi.fn();
const testLogger = {
getChild: (): unknown => testLogger,
info,
debug: vi.fn(),
warn,
error: vi.fn(),
} as unknown as Logger;
new Connection(
{
client,
roomId: ROOM_ID,
transport: livekitFocus,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
},
testLogger,
);
info.mockClear(); // drop the constructor log line
const bob = mockRemoteParticipant({
identity: "@bob:example.org:DEV111",
sid: "PA_bob",
});
const pub = {
kind: "audio",
source: "microphone",
trackSid: "TR_mic",
isMuted: false,
isEncrypted: true,
} as unknown as RemoteTrackPublication;
const messages = (): string[] => info.mock.calls.map((c) => c[0] as string);
fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, bob);
fakeLivekitRoom.emit(
RoomEvent.TrackSubscribed,
{} as RemoteTrack,
pub,
bob,
);
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, bob);
fakeLivekitRoom.emit(
RoomEvent.TrackStreamStateChanged,
pub,
Track.StreamState.Paused,
bob,
);
expect(messages()).toEqual([
"Participant connected: @bob:example.org:DEV111 (PA_bob)",
"Subscribed: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true muted=false",
"Muted: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
"Stream paused: audio microphone TR_mic of @bob:example.org:DEV111 encrypted=true",
]);
// Encryption status changes are logged; cryptor errors are warnings
fakeLivekitRoom.emit(
RoomEvent.ParticipantEncryptionStatusChanged,
false,
bob,
);
expect(messages().at(-1)).toBe(
"Encryption status of @bob:example.org:DEV111: encrypted=false",
);
const cryptorError = new Error("missing key at index 3");
fakeLivekitRoom.emit(RoomEvent.EncryptionError, cryptorError, bob);
expect(warn).toHaveBeenCalledWith(
"Encryption error for @bob:example.org:DEV111:",
cryptorError,
);
// Local mute events are already logged by the Publisher
fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, {
...fakeLocalParticipant,
isLocal: true,
} as unknown as LocalParticipant);
expect(messages().filter((m) => m.startsWith("Muted"))).toHaveLength(1);
// Listeners are removed when the scope ends
testScope.end();
fakeLivekitRoom.emit(RoomEvent.ParticipantDisconnected, bob);
expect(
messages().filter((m) => m.startsWith("Participant disconnected")),
).toHaveLength(0);
});
});
//
// NOT USED ANYMORE ?
//
// This setup look like sth for the Publisher. Not a connection.
// describe("PublishConnection", () => {
// // let fakeBlurProcessor: ProcessorWrapper<BackgroundOptions>;
// let roomFactoryMock: Mock<() => LivekitRoom>;
// let muteStates: MockedObject<MuteStates>;
// function setUpPublishConnection(): void {
// setupTest();
// roomFactoryMock = vi.fn().mockReturnValue(fakeLivekitRoom);
// muteStates = mockMuteStates();
// // fakeBlurProcessor = vi.mocked<ProcessorWrapper<BackgroundOptions>>({
// // name: "BackgroundBlur",
// // restart: vi.fn().mockResolvedValue(undefined),
// // setOptions: vi.fn().mockResolvedValue(undefined),
// // getOptions: vi.fn().mockReturnValue({ strength: 0.5 }),
// // isRunning: vi.fn().mockReturnValue(false)
// // });
// }
// describe("Livekit room creation", () => {
// function createSetup(): void {
// setUpPublishConnection();
// const fakeTrackProcessorSubject$ = new BehaviorSubject<ProcessorState>({
// supported: true,
// processor: undefined,
// });
// const opts: ConnectionOpts = {
// client: client,
// transport: livekitFocus,
// scope: testScope,
// livekitRoomFactory: roomFactoryMock,
// };
// const audioInput = {
// available$: of(new Map([["mic1", { id: "mic1" }]])),
// selected$: new BehaviorSubject({ id: "mic1" }),
// select(): void {},
// };
// const videoInput = {
// available$: of(new Map([["cam1", { id: "cam1" }]])),
// selected$: new BehaviorSubject({ id: "cam1" }),
// select(): void {},
// };
// const audioOutput = {
// available$: of(new Map([["speaker", { id: "speaker" }]])),
// selected$: new BehaviorSubject({ id: "speaker" }),
// select(): void {},
// };
// // TODO understand what is wrong with our mocking that requires ts-expect-error
// const fakeDevices = mockMediaDevices({
// // @ts-expect-error Mocking only
// audioInput,
// // @ts-expect-error Mocking only
// videoInput,
// // @ts-expect-error Mocking only
// audioOutput,
// });
// new Connection(
// opts,
// fakeDevices,
// muteStates,
// undefined,
// fakeTrackProcessorSubject$,
// );
// }
// it("should create room with proper initial audio and video settings", () => {
// createSetup();
// expect(roomFactoryMock).toHaveBeenCalled();
// const lastCallArgs =
// roomFactoryMock.mock.calls[roomFactoryMock.mock.calls.length - 1];
// const roomOptions = lastCallArgs.pop() as unknown as RoomOptions;
// expect(roomOptions).toBeDefined();
// expect(roomOptions!.videoCaptureDefaults?.deviceId).toEqual("cam1");
// expect(roomOptions!.audioCaptureDefaults?.deviceId).toEqual("mic1");
// expect(roomOptions!.audioOutput?.deviceId).toEqual("speaker");
// });
// it("respect controlledAudioDevices", () => {
// // TODO: Refactor the code to make it testable.
// // The UrlParams module is a singleton has a cache and is very hard to test.
// // This breaks other tests as well if not handled properly.
// // vi.mock(import("./../UrlParams"), () => {
// // return {
// // getUrlParams: vi.fn().mockReturnValue({
// // controlledAudioDevices: true
// // })
// // };
// // });
// });
// });
// });
+426
View File
@@ -0,0 +1,426 @@
/*
Copyright 2025 Element Creations Ltd.
Copyright 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 {
connectedParticipantsObserver,
connectionStateObserver,
} from "@livekit/components-core";
import {
ConnectionError,
ConnectionErrorReason,
type Participant,
type RemoteParticipant,
type RemoteTrackPublication,
type Room as LivekitRoom,
RoomEvent,
type SubscriptionError,
type Track,
type TrackPublication,
} from "livekit-client";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, map } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import {
getSFUConfigWithOpenID,
type OpenIDClientParts,
type SFUConfig,
} from "./openIDSFU";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import {
MatrixRTCError,
InsufficientCapacityError,
LivekitConnectionError,
PeerConnectionTimeoutError,
SFURoomCreationRestrictedError,
UnknownRTCError,
} from "../../errors";
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;
/** The media transport to connect to. */
transport: LivekitTransport;
/** The Matrix client to use for OpenID and SFU config requests. */
client: OpenIDClientParts;
/** The room ID this connection is associated with. */
roomId: string;
/** The observable scope to use for this connection. */
scope: ObservableScope;
/** Optional factory to create the LiveKit room, mainly for testing purposes. */
livekitRoomFactory: () => LivekitRoom;
}
export enum ConnectionState {
/** The start state of a connection. It has been created but nothing has loaded yet. */
Initialized = "Initialized",
/** `start` has been called on the connection. It aquires the jwt info to conenct to the LK Room */
FetchingConfig = "FetchingConfig",
Stopped = "Stopped",
/** The same as ConnectionState.Disconnected from `livekit-client` */
LivekitDisconnected = "disconnected",
/** The same as ConnectionState.Connecting from `livekit-client` */
LivekitConnecting = "connecting",
/** The same as ConnectionState.Connected from `livekit-client` */
LivekitConnected = "connected",
/** The same as ConnectionState.Reconnecting from `livekit-client` */
LivekitReconnecting = "reconnecting",
/** The same as ConnectionState.SignalReconnecting from `livekit-client` */
LivekitSignalReconnecting = "signalReconnecting",
}
/**
* A connection to a Matrix RTC LiveKit backend.
*
* Expose observables for participants and connection state.
*/
export class Connection {
// Private Behavior
private readonly _state$ = new BehaviorSubject<
ConnectionState | MatrixRTCError
>(ConnectionState.Initialized);
/**
* The current state of the connection to the media transport.
*/
public readonly state$: Behavior<ConnectionState | Error> = this._state$;
/**
* The media transport to connect to.
*/
public readonly transport: LivekitTransport;
public readonly livekitRoom: LivekitRoom;
private scope: ObservableScope;
/**
* The remote LiveKit participants that are visible on this connection.
*
* Note that this may include participants that are connected only to
* subscribe, or publishers that are otherwise unattested in MatrixRTC state.
* It is therefore more low-level than what should be presented to the user.
*/
public readonly remoteParticipants$: Behavior<RemoteParticipant[]>;
/**
* The alias of the LiveKit room.
*/
public get livekitAlias(): string | undefined {
return this._livekitAlias;
}
private _livekitAlias?: string;
/**
* Whether the connection has been stopped.
* @see Connection.stop
* */
protected stopped = false;
// TODO: can we just keep the ConnectionOpts object instead of spreading?
private readonly client: OpenIDClientParts;
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.
*
* @param opts - Connection options {@link ConnectionOpts}.
*
* @param logger - The logger to use.
*/
public constructor(opts: ConnectionOpts, logger: Logger) {
this.ownMembershipIdentity = opts.ownMembershipIdentity;
this.existingSFUConfig = opts.existingSFUConfig;
this.roomId = opts.roomId;
this.logger = logger.getChild(
"[Connection " + opts.transport.livekit_service_url + "]",
);
this.logger.info(
`constructor: ${opts.transport.livekit_service_url} roomId: ${this.roomId} withSfuConfig?: ${opts.existingSFUConfig ? JSON.stringify(opts.existingSFUConfig) : "undefined"}`,
);
const { transport, client, scope } = opts;
this.scope = scope;
this.livekitRoom = opts.livekitRoomFactory();
this.transport = transport;
this.client = client;
this.remoteParticipants$ = scope.behavior(
// Only tracks remote participants
connectedParticipantsObserver(this.livekitRoom),
);
this.logRemoteTrackEvents();
scope.onEnd(() => {
this.logger.info(`Connection scope ended, stopping connection`);
void this.stop();
});
}
/**
* Logs the lifecycle of remote participants and their tracks as seen by
* livekit-client, so that rageshakes show what a tile should have been
* rendering (published, subscribed, muted, paused) for each remote member.
*/
private logRemoteTrackEvents(): void {
const room = this.livekitRoom;
const log = this.logger.getChild("[RemoteTracks]");
// The encryption flag matters: if the publisher encrypts but this client
// believes the track is unencrypted, the cryptor is bypassed and raw
// ciphertext reaches the decoder (audible as loud noise bursts).
const track = (pub: TrackPublication, p: Participant): string =>
`${pub.kind} ${pub.source} ${pub.trackSid} of ${p.identity} encrypted=${pub.isEncrypted}`;
const onParticipantConnected = (p: RemoteParticipant): void =>
log.info(`Participant connected: ${p.identity} (${p.sid})`);
const onParticipantDisconnected = (p: RemoteParticipant): void =>
log.info(`Participant disconnected: ${p.identity} (${p.sid})`);
const onTrackPublished = (
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Published: ${track(pub, p)} muted=${pub.isMuted}`);
const onTrackUnpublished = (
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Unpublished: ${track(pub, p)}`);
const onTrackSubscribed = (
_t: Track,
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Subscribed: ${track(pub, p)} muted=${pub.isMuted}`);
const onTrackUnsubscribed = (
_t: Track,
pub: RemoteTrackPublication,
p: RemoteParticipant,
): void => log.info(`Unsubscribed: ${track(pub, p)}`);
const onTrackSubscriptionFailed = (
trackSid: string,
p: RemoteParticipant,
reason?: SubscriptionError,
): void =>
log.warn(
`Subscription failed: ${trackSid} of ${p.identity} reason=${reason}`,
);
// Mute events also fire for the local participant, which the Publisher
// already logs.
const onTrackMuted = (pub: TrackPublication, p: Participant): void => {
if (!p.isLocal) log.info(`Muted: ${track(pub, p)}`);
};
const onTrackUnmuted = (pub: TrackPublication, p: Participant): void => {
if (!p.isLocal) log.info(`Unmuted: ${track(pub, p)}`);
};
const onTrackStreamStateChanged = (
pub: RemoteTrackPublication,
state: Track.StreamState,
p: RemoteParticipant,
): void => log.info(`Stream ${state}: ${track(pub, p)}`);
const onEncryptionStatusChanged = (
encrypted: boolean,
p?: Participant,
): void =>
log.info(
`Encryption status of ${p?.identity ?? "unknown participant"}: encrypted=${encrypted}`,
);
// livekit-client throttles these per cryptor; they indicate frames being
// dropped (missing/invalid key).
const onEncryptionError = (error: Error, p?: Participant): void =>
log.warn(
`Encryption error for ${p?.identity ?? "unknown participant"}:`,
error,
);
room
.on(RoomEvent.ParticipantConnected, onParticipantConnected)
.on(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
.on(RoomEvent.TrackPublished, onTrackPublished)
.on(RoomEvent.TrackUnpublished, onTrackUnpublished)
.on(RoomEvent.TrackSubscribed, onTrackSubscribed)
.on(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
.on(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
.on(RoomEvent.TrackMuted, onTrackMuted)
.on(RoomEvent.TrackUnmuted, onTrackUnmuted)
.on(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
.on(
RoomEvent.ParticipantEncryptionStatusChanged,
onEncryptionStatusChanged,
)
.on(RoomEvent.EncryptionError, onEncryptionError);
this.scope.onEnd(() => {
room
.off(RoomEvent.ParticipantConnected, onParticipantConnected)
.off(RoomEvent.ParticipantDisconnected, onParticipantDisconnected)
.off(RoomEvent.TrackPublished, onTrackPublished)
.off(RoomEvent.TrackUnpublished, onTrackUnpublished)
.off(RoomEvent.TrackSubscribed, onTrackSubscribed)
.off(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed)
.off(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed)
.off(RoomEvent.TrackMuted, onTrackMuted)
.off(RoomEvent.TrackUnmuted, onTrackUnmuted)
.off(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged)
.off(
RoomEvent.ParticipantEncryptionStatusChanged,
onEncryptionStatusChanged,
)
.off(RoomEvent.EncryptionError, onEncryptionError);
});
}
/**
* Starts the connection.
*
* This will:
* 1. Request an OpenId token `request_token` (allows matrix users to verify their identity with a third-party service.)
* 2. Use this token to request the SFU config to the MatrixRtc authentication service.
* 3. Connect to the configured LiveKit room.
*
* The errors are also represented as a state in the `state$` observable.
* It is safe to ignore those errors and handle them accordingly via the `state$` observable.
* @throws {InsufficientCapacityError} if the LiveKit server indicates that it has insufficient capacity to accept the connection.
* @throws {SFURoomCreationRestrictedError} if the LiveKit server indicates that the room does not exist and cannot be created.
*/
// TODO consider an autostart pattern...
public async start(): Promise<void> {
this.logger.debug("Starting 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());
this.logger.debug(
"Starting Connection to: ",
this.transport.livekit_service_url,
"jwt: ",
jwt,
"wss: ",
url,
"livekitAlias: ",
livekitAlias,
);
this._livekitAlias = livekitAlias;
// If we were stopped while fetching the config, don't proceed to connect
if (this.stopped) return;
// Setup observer once we are done with getSFUConfigWithOpenID
connectionStateObserver(this.livekitRoom)
.pipe(
this.scope.bind(),
map((s) => s as unknown as ConnectionState),
)
.subscribe((lkState) => {
// It is save to cast lkState to ConnectionState as they are fully overlapping.
this._state$.next(lkState);
});
try {
this.logger.info(`livekitRoom.connect ${url}`);
await this.livekitRoom.connect(url, jwt);
this.logger.info(`livekitRoom.connect SUCCESS ${url}`);
} catch (e) {
this.logger.info(`livekitRoom.connect FAILED ${url}`, e);
// LiveKit uses 503 to indicate that the server has hit its track limits.
// https://github.com/livekit/livekit/blob/fcb05e97c5a31812ecf0ca6f7efa57c485cea9fb/pkg/service/rtcservice.go#L171
// It also errors with a status code of 200 (yes, really) for room
// participant limits.
// LiveKit Cloud uses 429 for connection limits.
// Either way, all these errors can be explained as "insufficient capacity".
if (e instanceof ConnectionError) {
if (e.status === 503 || e.status === 200 || e.status === 429) {
throw new InsufficientCapacityError();
}
if (e.status === 404) {
// error msg is "Failed to create call"
// error description is "Call creation might be restricted to authorized users only. Try again later, or contact your server admin if the problem persists."
// The room does not exist. There are two different modes of operation for the SFU:
// - the room is created on the fly when connecting (livekit `auto_create` option)
// - Only authorized users can create rooms, so the room must exist before connecting (done by the auth jwt service)
// In the first case there will not be a 404, so we are in the second case.
throw new SFURoomCreationRestrictedError();
}
if (e.reason === ConnectionErrorReason.Timeout) {
// Unabled to establish peer connection within the timeout
throw new PeerConnectionTimeoutError();
}
throw new LivekitConnectionError(e);
}
throw e;
}
// If we were stopped while connecting, don't proceed to update state.
if (this.stopped) return;
} catch (error) {
if (this.stopped) {
// stop() was called while we were connecting, which makes the pending
// connect reject. That is the abort we asked for, not a failure, so
// don't record an error state on a stopped connection or rethrow it
// (start() is not awaited by the ConnectionManager, so a throw here
// becomes an unhandled promise rejection).
this.logger.debug(`Connect aborted because the connection was stopped`);
return;
}
this.logger.debug(`Failed to connect to LiveKit room: ${error}`);
this._state$.next(
error instanceof MatrixRTCError
? error
: error instanceof Error
? new UnknownRTCError(error)
: new UnknownRTCError(new Error(`${error}`)),
);
// Its okay to ignore the throw. The error is part of the state.
throw error;
}
}
protected async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
// 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 getSFUConfigWithOpenID(
this.client,
this.ownMembershipIdentity,
this.transport.livekit_service_url,
this.roomId,
// dont pass any custom opts for the subscribe only connections
{},
this.logger,
);
}
/**
* Stops the connection.
*
* This will disconnect from the LiveKit room.
* If the connection is already stopped, this is a no-op.
*/
public async stop(): Promise<void> {
this.logger.debug(
`stop: disconnecing from lk room ${this.transport.livekit_service_url}`,
);
if (this.stopped) return;
// Mark as stopped before disconnecting so that a connect() aborted by the
// disconnect sees the flag and does not report the abort as an error.
this.stopped = true;
await this.livekitRoom.disconnect();
this._state$.next(ConnectionState.Stopped);
this.logger.debug(
`stop: DONE disconnecing from lk room ${this.transport.livekit_service_url}`,
);
}
}
@@ -0,0 +1,165 @@
/*
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 BaseKeyProvider,
Room as LivekitRoom,
type RoomOptions,
} from "livekit-client";
// Inline so that the worker also loads when the SDK is served from another
// origin than the page
import E2EEWorker from "livekit-client/e2ee-worker?worker&inline";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { BehaviorSubject, combineLatest, map } from "rxjs";
import { buildLiveKitOptions } from "./livekitOptions";
import { type OpenIDClientParts, type SFUConfig } from "./openIDSFU";
import { type Behavior } from "../../reactive/Behavior";
import { Connection, type ConnectionOpts, ConnectionState } from "./Connection";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { type LocalMediaInputs, type ResolvedTransport } from "../../api";
import { type MediaQuality } from "../../config";
import {
audioCaptureOptions,
type PublishRequestFor,
videoCaptureOptions,
videoPublishOptions,
} from "./publishOptions";
export interface ConnectionFactory {
createConnection(
scope: ObservableScope,
transport: LivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
): Connection;
}
/**
* A connection that keeps what it fetched from the JWT service, so that the
* transport can report it as `resolved$`. Nothing while the config is still
* being fetched, and nothing again once the connection has stopped or failed.
*/
export class ResolvedConnection extends Connection {
private readonly sfuConfig$: BehaviorSubject<SFUConfig | undefined>;
public readonly resolved$: Behavior<ResolvedTransport | undefined>;
public constructor(opts: ConnectionOpts, logger: Logger) {
super(opts, logger);
this.sfuConfig$ = new BehaviorSubject(opts.existingSFUConfig);
// One object per fetch, so that a change of state does not read as a
// change of what was fetched
const resolved$ = this.sfuConfig$.pipe(
map((config) => config && resolvedTransport(config)),
);
this.resolved$ = opts.scope.behavior(
combineLatest([resolved$, this.state$]).pipe(
map(([resolved, state]) =>
state === ConnectionState.Stopped || state instanceof Error
? undefined
: resolved,
),
),
);
}
protected override async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
const config = await super.getSFUConfigForRemoteConnection();
this.sfuConfig$.next(config);
return config;
}
}
function resolvedTransport(config: SFUConfig): ResolvedTransport {
return {
type: "livekit",
url: config.url,
token: config.jwt,
roomAlias: config.livekitAlias,
identity: config.livekitIdentity,
};
}
/**
* Creates a `ResolvedConnection` per transport, each with a LiveKit room of
* its own that captures from the devices the host selected.
*/
export class LivekitConnectionFactory implements ConnectionFactory {
public constructor(
private readonly client: OpenIDClientParts,
private readonly roomId: string,
private readonly localMedia: LocalMediaInputs,
private readonly keyProvider: BaseKeyProvider | undefined,
private readonly mediaQuality: MediaQuality | undefined,
) {}
public createConnection(
scope: ObservableScope,
transport: LivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
): Connection {
return new ResolvedConnection(
{
existingSFUConfig: sfuConfig,
roomId: this.roomId,
transport,
client: this.client,
scope,
livekitRoomFactory: () =>
new LivekitRoom(
roomOptions(this.localMedia, this.keyProvider, this.mediaQuality),
),
ownMembershipIdentity,
},
logger,
);
}
}
/** The initial requests become the room's defaults, so that one permission prompt covers both. */
function roomOptions(
localMedia: LocalMediaInputs,
keyProvider: BaseKeyProvider | undefined,
mediaQuality: MediaQuality | undefined,
): RoomOptions {
const base = buildLiveKitOptions(mediaQuality);
const microphone = initialRequest(localMedia, "microphone");
const camera = initialRequest(localMedia, "camera");
return {
...base,
videoCaptureDefaults: {
...base.videoCaptureDefaults,
...(camera && videoCaptureOptions(camera)),
},
publishDefaults: {
...base.publishDefaults,
...(camera && videoPublishOptions(camera.capture)),
},
audioCaptureDefaults: {
...base.audioCaptureDefaults,
...(microphone && audioCaptureOptions(microphone)),
},
audioOutput: { deviceId: localMedia.audioOutputDeviceId$.value },
// Every room needs a worker of its own: one gets confused by streams from
// several rooms
e2ee: keyProvider && { keyProvider, worker: new E2EEWorker() },
};
}
function initialRequest<S extends "microphone" | "camera">(
localMedia: LocalMediaInputs,
source: S,
): PublishRequestFor<S> | undefined {
return localMedia.publish.find(
(request): request is PublishRequestFor<S> => request.source === source,
);
}
@@ -0,0 +1,362 @@
/*
Copyright 2025 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 { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { BehaviorSubject, NEVER } from "rxjs";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type RemoteParticipant } from "livekit-client";
import { logger } from "matrix-js-sdk/lib/logger";
import {
Epoch,
mapEpoch,
ObservableScope,
} from "../../reactive/ObservableScope";
import {
createConnectionManager$,
ConnectionManagerData,
areLivekitTransportsEqual,
} from "./ConnectionManager.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
import { ownMemberMock, withTestScheduler } from "../../utils/test";
import { type Behavior } from "../../reactive/Behavior";
// Some test constants
const TRANSPORT_1: LivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const TRANSPORT_2: LivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
};
let fakeConnectionFactory: ConnectionFactory;
let testScope: ObservableScope;
// Can be useful to track all created connections in tests, even the disposed ones
let allCreatedConnections: Connection[];
beforeEach(() => {
testScope = new ObservableScope();
allCreatedConnections = [];
fakeConnectionFactory = {} as unknown as ConnectionFactory;
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
const mockConnection = {
transport,
remoteParticipants$: new BehaviorSubject([]),
} as unknown as Connection;
vi.mocked(mockConnection).start = vi.fn();
vi.mocked(mockConnection).stop = vi.fn();
// Tie the connection's lifecycle to the scope to test scope lifecycle management
scope.onEnd(() => {
void mockConnection.stop();
});
allCreatedConnections.push(mockConnection);
return mockConnection;
},
);
});
afterEach(() => {
testScope.end();
});
describe("connections$ stream", () => {
test("Should create and start new connections for each transports", () => {
withTestScheduler(({ behavior, expectObservable }) => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
).toBe("a", {
a: expect.toSatisfy((e: Epoch<Connection[]>) => {
const connections = e.value;
expect(connections.length).toBe(2);
expect(
vi.mocked(fakeConnectionFactory).createConnection,
).toHaveBeenCalledTimes(2);
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalled();
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
expect(conn2!.start).toHaveBeenCalled();
return true;
}),
});
});
});
test("Should start connection only once", () => {
withTestScheduler(({ behavior, expectObservable }) => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
remoteTransports$: behavior("abcdef", {
a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1], 1),
c: new Epoch([TRANSPORT_1], 2),
d: new Epoch([TRANSPORT_1], 3),
e: new Epoch([TRANSPORT_1], 4),
f: new Epoch([TRANSPORT_1, TRANSPORT_2], 5),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
).toBe("xxxxxa", {
x: expect.anything(),
a: expect.toSatisfy((e: Epoch<Connection[]>) => {
const connections = e.value;
expect(connections.length).toBe(2);
expect(
vi.mocked(fakeConnectionFactory).createConnection,
).toHaveBeenCalledTimes(2);
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalledOnce();
return true;
}),
});
});
});
test("Should cleanup connections when not needed anymore", () => {
withTestScheduler(({ behavior, expectObservable }) => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
remoteTransports$: behavior("abc", {
a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
c: new Epoch([TRANSPORT_1], 2),
}),
logger: logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(
connectionManagerData$.pipe(mapEpoch((d) => d.getConnections())),
).toBe("xab", {
x: expect.anything(),
a: expect.toSatisfy((e: Epoch<Connection[]>) => {
const connections = e.value;
expect(connections.length).toBe(2);
return true;
}),
b: expect.toSatisfy((e: Epoch<Connection[]>) => {
const connections = e.value;
expect(connections.length).toBe(1);
// The second connection should have been stopped has it is no longer needed.
const connection2 = allCreatedConnections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(connection2).toBeDefined();
expect(connection2!.stop).toHaveBeenCalled();
// The first connection should still be active
const conn1 = connections[0];
expect(conn1.stop).not.toHaveBeenCalledOnce();
return true;
}),
});
});
});
});
describe("ConnectionManagerData", () => {
test("warns when a second connection to the same URL is merged", () => {
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
const data = new ConnectionManagerData(logger);
const connection = { transport: TRANSPORT_1 } as unknown as Connection;
const p = (identity: string): RemoteParticipant =>
({ identity }) as unknown as RemoteParticipant;
data.add(connection, [p("a")]);
data.add({ ...connection } as Connection, [p("b")]);
expect(warn).toHaveBeenCalledWith(
expect.stringContaining(
"second connection to https://lk.example.org: existing [a], adding [b]",
),
);
expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2);
});
});
describe("connectionManagerData$ stream", () => {
// Used in test to control fake connections' remoteParticipants$ streams
let fakeRemoteParticipantsStreams: Map<string, Behavior<RemoteParticipant[]>>;
function keyForTransport(transport: LivekitTransport): string {
return `${transport.livekit_service_url}`;
}
beforeEach(() => {
fakeRemoteParticipantsStreams = new Map();
function getRemoteParticipantsFor(
transport: LivekitTransport,
): Behavior<RemoteParticipant[]> {
return (
fakeRemoteParticipantsStreams.get(keyForTransport(transport)) ??
new BehaviorSubject([])
);
}
// need a more advanced fake connection factory
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
const fakeRemoteParticipants$ = new BehaviorSubject<
RemoteParticipant[]
>([]);
const mockConnection = {
transport,
remoteParticipants$: getRemoteParticipantsFor(transport),
} as unknown as Connection;
vi.mocked(mockConnection).start = vi.fn();
vi.mocked(mockConnection).stop = vi.fn();
// Tie the connection's lifecycle to the scope to test scope lifecycle management
scope.onEnd(() => {
void mockConnection.stop();
});
fakeRemoteParticipantsStreams.set(
keyForTransport(transport),
fakeRemoteParticipants$,
);
return mockConnection;
},
);
});
test("Should report connections with the remote participants", () => {
withTestScheduler(({ expectObservable, schedule, behavior }) => {
// Setup the fake participants streams behavior
// ==============================
fakeRemoteParticipantsStreams.set(
keyForTransport(TRANSPORT_1),
behavior("oa-b", {
o: [],
a: [{ identity: "user1A" } as RemoteParticipant],
b: [
{ identity: "user1A" } as RemoteParticipant,
{ identity: "user1B" } as RemoteParticipant,
],
}),
);
fakeRemoteParticipantsStreams.set(
keyForTransport(TRANSPORT_2),
behavior("o-a", {
o: [],
a: [{ identity: "user2A" } as RemoteParticipant],
}),
);
// ==============================
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
logger,
ownMembershipIdentity: ownMemberMock,
});
expectObservable(connectionManagerData$).toBe("abcd", {
a: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(0);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
return true;
}),
b: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(0);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
return true;
}),
c: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(1);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
expect(
data.getParticipantsForTransport(TRANSPORT_2)[0].identity,
).toBe("user2A");
return true;
}),
d: expect.toSatisfy((e) => {
const data: ConnectionManagerData = e.value;
expect(data.getConnections().length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_1).length).toBe(2);
expect(data.getParticipantsForTransport(TRANSPORT_2).length).toBe(1);
expect(
data.getParticipantsForTransport(TRANSPORT_1)[0].identity,
).toBe("user1A");
expect(
data.getParticipantsForTransport(TRANSPORT_1)[1].identity,
).toBe("user1B");
expect(
data.getParticipantsForTransport(TRANSPORT_2)[0].identity,
).toBe("user2A");
return true;
}),
});
});
});
});
@@ -0,0 +1,297 @@
/*
Copyright 2025 Element Creations Ltd.
Copyright 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 { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
combineLatest,
map,
type Observable,
of,
switchMap,
startWith,
catchError,
NEVER,
} 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 "../../reactive/Behavior";
import { type Connection } from "./Connection.ts";
import { Epoch, type ObservableScope } from "../../reactive/ObservableScope";
import { generateItemsWithEpoch } from "../../reactive/observable";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type SFUConfig } from "./openIDSFU";
export class ConnectionManagerData {
private readonly store: Map<
string,
{ connection: Connection; participants: RemoteParticipant[] }
> = new Map();
public constructor(private readonly logger?: Logger) {}
public add(connection: Connection, participants: RemoteParticipant[]): void {
const key = this.getKey(connection.transport);
const existing = this.store.get(key);
if (!existing) {
this.store.set(key, { connection, participants });
} else {
// Transports are deduplicated by URL upstream, so this should never
// happen; if it does, members may be matched against the wrong room.
this.logger?.warn(
`Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`,
);
existing.participants.push(...participants);
}
}
private getKey(transport: LivekitTransport): string {
// This is enough as a key because the ConnectionManager is already scoped by room.
// We also do not need to consider the slotId at this point since each `MatrixRTCSession` is already scoped by `slotDescription: {id, application}`.
return transport.livekit_service_url;
}
public getConnections(): Connection[] {
return Array.from(this.store.values()).map(({ connection }) => connection);
}
public getConnectionForTransport(
transport: LivekitTransport,
): Connection | null {
return this.store.get(this.getKey(transport))?.connection ?? null;
}
public getParticipantsForTransport(
transport: LivekitTransport,
): RemoteParticipant[] {
const key = this.getKey(transport);
const existing = this.store.get(key);
if (existing) {
return existing.participants;
}
return [];
}
}
interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
localTransport$: Observable<LocalTransport>;
remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
logger: Logger;
ownMembershipIdentity: CallMembershipIdentityParts;
}
// TODO - write test for scopes (do we really need to bind scope)
export interface IConnectionManager {
connectionManagerData$: Behavior<Epoch<ConnectionManagerData>>;
}
/**
* Crete a `ConnectionManager`
* @param props - Configuration object
* @param props.scope - The observable scope used by this object
* @param props.connectionFactory - Used to create new connections
* @param props.localTransport$ - The transport to publish local media on. (deduplicated with remoteTransports$)
* @param props.remoteTransports$ - All other transports. The connection manager will create connections for each transport. (deduplicated with localTransport$)
* @param props.ownMembershipIdentity - The own membership identity to use.
* @param props.logger - The logger to use.
*
* Each of these behaviors can be interpreted as subscribed list of transports.
*
* Using `registerTransports` independent external modules can control what connections
* are created by the ConnectionManager.
*
* The connection manager will remove all duplicate transports in each subscibed list.
*
* See `unregisterAllTransports` and `unregisterTransport` for details on how to unsubscribe.
*/
export function createConnectionManager$({
scope,
connectionFactory,
localTransport$,
remoteTransports$,
logger: parentLogger,
ownMembershipIdentity,
}: Props): IConnectionManager {
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()`.
*/
const localAndRemoteTransports$: Behavior<
Epoch<(LivekitTransport | 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) =>
areLivekitTransportsEqual(l.transport, transport),
),
);
logger.debug(
"remoteWithoutLocal",
remoteWithoutLocal,
"localTransportAsArray",
localTransportAsArray,
);
return new Epoch(
[...localTransportAsArray, ...remoteWithoutLocal],
remoteTransports.epoch,
);
}),
),
);
/**
* Connections for each transport in use by one or more session members.
*/
const connections$ = scope.behavior(
localAndRemoteTransports$.pipe(
generateItemsWithEpoch(
"ConnectionManager connections$",
function* (transports) {
for (const transport of transports) {
if (isLocalTransport(transport)) {
// This is the local transport; only the `LocalTransport`
// interface has a `sfuConfig` field.
yield {
keys: [
transport.transport.livekit_service_url,
transport.sfuConfig,
],
data: undefined,
};
} else {
yield {
keys: [
transport.livekit_service_url,
undefined as SFUConfig | undefined,
],
data: undefined,
};
}
}
},
(scope, _data$, serviceUrl, sfuConfig) => {
const connection = connectionFactory.createConnection(
scope,
{
type: "livekit",
livekit_service_url: serviceUrl,
},
ownMembershipIdentity,
logger,
// TODO: This whole optional SFUConfig parameter is not particularly elegant.
// I would like it if connections always fetched the SFUConfig by themselves.
sfuConfig,
);
// Start the connection immediately
// Use connection state to track connection progress
void connection.start();
// TODO subscribe to connection state to retry or log issues?
return connection;
},
),
),
);
const connectionManagerData$ = scope.behavior(
connections$.pipe(
switchMap((connections) => {
const epoch = connections.epoch;
// Map the connections to list of {connection, participants}[]
const listOfConnectionsWithRemoteParticipants = connections.value.map(
(connection) => {
return connection.remoteParticipants$.pipe(
map((participants) => ({
connection,
participants,
})),
);
},
);
// probably not required
if (listOfConnectionsWithRemoteParticipants.length === 0) {
return of(new Epoch(new ConnectionManagerData(), epoch));
}
// combineLatest the several streams into a single stream with the ConnectionManagerData
return combineLatest(listOfConnectionsWithRemoteParticipants).pipe(
map(
(lists) =>
new Epoch(
lists.reduce((data, { connection, participants }) => {
data.add(connection, participants);
return data;
}, new ConnectionManagerData(logger)),
epoch,
),
),
);
}),
),
new Epoch(new ConnectionManagerData(), -1),
);
return { connectionManagerData$ };
}
function removeDuplicateTransports<T extends LivekitTransport>(
transports: T[],
): T[] {
return transports.reduce((acc, transport) => {
if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
acc.push(transport);
return acc;
}, [] as T[]);
}
/** The transport the local member publishes on, authenticated with. */
export interface LocalTransport {
transport: LivekitTransport;
sfuConfig: SFUConfig;
}
export function isLocalTransport(
obj: LivekitTransport | LocalTransport,
): obj is LocalTransport {
return "transport" in obj && "sfuConfig" in obj;
}
export function areLivekitTransportsEqual<T extends LivekitTransport>(
t1: T | null,
t2: T | null,
): boolean {
if (t1 && t2) return t1.livekit_service_url === t2.livekit_service_url;
return !t1 && !t2;
}
@@ -0,0 +1,134 @@
/*
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 { EventEmitter } from "events";
import { type RemoteParticipant, RoomEvent } from "livekit-client";
import { BehaviorSubject } from "rxjs";
import { describe, expect, it, vi } from "vitest";
import { Epoch } from "../../reactive/ObservableScope";
import { constant } from "../../reactive/Behavior";
import { type DataPacket } from "../api";
import { type Connection, ConnectionState } from "./Connection";
import { ConnectionManagerData } from "./ConnectionManager";
import { createDataChannel$ } from "./DataChannel";
import { testScope } from "../../utils/test";
import { exampleTransport } from "./test";
const encode = (text: string): Uint8Array => new TextEncoder().encode(text);
function fakeConnection(
url: string,
state = ConnectionState.LivekitConnected,
): {
connection: Connection;
room: EventEmitter;
publishData: ReturnType<typeof vi.fn>;
} {
const room = new EventEmitter();
const publishData = vi.fn(async () => Promise.resolve());
const connection = {
transport: { ...exampleTransport, livekit_service_url: url },
state$: constant(state),
livekitRoom: Object.assign(room, { localParticipant: { publishData } }),
} as unknown as Connection;
return { connection, room, publishData };
}
function setup(connections: Connection[]): {
received: DataPacket[];
connections$: BehaviorSubject<Epoch<ConnectionManagerData>>;
sendData: (topic: string, text: string) => Promise<void>;
} {
const data = new ConnectionManagerData();
for (const connection of connections) data.add(connection, []);
const connections$ = new BehaviorSubject(new Epoch(data));
const { data$, sendData } = createDataChannel$({
scope: testScope(),
connectionManager: { connectionManagerData$: connections$ },
connection$: constant(connections[0] ?? null),
});
const received: DataPacket[] = [];
data$.subscribe((message) => received.push(message));
return { received, connections$, sendData };
}
describe("data$", () => {
it("delivers a packet with its sender, topic and text", () => {
const { connection, room } = fakeConnection("https://sfu");
const { received } = setup([connection]);
room.emit(
RoomEvent.DataReceived,
encode("hello"),
{ identity: "alice" } as RemoteParticipant,
undefined,
"chat",
);
expect(received).toEqual([
{ senderId: "alice", topic: "chat", text: "hello" },
]);
});
it("drops packets from the server", () => {
const { connection, room } = fakeConnection("https://sfu");
const { received } = setup([connection]);
room.emit(RoomEvent.DataReceived, encode("server"), undefined);
expect(received).toEqual([]);
});
it("listens on every connection, including ones added later", () => {
const first = fakeConnection("https://sfu-1");
const second = fakeConnection("https://sfu-2");
const { received, connections$ } = setup([first.connection]);
const data = new ConnectionManagerData();
data.add(first.connection, []);
data.add(second.connection, []);
connections$.next(new Epoch(data, 1));
first.room.emit(RoomEvent.DataReceived, encode("one"), {
identity: "alice",
} as RemoteParticipant);
second.room.emit(RoomEvent.DataReceived, encode("two"), {
identity: "bob",
} as RemoteParticipant);
expect(received.map((m) => [m.senderId, m.text])).toEqual([
["alice", "one"],
["bob", "two"],
]);
});
});
describe("sendData", () => {
it("publishes the text as a reliable packet on the local connection", async () => {
const { connection, publishData } = fakeConnection("https://sfu");
const { sendData } = setup([connection]);
await sendData("chat", "hello");
expect(publishData).toHaveBeenCalledWith(encode("hello"), {
reliable: true,
topic: "chat",
});
});
it("rejects while the local transport is not connected", async () => {
const { connection, publishData } = fakeConnection(
"https://sfu",
ConnectionState.Initialized,
);
const { sendData } = setup([connection]);
await expect(sendData("chat", "hello")).rejects.toThrow("not connected");
expect(publishData).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,110 @@
/*
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 RemoteParticipant,
type Room as LivekitRoom,
RoomEvent,
} from "livekit-client";
import {
distinctUntilChanged,
filter,
fromEvent,
map,
merge,
type Observable,
share,
switchMap,
} from "rxjs";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { type DataPacket } from "../api";
import { type Connection, ConnectionState } from "./Connection";
import { type IConnectionManager } from "./ConnectionManager";
interface Props {
scope: ObservableScope;
connectionManager: IConnectionManager;
/** The local member's connection, which carries what we send. */
connection$: Behavior<Connection | null>;
}
export interface DataChannel {
data$: Observable<DataPacket>;
sendData: (topic: string, text: string) => Promise<void>;
}
/**
* A text channel beside the media: LiveKit's reliable data packets, received
* on every connection the client holds and sent on the local member's. The
* sender is reported by its identity; the packet's own claim of who sent it
* never reaches the host.
*/
export function createDataChannel$({
scope,
connectionManager,
connection$,
}: Props): DataChannel {
const decoder = new TextDecoder();
const encoder = new TextEncoder();
const rooms$ = connectionManager.connectionManagerData$.pipe(
map(({ value }) => value.getConnections().map((c) => c.livekitRoom)),
distinctUntilChanged(
(a, b) => a.length === b.length && a.every((room, i) => room === b[i]),
),
);
const data$ = rooms$.pipe(
switchMap((rooms) => merge(...rooms.map(receivedOn))),
map(({ payload, participant, topic }): DataPacket | null =>
// Only the server sends without a participant; a host has no use for it
participant === undefined
? null
: {
senderId: participant.identity,
topic: topic ?? "",
text: decoder.decode(payload),
},
),
filter((packet) => packet !== null),
scope.bind(),
share(),
);
const sendData = async (topic: string, text: string): Promise<void> => {
const connection = connection$.value;
if (connection?.state$.value !== ConnectionState.LivekitConnected)
throw new Error("Cannot send data: the local transport is not connected");
await connection.livekitRoom.localParticipant.publishData(
encoder.encode(text),
{ reliable: true, topic },
);
};
return { data$, sendData };
}
interface Received {
payload: Uint8Array;
participant: RemoteParticipant | undefined;
topic: string | undefined;
}
function receivedOn(room: LivekitRoom): Observable<Received> {
return fromEvent(
room,
RoomEvent.DataReceived,
(
payload: Uint8Array,
participant?: RemoteParticipant,
_kind?: unknown,
topic?: string,
): Received => ({ payload, participant, topic }),
);
}
@@ -0,0 +1,38 @@
/*
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 BaseKeyProvider, ExternalE2EEKeyProvider } from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type Observable } from "rxjs";
import { E2eeType, type EncryptionSystem } from "../../encryption";
import { type MediaKey } from "../api";
import { MatrixKeyProvider } from "./MatrixKeyProvider";
/** The LiveKit key provider for the encryption the host asked for, or none. */
export function createKeyProvider(
encryptionSystem: EncryptionSystem,
mediaKeys$: Observable<MediaKey>,
logger: Logger,
): BaseKeyProvider | undefined {
switch (encryptionSystem.kind) {
case E2eeType.NONE:
return undefined;
case E2eeType.PER_PARTICIPANT: {
const keyProvider = new MatrixKeyProvider();
keyProvider.setMediaKeys(mediaKeys$);
return keyProvider;
}
case E2eeType.SHARED_KEY: {
const keyProvider = new ExternalE2EEKeyProvider();
keyProvider
.setKey(encryptionSystem.secret)
.catch((e) => logger.error("Failed to set the shared E2EE key", e));
return keyProvider;
}
}
}
@@ -0,0 +1,112 @@
/*
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 fetchMock from "fetch-mock";
import { logger } from "matrix-js-sdk/lib/logger";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject } from "rxjs";
import { afterEach, describe, expect, it } from "vitest";
import { ObservableScope } from "../../reactive/ObservableScope";
import { testScope } from "../../utils/test";
import { MediaConnectionState } from "../api";
import { ConnectionState } from "./Connection";
import { toMediaConnectionState } from "./connectionState";
import { createTransportRegistry, supportsDelegation } from "./LivekitBackend";
import { exampleTransport } from "./test";
const otherTransport: LivekitTransport = {
type: "livekit",
livekit_service_url: "https://other.example.org",
};
describe("createTransportRegistry", () => {
it("connects to a transport once however many members are on it, and drops it with the last", () => {
const registry = createTransportRegistry(testScope());
const urls = (): string[] =>
registry.transports$.value.value.map((t) => t.livekit_service_url);
const alice = new ObservableScope();
const bob = new ObservableScope();
registry.follow(alice, new BehaviorSubject(exampleTransport));
registry.follow(bob, new BehaviorSubject(exampleTransport));
expect(urls()).toEqual([exampleTransport.livekit_service_url]);
alice.end();
expect(urls()).toEqual([exampleTransport.livekit_service_url]);
bob.end();
expect(urls()).toEqual([]);
});
it("follows a member that changes transport", () => {
const registry = createTransportRegistry(testScope());
const transport$ = new BehaviorSubject<LivekitTransport | undefined>(
exampleTransport,
);
registry.follow(testScope(), transport$);
transport$.next(otherTransport);
expect(registry.transports$.value.value).toEqual([otherTransport]);
transport$.next(undefined);
expect(registry.transports$.value.value).toEqual([]);
});
it("is safe to release after the registry's own scope has ended", () => {
const scope = new ObservableScope();
const registry = createTransportRegistry(scope);
const member = new ObservableScope();
registry.follow(member, new BehaviorSubject(exampleTransport));
scope.end();
expect(() => member.end()).not.toThrow();
});
});
describe("supportsDelegation", () => {
afterEach(() => fetchMock.reset());
const homeserver = "https://matrix.example.org";
const homeserverEndpoint = `${homeserver}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`;
const transportEndpoint = `${exampleTransport.livekit_service_url}/delegate_delayed_leave`;
it.each([
["the homeserver supports it", 400, 404, true],
["only the transport supports it", 404, 400, true],
["neither supports it", 404, 404, false],
])("%s", async (_name, homeserverStatus, transportStatus, expected) => {
fetchMock.post(homeserverEndpoint, homeserverStatus);
fetchMock.post(transportEndpoint, transportStatus);
await expect(
supportsDelegation(homeserver, exampleTransport, logger),
).resolves.toBe(expected);
});
it("assumes no support where the probe itself fails", async () => {
fetchMock.post(homeserverEndpoint, { throws: new TypeError("CORS") });
fetchMock.post(transportEndpoint, { throws: new TypeError("CORS") });
await expect(
supportsDelegation(homeserver, exampleTransport, logger),
).resolves.toBe(false);
});
});
describe("toMediaConnectionState", () => {
it("maps every LiveKit state and passes an error through", () => {
expect(toMediaConnectionState(ConnectionState.FetchingConfig)).toBe(
MediaConnectionState.Preparing,
);
expect(toMediaConnectionState(ConnectionState.LivekitConnected)).toBe(
MediaConnectionState.Connected,
);
expect(
toMediaConnectionState(ConnectionState.LivekitSignalReconnecting),
).toBe(MediaConnectionState.Reconnecting);
const error = new Error("gone");
expect(toMediaConnectionState(error)).toBe(error);
});
});
@@ -0,0 +1,357 @@
/*
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 } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
isLivekitTransport,
type LivekitTransport,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import {
BehaviorSubject,
combineLatest,
distinctUntilChanged,
map,
type Observable,
of,
ReplaySubject,
switchMap,
} from "rxjs";
import {
type MemberMedia,
type ResolvedTransport,
type RTCMembership,
} from "../../api";
import { type MediaQuality } from "../../config";
import { MatrixRTCError } from "../../errors";
import { type Behavior } from "../../reactive/Behavior";
import {
type Epoch,
type ObservableScope,
trackEpoch,
} from "../../reactive/ObservableScope";
import { mapScoped } from "../../utils/mapScoped";
import {
type BackendConnection,
type MediaBackend,
type MediaBackendContext,
type TransportCapabilities,
} from "../api";
import { type Connection } from "./Connection";
import {
LivekitConnectionFactory,
ResolvedConnection,
} from "./ConnectionFactory";
import {
createConnectionManager$,
type LocalTransport,
} from "./ConnectionManager";
import { toMediaConnectionState } from "./connectionState";
import { createDataChannel$ } from "./DataChannel";
import { createKeyProvider } from "./KeyProvider";
import { createLivekitMemberMedia } from "./LivekitMemberMedia";
import { createLivekitLocalMedia } from "./LocalMedia";
import { getSFUConfigWithOpenID, type TokenEndpoint } from "./openIDSFU";
export interface LivekitBackendOptions {
/** `getOpenIdToken` for the token exchange, `baseUrl` for the delegation probe and handover. */
client: Pick<MatrixClient, "getOpenIdToken" | "baseUrl">;
mediaQuality?: MediaQuality;
/** Which token service endpoint the local transport uses; remote ones try both. */
tokenEndpoint: TokenEndpoint;
}
/**
* MatrixRTC media over LiveKit SFUs: one room per transport, the local one
* published on, every remote one subscribed to, with the token service
* protocol in front of each.
*/
export function createLivekitBackend(
scope: ObservableScope,
{
roomId,
ownMembershipIdentity,
localMedia,
encryptionSystem,
mediaKeys$,
timings,
logger: parentLogger,
}: MediaBackendContext,
{ client, mediaQuality, tokenEndpoint }: LivekitBackendOptions,
): MediaBackend {
const logger = parentLogger.getChild("[LivekitBackend]");
const keyProvider = createKeyProvider(encryptionSystem, mediaKeys$, logger);
const transports = createTransportRegistry(scope);
const localTransport$ = new ReplaySubject<LocalTransport>(1);
const connectionManager = createConnectionManager$({
scope,
connectionFactory: new LivekitConnectionFactory(
client,
roomId,
localMedia,
keyProvider,
mediaQuality,
),
localTransport$,
remoteTransports$: transports.transports$,
logger,
ownMembershipIdentity,
});
let local: LivekitTransport | undefined;
const prepareLocalTransport = async (
transport: Transport,
): Promise<TransportCapabilities> => {
if (!isLivekitTransport(transport))
throw new MatrixRTCError(`Not a LiveKit transport: ${transport.type}`);
const [sfuConfig, canDelegateDelayedLeave] = await Promise.all([
getSFUConfigWithOpenID(
client,
ownMembershipIdentity,
transport.livekit_service_url,
roomId,
{ tokenEndpoint },
logger,
),
supportsDelegation(client.baseUrl, transport, logger),
]);
local = transport;
localTransport$.next({ transport, sfuConfig });
return { canDelegateDelayedLeave };
};
// The token this issues is discarded; the request is what hands the leave over
const delegateDelayedLeave = async (delayId: string): Promise<void> => {
if (local === undefined)
throw new MatrixRTCError("No local transport to delegate the leave to");
await getSFUConfigWithOpenID(
client,
ownMembershipIdentity,
local.livekit_service_url,
roomId,
{
tokenEndpoint,
delayEndpointBaseUrl: client.baseUrl,
delayId,
delayTimeoutMs: timings.delegatedDelayedLeave.delay_ms,
},
logger,
);
};
const mediaFor$ = (
memberScope: ObservableScope,
membership$: Behavior<RTCMembership>,
): Behavior<MemberMedia | null> => {
const identity = membership$.value.rtcBackendIdentity;
const transport$ = memberScope.behavior(
membership$.pipe(
map((membership) => membership.getTransport()),
map((transport) =>
isLivekitTransport(transport) ? transport : undefined,
),
distinctUntilChanged(
(a, b) => a?.livekit_service_url === b?.livekit_service_url,
),
),
);
transports.follow(memberScope, transport$);
const source$ = memberScope.behavior(
combineLatest([
transport$,
connectionManager.connectionManagerData$,
]).pipe(
map(([transport, { value: connections }]) => {
if (transport === undefined) return null;
const connection = connections.getConnectionForTransport(transport);
const participant = connections
.getParticipantsForTransport(transport)
.find((p) => p.identity === identity);
return connection && participant
? { participant, room: connection.livekitRoom }
: null;
}),
distinctUntilChanged(
(a, b) => a?.participant === b?.participant && a?.room === b?.room,
),
),
);
source$.pipe(memberScope.bind()).subscribe((source) => {
logger.info(
`${identity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"}`,
);
});
return memberScope.behavior(
mapScoped(memberScope, source$, (mediaScope, { participant, room }) =>
createLivekitMemberMedia(
mediaScope,
participant,
room,
encryptionSystem,
),
).pipe(map((media) => media ?? null)),
);
};
const connections$ = scope.behavior<BackendConnection[]>(
connectionManager.connectionManagerData$.pipe(
switchMap(({ value }) => {
const connections = value.getConnections();
if (connections.length === 0) return of([]);
return combineLatest(connections.map(backendConnection$));
}),
),
);
const localMedia_ = createLivekitLocalMedia({
scope,
connectionManager,
localTransport$,
localMedia,
encryptionSystem,
logger,
});
const { data$, sendData } = createDataChannel$({
scope,
connectionManager,
connection$: localMedia_.connection$,
});
return {
transportType: "livekit",
prepareLocalTransport,
delegateDelayedLeave,
local: localMedia_,
mediaFor$,
connections$,
sendData,
data$,
};
}
function backendConnection$(
connection: Connection,
): Observable<BackendConnection> {
const resolved$: Observable<ResolvedTransport | undefined> =
connection instanceof ResolvedConnection
? connection.resolved$
: of(undefined);
return combineLatest([connection.state$, resolved$]).pipe(
map(([state, resolved]): BackendConnection => ({
transport: connection.transport,
state: toMediaConnectionState(state),
resolved,
})),
);
}
export interface TransportRegistry {
/** Keeps a member's transport registered for as long as its scope lives. */
follow(
scope: ObservableScope,
transport$: Behavior<LivekitTransport | undefined>,
): void;
/** Every transport some member is on, for the connection manager. */
transports$: Behavior<Epoch<LivekitTransport[]>>;
}
/**
* The remote transports to connect to, counted per member so that a transport
* stays while any member names it. A release after the scope has ended only
* touches the map, so it is safe in any order of teardown.
*/
export function createTransportRegistry(
scope: ObservableScope,
): TransportRegistry {
const counts = new Map<
string,
{ transport: LivekitTransport; count: number }
>();
const transports = new BehaviorSubject<LivekitTransport[]>([]);
const publish = (): void =>
transports.next([...counts.values()].map(({ transport }) => transport));
const register = (transport: LivekitTransport): (() => void) => {
const url = transport.livekit_service_url;
const entry = counts.get(url) ?? { transport, count: 0 };
entry.count++;
if (entry.count === 1) {
counts.set(url, entry);
publish();
}
return () => {
if (--entry.count > 0) return;
counts.delete(url);
publish();
};
};
return {
follow: (memberScope, transport$) => {
let release: (() => void) | undefined;
transport$.pipe(memberScope.bind()).subscribe((transport) => {
release?.();
release = transport && register(transport);
});
memberScope.onEnd(() => release?.());
},
transports$: scope.behavior(transports.pipe(trackEpoch())),
};
}
/**
* Whether the SFU can take over restarting the delayed leave. Either the
* homeserver or the transport has to support it. Each endpoint is hit without
* credentials and read for a 404, not retried: many servers predate the
* endpoint altogether and answer with a CORS failure that a retry loop would
* only repeat.
*/
export async function supportsDelegation(
homeserverUrl: string,
transport: LivekitTransport,
logger: Logger,
): Promise<boolean> {
const [homeserver, sfu] = await Promise.all([
endpointExists(
`${homeserverUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
"homeserver",
logger,
),
endpointExists(
`${transport.livekit_service_url}/delegate_delayed_leave`,
`transport ${transport.livekit_service_url}`,
logger,
),
]);
return homeserver || sfu;
}
async function endpointExists(
url: string,
serviceName: string,
logger: Logger,
): Promise<boolean> {
try {
const res = await fetch(url, { method: "POST" });
const supported = res.status !== 404;
logger.info(
`${serviceName} ${supported ? "supports" : "does not support"} delegation`,
);
return supported;
} catch (e) {
logger.warn(
`Failed to determine whether ${serviceName} supports delegation, assuming no support`,
e,
);
return false;
}
}
@@ -0,0 +1,206 @@
/*
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 { EventEmitter } from "events";
import {
type LocalParticipant,
ParticipantEvent,
RemoteAudioTrack,
type RemoteParticipant,
type Room as LivekitRoom,
Track,
TrackEvent,
type TrackPublication,
} from "livekit-client";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { ObservableScope } from "../../reactive/ObservableScope";
import { type AudioMediaTrack } from "../../api";
import {
createLivekitMediaTrack,
createLocalLivekitMediaTrack,
} from "./LivekitMediaTrack";
describe("createLivekitMediaTrack", () => {
let scope: ObservableScope;
beforeEach(() => {
scope = new ObservableScope();
});
afterEach(() => scope.end());
it("describes the publication", () => {
const { participant, publication, room } = fakes();
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
);
expect(track.kind).toBe("video");
expect(track.source).toBe("camera");
expect(track.id).toBe("TR_1");
expect(track.encrypted$.value).toBe(true);
});
it("attaches elements to the track, even one that arrives later", () => {
const { participant, publication, room, emitter } = fakes({
track: undefined,
});
const element = document.createElement("video");
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
);
track.attach(element);
track.attach(element);
const livekitTrack = fakeTrack();
publication.track = livekitTrack;
emitter.emit(TrackEvent.Subscribed, livekitTrack);
expect(livekitTrack.attach).toHaveBeenCalledTimes(1);
expect(livekitTrack.attach).toHaveBeenCalledWith(element);
track.detach(element);
expect(livekitTrack.detach).toHaveBeenCalledWith(element);
});
it("detaches everything when its scope ends", () => {
const { participant, publication, room } = fakes();
const element = document.createElement("video");
const scope = new ObservableScope();
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
);
track.attach(element);
scope.end();
expect(publication.track!.detach).toHaveBeenCalledWith(element);
});
it("follows the mute state", () => {
const { participant, publication, room, emitter } = fakes();
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
);
expect(track.muted$.value).toBe(false);
publication.isMuted = true;
emitter.emit(ParticipantEvent.TrackMuted, publication);
expect(track.muted$.value).toBe(true);
});
it("scales a remote member's volume on the participant", () => {
const { participant, publication, room } = fakes({
kind: Track.Kind.Audio,
});
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
) as AudioMediaTrack;
track.setVolume(0.5);
expect(publication.track!.setVolume).toHaveBeenCalledWith(0.5);
});
it("is active while the member speaks and the track is not muted", () => {
const { participant, publication, room, emitter } = fakes({
kind: Track.Kind.Audio,
});
const track = createLivekitMediaTrack(
scope,
participant,
publication as unknown as TrackPublication,
room,
) as AudioMediaTrack;
expect(track.isActive$.value).toBe(false);
participant.isSpeaking = true;
emitter.emit(ParticipantEvent.IsSpeakingChanged, true);
expect(track.isActive$.value).toBe(true);
publication.isMuted = true;
emitter.emit(ParticipantEvent.TrackMuted, publication);
expect(track.isActive$.value).toBe(false);
});
it("gives our own track its controls", async () => {
const { participant, publication, room } = fakes({
kind: Track.Kind.Audio,
});
const setEnabled = vi.fn().mockResolvedValue(false);
const switchActiveDevice = vi.fn().mockResolvedValue(true);
const track = createLocalLivekitMediaTrack(
scope,
Object.assign(participant, {
isLocal: true,
}) as unknown as LocalParticipant,
publication as unknown as TrackPublication,
Object.assign(room, { switchActiveDevice }),
setEnabled,
);
await expect(track.setEnabled(false)).resolves.toBe(false);
expect(setEnabled).toHaveBeenCalledWith("microphone", false);
await track.setDevice("mic-2");
expect(switchActiveDevice).toHaveBeenCalledWith("audioinput", "mic-2");
});
});
type FakeTrack = Track & {
attach: ReturnType<typeof vi.fn>;
detach: ReturnType<typeof vi.fn>;
setVolume: ReturnType<typeof vi.fn>;
};
type FakePublication = Omit<TrackPublication, "track" | "isMuted"> & {
track: FakeTrack | undefined;
isMuted: boolean;
};
function fakeTrack(): FakeTrack {
return Object.assign(Object.create(RemoteAudioTrack.prototype), {
attach: vi.fn(),
detach: vi.fn(),
setVolume: vi.fn(),
setAudioContext: vi.fn(),
setWebAudioPlugins: vi.fn(),
}) as FakeTrack;
}
function fakes({
track = fakeTrack(),
kind = Track.Kind.Video,
}: { track?: FakeTrack | undefined; kind?: Track.Kind } = {}): {
participant: RemoteParticipant & { isSpeaking: boolean };
publication: FakePublication;
room: LivekitRoom;
emitter: EventEmitter;
} {
const emitter = new EventEmitter();
const participant = Object.assign(emitter, {
isLocal: false,
isSpeaking: false,
identity: "@alice:example.org:DEVICE",
getTrackPublication: (): TrackPublication =>
publication as unknown as TrackPublication,
}) as unknown as RemoteParticipant & { isSpeaking: boolean };
const publication = Object.assign(emitter, {
kind,
source:
kind === Track.Kind.Audio ? Track.Source.Microphone : Track.Source.Camera,
trackSid: "TR_1",
isMuted: false,
isEncrypted: true,
track,
}) as unknown as FakePublication;
const room = new EventEmitter() as unknown as LivekitRoom;
return { participant, publication, room, emitter };
}
@@ -0,0 +1,290 @@
/*
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 {
observeParticipantEvents,
observeParticipantMedia,
roomEventSelector,
} from "@livekit/components-core";
import {
facingModeFromLocalTrack,
type LocalParticipant,
LocalTrack,
LocalVideoTrack,
type Participant,
ParticipantEvent,
RemoteAudioTrack,
RemoteTrack,
type Room as LivekitRoom,
RoomEvent,
Track,
TrackEvent,
type TrackPublication,
} from "livekit-client";
import {
combineLatest,
distinctUntilChanged,
fromEvent,
interval,
map,
merge,
of,
share,
startWith,
switchMap,
} from "rxjs";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import {
type AudioMediaTrack,
type LocalAudioMediaTrack,
type LocalVideoMediaTrack,
type MediaSource,
type MediaStreamStats,
type MediaTrack,
type VideoMediaTrack,
} from "../../api";
import { MatrixRTCError } from "../../errors";
import { LazyBehavior } from "../../utils/LazyBehavior";
export const mediaSources: Record<Track.Source, MediaSource> = {
[Track.Source.Microphone]: "microphone",
[Track.Source.Camera]: "camera",
[Track.Source.ScreenShare]: "screenShare",
[Track.Source.ScreenShareAudio]: "screenShareAudio",
[Track.Source.Unknown]: "unknown",
};
export const livekitSources: Record<MediaSource, Track.Source> = {
microphone: Track.Source.Microphone,
camera: Track.Source.Camera,
screenShare: Track.Source.ScreenShare,
screenShareAudio: Track.Source.ScreenShareAudio,
unknown: Track.Source.Unknown,
};
// One timer for every track so that a large session does not keep hundreds of
// them, each firing a statistics request, in the event loop.
const refreshStats$ = interval(1000).pipe(startWith(0), share());
/**
* One publication of a participant as a `MediaTrack`. The publication is
* fixed; the track behind it may come and go (a remote track arrives on
* subscription), and every attached element follows it.
*/
export function createLivekitMediaTrack(
scope: ObservableScope,
participant: Participant,
publication: TrackPublication,
room: LivekitRoom,
): AudioMediaTrack | VideoMediaTrack {
const mediaChanged$ = observeParticipantMedia(participant);
const track$ = publicationTrack$(scope, participant, publication);
const attached = new Set<HTMLMediaElement>();
let audioContext: AudioContext | undefined;
let audioPlugins: AudioNode[] = [];
let volume: number | undefined;
const applyAudioSettings = (track: Track | undefined): void => {
if (!(track instanceof RemoteAudioTrack)) return;
track.setAudioContext(audioContext);
track.setWebAudioPlugins(audioPlugins);
if (volume !== undefined) track.setVolume(volume);
};
let current = track$.value;
track$.pipe(scope.bind()).subscribe((track) => {
for (const element of attached) current?.detach(element);
current = track;
applyAudioSettings(track);
for (const element of attached) track?.attach(element);
});
scope.onEnd(() => {
for (const element of attached) current?.detach(element);
attached.clear();
});
const muted$ = scope.behavior(
mediaChanged$.pipe(map(() => publication.isMuted)),
publication.isMuted,
);
const base: MediaTrack = {
source: mediaSources[publication.source],
kind: publication.kind === Track.Kind.Audio ? "audio" : "video",
id: publication.trackSid,
muted$,
encrypted$: scope.behavior(
merge(
mediaChanged$,
roomEventSelector(room, RoomEvent.ParticipantEncryptionStatusChanged),
).pipe(map(() => publication.isEncrypted)),
publication.isEncrypted,
),
stats$: new LazyBehavior<MediaStreamStats>(
refreshStats$.pipe(
switchMap(async () => rtpStreamStats(publication, participant.isLocal)),
scope.bind(),
),
undefined,
),
attach: (element) => {
if (attached.has(element)) return;
attached.add(element);
current?.attach(element);
},
detach: (element) => {
if (!attached.delete(element)) return;
current?.detach(element);
},
};
if (base.kind === "audio")
return {
...base,
kind: "audio",
// LiveKit detects speakers per participant, so this is the member's
// activity, narrowed to the tracks that can be carrying it
isActive$: scope.behavior(
combineLatest(
[
observeParticipantEvents(
participant,
ParticipantEvent.IsSpeakingChanged,
).pipe(map((p) => p.isSpeaking)),
muted$,
],
(speaking, muted) => speaking && !muted,
),
participant.isSpeaking && !publication.isMuted,
),
setAudioContext: (ctx, plugins = []) => {
audioContext = ctx;
audioPlugins = plugins;
applyAudioSettings(current);
},
setVolume: (v) => {
// Our own audio is never played back, so there is nothing to scale
volume = v;
applyAudioSettings(current);
},
};
return { ...base, kind: "video" };
}
/**
* One of our own publications, with the controls over it. `setEnabled` goes
* through the local member so that what it remembers for a reconnection
* stays in step.
*/
export function createLocalLivekitMediaTrack(
scope: ObservableScope,
participant: LocalParticipant,
publication: TrackPublication,
room: LivekitRoom,
setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>,
): LocalAudioMediaTrack | LocalVideoMediaTrack {
const base = createLivekitMediaTrack(scope, participant, publication, room);
const local = {
setEnabled: async (enabled: boolean) => setEnabled(base.source, enabled),
setDevice: async (deviceId: string): Promise<void> => {
if (base.source !== "microphone" && base.source !== "camera")
throw new MatrixRTCError(`A ${base.source} track has no device`);
await room.switchActiveDevice(
base.kind === "audio" ? "audioinput" : "videoinput",
deviceId,
);
},
};
if (base.kind === "audio") return { ...base, ...local };
const videoTrack = (): LocalVideoTrack | undefined =>
publication.track instanceof LocalVideoTrack
? publication.track
: undefined;
return {
...base,
...local,
facingMode$: facingMode$(
scope,
publicationTrack$(scope, participant, publication),
),
switchFacingMode: async () => {
const track = videoTrack();
if (!track) return;
const { facingMode } = facingModeFromLocalTrack(track);
if (facingMode !== "user" && facingMode !== "environment") return;
await track.restartTrack({
facingMode: facingMode === "user" ? "environment" : "user",
});
return track.mediaStreamTrack.getSettings().deviceId;
},
setProcessor: async (processor) => {
const track = videoTrack();
// A processor cannot be built on a track that has already ended
if (!track || track.mediaStreamTrack.readyState === "ended") return;
if (processor) await track.setProcessor(processor);
else if (track.getProcessor()) await track.stopProcessor();
},
};
}
/** The track behind a publication; a remote one arrives on subscription. */
function publicationTrack$(
scope: ObservableScope,
participant: Participant,
publication: TrackPublication,
): Behavior<Track | undefined> {
return scope.behavior(
merge(
observeParticipantMedia(participant),
fromEvent(publication, TrackEvent.Subscribed),
fromEvent(publication, TrackEvent.Unsubscribed),
).pipe(
map(() => publication.track),
startWith(publication.track),
distinctUntilChanged(),
),
);
}
async function rtpStreamStats(
publication: TrackPublication,
local: boolean,
): Promise<MediaStreamStats> {
const track = publication.track;
if (!(track instanceof RemoteTrack || track instanceof LocalTrack))
return undefined;
const report = await track.getRTCStatsReport();
if (!report) return undefined;
const type = local ? "outbound-rtp" : "inbound-rtp";
for (const stats of report.values()) if (stats.type === type) return stats;
return undefined;
}
function facingMode$(
scope: ObservableScope,
track$: Behavior<Track | undefined>,
): Behavior<"user" | "environment" | undefined> {
return scope.behavior(
track$.pipe(
switchMap((track) => {
if (!(track instanceof LocalVideoTrack)) return of(undefined);
return fromEvent(track, TrackEvent.Restarted).pipe(
startWith(null),
map(() => {
const { facingMode } = facingModeFromLocalTrack(track);
return facingMode === "user" || facingMode === "environment"
? facingMode
: undefined;
}),
);
}),
),
);
}
@@ -0,0 +1,101 @@
/*
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 { EventEmitter } from "events";
import {
ParticipantEvent,
type RemoteParticipant,
type Room as LivekitRoom,
Track,
type TrackPublication,
} from "livekit-client";
import { describe, expect, it } from "vitest";
import { E2eeType } from "../../encryption";
import { testScope } from "../../utils/test";
import { createLivekitMemberMedia } from "./LivekitMemberMedia";
describe("createLivekitMemberMedia", () => {
it("lists one stable track per publication, in publication order", () => {
const { participant, publications, room, emitter } = fakes();
const media = createLivekitMemberMedia(testScope(), participant, room, {
kind: E2eeType.NONE,
});
expect(media.tracks$.value).toEqual([]);
const microphone = publish(publications, "TR_1", Track.Source.Microphone);
const camera = publish(publications, "TR_2", Track.Source.Camera);
emitter.emit(ParticipantEvent.TrackPublished, camera);
const [first, second] = media.tracks$.value;
expect(first.source).toBe("microphone");
expect(second.source).toBe("camera");
// An unrelated media event keeps the same objects
microphone.isMuted = true;
emitter.emit(ParticipantEvent.TrackMuted, microphone);
expect(media.tracks$.value[0]).toBe(first);
expect(media.tracks$.value[1]).toBe(second);
publications.delete("TR_1");
emitter.emit(ParticipantEvent.TrackUnpublished, microphone);
expect(media.tracks$.value).toEqual([second]);
});
it("includes a track published without a source", () => {
const { participant, publications, room, emitter } = fakes();
const media = createLivekitMemberMedia(testScope(), participant, room, {
kind: E2eeType.NONE,
});
emitter.emit(
ParticipantEvent.TrackPublished,
publish(publications, "TR_1", Track.Source.Unknown),
);
expect(media.tracks$.value.map((t) => t.source)).toEqual(["unknown"]);
});
});
type FakePublication = { isMuted: boolean } & Omit<TrackPublication, "isMuted">;
function publish(
publications: Map<string, TrackPublication>,
trackSid: string,
source: Track.Source,
): FakePublication {
const publication = Object.assign(new EventEmitter(), {
trackSid,
source,
kind:
source === Track.Source.Camera || source === Track.Source.ScreenShare
? Track.Kind.Video
: Track.Kind.Audio,
isMuted: false,
isEncrypted: true,
track: undefined,
}) as unknown as FakePublication;
publications.set(trackSid, publication as unknown as TrackPublication);
return publication;
}
function fakes(): {
participant: RemoteParticipant;
publications: Map<string, TrackPublication>;
room: LivekitRoom;
emitter: EventEmitter;
} {
const emitter = new EventEmitter();
const publications = new Map<string, TrackPublication>();
const participant = Object.assign(emitter, {
isLocal: false,
isSpeaking: false,
identity: "@alice:example.org:DEVICE",
trackPublications: publications,
getTrackPublication: (source: Track.Source) =>
[...publications.values()].find((p) => p.source === source),
}) as unknown as RemoteParticipant;
const room = new EventEmitter() as unknown as LivekitRoom;
return { participant, publications, room, emitter };
}
@@ -0,0 +1,135 @@
/*
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 {
observeParticipantMedia,
roomEventSelector,
} from "@livekit/components-core";
import {
type LocalParticipant,
type Participant,
type Room as LivekitRoom,
RoomEvent,
type TrackPublication,
} from "livekit-client";
import { filter, map, type Observable } from "rxjs";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { generateItems } from "../../reactive/observable";
import { E2eeType } from "../../encryption";
import { type EncryptionSystem } from "../../encryption";
import { type Behavior } from "../../reactive/Behavior";
import {
type EncryptionError,
type LocalMemberMedia,
type MediaSource,
type MemberMedia,
} from "../../api";
import {
createLivekitMediaTrack,
createLocalLivekitMediaTrack,
} from "./LivekitMediaTrack";
/**
* A participant as `MemberMedia`. The only place, with `LivekitMediaTrack`,
* that reads a LiveKit participant; everything above sees tracks and
* behaviors.
*/
export function createLivekitMemberMedia(
scope: ObservableScope,
participant: Participant,
room: LivekitRoom,
encryptionSystem: EncryptionSystem,
): MemberMedia {
return {
local: false,
tracks$: memberTracks$(scope, participant, (trackScope, publication) =>
createLivekitMediaTrack(trackScope, participant, publication, room),
),
encryptionError$: encryptionErrors$(
scope,
participant,
room,
encryptionSystem,
),
};
}
export function createLocalLivekitMemberMedia(
scope: ObservableScope,
participant: LocalParticipant,
room: LivekitRoom,
encryptionSystem: EncryptionSystem,
setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>,
): LocalMemberMedia {
return {
local: true,
tracks$: memberTracks$(scope, participant, (trackScope, publication) =>
createLocalLivekitMediaTrack(
trackScope,
participant,
publication,
room,
setEnabled,
),
),
encryptionError$: encryptionErrors$(
scope,
participant,
room,
encryptionSystem,
),
};
}
/** One track per publication, each living as long as its publication is in the participant's map. */
function memberTracks$<T>(
scope: ObservableScope,
participant: Participant,
factory: (scope: ObservableScope, publication: TrackPublication) => T,
): Behavior<T[]> {
return scope.behavior(
observeParticipantMedia(participant).pipe(
generateItems(
`${participant.identity} tracks$`,
function* () {
for (const publication of participant.trackPublications.values())
yield { keys: [publication.trackSid], data: publication };
},
(trackScope, publication$) => factory(trackScope, publication$.value),
),
),
);
}
function encryptionErrors$(
scope: ObservableScope,
participant: Participant,
room: LivekitRoom,
encryptionSystem: EncryptionSystem,
): Observable<EncryptionError> {
return roomEventSelector(room, RoomEvent.EncryptionError).pipe(
map(([error]) => error?.message ?? ""),
// The participant the error is about does not survive the trip from the
// worker, so the identity is matched in the message. A shared key is the
// same for everyone, so its errors are too.
filter(
(message) =>
encryptionSystem.kind === E2eeType.SHARED_KEY ||
message.includes(participant.identity),
),
map((message): EncryptionError | undefined =>
message.includes("MissingKey")
? "MissingKey"
: message.includes("InvalidKey")
? "InvalidKey"
: undefined,
),
filter((error) => error !== undefined),
scope.bind(),
);
}
+272
View File
@@ -0,0 +1,272 @@
/*
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 { observeParticipantEvents } from "@livekit/components-core";
import {
type LocalParticipant,
type LocalTrackPublication,
ParticipantEvent,
type Room as LivekitRoom,
} from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
BehaviorSubject,
combineLatest,
distinctUntilChanged,
filter,
firstValueFrom,
map,
NEVER,
type Observable,
of,
switchMap,
} from "rxjs";
import {
type LocalAudioMediaTrack,
type LocalMediaInputs,
type LocalMemberMedia,
type LocalVideoMediaTrack,
type MediaSource,
type PublishRequest,
} from "../../api";
import { type EncryptionSystem } from "../../encryption";
import { FailToStartLivekitConnection, toMatrixRTCError } from "../../errors";
import { type Behavior } from "../../reactive/Behavior";
import { type ObservableScope } from "../../reactive/ObservableScope";
import { mapScoped } from "../../utils/mapScoped";
import { type LocalMediaBackend, MediaConnectionState } from "../api";
import { type Connection } from "./Connection";
import {
type IConnectionManager,
type LocalTransport,
} from "./ConnectionManager";
import { createLocalLivekitMemberMedia } from "./LivekitMemberMedia";
import { livekitSources } from "./LivekitMediaTrack";
import { type DesiredMedia, Publisher } from "./Publisher";
import { toMediaConnectionState } from "./connectionState";
interface Props {
scope: ObservableScope;
connectionManager: IConnectionManager;
localTransport$: Observable<LocalTransport>;
localMedia: LocalMediaInputs;
encryptionSystem: EncryptionSystem;
logger: Logger;
}
export interface LivekitLocalMedia extends LocalMediaBackend {
/** The connection the local member publishes on; the data channel sends on it. */
connection$: Behavior<Connection | null>;
}
/**
* The local member's side of the LiveKit backend: one `Publisher` per
* connection to the local transport, driven by the publish intent and the
* requests the host made, both of which outlive any one connection.
*/
export function createLivekitLocalMedia({
scope,
connectionManager,
localTransport$,
localMedia,
encryptionSystem,
logger: parentLogger,
}: Props): LivekitLocalMedia {
const logger = parentLogger.getChild("[LocalMedia]");
const connection$ = scope.behavior(
combineLatest([
connectionManager.connectionManagerData$,
localTransport$,
]).pipe(
map(([{ value: connections }, { transport }]) =>
connections.getConnectionForTransport(transport),
),
),
null,
);
const connectionState$ = scope.behavior<MediaConnectionState | Error>(
connection$.pipe(
switchMap(
(connection) =>
connection?.state$.pipe(map(toMediaConnectionState)) ??
of(MediaConnectionState.Initialized),
),
),
);
const desired: DesiredMedia = new Map(
localMedia.publish.map((request) => [
request.source,
{ request, enabled: true },
]),
);
const publishing$ = new BehaviorSubject(false);
const publisher$ = new BehaviorSubject<Publisher | null>(null);
const publishError$ = new BehaviorSubject<Error | null>(null);
scope.reconcile(connection$, async (connection) => {
if (connection === null) return;
const publisher = new Publisher(
connection.livekitRoom,
localMedia,
desired,
logger.getChild(
`[Publisher ${connection.transport.livekit_service_url}]`,
),
);
publisher$.next(publisher);
return Promise.resolve(async (): Promise<void> => {
publisher$.next(null);
await publisher.destroy();
});
});
scope.reconcile(
scope.behavior(combineLatest([publisher$, publishing$])),
async ([publisher, publish]) => {
if (publisher === null) return;
try {
if (publish) {
publisher.start();
await publisher.startPublishing();
} else if (publisher.shouldPublish) await publisher.stopPublishing();
} catch (e) {
if (publishError$.value === null)
publishError$.next(
new FailToStartLivekitConnection(
e instanceof Error ? e.message : String(e),
),
);
else logger.error("Another publish error", e);
}
},
);
const participant$ = scope.behavior(
connection$.pipe(map((c) => c?.livekitRoom.localParticipant ?? null)),
);
const setEnabled = async (
source: MediaSource,
enabled: boolean,
): Promise<boolean> => {
const wanted = desired.get(source);
if (wanted) wanted.enabled = enabled;
// Without a publisher the request waits for the tracks to be created
const publisher = publisher$.value;
if (publisher === null) return enabled;
const result = await publisher.setEnabled(source, enabled);
if (wanted && result !== enabled) wanted.enabled = result;
return result;
};
const media$ = scope.behavior<LocalMemberMedia | null>(
mapScoped(
scope,
participantAndRoom$(scope, participant$, connection$),
(mediaScope, { participant, room }) =>
createLocalLivekitMemberMedia(
mediaScope,
participant,
room,
encryptionSystem,
setEnabled,
),
).pipe(map((media) => media ?? null)),
);
const publish = async (
request: PublishRequest,
): Promise<LocalAudioMediaTrack | LocalVideoMediaTrack> => {
desired.set(request.source, { request, enabled: true });
const publisher = publisher$.value;
try {
// Before the start, the publisher publishes everything desired itself
const publication =
(publisher?.started ? await publisher.publish(request) : undefined) ??
(await firstValueFrom(publication$(participant$, request.source)));
return await firstValueFrom(trackWithId$(media$, publication.trackSid));
} catch (e) {
desired.delete(request.source);
throw toMatrixRTCError(e);
}
};
const unpublish = async (id: string): Promise<void> => {
const track = media$.value?.tracks$.value.find((t) => t.id === id);
if (track === undefined) return;
// The screen share audio goes with its video
const source =
track.source === "screenShareAudio" ? "screenShare" : track.source;
desired.delete(source);
await publisher$.value?.unpublish(source);
};
return {
connection$,
connectionState$,
media$,
setPublishing: (publish) => publishing$.next(publish),
publishError$,
publish,
unpublish,
};
}
/** The pair a `MemberMedia` is built on, changing only when one of the two does. */
function participantAndRoom$(
scope: ObservableScope,
participant$: Behavior<LocalParticipant | null>,
connection$: Behavior<Connection | null>,
): Behavior<{ participant: LocalParticipant; room: LivekitRoom } | null> {
return scope.behavior(
combineLatest([participant$, connection$]).pipe(
map(([participant, connection]) =>
participant && connection
? { participant, room: connection.livekitRoom }
: null,
),
distinctUntilChanged(
(a, b) => a?.participant === b?.participant && a?.room === b?.room,
),
),
);
}
/** The publication of a source, once the current participant has made it. */
function publication$(
participant$: Behavior<LocalParticipant | null>,
source: MediaSource,
): Observable<LocalTrackPublication> {
return participant$.pipe(
switchMap((participant) =>
participant === null
? NEVER
: observeParticipantEvents(
participant,
ParticipantEvent.LocalTrackPublished,
).pipe(
map(() => participant.getTrackPublication(livekitSources[source])),
),
),
filter((publication) => publication !== undefined),
);
}
function trackWithId$(
media$: Behavior<LocalMemberMedia | null>,
id: string,
): Observable<LocalAudioMediaTrack | LocalVideoMediaTrack> {
return media$.pipe(
switchMap((media) => media?.tracks$ ?? of([])),
map((tracks) => tracks.find((track) => track.id === id)),
filter((track) => track !== undefined),
);
}
@@ -0,0 +1,51 @@
/*
Copyright 2024 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 { KeyProviderEvent } from "livekit-client";
import { Subject } from "rxjs";
import { describe, expect, test, vi } from "vitest";
import { type MediaKey } from "../api";
import { MatrixKeyProvider } from "./MatrixKeyProvider";
describe("matrixKeyProvider", () => {
test("initializes", () => {
const keyProvider = new MatrixKeyProvider();
expect(keyProvider).toBeTruthy();
});
test("follows the key stream it is given", async () => {
// jsdom has no HKDF; the derived material is not what is under test
vi.spyOn(crypto.subtle, "importKey").mockResolvedValue({} as CryptoKey);
const keyProvider = new MatrixKeyProvider();
const setKey = vi.fn();
keyProvider.on(KeyProviderEvent.SetKey, setKey);
const keys$ = new Subject<MediaKey>();
keyProvider.setMediaKeys(keys$);
keys$.next({ participantId: "alice", index: 2, key: new Uint8Array(16) });
await vi.waitFor(() =>
expect(setKey.mock.lastCall?.[0]).toMatchObject({
participantIdentity: "alice",
keyIndex: 2,
}),
);
});
test("stops following a stream when given another", () => {
const keyProvider = new MatrixKeyProvider();
const first$ = new Subject<MediaKey>();
const second$ = new Subject<MediaKey>();
keyProvider.setMediaKeys(first$);
expect(first$.observed).toBe(true);
keyProvider.setMediaKeys(second$);
expect(first$.observed).toBe(false);
expect(second$.observed).toBe(true);
});
});
@@ -0,0 +1,46 @@
/*
Copyright 2023, 2024 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 { BaseKeyProvider } from "livekit-client";
import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
import { type Observable, type Subscription } from "rxjs";
import { type MediaKey } from "../api";
/** LiveKit's key provider, fed from the session's per-participant media keys. */
export class MatrixKeyProvider extends BaseKeyProvider {
private subscription?: Subscription;
private logger: Logger;
public constructor() {
super({ ratchetWindowSize: 10, keyringSize: 256 });
this.logger = rootLogger.getChild("[MatrixKeyProvider]");
}
public setMediaKeys(mediaKeys$: Observable<MediaKey>): void {
this.subscription?.unsubscribe();
this.subscription = mediaKeys$.subscribe(this.onMediaKey);
}
private onMediaKey = ({ key, index, participantId }: MediaKey): void => {
crypto.subtle
.importKey("raw", key, "HKDF", false, ["deriveBits", "deriveKey"])
.then(
(keyMaterial) => {
this.onSetEncryptionKey(keyMaterial, participantId, index);
this.logger.debug(
`Sent new key to livekit participantId=${participantId} encryptionKeyIndex=${index}`,
);
},
(e) => {
this.logger.error(
`Failed to create key material from buffer for livekit participantId=${participantId} encryptionKeyIndex=${index}`,
e,
);
},
);
};
}
+234
View File
@@ -0,0 +1,234 @@
/*
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 LocalTrackPublication,
ParticipantEvent,
type Room as LivekitRoom,
type Track,
} from "livekit-client";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { ObservableScope } from "../../reactive/ObservableScope";
import {
type LocalMediaInputs,
type MediaSource,
type PublishRequest,
} from "../../api";
import { livekitSources, mediaSources } from "./LivekitMediaTrack";
import {
audioCaptureOptions,
screenShareCaptureOptions,
screenSharePublishOptions,
videoCaptureOptions,
videoPublishOptions,
} from "./publishOptions";
/** What the host asked to publish, kept across publishers for a reconnection. */
export interface DesiredPublication {
request: PublishRequest;
enabled: boolean;
}
export type DesiredMedia = Map<MediaSource, DesiredPublication>;
/**
* Publishes the local media on one LiveKit room, following the host's
* requests.
*
* LiveKit publishes a track the moment it is created, but a member must not be
* heard before it has joined the MatrixRTC session, nor after it has left. The
* tracks are therefore kept, and their upstream paused, while `shouldPublish`
* is false: the local preview stays live while nothing reaches the room.
*/
export class Publisher {
public shouldPublish = false;
/** Whether the initial requests have been published; `publish` is immediate from then on. */
public started = false;
private readonly scope = new ObservableScope();
private readonly room: LivekitRoom;
public constructor(
room: LivekitRoom,
private readonly inputs: LocalMediaInputs,
private readonly desired: DesiredMedia,
private readonly logger: Logger,
) {
this.room = room;
room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
this.logger.error("Failed to enable E2EE on the room", e);
});
this.followAudioOutput();
this.onLocalTrackPublished = this.onLocalTrackPublished.bind(this);
room.localParticipant.on(
ParticipantEvent.LocalTrackPublished,
this.onLocalTrackPublished,
);
}
public async destroy(): Promise<void> {
this.scope.end();
this.room.localParticipant.off(
ParticipantEvent.LocalTrackPublished,
this.onLocalTrackPublished,
);
try {
for (const { track } of this.publications())
if (track) await this.room.localParticipant.unpublishTrack(track, true);
} catch (e) {
this.logger.error("Failed to stop the local tracks", e);
}
}
/** Publishes what the host asked for, less what it has muted since. Safe to call more than once. */
public start(): void {
if (this.started) return;
this.started = true;
const requests = [...this.desired.values()]
.filter(({ enabled }) => enabled)
.map(({ request }) => request);
const sources = new Set(requests.map((request) => request.source));
// Microphone and camera in one call so that the browser asks for
// permission once; the room defaults carry their options. LiveKit
// resolves these once the track is published, which may block on the
// connection; LocalTrackPublished is what tells us a track exists.
if (sources.has("microphone") && sources.has("camera"))
void this.room.localParticipant.enableCameraAndMicrophone();
for (const request of requests)
if (
request.source === "screenShare" ||
!(sources.has("microphone") && sources.has("camera"))
)
this.publish(request).catch((e) => {
this.logger.error(`Failed to publish the ${request.source}`, e);
});
}
/** Publishes one request and resolves with its publication once LiveKit has it. */
public async publish(
request: PublishRequest,
): Promise<LocalTrackPublication | undefined> {
const participant = this.room.localParticipant;
let publication: LocalTrackPublication | undefined;
if (request.source === "microphone")
publication = await participant.setMicrophoneEnabled(
true,
audioCaptureOptions(request),
);
else if (request.source === "camera")
publication = await participant.setCameraEnabled(
true,
videoCaptureOptions(request),
videoPublishOptions(request.capture),
);
else
publication = await participant.setScreenShareEnabled(
true,
screenShareCaptureOptions(request),
screenSharePublishOptions(request.capture),
);
// Until the member has joined, the upstream has to stay paused
if (!this.shouldPublish) await this.pauseUpstreams();
return publication;
}
public async unpublish(source: MediaSource): Promise<void> {
const participant = this.room.localParticipant;
// Takes the screen share audio with it
if (source === "screenShare") {
await participant.setScreenShareEnabled(false);
return;
}
const track = participant.getTrackPublication(
livekitSources[source],
)?.track;
if (track) await participant.unpublishTrack(track, true);
}
/**
* Mutes or unmutes a source and reports what resulted: the request where
* it succeeded, the state LiveKit is left in where it did not (a denied
* permission, a missing device).
*/
public async setEnabled(
source: MediaSource,
enabled: boolean,
): Promise<boolean> {
const participant = this.room.localParticipant;
const livekitSource = livekitSources[source];
try {
if (source === "microphone")
await participant.setMicrophoneEnabled(enabled);
else if (source === "camera") await participant.setCameraEnabled(enabled);
else {
const publication = participant.getTrackPublication(livekitSource);
if (!publication) return false;
if (enabled) await publication.unmute();
else await publication.mute();
}
// Unmuting restarts the upstream; until the member has joined, it has
// to stay paused
if (enabled && !this.shouldPublish) await this.pauseUpstreams();
return enabled;
} catch (e) {
this.logger.error(`Failed to set ${source} enabled=${enabled}`, e);
return !(participant.getTrackPublication(livekitSource)?.isMuted ?? true);
}
}
public async startPublishing(): Promise<void> {
if (this.shouldPublish) return;
this.shouldPublish = true;
// Enabling a track does not resume an upstream that was paused while it
// was already enabled, so it is done explicitly
for (const { track } of this.publications())
if (track?.isUpstreamPaused) await track.resumeUpstream();
}
public async stopPublishing(): Promise<void> {
this.shouldPublish = false;
await this.pauseUpstreams();
}
private onLocalTrackPublished(publication: LocalTrackPublication): void {
this.logger.info(`Local ${publication.source} track published`);
if (!this.shouldPublish)
this.pauseUpstreams().catch((e) => {
this.logger.error("Failed to pause the upstream", e);
});
// The host may have changed its mind while the track was being created
const source = mediaSources[publication.source as Track.Source];
if (this.desired.get(source)?.enabled === false)
void this.setEnabled(source, false);
}
private async pauseUpstreams(): Promise<void> {
for (const { track } of this.publications())
if (track && !track.isUpstreamPaused) await track.pauseUpstream();
}
private publications(): LocalTrackPublication[] {
return [...this.room.localParticipant.trackPublications.values()];
}
private followAudioOutput(): void {
this.inputs.audioOutputDeviceId$
.pipe(this.scope.bind())
.subscribe((deviceId) => {
if (
deviceId === undefined ||
this.room.state !== LivekitConnectionState.Connected ||
this.room.getActiveDevice("audiooutput") === deviceId
)
return;
this.room
.switchActiveDevice("audiooutput", deviceId)
.catch((e) => this.logger.error("Failed to switch audiooutput", e));
});
}
}
@@ -0,0 +1,28 @@
/*
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 { MediaConnectionState } from "../api";
import { ConnectionState } from "./Connection";
const states: Record<ConnectionState, MediaConnectionState> = {
[ConnectionState.Initialized]: MediaConnectionState.Initialized,
[ConnectionState.FetchingConfig]: MediaConnectionState.Preparing,
[ConnectionState.Stopped]: MediaConnectionState.Stopped,
[ConnectionState.LivekitDisconnected]: MediaConnectionState.Disconnected,
[ConnectionState.LivekitConnecting]: MediaConnectionState.Connecting,
[ConnectionState.LivekitConnected]: MediaConnectionState.Connected,
[ConnectionState.LivekitReconnecting]: MediaConnectionState.Reconnecting,
[ConnectionState.LivekitSignalReconnecting]:
MediaConnectionState.Reconnecting,
};
/** A connection's state as the session sees it; an error passes through. */
export function toMediaConnectionState(
state: ConnectionState | Error,
): MediaConnectionState | Error {
return state instanceof Error ? state : states[state];
}
@@ -0,0 +1,189 @@
/*
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 { describe, expect, it } from "vitest";
import { VideoPresets, type VideoPreset } from "livekit-client";
import { buildLiveKitOptions } from "./livekitOptions";
describe("buildLiveKitOptions", () => {
it("returns sensible defaults with no config", () => {
const opts = buildLiveKitOptions();
expect(opts.adaptiveStream).toBe(true);
expect(opts.dynacast).toBe(true);
expect(opts.videoCaptureDefaults?.resolution).toEqual(
VideoPresets.h720.resolution,
);
expect(opts.publishDefaults?.videoCodec).toBe("vp8");
expect(opts.publishDefaults?.videoEncoding).toEqual({
maxBitrate: 1_700_000,
maxFramerate: 30,
});
expect(opts.publishDefaults?.screenShareEncoding).toEqual({
maxBitrate: 5_000_000,
maxFramerate: 30,
});
expect(opts.publishDefaults?.videoSimulcastLayers).toEqual([
VideoPresets.h180,
VideoPresets.h360,
]);
});
it("applies video codec from config", () => {
const opts = buildLiveKitOptions({ video_codec: "vp9" });
expect(opts.publishDefaults?.videoCodec).toBe("vp9");
});
it("applies video resolution and encoding from config", () => {
const baseVideoConfig = {
max_resolution: 1080,
max_bitrate: 3_000_000,
max_framerate: 60,
};
const opts1080 = buildLiveKitOptions({
video: baseVideoConfig,
});
const opts1440 = buildLiveKitOptions({
video: { ...baseVideoConfig, max_resolution: 1440 },
});
const opts2160 = buildLiveKitOptions({
video: { ...baseVideoConfig, max_resolution: 2160 },
});
expect(opts1080.videoCaptureDefaults?.resolution).toEqual(
VideoPresets.h1080.resolution,
);
expect(opts1440.videoCaptureDefaults?.resolution).toEqual(
VideoPresets.h1440.resolution,
);
expect(opts2160.videoCaptureDefaults?.resolution).toEqual(
VideoPresets.h2160.resolution,
);
expect(opts1080.publishDefaults?.videoEncoding).toEqual({
maxBitrate: 3_000_000,
maxFramerate: 60,
});
});
it("applies screen share encoding from config", () => {
const opts = buildLiveKitOptions({
screen_share: {
max_bitrate: 8_000_000,
max_framerate: 15,
},
});
expect(opts.publishDefaults?.screenShareEncoding).toEqual({
maxBitrate: 8_000_000,
maxFramerate: 15,
});
});
it("uses DEFAULT_CONFIG defaults when only resolution is set", () => {
const opts = buildLiveKitOptions({
screen_share: {
max_resolution: 720,
},
});
// Bitrate and framerate fall back to DEFAULT_CONFIG, not the preset
expect(opts.publishDefaults?.screenShareEncoding).toEqual({
maxBitrate: 5_000_000,
maxFramerate: 30,
});
});
it("maps low resolutions to the closest preset, rounding up", () => {
const expectations: [number, VideoPreset][] = [
[180, VideoPresets.h180],
[360, VideoPresets.h360],
[480, VideoPresets.h540],
[540, VideoPresets.h540],
[720, VideoPresets.h720],
];
for (const [height, preset] of expectations) {
const opts = buildLiveKitOptions({ video: { max_resolution: height } });
expect(opts.videoCaptureDefaults?.resolution).toEqual(preset.resolution);
}
});
it("screen share layers fall back to max_framerate, then 30", () => {
const fromMax = buildLiveKitOptions({
screen_share: {
max_framerate: 15,
simulcast_layers: [{ height: 540, bitrate: 1_000_000 }],
},
});
expect(
fromMax.publishDefaults?.screenShareSimulcastLayers?.[0],
).toMatchObject({ encoding: { maxFramerate: 15 } });
const fromDefault = buildLiveKitOptions({
screen_share: {
simulcast_layers: [{ height: 540, bitrate: 1_000_000 }],
},
});
expect(
fromDefault.publishDefaults?.screenShareSimulcastLayers?.[0],
).toMatchObject({ encoding: { maxFramerate: 30 } });
});
it("applies custom video simulcast layers", () => {
const opts = buildLiveKitOptions({
video: {
simulcast_layers: [
{ height: 180, bitrate: 100_000 },
{ height: 360, bitrate: 300_000 },
{ height: 540, bitrate: 600_000 },
],
max_framerate: 24,
},
});
const layers = opts.publishDefaults?.videoSimulcastLayers;
expect(layers).toHaveLength(3);
expect(layers?.[0]).toMatchObject({
width: 320,
height: 180,
encoding: { maxBitrate: 100_000, maxFramerate: 24 },
});
expect(layers?.[2]).toMatchObject({
width: 960,
height: 540,
encoding: { maxBitrate: 600_000, maxFramerate: 24 },
});
});
it("applies custom screen share simulcast layers", () => {
const opts = buildLiveKitOptions({
screen_share: {
simulcast_layers: [{ height: 540, bitrate: 1_000_000, framerate: 5 }],
},
});
const layers = opts.publishDefaults?.screenShareSimulcastLayers;
expect(layers).toHaveLength(1);
expect(layers?.[0]).toMatchObject({
width: 960,
height: 540,
encoding: { maxBitrate: 1_000_000, maxFramerate: 5 },
});
});
it("does not include screenShareSimulcastLayers when not configured", () => {
const opts = buildLiveKitOptions();
expect(opts.publishDefaults?.screenShareSimulcastLayers).toBeUndefined();
});
it("backupCodec always uses stock VP8 720p encoding", () => {
const opts = buildLiveKitOptions({
video_codec: "av1",
video: { max_bitrate: 10_000_000, max_framerate: 60 },
});
const backup = opts.publishDefaults?.backupCodec as {
codec: string;
encoding: { maxBitrate: number; maxFramerate: number };
};
expect(backup.codec).toBe("vp8");
expect(backup.encoding).toEqual(VideoPresets.h720.encoding);
});
});
@@ -0,0 +1,140 @@
/*
Copyright 2023, 2024 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 {
AudioPresets,
DefaultReconnectPolicy,
type RoomOptions,
type TrackPublishDefaults,
type VideoPreset,
VideoPresets,
VideoPreset as VideoPresetClass,
} from "livekit-client";
import { defaultMediaQuality, type MediaQuality } from "../../config";
/**
* Find the closest matching VideoPreset for a given height.
*/
function videoPresetForHeight(height: number): VideoPreset {
if (height <= 180) return VideoPresets.h180;
if (height <= 360) return VideoPresets.h360;
if (height <= 540) return VideoPresets.h540;
if (height <= 720) return VideoPresets.h720;
if (height <= 1080) return VideoPresets.h1080;
if (height <= 1440) return VideoPresets.h1440;
return VideoPresets.h2160;
}
/**
* Build LiveKit publish options from config, falling back to sensible defaults.
*/
function buildPublishOptions(
mediaQuality: MediaQuality | undefined,
): TrackPublishDefaults {
const defaults = defaultMediaQuality;
const videoConf = mediaQuality?.video;
const screenConf = mediaQuality?.screen_share;
const codec = mediaQuality?.video_codec ?? defaults.video_codec;
// Camera video encoding
const videoEncoding = {
maxBitrate: videoConf?.max_bitrate ?? defaults.video.max_bitrate,
maxFramerate: videoConf?.max_framerate ?? defaults.video.max_framerate,
};
// Camera simulcast layers
let videoSimulcastLayers: VideoPreset[];
if (videoConf?.simulcast_layers) {
videoSimulcastLayers = videoConf.simulcast_layers.map(
(layer) =>
new VideoPresetClass(
Math.round((layer.height * 16) / 9),
layer.height,
layer.bitrate,
videoConf?.max_framerate ?? defaults.video.max_framerate,
),
);
} else {
videoSimulcastLayers = [VideoPresets.h180, VideoPresets.h360];
}
// Screen share encoding
const screenShareEncoding = {
maxBitrate: screenConf?.max_bitrate ?? defaults.screen_share.max_bitrate,
maxFramerate:
screenConf?.max_framerate ?? defaults.screen_share.max_framerate,
};
// Screen share simulcast layers
let screenShareSimulcastLayers: VideoPreset[] | undefined;
if (screenConf?.simulcast_layers) {
screenShareSimulcastLayers = screenConf.simulcast_layers.map(
(layer) =>
new VideoPresetClass(
Math.round((layer.height * 16) / 9),
layer.height,
layer.bitrate,
layer.framerate ?? screenConf?.max_framerate ?? 30,
),
);
}
return {
audioPreset: AudioPresets.music,
dtx: true,
// disable red because the livekit server strips out red packets for clients
// that don't support it (firefox) but of course that doesn't work with e2ee.
red: false,
forceStereo: false,
simulcast: true,
videoSimulcastLayers: videoSimulcastLayers as VideoPreset[],
screenShareEncoding,
...(screenShareSimulcastLayers && {
screenShareSimulcastLayers: screenShareSimulcastLayers as VideoPreset[],
}),
stopMicTrackOnMute: false,
videoCodec: codec,
videoEncoding,
backupCodec: {
codec: "vp8",
encoding: VideoPresets.h720.encoding,
},
} as TrackPublishDefaults;
}
/**
* Build LiveKit RoomOptions from the media quality a host asks for.
*/
export function buildLiveKitOptions(mediaQuality?: MediaQuality): RoomOptions {
const videoHeight =
mediaQuality?.video?.max_resolution ??
defaultMediaQuality.video.max_resolution;
const basePreset = videoPresetForHeight(videoHeight);
return {
// automatically manage subscribed video quality
adaptiveStream: true,
// optimize publishing bandwidth and CPU for published tracks
dynacast: true,
// capture settings
videoCaptureDefaults: {
resolution: basePreset.resolution,
},
// publish settings
publishDefaults: buildPublishOptions(mediaQuality),
// default LiveKit options that seem to be sane
stopLocalTrackOnUnpublish: true,
reconnectPolicy: new DefaultReconnectPolicy(),
disconnectOnPageLeave: true,
webAudioMix: false,
};
}
@@ -0,0 +1,332 @@
/*
Copyright 2025 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 {
beforeEach,
afterEach,
describe,
expect,
it,
type MockedObject,
vitest,
} from "vitest";
import fetchMock from "fetch-mock";
import { MatrixError } from "matrix-js-sdk";
import { getSFUConfigWithOpenID, type OpenIDClientParts } from "./openIDSFU";
import { testJWTToken } from "../../utils/test-fixtures";
import { ownMemberMock } from "../../utils/test";
import { FailToGetOpenIdToken } from "../../errors";
const sfuUrl = "https://sfu.example.org";
describe("getSFUConfigWithOpenID", () => {
let matrixClient: MockedObject<OpenIDClientParts>;
beforeEach(() => {
fetchMock.catch(404);
matrixClient = {
getOpenIdToken: vitest.fn(),
};
});
afterEach(() => {
vitest.clearAllMocks();
fetchMock.reset();
});
it("should handle fetching a token", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
void (await fetchMock.flush());
});
it("should fail if the SFU errors", async () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
} catch (ex: unknown) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
const mxError = (ex as Error).cause as MatrixError;
expect(mxError.message).toEqual(
"MatrixError: [500] Failed to look up user info from homeserver",
);
void (await fetchMock.flush());
return;
}
expect.fail("Expected test to throw;");
});
it("should retry without delay params if the JWT service legacy endpoint returns M_BAD_JSON 400", async () => {
let callCount = 0;
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
callCount++;
const body = JSON.parse(opts.body as string);
// First call: check if it has delay parts and return 400
if (callCount === 1) {
expect(body).toHaveProperty("delay_id", "mock_delay_id");
return {
status: 400,
body: { errcode: "M_BAD_JSON", error: "Unsupported parameters" },
};
}
// Second call: check if delay parts were stripped and return success
expect(body).not.toHaveProperty("delay_id");
expect(body).not.toHaveProperty("delay_timeout");
expect(body).not.toHaveProperty("delay_cs_api_url");
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
},
{ overwriteRoutes: true },
);
// Note: Assuming getSFUConfigWithOpenID eventually calls getLiveKitJWT
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
delayTimeoutMs: 3_600_000,
},
);
expect(config.jwt).toBe(testJWTToken);
expect(callCount).toBe(2);
void (await fetchMock.flush());
});
it("should successfully send delay parameters to the JWT service legacy endpoint", async () => {
fetchMock.post(
"https://sfu.example.org/sfu/get",
(url, opts) => {
const body = JSON.parse(opts.body as string);
// Verify, that the request contains the expected delay parameters
if (
body.delay_id === "mock_delay_id" &&
body.delay_timeout === 3600000 &&
body.delay_cs_api_url === "https://homeserverserver.org/cs_api"
) {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
}
return {
status: 400,
body: { error: "Missing expected delay params" },
};
},
{ overwriteRoutes: true },
);
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://homeserverserver.org/cs_api",
delayId: "mock_delay_id",
delayTimeoutMs: 3_600_000,
},
);
// Prüfe das Ergebnis
expect(config).toMatchObject({
jwt: testJWTToken,
url: sfuUrl,
});
void (await fetchMock.flush());
});
it("should try legacy and then new endpoint with delay delegation", async () => {
fetchMock.post("https://sfu.example.org/get_token", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: {
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
delayTimeoutMs: 3_600_000,
},
);
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
const mxError = (ex as Error).cause as MatrixError;
expect(mxError.message).toEqual(
"MatrixError: [500] Failed to look up user info from homeserver",
);
void (await fetchMock.flush());
}
const calls = fetchMock.calls();
expect(calls.length).toBe(2);
expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token");
expect(calls[0][1]).toStrictEqual({
// check if it uses correct delayID!
body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
method: "POST",
headers: {
"Content-Type": "application/json",
},
});
expect(calls[1][0]).toStrictEqual("https://sfu.example.org/sfu/get");
expect(calls[1][1]).toStrictEqual({
body: '{"room":"!example_room_id","device_id":"DEVICE","delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
headers: {
"Content-Type": "application/json",
},
method: "POST",
});
});
it("dont try legacy if endpoint with delay delegation is sucessful", async () => {
fetchMock.post("https://sfu.example.org/get_token", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 500,
body: { error: "Test failure" },
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
delayTimeoutMs: 3_600_000,
},
);
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toEqual(
new Error("SFU Config fetch failed with status code 500"),
);
void (await fetchMock.flush());
}
const calls = fetchMock.calls();
expect(calls.length).toBe(1);
expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token");
expect(calls[0][1]).toStrictEqual({
// check if it uses correct delayID!
body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}',
method: "POST",
headers: {
"Content-Type": "application/json",
},
});
});
it("should retry fetching the openid token", async () => {
let count = 0;
matrixClient.getOpenIdToken.mockImplementation(async () => {
count++;
if (count < 2) {
throw Error("Test failure");
}
return Promise.resolve({
token_type: "Bearer",
access_token: "foobar",
matrix_server_name: "example.org",
expires_in: 30,
});
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
void (await fetchMock.flush());
});
});
+321
View File
@@ -0,0 +1,321 @@
/*
Copyright 2023, 2024 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 {
type IOpenIDToken,
type MatrixClient,
parseErrorResponse,
} from "matrix-js-sdk";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
FailToGetOpenIdToken,
NoMatrix2AuthorizationService,
} from "../../errors";
import { doNetworkOperationWithRetry } from "../../utils/network";
/**
* Configuration and access tokens provided by the SFU on successful authentication.
*/
export interface SFUConfig {
url: string;
jwt: string;
livekitAlias: string;
// NOTE: Currently unused.
livekitIdentity: string;
}
/**
* Decoded details from the JWT.
*/
interface SFUJWTPayload {
/**
* Expiration time for the JWT.
* Note: This value is in seconds since Unix epoch.
*/
exp: number;
/**
* Name of the instance which authored the JWT
*/
iss: string;
/**
* Time at which the JWT can start to be used.
* Note: This value is in seconds since Unix epoch.
*/
nbf: number;
/**
* Subject. The Livekit alias in this context.
*/
sub: string;
/**
* The set of permissions for the user.
*/
video: {
canPublish: boolean;
canSubscribe: boolean;
room: string;
roomJoin: boolean;
};
}
// The bits we need from MatrixClient
export type OpenIDClientParts = Pick<MatrixClient, "getOpenIdToken">;
/**
* Which token service endpoint to use: the MSC4195 one, or the legacy one
* with the old identity format. Undefined tries the MSC4195 one and falls
* back to the legacy one, for transports we only subscribe on.
*/
export type TokenEndpoint = "msc4195" | "legacy";
/**
* Gets a bearer token from the homeserver and then use it to authenticate
* to the matrix RTC backend in order to get acces to the SFU.
* It has built-in retry for calls to the homeserver with a backoff policy.
* @param client The Matrix client
* @param membership Our own membership identity parts used to send to jwt service.
* @param serviceUrl The URL of the livekit SFU service
* @param roomId The room id used in the jwt request. This is NOT the livekit_alias. The jwt service will provide the alias. It maps matrix room ids <-> Livekit aliases.
* @param opts Additional options to modify which endpoint with which data will be used to acquire the jwt token.
* @param opts.tokenEndpoint Which version of the JWT endpoint to use, which affects whether the
* RTC backend identity is based on string concatenation (legacy) or a hash (MSC4195).
* This function by default uses whatever is possible with the current jwt service installed next to the SFU.
* For remote connections this does not matter, since we will not publish there we can rely on the newest option.
* @param opts.delayEndpointBaseUrl The URL of the matrix homeserver.
* @param opts.delayId The delay id used for the jwt service to manage.
* @param logger optional logger.
* @returns Object containing the token information
* @throws FailToGetOpenIdToken
*/
export async function getSFUConfigWithOpenID(
client: OpenIDClientParts,
membership: CallMembershipIdentityParts,
serviceUrl: string,
roomId: string,
opts?: {
tokenEndpoint?: TokenEndpoint;
delayEndpointBaseUrl?: string;
delayId?: string;
/** How long the SFU waits before sending the delegated leave. Needed with `delayId`. */
delayTimeoutMs?: number;
},
logger?: Logger,
): Promise<SFUConfig> {
let openIdToken: IOpenIDToken;
try {
openIdToken = await doNetworkOperationWithRetry(async () =>
client.getOpenIdToken(),
);
} catch (error) {
throw new FailToGetOpenIdToken(
error instanceof Error ? error : new Error("Unknown error"),
);
}
logger?.debug("Got openID token", openIdToken);
let sfuConfig: { url: string; jwt: string } | undefined;
const tryBothJwtEndpoints = opts?.tokenEndpoint === undefined; // This is for SFUs where we do not publish.
const forceMatrix2Jwt = opts?.tokenEndpoint === "msc4195";
// We want to start using the new endpoint (with optional delay delegation)
// if we can use both or if we are forced to use the new one.
if (tryBothJwtEndpoints || forceMatrix2Jwt) {
try {
logger?.info(
`Trying to get JWT via default endpoint for focus ${serviceUrl}...`,
);
const sfuConfig = await getLiveKitJWT(
membership,
serviceUrl,
roomId,
openIdToken,
opts?.delayEndpointBaseUrl,
opts?.delayId,
opts?.delayTimeoutMs,
);
return extractFullConfigFromToken(sfuConfig);
} catch (e) {
logger?.debug(`Failed fetching jwt with matrix 2.0 endpoint:`, e);
// Make this throw a hard error in case we force the matrix2.0 endpoint.
if (forceMatrix2Jwt) {
throw new NoMatrix2AuthorizationService(e as Error);
}
}
}
// DEPRECATED
// here we either have a sfuConfig or we already exited because of `if (forceMatrix2) throw ...`
// The only case we can get into this condition is, if `forceMatrix2` is `false`
try {
logger?.info(
`Trying to get JWT with legacy endpoint for focus ${serviceUrl}...`,
);
sfuConfig = await getLiveKitJWTLegacy(
membership.deviceId,
serviceUrl,
roomId,
openIdToken,
opts?.delayEndpointBaseUrl,
opts?.delayId,
opts?.delayTimeoutMs,
);
logger?.info(`Got JWT from call's active focus URL.`);
return extractFullConfigFromToken(sfuConfig);
} catch (ex) {
throw new FailToGetOpenIdToken(
ex instanceof Error ? ex : new Error(`Unknown error ${ex}`),
);
}
}
function extractFullConfigFromToken(sfuConfig: {
url: string;
jwt: string;
}): SFUConfig {
const [, payloadStr] = sfuConfig.jwt.split(".");
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
return {
jwt: sfuConfig.jwt,
url: sfuConfig.url,
livekitAlias: payload.video.room,
// NOTE: Currently unused.
// Probably also not helpful since we now compute the backendIdentity on joining the call so we can use it for the encryption manager.
// The only reason for us to know it locally is to connect the right users with the lk world. (and to set our own keys)
livekitIdentity: payload.sub,
};
}
async function getLiveKitJWTLegacy(
deviceId: string,
livekitServiceURL: string,
matrixRoomId: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
delayTimeoutMs?: number,
): Promise<{ url: string; jwt: string }> {
interface IDelayParams {
delay_id?: string;
delay_timeout?: number;
delay_cs_api_url?: string;
}
let bodyDalayParts: IDelayParams = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout: delayTimeoutMs,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const makeRequest = async (delayParts: IDelayParams): Promise<Response> => {
return await fetch(livekitServiceURL + "/sfu/get", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({
// The legacy JWT endpoint uses only the matrix room id to calculate the livekit room alias.
// However, the livekit room alias is provided as part of the JWT payload.
room: matrixRoomId,
openid_token: openIDToken,
device_id: deviceId,
...delayParts,
}),
});
};
const res = await doNetworkOperationWithRetry(async () => {
let response = await makeRequest(bodyDalayParts);
// Old service compatibility check
const oldServiceDoesNotSupportDelayParts =
response.status === 400 && Object.keys(bodyDalayParts).length > 0;
// If http status 400 with M_BAD_JSON and we sent delay parts, retry without them
if (oldServiceDoesNotSupportDelayParts) {
try {
const errorBody = await response.json();
if (errorBody.errcode === "M_BAD_JSON") {
response = await makeRequest({});
}
} catch {
// If we can't parse the error, treat as real error
}
}
return response;
});
if (!res.ok) {
throw parseErrorResponse(res, await res.text());
}
return await res.json();
}
class NotSupportedError extends Error {
public constructor(message: string) {
super(message);
this.name = "NotSupported";
}
}
export async function getLiveKitJWT(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
openIDToken: IOpenIDToken,
delayEndpointBaseUrl?: string,
delayId?: string,
delayTimeoutMs?: number,
): Promise<{ url: string; jwt: string }> {
const { userId, deviceId, memberId } = membership;
const body = {
room_id: matrixRoomId,
slot_id: "m.call#ROOM",
openid_token: openIDToken,
member: {
id: memberId,
claimed_user_id: userId,
claimed_device_id: deviceId,
},
};
let bodyDalayParts = {};
// Also check for empty string
if (delayId && delayEndpointBaseUrl) {
bodyDalayParts = {
delay_id: delayId,
delay_timeout: delayTimeoutMs,
delay_cs_api_url: delayEndpointBaseUrl,
};
}
const res = await doNetworkOperationWithRetry(async () => {
return await fetch(livekitServiceURL + "/get_token", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ ...body, ...bodyDalayParts }),
});
});
if (!res.ok) {
const msg = "SFU Config fetch failed with status code " + res.status;
if (res.status === 404) {
throw new NotSupportedError(msg);
} else {
throw parseErrorResponse(res, await res.text());
}
}
return await res.json();
}
@@ -0,0 +1,91 @@
/*
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 AudioCaptureOptions,
type ScreenShareCaptureOptions,
type TrackPublishOptions,
type VideoCaptureOptions,
} from "livekit-client";
import { type PublishRequest, type VideoCaptureSettings } from "../../api";
export type PublishRequestFor<S extends PublishRequest["source"]> = Extract<
PublishRequest,
{ source: S }
>;
/** LiveKit fills in the room defaults for whatever a request leaves out. */
export function audioCaptureOptions({
deviceId,
capture,
}: PublishRequestFor<"microphone">): AudioCaptureOptions | undefined {
if (deviceId === undefined && capture === undefined) return undefined;
return { ...(deviceId && { deviceId }), ...capture };
}
export function videoCaptureOptions({
deviceId,
processor,
capture,
}: PublishRequestFor<"camera">): VideoCaptureOptions | undefined {
if (!deviceId && !processor && !capture?.resolution) return undefined;
return {
...(deviceId && { deviceId }),
...(processor && { processor }),
...(capture?.resolution && { resolution: capture.resolution }),
};
}
export function videoPublishOptions(
settings: VideoCaptureSettings | undefined,
): TrackPublishOptions | undefined {
if (settings === undefined) return undefined;
return {
...(settings.maxBitrate !== undefined && {
videoEncoding: {
maxBitrate: settings.maxBitrate,
maxFramerate: settings.maxFramerate,
},
}),
...(settings.codec && { videoCodec: settings.codec }),
};
}
export function screenShareCaptureOptions({
audio = true,
capture,
}: PublishRequestFor<"screenShare">): ScreenShareCaptureOptions {
return {
// No echo cancellation: it would cancel the other members' voices out of
// the shared audio
audio: audio && {
autoGainControl: false,
noiseSuppression: false,
voiceIsolation: false,
},
selfBrowserSurface: "include",
surfaceSwitching: "include",
systemAudio: "include",
...(capture?.resolution && { resolution: capture.resolution }),
};
}
export function screenSharePublishOptions(
settings: VideoCaptureSettings | undefined,
): TrackPublishOptions | undefined {
if (settings === undefined) return undefined;
return {
...(settings.maxBitrate !== undefined && {
screenShareEncoding: {
maxBitrate: settings.maxBitrate,
maxFramerate: settings.maxFramerate,
},
}),
...(settings.codec && { videoCodec: settings.codec }),
};
}
+36
View File
@@ -0,0 +1,36 @@
/*
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.
*/
/** LiveKit test doubles shared by the backend's unit tests. Not part of the package. */
import {
type RemoteParticipant,
type RemoteTrackPublication,
} from "livekit-client";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { mockEmitter } from "../../utils/test";
export const exampleTransport: LivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
export function mockRemoteParticipant(
participant: Partial<RemoteParticipant>,
): RemoteParticipant {
return {
isLocal: false,
setVolume() {},
getTrackPublication: () =>
({}) as Partial<RemoteTrackPublication> as RemoteTrackPublication,
// this will only get used for `getTrackPublications().length`
getTrackPublications: () => [0],
...mockEmitter(),
...participant,
} as RemoteParticipant;
}