WIP

src/state/CallViewModel/CallViewModel.ts

@@ -56,6 +56,7 @@ import {
   createToggle$,
   filterBehavior,
   generateItems,
+  mapScoped,
   pauseWhen,
 } from "../../utils/observable";
 import {
@@ -104,17 +105,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 +123,9 @@ import {
   ECConnectionFactory,
 } from "./remoteMembers/ConnectionFactory.ts";
 import {
-  type ConnectionManagerData,
+  ConnectionManagerData,
   createConnectionManager$,
+  type IConnectionManager,
 } from "./remoteMembers/ConnectionManager.ts";
 import {
   createRemoteMatrixLivekitMembers$,
@@ -215,7 +214,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. */
@@ -575,13 +574,7 @@ export function createCallViewModel$(
 
   const localTransport$ = options.localTransport
     ? constant(options.localTransport)
-    : from(
-        getLocalTransport({
-          ownMembershipIdentity,
-          client,
-          roomId: matrixRoom.roomId,
-        }),
-      );
+    : from(getLocalTransport(client));
 
   const connectionFactory =
     options.connectionFactory ??
@@ -595,14 +588,24 @@ 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(
+        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[]>
@@ -737,49 +740,27 @@ export function createCallViewModel$(
   const allConnections$ = scope.behavior(
     connectionManager.connectionManagerData$.pipe(map((d) => d.value)),
   );
-  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) =>
+        combineLatest(
+          connections.getConnections().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(
@@ -837,7 +818,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$,

src/state/CallViewModel/localMember/LocalMember.ts

@@ -78,7 +78,6 @@ import {
   type FailedToStartError,
 } from "../remoteMembers/Connection.ts";
 import { type HomeserverConnected } from "./HomeserverConnected.ts";
-import { type LocalTransport } from "./LocalTransport.ts";
 import {
   type ClientDelegationParts,
   delegateDelayedLeave,
@@ -152,7 +151,7 @@ interface Props {
   homeserverConnected: HomeserverConnected;
   roomId: string;
   ownMembershipIdentity: CallMembershipIdentityParts;
-  localTransport$: Observable<LocalTransport>;
+  localTransport$: Observable<UnstableLivekitTransport>;
   client: ClientDelegationParts & ClientOpenIDParts;
   matrixRTCSession: Pick<MatrixRTCSession, "updateCallIntent" | "leave">;
   /** Whether to hide the screen-sharing button. */
@@ -308,7 +307,7 @@ export const createLocalMembership$ = ({
   // to whether delayed event delegation is supported
   const joinParams$ = scope.behavior(
     localTransport$.pipe(
-      switchMap(async ({ transport }) => {
+      switchMap(async (transport) => {
         const transportSupportsDelegation =
           "livekit_service_url" in transport &&
           (await checkDelegationSupport(
@@ -331,7 +330,7 @@ export const createLocalMembership$ = ({
       connectionManager.connectionManagerData$,
       localTransport$,
     ]).pipe(
-      map(([{ value: connectionData }, { transport }]) =>
+      map(([{ value: connectionData }, transport]) =>
         connectionData.getConnectionForTransport(transport),
       ),
       tap((connection) => {

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -8,62 +8,23 @@ Please see LICENSE in the repository root for full details.
 import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import { type MatrixClient } from "matrix-js-sdk";
 import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
-import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
 
 import { Config } from "../../../config/Config.ts";
-import {
-  FailToGetOpenIdToken,
-  MatrixRTCTransportMissingError,
-  NoMatrix2AuthorizationService,
-} from "../../../utils/errors.ts";
+import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
 import { customLivekitUrl } from "../../../settings/settings.ts";
 import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
-import {
-  getSFUConfig,
-  type ClientGetTokenParts,
-  type ClientOpenIDParts,
-  type SFUConfig,
-} from "../../../livekit/auth";
-
-interface Props {
-  ownMembershipIdentity: CallMembershipIdentityParts;
-  client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports"> &
-    ClientGetTokenParts &
-    ClientOpenIDParts;
-  // Used by the jwt service to create the livekit room and compute the livekit alias.
-  roomId: string;
-}
-
-export interface LocalTransport {
-  transport: UnstableLivekitTransport;
-  sfuConfig: SFUConfig;
-}
-
-export function isLocalTransport(
-  obj: UnstableLivekitTransport | 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,
-}: 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,
     resolvedConfig: Config.get(),
-    logger: logger,
+    logger: rootLogger.getChild("[LocalTransport]"),
   });
   const customUrl = customLivekitUrl.value$.value;
 
@@ -75,37 +36,8 @@ export async function getLocalTransport({
   if (transport === null)
     throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
 
-  try {
-    return {
-      transport,
-      sfuConfig: await getSFUConfig({
-        client,
-        membership: ownMembershipIdentity,
-        transport,
-        roomId,
-        role: "publisher",
-        logger,
-      }),
-    };
-  } catch (e) {
-    logger.error(
-      `Failed to authenticate to transport ${JSON.stringify(transport)}`,
-      e,
-    );
-    throw mapAuthErrorToUserFriendlyError(e);
-  }
-}
-
-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;
 }

src/state/CallViewModel/localMember/RtcTransportAutoDiscovery.ts

@@ -45,7 +45,7 @@ export class RtcTransportAutoDiscovery {
     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;
     }

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -45,13 +45,10 @@ import {
 } 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: UnstableLivekitTransport;
   /** The Matrix client to use for OpenID and SFU config requests. */
@@ -105,6 +102,8 @@ export class Connection {
    */
   public readonly state$: Behavior<ConnectionState | Error> = this._state$;
 
+  private readonly role: "publisher" | "subscriber";
+
   /**
    * The media transport to connect to.
    */
@@ -142,7 +141,6 @@ export class Connection {
   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.
    *
@@ -152,18 +150,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 ${JSON.stringify(opts.transport)}]`,
     );
-    this.logger.info(
-      `constructor - 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;
 
@@ -306,11 +302,7 @@ 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 - jwt: ",
         jwt,
@@ -394,15 +386,15 @@ 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 getSFUConfig({
       client: this.client,
       membership: this.ownMembershipIdentity,
+      role: this.role,
       transport: this.transport,
       roomId: this.roomId,
-      role: "subscriber",
       logger: this.logger,
     });
   }

src/state/CallViewModel/remoteMembers/ConnectionFactory.ts

@@ -38,17 +38,16 @@ import {
 import {
   type ClientGetTokenParts,
   type ClientOpenIDParts,
-  type SFUConfig,
 } from "../../../livekit/auth";
 
 // TODO evaluate if this should be done like the Publisher Factory
 export interface ConnectionFactory {
   createConnection(
     scope: ObservableScope,
+    role: "publisher" | "subscriber",
     transport: UnstableLivekitTransport,
     ownMembershipIdentity: CallMembershipIdentityParts,
     logger: Logger,
-    sfuConfig?: SFUConfig,
   ): Connection;
 }
 
@@ -104,15 +103,15 @@ export class ECConnectionFactory implements ConnectionFactory {
    */
   public createConnection(
     scope: ObservableScope,
+    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,

src/state/CallViewModel/remoteMembers/ConnectionManager.ts

@@ -7,31 +7,21 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
-import {
-  combineLatest,
-  map,
-  type Observable,
-  of,
-  switchMap,
-  startWith,
-  catchError,
-  NEVER,
-} from "rxjs";
+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 {
+  Epoch,
+  mapEpoch,
+  type ObservableScope,
+} from "../../ObservableScope.ts";
 import { generateItemsWithEpoch } from "../../../utils/observable.ts";
 import { areUnstableLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
-import {
-  isLocalTransport,
-  type LocalTransport,
-} from "../localMember/LocalTransport.ts";
-import { type SFUConfig } from "../../../livekit/auth";
 
 export class ConnectionManagerData {
   private readonly store: Map<
@@ -92,7 +82,7 @@ export class ConnectionManagerData {
 interface Props {
   scope: ObservableScope;
   connectionFactory: ConnectionFactory;
-  localTransport$: Observable<LocalTransport>;
+  localTransport: UnstableLivekitTransport;
   remoteTransports$: Behavior<Epoch<UnstableLivekitTransport[]>>;
 
   logger: Logger;
@@ -127,7 +117,7 @@ export interface IConnectionManager {
 export function createConnectionManager$({
   scope,
   connectionFactory,
-  localTransport$,
+  localTransport,
   remoteTransports$,
   logger: parentLogger,
   ownMembershipIdentity,
@@ -135,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<(UnstableLivekitTransport | 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) =>
-              areUnstableLivekitTransportsEqual(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.
@@ -188,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

src/utils/observable.ts

@@ -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[]],