[SDK] Move SFU authentication into livekit/auth and choose the endpoint by role

The OpenID flow moves to src/livekit/auth/openID.ts with the SFU config
types in a sibling file, takes an options object, and replaces the
MatrixRTC mode with the role we take on the transport plus the session's
slot ID instead of a hard-coded m.call#ROOM.

This changes behaviour: publishers on a livekit_service_url transport
always use the legacy /sfu/get endpoint, so that subscribers can predict
their user:device participant identity, and subscribers always use the
default /get_token endpoint. The MatrixRTC mode no longer influences which
endpoint is used.

On the SDK this lands as sdk/src/media-backend/livekit/auth/ with the connection map, the discriminated connection state and the client-server API flow of the following commits of element-hq/element-call#4331, which rewrite the same files:
- Model ConnectionState as a discriminated union
- Show the whole transport in debug UI and logs (SDK part)
- Let the publishing Connection authenticate itself
- Reduce ConnectionManager to a ConnectionMap
- Use the unstable LiveKit transport type
- Identify transports by server name as well as URLs
- Get SFU tokens and delegate delayed leaves through the MSC4195 CS API
- Fall back to get_token as subscriber on widget API errors
- Share a connection between homeservers using the same JWT service
- Make the auth choice mode independent again

Co-authored-by: Johannes Marbach <n0-0ne+github@mailbox.org>
(cherry picked from commit adb380cd9ed108856e42d992fc03ced8c44f1e56)

sdk/src/media-backend/livekit/LivekitBackend.ts

@@ -8,71 +8,86 @@ Please see LICENSE in the repository root for full details.
 import { type MatrixClient } from "matrix-js-sdk";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 import {
-  isLivekitTransport,
-  type LivekitTransport,
+  isUnstableLivekitTransport,
   type Transport,
 } from "matrix-js-sdk/lib/matrixrtc";
 import {
   BehaviorSubject,
   combineLatest,
   distinctUntilChanged,
+  filter,
+  firstValueFrom,
   map,
   type Observable,
   of,
-  ReplaySubject,
   switchMap,
 } from "rxjs";
 
 import { type ResolvedTransport, type RTCMember } from "../../api";
-
 import { type MediaQuality } from "../../config";
 import { MatrixRTCError } from "../../errors";
-import { type Behavior } from "../../reactive/Behavior";
 import { type MemberMedia } from "../../media-api";
+import { type Behavior } from "../../reactive/Behavior";
 import {
-  type Epoch,
+  Epoch,
   type ObservableScope,
   trackEpoch,
 } from "../../reactive/ObservableScope";
+import { mapScoped } from "../../utils/mapScoped";
+import { serverNameOf } from "../../utils/serverName";
 import {
   type BackendConnection,
   type MediaBackend,
   type MediaBackendContext,
   type TransportCapabilities,
 } from "../api";
+import {
+  chooseAuthFlow,
+  delegateDelayedLeave,
+  type TransportLocator,
+} from "./auth";
 import { type Connection } from "./Connection";
+import {
+  ConnectionMap,
+  createConnectionMap$,
+  keyFor,
+  type TransportKey,
+} from "./ConnectionMap";
 import {
   LivekitConnectionFactory,
   ResolvedConnection,
 } from "./ConnectionFactory";
-import {
-  createConnectionManager$,
-  type LocalTransport,
-} from "./ConnectionManager";
 import { toMediaConnectionState } from "./connectionState";
 import { createDataChannel$ } from "./DataChannel";
 import { createKeyProvider } from "./KeyProvider";
 import { createLivekitMemberMedia } from "./LivekitMemberMedia";
 import { createLivekitLocalMedia } from "./LocalMedia";
-import { getSFUConfigWithOpenID, type TokenEndpoint } from "./openIDSFU";
 
 export interface LivekitBackendOptions {
-  /** `getOpenIdToken` for the token exchange, `baseUrl` for the delegation probe and handover. */
-  client: Pick<MatrixClient, "getOpenIdToken" | "baseUrl">;
+  /**
+   * `getOpenIdToken` for the JWT service flow, the MSC4195 methods for the
+   * client-server API flow, `baseUrl` for the delegation probe and handover.
+   */
+  client: Pick<
+    MatrixClient,
+    | "getOpenIdToken"
+    | "_unstable_getLivekitToken"
+    | "_unstable_delegateDelayedLeave"
+    | "baseUrl"
+  >;
   mediaQuality?: MediaQuality;
-  /** Which token service endpoint the local transport uses; remote ones try both. */
-  tokenEndpoint: TokenEndpoint;
 }
 
 /**
  * MatrixRTC media over LiveKit SFUs: one room per transport, the local one
- * published on, every remote one subscribed to, with the token service
- * protocol in front of each.
+ * published on, every remote one subscribed to, each authenticated for its
+ * role through the homeserver or the transport's JWT service.
  */
 export function createLivekitBackend(
   scope: ObservableScope,
   {
     roomId,
+    slotId,
     ownMembershipIdentity,
     publish,
     encryptionSystem,
@@ -80,102 +95,133 @@ export function createLivekitBackend(
     timings,
     logger: parentLogger,
   }: MediaBackendContext,
-  { client, mediaQuality, tokenEndpoint }: LivekitBackendOptions,
+  { client, mediaQuality }: LivekitBackendOptions,
 ): MediaBackend {
   const logger = parentLogger.getChild("[LivekitBackend]");
   const keyProvider = createKeyProvider(encryptionSystem, mediaKeys$, logger);
   const transports = createTransportRegistry(scope);
-  const localTransport$ = new ReplaySubject<LocalTransport>(1);
+  const localTransport$ = new BehaviorSubject<TransportLocator | null>(null);
   const audioOutputDeviceId$ = new BehaviorSubject<string | undefined>(
     undefined,
   );
+  const connectionFactory = new LivekitConnectionFactory(
+    client,
+    roomId,
+    slotId,
+    publish,
+    audioOutputDeviceId$,
+    keyProvider,
+    mediaQuality,
+  );
 
-  const connectionManager = createConnectionManager$({
+  // A new map, in a scope of its own, whenever the local transport changes:
+  // the publisher role sits on one connection, and which one is fixed per map
+  const connectionMap$ = scope.behavior<Epoch<ConnectionMap>>(
+    mapScoped(scope, localTransport$, (mapScope, localTransport) =>
+      createConnectionMap$({
+        scope: mapScope,
+        connectionFactory,
+        localTransport,
+        remoteTransports$: transports.transports$,
+        logger,
+        ownMembershipIdentity,
+      }),
+    ).pipe(switchMap((map$) => map$ ?? of(new Epoch(new ConnectionMap())))),
+  );
+
+  const localMedia_ = createLivekitLocalMedia({
     scope,
-    connectionFactory: new LivekitConnectionFactory(
-      client,
-      roomId,
-      publish,
-      audioOutputDeviceId$,
-      keyProvider,
-      mediaQuality,
-    ),
+    connectionMap$,
     localTransport$,
-    remoteTransports$: transports.transports$,
+    publish,
+    audioOutputDeviceId$,
+    encryptionSystem,
     logger,
-    ownMembershipIdentity,
   });
 
-  let local: LivekitTransport | undefined;
+  // Probed once: the answer cannot change within a session
+  const homeserverSupportsDelegation = supportsDelegation(
+    `${client.baseUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
+    "homeserver",
+    logger,
+  );
+
   const prepareLocalTransport = async (
     transport: Transport,
   ): Promise<TransportCapabilities> => {
-    if (!isLivekitTransport(transport))
+    if (!isUnstableLivekitTransport(transport))
       throw new MatrixRTCError(`Not a LiveKit transport: ${transport.type}`);
-    const [sfuConfig, canDelegateDelayedLeave] = await Promise.all([
-      getSFUConfigWithOpenID(
-        client,
-        ownMembershipIdentity,
-        transport.livekit_service_url,
-        roomId,
-        { tokenEndpoint },
-        logger,
-      ),
-      supportsDelegation(client.baseUrl, transport, logger),
+    const locator: TransportLocator = {
+      transport,
+      serverName: serverNameOf(ownMembershipIdentity.userId),
+    };
+    // Delegation goes through whichever service authenticates us as a
+    // publisher, so only that service's support counts
+    const flow = chooseAuthFlow(transport, "publisher");
+    const canDelegate =
+      flow.kind === "openID"
+        ? supportsDelegation(
+            `${flow.serviceUrl}/delegate_delayed_leave`,
+            `transport ${flow.serviceUrl}`,
+            logger,
+          )
+        : homeserverSupportsDelegation;
+    localTransport$.next(locator);
+    // The membership is only sent once the publisher connection holds its
+    // token, so that the LiveKit room exists before anyone subscribes to it:
+    // a subscriber lacks the permission to create it and would fail without
+    // retrying
+    const [, canDelegateDelayedLeave] = await Promise.all([
+      authenticated(localMedia_.connection$),
+      canDelegate,
     ]);
-    local = transport;
-    localTransport$.next({ transport, sfuConfig });
     return { canDelegateDelayedLeave };
   };
 
-  // The token this issues is discarded; the request is what hands the leave over
-  const delegateDelayedLeave = async (delayId: string): Promise<void> => {
-    if (local === undefined)
+  const delegateLeave = async (delayId: string): Promise<void> => {
+    const local = localTransport$.value;
+    if (local === null)
       throw new MatrixRTCError("No local transport to delegate the leave to");
-    await getSFUConfigWithOpenID(
+    await delegateDelayedLeave({
       client,
-      ownMembershipIdentity,
-      local.livekit_service_url,
+      membership: ownMembershipIdentity,
+      transport: local.transport,
       roomId,
-      {
-        tokenEndpoint,
-        delayEndpointBaseUrl: client.baseUrl,
-        delayId,
-        delayTimeoutMs: timings.delegatedDelayedLeave.delay_ms,
-      },
+      slotId,
+      delayId,
+      delayTimeoutMs: timings.delegatedDelayedLeave.delay_ms,
       logger,
-    );
+    });
   };
 
   const mediaFor = (
     memberScope: ObservableScope,
     member: RTCMember,
   ): MemberMedia => {
-    const identity = member.rtcBackendIdentity;
-    const transport$ = memberScope.behavior(
+    const { backendIdentities } = member;
+    const locator$ = memberScope.behavior(
       member.transport$.pipe(
-        map((transport) => transport?.raw),
-        map((transport) =>
-          isLivekitTransport(transport) ? transport : undefined,
-        ),
-        distinctUntilChanged(
-          (a, b) => a?.livekit_service_url === b?.livekit_service_url,
-        ),
+        map((transport): TransportLocator | undefined => {
+          const raw = transport?.raw;
+          return isUnstableLivekitTransport(raw)
+            ? { transport: raw, serverName: serverNameOf(member.userId) }
+            : undefined;
+        }),
+        distinctUntilChanged((a, b) => (a && keyFor(a)) === (b && keyFor(b))),
       ),
     );
-    transports.follow(memberScope, transport$);
+    transports.follow(memberScope, locator$);
 
     const source$ = memberScope.behavior(
-      combineLatest([
-        transport$,
-        connectionManager.connectionManagerData$,
-      ]).pipe(
-        map(([transport, { value: connections }]) => {
-          if (transport === undefined) return null;
-          const connection = connections.getConnectionForTransport(transport);
+      combineLatest([locator$, connectionMap$]).pipe(
+        map(([locator, { value: connections }]) => {
+          if (locator === undefined) return null;
+          const connection = connections.getConnectionForTransport(locator);
+          // A member advertising both a `url` and a `livekit_service_url` may
+          // have authenticated through either flow, under either identity
           const participant = connections
-            .getParticipantsForTransport(transport)
-            .find((p) => p.identity === identity);
+            .getParticipantsForTransport(locator)
+            .find((p) => backendIdentities.includes(p.identity));
           return connection && participant
             ? { participant, room: connection.livekitRoom }
             : null;
@@ -187,14 +233,14 @@ export function createLivekitBackend(
     );
     source$.pipe(memberScope.bind()).subscribe((source) => {
       logger.info(
-        `${identity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"}`,
+        `${member.rtcBackendIdentity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"} on ${locator$.value ? JSON.stringify(locator$.value.transport) : "no transport"}`,
       );
     });
     return createLivekitMemberMedia(memberScope, source$, encryptionSystem);
   };
 
   const connections$ = scope.behavior<BackendConnection[]>(
-    connectionManager.connectionManagerData$.pipe(
+    connectionMap$.pipe(
       switchMap(({ value }) => {
         const connections = value.getConnections();
         if (connections.length === 0) return of([]);
@@ -203,26 +249,16 @@ export function createLivekitBackend(
     ),
   );
 
-  const localMedia_ = createLivekitLocalMedia({
-    scope,
-    connectionManager,
-    localTransport$,
-    publish,
-    audioOutputDeviceId$,
-    encryptionSystem,
-    logger,
-  });
-
   const { data$, sendData } = createDataChannel$({
     scope,
-    connectionManager,
+    connectionMap$,
     connection$: localMedia_.connection$,
   });
 
   return {
     transportType: "livekit",
     prepareLocalTransport,
-    delegateDelayedLeave,
+    delegateDelayedLeave: delegateLeave,
     local: localMedia_,
     mediaFor,
     connections$,
@@ -232,6 +268,29 @@ export function createLivekitBackend(
   };
 }
 
+/** Resolves once the connection holds its token; rejects with its error otherwise. */
+async function authenticated(
+  connection$: Behavior<Connection | null>,
+): Promise<void> {
+  const state = await firstValueFrom(
+    connection$.pipe(
+      switchMap((connection) => connection?.state$ ?? of(null)),
+      filter(
+        (state) =>
+          state !== null &&
+          (state.state === "authenticated" ||
+            state.state === "error" ||
+            state.state === "stopped"),
+      ),
+    ),
+  );
+  if (state?.state === "error") throw state.error;
+  if (state?.state === "stopped")
+    throw new MatrixRTCError(
+      "The local connection stopped before it was authenticated",
+    );
+}
+
 function backendConnection$(
   connection: Connection,
 ): Observable<BackendConnection> {
@@ -252,10 +311,10 @@ export interface TransportRegistry {
   /** Keeps a member's transport registered for as long as its scope lives. */
   follow(
     scope: ObservableScope,
-    transport$: Behavior<LivekitTransport | undefined>,
+    locator$: Behavior<TransportLocator | undefined>,
   ): void;
-  /** Every transport some member is on, for the connection manager. */
-  transports$: Behavior<Epoch<LivekitTransport[]>>;
+  /** Every transport some member is on, for the connection map. */
+  transports$: Behavior<Epoch<TransportLocator[]>>;
 }
 
 /**
@@ -267,34 +326,34 @@ export function createTransportRegistry(
   scope: ObservableScope,
 ): TransportRegistry {
   const counts = new Map<
-    string,
-    { transport: LivekitTransport; count: number }
+    TransportKey,
+    { locator: TransportLocator; count: number }
   >();
-  const transports = new BehaviorSubject<LivekitTransport[]>([]);
+  const transports = new BehaviorSubject<TransportLocator[]>([]);
   const publish = (): void =>
-    transports.next([...counts.values()].map(({ transport }) => transport));
+    transports.next([...counts.values()].map(({ locator }) => locator));
 
-  const register = (transport: LivekitTransport): (() => void) => {
-    const url = transport.livekit_service_url;
-    const entry = counts.get(url) ?? { transport, count: 0 };
+  const register = (locator: TransportLocator): (() => void) => {
+    const key = keyFor(locator);
+    const entry = counts.get(key) ?? { locator, count: 0 };
     entry.count++;
     if (entry.count === 1) {
-      counts.set(url, entry);
+      counts.set(key, entry);
       publish();
     }
     return () => {
       if (--entry.count > 0) return;
-      counts.delete(url);
+      counts.delete(key);
       publish();
     };
   };
 
   return {
-    follow: (memberScope, transport$) => {
+    follow: (memberScope, locator$) => {
       let release: (() => void) | undefined;
-      transport$.pipe(memberScope.bind()).subscribe((transport) => {
+      locator$.pipe(memberScope.bind()).subscribe((locator) => {
         release?.();
-        release = transport && register(transport);
+        release = locator && register(locator);
       });
       memberScope.onEnd(() => release?.());
     },
@@ -303,33 +362,12 @@ export function createTransportRegistry(
 }
 
 /**
- * Whether the SFU can take over restarting the delayed leave. Either the
- * homeserver or the transport has to support it. Each endpoint is hit without
- * credentials and read for a 404, not retried: many servers predate the
- * endpoint altogether and answer with a CORS failure that a retry loop would
- * only repeat.
+ * Whether a service can take over restarting the delayed leave. The endpoint
+ * is hit without credentials and read for a 404, not retried: many servers
+ * predate the endpoint altogether and answer with a CORS failure that a retry
+ * loop would only repeat.
  */
 export async function supportsDelegation(
-  homeserverUrl: string,
-  transport: LivekitTransport,
-  logger: Logger,
-): Promise<boolean> {
-  const [homeserver, sfu] = await Promise.all([
-    endpointExists(
-      `${homeserverUrl}/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave`,
-      "homeserver",
-      logger,
-    ),
-    endpointExists(
-      `${transport.livekit_service_url}/delegate_delayed_leave`,
-      `transport ${transport.livekit_service_url}`,
-      logger,
-    ),
-  ]);
-  return homeserver || sfu;
-}
-
-async function endpointExists(
   url: string,
   serviceName: string,
   logger: Logger,