Get SFU tokens and delegate delayed leaves through the MSC4195 CS API

src/livekit/auth now fronts two flows. For a transport with a LiveKit SFU
`url`, tokens come from our own homeserver's MSC4195 get_token endpoint
and delayed leaves are delegated through delegate_delayed_leave, both via
the js-sdk's unstable client methods. Transports with only a
livekit_service_url keep using the OpenID flow against the JWT service.

A transport advertising both shapes lets a subscriber fall back to the
OpenID flow when the homeserver answers get_token with a 404 (Synapse
reports M_UNRECOGNIZED for unknown routes rather than M_NOT_FOUND), so a
client can still join calls hosted on servers that have not deployed
MSC4195 yet. Publishers never fall back: a homeserver that advertises a
`url` but does not implement the endpoints is misconfigured and the
admin should hear about it, so they get NoMatrix2AuthorizationService.
Delegation uses the same flow as the token so that the participant
identity the service computes matches.

The local member only probes a transport's delegate_delayed_leave
endpoint when it has a service URL. The developer settings tab validates
a custom service URL through the same facade.

src/state/CallViewModel/CallViewModelTestUtils.ts

@@ -122,6 +122,8 @@ export function withCallViewModel(mode: MatrixRTCMode) {
     };
     const room = mockMatrixRoom({
       client: new (class extends EventEmitter {
+        public readonly baseUrl = "https://matrix.example.com";
+
         public getUserId(): string | undefined {
           return localRtcMember.userId;
         }

src/state/CallViewModel/localMember/LocalMember.test.ts

@@ -116,6 +116,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({
@@ -124,6 +125,7 @@ const mockedClient = {
     matrix_server_name: "localhost",
     expires_in: 10000,
   }),
+  _unstable_delegateDelayedLeave: vi.fn().mockResolvedValue({}),
 };
 
 describe("enterRTCSession", () => {
@@ -239,6 +241,7 @@ describe("LocalMembership", () => {
     ownMembershipIdentity: ownMemberMock,
     client: mockedClient,
     delayId$: constant(null),
+    matrixRTCMode: MATRIX_RTC_MODE,
   };
 
   beforeEach(() => {
@@ -304,9 +307,11 @@ describe("LocalMembership", () => {
     expect(internalLogger).toHaveBeenCalledWith(expextedLog);
   });
 
-  const mockTransport = {
-    livekit_service_url: "a",
-  } as UnstableLivekitTransport;
+  const mockTransport: UnstableLivekitTransport = {
+    type: "livekit",
+    url: "https://sfu.example.org",
+    livekit_service_url: "https://jwt.example.org",
+  };
 
   const mockTransportLocator: livekitAuth.TransportLocator = {
     transport: mockTransport,
@@ -336,10 +341,10 @@ describe("LocalMembership", () => {
     livekitRoom: mockLivekitRoom({}),
   } as unknown as Connection;
 
-  const authCallSpy = vi
-    .spyOn(livekitAuth, "getSFUConfig")
-    .mockImplementation(() => mockedClient.getOpenIdToken());
-  afterEach(() => authCallSpy.mockClear());
+  const delegationSpy = vi
+    .spyOn(livekitAuth, "delegateDelayedLeave")
+    .mockResolvedValue();
+  afterEach(() => delegationSpy.mockClear());
 
   it.each([
     ["no", null, timings],
@@ -348,7 +353,11 @@ describe("LocalMembership", () => {
       "https://matrix.example.org/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave",
       delegatedTimings,
     ],
-    ["transport", "/a/delegate_delayed_leave", delegatedTimings],
+    [
+      "transport",
+      "https://jwt.example.org/delegate_delayed_leave",
+      delegatedTimings,
+    ],
   ])(
     "joins session with %s delegation support",
     async (_serviceName, delegationUrl, delayedLeaveTimings) => {
@@ -379,22 +388,20 @@ describe("LocalMembership", () => {
         delayedLeaveTimings,
       );
 
-      expect(authCallSpy).not.toHaveBeenCalled();
+      expect(delegationSpy).not.toHaveBeenCalled();
       delayId$.next("leave1");
       await flushPromises();
       if (delegationUrl === null) {
-        expect(authCallSpy).not.toHaveBeenCalled();
+        expect(delegationSpy).not.toHaveBeenCalled();
       } else {
         // Delegation is supported in this test case, so go on to check that
         // LocalMember actually performs delegation
         const expectDelegation = (delayId: string) =>
-          expect(authCallSpy).toHaveBeenLastCalledWith({
+          expect(delegationSpy).toHaveBeenLastCalledWith({
             client: mockedClient,
             membership: ownMemberMock,
-            serviceUrl: "a",
+            transport: mockTransport,
             roomId: "!test-room-id:example.org",
-            role: "publisher",
-            delayEndpointBaseUrl: "https://matrix.example.org",
             delayId,
             logger: expect.anything(),
           });

src/state/CallViewModel/localMember/LocalMember.ts

@@ -79,8 +79,9 @@ import {
 } from "../remoteMembers/Connection.ts";
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
 import {
+  type ClientDelegationParts,
+  delegateDelayedLeave,
   type ClientOpenIDParts,
-  getSFUConfig,
   type TransportLocator,
 } from "../../../livekit/auth";
 
@@ -152,7 +153,7 @@ interface Props {
   roomId: string;
   ownMembershipIdentity: CallMembershipIdentityParts;
   localTransport$: Observable<TransportLocator>;
-  client: ClientOpenIDParts;
+  client: ClientDelegationParts & ClientOpenIDParts;
   matrixRTCSession: Pick<
     MatrixRTCSession,
     "slotId" | "updateCallIntent" | "leave"
@@ -720,22 +721,19 @@ export const createLocalMembership$ = ({
     async ([joinParams, delayId]) => {
       if (joinParams?.delegationSupported && delayId !== null) {
         try {
-          if (!("livekit_service_url" in joinParams.transport))
-            throw new Error("Transport has no livekit_service_url");
-          // 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 getSFUConfig({
+          logger.info(
+            `Delegating delayed leave to ${JSON.stringify(joinParams.transport)}…`,
+          );
+          await delegateDelayedLeave({
             client,
             membership: ownMembershipIdentity,
-            serviceUrl: joinParams.transport.livekit_service_url,
-            roomId,
+            transport: joinParams.transport,
             slotId: matrixRTCSession.slotId,
-            role: "publisher",
-            delayEndpointBaseUrl: baseUrl,
+            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`, e);

src/state/CallViewModel/remoteMembers/Connection.test.ts

@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
 
 import {
   afterEach,
+  beforeEach,
   describe,
   expect,
   it,
@@ -27,9 +28,12 @@ import {
 } from "livekit-client";
 import fetchMock from "fetch-mock";
 import EventEmitter from "events";
-import { type IOpenIDToken } from "matrix-js-sdk";
+import { MatrixError } from "matrix-js-sdk";
 import { logger, type Logger } from "matrix-js-sdk/lib/logger";
-import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+import {
+  type LivekitGetTokenResponse,
+  type UnstableLivekitTransport,
+} from "matrix-js-sdk/lib/matrixrtc";
 
 import {
   Connection,
@@ -40,11 +44,14 @@ import { ObservableScope } from "../../ObservableScope.ts";
 import { FailToGetOpenIdToken } from "../../../utils/errors.ts";
 import { testJWTToken } from "../../../utils/test-fixtures.ts";
 import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts";
-import { type ClientOpenIDParts } from "../../../livekit/auth";
+import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
 
 let testScope: ObservableScope;
 
-let client: MockedObject<ClientOpenIDParts>;
+let client: MockedObject<ClientGetTokenParts & ClientOpenIDParts>;
 
 let fakeLivekitRoom: MockedObject<LivekitRoom>;
 
@@ -54,21 +61,21 @@ let fakeLocalParticipant: MockedObject<LocalParticipant>;
 const ROOM_ID = "!roomID:example.org";
 const SLOT_ID = "m.call#ROOM";
 
-const livekitServiceUrl = "https://matrix-rtc.example.org/livekit/jwt";
-const livekitFocus: UnstableLivekitTransport = {
-  livekit_service_url: livekitServiceUrl,
+const transport: UnstableLivekitTransport = {
   type: "livekit",
+  url: "wss://matrix-rtc.m.localhost/livekit/sfu",
 };
 
 function setupTest(): void {
   testScope = new ObservableScope();
-  client = vi.mocked<ClientOpenIDParts>({
+  client = vi.mocked<ClientGetTokenParts & ClientOpenIDParts>({
     getOpenIdToken: vi.fn().mockResolvedValue({
       access_token: "rYsmGUEwNjKgJYyeNUkZseJN",
       token_type: "Bearer",
       matrix_server_name: "example.org",
       expires_in: 3600,
     }),
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
   });
 
   localParticipantEventEmiter = new EventEmitter();
@@ -117,23 +124,13 @@ function setupRemoteConnection(): Connection {
     roomId: ROOM_ID,
     slotId: SLOT_ID,
     role: "subscriber",
-    transport: livekitFocus,
+    transport,
     serverName: "m.localhost",
     scope: testScope,
     ownMembershipIdentity: ownMemberMock,
     livekitRoomFactory: () => fakeLivekitRoom,
   };
 
-  fetchMock.post(`${livekitServiceUrl}/get_token`, () => {
-    return {
-      status: 200,
-      body: {
-        url: "wss://matrix-rtc.m.localhost/livekit/sfu",
-        jwt: testJWTToken,
-      },
-    };
-  });
-
   fakeLivekitRoom.connect.mockImplementation(async (): Promise<void> => {
     const changeEv = RoomEvent.ConnectionStateChanged;
 
@@ -155,6 +152,19 @@ afterEach(() => {
 });
 
 describe("Start connection states", () => {
+  beforeEach(() => {
+    fetchMock.post(
+      `https://matrix-rtc.example.org/livekit/jwt/get_token`,
+      () => {
+        return {
+          // Return a non-retryable error, if not, the retry logic will
+          // wait and fail the test with a timeout.
+          status: 404,
+        };
+      },
+    );
+  });
+
   it("start in initialized state", () => {
     setupTest();
 
@@ -163,7 +173,7 @@ describe("Start connection states", () => {
       roomId: ROOM_ID,
       slotId: SLOT_ID,
       role: "subscriber",
-      transport: livekitFocus,
+      transport,
       serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -174,7 +184,7 @@ describe("Start connection states", () => {
     expect(connection.state$.getValue()).toEqual({ state: "initialized" });
   });
 
-  it("fail to getOpenId token then error state", async () => {
+  it("fail to get SFU token then error state", async () => {
     setupTest();
     vi.useFakeTimers();
 
@@ -183,7 +193,7 @@ describe("Start connection states", () => {
       roomId: ROOM_ID,
       slotId: SLOT_ID,
       role: "subscriber",
-      transport: livekitFocus,
+      transport,
       serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -198,10 +208,10 @@ describe("Start connection states", () => {
     });
     onTestFinished(() => s.unsubscribe());
 
-    const deferred = Promise.withResolvers<IOpenIDToken>();
+    const deferred = Promise.withResolvers<LivekitGetTokenResponse>();
 
-    client.getOpenIdToken.mockImplementation(
-      async (): Promise<IOpenIDToken> => {
+    client._unstable_getLivekitToken.mockImplementation(
+      async (): Promise<LivekitGetTokenResponse> => {
         return await deferred.promise;
       },
     );
@@ -221,7 +231,7 @@ describe("Start connection states", () => {
     capturedState = capturedStates.pop();
     if (capturedState?.state === "error") {
       expect(capturedState.error.message).toEqual("Something went wrong");
-      expect(connection.transport).toEqual(livekitFocus);
+      expect(connection.transport).toEqual(transport);
     } else {
       expect.fail(
         "Expected FailedToStart state but got " + JSON.stringify(capturedState),
@@ -238,7 +248,7 @@ describe("Start connection states", () => {
       roomId: ROOM_ID,
       slotId: SLOT_ID,
       role: "subscriber",
-      transport: livekitFocus,
+      transport,
       serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -254,16 +264,15 @@ describe("Start connection states", () => {
     onTestFinished(() => s.unsubscribe());
 
     const deferredSFU = Promise.withResolvers<void>();
-    // mock the /get_token call
-    fetchMock.post(`${livekitServiceUrl}/get_token`, 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(() => {
@@ -286,7 +295,7 @@ describe("Start connection states", () => {
       expect(capturedState.error.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 " + JSON.stringify(capturedState),
@@ -303,7 +312,7 @@ describe("Start connection states", () => {
       roomId: ROOM_ID,
       slotId: SLOT_ID,
       role: "subscriber",
-      transport: livekitFocus,
+      transport,
       serverName: "m.localhost",
       scope: testScope,
       ownMembershipIdentity: ownMemberMock,
@@ -319,17 +328,6 @@ describe("Start connection states", () => {
     onTestFinished(() => s.unsubscribe());
 
     const deferredSFU = Promise.withResolvers<void>();
-    // mock the /get_token call
-    fetchMock.post(`${livekitServiceUrl}/get_token`, () => {
-      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");
@@ -356,7 +354,7 @@ describe("Start connection states", () => {
       expect(capturedState.error.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),
@@ -564,7 +562,7 @@ describe("remote track logging", () => {
         roomId: ROOM_ID,
         slotId: SLOT_ID,
         role: "subscriber",
-        transport: livekitFocus,
+        transport,
         serverName: "m.localhost",
         scope: testScope,
         ownMembershipIdentity: ownMemberMock,

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -39,6 +39,7 @@ import {
   UnknownCallError,
 } from "../../../utils/errors.ts";
 import {
+  type ClientGetTokenParts,
   getSFUConfig,
   type ClientOpenIDParts,
   type SFUConfig,
@@ -54,7 +55,7 @@ export interface ConnectionOpts {
   /** The name of the homeserver to which the {@link transport} belongs. */
   serverName: string;
   /** The Matrix client to use for OpenID and SFU config requests. */
-  client: ClientOpenIDParts;
+  client: ClientGetTokenParts & ClientOpenIDParts;
   /** 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. */
@@ -134,7 +135,7 @@ export class Connection {
   protected stopped = false;
 
   // TODO: can we just keep the ConnectionOpts object instead of spreading?
-  private readonly client: ClientOpenIDParts;
+  private readonly client: ClientGetTokenParts & ClientOpenIDParts;
   private readonly roomId: string;
   private readonly slotId: string;
   private readonly logger: Logger;
@@ -385,13 +386,12 @@ export class Connection {
   }
 
   protected async getSFUConfig(): Promise<SFUConfig> {
-    if (!("livekit_service_url" in this.transport))
-      throw new Error("Transport has no livekit_service_url");
     return await getSFUConfig({
       client: this.client,
       membership: this.ownMembershipIdentity,
       role: this.role,
-      serviceUrl: this.transport.livekit_service_url,
+      transport: this.transport,
+      serverName: this.serverName,
       roomId: this.roomId,
       slotId: this.slotId,
       logger: this.logger,

src/state/CallViewModel/remoteMembers/ConnectionFactory.ts

@@ -35,7 +35,10 @@ import {
   noiseSuppressionSetting,
   autoGainControlSetting,
 } from "../../../settings/settings.ts";
-import { type ClientOpenIDParts } from "../../../livekit/auth";
+import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
 
 // TODO evaluate if this should be done like the Publisher Factory
 export interface ConnectionFactory {
@@ -65,7 +68,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: ClientOpenIDParts,
+    private client: ClientGetTokenParts & ClientOpenIDParts,
     private readonly roomId: string,
     private readonly slotId: string,
     private devices: MediaDevices,

src/state/CallViewModel/remoteMembers/ECConnectionFactory.test.ts

@@ -14,7 +14,10 @@ import EventEmitter from "events";
 
 import { ObservableScope } from "../../ObservableScope.ts";
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
-import { type ClientOpenIDParts } from "../../../livekit/auth";
+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,12 +55,13 @@ vi.mock("livekit-client", async (importOriginal) => {
 });
 
 let testScope: ObservableScope;
-let mockClient: ClientOpenIDParts;
+let mockClient: ClientGetTokenParts & ClientOpenIDParts;
 
 beforeEach(() => {
   testScope = new ObservableScope();
   mockClient = {
     getOpenIdToken: vi.fn().mockReturnValue(""),
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
   };
 });
 

src/state/CallViewModel/remoteMembers/integration.test.ts

@@ -18,7 +18,10 @@ import {
   trackEpoch,
 } from "../../ObservableScope.ts";
 import { ECConnectionFactory } from "./ConnectionFactory.ts";
-import { type ClientOpenIDParts } from "../../../livekit/auth";
+import {
+  type ClientGetTokenParts,
+  type ClientOpenIDParts,
+} from "../../../livekit/auth";
 import {
   exampleTransportLocator,
   mockMediaDevices,
@@ -40,7 +43,7 @@ import { deepCompare } from "matrix-js-sdk/lib/utils";
 
 let testScope: ObservableScope;
 let ecConnectionFactory: ECConnectionFactory;
-let mockClient: ClientOpenIDParts;
+let mockClient: ClientGetTokenParts & ClientOpenIDParts;
 let lkRoomFactory: () => LivekitRoom;
 
 const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
@@ -49,6 +52,7 @@ beforeEach(() => {
   testScope = new ObservableScope();
   mockClient = {
     getOpenIdToken: vi.fn().mockReturnValue(""),
+    _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: testJWTToken }),
   };
 
   lkRoomFactory = vi.fn().mockImplementation(() => {
@@ -82,7 +86,7 @@ beforeEach(() => {
   );
 
   //TODO a bit annoying to have to do a http mock?
-  const sfuConfigResponse = (url: string): { status: number; body: object } => {
+  fetchMock.post(`path:/sfu/get`, (url) => {
     const domain = new URL(url).hostname; // Extract the domain from the URL
     return {
       status: 200,
@@ -91,9 +95,7 @@ beforeEach(() => {
         jwt: testJWTToken,
       },
     };
-  };
-  fetchMock.post(`path:/sfu/get`, sfuConfigResponse);
-  fetchMock.post(`path:/get_token`, sfuConfigResponse);
+  });
 });
 
 afterEach(() => {