studio/element-call
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support MSC4143 RTC Transport endpoint (#3629)

Browse Source 
* Use rtc-focus branch of js-sdk

* Update makeTransport to fetch backend transports and validate all transports before response.

* Fix test

* Add test

* Loads more tests

* Add tests for openid errors

* improve comment

* update to develop commit

* Add JWT parsing

* Use JWT

* Cleanup

* fixup tests

* fixup tests

* lint

* lint lint

* Fix `Reconnecting`
This commit is contained in:
Will Hunt
2025-12-29 17:45:41 +00:00
committed by GitHub
parent 67d20a8f3d
commit 72ec1439f4
13 changed files with 522 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/state/CallViewModel/CallViewModel.test.ts
View File

@@ -1256,7 +1256,9 @@ describe.each([
rtcSession.membershipStatus = Status.Connected;
},
n: () => {
rtcSession.membershipStatus = Status.Reconnecting;
// NOTE: This was removed in https://github.com/matrix-org/matrix-js-sdk/pull/5103 accidentally.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
rtcSession.membershipStatus = "Reconnecting" as any;
},
});
schedule(probablyLeftMarbles, {

238
src/state/CallViewModel/localMember/LocalTransport.test.ts
View File

@@ -5,9 +5,18 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import {
afterEach,
beforeEach,
describe,
expect,
it,
type MockedObject,
vi,
} from "vitest";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { BehaviorSubject } from "rxjs";
import { BehaviorSubject, lastValueFrom } from "rxjs";
import fetchMock from "fetch-mock";
import { mockConfig, flushPromises } from "../../../utils/test";
import { createLocalTransport$ } from "./LocalTransport";
@@ -18,31 +27,22 @@ import {
FailToGetOpenIdToken,
} from "../../../utils/errors";
import * as openIDSFU from "../../../livekit/openIDSFU";
import { customLivekitUrl } from "../../../settings/settings";
import { testJWTToken } from "../../../utils/test-fixtures";
describe("LocalTransport", () => {
const openIdResponse: openIDSFU.SFUConfig = {
url: "https://lk.example.org",
jwt: testJWTToken,
livekitAlias: "!example_room_id",
livekitIdentity: "@lk_user:ABCDEF",
};
let scope: ObservableScope;
beforeEach(() => (scope = new ObservableScope()));
afterEach(() => scope.end());
it("throws if config is missing", async () => {
const localTransport$ = createLocalTransport$({
scope,
roomId: "!room:example.org",
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: () => "",
// These won't be called in this error path but satisfy the type
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
});
await flushPromises();
expect(() => localTransport$.value).toThrow(
new MatrixRTCTransportMissingError(""),
);
beforeEach(() => {
scope = new ObservableScope();
});
afterEach(() => scope.end());
it("throws FailToGetOpenIdToken when OpenID fetch fails", async () => {
// Provide a valid config so makeTransportInternal resolves a transport
@@ -61,12 +61,14 @@ describe("LocalTransport", () => {
const errors: Error[] = [];
const localTransport$ = createLocalTransport$({
scope,
roomId: "!room:example.org",
roomId: "!example_room_id",
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
// Use empty domain to skip .well-known and use config directly
getDomain: () => "",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
@@ -84,41 +86,6 @@ describe("LocalTransport", () => {
expect(() => localTransport$.value).toThrow(expectedError);
});
it("emits preferred transport after OpenID resolves", async () => {
// Use config so transport discovery succeeds, but delay OpenID JWT fetch
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
const openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
openIdResolver.promise,
);
const localTransport$ = createLocalTransport$({
scope,
roomId: "!room:example.org",
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: () => "",
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
});
openIdResolver.resolve?.({ url: "https://lk.example.org", jwt: "jwt" });
expect(localTransport$.value).toBe(null);
await flushPromises();
// final
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!room:example.org",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
});
it("updates local transport when oldest member changes", async () => {
// Use config so transport discovery succeeds, but delay OpenID JWT fetch
mockConfig({
@@ -133,24 +100,171 @@ describe("LocalTransport", () => {
const localTransport$ = createLocalTransport$({
scope,
roomId: "!room:example.org",
roomId: "!example_room_id",
useOldestMember$: constant(true),
memberships$,
client: {
getDomain: () => "",
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: async () => Promise.resolve([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
});
openIdResolver.resolve?.({ url: "https://lk.example.org", jwt: "jwt" });
openIdResolver.resolve?.(openIdResponse);
expect(localTransport$.value).toBe(null);
await flushPromises();
// final
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!room:example.org",
livekit_alias: "!example_room_id",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
});
type LocalTransportProps = Parameters<typeof createLocalTransport$>[0];
describe("transport configuration mechanisms", () => {
let localTransportOpts: LocalTransportProps & {
client: MockedObject<LocalTransportProps["client"]>;
};
let openIdResolver: PromiseWithResolvers<openIDSFU.SFUConfig>;
beforeEach(() => {
mockConfig({});
customLivekitUrl.setValue(customLivekitUrl.defaultValue);
localTransportOpts = {
scope,
roomId: "!example_room_id",
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: vi.fn().mockReturnValue(""),
// eslint-disable-next-line @typescript-eslint/naming-convention
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
getOpenIdToken: vi.fn(),
getDeviceId: vi.fn(),
},
};
openIdResolver = Promise.withResolvers<openIDSFU.SFUConfig>();
vi.spyOn(openIDSFU, "getSFUConfigWithOpenID").mockReturnValue(
openIdResolver.promise,
);
});
afterEach(() => {
fetchMock.reset();
});
it("supports getting transport via application config", async () => {
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
const localTransport$ = createLocalTransport$(localTransportOpts);
openIdResolver.resolve?.(openIdResponse);
expect(localTransport$.value).toBe(null);
await flushPromises();
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!example_room_id",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
});
it("supports getting transport via user settings", async () => {
customLivekitUrl.setValue("https://lk.example.org");
const localTransport$ = createLocalTransport$(localTransportOpts);
openIdResolver.resolve?.(openIdResponse);
expect(localTransport$.value).toBe(null);
await flushPromises();
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!example_room_id",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
});
it("supports getting transport via backend", async () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
const localTransport$ = createLocalTransport$(localTransportOpts);
openIdResolver.resolve?.(openIdResponse);
expect(localTransport$.value).toBe(null);
await flushPromises();
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!example_room_id",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
});
it("fails fast if the openID request fails for backend config", async () => {
localTransportOpts.client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
]);
openIdResolver.reject(
new FailToGetOpenIdToken(new Error("Test driven error")),
);
try {
await lastValueFrom(createLocalTransport$(localTransportOpts));
throw Error("Expected test to throw");
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
}
});
it("supports getting transport via well-known", async () => {
localTransportOpts.client.getDomain.mockReturnValue("example.org");
fetchMock.getOnce("https://example.org/.well-known/matrix/client", {
"org.matrix.msc4143.rtc_foci": [
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
],
});
const localTransport$ = createLocalTransport$(localTransportOpts);
openIdResolver.resolve?.(openIdResponse);
expect(localTransport$.value).toBe(null);
await flushPromises();
expect(localTransport$.value).toStrictEqual({
livekit_alias: "!example_room_id",
livekit_service_url: "https://lk.example.org",
type: "livekit",
});
expect(fetchMock.done()).toEqual(true);
});
it("fails fast if the openId request fails for the well-known config", async () => {
localTransportOpts.client.getDomain.mockReturnValue("example.org");
fetchMock.getOnce("https://example.org/.well-known/matrix/client", {
"org.matrix.msc4143.rtc_foci": [
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
],
});
openIdResolver.reject(
new FailToGetOpenIdToken(new Error("Test driven error")),
);
try {
await lastValueFrom(createLocalTransport$(localTransportOpts));
throw Error("Expected test to throw");
} catch (ex) {
expect(ex).toBeInstanceOf(FailToGetOpenIdToken);
}
});
it("throws if no options are available", async () => {
const localTransport$ = createLocalTransport$({
scope,
roomId: "!example_room_id",
useOldestMember$: constant(false),
memberships$: constant(new Epoch<CallMembership[]>([])),
client: {
getDomain: () => "",
// 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(),
},
});
await flushPromises();
expect(() => localTransport$.value).toThrow(
new MatrixRTCTransportMissingError(""),
);
});
});
});

169
src/state/CallViewModel/localMember/LocalTransport.ts
View File

@@ -8,11 +8,11 @@ Please see LICENSE in the repository root for full details.
import {
type CallMembership,
isLivekitTransport,
type LivekitTransportConfig,
type LivekitTransport,
isLivekitTransportConfig,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type MatrixClient } from "matrix-js-sdk";
import { MatrixError, type MatrixClient } from "matrix-js-sdk";
import {
combineLatest,
distinctUntilChanged,
@@ -27,7 +27,10 @@ import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
import { type Behavior } from "../../Behavior.ts";
import { type Epoch, type ObservableScope } from "../../ObservableScope.ts";
import { Config } from "../../../config/Config.ts";
import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
import {
FailToGetOpenIdToken,
MatrixRTCTransportMissingError,
} from "../../../utils/errors.ts";
import {
getSFUConfigWithOpenID,
type OpenIDClientParts,
@@ -45,7 +48,8 @@ const logger = rootLogger.getChild("[LocalTransport]");
interface Props {
scope: ObservableScope;
memberships$: Behavior<Epoch<CallMembership[]>>;
client: Pick<MatrixClient, "getDomain"> & OpenIDClientParts;
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts;
roomId: string;
useOldestMember$: Behavior<boolean>;
}
@@ -116,73 +120,150 @@ export const createLocalTransport$ = ({
const FOCI_WK_KEY = "org.matrix.msc4143.rtc_foci";
/**
* Determine the correct Transport for the current session, including
* validating auth against the service to ensure it's correct.
* Prefers in order:
*
* @param client
* @param roomId
* @returns
* 1. The `urlFromDevSettings` value. If this cannot be validated, the function will throw.
* 2. The transports returned via the homeserver.
* 3. The transports returned via .well-known.
* 4. The transport configured in Element Call's config.
*
* @param client The authenticated Matrix client for the current user
* @param roomId The ID of the room to be connected to.
* @param urlFromDevSettings Override URL provided by the user's local config.
* @returns A fully validated transport config.
* @throws MatrixRTCTransportMissingError | FailToGetOpenIdToken
*/
async function makeTransport(
client: Pick<MatrixClient, "getDomain"> & OpenIDClientParts,
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
OpenIDClientParts,
roomId: string,
urlFromDevSettings: string | null,
): Promise<LivekitTransport> {
let transport: LivekitTransport | undefined;
logger.trace("Searching for a preferred transport");
//TODO refactor this to use the jwt service returned alias.
const livekitAlias = roomId;
// We will call `getSFUConfigWithOpenID` once per transport here as it's our
// only mechanism of valiation. 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.
//
// If the OpenID request were to fail then it's acceptable for us to fail
// this function early, as we assume the homeserver has got some problems.
// DEVTOOL: Highest priority: Load from devtool setting
if (urlFromDevSettings !== null) {
const transportFromStorage: LivekitTransport = {
logger.info("Using LiveKit transport from dev tools: ", urlFromDevSettings);
// Validate that the SFU is up. Otherwise, we want to fail on this
// as we don't permit other SFUs.
const config = await getSFUConfigWithOpenID(
client,
urlFromDevSettings,
roomId,
);
return {
type: "livekit",
livekit_service_url: urlFromDevSettings,
livekit_alias: livekitAlias,
livekit_alias: config.livekitAlias,
};
logger.info(
"Using LiveKit transport from dev tools: ",
transportFromStorage,
);
transport = transportFromStorage;
}
// WELL_KNOWN: Prioritize the .well-known/matrix/client, if available, over the configured SFU
async function getFirstUsableTransport(
transports: Transport[],
): Promise<LivekitTransport | null> {
for (const potentialTransport of transports) {
if (isLivekitTransportConfig(potentialTransport)) {
try {
const { livekitAlias } = await getSFUConfigWithOpenID(
client,
potentialTransport.livekit_service_url,
roomId,
);
return {
...potentialTransport,
livekit_alias: livekitAlias,
};
} catch (ex) {
if (ex instanceof FailToGetOpenIdToken) {
// Explictly throw these
throw ex;
}
logger.debug(
`Could not use SFU service "${potentialTransport.livekit_service_url}" as SFU`,
ex,
);
}
}
}
return null;
}
// MSC4143: Attempt to fetch transports from backend.
if ("_unstable_getRTCTransports" in client) {
try {
const selectedTransport = await getFirstUsableTransport(
await client._unstable_getRTCTransports(),
);
if (selectedTransport) {
logger.info("Using backend-configured SFU", selectedTransport);
return selectedTransport;
}
} catch (ex) {
if (ex instanceof MatrixError && ex.httpStatus === 404) {
// Expected, this is an unstable endpoint and it's not required.
logger.debug("Backend does not provide any RTC transports", ex);
} else if (ex instanceof FailToGetOpenIdToken) {
throw ex;
} else {
// We got an error that wasn't just missing support for the feature, so log it loudly.
logger.error(
"Unexpected error fetching RTC transports from backend",
ex,
);
}
}
}
// Legacy MSC4143 (to be removed) WELL_KNOWN: Prioritize the .well-known/matrix/client, if available.
const domain = client.getDomain();
if (domain && transport === undefined) {
if (domain) {
// we use AutoDiscovery instead of relying on the MatrixClient having already
// been fully configured and started
const wellKnownFoci = (await AutoDiscovery.getRawClientConfig(domain))?.[
FOCI_WK_KEY
];
if (Array.isArray(wellKnownFoci)) {
const wellKnownTransport: LivekitTransportConfig | undefined =
wellKnownFoci.find((f) => f && isLivekitTransportConfig(f));
if (wellKnownTransport !== undefined) {
logger.info("Using LiveKit transport from .well-known: ", transport);
transport = { ...wellKnownTransport, livekit_alias: livekitAlias };
}
const selectedTransport = Array.isArray(wellKnownFoci)
? await getFirstUsableTransport(wellKnownFoci)
: null;
if (selectedTransport) {
logger.info("Using .well-known SFU", selectedTransport);
return selectedTransport;
}
}
// CONFIG: Least prioritized; Load from config file
const urlFromConf = Config.get().livekit?.livekit_service_url;
if (urlFromConf && transport === undefined) {
const transportFromConf: LivekitTransport = {
type: "livekit",
livekit_service_url: urlFromConf,
livekit_alias: livekitAlias,
};
logger.info("Using LiveKit transport from config: ", transportFromConf);
transport = transportFromConf;
if (urlFromConf) {
try {
const { livekitAlias } = await getSFUConfigWithOpenID(
client,
urlFromConf,
roomId,
);
const selectedTransport: LivekitTransport = {
type: "livekit",
livekit_service_url: urlFromConf,
livekit_alias: livekitAlias,
};
logger.info("Using config SFU", selectedTransport);
return selectedTransport;
} catch (ex) {
if (ex instanceof FailToGetOpenIdToken) {
throw ex;
}
logger.error("Failed to validate config SFU", ex);
}
}
if (!transport) throw new MatrixRTCTransportMissingError(domain ?? ""); // this will call the jwt/sfu/get endpoint to pre create the livekit room.
await getSFUConfigWithOpenID(
client,
transport.livekit_service_url,
transport.livekit_alias,
);
return transport;
throw new MatrixRTCTransportMissingError(domain ?? "");
}

7
src/state/CallViewModel/remoteMembers/Connection.test.ts
View File

@@ -39,6 +39,7 @@ import {
ElementCallError,
FailToGetOpenIdToken,
} from "../../../utils/errors.ts";
import { testJWTToken } from "../../../utils/test-fixtures.ts";
import { mockRemoteParticipant } from "../../../utils/test.ts";
let testScope: ObservableScope;
@@ -121,7 +122,7 @@ function setupRemoteConnection(): Connection {
status: 200,
body: {
url: "wss://matrix-rtc.m.localhost/livekit/sfu",
jwt: "ATOKEN",
jwt: testJWTToken,
},
};
});
@@ -258,7 +259,7 @@ describe("Start connection states", () => {
capturedState.cause instanceof Error
) {
expect(capturedState.cause.message).toContain(
"SFU Config fetch failed with exception Error",
"SFU Config fetch failed with exception",
);
expect(connection.transport.livekit_alias).toEqual(
livekitFocus.livekit_alias,
@@ -294,7 +295,7 @@ describe("Start connection states", () => {
status: 200,
body: {
url: "wss://matrix-rtc.m.localhost/livekit/sfu",
jwt: "ATOKEN",
jwt: testJWTToken,
},
};
});

3
src/state/CallViewModel/remoteMembers/integration.test.ts
View File

@@ -33,6 +33,7 @@ import {
} from "./MatrixLivekitMembers.ts";
import { createConnectionManager$ } from "./ConnectionManager.ts";
import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
import { testJWTToken } from "../../../utils/test-fixtures.ts";
// Test the integration of ConnectionManager and MatrixLivekitMerger
@@ -85,7 +86,7 @@ beforeEach(() => {
status: 200,
body: {
url: `wss://${domain}/livekit/sfu`,
jwt: "ATOKEN",
jwt: testJWTToken,
},
};
});