Compare commits

...
Author SHA1 Message Date
Johannes Marbach 56d6091b98 Add token federation test 2026-10-08 08:55:19 +02:00
Johannes Marbach f5d3ee0891 Add delegation test 2026-10-08 08:45:07 +02:00
Johannes Marbach f86c3f7114 Add error tests 2026-10-08 08:26:03 +02:00
Johannes Marbach 4010411b81 Add e2e tests for endpoint usage 2026-10-07 18:08:16 +02:00
Robin ddfcde4a0d WIP 2026-10-07 15:08:29 +02:00
Robin e6fed4c458 WIP 2026-10-07 14:48:26 +02:00
Robin 0df8b98a95 Revert "Reset to matrix-js-sdk develop branch"
This reverts commit d74becdd22.
2026-10-06 17:50:14 +02:00
Robin 6d304d48dc WIP 2026-10-06 17:50:11 +02:00
Robin 395f7c363f WIP 2026-10-06 16:08:59 +02:00
Robin d74becdd22 Reset to matrix-js-sdk develop branch 2026-10-06 14:47:18 +02:00
Robin 4326455249 WIP 2026-10-05 12:00:46 +02:00
Robin 8a06976494 WIP 2026-10-03 01:27:05 +02:00
Robin 0030371ea7 WIP 2026-10-02 23:26:21 +02:00
42 changed files with 1377 additions and 913 deletions
+1 -1
View File
@@ -5,7 +5,7 @@ sender_localpart: "_lk_jwt_service"
namespaces:
users:
- exclusive: false
regex: "@.*:homeserver-1" # Cover all local users
regex: "@.*:synapse\\.othersite\\.m\\.localhost" # Cover all local users
# Required for Synapse to call POST /_matrix/app/v1/ping on this service.
url: "http://auth-server-1:16080"
# Enable membership look-ups via /is_joined.
+1 -1
View File
@@ -5,7 +5,7 @@ sender_localpart: "_lk_jwt_service"
namespaces:
users:
- exclusive: false
regex: "@.*:homeserver" # Cover all local users
regex: "@.*:synapse\\.m\\.localhost" # Cover all local users
# Required for Synapse to call POST /_matrix/app/v1/ping on this service.
url: "http://auth-server:6080"
# Enable membership look-ups via /is_joined.
+5
View File
@@ -6,6 +6,11 @@ rtc:
port_range_start: 50300
port_range_end: 50400
use_external_ip: false
# Advertise the loopback address instead of the container IP, which is not
# routable from the host when Docker runs in a VM (colima, Docker Desktop).
# The browser then reaches the SFU through the published ports. Firefox
# ignores loopback candidates, so this only works in Chromium-based browsers.
node_ip: 127.0.0.1
turn:
enabled: false
domain: localhost
+5
View File
@@ -6,6 +6,11 @@ rtc:
port_range_start: 50100
port_range_end: 50200
use_external_ip: false
# Advertise the loopback address instead of the container IP, which is not
# routable from the host when Docker runs in a VM (colima, Docker Desktop).
# The browser then reaches the SFU through the published ports. Firefox
# ignores loopback candidates, so this only works in Chromium-based browsers.
node_ip: 127.0.0.1
turn:
enabled: false
domain: localhost
+4 -2
View File
@@ -3,7 +3,8 @@ networks:
services:
auth-service:
image: ghcr.io/element-hq/lk-jwt-service:0.7.0
# Need to use rolling main until the `member_id` change has been released.
image: ghcr.io/element-hq/lk-jwt-service:latest-ci
pull_policy: always
hostname: auth-server
environment:
@@ -33,7 +34,8 @@ services:
- ecbackend
auth-service-1:
image: ghcr.io/element-hq/lk-jwt-service:0.7.0
# Need to use rolling main until the `member_id` change has been released.
image: ghcr.io/element-hq/lk-jwt-service:latest-ci
pull_policy: always
hostname: auth-server-1
environment:
+3 -3
View File
@@ -59,10 +59,10 @@
"developer_mode": {
"always_show_iphone_earpiece": "Show iPhone earpiece option on all platforms",
"crypto_version": "Crypto version: {{version}}",
"custom_livekit_url": {
"current_url": "Currently set to: ",
"custom_transport": {
"current": "Currently set to: ",
"from_config": "Currently, no overwrite is set. Url from config is used.",
"label": "Custom Livekit-url",
"label": "Custom transport JSON",
"reset": "Reset overwrite",
"save": "Save",
"saving": "Saving..."
+2 -2
View File
@@ -103,8 +103,8 @@
"livekit-client": "^2.18.1",
"lodash-es": "^4.17.21",
"loglevel": "^1.9.1",
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#develop",
"matrix-widget-api": "^1.18.0",
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#robin/livekit-updates",
"matrix-widget-api": "^1.20.0",
"node-stdlib-browser": "^1.3.1",
"normalize.css": "^8.0.1",
"oxfmt": "^0.66.0",
+421
View File
@@ -0,0 +1,421 @@
/*
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 { expect, type Page, test } from "@playwright/test";
import { SpaHelpers } from "./spa-helpers";
const sfuUrl = "wss://matrix-rtc.m.localhost/livekit/sfu";
const serviceUrl = "https://matrix-rtc.m.localhost/livekit/jwt";
const otherHomeserver = {
base_url: "https://synapse.othersite.m.localhost",
server_name: "synapse.othersite.m.localhost",
};
const otherSfuUrl = "wss://matrix-rtc.othersite.m.localhost/livekit/sfu";
interface Transport {
type: "livekit";
url?: string;
livekit_service_url?: string;
}
/**
* The transport shapes a homeserver may advertise while migrating to the
* latest MSC4195. `legacy` is what the e2e Synapse advertises on its own.
*/
const shapes = {
url: { type: "livekit", url: sfuUrl },
both: { type: "livekit", url: sfuUrl, livekit_service_url: serviceUrl },
legacy: null,
} satisfies Record<string, Transport | null>;
type Shape = keyof typeof shapes;
/** Which endpoint a page asks for its SFU tokens. */
type Flow = "cs-api" | "service-legacy" | "service-default";
const cases: {
hostShape: Shape;
guestShape: Shape;
hostFlows: Flow[];
guestFlows: Flow[];
}[] = [
{
hostShape: "url",
guestShape: "url",
hostFlows: ["cs-api"],
guestFlows: ["cs-api"],
},
{
hostShape: "both",
guestShape: "both",
hostFlows: ["cs-api"],
guestFlows: ["cs-api"],
},
{
hostShape: "url",
guestShape: "legacy",
hostFlows: ["cs-api", "service-default"],
guestFlows: ["service-legacy", "cs-api"],
},
];
for (const { hostShape, guestShape, hostFlows, guestFlows } of cases) {
test(`One to one call with ${hostShape} host and ${guestShape} guest transport`, async ({
browser,
page,
browserName,
}) => {
skipOnFirefox(browserName);
// The host starts the call.
await advertiseTransport(page, shapes[hostShape]);
const hostRequests = recordAuthRequests(page);
await page.goto("/");
await SpaHelpers.createCall(page, "Androl", "HelloCall", true, "2_0");
const inviteLink = await SpaHelpers.getCallInviteLink(page);
// The guest joins.
const guestContext = await browser.newContext({ reducedMotion: "reduce" });
const guestPage = await guestContext.newPage();
await advertiseTransport(guestPage, shapes[guestShape]);
const guestRequests = recordAuthRequests(guestPage);
await SpaHelpers.joinCallFromInviteLink(
guestPage,
inviteLink,
"Pevara",
"2_0",
);
// Wait for the call to connect and render.
await SpaHelpers.expectVideoTilesCount(page, 2);
await SpaHelpers.expectVideoTilesCount(guestPage, 2);
// The host and guest ran through the expected flows.
expect(flowsOf(hostRequests)).toEqual(new Set(hostFlows));
expect(flowsOf(guestRequests)).toEqual(new Set(guestFlows));
// All requests succeeded and the SFU URL was set on C-S requests.
for (const request of [...hostRequests, ...guestRequests]) {
expect(request.status).toBe(200);
if (request.flow === "cs-api") expect(request.sfuUrl).toBe(sfuUrl);
}
});
}
test("Subscriber falls back to the JWT service when the homeserver lacks MSC4195", async ({
browser,
page,
browserName,
}) => {
skipOnFirefox(browserName);
// The host starts the call with both transport shapes.
await advertiseTransport(page, shapes.both);
await page.goto("/");
await SpaHelpers.createCall(page, "Androl", "HelloCall", true, "2_0");
const inviteLink = await SpaHelpers.getCallInviteLink(page);
// The guest joins but their C-S requests fail.
const guestContext = await browser.newContext({ reducedMotion: "reduce" });
const guestPage = await guestContext.newPage();
await rejectCsApiTokens(guestPage);
const guestRequests = recordAuthRequests(guestPage);
await SpaHelpers.joinCallFromInviteLink(
guestPage,
inviteLink,
"Pevara",
"2_0",
);
// Wait for the call to connect and render.
await SpaHelpers.expectVideoTilesCount(page, 2);
await SpaHelpers.expectVideoTilesCount(guestPage, 2);
// The guest ran through all flows.
expect(flowsOf(guestRequests)).toEqual(
new Set<Flow>(["cs-api", "service-legacy", "service-default"]),
);
// The C-S token requests failed.
for (const request of guestRequests) {
expect(request.status).toBe(request.flow === "cs-api" ? 404 : 200);
}
// The C-S token requests preceeded the fallback request.
const flows = guestRequests.map((r) => r.flow);
expect(flows.indexOf("service-default")).toBeGreaterThan(
flows.indexOf("cs-api"),
);
});
test("Publisher shows an error when the homeserver lacks MSC4195", async ({
page,
browserName,
}) => {
skipOnFirefox(browserName);
// The host tries to start a call with only the C-S transport shape available but
// their C-S requests fail.
await advertiseTransport(page, shapes.url);
await rejectCsApiTokens(page);
await page.goto("/");
await SpaHelpers.createCall(page, "Androl", "HelloCall", true, "2_0");
// We land on an error page.
await expect(
page.getByRole("heading", { name: "Something went wrong" }),
).toBeVisible();
await expect(
page.getByText(
"The authorization service for your media server (SFU) is out of date.",
),
).toBeVisible();
});
test("Delegated leave on a url transport ends the membership when the guest drops", async ({
browser,
page,
browserName,
}) => {
skipOnFirefox(browserName);
// The host starts the call.
await advertiseTransport(page, shapes.url);
await page.goto("/");
await SpaHelpers.createCall(page, "Androl", "HelloCall", true, "2_0");
const inviteLink = await SpaHelpers.getCallInviteLink(page);
// The guest joins.
const guestContext = await browser.newContext({ reducedMotion: "reduce" });
const guestPage = await guestContext.newPage();
await advertiseTransport(guestPage, shapes.url);
const delayedLeave = recordDelayedLeave(guestPage);
const delegations = recordDelegations(guestPage);
await SpaHelpers.joinCallFromInviteLink(
guestPage,
inviteLink,
"Pevara",
"2_0",
);
// Wait for the call to connect and render.
await SpaHelpers.expectVideoTilesCount(page, 2);
await SpaHelpers.expectVideoTilesCount(guestPage, 2);
// The guest handed its delayed leave to the homeserver, after probing it for
// delegation support with bodiless requests.
await expect
.poll(() => delegations.at(-1))
.toEqual({ status: 200, sfuUrl, delayId: delayedLeave.delayId });
const probes = delegations.slice(0, -1);
expect(probes.length).toBeGreaterThan(0);
for (const probe of probes) expect(probe).toEqual({ status: 401 });
// The delay is long enough that a leave within the window below cannot come
// from the delayed event expiring on its own.
expect(delayedLeave.delayMs).toBeGreaterThan(60_000);
// The guest drops without leaving.
await guestPage.reload();
// The host eventually sees the guest go once the delayed event fires.
await expect(page.getByTestId("videoTile")).toHaveCount(1, {
timeout: 30_000,
});
});
test("Subscribers get tokens for a remote homeserver's SFU over federation", async ({
browser,
page,
browserName,
}) => {
skipOnFirefox(browserName);
// The host lives on the other homeserver and publishes on its SFU.
await useHomeserver(page, otherHomeserver);
await advertiseTransport(page, { type: "livekit", url: otherSfuUrl });
const hostRequests = recordAuthRequests(page);
await page.goto("/");
await SpaHelpers.createCall(page, "Androl", "HelloCall", true, "2_0");
const inviteLink = await SpaHelpers.getCallInviteLink(page);
// The guest lives on the default homeserver and joins over federation.
const guestContext = await browser.newContext({ reducedMotion: "reduce" });
const guestPage = await guestContext.newPage();
await advertiseTransport(guestPage, shapes.url);
const guestRequests = recordAuthRequests(guestPage);
await SpaHelpers.joinCallFromInviteLink(
guestPage,
`${inviteLink}&viaServers=${otherHomeserver.server_name}`,
"Pevara",
"2_0",
);
// Wait for the call to connect and render.
await SpaHelpers.expectVideoTilesCount(page, 2);
await SpaHelpers.expectVideoTilesCount(guestPage, 2);
// Each side got a token for its own SFU and for the other side's SFU, all
// through its own homeserver.
for (const requests of [hostRequests, guestRequests]) {
expect(flowsOf(requests)).toEqual(new Set<Flow>(["cs-api"]));
expect(new Set(requests.map((r) => r.sfuUrl))).toEqual(
new Set([sfuUrl, otherSfuUrl]),
);
for (const request of requests) expect(request.status).toBe(200);
}
});
/** Skips the calling test if we're running on Firefox. */
function skipOnFirefox(browserName: string): void {
test.skip(
browserName === "firefox",
"The is test is not working on firefox CI environment. No mic/audio device inputs so cam/mic are disabled",
);
}
/** Points this page's Element Call at another homeserver. */
async function useHomeserver(
page: Page,
homeserver: typeof otherHomeserver,
): Promise<void> {
await page.route("**/config.json", async (route) => {
const response = await route.fetch();
const config = (await response.json()) as Record<string, unknown>;
await route.fulfill({
response,
json: {
...config,
default_server_config: { "m.homeserver": homeserver },
},
});
});
}
/**
* Maps recorded requests to a set of flows. Publisher and subscriber connections
* start concurrently so using arrays for comparison isn't convenient.
*/
function flowsOf(requests: AuthRequest[]): Set<Flow> {
return new Set(requests.map((r) => r.flow));
}
/**
* Replaces the transports the homeserver advertises to this page. The real
* response is fetched first so its CORS headers carry over.
*/
async function advertiseTransport(
page: Page,
transport: Transport | null,
): Promise<void> {
if (transport === null) return;
await page.route("**/org.matrix.msc4143/rtc/transports", async (route) => {
if (route.request().method() !== "GET") return route.continue();
const response = await route.fetch();
await route.fulfill({ response, json: { rtc_transports: [transport] } });
});
}
/**
* Makes the homeserver look like one without MSC4195: its `get_token`
* endpoint answers as Synapse does for an unknown route.
*/
async function rejectCsApiTokens(page: Page): Promise<void> {
await page.route("**/rtc/livekit/get_token", async (route) => {
if (route.request().method() !== "POST") return route.continue();
await route.fulfill({
status: 404,
json: { errcode: "M_UNRECOGNIZED", error: "Unrecognized request" },
headers: { "Access-Control-Allow-Origin": "*" },
});
});
}
/** A recorded delayed leave event. */
interface DelayedLeave {
delayMs?: number;
delayId?: string;
}
/** Records the delayed leave event the page schedules on joining. */
function recordDelayedLeave(page: Page): DelayedLeave {
const delayedLeave: DelayedLeave = {};
page.on("response", (response) => {
const delay = new URL(response.request().url()).searchParams.get(
"org.matrix.msc4140.delay",
);
if (delay === null || response.status() !== 200) return;
delayedLeave.delayMs = Number(delay);
response
.json()
.then((body: { delay_id: string }) => {
delayedLeave.delayId = body.delay_id;
})
.catch(() => {});
});
return delayedLeave;
}
/** A recorded delegation request for a delayed leave event. */
interface Delegation {
status: number;
sfuUrl?: string;
delayId?: string;
}
/** Records requests to the MSC4195 `delegate_delayed_leave` endpoint. */
function recordDelegations(page: Page): Delegation[] {
const delegations: Delegation[] = [];
page.on("response", (response) => {
const request = response.request();
if (
request.method() !== "POST" ||
!request.url().endsWith("/rtc/livekit/delegate_delayed_leave")
)
return;
const body = request.postDataJSON() as {
url: string;
delay_id: string;
} | null;
delegations.push({
status: response.status(),
...(body && { sfuUrl: body.url, delayId: body.delay_id }),
});
});
return delegations;
}
/** A recorded token request. */
interface AuthRequest {
flow: Flow;
status: number;
/** The SFU a request to the MSC4195 `get_token` endpoint asked for. */
sfuUrl?: string;
}
/** Records token requests. */
function recordAuthRequests(page: Page): AuthRequest[] {
const requests: AuthRequest[] = [];
page.on("response", (response) => {
const request = response.request();
if (request.method() !== "POST") return;
const url = request.url();
if (url.endsWith("/rtc/livekit/get_token")) {
requests.push({
flow: "cs-api",
status: response.status(),
sfuUrl: (request.postDataJSON() as { url: string }).url,
});
} else if (url.endsWith("/livekit/jwt/sfu/get")) {
requests.push({ flow: "service-legacy", status: response.status() });
} else if (url.endsWith("/livekit/jwt/get_token")) {
requests.push({ flow: "service-default", status: response.status() });
}
});
return requests;
}
+11 -11
View File
@@ -191,11 +191,11 @@ importers:
specifier: ^1.9.1
version: 1.9.2
matrix-js-sdk:
specifier: github:matrix-org/matrix-js-sdk#develop
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/fa4772a4ddd335a5a8bfbf8bb21df49b70f7a999
specifier: github:matrix-org/matrix-js-sdk#robin/livekit-updates
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/621b225e227dc6dc2b1c9d53413235d8999e6d54
matrix-widget-api:
specifier: ^1.18.0
version: 1.19.0
specifier: ^1.20.0
version: 1.20.0
node-stdlib-browser:
specifier: ^1.3.1
version: 1.3.1
@@ -4673,13 +4673,13 @@ packages:
matrix-events-sdk@0.0.1:
resolution: {integrity: sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==}
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/fa4772a4ddd335a5a8bfbf8bb21df49b70f7a999:
resolution: {gitHosted: true, integrity: sha512-qF8++74MX36KS1KEw6Lc8HHXgZ6nHD1oUU4yBZUj5vVPvZTIspk1lOGztkFmizE5vmLAHd5tdadL9Xu1Asy4bA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/fa4772a4ddd335a5a8bfbf8bb21df49b70f7a999}
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/621b225e227dc6dc2b1c9d53413235d8999e6d54:
resolution: {gitHosted: true, integrity: sha512-HeTjG3NPGeCMkfPfGPaYl1QBnn7XYRiC1NpBgOxflujzQmTBNTPshGSTzMsMQKH/hpfi8QtjPE1SddNzzax3vA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/621b225e227dc6dc2b1c9d53413235d8999e6d54}
version: 42.3.0
engines: {node: '>=22.0.0'}
matrix-widget-api@1.19.0:
resolution: {integrity: sha512-Qe7L6G1mQJphBC47/Bzp5ljPIK1UARgLiuGgS+/KmEs/eaNHkQ3+ArD0/0kvMV+d6kCGAZ9OKRqsnuPjizI+tw==}
matrix-widget-api@1.20.0:
resolution: {integrity: sha512-qwFP37rgns7mPesM5/vvZgknzb+kvyRB5BISv5sCuBxsGMceMr+Kogf7OkjRxBXdMCPBFuf6gIuYRe5S+lpzQw==}
md5.js@1.3.5:
resolution: {integrity: sha512-xitP+WxNPcTTOgnTJcrhM0xvdPepipPSf3I8EIpGKeFLjt3PlJLIDG3u8EX53ZIubkb+5U2+3rELYpEhHhzdkg==}
@@ -10227,7 +10227,7 @@ snapshots:
matrix-events-sdk@0.0.1: {}
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/fa4772a4ddd335a5a8bfbf8bb21df49b70f7a999:
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/621b225e227dc6dc2b1c9d53413235d8999e6d54:
dependencies:
'@babel/runtime': 8.0.0
'@matrix-org/matrix-sdk-crypto-wasm': 18.9.0
@@ -10236,12 +10236,12 @@ snapshots:
content-type: 3.1.1
loglevel: 1.9.2
matrix-events-sdk: 0.0.1
matrix-widget-api: 1.19.0
matrix-widget-api: 1.20.0
p-retry: 8.0.1
sdp-transform: 3.0.0
unhomoglyph: 1.0.6
matrix-widget-api@1.19.0:
matrix-widget-api@1.20.0:
dependencies:
'@types/events': 3.0.3
events: 3.3.0
+76
View File
@@ -0,0 +1,76 @@
/*
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 CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { extractFullConfigFromToken, type SFUConfig } from "./types.ts";
import { doNetworkOperationWithRetry } from "../../utils/matrix.ts";
// TODO: This should come from the `MatrixRTCSession`s slot description instead
// of being hardcoded here. (the legacy flow hardcodes it as well)
const SLOT_ID = "m.call#ROOM";
export type ClientGetTokenParts = Pick<
MatrixClient,
"_unstable_getLivekitToken"
>;
export interface GetSFUConfigParams {
client: ClientGetTokenParts;
membership: CallMembershipIdentityParts;
url: string;
roomId: string;
}
export async function getSFUConfig({
client,
membership,
url,
roomId,
}: GetSFUConfigParams): Promise<SFUConfig> {
const res = await doNetworkOperationWithRetry(async () =>
client._unstable_getLivekitToken({
url,
room_id: roomId,
slot_id: "m.call#ROOM",
member_id: membership.memberId,
}),
);
return extractFullConfigFromToken({ url, jwt: res.jwt });
}
export type ClientDelegationParts = Pick<
MatrixClient,
"_unstable_delegateDelayedLeave" | "baseUrl"
>;
export interface DelegateDelayedLeaveParams {
client: ClientDelegationParts;
membership: CallMembershipIdentityParts;
url: string;
roomId: string;
delayId: string;
}
export async function delegateDelayedLeave({
client,
membership,
url,
roomId,
delayId,
}: DelegateDelayedLeaveParams): Promise<void> {
await doNetworkOperationWithRetry(async () =>
client._unstable_delegateDelayedLeave({
url,
room_id: roomId,
slot_id: SLOT_ID,
member_id: membership.memberId,
delay_id: delayId,
}),
);
}
+143
View File
@@ -0,0 +1,143 @@
/*
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 CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { MatrixError } from "matrix-js-sdk";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import * as openID from "./openID";
import * as csApi from "./csApi";
import { type SFUConfig } from "./types";
import { NoMatrix2AuthorizationService } from "../../utils/errors";
export type { ClientOpenIDParts } from "./openID";
export type { ClientGetTokenParts, ClientDelegationParts } from "./csApi";
export type * from "./types";
/**
* Checks whether an error means "this homeserver does not implement the
* (unstable) MSC4195 endpoints".
*
* MSC4195 documents `M_NOT_FOUND` for this, but a homeserver without support
* answers with whatever it uses for unknown endpoints (Synapse: 404 with
* `M_UNRECOGNIZED`), so we accept any 404 as "unsupported".
* @param e The error to check.
*/
function isEndpointUnsupported(e: unknown): boolean {
return (
e instanceof MatrixError &&
(e.httpStatus === 404 ||
e.errcode === "M_NOT_FOUND" ||
e.errcode === "M_UNRECOGNIZED")
);
}
export interface GetSFUConfigParams {
client: csApi.ClientGetTokenParts & openID.ClientOpenIDParts;
membership: CallMembershipIdentityParts;
transport: UnstableLivekitTransport;
roomId: string;
role: "publisher" | "subscriber";
logger: Logger;
}
export async function getSFUConfig({
client,
membership,
transport,
roomId,
role,
logger,
}: GetSFUConfigParams): Promise<SFUConfig> {
if ("url" in transport) {
try {
return await csApi.getSFUConfig({
client,
membership,
url: transport.url,
roomId,
});
} catch (e) {
if (isEndpointUnsupported(e)) {
const mayFallBack =
role === "subscriber" && "livekit_service_url" in transport;
if (mayFallBack) {
logger.warn(
`Homeserver does not support the MSC4195 get_token endpoint, falling back to the OpenID flow with service ${transport.livekit_service_url}`,
);
// Fall through
} else {
logger.error(
`Homeserver does not support the MSC4195 get_token endpoint, cannot get token for transport ${transport.url}`,
);
throw new NoMatrix2AuthorizationService(e as Error);
}
} else {
throw e;
}
}
}
return await openID.getSFUConfig({
client,
membership,
serviceUrl: transport.livekit_service_url,
roomId,
role,
logger,
});
}
export interface DelegateDelayedLeaveParams {
client: csApi.ClientDelegationParts & openID.ClientOpenIDParts;
membership: CallMembershipIdentityParts;
transport: UnstableLivekitTransport;
roomId: string;
delayId: string;
logger: Logger;
}
export async function delegateDelayedLeave({
client,
membership,
transport,
roomId,
delayId,
logger,
}: DelegateDelayedLeaveParams): Promise<void> {
if ("url" in transport) {
try {
await csApi.delegateDelayedLeave({
client,
membership,
url: transport.url,
roomId,
delayId,
});
} catch (e) {
if (isEndpointUnsupported(e))
logger.error(
"Server misconfigured: transport contains a 'url' field but server does not support the MSC4195 delegate_delayed_leave endpoint",
);
throw e;
}
} else {
// This will technically cause the service to issue a new JWT token, but
// it's safe to discard. We're only interested in triggering delegation.
await openID.getSFUConfig({
client,
membership,
serviceUrl: transport.livekit_service_url,
roomId,
role: "publisher",
delayEndpointBaseUrl: client.baseUrl,
delayId,
logger,
});
}
}
@@ -17,21 +17,40 @@ import {
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 "../utils/errors";
import {
getSFUConfig,
type GetSFUConfigParams,
type ClientOpenIDParts,
} from "./openID";
import { testJWTToken } from "../../utils/test-fixtures";
import { ownMemberMock } from "../../utils/test";
import { FailToGetOpenIdToken } from "../../utils/errors";
import { logger } from "matrix-js-sdk/lib/logger";
const sfuUrl = "https://sfu.example.org";
const serviceUrl = "https://sfu.example.org";
describe("getSFUConfigWithOpenID", () => {
let matrixClient: MockedObject<OpenIDClientParts>;
const delayParams = {
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
};
describe("getSFUConfig", () => {
let matrixClient: MockedObject<ClientOpenIDParts>;
let params: GetSFUConfigParams;
beforeEach(() => {
fetchMock.catch(404);
matrixClient = {
getOpenIdToken: vitest.fn(),
getDeviceId: vitest.fn(),
};
params = {
client: matrixClient,
membership: ownMemberMock,
serviceUrl: "https://sfu.example.org",
roomId: "!example_room_id",
role: "subscriber",
logger,
};
});
afterEach(() => {
vitest.clearAllMocks();
@@ -42,18 +61,13 @@ describe("getSFUConfigWithOpenID", () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
body: { url: serviceUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
const config = await getSFUConfig(params);
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
url: serviceUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
@@ -71,12 +85,7 @@ describe("getSFUConfigWithOpenID", () => {
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
await getSFUConfig(params);
} catch (ex: unknown) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
@@ -116,23 +125,14 @@ describe("getSFUConfigWithOpenID", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
body: { url: serviceUrl, 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",
},
);
// Note: Assuming getSFUConfig eventually calls getLiveKitJWT
const config = await getSFUConfig({ ...params, ...delayParams });
expect(config.jwt).toBe(testJWTToken);
expect(callCount).toBe(2);
@@ -153,7 +153,7 @@ describe("getSFUConfigWithOpenID", () => {
) {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
body: { url: serviceUrl, jwt: testJWTToken },
};
}
return {
@@ -164,21 +164,12 @@ describe("getSFUConfigWithOpenID", () => {
{ 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",
},
);
const config = await getSFUConfig({ ...params, ...delayParams });
// Prüfe das Ergebnis
expect(config).toMatchObject({
jwt: testJWTToken,
url: sfuUrl,
url: serviceUrl,
});
void (await fetchMock.flush());
@@ -204,16 +195,7 @@ describe("getSFUConfigWithOpenID", () => {
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
},
);
await getSFUConfig({ ...params, ...delayParams });
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toBeInstanceOf(MatrixError);
@@ -251,7 +233,7 @@ describe("getSFUConfigWithOpenID", () => {
fetchMock.post("https://sfu.example.org/get_token", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
body: { url: serviceUrl, jwt: testJWTToken },
};
});
fetchMock.post("https://sfu.example.org/sfu/get", () => {
@@ -261,16 +243,7 @@ describe("getSFUConfigWithOpenID", () => {
};
});
try {
await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
{
delayEndpointBaseUrl: "https://matrix.homeserverserver.org",
delayId: "mock_delay_id",
},
);
await getSFUConfig({ ...params, ...delayParams });
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
expect((ex as FailToGetOpenIdToken).cause).toEqual(
@@ -309,18 +282,13 @@ describe("getSFUConfigWithOpenID", () => {
fetchMock.post("https://sfu.example.org/sfu/get", () => {
return {
status: 200,
body: { url: sfuUrl, jwt: testJWTToken },
body: { url: serviceUrl, jwt: testJWTToken },
};
});
const config = await getSFUConfigWithOpenID(
matrixClient,
ownMemberMock,
"https://sfu.example.org",
"!example_room_id",
);
const config = await getSFUConfig({ ...params, ...delayParams });
expect(config).toEqual({
jwt: testJWTToken,
url: sfuUrl,
url: serviceUrl,
livekitIdentity: "@me:example.org:ABCDEF",
livekitAlias: "!example_room_id",
});
@@ -13,95 +13,64 @@ import {
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
FailToGetOpenIdToken,
NoMatrix2AuthorizationService,
} from "../utils/errors";
import { doNetworkOperationWithRetry } from "../utils/matrix";
import { Config } from "../config/Config";
import { MatrixRTCMode } from "../config/ConfigOptions";
/**
* 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;
};
}
import { FailToGetOpenIdToken } from "../../utils/errors";
import { doNetworkOperationWithRetry } from "../../utils/matrix";
import { Config } from "../../config/Config";
import { extractFullConfigFromToken, type SFUConfig } from "./types";
// The bits we need from MatrixClient
export type OpenIDClientParts = Pick<
export type ClientOpenIDParts = Pick<
MatrixClient,
"getOpenIdToken" | "getDeviceId"
>;
export interface GetSFUConfigParams {
client: ClientOpenIDParts;
/**
* Our own membership identity parts used to send to the JWT service.
*/
membership: CallMembershipIdentityParts;
/**
* The base URL of the LiveKit JWT service.
*/
serviceUrl: string;
/**
* The room ID used in the request. This is NOT the livekit_alias. The JWT
* service will provide the alias. It maps Matrix room IDs to LiveKit aliases.
*/
roomId: string;
/**
* Whether we want to publish or only subscribe on this transport.
*/
role: "publisher" | "subscriber";
/**
* The base URL of the Matrix homeserver, for delayed leave delegation.
*/
delayEndpointBaseUrl?: string;
/**
* The delay ID of the leave event to be delegated to the JWT service.
*/
delayId?: string;
logger: Logger;
}
/**
* 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.matrixRTCMode Determines which version of the JWT endpoint to use, which affects whether the
* RTC backend identity is based on string concatenation (legacy) or a hash (Matrix 2.0).
* 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?: {
matrixRTCMode?: MatrixRTCMode;
delayEndpointBaseUrl?: string;
delayId?: string;
},
logger?: Logger,
): Promise<SFUConfig> {
export async function getSFUConfig({
client,
membership,
serviceUrl,
roomId,
role,
delayEndpointBaseUrl,
delayId,
logger,
}: GetSFUConfigParams): Promise<SFUConfig> {
let openIdToken: IOpenIDToken;
try {
openIdToken = await doNetworkOperationWithRetry(async () =>
@@ -112,82 +81,50 @@ export async function getSFUConfigWithOpenID(
error instanceof Error ? error : new Error("Unknown error"),
);
}
logger?.debug("Got openID token", openIdToken);
logger.debug("Got openID token", openIdToken);
let sfuConfig: { url: string; jwt: string } | undefined;
const tryBothJwtEndpoints = opts?.matrixRTCMode === undefined; // This is for SFUs where we do not publish.
const forceMatrix2Jwt = opts?.matrixRTCMode === MatrixRTCMode.Matrix_2_0;
// 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,
);
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);
}
}
let endpoint: "default" | "legacy";
switch (role) {
case "publisher":
// When publishing a legacy transport (one with a `livekit_service_url`),
// subscribers will expect our participant identity to use the legacy
// `@user_id:device_id` format. Only the legacy JWT service endpoint
// assigns identities in this format, so we must use it.
endpoint = "legacy";
break;
case "subscriber":
// Use the default endpoint as it's more likely to allow remote access.
endpoint = "default";
break;
}
// 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}...`,
logger.info(
`Trying to get JWT with ${endpoint} endpoint for focus ${serviceUrl}...`,
);
sfuConfig = await getLiveKitJWTLegacy(
membership.deviceId,
sfuConfig = await (
endpoint === "legacy" ? getLiveKitJWTLegacy : getLiveKitJWT
)(
membership,
serviceUrl,
roomId,
openIdToken,
opts?.delayEndpointBaseUrl,
opts?.delayId,
delayEndpointBaseUrl,
delayId,
);
logger?.info(`Got JWT from call's active focus URL.`);
logger.info(`Got JWT from call's active focus URL.`);
return extractFullConfigFromToken(sfuConfig);
} catch (ex) {
} catch (e) {
logger.error(`Failed fetching jwt with ${endpoint} endpoint:`, e);
throw new FailToGetOpenIdToken(
ex instanceof Error ? ex : new Error(`Unknown error ${ex}`),
e instanceof Error ? e : new Error(`Unknown error ${e}`),
);
}
}
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,
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
openIDToken: IOpenIDToken,
@@ -221,7 +158,7 @@ async function getLiveKitJWTLegacy(
// However, the livekit room alias is provided as part of the JWT payload.
room: matrixRoomId,
openid_token: openIDToken,
device_id: deviceId,
device_id: membership.deviceId,
...delayParts,
}),
});
@@ -261,7 +198,7 @@ class NotSupportedError extends Error {
}
}
export async function getLiveKitJWT(
async function getLiveKitJWT(
membership: CallMembershipIdentityParts,
livekitServiceURL: string,
matrixRoomId: string,
+79
View File
@@ -0,0 +1,79 @@
/*
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.
*/
/**
* Configuration and access tokens provided by the SFU on successful authentication.
*/
export interface SFUConfig {
/**
* The WebSocket URL of the LiveKit SFU. This is what we connect the LiveKit
* room to. Note that this is NOT the JWT service URL
* (`livekit_service_url`): the legacy JWT service returns this URL as part of
* its response, while the CS-Api based flow requires us to know it upfront.
*/
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;
};
}
/**
* Complements the SFU websocket url and the JWT with the information encoded in
* the JWT payload itself.
* @param sfuConfig The SFU websocket url and the JWT to connect with.
* @returns The full SFU config, including the LiveKit alias and identity.
*/
export 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,
};
}
+2 -2
View File
@@ -17,7 +17,7 @@ import {
import { act } from "react";
import { type RoomMember } from "matrix-js-sdk";
import {
type LivekitTransport,
type UnstableLivekitTransport,
type CallMembership,
} from "matrix-js-sdk/lib/matrixrtc";
@@ -46,7 +46,7 @@ vitest.mock("../useAudioContext");
vitest.mock("../soundUtils");
vitest.mock("../rtcSessionHelpers", async (importOriginal) => ({
...(await importOriginal()),
makeTransport: (): [LivekitTransport] => [exampleTransport],
makeTransport: (): [UnstableLivekitTransport] => [exampleTransport],
}));
afterEach(() => {
+13 -9
View File
@@ -694,15 +694,19 @@ export const InCallView: FC<InCallViewProps> = ({
tab={settingsTab}
onTabChange={setSettingsTab}
developerSettingsVm={developerSettingsVm}
livekitRooms={allConnections
.getConnections()
.map((connectionItem) => ({
room: connectionItem.livekitRoom,
livekitAlias: connectionItem.livekitAlias,
// TODO compute is local or tag it in the livekit room items already
isLocal: undefined,
url: connectionItem.transport.livekit_service_url,
}))}
livekitRooms={allConnections.map(
({ livekitRoom: room, livekitAlias, transport, role }) => ({
room,
livekitAlias,
role,
url:
// It should be relatively clear to the user which kind of URL
// this is, based on the scheme (ws:// vs. http://)
"url" in transport
? transport.url
: transport.livekit_service_url,
}),
)}
/>
</>
)}
+15 -15
View File
@@ -18,14 +18,14 @@ import {
} from "./DeveloperSettingsTab";
import { outOfCallDeveloperSettingsTabViewModel } from "./DeveloperSettingsTabViewModel";
import { createStaticViewModel } from "../state/ViewModel";
import { getSFUConfigWithOpenID } from "../livekit/openIDSFU";
import {
customLivekitUrl as customLivekitUrlSetting,
customTransport as customTransportSetting,
enableExtendedLivekitLogs as enableExtendedLivekitLogsSetting,
matrixRTCMode as matrixRTCModeSetting,
} from "./settings";
import { MatrixRTCMode } from "../config/ConfigOptions";
import { mockConfig } from "../utils/test";
import { getSFUConfig } from "../livekit/auth";
// Mock url params hook to avoid environment-dependent snapshot churn.
vi.mock("../UrlParams", () => ({
@@ -36,8 +36,8 @@ vi.mock("../UrlParams", () => ({
}));
// IMPORTANT: mock the same specifier used by DeveloperSettingsTab
vi.mock("../livekit/openIDSFU", () => ({
getSFUConfigWithOpenID: vi.fn().mockResolvedValue({
vi.mock("../livekit/auth", () => ({
getSFUConfig: vi.fn().mockResolvedValue({
url: "mock-url",
jwt: "mock-jwt",
}),
@@ -126,7 +126,7 @@ describe("DeveloperSettingsTab", () => {
});
describe("custom livekit url", () => {
afterEach(() => {
customLivekitUrlSetting.setValue(null);
customTransportSetting.setValue(null);
});
const client = {
doesServerSupportUnstableFeature: vi.fn().mockResolvedValue(true),
@@ -153,9 +153,9 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
expect(customTransportSetting.getValue()).toBe(null);
});
it("will not update custom livekit url without text in input", async () => {
const user = userEvent.setup();
@@ -176,9 +176,9 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
expect(customTransportSetting.getValue()).toBe(null);
});
it("will not update custom livekit url when pressing cancel", async () => {
const user = userEvent.setup();
@@ -202,9 +202,9 @@ describe("DeveloperSettingsTab", () => {
name: "Reset overwrite",
});
await user.click(cancelButton);
expect(getSFUConfigWithOpenID).not.toHaveBeenCalled();
expect(getSFUConfig).not.toHaveBeenCalled();
expect(customLivekitUrlSetting.getValue()).toBe(null);
expect(customTransportSetting.getValue()).toBe(null);
});
it("will update custom livekit url", async () => {
const user = userEvent.setup();
@@ -226,14 +226,14 @@ describe("DeveloperSettingsTab", () => {
const saveButton = screen.getByRole("button", { name: "Save" });
await user.click(saveButton);
expect(getSFUConfigWithOpenID).toHaveBeenCalledWith(
expect(getSFUConfig).toHaveBeenCalledWith(
expect.anything(),
expect.anything(),
"wss://example.livekit.valid",
"#testRoom",
);
expect(customLivekitUrlSetting.getValue()).toBe(
expect(customTransportSetting.getValue()).toBe(
"wss://example.livekit.valid",
);
});
@@ -256,14 +256,14 @@ describe("DeveloperSettingsTab", () => {
await user.type(input, "wss://example.livekit.valid");
const saveButton = screen.getByRole("button", { name: "Save" });
(getSFUConfigWithOpenID as Mock).mockImplementation(() => {
(getSFUConfig as Mock).mockImplementation(() => {
throw new Error("Invalid URL");
});
await user.click(saveButton);
expect(
screen.getByText("invalid URL (did not update)"),
).toBeInTheDocument();
expect(customLivekitUrlSetting.getValue()).toBe(null);
expect(customTransportSetting.getValue()).toBe(null);
});
});
+42 -34
View File
@@ -45,7 +45,7 @@ import {
muteAllAudio as muteAllAudioSetting,
alwaysShowIphoneEarpiece as alwaysShowIphoneEarpieceSetting,
matrixRTCMode as matrixRTCModeSetting,
customLivekitUrl as customLivekitUrlSetting,
customTransport as customTransportSetting,
advancedScreenShare as advancedScreenShareSetting,
screenShareResolution as screenShareResolutionSetting,
screenShareFramerate as screenShareFramerateSetting,
@@ -67,9 +67,9 @@ import styles from "./DeveloperSettingsTab.module.css";
import settingsStyles from "./SettingsModal.module.css";
import { Slider } from "../Slider";
import { useUrlParams } from "../UrlParams";
import { getSFUConfigWithOpenID } from "../livekit/openIDSFU";
import { useBehavior } from "../useBehavior";
import { type ViewModel } from "../state/ViewModel.ts";
import { getSFUConfig } from "../livekit/auth/index.ts";
/**
* The state of MatrixRTC's media key rotation.
@@ -167,16 +167,18 @@ export const DeveloperSettingsTab: FC<Props> = ({
enableExtendedLivekitLogsSetting,
);
const [customLivekitUrlUpdateError, setCustomLivekitUrlUpdateError] =
useState<string | null>(null);
const [customLivekitUrl, setCustomLivekitUrl] = useSetting(
customLivekitUrlSetting,
const [customTransportUpdateError, setCustomLivekitUrlUpdateError] = useState<
string | null
>(null);
const [customTransport, setCustomTransport] = useSetting(
customTransportSetting,
);
const [customTransportTextBuffer, setCustomLivekitUrlTextBuffer] = useState(
() => JSON.stringify(customTransport),
);
const [customLivekitUrlTextBuffer, setCustomLivekitUrlTextBuffer] =
useState(customLivekitUrl);
useEffect(() => {
setCustomLivekitUrlTextBuffer(customLivekitUrl);
}, [customLivekitUrl]);
setCustomLivekitUrlTextBuffer(JSON.stringify(customTransport));
}, [customTransport]);
const [muteAllAudio, setMuteAllAudio] = useSetting(muteAllAudioSetting);
@@ -491,23 +493,23 @@ export const DeveloperSettingsTab: FC<Props> = ({
<EditInPlace
onSubmit={(e) => e.preventDefault()}
helpLabel={
customLivekitUrl === null
? t("developer_mode.custom_livekit_url.from_config")
: t("developer_mode.custom_livekit_url.current_url") +
customLivekitUrl
customTransport === null
? t("developer_mode.custom_transport.from_config")
: t("developer_mode.custom_transport.current") +
JSON.stringify(customTransport)
}
label={t("developer_mode.custom_livekit_url.label")}
saveButtonLabel={t("developer_mode.custom_livekit_url.save")}
savingLabel={t("developer_mode.custom_livekit_url.saving")}
cancelButtonLabel={t("developer_mode.custom_livekit_url.reset")}
label={t("developer_mode.custom_transport.label")}
saveButtonLabel={t("developer_mode.custom_transport.save")}
savingLabel={t("developer_mode.custom_transport.saving")}
cancelButtonLabel={t("developer_mode.custom_transport.reset")}
onSave={useCallback(
async (e: React.FormEvent<HTMLFormElement>): Promise<void> => {
if (
roomId === undefined ||
customLivekitUrlTextBuffer === "" ||
customLivekitUrlTextBuffer === null
customTransportTextBuffer === "" ||
customTransportTextBuffer === null
) {
setCustomLivekitUrl(null);
setCustomTransport(null);
return;
}
@@ -518,21 +520,27 @@ export const DeveloperSettingsTab: FC<Props> = ({
if (userId === null || deviceId === null) {
throw new Error("Invalid user or device ID");
}
await getSFUConfigWithOpenID(
const transport = JSON.parse(customTransportTextBuffer);
// Validate the transport by attempting to get a token for it
await getSFUConfig({
client,
{ userId, deviceId, memberId: "" },
customLivekitUrlTextBuffer,
membership: { userId, deviceId, memberId: "" },
transport,
roomId,
);
role: "publisher",
logger,
});
setCustomLivekitUrlUpdateError(null);
setCustomLivekitUrl(customLivekitUrlTextBuffer);
setCustomTransport(transport);
} catch {
setCustomLivekitUrlUpdateError("invalid URL (did not update)");
setCustomLivekitUrlUpdateError(
"invalid transport (did not update)",
);
}
},
[customLivekitUrlTextBuffer, setCustomLivekitUrl, client, roomId],
[customTransportTextBuffer, setCustomTransport, client, roomId],
)}
value={customLivekitUrlTextBuffer ?? ""}
value={customTransportTextBuffer ?? ""}
onChange={useCallback(
(event: ChangeEvent<HTMLInputElement>): void => {
setCustomLivekitUrlTextBuffer(event.target.value);
@@ -541,14 +549,14 @@ export const DeveloperSettingsTab: FC<Props> = ({
)}
onCancel={useCallback(
(e: React.FormEvent<HTMLFormElement>) => {
setCustomLivekitUrl(null);
setCustomTransport(null);
},
[setCustomLivekitUrl],
[setCustomTransport],
)}
serverInvalid={customLivekitUrlUpdateError !== null}
serverInvalid={customTransportUpdateError !== null}
>
{customLivekitUrlUpdateError !== null && (
<ErrorMessage>{customLivekitUrlUpdateError}</ErrorMessage>
{customTransportUpdateError !== null && (
<ErrorMessage>{customTransportUpdateError}</ErrorMessage>
)}
</EditInPlace>
<Separator />
+1 -1
View File
@@ -59,7 +59,7 @@ interface Props {
livekitRooms?: {
room: LivekitRoom;
url: string;
isLocal?: boolean;
role: "publisher" | "subscriber";
}[];
/** Only available while in a call. Used by the developer tab. */
developerSettingsVm?: ViewModel<DeveloperSettingsSnapshot>;
+3 -2
View File
@@ -12,6 +12,7 @@ import { PosthogAnalytics } from "../analytics/PosthogAnalytics";
import { type Behavior } from "../state/Behavior";
import { useBehavior } from "../useBehavior";
import { MatrixRTCMode } from "../config/ConfigOptions";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
export class Setting<T> {
public constructor(
@@ -153,8 +154,8 @@ export const matrixRTCMode = new Setting<MatrixRTCMode>(
MatrixRTCMode.Compatibility,
);
export const customLivekitUrl = new Setting<string | null>(
"custom-livekit-url",
export const customTransport = new Setting<UnstableLivekitTransport | null>(
"custom-transport",
null,
);
@@ -31,7 +31,7 @@ import {
type CallMembership,
type IRTCNotificationContent,
MatrixRTCSessionEvent,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { deepCompare } from "matrix-js-sdk/lib/utils";
@@ -97,7 +97,7 @@ vi.mock(
"../state/CallViewModel/localMember/localTransport",
async (importOriginal) => ({
...(await importOriginal()),
makeTransport: async (): Promise<LivekitTransport> =>
makeTransport: async (): Promise<UnstableLivekitTransport> =>
Promise.resolve(exampleTransport),
}),
);
+52 -70
View File
@@ -39,11 +39,12 @@ import {
timer,
takeUntil,
from,
concatWith,
} from "rxjs";
import { type Logger, logger as rootLogger } from "matrix-js-sdk/lib/logger";
import {
MembershipManagerEvent,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
type RTCCallIntent,
type RTCNotificationType,
@@ -56,6 +57,7 @@ import {
createToggle$,
filterBehavior,
generateItems,
mapScoped,
pauseWhen,
} from "../../utils/observable";
import {
@@ -104,17 +106,14 @@ import {
type SpotlightPortraitLayoutMedia,
} from "../layout-types.ts";
import { ElementCallError, UnknownCallError } from "../../utils/errors.ts";
import { type Epoch, type ObservableScope } from "../ObservableScope.ts";
import { Epoch, type ObservableScope } from "../ObservableScope.ts";
import { createHomeserverConnected$ } from "./localMember/HomeserverConnected.ts";
import {
createLocalMembership$,
enterRTCSession,
TransportState,
} from "./localMember/LocalMember.ts";
import {
getLocalTransport,
type LocalTransport,
} from "./localMember/LocalTransport.ts";
import { getLocalTransport } from "./localMember/LocalTransport.ts";
import {
createKeyRotationSuppressed$,
createMemberships$,
@@ -125,8 +124,9 @@ import {
ECConnectionFactory,
} from "./remoteMembers/ConnectionFactory.ts";
import {
type ConnectionManagerData,
ConnectionManagerData,
createConnectionManager$,
type IConnectionManager,
} from "./remoteMembers/ConnectionManager.ts";
import {
createRemoteMatrixLivekitMembers$,
@@ -215,7 +215,7 @@ export interface CallViewModelOptions {
*/
windowSize$: Behavior<{ width: number; height: number }>;
/** Optional value overriding the local transport, for testing purposes. */
localTransport?: LocalTransport;
localTransport?: UnstableLivekitTransport;
/** Optional value overriding the connection factory, for testing purposes. */
connectionFactory?: ConnectionFactory;
/** The version & compatibility mode of MatrixRTC that we should use. */
@@ -384,7 +384,7 @@ export interface CallViewModel {
* key is generated when someone joins or leaves.
*/
keyRotationSuppressed$: Behavior<boolean>;
allConnections$: Behavior<ConnectionManagerData>;
allConnections$: Behavior<Connection[]>;
/** Participants sorted by livekit room so they can be used in the audio rendering */
livekitRoomItems$: Behavior<LivekitRoomItem[]>;
/** use the layout instead, this is just for the sdk export. */
@@ -575,14 +575,7 @@ export function createCallViewModel$(
const localTransport$ = options.localTransport
? constant(options.localTransport)
: from(
getLocalTransport({
ownMembershipIdentity,
client,
roomId: matrixRoom.roomId,
matrixRTCMode,
}),
);
: from(getLocalTransport(client));
const connectionFactory =
options.connectionFactory ??
@@ -596,14 +589,25 @@ export function createCallViewModel$(
options.livekitRoomFactory,
);
const connectionManager = createConnectionManager$({
scope: scope,
connectionFactory: connectionFactory,
localTransport$,
remoteTransports$: membershipsAndTransports.transports$,
logger: logger,
ownMembershipIdentity,
});
const connectionManager: IConnectionManager = {
connectionManagerData$: scope.behavior(
localTransport$.pipe(
concatWith(NEVER), // So the Observable doesn't complete prematurely
mapScoped("connectionManager$", (scope, localTransport) =>
createConnectionManager$({
scope,
connectionFactory,
localTransport,
remoteTransports$: membershipsAndTransports.transports$,
logger,
ownMembershipIdentity,
}),
),
switchMap(({ connectionManagerData$ }) => connectionManagerData$),
),
new Epoch(new ConnectionManagerData()),
),
};
const remoteMatrixLivekitMembers$: Behavior<
Epoch<RemoteMatrixLivekitMember[]>
@@ -624,7 +628,7 @@ export function createCallViewModel$(
),
muteStates,
joinMatrixRTC: (
transport: LivekitTransport,
transport: UnstableLivekitTransport,
delayedLeaveTimings: ResolvedDelayedLeaveTimings,
) => {
return enterRTCSession(
@@ -646,9 +650,7 @@ export function createCallViewModel$(
mediaDevices,
muteStates,
trackProcessorState$,
logger.getChild(
"[Publisher " + connection.transport.livekit_service_url + "]",
),
logger.getChild(`[Publisher ${JSON.stringify(connection.transport)}]`),
controlledAudioDevices,
);
},
@@ -671,7 +673,6 @@ export function createCallViewModel$(
).pipe(map(([delayId]) => delayId ?? null)),
matrixRTCSession.delayId ?? null,
),
matrixRTCMode,
logger: logger.getChild(`[${Date.now()}]`),
});
@@ -739,51 +740,32 @@ export function createCallViewModel$(
});
const allConnections$ = scope.behavior(
connectionManager.connectionManagerData$.pipe(map((d) => d.value)),
connectionManager.connectionManagerData$.pipe(
map((d) => d.value.getConnections()),
),
);
const livekitRoomItems$ = scope.behavior(
remoteMatrixLivekitMembers$.pipe(
switchMap((members) => {
const a$ = combineLatest(
members.value.map((member) =>
combineLatest([member.connection$, member.participant.value$]).pipe(
map(([connection, participant]) => {
// do not render audio for local participant
if (!connection || !participant || participant.isLocal)
return null;
const livekitRoom = connection.livekitRoom;
const url = connection.transport.livekit_service_url;
return {
url,
livekitRoom,
participant: participant.identity,
};
}),
const livekitRoomItems$ = scope.behavior<LivekitRoomItem[]>(
allConnections$.pipe(
switchMap((connections) => {
if (connections.length === 0) return of([]);
return combineLatest(
connections.map((connection) =>
remoteMatrixLivekitMembers$.pipe(
switchMap((members) =>
combineLatest(members.value.map((m) => m.participant.value$)),
),
map((participants) => ({
url: JSON.stringify(connection.transport), // TODO
livekitRoom: connection.livekitRoom,
participants: participants
.filter((p) => p !== null)
.map((p) => p.identity),
})),
),
),
);
return a$;
}),
map((members) =>
members.reduce<LivekitRoomItem[]>((acc, curr) => {
if (!curr) return acc;
const existing = acc.find((item) => item.url === curr.url);
if (existing) {
existing.participants.push(curr.participant);
} else {
acc.push({
livekitRoom: curr.livekitRoom,
participants: [curr.participant],
url: curr.url,
});
}
return acc;
}, []),
),
),
[],
);
const handsRaised$ = scope.behavior(
@@ -841,7 +823,7 @@ export function createCallViewModel$(
connection$.pipe(map((c) => c?.livekitRoom)),
),
focusUrl$: scope.behavior(
connection$.pipe(map((c) => c?.transport.livekit_service_url)),
connection$.pipe(map((c) => JSON.stringify(c?.transport))), // TODO
),
mediaDevices,
pretendToBeDisconnected$: localMembership.reconnecting$,
@@ -30,7 +30,6 @@ import {
type CallViewModelOptions,
} from "./CallViewModel";
import {
exampleSfuConfig,
exampleTransport,
mockConfig,
MockConnection,
@@ -209,24 +208,21 @@ export function withCallViewModel(mode: MatrixRTCMode) {
livekitRoomFactory,
connectionState$,
windowSize$,
localTransport: {
transport: exampleTransport,
sfuConfig: exampleSfuConfig,
},
localTransport: exampleTransport,
connectionFactory: {
createConnection(
scope,
role,
transport,
ownMembershipIdentity,
logger,
sfuConfig,
) {
return new MockConnection(
{
scope,
role,
transport,
ownMembershipIdentity,
existingSFUConfig: sfuConfig,
client: room.client,
roomId: room.roomId,
livekitRoomFactory,
@@ -8,7 +8,7 @@ Please see LICENSE in the repository root for full details.
import {
Status as RTCMemberStatus,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
} from "matrix-js-sdk/lib/matrixrtc";
import {
@@ -58,8 +58,8 @@ import { ConnectionState, type Connection } from "../remoteMembers/Connection";
import { type Publisher } from "./Publisher";
import { initializeWidget } from "../../../widget";
import { nullHostBridge } from "../../../HostBridge";
import { type LocalTransport } from "./LocalTransport";
import * as openIDSFU from "../../../livekit/openIDSFU";
import { type SFUConfig } from "../../../livekit/auth";
import * as livekitAuth from "../../../livekit/auth";
initializeWidget();
@@ -110,6 +110,7 @@ const delegatedTimings: ResolvedDelayedLeaveTimings = {
};
const mockedClient = {
baseUrl: "https://matrix.example.org",
getDomain: vi.fn().mockReturnValue("example.org"),
getDeviceId: vi.fn().mockReturnValue("AAAA"),
getOpenIdToken: vi.fn().mockResolvedValue({
@@ -118,10 +119,11 @@ const mockedClient = {
matrix_server_name: "localhost",
expires_in: 10000,
}),
_unstable_delegateDelayedLeave: vi.fn().mockResolvedValue({}),
};
describe("enterRTCSession", () => {
const transport: LivekitTransport = {
const transport: UnstableLivekitTransport = {
livekit_service_url: "http://my-livekit-service-url.com",
type: "livekit",
};
@@ -252,7 +254,7 @@ describe("LocalMembership", () => {
it("throws error on missing RTC config error", () => {
withTestScheduler(({ scope, hot, expectObservable }) => {
const localTransport$ = scope.behavior<null | LivekitTransport>(
const localTransport$ = scope.behavior<null | UnstableLivekitTransport>(
hot("1ms #", {}, new MatrixRTCTransportMissingError("domain.com")),
null,
);
@@ -315,19 +317,17 @@ describe("LocalMembership", () => {
scope.end();
});
const mockTransportConfig = {
livekit_service_url: "a",
} as LivekitTransport;
const mockTransport: UnstableLivekitTransport = {
type: "livekit",
url: "https://sfu.example.org",
} as UnstableLivekitTransport;
const mockTransport = {
transport: mockTransportConfig,
sfuConfig: {
jwt: "foo",
livekitAlias: "bar",
livekitIdentity: "baz",
url: "bro",
},
} as LocalTransport;
const mockSFUConfig: SFUConfig = {
jwt: "foo",
livekitAlias: "bar",
livekitIdentity: "baz",
url: "bro",
};
const connectionTransportAConnected = {
livekitRoom: mockLivekitRoom({
@@ -337,7 +337,7 @@ describe("LocalMembership", () => {
} as unknown as LocalParticipant,
}),
state$: constant(ConnectionState.LivekitConnected),
transport: mockTransportConfig,
transport: mockTransport,
} as Connection;
const connectionTransportAConnecting = {
...connectionTransportAConnected,
@@ -346,8 +346,8 @@ describe("LocalMembership", () => {
} as unknown as Connection;
const authCallSpy = vi
.spyOn(openIDSFU, "getSFUConfigWithOpenID")
.mockImplementation(() => mockedClient.getOpenIdToken());
.spyOn(livekitAuth, "getSFUConfig")
.mockResolvedValue(mockSFUConfig);
afterEach(() => authCallSpy.mockClear());
it.each([
@@ -386,7 +386,7 @@ describe("LocalMembership", () => {
await flushPromises();
// Joins with timings appropriate for the level of delegation support
expect(joinMatrixRTC).toHaveBeenCalledWith(
mockTransportConfig,
mockTransport,
delayedLeaveTimings,
);
@@ -480,7 +480,7 @@ describe("LocalMembership", () => {
it("tracks livekit state correctly", async () => {
const scope = new ObservableScope();
const connectionManagerData = new ConnectionManagerData();
const localTransport$ = new Subject<LocalTransport>();
const localTransport$ = new Subject<UnstableLivekitTransport>();
const connectionManagerData$ = new BehaviorSubject(
new Epoch(connectionManagerData),
@@ -553,7 +553,7 @@ describe("LocalMembership", () => {
});
(
connectionManagerData2.getConnectionForTransport(mockTransportConfig)!
connectionManagerData2.getConnectionForTransport(mockTransport)!
.state$ as BehaviorSubject<ConnectionState>
).next(ConnectionState.LivekitConnected);
expect(localMembership.localMemberState$.value).toStrictEqual({
@@ -858,7 +858,7 @@ describe("LocalMembership", () => {
const setScreenShareEnabled = vi.fn().mockRejectedValue(error);
const connection = {
state$: constant(ConnectionState.LivekitConnected),
transport: mockTransportConfig,
transport: mockTransport,
livekitRoom: mockLivekitRoom({
localParticipant: mockLocalParticipant({
isScreenShareEnabled: false,
@@ -15,10 +15,9 @@ import {
MediaDeviceFailure,
} from "livekit-client";
import { observeParticipantEvents } from "@livekit/components-core";
import { type MatrixClient } from "matrix-js-sdk";
import {
Status as RTCSessionStatus,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
type RTCCallIntent,
type RTCNotificationType,
@@ -79,8 +78,11 @@ import {
type FailedToStartError,
} from "../remoteMembers/Connection.ts";
import { type HomeserverConnected } from "./HomeserverConnected.ts";
import { type LocalTransport } from "./LocalTransport.ts";
import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
import {
type ClientDelegationParts,
delegateDelayedLeave,
type ClientOpenIDParts,
} from "../../../livekit/auth";
export enum TransportState {
/** Not even a transport is available to the LocalMembership */
@@ -143,14 +145,14 @@ interface Props {
connectionManager: IConnectionManager;
createPublisherFactory: (connection: Connection) => Publisher;
joinMatrixRTC: (
transport: LivekitTransport,
transport: UnstableLivekitTransport,
delayedLeaveTimings: ResolvedDelayedLeaveTimings,
) => void;
homeserverConnected: HomeserverConnected;
roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts;
localTransport$: Observable<LocalTransport>;
client: Pick<MatrixClient, "getDeviceId" | "getOpenIdToken">;
localTransport$: Observable<UnstableLivekitTransport>;
client: ClientDelegationParts & ClientOpenIDParts;
matrixRTCSession: Pick<MatrixRTCSession, "updateCallIntent" | "leave">;
/** Whether to hide the screen-sharing button. */
hideScreensharing: boolean;
@@ -158,7 +160,6 @@ interface Props {
hostBridge: HostBridge;
baseUrl: string;
delayId$: Behavior<string | null>;
matrixRTCMode: MatrixRTCMode;
logger: Logger;
}
@@ -206,7 +207,6 @@ export const createLocalMembership$ = ({
hostBridge,
ownMembershipIdentity,
delayId$,
matrixRTCMode,
}: Props): {
/**
* This request to start audio and video tracks.
@@ -307,16 +307,17 @@ export const createLocalMembership$ = ({
// to whether delayed event delegation is supported
const joinParams$ = scope.behavior(
localTransport$.pipe(
switchMap(async ({ transport }) => {
const transportSupportsDelegation = checkDelegationSupport(
transport.livekit_service_url + "/delegate_delayed_leave",
`transport ${transport.livekit_service_url}`,
);
switchMap(async (transport) => {
const transportSupportsDelegation =
"livekit_service_url" in transport &&
(await checkDelegationSupport(
transport.livekit_service_url + "/delegate_delayed_leave",
`transport ${transport.livekit_service_url}`,
));
return {
transport,
delegationSupported:
(await homeserverSupportsDelegation) ||
(await transportSupportsDelegation),
transportSupportsDelegation || (await homeserverSupportsDelegation),
};
}),
),
@@ -329,12 +330,12 @@ export const createLocalMembership$ = ({
connectionManager.connectionManagerData$,
localTransport$,
]).pipe(
map(([{ value: connectionData }, { transport }]) =>
map(([{ value: connectionData }, transport]) =>
connectionData.getConnectionForTransport(transport),
),
tap((connection) => {
logger.info(
`Local connection updated: ${connection?.transport?.livekit_service_url}`,
`Local connection updated: ${JSON.stringify(connection?.transport)}`,
);
}),
),
@@ -406,7 +407,7 @@ export const createLocalMembership$ = ({
scope.reconcile(localConnection$, async (connection) => {
logger.info(
"reconcile based on new localConnection:",
connection?.transport.livekit_service_url,
JSON.stringify(connection?.transport),
);
if (connection !== null) {
const publisher = createPublisherFactory(connection);
@@ -702,23 +703,21 @@ export const createLocalMembership$ = ({
async ([joinParams, delayId]) => {
if (joinParams?.delegationSupported && delayId !== null) {
try {
// This will technically cause the service to issue a new JWT token,
// but it's safe to discard. We're only interested in triggering
// delegation.
await getSFUConfigWithOpenID(
client,
ownMembershipIdentity,
joinParams.transport.livekit_service_url,
roomId,
{ matrixRTCMode, delayEndpointBaseUrl: baseUrl, delayId },
logger,
logger.info(
`Delegating delayed leave to ${JSON.stringify(joinParams.transport)}…`,
);
await delegateDelayedLeave({
client,
membership: ownMembershipIdentity,
transport: joinParams.transport,
roomId,
delayId,
logger,
});
logger.info("Delayed leave successfully delegated");
} catch (e) {
// TODO: Surface this to the user as a service interruption?
logger.error(
`Failed to delegate leave to ${joinParams.transport.livekit_service_url}`,
e,
);
logger.error(`Failed to delegate leave`, e);
}
}
},
@@ -951,7 +950,7 @@ interface EnterRTCSessionOptions {
*
* @param rtcSession - The MatrixRTCSession to join.
* @param ownMembershipIdentity - Options for entering the RTC session.
* @param transport - The LivekitTransport to use for this session.
* @param transport - The UnstableLivekitTransport to use for this session.
* @param options - `encryptMedia`: Whether to encrypt media. `matrixRTCMode`: The
* Matrix RTC mode to use. `delayedLeaveTimings`: The preferred timings for
* delayed leave events. `sendNotificationType`: Whether and what kind of
@@ -962,7 +961,7 @@ interface EnterRTCSessionOptions {
export function enterRTCSession(
rtcSession: MatrixRTCSession,
ownMembershipIdentity: CallMembershipIdentityParts,
transport: LivekitTransport,
transport: UnstableLivekitTransport,
{
encryptMedia,
matrixRTCMode,
@@ -22,13 +22,13 @@ import {
MatrixRTCTransportMissingError,
FailToGetOpenIdToken,
} from "../../../utils/errors";
import * as openIDSFU from "../../../livekit/openIDSFU";
import { customLivekitUrl } from "../../../settings/settings";
import * as livekitAuth from "../../../livekit/auth";
import { customTransport } from "../../../settings/settings";
import { testJWTToken } from "../../../utils/test-fixtures";
import { MatrixRTCMode } from "../../../config/ConfigOptions";
import { type MatrixClient } from "matrix-js-sdk";
describe("LocalTransport", () => {
const openIdResponse: openIDSFU.SFUConfig = {
const sfuConfig: livekitAuth.SFUConfig = {
url: "https://lk.example.org",
jwt: testJWTToken,
livekitAlias: "Akph4alDMhen",
@@ -40,42 +40,25 @@ describe("LocalTransport", () => {
it("throws if config is missing", async () => {
await expect(
getLocalTransport({
roomId: "!room:example.org",
client: {
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
}),
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
});
it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
it("threes if SFU config fetch fails", async () => {
// Provide a valid config so makeTransportInternal resolves a transport
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
new FailToGetOpenIdToken(new Error("no openid")),
);
await expect(
getLocalTransport({
roomId: "!example_room_id",
client: {
getDomain: () => "example.org",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
getDomain: () => "example.org",
_unstable_getRTCTransports: async () => Promise.resolve([]),
}),
).rejects.toThrow(new FailToGetOpenIdToken(new Error("no openid")));
});
@@ -86,7 +69,7 @@ describe("LocalTransport", () => {
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue({
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue({
url: "https://lk.example.org",
jwt: "jwt",
livekitAlias: "Akph4alDMhen",
@@ -95,16 +78,8 @@ describe("LocalTransport", () => {
expect(
await getLocalTransport({
roomId: "!room:example.org",
client: {
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
ownMembershipIdentity: ownMemberMock,
matrixRTCMode: MatrixRTCMode.Compatibility,
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
}),
).toStrictEqual({
transport: {
@@ -120,26 +95,16 @@ describe("LocalTransport", () => {
});
});
type LocalTransportProps = Parameters<typeof getLocalTransport>[0];
describe("transport configuration mechanisms", () => {
let localTransportOpts: LocalTransportProps & {
client: MockedObject<LocalTransportProps["client"]>;
};
let client: MockedObject<
Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">
>;
beforeEach(() => {
mockConfig({});
customLivekitUrl.setValue(customLivekitUrl.defaultValue);
localTransportOpts = {
ownMembershipIdentity: ownMemberMock,
roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility,
client: {
getDomain: vi.fn().mockReturnValue("example.org"),
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
customTransport.setValue(customTransport.defaultValue);
client = {
getDomain: vi.fn().mockReturnValue("example.org"),
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
};
});
@@ -151,11 +116,9 @@ describe("LocalTransport", () => {
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
expect(await getLocalTransport(client)).toStrictEqual({
transport: {
livekit_service_url: "https://lk.example.org",
type: "livekit",
@@ -170,12 +133,13 @@ describe("LocalTransport", () => {
});
it("supports getting transport via user settings", async () => {
customLivekitUrl.setValue("https://lk.example.org");
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
customTransport.setValue({
type: "livekit",
livekit_service_url: "https://lk.example.org",
});
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
expect(await getLocalTransport(client)).toStrictEqual({
transport: {
livekit_service_url: "https://lk.example.org",
type: "livekit",
@@ -190,14 +154,12 @@ describe("LocalTransport", () => {
});
it("supports getting transport via backend", async () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockResolvedValue(
openIdResponse,
);
vi.spyOn(livekitAuth, "getSFUConfig").mockResolvedValue(sfuConfig);
expect(await getLocalTransport(localTransportOpts)).toStrictEqual({
expect(await getLocalTransport(client)).toStrictEqual({
transport: {
livekit_service_url: "https://lk.example.org",
type: "livekit",
@@ -212,14 +174,14 @@ describe("LocalTransport", () => {
});
it("fails fast if the openID request fails for backend config", async () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockRejectedValue(
vi.spyOn(livekitAuth, "getSFUConfig").mockRejectedValue(
new FailToGetOpenIdToken(new Error("Test driven error")),
);
await expect(getLocalTransport(localTransportOpts)).rejects.toThrow(
await expect(getLocalTransport(client)).rejects.toThrow(
expect.any(FailToGetOpenIdToken),
);
});
@@ -227,17 +189,8 @@ describe("LocalTransport", () => {
it("throws if no options are available", async () => {
await expect(
getLocalTransport({
ownMembershipIdentity: ownMemberMock,
roomId: "!example_room_id",
matrixRTCMode: MatrixRTCMode.Compatibility,
client: {
getDomain: () => "example.org",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
getDomain: () => "example.org",
_unstable_getRTCTransports: async () => Promise.resolve([]),
}),
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
});
@@ -5,142 +5,38 @@ SPDX-License-IdFentifier: 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 { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { Config } from "../../../config/Config.ts";
import {
FailToGetOpenIdToken,
MatrixRTCTransportMissingError,
NoMatrix2AuthorizationService,
} from "../../../utils/errors.ts";
import {
getSFUConfigWithOpenID,
type SFUConfig,
type OpenIDClientParts,
} from "../../../livekit/openIDSFU.ts";
import { customLivekitUrl } from "../../../settings/settings.ts";
import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
import { customTransport as customTransportSetting } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type MatrixRTCMode } from "../../../config/ConfigOptions.ts";
interface Props {
ownMembershipIdentity: CallMembershipIdentityParts;
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts;
// Used by the jwt service to create the livekit room and compute the livekit alias.
roomId: string;
matrixRTCMode: MatrixRTCMode;
}
export interface LocalTransport {
transport: LivekitTransport;
sfuConfig: SFUConfig;
}
export function isLocalTransport(
obj: LivekitTransport | LocalTransport,
): obj is LocalTransport {
return "transport" in obj && "sfuConfig" in obj;
}
/**
* Connects to the JWT service and determines the transport that the local member should use.
*
* @prop useOldJwtEndpoint Whether to set forceOldJwtEndpoint on the returned transport and to use the old JWT endpoint.
* This is used when the connection manager needs to know if it has to use the legacy endpoint which implies a string concatenated rtcBackendIdentity.
* (which is expected for non sticky event based rtc member events)
* @returns The transport to advertise in our MatrixRTC membership and publish media on.
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
* Determines the transport to advertise in our MatrixRTC membership and publish
* media on.
*/
export async function getLocalTransport({
ownMembershipIdentity,
client,
roomId,
matrixRTCMode,
}: Props): Promise<LocalTransport> {
const logger = rootLogger.getChild("[LocalTransport]");
export async function getLocalTransport(
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
): Promise<UnstableLivekitTransport> {
const discovery = new RtcTransportAutoDiscovery({
client: client,
client,
resolvedConfig: Config.get(),
logger: logger,
logger: rootLogger.getChild("[LocalTransport]"),
});
const customUrl = customLivekitUrl.value$.value;
// Respect the user's custom URL, if set
const transport: LivekitTransport | null = customUrl
? { type: "livekit", livekit_service_url: customUrl }
: await discovery.discoverPreferredTransport();
// Respect the user's custom transport, if set
const transport =
customTransportSetting.value$.value ??
(await discovery.discoverPreferredTransport());
if (transport === null)
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
try {
return await doOpenIdAndJWTFromUrl(
transport,
matrixRTCMode,
ownMembershipIdentity,
roomId,
client,
logger,
);
} catch (e) {
logger.error(
`Failed to authenticate to transport ${transport.livekit_service_url}`,
e,
);
throw mapAuthErrorToUserFriendlyError(e);
}
}
/**
* Utility to ensure the user can authenticate with the SFU.
* We will call `getSFUConfigWithOpenID` once per transport here as it's our
* only mechanism of validation. This means we will also ask the
* homeserver for a OpenID token a few times. Since OpenID tokens are single
* use we don't want to risk any issues by re-using a token.
*
* @param transport The transport to authenticate with.
* @param matrixRTCMode Whether to force the JWT endpoint to be used.
* @param membership The identity of the local member.
* @param roomId The room ID to use for the JWT.
* @param client The client to use for the OpenID token.
*
* @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService
*/
async function doOpenIdAndJWTFromUrl(
transport: LivekitTransport,
matrixRTCMode: MatrixRTCMode,
membership: CallMembershipIdentityParts,
roomId: string,
client: Pick<MatrixClient, "_unstable_getRTCTransports"> & OpenIDClientParts,
logger?: Logger,
): Promise<LocalTransport> {
const sfuConfig = await getSFUConfigWithOpenID(
client,
membership,
transport.livekit_service_url,
roomId,
{ matrixRTCMode },
logger,
);
return {
transport,
sfuConfig,
};
}
function mapAuthErrorToUserFriendlyError(e: unknown): Error {
if (
e instanceof FailToGetOpenIdToken ||
e instanceof NoMatrix2AuthorizationService
) {
// rethrow as is
return e;
}
// Catch others and rethrow as FailToGetOpenIdToken that has user friendly message.
return new FailToGetOpenIdToken(
e instanceof Error ? e : new Error(String(e)),
);
// TODO: Since this module no longer maps auth errors to fatal user-facing
// errors (as it doesn't even perform auth), that needs to happen somewhere
// else.
return transport;
}
@@ -16,7 +16,7 @@ import {
import { MatrixError } from "matrix-js-sdk";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import {
type LivekitTransport,
type UnstableLivekitTransport,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
@@ -28,12 +28,12 @@ import {
type DiscoveryClient = RtcTransportAutoDiscoveryProps["client"];
const backendTransport: LivekitTransport = {
const backendTransport: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://backend.example.org",
};
const configTransport: LivekitTransport = {
const configTransport: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://config.example.org",
};
@@ -5,8 +5,8 @@ SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import {
isLivekitTransport,
type LivekitTransport,
isUnstableLivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger";
@@ -40,12 +40,12 @@ export class RtcTransportAutoDiscovery {
this.logger = logger.getChild("[RtcTransportAutoDiscovery]");
}
public async discoverPreferredTransport(): Promise<LivekitTransport | null> {
public async discoverPreferredTransport(): Promise<UnstableLivekitTransport | null> {
// 1) backend transports
const backendTransport = await this.tryBackendTransports();
if (backendTransport) {
this.logger.info(
`Found backend transport: ${backendTransport.livekit_service_url}`,
`Found backend transport: ${JSON.stringify(backendTransport)}`,
);
return backendTransport;
}
@@ -54,7 +54,7 @@ export class RtcTransportAutoDiscovery {
const configTransport = this.tryConfigTransport();
if (configTransport) {
this.logger.info(
`Found app config transport: ${configTransport.livekit_service_url}`,
`Found app config transport: ${JSON.stringify(configTransport)}`,
);
return configTransport;
}
@@ -67,7 +67,7 @@ export class RtcTransportAutoDiscovery {
* This will not throw errors, but instead just log them and return null if the expected config is not found or malformed.
* @private
*/
private async tryBackendTransports(): Promise<LivekitTransport | null> {
private async tryBackendTransports(): Promise<UnstableLivekitTransport | null> {
const client = this.client;
// MSC4143: Attempt to fetch transports from backend.
this.logger.info("First try to use getRTCTransports end point ...");
@@ -75,7 +75,7 @@ export class RtcTransportAutoDiscovery {
const transportList = await doNetworkOperationWithRetry(async () =>
client._unstable_getRTCTransports(),
);
const first = transportList.find(isLivekitTransport);
const first = transportList.find(isUnstableLivekitTransport);
if (first) {
return first;
} else {
@@ -90,7 +90,7 @@ export class RtcTransportAutoDiscovery {
return null;
}
private tryConfigTransport(): LivekitTransport | null {
private tryConfigTransport(): UnstableLivekitTransport | null {
const url = this.resolvedConfig.livekit?.livekit_service_url;
if (url) {
return {
@@ -28,9 +28,9 @@ import {
} from "livekit-client";
import fetchMock from "fetch-mock";
import EventEmitter from "events";
import { type IOpenIDToken } from "matrix-js-sdk";
import { MatrixError, 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 { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import {
Connection,
@@ -38,17 +38,20 @@ import {
type ConnectionOpts,
} from "./Connection.ts";
import { ObservableScope } from "../../ObservableScope.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import {
ElementCallError,
FailToGetOpenIdToken,
} from "../../../utils/errors.ts";
import { testJWTToken } from "../../../utils/test-fixtures.ts";
import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts";
import {
type ClientGetTokenParts,
type ClientOpenIDParts,
} from "../../../livekit/auth";
let testScope: ObservableScope;
let client: MockedObject<OpenIDClientParts>;
let client: MockedObject<ClientGetTokenParts & ClientOpenIDParts>;
let fakeLivekitRoom: MockedObject<LivekitRoom>;
@@ -57,14 +60,14 @@ let fakeLocalParticipant: MockedObject<LocalParticipant>;
const ROOM_ID = "!roomID:example.org";
const livekitFocus: LivekitTransport = {
livekit_service_url: "https://matrix-rtc.example.org/livekit/jwt",
const transport: UnstableLivekitTransport = {
type: "livekit",
url: "wss://matrix-rtc.m.localhost/livekit/sfu",
};
function setupTest(): void {
testScope = new ObservableScope();
client = vi.mocked<OpenIDClientParts>({
client = vi.mocked<ClientGetTokenParts & ClientOpenIDParts>({
getOpenIdToken: vi.fn().mockResolvedValue({
access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
token_type: "Bearer",
@@ -72,7 +75,8 @@ function setupTest(): void {
expires_in: 3600,
}),
getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
} as unknown as OpenIDClientParts);
_unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
});
localParticipantEventEmiter = new EventEmitter();
@@ -118,22 +122,13 @@ function setupRemoteConnection(): Connection {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
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;
@@ -174,7 +169,8 @@ describe("Start connection states", () => {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -191,7 +187,8 @@ describe("Start connection states", () => {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -228,7 +225,7 @@ describe("Start connection states", () => {
capturedState = capturedStates.pop();
if (capturedState instanceof Error) {
expect(capturedState.message).toEqual("Something went wrong");
expect(connection.transport).toEqual(livekitFocus);
expect(connection.transport).toEqual(transport);
} else {
expect.fail("Expected FailedToStart state but got " + capturedState);
}
@@ -241,7 +238,8 @@ describe("Start connection states", () => {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -256,16 +254,15 @@ describe("Start connection states", () => {
onTestFinished(() => s.unsubscribe());
const deferredSFU = Promise.withResolvers<void>();
// mock the /sfu/get call
fetchMock.post(`${livekitFocus.livekit_service_url}/sfu/get`, async () => {
client._unstable_getLivekitToken.mockImplementation(async () => {
await deferredSFU.promise;
return {
status: 500,
body: {
throw new MatrixError(
{
errcode: "M_LOOKUP_FAILED",
error: "Failed to look up user info from homeserver",
},
};
500,
);
});
connection.start().catch(() => {
@@ -288,7 +285,7 @@ describe("Start connection states", () => {
expect(capturedState.cause.message).toContain(
"Failed to look up user info from homeserver",
);
expect(connection.transport).toEqual(livekitFocus);
expect(connection.transport).toEqual(transport);
} else {
expect.fail("Expected FailedToStart state but got " + capturedState);
}
@@ -301,7 +298,8 @@ describe("Start connection states", () => {
const opts: ConnectionOpts = {
client: client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -316,17 +314,6 @@ describe("Start connection states", () => {
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");
@@ -353,7 +340,7 @@ describe("Start connection states", () => {
expect(capturedState.cause.message).toContain(
"Failed to connect to livekit",
);
expect(connection.transport).toEqual(livekitFocus);
expect(connection.transport).toEqual(transport);
} else {
expect.fail(
"Expected FailedToStart state but got " + JSON.stringify(capturedState),
@@ -550,7 +537,8 @@ describe("remote track logging", () => {
{
client,
roomId: ROOM_ID,
transport: livekitFocus,
role: "subscriber",
transport: transport,
scope: testScope,
ownMembershipIdentity: ownMemberMock,
livekitRoomFactory: () => fakeLivekitRoom,
@@ -22,16 +22,11 @@ import {
type Track,
type TrackPublication,
} from "livekit-client";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } 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 "../../../livekit/openIDSFU.ts";
import { type Behavior } from "../../Behavior.ts";
import { type ObservableScope } from "../../ObservableScope.ts";
import {
@@ -42,19 +37,22 @@ import {
SFURoomCreationRestrictedError,
UnknownCallError,
} from "../../../utils/errors.ts";
import {
type ClientGetTokenParts,
getSFUConfig,
type ClientOpenIDParts,
type SFUConfig,
} from "../../../livekit/auth";
export interface ConnectionOpts {
/**
* For the local transport we already do know the jwt token and url. We can reuse it.
* On top the local transport will send additional data to the jwt server to use delayed event delegation.
*/
existingSFUConfig?: SFUConfig;
/** The identity parts to use on this connection */
ownMembershipIdentity: CallMembershipIdentityParts;
/** Whether we want to publish or only subscribe on this connection. */
role: "publisher" | "subscriber";
/** The media transport to connect to. */
transport: LivekitTransport;
transport: UnstableLivekitTransport;
/** The Matrix client to use for OpenID and SFU config requests. */
client: OpenIDClientParts;
client: ClientGetTokenParts & ClientOpenIDParts;
/** The room ID this connection is associated with. */
roomId: string;
/** The observable scope to use for this connection. */
@@ -104,10 +102,15 @@ export class Connection {
*/
public readonly state$: Behavior<ConnectionState | Error> = this._state$;
/**
* Whether we want to publish or only subscribe on this connection.
*/
public readonly role: "publisher" | "subscriber";
/**
* The media transport to connect to.
*/
public readonly transport: LivekitTransport;
public readonly transport: UnstableLivekitTransport;
public readonly livekitRoom: LivekitRoom;
@@ -137,11 +140,10 @@ export class Connection {
protected stopped = false;
// TODO: can we just keep the ConnectionOpts object instead of spreading?
private readonly client: OpenIDClientParts;
private readonly client: ClientGetTokenParts & ClientOpenIDParts;
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.
*
@@ -151,18 +153,16 @@ export class Connection {
*/
public constructor(opts: ConnectionOpts, logger: Logger) {
this.ownMembershipIdentity = opts.ownMembershipIdentity;
this.existingSFUConfig = opts.existingSFUConfig;
this.roomId = opts.roomId;
this.logger = logger.getChild(
"[Connection " + opts.transport.livekit_service_url + "]",
`[Connection ${JSON.stringify(opts.transport)}]`,
);
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.logger.debug(`constructor called`);
const { role, transport, client, scope } = opts;
this.scope = scope;
this.livekitRoom = opts.livekitRoomFactory();
this.role = role;
this.transport = transport;
this.client = client;
@@ -305,15 +305,9 @@ export class Connection {
this.stopped = false;
try {
this._state$.next(ConnectionState.FetchingConfig);
// We should already have this information after creating the localTransport.
// only call getSFUConfigWithOpenID for connections where we do not have a token yet. (existingJwtTokenData === undefined)
const { url, jwt, livekitAlias } =
this.existingSFUConfig ??
(await this.getSFUConfigForRemoteConnection());
const { url, jwt, livekitAlias } = await this.getSFUConfig();
this.logger.debug(
"Starting Connection to: ",
this.transport.livekit_service_url,
"jwt: ",
"Starting Connection - jwt: ",
jwt,
"wss: ",
url,
@@ -395,18 +389,17 @@ export class Connection {
}
}
protected async getSFUConfigForRemoteConnection(): Promise<SFUConfig> {
protected async getSFUConfig(): 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,
);
return await getSFUConfig({
client: this.client,
membership: this.ownMembershipIdentity,
role: this.role,
transport: this.transport,
roomId: this.roomId,
logger: this.logger,
});
}
/**
@@ -416,17 +409,13 @@ export class Connection {
* 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}`,
);
this.logger.debug("stop: disconnecing from lk room");
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}`,
);
this.logger.debug("stop: DONE disconnecing from lk room");
}
}
@@ -16,14 +16,10 @@ import { logger, type Logger } from "matrix-js-sdk/lib/logger";
// imported as inline to support worker when loaded from a cdn (cross domain)
import E2EEWorker from "livekit-client/e2ee-worker?worker&inline";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type ObservableScope } from "../../ObservableScope.ts";
import { Connection } from "./Connection.ts";
import type {
OpenIDClientParts,
SFUConfig,
} from "../../../livekit/openIDSFU.ts";
import type { MediaDevices } from "../../MediaDevices.ts";
import type { Behavior } from "../../Behavior.ts";
import type { ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
@@ -39,15 +35,19 @@ import {
noiseSuppressionSetting,
autoGainControlSetting,
} from "../../../settings/settings.ts";
import {
type ClientGetTokenParts,
type ClientOpenIDParts,
} from "../../../livekit/auth";
// TODO evaluate if this should be done like the Publisher Factory
export interface ConnectionFactory {
createConnection(
scope: ObservableScope,
transport: LivekitTransport,
role: "publisher" | "subscriber",
transport: UnstableLivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
): Connection;
}
@@ -66,7 +66,7 @@ export class ECConnectionFactory implements ConnectionFactory {
* @param livekitRoomFactory - Optional factory function (for testing) to create LivekitRoom instances. If not provided, a default factory is used.
*/
public constructor(
private client: OpenIDClientParts,
private client: ClientGetTokenParts & ClientOpenIDParts,
private readonly roomId: string,
private devices: MediaDevices,
private processorState$: Behavior<ProcessorState>,
@@ -103,15 +103,15 @@ export class ECConnectionFactory implements ConnectionFactory {
*/
public createConnection(
scope: ObservableScope,
transport: LivekitTransport,
role: "publisher" | "subscriber",
transport: UnstableLivekitTransport,
ownMembershipIdentity: CallMembershipIdentityParts,
logger: Logger,
sfuConfig?: SFUConfig,
): Connection {
return new Connection(
{
existingSFUConfig: sfuConfig,
roomId: this.roomId,
role,
transport,
client: this.client,
scope: scope,
@@ -6,8 +6,8 @@ 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 { BehaviorSubject } from "rxjs";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type RemoteParticipant } from "livekit-client";
import { logger } from "matrix-js-sdk/lib/logger";
@@ -19,17 +19,17 @@ import {
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import { type Connection } from "./Connection.ts";
import { ownMemberMock, withTestScheduler } from "../../../utils/test.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type Behavior } from "../../Behavior.ts";
// Some test constants
const TRANSPORT_1: LivekitTransport = {
const TRANSPORT_1: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const TRANSPORT_2: LivekitTransport = {
const TRANSPORT_2: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
};
@@ -47,7 +47,7 @@ beforeEach(() => {
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
(scope: ObservableScope, transport: UnstableLivekitTransport) => {
const mockConnection = {
transport,
remoteParticipants$: new BehaviorSubject([]),
@@ -74,7 +74,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
localTransport: TRANSPORT_1,
remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
@@ -94,13 +94,13 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalled();
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
expect(conn2!.start).toHaveBeenCalled();
@@ -115,7 +115,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
localTransport: TRANSPORT_1,
remoteTransports$: behavior("abcdef", {
a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1], 1),
@@ -141,12 +141,12 @@ describe("connections$ stream", () => {
).toHaveBeenCalledTimes(2);
const conn2 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_2),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(conn2).toBeDefined();
const conn1 = connections.find((c) =>
areLivekitTransportsEqual(c.transport, TRANSPORT_1),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_1),
);
expect(conn1).toBeDefined();
expect(conn1!.start).toHaveBeenCalledOnce();
@@ -162,7 +162,7 @@ describe("connections$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
localTransport: TRANSPORT_1,
remoteTransports$: behavior("abc", {
a: new Epoch([TRANSPORT_1], 0),
b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
@@ -187,7 +187,7 @@ describe("connections$ stream", () => {
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),
areUnstableLivekitTransportsEqual(c.transport, TRANSPORT_2),
);
expect(connection2).toBeDefined();
expect(connection2!.stop).toHaveBeenCalled();
@@ -225,15 +225,20 @@ 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}`;
function keyForTransport(transport: UnstableLivekitTransport): string {
return JSON.stringify([
"url" in transport ? transport.url : undefined,
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined,
]);
}
beforeEach(() => {
fakeRemoteParticipantsStreams = new Map();
function getRemoteParticipantsFor(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): Behavior<RemoteParticipant[]> {
return (
fakeRemoteParticipantsStreams.get(keyForTransport(transport)) ??
@@ -245,7 +250,7 @@ describe("connectionManagerData$ stream", () => {
vi.mocked(fakeConnectionFactory).createConnection = vi
.fn()
.mockImplementation(
(scope: ObservableScope, transport: LivekitTransport) => {
(scope: ObservableScope, transport: UnstableLivekitTransport) => {
const fakeRemoteParticipants$ = new BehaviorSubject<
RemoteParticipant[]
>([]);
@@ -297,7 +302,7 @@ describe("connectionManagerData$ stream", () => {
const { connectionManagerData$ } = createConnectionManager$({
scope: testScope,
connectionFactory: fakeConnectionFactory,
localTransport$: NEVER,
localTransport: TRANSPORT_1,
remoteTransports$: behavior("a", {
a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
}),
@@ -6,32 +6,22 @@ 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 UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, map, of, switchMap, tap } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger";
import { type RemoteParticipant } from "livekit-client";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Behavior } from "../../Behavior.ts";
import { type Connection } from "./Connection.ts";
import { Epoch, type ObservableScope } from "../../ObservableScope.ts";
import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
import {
isLocalTransport,
type LocalTransport,
} from "../localMember/LocalTransport.ts";
import { type SFUConfig } from "../../../livekit/openIDSFU.ts";
Epoch,
mapEpoch,
type ObservableScope,
} from "../../ObservableScope.ts";
import { generateItemsWithEpoch } from "../../../utils/observable.ts";
import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
import { type ConnectionFactory } from "./ConnectionFactory.ts";
export class ConnectionManagerData {
private readonly store: Map<
@@ -56,10 +46,15 @@ export class ConnectionManagerData {
}
}
private getKey(transport: LivekitTransport): string {
private getKey(transport: UnstableLivekitTransport): 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;
return JSON.stringify([
"url" in transport ? transport.url : undefined,
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined,
]);
}
public getConnections(): Connection[] {
@@ -67,13 +62,13 @@ export class ConnectionManagerData {
}
public getConnectionForTransport(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): Connection | null {
return this.store.get(this.getKey(transport))?.connection ?? null;
}
public getParticipantsForTransport(
transport: LivekitTransport,
transport: UnstableLivekitTransport,
): RemoteParticipant[] {
const key = this.getKey(transport);
const existing = this.store.get(key);
@@ -87,8 +82,8 @@ export class ConnectionManagerData {
interface Props {
scope: ObservableScope;
connectionFactory: ConnectionFactory;
localTransport$: Observable<LocalTransport>;
remoteTransports$: Behavior<Epoch<LivekitTransport[]>>;
localTransport: UnstableLivekitTransport;
remoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
logger: Logger;
ownMembershipIdentity: CallMembershipIdentityParts;
@@ -122,7 +117,7 @@ export interface IConnectionManager {
export function createConnectionManager$({
scope,
connectionFactory,
localTransport$,
localTransport,
remoteTransports$,
logger: parentLogger,
ownMembershipIdentity,
@@ -130,49 +125,21 @@ export function createConnectionManager$({
const logger = parentLogger.getChild("[ConnectionManager]");
// TODO logger: only construct one logger from the client and make it compatible via a EC specific sing
const localTransportAsArray$ = localTransport$.pipe(
// LocalMember already surfaces local transport errors properly in the UI,
// here we can just swallow them
catchError(() => NEVER),
map((transport) => [transport]),
startWith([]),
);
/**
* All transports currently managed by the ConnectionManager.
*
* This list does not include duplicate transports.
*
* It is build based on the list of subscribed transports (`transportsSubscriptions$`).
* externally this is modified via `registerTransports()`.
* All transports currently managed by the ConnectionManager. This list does
* not include duplicate transports.
*/
const localAndRemoteTransports$: Behavior<
Epoch<(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,
);
}),
),
);
const localAndRemoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>> =
scope.behavior(
remoteTransports$.pipe(
mapEpoch((remoteTransports) =>
removeDuplicateTransports([localTransport, ...remoteTransports]),
),
tap((transports) =>
logger.debug("localAndRemoteTransports$ = ", transports),
),
),
);
/**
* Connections for each transport in use by one or more session members.
@@ -183,39 +150,31 @@ export function createConnectionManager$({
"ConnectionManager connections$",
function* (transports) {
for (const transport of transports) {
if (isLocalTransport(transport)) {
// This is the local transport; only the `LocalTransport`
// interface has a `sfuConfig` field.
yield {
keys: [
transport.transport.livekit_service_url,
transport.sfuConfig,
],
data: undefined,
};
} else {
yield {
keys: [
transport.livekit_service_url,
undefined as SFUConfig | undefined,
],
data: undefined,
};
}
const role =
transport === localTransport
? ("publisher" as const)
: ("subscriber" as const);
const url = "url" in transport ? transport.url : undefined;
const serviceUrl =
"livekit_service_url" in transport
? transport.livekit_service_url
: undefined;
yield { keys: [role, url, serviceUrl], data: undefined };
}
},
(scope, _data$, serviceUrl, sfuConfig) => {
(scope, _data$, role, url, serviceUrl) => {
const connection = connectionFactory.createConnection(
scope,
role,
{
type: "livekit",
livekit_service_url: serviceUrl,
},
...(url === undefined ? {} : { url }),
...(serviceUrl === undefined
? {}
: { livekit_service_url: serviceUrl }),
} as UnstableLivekitTransport,
ownMembershipIdentity,
logger,
// TODO: This whole optional SFUConfig parameter is not particularly elegant.
// I would like it if connections always fetched the SFUConfig by themselves.
sfuConfig,
);
// Start the connection immediately
// Use connection state to track connection progress
@@ -271,11 +230,11 @@ export function createConnectionManager$({
return { connectionManagerData$ };
}
function removeDuplicateTransports<T extends LivekitTransport>(
function removeDuplicateTransports<T extends UnstableLivekitTransport>(
transports: T[],
): T[] {
return transports.reduce((acc, transport) => {
if (!acc.some((t) => areLivekitTransportsEqual(t, transport)))
if (!acc.some((t) => areUnstableLivekitTransportsEqual(t, transport)))
acc.push(transport);
return acc;
}, [] as T[]);
@@ -14,7 +14,10 @@ import EventEmitter from "events";
import { ObservableScope } from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import type { OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import {
type ClientGetTokenParts,
type ClientOpenIDParts,
} from "../../../livekit/auth";
import {
exampleTransport,
mockMediaDevices,
@@ -32,6 +35,7 @@ import {
cameraBitrate,
cameraCodec,
} from "../../../settings/settings.ts";
import { testJWTToken } from "../../../utils/test-fixtures.ts";
// At the top of your test file, after imports
vi.mock("livekit-client", async (importOriginal) => {
@@ -51,13 +55,14 @@ vi.mock("livekit-client", async (importOriginal) => {
});
let testScope: ObservableScope;
let mockClient: OpenIDClientParts;
let mockClient: ClientGetTokenParts & ClientOpenIDParts;
beforeEach(() => {
testScope = new ObservableScope();
mockClient = {
getOpenIdToken: vi.fn().mockReturnValue(""),
getDeviceId: vi.fn().mockReturnValue("DEV000"),
_unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
};
});
@@ -89,6 +94,7 @@ describe("ECConnectionFactory - Audio inputs options", () => {
);
ecConnectionFactory.createConnection(
testScope,
"subscriber",
exampleTransport,
ownMemberMock,
logger,
@@ -137,6 +143,7 @@ describe("ECConnectionFactory - ControlledAudioDevice", () => {
);
ecConnectionFactory.createConnection(
testScope,
"subscriber",
exampleTransport,
ownMemberMock,
logger,
@@ -172,6 +179,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
);
ecConnectionFactory.createConnection(
testScope,
"subscriber",
exampleTransport,
ownMemberMock,
logger,
@@ -209,6 +217,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
);
ecConnectionFactory.createConnection(
testScope,
"subscriber",
exampleTransport,
ownMemberMock,
logger,
@@ -247,6 +256,7 @@ describe("ECConnectionFactory - Camera quality settings", () => {
);
ecConnectionFactory.createConnection(
testScope,
"subscriber",
exampleTransport,
ownMemberMock,
logger,
@@ -9,7 +9,7 @@ import { describe, test, expect, beforeEach, afterEach, vi } from "vitest";
import { logger } from "matrix-js-sdk/lib/logger";
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject, combineLatest, map, type Observable } from "rxjs";
@@ -39,12 +39,12 @@ let testScope: ObservableScope;
const fallbackMemberId = (userId: string, deviceId: string): string =>
`${userId}:${deviceId}`;
const transportA: LivekitTransport = {
const transportA: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const transportB: LivekitTransport = {
const transportB: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.sample.com",
};
@@ -117,7 +117,7 @@ test("should signal participant not yet connected to livekit", async () => {
function createEpochedMemberships$(m$: Observable<CallMembership[]>): {
memberships$: Observable<Epoch<CallMembership[]>>;
membershipsWithTransport$: Observable<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
>;
} {
const memberships$ = m$.pipe(trackEpoch());
@@ -128,7 +128,7 @@ function createEpochedMemberships$(m$: Observable<CallMembership[]>): {
return {
membership: m,
transport:
tr?.type === "livekit" ? (tr as LivekitTransport) : undefined,
tr?.type === "livekit" ? (tr as UnstableLivekitTransport) : undefined,
};
});
}),
@@ -8,7 +8,7 @@ Please see LICENSE in the repository root for full details.
import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { combineLatest, filter, map } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
@@ -60,7 +60,9 @@ export interface RemoteMatrixLivekitMember extends MatrixLivekitMember {
interface Props {
scope: ObservableScope;
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<
{ membership: CallMembership; transport?: UnstableLivekitTransport }[]
>
>;
connectionManager: IConnectionManager;
localUser: { deviceId: string; userId: string };
@@ -111,8 +113,8 @@ export function createRemoteMatrixLivekitMembers$({
const participants = transport
? managerData.getParticipantsForTransport(transport)
: [];
const matches = participants.filter(
(p) => p.identity == membership.rtcBackendIdentity,
const matches = participants.filter((p) =>
membership.backendIdentities.includes(p.identity),
);
const participant = matches[0] ?? null;
const connection = transport
@@ -143,9 +145,9 @@ export function createRemoteMatrixLivekitMembers$({
// Log whether the member could be matched to a LiveKit participant,
// since a tile shows "waiting for media" for as long as it cannot.
participant$.pipe(scope.bind()).subscribe((p) => {
const url = data$.value.connection?.transport.livekit_service_url;
const transport = data$.value.connection?.transport;
logger.info(
`[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${url ?? "no connection"}`,
`[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${transport ? JSON.stringify(transport) : "no connection"}`,
);
});
// will only get called once per backend identity.
@@ -165,12 +167,26 @@ export function createRemoteMatrixLivekitMembers$({
// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
// TODO add this to the JS-SDK
export function areLivekitTransportsEqual<T extends LivekitTransport>(
t1: T | null,
t2: T | null,
): boolean {
export function areUnstableLivekitTransportsEqual<
T extends UnstableLivekitTransport,
>(t1: T | null, t2: T | null): boolean {
if (t1 && t2) {
return t1.livekit_service_url === t2.livekit_service_url;
if ("url" in t1 !== "url" in t2) return false;
if ("livekit_service_url" in t1 !== "livekit_service_url" in t2)
return false;
if (
"url" in t1 &&
t1.url !== (t2 as UnstableLivekitTransport & { url: string }).url
)
return false;
if (
"livekit_service_url" in t1 &&
t1.livekit_service_url !==
(t2 as UnstableLivekitTransport & { livekit_service_url: string })
.livekit_service_url
)
return false;
return true;
}
return !t1 && !t2;
}
@@ -6,11 +6,11 @@ Please see LICENSE in the repository root for full details.
*/
import { test, vi, expect, beforeEach, afterEach } from "vitest";
import { BehaviorSubject, NEVER } from "rxjs";
import { BehaviorSubject } from "rxjs";
import { type Room as LivekitRoom } from "livekit-client";
import EventEmitter from "events";
import fetchMock from "fetch-mock";
import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
import { logger } from "matrix-js-sdk/lib/logger";
import {
@@ -19,8 +19,12 @@ import {
trackEpoch,
} from "../../ObservableScope.ts";
import { ECConnectionFactory } from "./ConnectionFactory.ts";
import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
import {
type ClientGetTokenParts,
type ClientOpenIDParts,
} from "../../../livekit/auth";
import {
exampleTransport,
mockMediaDevices,
mockRtcMembership,
ownMemberMock,
@@ -28,7 +32,7 @@ import {
} from "../../../utils/test.ts";
import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
import {
areLivekitTransportsEqual,
areUnstableLivekitTransportsEqual,
createRemoteMatrixLivekitMembers$,
type RemoteMatrixLivekitMember,
} from "./MatrixLivekitMembers.ts";
@@ -40,7 +44,7 @@ import { localRtcMember, testJWTToken } from "../../../utils/test-fixtures.ts";
let testScope: ObservableScope;
let ecConnectionFactory: ECConnectionFactory;
let mockClient: OpenIDClientParts;
let mockClient: ClientGetTokenParts & ClientOpenIDParts;
let lkRoomFactory: () => LivekitRoom;
const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
@@ -50,6 +54,7 @@ beforeEach(() => {
mockClient = {
getOpenIdToken: vi.fn().mockReturnValue(""),
getDeviceId: vi.fn().mockReturnValue("DEV000"),
_unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
};
lkRoomFactory = vi.fn().mockImplementation(() => {
@@ -123,7 +128,7 @@ test("bob, carl, then bob joining no tracks yet", () => {
const connectionManager = createConnectionManager$({
scope: testScope,
connectionFactory: ecConnectionFactory,
localTransport$: NEVER,
localTransport: exampleTransport,
remoteTransports$: membershipsAndTransports.transports$,
logger: logger,
ownMembershipIdentity: ownMemberMock,
@@ -147,9 +152,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
});
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((co) =>
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
co.transport,
bobMembership.transports[0]! as LivekitTransport,
bobMembership.transports[0]! as UnstableLivekitTransport,
),
),
});
@@ -184,9 +189,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((connection) => {
expect(
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
connection.transport,
carlMembership.transports[0]! as LivekitTransport,
carlMembership.transports[0]! as UnstableLivekitTransport,
),
).toBe(true);
return true;
@@ -214,9 +219,9 @@ test("bob, carl, then bob joining no tracks yet", () => {
expectObservable(item.connection$).toBe("a", {
a: expect.toSatisfy((connection) => {
expect(
areLivekitTransportsEqual(
areUnstableLivekitTransportsEqual(
connection.transport,
daveMembership.transports[0]! as LivekitTransport,
daveMembership.transports[0]! as UnstableLivekitTransport,
),
).toBe(true);
return true;
+7 -7
View File
@@ -7,10 +7,10 @@ Please see LICENSE in the repository root for full details.
import {
type CallMembership,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
isLivekitTransport,
isUnstableLivekitTransport,
} from "matrix-js-sdk/lib/matrixrtc";
import { fromEvent } from "rxjs";
@@ -31,15 +31,15 @@ export const membershipsAndTransports$ = (
memberships$: Behavior<Epoch<CallMembership[]>>,
): {
membershipsWithTransport$: Behavior<
Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
Epoch<{ membership: CallMembership; transport?: UnstableLivekitTransport }[]>
>;
transports$: Behavior<Epoch<LivekitTransport[]>>;
transports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
} => {
const membershipsWithTransport$: Behavior<
Epoch<
{
membership: CallMembership;
transport: LivekitTransport | undefined;
transport: UnstableLivekitTransport | undefined;
}[]
>
> = scope.behavior(
@@ -49,14 +49,14 @@ export const membershipsAndTransports$ = (
const transport = membership.getTransport();
return {
membership,
transport: isLivekitTransport(transport) ? transport : undefined,
transport: isUnstableLivekitTransport(transport) ? transport : undefined,
};
});
}),
),
);
const transports$: Behavior<Epoch<LivekitTransport[]>> = scope.behavior(
const transports$: Behavior<Epoch<UnstableLivekitTransport[]>> = scope.behavior(
membershipsWithTransport$.pipe(
mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
),
+22
View File
@@ -218,6 +218,28 @@ export function filterBehavior<T, S extends T>(
);
}
/**
* Maps a changing input value to an item whose lifetime is tied to a certain
* computed key. The item may capture some dynamic data from the input.
*/
export function mapScoped<A, B>(
name: string,
project: (scope: ObservableScope, a: A) => B,
): OperatorFunction<A, B> {
return (a$) =>
a$.pipe(
generateItemsInternal(
name,
function* (a) {
yield { keys: [a], data: undefined };
},
(scope, _data$, a) => project(scope, a),
(items) => items,
),
map(([item]) => item),
);
}
function generateItemsInternal<
Input,
Keys extends [unknown, ...unknown[]],
+6 -13
View File
@@ -26,7 +26,7 @@ import {
import {
CallMembership,
type LivekitFocusSelection,
type LivekitTransport,
type UnstableLivekitTransport,
type MatrixRTCSession,
MatrixRTCSessionEvent,
type MatrixRTCSessionEventHandlerMap,
@@ -78,7 +78,6 @@ import {
type RemoteScreenShareViewModel,
} from "../state/media/RemoteScreenShareViewModel";
import { Connection } from "../state/CallViewModel/remoteMembers/Connection";
import { type SFUConfig } from "../livekit/openIDSFU";
export function withFakeTimers(continuation: () => void): void {
vi.useFakeTimers();
@@ -209,18 +208,11 @@ export function mockEmitter<T>(): EmitterMock<T> {
};
}
export const exampleTransport: LivekitTransport = {
export const exampleTransport: UnstableLivekitTransport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
export const exampleSfuConfig: SFUConfig = {
jwt: "foo",
livekitAlias: "bar",
livekitIdentity: "baz",
url: "bro",
};
export function mockRtcMembership(
user: string | RoomMember,
deviceId: string,
@@ -261,11 +253,12 @@ export function mockRtcMembership(
content: data,
});
const membershipData = CallMembership.membershipDataFromMatrixEvent(event);
const backendIdentity = rtcBackendIdentity ?? `${userId}:${deviceId}`;
const cms = new CallMembership(
event,
membershipData,
rtcBackendIdentity ?? `${userId}:${deviceId}`,
CallMembership.membershipDataFromMatrixEvent(event),
[backendIdentity],
backendIdentity,
);
vi.mocked(cms).getTransport = vi.fn().mockReturnValue(fociPreferred[0]);
+2
View File
@@ -175,6 +175,8 @@ export const initializeWidget = (
receiveToDevice: sendRecvToDevice,
turnServers: false,
rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
sendDelayedEvents: true,
updateDelayedEvents: true,
sendSticky: true,