diff --git a/locales/en/app.json b/locales/en/app.json index 5be1bfff6..d57a4e8e3 100644 --- a/locales/en/app.json +++ b/locales/en/app.json @@ -106,7 +106,7 @@ "insufficient_capacity_description": "The server has reached its maximum capacity and you cannot join the call at this time. Try again later, or contact your server admin if the problem persists.", "livekit_connection_error": "Failed to connect to Livekit server", "livekit_connection_error_description": "An error occurred while connecting to the Livekit server (<1>Reason: <2>{{ reason }}).", - "matrix_rtc_transport_missing": "The server is not configured to work with {{brand}}. Please contact your server admin (Domain: {{domain}}, Error Code: {{ errorCode }}).", + "matrix_rtc_transport_missing": "The server is not configured to work with {{brand}} and no other call participant provides a usable transport. Please contact your server admin (Domain: {{domain}}, Error Code: {{ errorCode }}).", "membership_manager": "Membership Manager Error", "membership_manager_description": "The Membership Manager had to shut down. This is caused by many consecutive failed network requests.", "no_matrix_2_authorization_service": "The authorization service for your media server (SFU) is out of date.", diff --git a/package.json b/package.json index 3c7623366..a74c2d92c 100644 --- a/package.json +++ b/package.json @@ -103,7 +103,7 @@ "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-js-sdk": "github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8", "matrix-widget-api": "^1.18.0", "node-stdlib-browser": "^1.3.1", "normalize.css": "^8.0.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index cc0d93beb..ecafc61f5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -191,8 +191,8 @@ 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/4df2f2bfb5826c25f05d9535a8556acecaf2e570 + specifier: github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8 + version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8 matrix-widget-api: specifier: ^1.18.0 version: 1.20.0 @@ -4673,9 +4673,9 @@ 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/4df2f2bfb5826c25f05d9535a8556acecaf2e570: - resolution: {gitHosted: true, integrity: sha512-b1o6ywaPY9jakj+3GIgx3samM9AbqohzPQ5nZKbyOGYB16IHQtNMyjEo/PY0fW/s0uwBVApEHNb/MMtN40c/rQ==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570} - version: 43.0.0 + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8: + resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8} + version: 44.0.0 engines: {node: '>=22.0.0'} matrix-widget-api@1.20.0: @@ -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/4df2f2bfb5826c25f05d9535a8556acecaf2e570: + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8: dependencies: '@babel/runtime': 8.0.0 '@matrix-org/matrix-sdk-crypto-wasm': 18.9.0 diff --git a/src/livekit/auth/index.ts b/src/livekit/auth/index.ts index 292dac4d4..0f4c4f2fc 100644 --- a/src/livekit/auth/index.ts +++ b/src/livekit/auth/index.ts @@ -45,7 +45,7 @@ function isEndpointUnsupported(e: unknown): boolean { * Matrix error. A connection error is neither, as the endpoint may well exist. * @param e The error to check. */ -function isCsApiUnavailable(e: unknown): boolean { +export function isCsApiUnavailable(e: unknown): boolean { if (isEndpointUnsupported(e)) return true; return !(e instanceof MatrixError) && !(e instanceof ConnectionError); } diff --git a/src/livekit/auth/types.ts b/src/livekit/auth/types.ts index 409d88a6b..27c57d2f3 100644 --- a/src/livekit/auth/types.ts +++ b/src/livekit/auth/types.ts @@ -36,7 +36,7 @@ export interface SFUConfig { /** * Decoded details from the JWT. */ -interface SFUJWTPayload { +export interface SFUJWTPayload { /** * Expiration time for the JWT. * Note: This value is in seconds since Unix epoch. @@ -76,8 +76,7 @@ export function extractFullConfigFromToken(sfuConfig: { url: string; jwt: string; }): SFUConfig { - const [, payloadStr] = sfuConfig.jwt.split("."); - const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload; + const payload = decodeSFUJWTPayload(sfuConfig.jwt); return { jwt: sfuConfig.jwt, url: sfuConfig.url, @@ -88,3 +87,11 @@ export function extractFullConfigFromToken(sfuConfig: { livekitIdentity: payload.sub, }; } + +/** + * Decodes the payload of an SFU JWT, without verifying its signature. + */ +export function decodeSFUJWTPayload(jwt: string): SFUJWTPayload { + const [, payloadStr] = jwt.split("."); + return JSON.parse(global.atob(payloadStr)) as SFUJWTPayload; +} diff --git a/src/settings/DeveloperSettingsTab.tsx b/src/settings/DeveloperSettingsTab.tsx index e31282e81..766c492eb 100644 --- a/src/settings/DeveloperSettingsTab.tsx +++ b/src/settings/DeveloperSettingsTab.tsx @@ -527,7 +527,7 @@ export const DeveloperSettingsTab: FC = ({ client, membership: { userId, deviceId, memberId: "" }, transport, - serverName: client.getDomain()!, + serverName: transport.server_name ?? client.getDomain()!, roomId, slotId: "org.example.fake-slot#room", role: "publisher", diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 169abb98c..36d8215de 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -567,7 +567,18 @@ export function createCallViewModel$( const localTransport$ = options.localTransport ? constant(options.localTransport) - : from(getLocalTransport(client)); + : from( + getLocalTransport({ + client, + // A snapshot: if nobody with a usable SFU is here yet, the user has + // to come back later rather than wait in a half-working lobby. + memberships: matrixRTCSession.memberships, + ownMembershipIdentity, + roomId: matrixRoom.roomId, + slotId: matrixRTCSession.slotId, + logger, + }), + ); const connectionFactory = options.connectionFactory ?? diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index f8b4e6b6b..ec0d0d8d9 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -427,6 +427,56 @@ describe("LocalMembership", () => { }, ); + it("joins without delegation when publishing on a remote SFU", async () => { + // Our homeserver only delegates to its own SFUs (MSC4195), so publishing + // on another homeserver's SFU means restarting the delayed leave ourselves. + const scope = testScope(); + const joinMatrixRTC = vi.fn(); + const delayId$ = new BehaviorSubject(null); + const remoteTransport: UnstableLivekitTransport = { + type: "livekit", + url: "https://sfu.sample.com", + server_name: "sample.com", + }; + + const connectionMap = new ConnectionMap(); + connectionMap.add( + { + ...connectionTransportAConnected, + transport: remoteTransport, + serverName: "sample.com", + } as Connection, + [], + ); + // Our own homeserver does support delegation, so only the SFU being + // remote can explain delegation being skipped + fetchMock.post( + "https://matrix.example.org/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave", + () => ({ status: 401, body: {} }), + ); + + const localMembership = createLocalMembership$({ + scope, + ...defaultCreateLocalMemberValues, + connectionMap$: constant(new Epoch(connectionMap)), + joinMatrixRTC, + localTransport$: constant({ + transport: remoteTransport, + serverName: "sample.com", + }), + delayId$, + }); + + localMembership.requestJoinAndPublish(); + void (await fetchMock.flush()); + await flushPromises(); + expect(joinMatrixRTC).toHaveBeenCalledWith(remoteTransport, timings); + + delayId$.next("leave1"); + await flushPromises(); + expect(delegationSpy).not.toHaveBeenCalled(); + }); + it("only starts tracks if requested", async () => { const scope = new ObservableScope(); diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index baa6b473d..90c78598d 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -43,6 +43,7 @@ import { Subject, } from "rxjs"; import { type Logger } from "matrix-js-sdk/lib/logger"; +import { type MatrixClient } from "matrix-js-sdk"; import { deepCompare } from "matrix-js-sdk/lib/utils"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; @@ -154,7 +155,9 @@ interface Props { roomId: string; ownMembershipIdentity: CallMembershipIdentityParts; localTransport$: Observable; - client: ClientDelegationParts & ClientOpenIDParts; + client: ClientDelegationParts & + ClientOpenIDParts & + Pick; matrixRTCSession: Pick< MatrixRTCSession, "slotId" | "updateCallIntent" | "leave" @@ -312,17 +315,26 @@ export const createLocalMembership$ = ({ // to whether delayed event delegation is supported const joinParams$ = scope.behavior( localTransport$.pipe( - switchMap(async ({ transport }) => { - // Delegation goes through whichever service authenticates us as a - // publisher, so only that service's support counts - const flow = chooseAuthFlow(transport, "publisher"); - const delegationSupported = - flow.kind === "openID" - ? await checkDelegationSupport( - flow.serviceUrl + "/delegate_delayed_leave", - `transport ${flow.serviceUrl}`, - ) - : await homeserverSupportsDelegation; + switchMap(async ({ transport, serverName }) => { + let delegationSupported: boolean; + if (serverName !== client.getDomain()) { + // Delegation is only possible when publishing on a local SFU. + logger.info( + `Publishing on a remote SFU of ${serverName}, delegation is not possible`, + ); + delegationSupported = false; + } else { + // Delegation goes through whichever service authenticates us as a + // publisher, so only that service's support counts + const flow = chooseAuthFlow(transport, "publisher"); + delegationSupported = + flow.kind === "openID" + ? await checkDelegationSupport( + flow.serviceUrl + "/delegate_delayed_leave", + `transport ${flow.serviceUrl}`, + ) + : await homeserverSupportsDelegation; + } return { transport, delegationSupported }; }), ), diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index ed6654e2f..a74949ce9 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -15,68 +15,94 @@ import { vi, } from "vitest"; import fetchMock from "fetch-mock"; +import { MatrixError } from "matrix-js-sdk"; +import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc"; -import { mockConfig } from "../../../utils/test"; -import { getLocalTransport } from "./LocalTransport"; +import { + mockConfig, + mockRtcMembership, + ownMemberMock, +} from "../../../utils/test"; +import { getLocalTransport, type LocalTransportClient } from "./LocalTransport"; import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { customTransport } from "../../../settings/settings"; -import { type MatrixClient } from "matrix-js-sdk"; + +/** + * Builds a JWT whose payload states whether the issuing server lets us publish. + */ +function jwtWithPublishGrant(canPublish: boolean): string { + return [ + {}, // header + { sub: "identity", video: { room: "room", canPublish } }, // payload + {}, // signature + ] + .map((d) => global.btoa(JSON.stringify(d))) + .join("."); +} + +const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" }; +const carolSfu = { type: "livekit", url: "wss://sfu.carol.com" }; + +// Bob is the oldest member, Carol joined after him +const bob = mockRtcMembership("@bob:bob.com", "BOB", { + fociPreferred: [bobSfu], + membership: { created_ts: 100 }, +}); +const carol = mockRtcMembership("@carol:carol.com", "CAROL", { + fociPreferred: [carolSfu], + membership: { created_ts: 200 }, +}); +const legacyDave = mockRtcMembership("@dave:dave.com", "DAVE", { + fociPreferred: [ + { type: "livekit", livekit_service_url: "https://jwt.dave.com" }, + ], + membership: { created_ts: 50 }, +}); describe("LocalTransport", () => { - beforeEach(() => vi.clearAllMocks()); + let client: MockedObject; + const params = ( + memberships: CallMembership[] = [], + ): Parameters[0] => ({ + client, + memberships, + ownMembershipIdentity: ownMemberMock, + roomId: "!room:example.org", + slotId: "m.call#ROOM", + }); + + beforeEach(() => { + vi.clearAllMocks(); + mockConfig({}); + customTransport.setValue(customTransport.defaultValue); + client = { + getDomain: vi.fn().mockReturnValue("example.org"), + _unstable_getRTCTransports: vi.fn().mockResolvedValue([]), + _unstable_getLivekitToken: vi + .fn() + .mockRejectedValue( + new MatrixError({ errcode: "M_UNRECOGNIZED", error: "" }, 404), + ), + }; + }); + + afterEach(() => { + fetchMock.reset(); + }); it("throws if config is missing", async () => { - await expect( - getLocalTransport({ - _unstable_getRTCTransports: async () => Promise.resolve([]), - getDomain: () => "example.org", - }), - ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org")); + await expect(getLocalTransport(params())).rejects.toThrow( + new MatrixRTCTransportMissingError("example.org"), + ); }); - it("returns preferred transport", async () => { - // Use config so transport discovery succeeds - mockConfig({ - livekit: { livekit_service_url: "https://lk.example.org" }, - }); - - expect( - await getLocalTransport({ - _unstable_getRTCTransports: async () => Promise.resolve([]), - getDomain: () => "example.org", - }), - ).toStrictEqual({ - transport: { - livekit_service_url: "https://lk.example.org", - type: "livekit", - }, - serverName: "example.org", - }); - }); - - describe("transport configuration mechanisms", () => { - let client: MockedObject< - Pick - >; - beforeEach(() => { - mockConfig({}); - customTransport.setValue(customTransport.defaultValue); - client = { - getDomain: vi.fn().mockReturnValue("example.org"), - _unstable_getRTCTransports: vi.fn().mockResolvedValue([]), - }; - }); - - afterEach(() => { - fetchMock.reset(); - }); - + describe("local transport configuration mechanisms", () => { it("supports getting transport via application config", async () => { mockConfig({ livekit: { livekit_service_url: "https://lk.example.org" }, }); - expect(await getLocalTransport(client)).toStrictEqual({ + expect(await getLocalTransport(params())).toStrictEqual({ transport: { livekit_service_url: "https://lk.example.org", type: "livekit", @@ -91,7 +117,7 @@ describe("LocalTransport", () => { livekit_service_url: "https://lk.example.org", }); - expect(await getLocalTransport(client)).toStrictEqual({ + expect(await getLocalTransport(params())).toStrictEqual({ transport: { livekit_service_url: "https://lk.example.org", type: "livekit", @@ -100,12 +126,29 @@ describe("LocalTransport", () => { }); }); + it("locates a custom transport with server_name on that homeserver", async () => { + customTransport.setValue({ + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", + }); + + expect(await getLocalTransport(params())).toStrictEqual({ + transport: { + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", + }, + serverName: "bob.com", + }); + }); + it("supports getting transport via backend", async () => { client._unstable_getRTCTransports.mockResolvedValue([ { type: "livekit", livekit_service_url: "https://lk.example.org" }, ]); - expect(await getLocalTransport(client)).toStrictEqual({ + expect(await getLocalTransport(params())).toStrictEqual({ transport: { livekit_service_url: "https://lk.example.org", type: "livekit", @@ -114,13 +157,110 @@ describe("LocalTransport", () => { }); }); + it("prefers a local transport over other members' SFUs", async () => { + client._unstable_getRTCTransports.mockResolvedValue([ + { type: "livekit", url: "wss://sfu.example.org" }, + ]); + + expect(await getLocalTransport(params([bob]))).toStrictEqual({ + transport: { type: "livekit", url: "wss://sfu.example.org" }, + serverName: "example.org", + }); + expect(client._unstable_getLivekitToken).not.toHaveBeenCalled(); + }); + it("throws if no options are available", async () => { - await expect( - getLocalTransport({ - getDomain: () => "example.org", - _unstable_getRTCTransports: async () => Promise.resolve([]), - }), - ).rejects.toThrow(new MatrixRTCTransportMissingError("example.org")); + await expect(getLocalTransport(params())).rejects.toThrow( + new MatrixRTCTransportMissingError("example.org"), + ); + }); + }); + + describe("publishing on another member's SFU", () => { + beforeEach(() => { + client._unstable_getLivekitToken.mockResolvedValue({ + jwt: jwtWithPublishGrant(true), + }); + }); + + it("publishes on the oldest member's SFU with its server_name", async () => { + expect(await getLocalTransport(params([carol, bob]))).toStrictEqual({ + transport: { + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", + }, + serverName: "bob.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({ + url: "wss://sfu.bob.com", + server_name: "bob.com", + room_id: "!room:example.org", + slot_id: "m.call#ROOM", + member_id: ownMemberMock.memberId, + }); + }); + + it("ignores members publishing through a legacy JWT service", async () => { + expect(await getLocalTransport(params([legacyDave, bob]))).toMatchObject({ + serverName: "bob.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + }); + + it("moves on when a homeserver does not let us publish", async () => { + client._unstable_getLivekitToken.mockImplementation( + async ({ server_name }) => + Promise.resolve({ + jwt: jwtWithPublishGrant(server_name === "carol.com"), + }), + ); + + expect(await getLocalTransport(params([bob, carol]))).toMatchObject({ + serverName: "carol.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); + }); + + it("moves on when a homeserver refuses to issue a token", async () => { + client._unstable_getLivekitToken.mockImplementation( + async ({ server_name }) => + server_name === "bob.com" + ? Promise.reject( + new MatrixError( + { errcode: "M_FORBIDDEN", error: "Not joined" }, + 403, + ), + ) + : Promise.resolve({ jwt: jwtWithPublishGrant(true) }), + ); + + expect(await getLocalTransport(params([bob, carol]))).toMatchObject({ + serverName: "carol.com", + }); + }); + + it("throws once all candidates are exhausted", async () => { + client._unstable_getLivekitToken.mockResolvedValue({ + jwt: jwtWithPublishGrant(false), + }); + + await expect(getLocalTransport(params([bob, carol]))).rejects.toThrow( + new MatrixRTCTransportMissingError("example.org"), + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); + }); + + it("gives up immediately if our homeserver lacks the CS API", async () => { + client._unstable_getLivekitToken.mockRejectedValue( + new MatrixError({ errcode: "M_UNRECOGNIZED", error: "" }, 404), + ); + + await expect(getLocalTransport(params([bob, carol]))).rejects.toThrow( + new MatrixRTCTransportMissingError("example.org"), + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); }); }); }); diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index 3b43f8112..7789b6666 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -6,28 +6,78 @@ Please see LICENSE in the repository root for full details. */ import { type MatrixClient } from "matrix-js-sdk"; -import { logger as rootLogger } from "matrix-js-sdk/lib/logger"; +import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger"; +import { + type CallMembership, + getRemoteLivekitPublishCandidates, + type LivekitPublishCandidate, +} from "matrix-js-sdk/lib/matrixrtc"; +import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { Config } from "../../../config/Config.ts"; import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts"; import { customTransport as customTransportSetting } from "../../../settings/settings.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; -import { type TransportLocator } from "../../../livekit/auth/types.ts"; +import { + type TransportLocator, + isCsApiUnavailable, +} from "../../../livekit/auth"; +import * as csApi from "../../../livekit/auth/csApi.ts"; +import { decodeSFUJWTPayload } from "../../../livekit/auth/types.ts"; + +export type LocalTransportClient = Pick< + MatrixClient, + "getDomain" | "_unstable_getRTCTransports" | "_unstable_getLivekitToken" +>; + +export interface LocalTransportParams { + client: LocalTransportClient; + /** + * The current members of the session. If our own homeserver provides no + * transport, we fall back to publishing on one of theirs. + */ + memberships: CallMembership[]; + /** + * Data identifying our own session membership, needed to request tokens. + */ + ownMembershipIdentity: CallMembershipIdentityParts; + /** + * The ID of the Matrix room in which the session takes place. + */ + roomId: string; + /** + * The ID of the MatrixRTC slot in which the session takes place. + */ + slotId: string; + logger?: Logger; +} /** * Determines the transport to advertise in our MatrixRTC membership and publish * media on. + * + * This tries to fall back to the SFU of another session member if needed. + * + * @throws {MatrixRTCTransportMissingError} if neither a local nor a remote + * transport is available. */ -export async function getLocalTransport( - client: Pick, -): Promise { - const serverName = client.getDomain(); - if (serverName === null) throw new Error("No server name"); +export async function getLocalTransport({ + client, + memberships, + ownMembershipIdentity, + roomId, + slotId, + logger: parentLogger = rootLogger, +}: LocalTransportParams): Promise { + const logger = parentLogger.getChild("[LocalTransport]"); + const domain = client.getDomain(); + if (domain === null) throw new Error("No server name"); + const ownServerName: string = domain; const discovery = new RtcTransportAutoDiscovery({ client, resolvedConfig: Config.get(), - logger: rootLogger.getChild("[LocalTransport]"), + logger, }); // Respect the user's custom transport, if set @@ -35,7 +85,64 @@ export async function getLocalTransport( customTransportSetting.value$.value ?? (await discovery.discoverPreferredTransport()); - if (transport === null) throw new MatrixRTCTransportMissingError(serverName); + if (transport !== null) { + // A custom transport may point at another homeserver's SFU + const serverName = + "server_name" in transport && transport.server_name !== undefined + ? transport.server_name + : ownServerName; + return { transport, serverName }; + } - return { transport, serverName }; + logger.info( + "No local transport available, looking for another member's SFU to publish on", + ); + const candidates = getRemoteLivekitPublishCandidates( + memberships, + ownMembershipIdentity, + ); + for (const candidate of candidates) { + const locator = await tryRemoteCandidate(candidate); + if (locator !== null) return locator; + } + throw new MatrixRTCTransportMissingError(ownServerName); + + /** + * Checks whether the homeserver behind a candidate SFU lets us publish on + * it. Servers MAY restrict remote publishers, in which case the token comes + * back without the publish grant and we move on to the next candidate. + * @returns The locator to publish with, or null if this candidate is unusable. + * @throws {MatrixRTCTransportMissingError} if the client-server API is not + * available at all, as no candidate can work then. + */ + async function tryRemoteCandidate({ + url, + serverName, + transport, + }: LivekitPublishCandidate): Promise { + try { + const { jwt } = await csApi.getSFUConfig({ + client, + membership: ownMembershipIdentity, + url, + serverName, + roomId, + slotId, + }); + if (decodeSFUJWTPayload(jwt).video.canPublish) { + logger.info(`Publishing on remote SFU ${url} of ${serverName}`); + return { transport, serverName }; + } + logger.info(`${serverName} does not let us publish on ${url}, skipping`); + } catch (e) { + if (isCsApiUnavailable(e)) { + logger.warn( + `Cannot use the MSC4195 get_token endpoint (${e}), so no remote SFU can be used`, + ); + throw new MatrixRTCTransportMissingError(ownServerName); + } + logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e); + } + return null; + } } diff --git a/src/state/SessionBehaviors.test.ts b/src/state/SessionBehaviors.test.ts index d9da6f830..2a3e1c67e 100644 --- a/src/state/SessionBehaviors.test.ts +++ b/src/state/SessionBehaviors.test.ts @@ -83,6 +83,26 @@ describe("SessionBehaviors", () => { ).toEqual(["example.org", "sample.com"]); }); + it("locates a remote publisher's transport on the SFU's homeserver", () => { + // Dave has no SFU of his own and publishes on sample.com's + const remoteTransport: UnstableLivekitTransport = { + ...urlTransport, + server_name: "sample.com", + }; + const dave = mockRtcMembership("@dave:example.org", "DEV222", { + fociPreferred: [remoteTransport], + }); + + const { transports$ } = membershipsAndTransports$( + testScope(), + constant(new Epoch([dave])), + ); + + expect(transports$.value.value).toEqual([ + { transport: remoteTransport, serverName: "sample.com" }, + ]); + }); + it("leaves out members without a LiveKit transport", () => { const dave = mockRtcMembership("@dave:example.org", "DEV222", { fociPreferred: [{ type: "something-else" } as Transport], diff --git a/src/state/SessionBehaviors.ts b/src/state/SessionBehaviors.ts index 6620350cf..eeaa655de 100644 --- a/src/state/SessionBehaviors.ts +++ b/src/state/SessionBehaviors.ts @@ -51,7 +51,9 @@ export const membershipsAndTransports$ = ( transport: isUnstableLivekitTransport(transport) ? { transport, - serverName: membership.userId.replace(/^.*?:/, ""), + // The SFU's own homeserver, which is not necessarily the + // member's if they publish on a remote SFU (MSC4195) + serverName: membership.getTransportServerName(transport), } : undefined, }; diff --git a/src/widget.test.ts b/src/widget.test.ts index c645c3b34..908159158 100644 --- a/src/widget.test.ts +++ b/src/widget.test.ts @@ -111,6 +111,8 @@ describe("widget", () => { receiveToDevice: sendRecvToDevice, turnServers: false, rtcTransports: true, + rtcLivekitGetToken: true, + rtcLivekitDelegateDelayedLeave: true, sendDelayedEvents: true, updateDelayedEvents: true, sendSticky: true, diff --git a/src/widget.ts b/src/widget.ts index b35e335bc..f1847004b 100644 --- a/src/widget.ts +++ b/src/widget.ts @@ -175,6 +175,8 @@ export const initializeWidget = ( receiveToDevice: sendRecvToDevice, turnServers: false, rtcTransports: true, + rtcLivekitGetToken: true, + rtcLivekitDelegateDelayedLeave: true, sendDelayedEvents: true, updateDelayedEvents: true, sendSticky: true,