add back docstrings

This comma was showing up in the bottom left corner of the UI.

locales/en/app.json

@@ -74,17 +74,18 @@
     "matrix_id": "Matrix ID: {{id}}",
     "matrixRTCMode": {
       "Comptibility": {
-        "label": "Compatibility: state events & multi SFU",
-        "description": "Compatible with homeservers that do not support sticky events (but all other EC clients are v0.17.0 or later)"
+        "description": "Compatible with homeservers that do not support sticky events (but all other EC clients are v0.17.0 or later)",
+        "label": "Compatibility: state events & multi SFU"
       },
       "Legacy": {
-        "label": "Legacy: state events & oldest membership SFU",
-        "description": "Compatible with old versions of EC that do not support multi SFU"
+        "description": "Compatible with old versions of EC that do not support multi SFU",
+        "label": "Legacy: state events & oldest membership SFU"
       },
       "Matrix_2_0": {
-        "label": "Matrix 2.0: sticky events & multi SFU",
-        "description": "Compatible only with homservers supporting sticky events and all EC clients v0.17.0 or later"
-      }
+        "description": "Compatible only with homservers supporting sticky events and all EC clients v0.17.0 or later",
+        "label": "Matrix 2.0: sticky events & multi SFU"
+      },
+      "title": "MatrixRTC mode"
     },
     "mute_all_audio": "Mute all audio (participants, reactions, join sounds)",
     "show_connection_stats": "Show connection statistics",

src/button/ReactionToggleButton.test.tsx

@@ -13,7 +13,7 @@ import { type ReactNode } from "react";
 
 import { ReactionToggleButton } from "./ReactionToggleButton";
 import { ElementCallReactionEventType } from "../reactions";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { getBasicCallViewModelEnvironment } from "../utils/test-viewmodel";
 import { alice, local, localRtcMember } from "../utils/test-fixtures";
 import { type MockRTCSession } from "../utils/test";

src/button/ReactionToggleButton.tsx

@@ -33,7 +33,7 @@ import {
   ReactionsRowSize,
 } from "../reactions";
 import { Modal } from "../Modal";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { useBehavior } from "../useBehavior";
 
 interface InnerButtonProps extends ComponentPropsWithoutRef<"button"> {

src/livekit/TrackProcessorContext.tsx

@@ -28,6 +28,7 @@ import {
 } from "../settings/settings";
 import { BlurBackgroundTransformer } from "./BlurBackgroundTransformer";
 import { type Behavior } from "../state/Behavior";
+import { type ObservableScope } from "../state/ObservableScope";
 
 //TODO-MULTI-SFU: This is not yet fully there.
 // it is a combination of exposing observable and react hooks.
@@ -63,13 +64,17 @@ export function useTrackProcessorObservable$(): Observable<ProcessorState> {
   return state$;
 }
 
+/**
+ * Updates your video tracks to always use the given processor.
+ */
 export const trackProcessorSync = (
+  scope: ObservableScope,
   videoTrack$: Behavior<LocalVideoTrack | null>,
   processor$: Behavior<ProcessorState>,
 ): void => {
-  // TODO-MULTI-SFU: Bind to an ObservableScope to avoid leaking resources.
-  combineLatest([videoTrack$, processor$]).subscribe(
-    ([videoTrack, processorState]) => {
+  combineLatest([videoTrack$, processor$])
+    .pipe(scope.bind())
+    .subscribe(([videoTrack, processorState]) => {
       if (!processorState) return;
       if (!videoTrack) return;
       const { processor } = processorState;
@@ -79,8 +84,7 @@ export const trackProcessorSync = (
       if (!processor && videoTrack.getProcessor()) {
         void videoTrack.stopProcessor();
       }
-    },
-  );
+    });
 };
 
 export const useTrackProcessorSync = (

src/main.tsx

@@ -62,7 +62,7 @@ Initializer.initBeforeReact()
   .then(() => {
     root.render(
       <StrictMode>
-        <App vm={new AppViewModel(globalScope)} />,
+        <App vm={new AppViewModel(globalScope)} />
       </StrictMode>,
     );
   })

src/reactions/ReactionsReader.ts

@@ -266,6 +266,7 @@ export class ReactionsReader {
         );
         return;
       }
+      // TODO refactor to use memer id `membershipEvent.membershipID` (needs to happen in combination with other memberId refactors)
       const identifier = `${membershipEvent.userId}:${membershipEvent.deviceId}`;
 
       if (!content.emoji) {

src/reactions/useReactionsSender.tsx

@@ -20,7 +20,7 @@ import { logger } from "matrix-js-sdk/lib/logger";
 import { useMatrixRTCSessionMemberships } from "../useMatrixRTCSessionMemberships";
 import { useClientState } from "../ClientContext";
 import { ElementCallReactionEventType, type ReactionOption } from ".";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { useBehavior } from "../useBehavior";
 
 interface ReactionsSenderContextType {

src/room/CallEventAudioRenderer.test.tsx

@@ -38,7 +38,7 @@ import {
   local,
   localRtcMember,
 } from "../utils/test-fixtures";
-import { MAX_PARTICIPANT_COUNT_FOR_SOUND } from "../state/CallViewModel";
+import { MAX_PARTICIPANT_COUNT_FOR_SOUND } from "../state/CallViewModel/CallViewModel";
 
 vitest.mock("livekit-client/e2ee-worker?worker");
 vitest.mock("../useAudioContext");

src/room/CallEventAudioRenderer.tsx

@@ -7,7 +7,7 @@ Please see LICENSE in the repository root for full details.
 
 import { type ReactNode, useEffect } from "react";
 
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import joinCallSoundMp3 from "../sound/join_call.mp3";
 import joinCallSoundOgg from "../sound/join_call.ogg";
 import leftCallSoundMp3 from "../sound/left_call.mp3";

src/room/GroupCallView.test.tsx

@@ -77,13 +77,13 @@ const leaveRTCSession = vi.hoisted(() =>
   ),
 );
 
-vi.mock("../rtcSessionHelpers", async (importOriginal) => {
-  // TODO: perhaps there is a more elegant way to manage the type import here?
-  // eslint-disable-next-line @typescript-eslint/consistent-type-imports
-  const orig = await importOriginal<typeof import("../rtcSessionHelpers")>();
-  // TODO: leaveRTCSession no longer exists! Tests need adapting.
-  return { ...orig, enterRTCSession, leaveRTCSession };
-});
+// vi.mock("../rtcSessionHelpers", async (importOriginal) => {
+//   // TODO: perhaps there is a more elegant way to manage the type import here?
+//   // eslint-disable-next-line @typescript-eslint/consistent-type-imports
+//   const orig = await importOriginal<typeof import("../rtcSessionHelpers")>();
+//   // TODO: leaveRTCSession no longer exists! Tests need adapting.
+//   return { ...orig, enterRTCSession, leaveRTCSession };
+// });
 
 let playSound: MockedFunction<
   NonNullable<ReturnType<typeof useAudioContext>>["playSound"]
@@ -266,6 +266,7 @@ test.skip("GroupCallView leaves the session when an error occurs", async () => {
 
 test.skip("GroupCallView shows errors that occur during joining", async () => {
   const user = userEvent.setup();
+  // This should not mock this error that deep. it should only mock the CallViewModel.
   enterRTCSession.mockRejectedValue(new MatrixRTCTransportMissingError(""));
   onTestFinished(() => {
     enterRTCSession.mockReset();

src/room/InCallView.test.tsx

@@ -43,9 +43,6 @@ import { LivekitRoomAudioRenderer } from "../livekit/MatrixAudioRenderer";
 import { MediaDevicesContext } from "../MediaDevicesContext";
 import { HeaderStyle } from "../UrlParams";
 
-// vi.hoisted(() => {
-//   localStorage = {} as unknown as Storage;
-// });
 vi.hoisted(
   () =>
     (global.ImageData = class MockImageData {
@@ -109,6 +106,7 @@ function createInCallView(): RenderResult & {
     getUserId: () => localRtcMember.userId,
     getDeviceId: () => localRtcMember.deviceId,
     getRoom: (rId) => (rId === roomId ? room : null),
+    getDomain: () => "example.com",
   } as Partial<MatrixClient> as MatrixClient;
   const room = mockMatrixRoom({
     relations: {
@@ -119,7 +117,8 @@ function createInCallView(): RenderResult & {
     } as unknown as RelationsContainer,
     client,
     roomId,
-    getMember: (userId) => roomMembers.get(userId) ?? null,
+    // getMember: (userId) => roomMembers.get(userId) ?? null,
+    getMembers: () => Array.from(roomMembers.values()),
     getMxcAvatarUrl: () => null,
     hasEncryptionStateEvent: vi.fn().mockReturnValue(true),
     getCanonicalAlias: () => null,

src/room/InCallView.tsx

@@ -58,7 +58,10 @@ import { type MuteStates } from "../state/MuteStates";
 import { type MatrixInfo } from "./VideoPreview";
 import { InviteButton } from "../button/InviteButton";
 import { LayoutToggle } from "./LayoutToggle";
-import { CallViewModel, type GridMode } from "../state/CallViewModel";
+import {
+  CallViewModel,
+  type GridMode,
+} from "../state/CallViewModel/CallViewModel.ts";
 import { Grid, type TileProps } from "../grid/Grid";
 import { useInitial } from "../useInitial";
 import { SpotlightTile } from "../tile/SpotlightTile";
@@ -115,16 +118,16 @@ export interface ActiveCallProps
 }
 
 export const ActiveCall: FC<ActiveCallProps> = (props) => {
-  const mediaDevices = useMediaDevices();
   const [vm, setVm] = useState<CallViewModel | null>(null);
 
-  const { autoLeaveWhenOthersLeft, waitForCallPickup, sendNotificationType } =
-    useUrlParams();
-
+  const urlParams = useUrlParams();
+  const mediaDevices = useMediaDevices();
   const trackProcessorState$ = useTrackProcessorObservable$();
   useEffect(() => {
     const scope = new ObservableScope();
     const reactionsReader = new ReactionsReader(scope, props.rtcSession);
+    const { autoLeaveWhenOthersLeft, waitForCallPickup, sendNotificationType } =
+      urlParams;
     const vm = new CallViewModel(
       scope,
       props.rtcSession,
@@ -149,13 +152,11 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
   }, [
     props.rtcSession,
     props.matrixRoom,
-    mediaDevices,
     props.muteStates,
     props.e2eeSystem,
-    autoLeaveWhenOthersLeft,
-    sendNotificationType,
-    waitForCallPickup,
     props.onLeft,
+    urlParams,
+    mediaDevices,
     trackProcessorState$,
   ]);
 
@@ -819,7 +820,7 @@ export const InCallView: FC<InCallViewProps> = ({
           key={url}
           url={url}
           livekitRoom={livekitRoom}
-          validIdentities={participants.map((p) => p.identity)}
+          validIdentities={participants}
           muted={muteAllAudio}
         />
       ))}

src/room/ReactionAudioRenderer.test.tsx

@@ -10,8 +10,9 @@ import {
   afterAll,
   afterEach,
   beforeEach,
+  describe,
   expect,
-  test,
+  it,
   vitest,
   type MockedFunction,
   type Mock,
@@ -27,7 +28,7 @@ import {
 import { useAudioContext } from "../useAudioContext";
 import { GenericReaction, ReactionSet } from "../reactions";
 import { prefetchSounds } from "../soundUtils";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { getBasicCallViewModelEnvironment } from "../utils/test-viewmodel";
 import {
   alice,
@@ -49,122 +50,125 @@ vitest.mock("livekit-client/e2ee-worker?worker");
 vitest.mock("../useAudioContext");
 vitest.mock("../soundUtils");
 
-afterEach(() => {
-  vitest.resetAllMocks();
-  playReactionsSoundSetting.setValue(playReactionsSoundSetting.defaultValue);
-  soundEffectVolumeSetting.setValue(soundEffectVolumeSetting.defaultValue);
-});
-
-afterAll(() => {
-  vitest.restoreAllMocks();
-});
-
 let playSound: Mock<
   NonNullable<ReturnType<typeof useAudioContext>>["playSound"]
 >;
 
-beforeEach(() => {
-  (prefetchSounds as MockedFunction<typeof prefetchSounds>).mockResolvedValue({
-    sound: new ArrayBuffer(0),
+describe("ReactionAudioRenderer", () => {
+  afterEach(() => {
+    playReactionsSoundSetting.setValue(playReactionsSoundSetting.defaultValue);
+    soundEffectVolumeSetting.setValue(soundEffectVolumeSetting.defaultValue);
   });
-  playSound = vitest.fn();
-  (useAudioContext as MockedFunction<typeof useAudioContext>).mockReturnValue({
-    playSound,
-    playSoundLooping: vitest.fn(),
-    soundDuration: {},
-  });
-});
-
-test("preloads all audio elements", () => {
-  const { vm } = getBasicCallViewModelEnvironment([local, alice]);
-  playReactionsSoundSetting.setValue(true);
-  render(<TestComponent vm={vm} />);
-  expect(prefetchSounds).toHaveBeenCalledOnce();
-});
-
-test("will play an audio sound when there is a reaction", () => {
-  const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
-    local,
-    alice,
-  ]);
-  playReactionsSoundSetting.setValue(true);
-  render(<TestComponent vm={vm} />);
-
-  // Find the first reaction with a sound effect
-  const chosenReaction = ReactionSet.find((r) => !!r.sound);
-  if (!chosenReaction) {
-    throw Error(
-      "No reactions have sounds configured, this test cannot succeed",
+  beforeEach(() => {
+    (prefetchSounds as MockedFunction<typeof prefetchSounds>).mockResolvedValue(
+      {
+        sound: new ArrayBuffer(0),
+      },
     );
-  }
-  act(() => {
-    reactionsSubject$.next({
-      [aliceRtcMember.deviceId]: {
-        reactionOption: chosenReaction,
-        expireAfter: new Date(0),
+    playSound = vitest.fn();
+    (useAudioContext as MockedFunction<typeof useAudioContext>).mockReturnValue(
+      {
+        playSound,
+        playSoundLooping: vitest.fn(),
+        soundDuration: {},
       },
-    });
-  });
-  expect(playSound).toHaveBeenCalledWith(chosenReaction.name);
-});
-
-test("will play the generic audio sound when there is soundless reaction", () => {
-  const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
-    local,
-    alice,
-  ]);
-  playReactionsSoundSetting.setValue(true);
-  render(<TestComponent vm={vm} />);
-
-  // Find the first reaction with a sound effect
-  const chosenReaction = ReactionSet.find((r) => !r.sound);
-  if (!chosenReaction) {
-    throw Error(
-      "No reactions have sounds configured, this test cannot succeed",
     );
-  }
-  act(() => {
-    reactionsSubject$.next({
-      [aliceRtcMember.deviceId]: {
-        reactionOption: chosenReaction,
-        expireAfter: new Date(0),
-      },
-    });
   });
-  expect(playSound).toHaveBeenCalledWith(GenericReaction.name);
-});
-
-test("will play multiple audio sounds when there are multiple different reactions", () => {
-  const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
-    local,
-    alice,
-  ]);
-  playReactionsSoundSetting.setValue(true);
-  render(<TestComponent vm={vm} />);
-
-  // Find the first reaction with a sound effect
-  const [reaction1, reaction2] = ReactionSet.filter((r) => !!r.sound);
-  if (!reaction1 || !reaction2) {
-    throw Error(
-      "No reactions have sounds configured, this test cannot succeed",
-    );
-  }
-  act(() => {
-    reactionsSubject$.next({
-      [aliceRtcMember.deviceId]: {
-        reactionOption: reaction1,
-        expireAfter: new Date(0),
-      },
-      [bobRtcMember.deviceId]: {
-        reactionOption: reaction2,
-        expireAfter: new Date(0),
-      },
-      [localRtcMember.deviceId]: {
-        reactionOption: reaction1,
-        expireAfter: new Date(0),
-      },
-    });
+  afterAll(() => {
+    vitest.restoreAllMocks();
+  });
+
+  it("preloads all audio elements", () => {
+    const { vm } = getBasicCallViewModelEnvironment([local, alice]);
+    playReactionsSoundSetting.setValue(true);
+    render(<TestComponent vm={vm} />);
+    expect(prefetchSounds).toHaveBeenCalledOnce();
+  });
+
+  it("will play an audio sound when there is a reaction", () => {
+    const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
+      local,
+      alice,
+    ]);
+    playReactionsSoundSetting.setValue(true);
+    render(<TestComponent vm={vm} />);
+
+    // Find the first reaction with a sound effect
+    const chosenReaction = ReactionSet.find((r) => !!r.sound);
+    if (!chosenReaction) {
+      throw Error(
+        "No reactions have sounds configured, this test cannot succeed",
+      );
+    }
+    act(() => {
+      reactionsSubject$.next({
+        [aliceRtcMember.deviceId]: {
+          reactionOption: chosenReaction,
+          expireAfter: new Date(0),
+        },
+      });
+    });
+    expect(playSound).toHaveBeenCalledWith(chosenReaction.name);
+  });
+
+  it("will play the generic audio sound when there is soundless reaction", () => {
+    const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
+      local,
+      alice,
+    ]);
+    playReactionsSoundSetting.setValue(true);
+    render(<TestComponent vm={vm} />);
+
+    // Find the first reaction with a sound effect
+    const chosenReaction = ReactionSet.find((r) => !r.sound);
+    if (!chosenReaction) {
+      throw Error(
+        "No reactions have sounds configured, this test cannot succeed",
+      );
+    }
+    act(() => {
+      reactionsSubject$.next({
+        [aliceRtcMember.deviceId]: {
+          reactionOption: chosenReaction,
+          expireAfter: new Date(0),
+        },
+      });
+    });
+    expect(playSound).toHaveBeenCalledWith(GenericReaction.name);
+  });
+
+  it("will play multiple audio sounds when there are multiple different reactions", () => {
+    const { vm, reactionsSubject$ } = getBasicCallViewModelEnvironment([
+      local,
+      alice,
+    ]);
+    playReactionsSoundSetting.setValue(true);
+    render(<TestComponent vm={vm} />);
+
+    // Find the first reaction with a sound effect
+    const [reaction1, reaction2] = ReactionSet.filter((r) => !!r.sound);
+    if (!reaction1 || !reaction2) {
+      throw Error(
+        "No reactions have sounds configured, this test cannot succeed",
+      );
+    }
+    act(() => {
+      reactionsSubject$.next({
+        [aliceRtcMember.deviceId]: {
+          reactionOption: reaction1,
+          expireAfter: new Date(0),
+        },
+        [bobRtcMember.deviceId]: {
+          reactionOption: reaction2,
+          expireAfter: new Date(0),
+        },
+        [localRtcMember.deviceId]: {
+          reactionOption: reaction1,
+          expireAfter: new Date(0),
+        },
+      });
+    });
+    expect(playSound).toHaveBeenCalledWith(reaction1.name);
+    expect(playSound).toHaveBeenCalledWith(reaction2.name);
   });
-  expect(playSound).toHaveBeenCalledWith(reaction1.name);
-  expect(playSound).toHaveBeenCalledWith(reaction2.name);
 });

src/room/ReactionAudioRenderer.tsx

@@ -12,7 +12,7 @@ import { GenericReaction, ReactionSet } from "../reactions";
 import { useAudioContext } from "../useAudioContext";
 import { prefetchSounds } from "../soundUtils";
 import { useLatest } from "../useLatest";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 
 const soundMap = Object.fromEntries([
   ...ReactionSet.filter((v) => v.sound !== undefined).map((v) => [

src/room/ReactionsOverlay.tsx

@@ -8,7 +8,7 @@ Please see LICENSE in the repository root for full details.
 import { type ReactNode } from "react";
 
 import styles from "./ReactionsOverlay.module.css";
-import { type CallViewModel } from "../state/CallViewModel";
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { useBehavior } from "../useBehavior";
 
 export function ReactionsOverlay({ vm }: { vm: CallViewModel }): ReactNode {

src/room/__snapshots__/InCallView.test.tsx.snap

@@ -83,9 +83,6 @@ exports[`InCallView > rendering > renders 1`] = `
         class="nav rightNav"
       />
     </header>
-    <div>
-      mocked: MatrixAudioRenderer
-    </div>
     <div
       class="scrollingGrid grid"
     >

src/rtcSessionHelpers.ts

@@ -1,160 +0,0 @@
-/*
-Copyright 2023, 2024 New Vector Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import {
-  type MatrixRTCSession,
-  isLivekitTransportConfig,
-  type LivekitTransportConfig,
-  type LivekitTransport,
-} from "matrix-js-sdk/lib/matrixrtc";
-import { logger } from "matrix-js-sdk/lib/logger";
-import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
-import { type MatrixClient } from "matrix-js-sdk";
-
-import { PosthogAnalytics } from "./analytics/PosthogAnalytics";
-import { Config } from "./config/Config";
-import { ElementWidgetActions, widget } from "./widget";
-import { MatrixRTCTransportMissingError } from "./utils/errors";
-import { getUrlParams } from "./UrlParams";
-import { getSFUConfigWithOpenID } from "./livekit/openIDSFU.ts";
-import { MatrixRTCMode } from "./settings/settings.ts";
-
-const FOCI_WK_KEY = "org.matrix.msc4143.rtc_foci";
-
-async function makeTransportInternal(
-  client: MatrixClient,
-  roomId: string,
-): Promise<LivekitTransport> {
-  logger.log("Searching for a preferred transport");
-  //TODO refactor this to use the jwt service returned alias.
-  const livekitAlias = roomId;
-
-  // TODO-MULTI-SFU: Either remove this dev tool or make it more official
-  const urlFromStorage =
-    localStorage.getItem("robin-matrixrtc-auth") ??
-    localStorage.getItem("timo-focus-url");
-  if (urlFromStorage !== null) {
-    const transportFromStorage: LivekitTransport = {
-      type: "livekit",
-      livekit_service_url: urlFromStorage,
-      livekit_alias: livekitAlias,
-    };
-    logger.log(
-      "Using LiveKit transport from local storage: ",
-      transportFromStorage,
-    );
-    return transportFromStorage;
-  }
-
-  // Prioritize the .well-known/matrix/client, if available, over the configured SFU
-  const domain = client.getDomain();
-  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 transport: LivekitTransportConfig | undefined = wellKnownFoci.find(
-        (f) => f && isLivekitTransportConfig(f),
-      );
-      if (transport !== undefined) {
-        logger.log("Using LiveKit transport from .well-known: ", transport);
-        return { ...transport, livekit_alias: livekitAlias };
-      }
-    }
-  }
-
-  const urlFromConf = Config.get().livekit?.livekit_service_url;
-  if (urlFromConf) {
-    const transportFromConf: LivekitTransport = {
-      type: "livekit",
-      livekit_service_url: urlFromConf,
-      livekit_alias: livekitAlias,
-    };
-    logger.log("Using LiveKit transport from config: ", transportFromConf);
-    return transportFromConf;
-  }
-
-  throw new MatrixRTCTransportMissingError(domain ?? "");
-}
-
-export async function makeTransport(
-  client: MatrixClient,
-  roomId: string,
-): Promise<LivekitTransport> {
-  const transport = await makeTransportInternal(client, roomId);
-  // 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;
-}
-
-export interface EnterRTCSessionOptions {
-  encryptMedia: boolean;
-  matrixRTCMode: MatrixRTCMode;
-}
-
-/**
- * TODO! document this function properly
- * @param rtcSession
- * @param transport
- * @param options
- */
-export async function enterRTCSession(
-  rtcSession: MatrixRTCSession,
-  transport: LivekitTransport,
-  { encryptMedia, matrixRTCMode }: EnterRTCSessionOptions,
-): Promise<void> {
-  PosthogAnalytics.instance.eventCallEnded.cacheStartCall(new Date());
-  PosthogAnalytics.instance.eventCallStarted.track(rtcSession.room.roomId);
-
-  // This must be called before we start trying to join the call, as we need to
-  // have started tracking by the time calls start getting created.
-  // groupCallOTelMembership?.onJoinCall();
-
-  const { features, matrix_rtc_session: matrixRtcSessionConfig } = Config.get();
-  const useDeviceSessionMemberEvents =
-    features?.feature_use_device_session_member_events;
-  const { sendNotificationType: notificationType, callIntent } = getUrlParams();
-  const multiSFU = matrixRTCMode !== MatrixRTCMode.Legacy;
-  // Multi-sfu does not need a preferred foci list. just the focus that is actually used.
-  rtcSession.joinRoomSession(
-    multiSFU ? [] : [transport],
-    multiSFU ? transport : undefined,
-    {
-      notificationType,
-      callIntent,
-      manageMediaKeys: encryptMedia,
-      ...(useDeviceSessionMemberEvents !== undefined && {
-        useLegacyMemberEvents: !useDeviceSessionMemberEvents,
-      }),
-      delayedLeaveEventRestartMs:
-        matrixRtcSessionConfig?.delayed_leave_event_restart_ms,
-      delayedLeaveEventDelayMs:
-        matrixRtcSessionConfig?.delayed_leave_event_delay_ms,
-      delayedLeaveEventRestartLocalTimeoutMs:
-        matrixRtcSessionConfig?.delayed_leave_event_restart_local_timeout_ms,
-      networkErrorRetryMs: matrixRtcSessionConfig?.network_error_retry_ms,
-      makeKeyDelay: matrixRtcSessionConfig?.wait_for_key_rotation_ms,
-      membershipEventExpiryMs:
-        matrixRtcSessionConfig?.membership_event_expiry_ms,
-      useExperimentalToDeviceTransport: true,
-      unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
-    },
-  );
-  if (widget) {
-    try {
-      await widget.api.transport.send(ElementWidgetActions.JoinCall, {});
-    } catch (e) {
-      logger.error("Failed to send join action", e);
-    }
-  }
-}

src/settings/DeveloperSettingsTab.tsx

@@ -12,6 +12,7 @@ import {
   useEffect,
   useMemo,
   useState,
+  useId,
 } from "react";
 import { useTranslation } from "react-i18next";
 import {
@@ -19,6 +20,14 @@ import {
   type MatrixClient,
 } from "matrix-js-sdk";
 import { logger } from "matrix-js-sdk/lib/logger";
+import {
+  Root as Form,
+  Heading,
+  HelpMessage,
+  InlineField,
+  Label,
+  RadioControl,
+} from "@vector-im/compound-web";
 
 import { FieldRow, InputField } from "../input/Input";
 import {
@@ -26,7 +35,6 @@ import {
   duplicateTiles as duplicateTilesSetting,
   debugTileLayout as debugTileLayoutSetting,
   showConnectionStats as showConnectionStatsSetting,
-  multiSfu as multiSfuSetting,
   muteAllAudio as muteAllAudioSetting,
   alwaysShowIphoneEarpiece as alwaysShowIphoneEarpieceSetting,
   matrixRTCMode as matrixRTCModeSetting,
@@ -61,6 +69,13 @@ export const DeveloperSettingsTab: FC<Props> = ({ client, livekitRooms }) => {
   }, [client]);
 
   const [matrixRTCMode, setMatrixRTCMode] = useSetting(matrixRTCModeSetting);
+  const matrixRTCModeRadioGroup = useId();
+  const onMatrixRTCModeChange = useCallback(
+    (e: ChangeEvent<HTMLInputElement>) => {
+      setMatrixRTCMode(e.target.value as MatrixRTCMode);
+    },
+    [setMatrixRTCMode],
+  );
 
   const [showConnectionStats, setShowConnectionStats] = useSetting(
     showConnectionStatsSetting,
@@ -86,7 +101,7 @@ export const DeveloperSettingsTab: FC<Props> = ({ client, livekitRooms }) => {
   }, [livekitRooms]);
 
   return (
-    <>
+    <Form>
       <p>
         {t("developer_mode.hostname", {
           hostname: window.location.hostname || "unknown",
@@ -143,52 +158,6 @@ export const DeveloperSettingsTab: FC<Props> = ({ client, livekitRooms }) => {
           }
         />
       </FieldRow>
-      <FieldRow>
-        <InputField
-          id="matrixRTCMode"
-          type="checkbox"
-          label={t("developer_mode.matrixRTCMode.Legacy.label")}
-          description={t("developer_mode.matrixRTCMode.Legacy.description")}
-          checked={matrixRTCMode === MatrixRTCMode.Legacy}
-          onChange={useCallback(
-            (event: ChangeEvent<HTMLInputElement>) => {
-              if (event.target.checked) setMatrixRTCMode(MatrixRTCMode.Legacy);
-            },
-            [setMatrixRTCMode],
-          )}
-        />
-        <InputField
-          id="matrixRTCMode"
-          type="checkbox"
-          label={t("developer_mode.matrixRTCMode.Comptibility.label")}
-          description={t(
-            "developer_mode.matrixRTCMode.Comptibility.description",
-          )}
-          checked={matrixRTCMode === MatrixRTCMode.Compatibil}
-          onChange={useCallback(
-            (event: ChangeEvent<HTMLInputElement>) => {
-              if (event.target.checked)
-                setMatrixRTCMode(MatrixRTCMode.Compatibil);
-            },
-            [setMatrixRTCMode],
-          )}
-        />
-        <InputField
-          id="matrixRTCMode"
-          type="checkbox"
-          label={t("developer_mode.matrixRTCMode.Matrix_2_0.label")}
-          description={t("developer_mode.matrixRTCMode.Matrix_2_0.description")}
-          disabled={!stickyEventsSupported}
-          checked={matrixRTCMode === MatrixRTCMode.Matrix_2_0}
-          onChange={useCallback(
-            (event: ChangeEvent<HTMLInputElement>) => {
-              if (event.target.checked)
-                setMatrixRTCMode(MatrixRTCMode.Matrix_2_0);
-            },
-            [setMatrixRTCMode],
-          )}
-        />
-      </FieldRow>
       <FieldRow>
         <InputField
           id="showConnectionStats"
@@ -231,6 +200,55 @@ export const DeveloperSettingsTab: FC<Props> = ({ client, livekitRooms }) => {
           )}
         />{" "}
       </FieldRow>
+      <Heading as="h3" type="body" weight="semibold" size="lg">
+        {t("developer_mode.matrixRTCMode.title")}
+      </Heading>
+      <InlineField
+        name={matrixRTCModeRadioGroup}
+        control={
+          <RadioControl
+            checked={matrixRTCMode === MatrixRTCMode.Legacy}
+            value={MatrixRTCMode.Legacy}
+            onChange={onMatrixRTCModeChange}
+          />
+        }
+      >
+        <Label>{t("developer_mode.matrixRTCMode.Legacy.label")}</Label>
+        <HelpMessage>
+          {t("developer_mode.matrixRTCMode.Legacy.description")}
+        </HelpMessage>
+      </InlineField>
+      <InlineField
+        name={matrixRTCModeRadioGroup}
+        control={
+          <RadioControl
+            checked={matrixRTCMode === MatrixRTCMode.Compatibil}
+            value={MatrixRTCMode.Compatibil}
+            onChange={onMatrixRTCModeChange}
+          />
+        }
+      >
+        <Label>{t("developer_mode.matrixRTCMode.Comptibility.label")}</Label>
+        <HelpMessage>
+          {t("developer_mode.matrixRTCMode.Comptibility.description")}
+        </HelpMessage>
+      </InlineField>
+      <InlineField
+        name={matrixRTCModeRadioGroup}
+        control={
+          <RadioControl
+            checked={matrixRTCMode === MatrixRTCMode.Matrix_2_0}
+            value={MatrixRTCMode.Matrix_2_0}
+            disabled={!stickyEventsSupported}
+            onChange={onMatrixRTCModeChange}
+          />
+        }
+      >
+        <Label>{t("developer_mode.matrixRTCMode.Matrix_2_0.label")}</Label>
+        <HelpMessage>
+          {t("developer_mode.matrixRTCMode.Matrix_2_0.description")}
+        </HelpMessage>
+      </InlineField>
       {livekitRooms?.map((livekitRoom) => (
         <>
           <h3>
@@ -255,6 +273,6 @@ export const DeveloperSettingsTab: FC<Props> = ({ client, livekitRooms }) => {
       <pre>{JSON.stringify(import.meta.env, null, 2)}</pre>
       <p>{t("developer_mode.url_params")}</p>
       <pre>{JSON.stringify(urlParams, null, 2)}</pre>
-    </>
+    </Form>
   );
 };

src/state/Async.ts

@@ -1,59 +0,0 @@
-/*
-Copyright 2025 New Vector Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import { catchError, from, map, type Observable, of, startWith } from "rxjs";
-
-/**
- * Data that may need to be loaded asynchronously.
- *
- * This type is for when you need to represent the current state of an operation
- * involving Promises as **immutable data**. See the async$ function below.
- */
-export type Async<A> =
-  | { state: "loading" }
-  | { state: "error"; value: Error }
-  | { state: "ready"; value: A };
-
-export const loading: Async<never> = { state: "loading" };
-export function error(value: Error): Async<never> {
-  return { state: "error", value };
-}
-
-export function ready<A>(value: A): Async<A> {
-  return { state: "ready", value };
-}
-
-/**
- * Turn a Promise into an Observable async value. The Observable will have the
- * value "loading" while the Promise is pending, "ready" when the Promise
- * resolves, and "error" when the Promise rejects.
- */
-export function async$<A>(promise: Promise<A>): Observable<Async<A>> {
-  return from(promise).pipe(
-    map(ready),
-    startWith(loading),
-    catchError((e: unknown) =>
-      of(error((e as Error) ?? new Error("Unknown error"))),
-    ),
-  );
-}
-
-/**
- * If the async value is ready, apply the given function to the inner value.
- */
-export function mapAsync<A, B>(
-  async: Async<A>,
-  project: (value: A) => B,
-): Async<B> {
-  return async.state === "ready" ? ready(project(async.value)) : async;
-}
-
-export function unwrapAsync<A>(fallback: A): (async: Async<A>) => A {
-  return (async: Async<A>) => {
-    return async.state === "ready" ? async.value : fallback;
-  };
-}

src/state/CallViewModel/CallNotificationLifecycle.test.ts

@@ -0,0 +1,354 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import {
+  type ICallNotifyContent,
+  type IRTCNotificationContent,
+} from "matrix-js-sdk/lib/matrixrtc";
+import { describe, it } from "vitest";
+import {
+  EventType,
+  type IEvent,
+  type IRoomTimelineData,
+  MatrixEvent,
+  type Room,
+} from "matrix-js-sdk";
+
+import { withTestScheduler } from "../../utils/test";
+import {
+  aliceRtcMember,
+  local,
+  localRtcMember,
+} from "../../utils/test-fixtures";
+import {
+  createCallNotificationLifecycle$,
+  type Props as CallNotificationLifecycleProps,
+} from "./CallNotificationLifecycle";
+import { trackEpoch } from "../ObservableScope";
+
+const mockLegacyRingEvent = {} as { event_id: string } & ICallNotifyContent;
+function mockRingEvent(
+  eventId: string,
+  lifetimeMs: number | undefined,
+  sender = local.userId,
+): { event_id: string } & IRTCNotificationContent {
+  return {
+    event_id: eventId,
+    ...(lifetimeMs === undefined ? {} : { lifetime: lifetimeMs }),
+    notification_type: "ring",
+    sender,
+  } as unknown as { event_id: string } & IRTCNotificationContent;
+}
+
+describe("waitForCallPickup$", () => {
+  it("unknown -> ringing -> timeout when notified and nobody joins", () => {
+    withTestScheduler(({ scope, expectObservable, behavior, hot }) => {
+      // No one ever joins (only local user)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a", { a: [] }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$notif1", 30), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot(""),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+
+      const lifecycle = createCallNotificationLifecycle$(props);
+
+      expectObservable(lifecycle.callPickupState$).toBe("a 9ms b 29ms c", {
+        a: "unknown",
+        b: "ringing",
+        c: "timeout",
+      });
+    });
+  });
+
+  it("ringing -> success if someone joins before timeout is reached", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a 19ms b", {
+            a: [localRtcMember],
+            b: [localRtcMember, aliceRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("5ms a", {
+          a: [mockRingEvent("$notif2", 100), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot(""),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      expectObservable(lifecycle.callPickupState$).toBe("a 4ms b 14ms c", {
+        a: "unknown",
+        b: "ringing",
+        c: "success",
+      });
+    });
+  });
+  it("success when someone joins before we notify", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a 9ms b", {
+            a: [localRtcMember],
+            b: [localRtcMember, aliceRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("20ms a", {
+          a: [mockRingEvent("$notif2", 50), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot(""),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      expectObservable(lifecycle.callPickupState$).toBe("a 9ms b", {
+        a: "unknown",
+        b: "success",
+      });
+    });
+  });
+  it("notify without lifetime -> immediate timeout", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a", {
+            a: [localRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$notif2", undefined), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot(""),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      expectObservable(lifecycle.callPickupState$).toBe("a 9ms b", {
+        a: "unknown",
+        b: "timeout",
+      });
+    });
+  });
+
+  it("stays null when waitForCallPickup=false", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const validProps: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a--b", {
+            a: [localRtcMember],
+            b: [localRtcMember, aliceRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$notif5", 30), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot(""),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const propsDeactivated = {
+        ...validProps,
+        options: {
+          ...validProps.options,
+          waitForCallPickup: false,
+        },
+      };
+      const lifecycle = createCallNotificationLifecycle$(propsDeactivated);
+      expectObservable(lifecycle.callPickupState$).toBe("n", {
+        n: null,
+      });
+      const lifecycleReference = createCallNotificationLifecycle$(validProps);
+      expectObservable(lifecycleReference.callPickupState$).toBe("u--s", {
+        u: "unknown",
+        s: "success",
+      });
+    });
+  });
+
+  it("decline before timeout window ends -> decline", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a", {
+            a: [localRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$decl1", 50), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot("40ms d", {
+          d: [
+            new MatrixEvent({
+              type: EventType.RTCDecline,
+              content: {
+                "m.relates_to": {
+                  rel_type: "m.reference",
+                  event_id: "$decl1",
+                },
+              },
+            }),
+            {} as Room,
+            undefined,
+            false,
+            {} as IRoomTimelineData,
+          ],
+        }),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      expectObservable(lifecycle.callPickupState$).toBe("a 9ms b 29ms e", {
+        a: "unknown",
+        b: "ringing",
+        e: "decline",
+      });
+    });
+  });
+  it("decline after timeout window ends -> stays timeout", () => {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a", {
+            a: [localRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$decl", 20), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot("40ms d", {
+          d: [
+            new MatrixEvent({
+              type: EventType.RTCDecline,
+              content: {
+                "m.relates_to": {
+                  rel_type: "m.reference",
+                  event_id: "$decl",
+                },
+              },
+            }),
+            {} as Room,
+            undefined,
+            false,
+            {} as IRoomTimelineData,
+          ],
+        }),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      expectObservable(lifecycle.callPickupState$, "50ms !").toBe(
+        "a 9ms b 19ms e",
+        {
+          a: "unknown",
+          b: "ringing",
+          e: "timeout",
+        },
+      );
+    });
+  });
+  //
+  function testStaysRinging(
+    declineEvent: Partial<IEvent>,
+    expectDecline: boolean,
+  ): void {
+    withTestScheduler(({ scope, hot, behavior, expectObservable }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const props: CallNotificationLifecycleProps = {
+        scope,
+        memberships$: scope.behavior(
+          behavior("a", {
+            a: [localRtcMember],
+          }).pipe(trackEpoch()),
+        ),
+        sentCallNotification$: hot("10ms a", {
+          a: [mockRingEvent("$right", 50), mockLegacyRingEvent],
+        }),
+        receivedDecline$: hot("20ms d", {
+          d: [
+            new MatrixEvent(declineEvent),
+            {} as Room,
+            undefined,
+            false,
+            {} as IRoomTimelineData,
+          ],
+        }),
+        options: {
+          waitForCallPickup: true,
+          autoLeaveWhenOthersLeft: false,
+        },
+        localUser: localRtcMember,
+      };
+      const lifecycle = createCallNotificationLifecycle$(props);
+      const marbles = expectDecline ? "a 9ms b 9ms d" : "a 9ms b";
+      expectObservable(lifecycle.callPickupState$, "21ms !").toBe(marbles, {
+        a: "unknown",
+        b: "ringing",
+        d: "decline",
+      });
+    });
+  }
+  const reference = (refId?: string, sender?: string): Partial<IEvent> => ({
+    event_id: "$decline",
+    type: EventType.RTCDecline,
+    sender: sender ?? "@other:example.org",
+    content: {
+      "m.relates_to": {
+        rel_type: "m.reference",
+        event_id: refId ?? "$right",
+      },
+    },
+  });
+  it("decline reference works", () => {
+    testStaysRinging(reference(), true);
+  });
+  it("decline with wrong id is ignored (stays ringing)", () => {
+    testStaysRinging(reference("$wrong"), false);
+  });
+  it("decline with wrong id is ignored (stays ringing)", () => {
+    testStaysRinging(reference(undefined, local.userId), false);
+  });
+});

src/state/CallViewModel/CallNotificationLifecycle.ts

@@ -0,0 +1,211 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import {
+  type CallMembership,
+  type MatrixRTCSession,
+  MatrixRTCSessionEvent,
+  type MatrixRTCSessionEventHandlerMap,
+} from "matrix-js-sdk/lib/matrixrtc";
+import {
+  combineLatest,
+  concat,
+  endWith,
+  filter,
+  fromEvent,
+  ignoreElements,
+  map,
+  merge,
+  NEVER,
+  type Observable,
+  of,
+  pairwise,
+  startWith,
+  switchMap,
+  takeUntil,
+  timer,
+} from "rxjs";
+import {
+  type EventTimelineSetHandlerMap,
+  EventType,
+  type Room as MatrixRoom,
+  RoomEvent,
+} from "matrix-js-sdk";
+
+import { type Behavior } from "../Behavior";
+import { type Epoch, mapEpoch, type ObservableScope } from "../ObservableScope";
+export type AutoLeaveReason = "allOthersLeft" | "timeout" | "decline";
+export type CallPickupState =
+  | "unknown"
+  | "ringing"
+  | "timeout"
+  | "decline"
+  | "success"
+  | null;
+export type CallNotificationWrapper = Parameters<
+  MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification]
+>;
+export function createSentCallNotification$(
+  scope: ObservableScope,
+  matrixRTCSession: MatrixRTCSession,
+): Behavior<CallNotificationWrapper | null> {
+  const sentCallNotification$ = scope.behavior(
+    fromEvent(matrixRTCSession, MatrixRTCSessionEvent.DidSendCallNotification),
+    null,
+  ) as Behavior<CallNotificationWrapper | null>;
+  return sentCallNotification$;
+}
+
+export function createReceivedDecline$(
+  matrixRoom: MatrixRoom,
+): Observable<Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>> {
+  return (
+    fromEvent(matrixRoom, RoomEvent.Timeline) as Observable<
+      Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
+    >
+  ).pipe(filter(([event]) => event.getType() === EventType.RTCDecline));
+}
+
+export interface Props {
+  scope: ObservableScope;
+  memberships$: Behavior<Epoch<CallMembership[]>>;
+  sentCallNotification$: Observable<CallNotificationWrapper | null>;
+  receivedDecline$: Observable<
+    Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
+  >;
+  options: { waitForCallPickup?: boolean; autoLeaveWhenOthersLeft?: boolean };
+  localUser: { deviceId: string; userId: string };
+}
+/**
+ * @returns {callPickupState$, autoLeave$}
+ * `callPickupState$` The current call pickup state of the call.
+ *  - "unknown": The client has not yet sent the notification event. We don't know if it will because it first needs to send its own membership.
+ *     Then we can conclude if we were the first one to join or not.
+ *     This may also be set if we are disconnected.
+ *  - "ringing": The call is ringing on other devices in this room (This client should give audiovisual feedback that this is happening).
+ *  - "timeout": No-one picked up in the defined time this call should be ringing on others devices.
+ *     The call failed. If desired this can be used as a trigger to exit the call.
+ *  - "success": Someone else joined. The call is in a normal state. No audiovisual feedback.
+ *  - null: EC is configured to never show any waiting for answer state.
+ *
+ * `autoLeave$` An observable that emits (null) when the call should be automatically left.
+ *  - if options.autoLeaveWhenOthersLeft is set to true it emits when all others left.
+ *  - if options.waitForCallPickup is set to true it emits if noone picked up the ring or if the ring got declined.
+ *  - if options.autoLeaveWhenOthersLeft && options.waitForCallPickup is false it will never emit.
+ *
+ */
+export function createCallNotificationLifecycle$({
+  scope,
+  memberships$,
+  sentCallNotification$,
+  receivedDecline$,
+  options,
+  localUser,
+}: Props): {
+  callPickupState$: Behavior<CallPickupState>;
+  autoLeave$: Observable<AutoLeaveReason>;
+} {
+  const allOthersLeft$ = memberships$.pipe(
+    pairwise(),
+    filter(
+      ([{ value: prev }, { value: current }]) =>
+        current.every((m) => m.userId === localUser.userId) &&
+        prev.some((m) => m.userId !== localUser.userId),
+    ),
+    map(() => {}),
+  );
+
+  /**
+   * Whether some Matrix user other than ourself is joined to the call.
+   */
+  const someoneElseJoined$ = memberships$.pipe(
+    mapEpoch((ms) => ms.some((m) => m.userId !== localUser.userId)),
+  ) as Behavior<Epoch<boolean>>;
+
+  /**
+   * Whenever the RTC session tells us that it intends to ring the remote
+   * participant's devices, this emits an Observable tracking the current state of
+   * that ringing process.
+   */
+  // This is a behavior since we need to store the latest state for when we subscribe to this after `didSendCallNotification$`
+  // has already emitted but we still need the latest observable with a timeout timer that only gets created on after receiving `notificationEvent`.
+  // A behavior will emit the latest observable with the running timer to new subscribers.
+  // see also: callPickupState$ and in particular the line: `return this.ring$.pipe(mergeAll());` here we otherwise might get an EMPTY observable if
+  // `ring$` would not be a behavior.
+  const remoteRingState$: Behavior<"ringing" | "timeout" | "decline" | null> =
+    scope.behavior(
+      sentCallNotification$.pipe(
+        filter(
+          (newAndLegacyEvents) =>
+            // only care about new events (legacy do not have decline pattern)
+            newAndLegacyEvents?.[0].notification_type === "ring",
+        ),
+        map((e) => e as CallNotificationWrapper),
+        switchMap(([notificationEvent]) => {
+          const lifetimeMs = notificationEvent?.lifetime ?? 0;
+          return concat(
+            lifetimeMs === 0
+              ? // If no lifetime, skip the ring state
+                of(null)
+              : // Ring until lifetime ms have passed
+                timer(lifetimeMs).pipe(
+                  ignoreElements(),
+                  startWith("ringing" as const),
+                ),
+            // The notification lifetime has timed out, meaning ringing has likely
+            // stopped on all receiving clients.
+            of("timeout" as const),
+            // This makes sure we will not drop into the `endWith("decline" as const)` state
+            NEVER,
+          ).pipe(
+            takeUntil(
+              receivedDecline$.pipe(
+                filter(
+                  ([event]) =>
+                    event.getRelation()?.rel_type === "m.reference" &&
+                    event.getRelation()?.event_id ===
+                      notificationEvent.event_id &&
+                    event.getSender() !== localUser.userId &&
+                    callPickupState$.value !== "timeout",
+                ),
+              ),
+            ),
+            endWith("decline" as const),
+          );
+        }),
+      ),
+      null,
+    );
+
+  const callPickupState$ = scope.behavior(
+    options.waitForCallPickup === true
+      ? combineLatest(
+          [someoneElseJoined$, remoteRingState$],
+          (someoneElseJoined, ring) => {
+            if (someoneElseJoined.value === true) {
+              return "success" as const;
+            }
+            // Show the ringing state of the most recent ringing attempt.
+            // as long as we have not yet sent an RTC notification event or noone else joined,
+            // ring will be null -> callPickupState$ = unknown.
+            return ring ?? ("unknown" as const);
+          },
+        )
+      : NEVER,
+    null,
+  );
+
+  const autoLeave$ = merge(
+    options.autoLeaveWhenOthersLeft === true
+      ? allOthersLeft$.pipe(map(() => "allOthersLeft" as const))
+      : NEVER,
+    callPickupState$.pipe(
+      filter((state) => state === "timeout" || state === "decline"),
+    ),
+  );
+  return { autoLeave$, callPickupState$ };
+}

src/state/CallViewModel/CallViewModelTestUtils.ts

@@ -0,0 +1,188 @@
+/*
+Copyright 2025 Element Corp.
+Copyright 2024 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import {
+  ConnectionState,
+  type LocalParticipant,
+  type Participant,
+  ParticipantEvent,
+  type RemoteParticipant,
+  type Room as LivekitRoom,
+} from "livekit-client";
+import { SyncState } from "matrix-js-sdk/lib/sync";
+import { BehaviorSubject, type Observable, map, of } from "rxjs";
+import { onTestFinished, vi } from "vitest";
+import { ClientEvent, type MatrixClient } from "matrix-js-sdk";
+import EventEmitter from "events";
+import * as ComponentsCore from "@livekit/components-core";
+
+import type { CallMembership } from "matrix-js-sdk/lib/matrixrtc";
+import { E2eeType } from "../../e2ee/e2eeType";
+import { type RaisedHandInfo, type ReactionInfo } from "../../reactions";
+import { CallViewModel, type CallViewModelOptions } from "./CallViewModel";
+import {
+  mockConfig,
+  mockLivekitRoom,
+  mockLocalParticipant,
+  mockMatrixRoom,
+  mockMatrixRoomMember,
+  mockMediaDevices,
+  mockMuteStates,
+  MockRTCSession,
+  testScope,
+} from "../../utils/test";
+import {
+  alice,
+  aliceDoppelganger,
+  bob,
+  bobZeroWidthSpace,
+  daveRTL,
+  daveRTLRtcMember,
+  local,
+  localRtcMember,
+} from "../../utils/test-fixtures";
+import { type Behavior, constant } from "../Behavior";
+import { type ProcessorState } from "../../livekit/TrackProcessorContext";
+import { type MediaDevices } from "../MediaDevices";
+
+mockConfig({
+  livekit: { livekit_service_url: "http://my-default-service-url.com" },
+});
+
+const carol = local;
+
+const dave = mockMatrixRoomMember(daveRTLRtcMember, { rawDisplayName: "Dave" });
+
+const roomMembers = new Map(
+  [alice, aliceDoppelganger, bob, bobZeroWidthSpace, carol, dave, daveRTL].map(
+    (p) => [p.userId, p],
+  ),
+);
+
+export interface CallViewModelInputs {
+  remoteParticipants$: Behavior<RemoteParticipant[]>;
+  rtcMembers$: Behavior<Partial<CallMembership>[]>;
+  livekitConnectionState$: Behavior<ConnectionState>;
+  speaking: Map<Participant, Observable<boolean>>;
+  mediaDevices: MediaDevices;
+  initialSyncState: SyncState;
+}
+
+const localParticipant = mockLocalParticipant({ identity: "" });
+
+export function withCallViewModel(
+  {
+    remoteParticipants$ = constant([]),
+    rtcMembers$ = constant([localRtcMember]),
+    livekitConnectionState$: connectionState$ = constant(
+      ConnectionState.Connected,
+    ),
+    speaking = new Map(),
+    mediaDevices = mockMediaDevices({}),
+    initialSyncState = SyncState.Syncing,
+  }: Partial<CallViewModelInputs> = {},
+  continuation: (
+    vm: CallViewModel,
+    rtcSession: MockRTCSession,
+    subjects: { raisedHands$: BehaviorSubject<Record<string, RaisedHandInfo>> },
+    setSyncState: (value: SyncState) => void,
+  ) => void,
+  options: CallViewModelOptions = {
+    encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+    autoLeaveWhenOthersLeft: false,
+  },
+): void {
+  let syncState = initialSyncState;
+  const setSyncState = (value: SyncState): void => {
+    const prev = syncState;
+    syncState = value;
+    room.client.emit(ClientEvent.Sync, value, prev);
+  };
+  const room = mockMatrixRoom({
+    client: new (class extends EventEmitter {
+      public getUserId(): string | undefined {
+        return localRtcMember.userId;
+      }
+
+      public getDeviceId(): string {
+        return localRtcMember.deviceId;
+      }
+
+      public getDomain(): string {
+        return "example.com";
+      }
+
+      public getSyncState(): SyncState {
+        return syncState;
+      }
+    })() as Partial<MatrixClient> as MatrixClient,
+    getMembers: () => Array.from(roomMembers.values()),
+  });
+  const rtcSession = new MockRTCSession(room, []).withMemberships(rtcMembers$);
+  const participantsSpy = vi
+    .spyOn(ComponentsCore, "connectedParticipantsObserver")
+    .mockReturnValue(remoteParticipants$);
+  const mediaSpy = vi
+    .spyOn(ComponentsCore, "observeParticipantMedia")
+    .mockImplementation((p) =>
+      of({ participant: p } as Partial<
+        ComponentsCore.ParticipantMedia<LocalParticipant>
+      > as ComponentsCore.ParticipantMedia<LocalParticipant>),
+    );
+  const eventsSpy = vi
+    .spyOn(ComponentsCore, "observeParticipantEvents")
+    .mockImplementation((p, ...eventTypes) => {
+      if (eventTypes.includes(ParticipantEvent.IsSpeakingChanged)) {
+        return (speaking.get(p) ?? of(false)).pipe(
+          map((s): Participant => ({ ...p, isSpeaking: s }) as Participant),
+        );
+      } else {
+        return of(p);
+      }
+    });
+
+  const roomEventSelectorSpy = vi
+    .spyOn(ComponentsCore, "roomEventSelector")
+    .mockImplementation((_room, _eventType) => of());
+  const muteStates = mockMuteStates();
+  const raisedHands$ = new BehaviorSubject<Record<string, RaisedHandInfo>>({});
+  const reactions$ = new BehaviorSubject<Record<string, ReactionInfo>>({});
+
+  const vm = new CallViewModel(
+    testScope(),
+    rtcSession.asMockedSession(),
+    room,
+    mediaDevices,
+    muteStates,
+    {
+      ...options,
+      livekitRoomFactory: (): LivekitRoom =>
+        mockLivekitRoom({
+          localParticipant,
+          disconnect: async () => Promise.resolve(),
+          setE2EEEnabled: async () => Promise.resolve(),
+        }),
+      connectionState$,
+    },
+    raisedHands$,
+    reactions$,
+    new BehaviorSubject<ProcessorState>({
+      processor: undefined,
+      supported: undefined,
+    }),
+  );
+
+  onTestFinished(() => {
+    participantsSpy.mockRestore();
+    mediaSpy.mockRestore();
+    eventsSpy.mockRestore();
+    roomEventSelectorSpy.mockRestore();
+  });
+
+  continuation(vm, rtcSession, { raisedHands$: raisedHands$ }, setSyncState);
+}

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

@@ -10,17 +10,16 @@ import { expect, test, vi } from "vitest";
 import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
 import EventEmitter from "events";
 
-import { enterRTCSession } from "../src/rtcSessionHelpers";
-import { mockConfig } from "./utils/test";
-import { MatrixRTCMode } from "./settings/settings";
+import { MatrixRTCMode } from "../../../settings/settings";
+import { mockConfig } from "../../../utils/test";
+import { enterRTCSession } from "./LocalMembership";
 
 const MATRIX_RTC_MODE = MatrixRTCMode.Legacy;
 const getUrlParams = vi.hoisted(() => vi.fn(() => ({})));
-vi.mock("./UrlParams", () => ({ getUrlParams }));
+vi.mock("../../../UrlParams", () => ({ getUrlParams }));
 
-const actualWidget = await vi.hoisted(async () => vi.importActual("./widget"));
-vi.mock("./widget", () => ({
-  ...actualWidget,
+vi.mock("../../../widget", async (importOriginal) => ({
+  ...(await importOriginal()),
   widget: {
     api: {
       setAlwaysOnScreen: (): void => {},

src/state/CallViewModel/localMember/LocalMembership.ts

@@ -0,0 +1,602 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import {
+  type LocalTrack,
+  type Participant,
+  ParticipantEvent,
+  type LocalParticipant,
+} from "livekit-client";
+import { observeParticipantEvents } from "@livekit/components-core";
+import {
+  type LivekitTransport,
+  type MatrixRTCSession,
+  MembershipManagerEvent,
+  Status,
+} from "matrix-js-sdk/lib/matrixrtc";
+import { ClientEvent, SyncState, type Room as MatrixRoom } from "matrix-js-sdk";
+import {
+  BehaviorSubject,
+  combineLatest,
+  fromEvent,
+  map,
+  type Observable,
+  of,
+  scan,
+  startWith,
+  switchMap,
+  tap,
+} from "rxjs";
+import { type Logger } from "matrix-js-sdk/lib/logger";
+
+import { type Behavior } from "../../Behavior";
+import { type IConnectionManager } from "../remoteMembers/ConnectionManager";
+import { ObservableScope } from "../../ObservableScope";
+import { Publisher } from "./Publisher";
+import { type MuteStates } from "../../MuteStates";
+import { type ProcessorState } from "../../../livekit/TrackProcessorContext";
+import { type MediaDevices } from "../../MediaDevices";
+import { and$ } from "../../../utils/observable";
+import { type ElementCallError } from "../../../utils/errors";
+import {
+  ElementWidgetActions,
+  widget,
+  type WidgetHelpers,
+} from "../../../widget";
+import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers";
+import { getUrlParams } from "../../../UrlParams.ts";
+import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics.ts";
+import { MatrixRTCMode } from "../../../settings/settings.ts";
+import { Config } from "../../../config/Config.ts";
+import { type Connection } from "../remoteMembers/Connection.ts";
+
+export enum LivekitState {
+  Uninitialized = "uninitialized",
+  Connecting = "connecting",
+  Connected = "connected",
+  Error = "error",
+  Disconnected = "disconnected",
+  Disconnecting = "disconnecting",
+}
+
+type LocalMemberLivekitState =
+  | { state: LivekitState.Error; error: string }
+  | { state: LivekitState.Connected }
+  | { state: LivekitState.Connecting }
+  | { state: LivekitState.Uninitialized }
+  | { state: LivekitState.Disconnected }
+  | { state: LivekitState.Disconnecting };
+
+export enum MatrixState {
+  Connected = "connected",
+  Disconnected = "disconnected",
+  Connecting = "connecting",
+}
+
+type LocalMemberMatrixState =
+  | { state: MatrixState.Connected }
+  | { state: MatrixState.Connecting }
+  | { state: MatrixState.Disconnected };
+
+export interface LocalMemberConnectionState {
+  livekit$: Behavior<LocalMemberLivekitState>;
+  matrix$: Behavior<LocalMemberMatrixState>;
+}
+
+/*
+ * - get well known
+ * - get oldest membership
+ * - get transport to use
+ * - get openId + jwt token
+ * - wait for createTrack() call
+ *    - create tracks
+ * - wait for join() call
+ *   - Publisher.publishTracks()
+ *   - send join state/sticky event
+ */
+interface Props {
+  options: Behavior<EnterRTCSessionOptions>;
+  scope: ObservableScope;
+  mediaDevices: MediaDevices;
+  muteStates: MuteStates;
+  connectionManager: IConnectionManager;
+  matrixRTCSession: MatrixRTCSession;
+  matrixRoom: MatrixRoom;
+  localTransport$: Behavior<LivekitTransport | null>;
+  trackProcessorState$: Behavior<ProcessorState>;
+  widget: WidgetHelpers | null;
+  logger: Logger;
+}
+
+/**
+ * This class is responsible for managing the own membership in a room.
+ * We want
+ *  - a publisher
+ *  -
+ * @param param0
+ * @returns
+ *  - publisher: The handle to create tracks and publish them to the room.
+ *  - connected$: the current connection state. Including matrix server and livekit server connection. (only considering the livekit server we are using for our own media publication)
+ *  - transport$: the transport object the ownMembership$ ended up using.
+ *  - connectionState: the current connection state. Including matrix server and livekit server connection.
+ *  - sharingScreen$: Whether we are sharing our screen. `undefined` if we cannot share the screen.
+ */
+export const createLocalMembership$ = ({
+  scope,
+  options,
+  muteStates,
+  mediaDevices,
+  connectionManager,
+  matrixRTCSession,
+  localTransport$,
+  matrixRoom,
+  trackProcessorState$,
+  widget,
+  logger: parentLogger,
+}: Props): {
+  // publisher: Publisher
+  requestConnect: () => LocalMemberConnectionState;
+  startTracks: () => Behavior<LocalTrack[]>;
+  requestDisconnect: () => Observable<LocalMemberLivekitState> | null;
+  connectionState: LocalMemberConnectionState;
+  sharingScreen$: Behavior<boolean>;
+  /**
+   * Callback to toggle screen sharing. If null, screen sharing is not possible.
+   */
+  toggleScreenSharing: (() => void) | null;
+  participant$: Behavior<LocalParticipant | null>;
+  connection$: Behavior<Connection | null>;
+  // deprecated fields
+  /** @deprecated use state instead*/
+  homeserverConnected$: Behavior<boolean>;
+  /** @deprecated use state instead*/
+  connected$: Behavior<boolean>;
+  // this needs to be discussed
+  /** @deprecated use state instead*/
+  reconnecting$: Behavior<boolean>;
+  // also needs to be disccues
+  /** @deprecated use state instead*/
+  configError$: Behavior<ElementCallError | null>;
+} => {
+  const logger = parentLogger.getChild("[LocalMembership]");
+  logger.debug(`Creating local membership..`);
+  const state = {
+    livekit$: new BehaviorSubject<LocalMemberLivekitState>({
+      state: LivekitState.Uninitialized,
+    }),
+    matrix$: new BehaviorSubject<LocalMemberMatrixState>({
+      state: MatrixState.Disconnected,
+    }),
+  };
+
+  // This should be used in a combineLatest with publisher$ to connect.
+  // to make it possible to call startTracks before the preferredTransport$ has resolved.
+  const trackStartRequested$ = new BehaviorSubject(false);
+
+  // This should be used in a combineLatest with publisher$ to connect.
+  // to make it possible to call startTracks before the preferredTransport$ has resolved.
+  const connectRequested$ = new BehaviorSubject(false);
+
+  // This should be used in a combineLatest with publisher$ to connect.
+  const tracks$ = new BehaviorSubject<LocalTrack[]>([]);
+
+  // Drop Epoch data here since we will not combine this anymore
+  const localConnection$ = scope.behavior(
+    combineLatest([connectionManager.connections$, localTransport$]).pipe(
+      map(([connections, localTransport]) => {
+        if (localTransport === null) {
+          return null;
+        }
+        return (
+          connections.value.find((connection) =>
+            areLivekitTransportsEqual(connection.transport, localTransport),
+          ) ?? null
+        );
+      }),
+      tap((connection) => {
+        logger.info(
+          `Local connection updated: ${connection?.transport?.livekit_service_url}`,
+        );
+      }),
+    ),
+  );
+  /**
+   * Whether we are connected to the MatrixRTC session.
+   */
+  const homeserverConnected$ = scope.behavior(
+    // To consider ourselves connected to MatrixRTC, we check the following:
+    and$(
+      // The client is connected to the sync loop
+      (
+        fromEvent(matrixRoom.client, ClientEvent.Sync) as Observable<
+          [SyncState]
+        >
+      ).pipe(
+        startWith([matrixRoom.client.getSyncState()]),
+        map(([state]) => state === SyncState.Syncing),
+      ),
+      // Room state observed by session says we're connected
+      fromEvent(matrixRTCSession, MembershipManagerEvent.StatusChanged).pipe(
+        startWith(null),
+        map(() => matrixRTCSession.membershipStatus === Status.Connected),
+      ),
+      // Also watch out for warnings that we've likely hit a timeout and our
+      // delayed leave event is being sent (this condition is here because it
+      // provides an earlier warning than the sync loop timeout, and we wouldn't
+      // see the actual leave event until we reconnect to the sync loop)
+      fromEvent(matrixRTCSession, MembershipManagerEvent.ProbablyLeft).pipe(
+        startWith(null),
+        map(() => matrixRTCSession.probablyLeft !== true),
+      ),
+    ).pipe(
+      tap((connected) => {
+        logger.info(`Homeserver connected update: ${connected}`);
+      }),
+    ),
+  );
+
+  // /**
+  //  * Whether we are "fully" connected to the call. Accounts for both the
+  //  * connection to the MatrixRTC session and the LiveKit publish connection.
+  //  */
+  // // TODO use this in combination with the MemberState.
+  const connected$ = scope.behavior(
+    and$(
+      homeserverConnected$,
+      localConnection$.pipe(
+        switchMap((c) =>
+          c
+            ? c.state$.pipe(map((state) => state.state === "ConnectedToLkRoom"))
+            : of(false),
+        ),
+      ),
+    ),
+  );
+
+  const publisher$ = new BehaviorSubject<Publisher | null>(null);
+  localConnection$.subscribe((connection) => {
+    if (connection !== null && publisher$.value === null) {
+      // TODO looks strange to not change publisher if connection changes.
+      publisher$.next(
+        new Publisher(
+          scope,
+          connection,
+          mediaDevices,
+          muteStates,
+          trackProcessorState$,
+        ),
+      );
+    }
+  });
+
+  combineLatest([publisher$, trackStartRequested$]).subscribe(
+    ([publisher, shouldStartTracks]) => {
+      if (publisher && shouldStartTracks) {
+        publisher
+          .createAndSetupTracks()
+          .then((tracks) => {
+            tracks$.next(tracks);
+          })
+          .catch((error) => {
+            logger.error("Error creating tracks:", error);
+          });
+      }
+    },
+  );
+
+  // MATRIX RELATED
+
+  // /**
+  //  * Whether we should tell the user that we're reconnecting to the call.
+  //  */
+  // DISCUSSION is there a better way to do this?
+  // sth that is more deriectly implied from the membership manager of the js sdk. (fromEvent(matrixRTCSession, Reconnecting)) ??? or similar
+  const reconnecting$ = scope.behavior(
+    connected$.pipe(
+      // We are reconnecting if we previously had some successful initial
+      // connection but are now disconnected
+      scan(
+        ({ connectedPreviously }, connectedNow) => ({
+          connectedPreviously: connectedPreviously || connectedNow,
+          reconnecting: connectedPreviously && !connectedNow,
+        }),
+        { connectedPreviously: false, reconnecting: false },
+      ),
+      map(({ reconnecting }) => reconnecting),
+    ),
+  );
+
+  const startTracks = (): Behavior<LocalTrack[]> => {
+    trackStartRequested$.next(true);
+    return tracks$;
+  };
+
+  combineLatest([publisher$, tracks$]).subscribe(([publisher, tracks]) => {
+    if (
+      tracks.length === 0 ||
+      // change this to !== Publishing
+      state.livekit$.value.state !== LivekitState.Uninitialized
+    ) {
+      return;
+    }
+    state.livekit$.next({ state: LivekitState.Connecting });
+    publisher
+      ?.startPublishing()
+      .then(() => {
+        state.livekit$.next({ state: LivekitState.Connected });
+      })
+      .catch((error) => {
+        state.livekit$.next({ state: LivekitState.Error, error });
+      });
+  });
+
+  combineLatest([localTransport$, connectRequested$]).subscribe(
+    // TODO reconnect on options change.
+    ([transport, connectRequested]) => {
+      if (
+        transport === null ||
+        !connectRequested ||
+        state.matrix$.value.state !== MatrixState.Disconnected
+      ) {
+        logger.info(
+          "Not yet connecting because: ",
+          "transport === null:",
+          transport === null,
+          "!connectRequested:",
+          !connectRequested,
+          "state.matrix$.value.state !== MatrixState.Disconnected:",
+          state.matrix$.value.state !== MatrixState.Disconnected,
+        );
+        return;
+      }
+      state.matrix$.next({ state: MatrixState.Connecting });
+      logger.info("Matrix State connecting");
+      enterRTCSession(matrixRTCSession, transport, options.value).catch(
+        (error) => {
+          logger.error(error);
+        },
+      );
+    },
+  );
+
+  const requestConnect = (): LocalMemberConnectionState => {
+    trackStartRequested$.next(true);
+    connectRequested$.next(true);
+
+    return state;
+  };
+
+  const requestDisconnect = (): Behavior<LocalMemberLivekitState> | null => {
+    if (state.livekit$.value.state !== LivekitState.Connected) return null;
+    state.livekit$.next({ state: LivekitState.Disconnecting });
+    combineLatest([publisher$, tracks$], (publisher, tracks) => {
+      publisher
+        ?.stopPublishing()
+        .then(() => {
+          tracks.forEach((track) => track.stop());
+          state.livekit$.next({ state: LivekitState.Disconnected });
+        })
+        .catch((error) => {
+          state.livekit$.next({ state: LivekitState.Error, error });
+        });
+    });
+
+    return state.livekit$;
+  };
+
+  // Pause upstream of all local media tracks when we're disconnected from
+  // MatrixRTC, because it can be an unpleasant surprise for the app to say
+  // 'reconnecting' and yet still be transmitting your media to others.
+  // We use matrixConnected$ rather than reconnecting$ because we want to
+  // pause tracks during the initial joining sequence too until we're sure
+  // that our own media is displayed on screen.
+  combineLatest([localConnection$, homeserverConnected$])
+    .pipe(scope.bind())
+    .subscribe(([connection, connected]) => {
+      if (connection?.state$.value.state !== "ConnectedToLkRoom") return;
+      const publications =
+        connection.livekitRoom.localParticipant.trackPublications.values();
+      if (connected) {
+        for (const p of publications) {
+          if (p.track?.isUpstreamPaused === true) {
+            const kind = p.track.kind;
+            logger.info(
+              `Resuming ${kind} track (MatrixRTC connection present)`,
+            );
+            p.track
+              .resumeUpstream()
+              .catch((e) =>
+                logger.error(
+                  `Failed to resume ${kind} track after MatrixRTC reconnection`,
+                  e,
+                ),
+              );
+          }
+        }
+      } else {
+        for (const p of publications) {
+          if (p.track?.isUpstreamPaused === false) {
+            const kind = p.track.kind;
+            logger.info(
+              `Pausing ${kind} track (uncertain MatrixRTC connection)`,
+            );
+            p.track
+              .pauseUpstream()
+              .catch((e) =>
+                logger.error(
+                  `Failed to pause ${kind} track after entering uncertain MatrixRTC connection`,
+                  e,
+                ),
+              );
+          }
+        }
+      }
+    });
+
+  const configError$ = new BehaviorSubject<ElementCallError | null>(null);
+  // TODO I do not fully understand what this does.
+  // Is it needed?
+  // Is this at the right place?
+  // Can this be simplified?
+  // Start and stop session membership as needed
+  scope.reconcile(localTransport$, async (advertised) => {
+    if (advertised !== null && advertised !== undefined) {
+      try {
+        configError$.next(null);
+        await enterRTCSession(matrixRTCSession, advertised, options.value);
+      } catch (e) {
+        logger.error("Error entering RTC session", e);
+      }
+
+      // Update our member event when our mute state changes.
+      const intentScope = new ObservableScope();
+      intentScope.reconcile(muteStates.video.enabled$, async (videoEnabled) =>
+        matrixRTCSession.updateCallIntent(videoEnabled ? "video" : "audio"),
+      );
+
+      return async (): Promise<void> => {
+        intentScope.end();
+        // Only sends Matrix leave event. The LiveKit session will disconnect
+        // as soon as either the stopConnection$ handler above gets to it or
+        // the view model is destroyed.
+        try {
+          await matrixRTCSession.leaveRoomSession();
+        } catch (e) {
+          logger.error("Error leaving RTC session", e);
+        }
+        try {
+          await widget?.api.transport.send(ElementWidgetActions.HangupCall, {});
+        } catch (e) {
+          logger.error("Failed to send hangup action", e);
+        }
+      };
+    }
+  });
+
+  /**
+   * Whether the user is currently sharing their screen.
+   */
+  const sharingScreen$ = scope.behavior(
+    localConnection$.pipe(
+      switchMap((c) =>
+        c === null
+          ? of(false)
+          : observeSharingScreen$(c.livekitRoom.localParticipant),
+      ),
+    ),
+  );
+
+  const toggleScreenSharing =
+    "getDisplayMedia" in (navigator.mediaDevices ?? {}) &&
+    !getUrlParams().hideScreensharing
+      ? (): void =>
+          // If a connection is ready, toggle screen sharing.
+          // We deliberately do nothing in the case of a null connection because
+          // it looks nice for the call control buttons to all become available
+          // at once upon joining the call, rather than introducing a disabled
+          // state. The user can just click again.
+          // We also allow screen sharing to be toggled even if the connection
+          // is still initializing or publishing tracks, because there's no
+          // technical reason to disallow this. LiveKit will publish if it can.
+          void localConnection$.value?.livekitRoom.localParticipant
+            .setScreenShareEnabled(!sharingScreen$.value, {
+              audio: true,
+              selfBrowserSurface: "include",
+              surfaceSwitching: "include",
+              systemAudio: "include",
+            })
+            .catch(logger.error)
+      : null;
+
+  const participant$ = scope.behavior(
+    localConnection$.pipe(map((c) => c?.livekitRoom.localParticipant ?? null)),
+  );
+  return {
+    startTracks,
+    requestConnect,
+    requestDisconnect,
+    connectionState: state,
+    homeserverConnected$,
+    connected$,
+    reconnecting$,
+    configError$,
+    sharingScreen$,
+    toggleScreenSharing,
+    participant$,
+    connection$: localConnection$,
+  };
+};
+
+export function observeSharingScreen$(p: Participant): Observable<boolean> {
+  return observeParticipantEvents(
+    p,
+    ParticipantEvent.TrackPublished,
+    ParticipantEvent.TrackUnpublished,
+    ParticipantEvent.LocalTrackPublished,
+    ParticipantEvent.LocalTrackUnpublished,
+  ).pipe(map((p) => p.isScreenShareEnabled));
+}
+
+interface EnterRTCSessionOptions {
+  encryptMedia: boolean;
+  matrixRTCMode: MatrixRTCMode;
+}
+
+/**
+ * TODO! document this function properly
+ * @param rtcSession
+ * @param transport
+ * @param options
+ * @throws If the widget could not send ElementWidgetActions.JoinCall action.
+ */
+// Exported for unit testing
+export async function enterRTCSession(
+  rtcSession: MatrixRTCSession,
+  transport: LivekitTransport,
+  { encryptMedia, matrixRTCMode }: EnterRTCSessionOptions,
+): Promise<void> {
+  PosthogAnalytics.instance.eventCallEnded.cacheStartCall(new Date());
+  PosthogAnalytics.instance.eventCallStarted.track(rtcSession.room.roomId);
+
+  // This must be called before we start trying to join the call, as we need to
+  // have started tracking by the time calls start getting created.
+  // groupCallOTelMembership?.onJoinCall();
+
+  const { features, matrix_rtc_session: matrixRtcSessionConfig } = Config.get();
+  const useDeviceSessionMemberEvents =
+    features?.feature_use_device_session_member_events;
+  const { sendNotificationType: notificationType, callIntent } = getUrlParams();
+  const multiSFU = matrixRTCMode !== MatrixRTCMode.Legacy;
+  // Multi-sfu does not need a preferred foci list. just the focus that is actually used.
+  rtcSession.joinRoomSession(
+    multiSFU ? [] : [transport],
+    multiSFU ? transport : undefined,
+    {
+      notificationType,
+      callIntent,
+      manageMediaKeys: encryptMedia,
+      ...(useDeviceSessionMemberEvents !== undefined && {
+        useLegacyMemberEvents: !useDeviceSessionMemberEvents,
+      }),
+      delayedLeaveEventRestartMs:
+        matrixRtcSessionConfig?.delayed_leave_event_restart_ms,
+      delayedLeaveEventDelayMs:
+        matrixRtcSessionConfig?.delayed_leave_event_delay_ms,
+      delayedLeaveEventRestartLocalTimeoutMs:
+        matrixRtcSessionConfig?.delayed_leave_event_restart_local_timeout_ms,
+      networkErrorRetryMs: matrixRtcSessionConfig?.network_error_retry_ms,
+      makeKeyDelay: matrixRtcSessionConfig?.wait_for_key_rotation_ms,
+      membershipEventExpiryMs:
+        matrixRtcSessionConfig?.membership_event_expiry_ms,
+      useExperimentalToDeviceTransport: true,
+      unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
+    },
+  );
+  if (widget) {
+    await widget.api.transport.send(ElementWidgetActions.JoinCall, {});
+  }
+}

src/state/CallViewModel/localMember/LocalTransport.ts

@@ -16,13 +16,13 @@ import { type MatrixClient } from "matrix-js-sdk";
 import { combineLatest, distinctUntilChanged, first, from, map } from "rxjs";
 import { logger } from "matrix-js-sdk/lib/logger";
 import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
-import { deepCompare } from "matrix-js-sdk/lib/utils";
 
-import { type Behavior } from "../Behavior.ts";
-import { type ObservableScope } from "../ObservableScope.ts";
-import { Config } from "../../config/Config.ts";
-import { MatrixRTCTransportMissingError } from "../../utils/errors.ts";
-import { getSFUConfigWithOpenID } from "../../livekit/openIDSFU.ts";
+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 { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts";
+import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
 
 /*
  * - get well known
@@ -37,7 +37,7 @@ import { getSFUConfigWithOpenID } from "../../livekit/openIDSFU.ts";
  */
 interface Props {
   scope: ObservableScope;
-  memberships$: Behavior<CallMembership[]>;
+  memberships$: Behavior<Epoch<CallMembership[]>>;
   client: MatrixClient;
   roomId: string;
   useOldestMember$: Behavior<boolean>;
@@ -56,36 +56,47 @@ export const createLocalTransport$ = ({
   client,
   roomId,
   useOldestMember$,
-}: Props): Behavior<LivekitTransport | undefined> => {
+}: Props): Behavior<LivekitTransport | null> => {
   /**
    * The transport over which we should be actively publishing our media.
    * undefined when not joined.
    */
   const oldestMemberTransport$ = scope.behavior(
     memberships$.pipe(
-      map((memberships) => memberships[0].getTransport(memberships[0])),
-      first((t) => t != undefined && isLivekitTransport(t)),
+      map(
+        (memberships) =>
+          memberships.value[0]?.getTransport(memberships.value[0]) ?? null,
+      ),
+      first((t) => t != null && isLivekitTransport(t)),
     ),
-    undefined,
+    null,
   );
 
   /**
    * The transport that we would personally prefer to publish on (if not for the
    * transport preferences of others, perhaps).
    */
-  const preferredTransport$: Behavior<LivekitTransport | undefined> =
-    scope.behavior(from(makeTransport(client, roomId)), undefined);
+  const preferredTransport$: Behavior<LivekitTransport | null> = scope.behavior(
+    from(makeTransport(client, roomId)),
+    null,
+  );
 
   /**
    * The transport we should advertise in our MatrixRTC membership.
    */
   const advertisedTransport$ = scope.behavior(
-    combineLatest(
-      [useOldestMember$, oldestMemberTransport$, preferredTransport$],
-      (useOldestMember, oldestMemberTransport, preferredTransport) =>
-        useOldestMember ? oldestMemberTransport : preferredTransport,
-    ).pipe<LivekitTransport>(distinctUntilChanged(deepCompare)),
-    undefined,
+    combineLatest([
+      useOldestMember$,
+      oldestMemberTransport$,
+      preferredTransport$,
+    ]).pipe(
+      map(([useOldestMember, oldestMemberTransport, preferredTransport]) =>
+        useOldestMember
+          ? (oldestMemberTransport ?? preferredTransport)
+          : preferredTransport,
+      ),
+      distinctUntilChanged(areLivekitTransportsEqual),
+    ),
   );
   return advertisedTransport$;
 };
@@ -99,7 +110,6 @@ async function makeTransportInternal(
   logger.log("Searching for a preferred transport");
   //TODO refactor this to use the jwt service returned alias.
   const livekitAlias = roomId;
-
   // TODO-MULTI-SFU: Either remove this dev tool or make it more official
   const urlFromStorage =
     localStorage.getItem("robin-matrixrtc-auth") ??

src/state/CallViewModel/localMember/Publisher.ts

@@ -5,7 +5,6 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 import {
-  type E2EEOptions,
   LocalVideoTrack,
   type Room as LivekitRoom,
   Track,
@@ -22,17 +21,17 @@ import {
 } from "rxjs";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 
-import type { Behavior } from "../Behavior.ts";
-import type { MediaDevices, SelectedDevice } from "../MediaDevices.ts";
-import type { MuteStates } from "../MuteStates.ts";
+import type { Behavior } from "../../Behavior.ts";
+import type { MediaDevices, SelectedDevice } from "../../MediaDevices.ts";
+import type { MuteStates } from "../../MuteStates.ts";
 import {
   type ProcessorState,
   trackProcessorSync,
-} from "../../livekit/TrackProcessorContext.tsx";
-import { getUrlParams } from "../../UrlParams.ts";
-import { observeTrackReference$ } from "../MediaViewModel.ts";
+} from "../../../livekit/TrackProcessorContext.tsx";
+import { getUrlParams } from "../../../UrlParams.ts";
+import { observeTrackReference$ } from "../../MediaViewModel.ts";
 import { type Connection } from "../remoteMembers/Connection.ts";
-import { type ObservableScope } from "../ObservableScope.ts";
+import { type ObservableScope } from "../../ObservableScope.ts";
 
 /**
  * A wrapper for a Connection object.
@@ -55,7 +54,6 @@ export class Publisher {
     private connection: Connection,
     devices: MediaDevices,
     private readonly muteStates: MuteStates,
-    e2eeLivekitOptions: E2EEOptions | undefined,
     trackerProcessorState$: Behavior<ProcessorState>,
     private logger?: Logger,
   ) {
@@ -64,7 +62,7 @@ export class Publisher {
 
     const room = connection.livekitRoom;
 
-    room.setE2EEEnabled(e2eeLivekitOptions !== undefined)?.catch((e) => {
+    room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
       this.logger?.error("Failed to set E2EE enabled on room", e);
     });
 
@@ -74,6 +72,12 @@ export class Publisher {
     this.observeMediaDevices(scope, devices, controlledAudioDevices);
 
     this.workaroundRestartAudioInputTrackChrome(devices, scope);
+    this.scope.onEnd(() => {
+      this.logger?.info(
+        "[PublishConnection] Scope ended -> stop publishing all tracks",
+      );
+      void this.stopPublishing();
+    });
   }
 
   /**
@@ -99,31 +103,36 @@ export class Publisher {
     // instead? This optimization would only be safe for a publish connection,
     // because we don't want to leak the user's intent to perhaps join a call to
     // remote servers before they actually commit to it.
-    const { promise, resolve, reject } = Promise.withResolvers<void>();
-    const sub = this.connection.state$.subscribe((s) => {
-      if (s.state !== "FailedToStart") {
-        reject(new Error("Disconnected from LiveKit server"));
-      } else {
-        resolve();
-      }
-    });
-    try {
-      await promise;
-    } catch (e) {
-      throw e;
-    } finally {
-      sub.unsubscribe();
-    }
+    // const { promise, resolve, reject } = Promise.withResolvers<void>();
+    // const sub = this.connection.state$.subscribe((s) => {
+    //   if (s.state === "FailedToStart") {
+    //     reject(new Error("Disconnected from LiveKit server"));
+    //   } else if (s.state === "ConnectedToLkRoom") {
+    //     resolve();
+    //   }
+    // });
+    // try {
+    //   await promise;
+    // } catch (e) {
+    //   throw e;
+    // } finally {
+    //   sub.unsubscribe();
+    // }
     // TODO-MULTI-SFU: Prepublish a microphone track
     const audio = this.muteStates.audio.enabled$.value;
     const video = this.muteStates.video.enabled$.value;
     // createTracks throws if called with audio=false and video=false
     if (audio || video) {
       // TODO this can still throw errors? It will also prompt for permissions if not already granted
-      this.tracks = await lkRoom.localParticipant.createTracks({
-        audio,
-        video,
-      });
+      this.tracks =
+        (await lkRoom.localParticipant
+          .createTracks({
+            audio,
+            video,
+          })
+          .catch((error) => {
+            this.logger?.error("Failed to create tracks", error);
+          })) ?? [];
     }
     return this.tracks;
   }
@@ -153,7 +162,9 @@ export class Publisher {
     for (const track of this.tracks) {
       // TODO: handle errors? Needs the signaling connection to be up, but it has some retries internally
       // with a timeout.
-      await lkRoom.localParticipant.publishTrack(track);
+      await lkRoom.localParticipant.publishTrack(track).catch((error) => {
+        this.logger?.error("Failed to publish track", error);
+      });
 
       // TODO: check if the connection is still active? and break the loop if not?
     }
@@ -249,7 +260,7 @@ export class Publisher {
         ) {
           lkRoom
             .switchActiveDevice(kind, device.id)
-            .catch((e) =>
+            .catch((e: Error) =>
               this.logger?.error(
                 `Failed to sync ${kind} device with LiveKit`,
                 e,
@@ -307,7 +318,8 @@ export class Publisher {
           return track instanceof LocalVideoTrack ? track : null;
         }),
       ),
+      null,
     );
-    trackProcessorSync(track$, trackerProcessorState$);
+    trackProcessorSync(scope, track$, trackerProcessorState$);
   }
 }

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

@@ -10,38 +10,32 @@ import {
   describe,
   expect,
   it,
-  type Mock,
   type MockedObject,
   onTestFinished,
   vi,
 } from "vitest";
-import { BehaviorSubject, of } from "rxjs";
 import {
   type LocalParticipant,
   type RemoteParticipant,
   type Room as LivekitRoom,
   RoomEvent,
-  type RoomOptions,
+  ConnectionState as LivekitConnectionState,
 } from "livekit-client";
 import fetchMock from "fetch-mock";
 import EventEmitter from "events";
 import { type IOpenIDToken } from "matrix-js-sdk";
+import { logger } from "matrix-js-sdk/lib/logger";
 
-import type {
-  CallMembership,
-  LivekitTransport,
-} from "matrix-js-sdk/lib/matrixrtc";
+import type { LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
 import {
+  Connection,
   type ConnectionOpts,
   type ConnectionState,
   type PublishingParticipant,
 } from "./Connection.ts";
-import { ObservableScope } from "../ObservableScope.ts";
-import { type OpenIDClientParts } from "../../livekit/openIDSFU.ts";
-import { FailToGetOpenIdToken } from "../../utils/errors.ts";
-import { mockMediaDevices, mockMuteStates } from "../../utils/test.ts";
-import type { ProcessorState } from "../../livekit/TrackProcessorContext.tsx";
-import { type MuteStates } from "../MuteStates.ts";
+import { ObservableScope } from "../../ObservableScope.ts";
+import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
+import { FailToGetOpenIdToken } from "../../../utils/errors.ts";
 
 let testScope: ObservableScope;
 
@@ -53,9 +47,9 @@ let localParticipantEventEmiter: EventEmitter;
 let fakeLocalParticipant: MockedObject<LocalParticipant>;
 
 let fakeRoomEventEmiter: EventEmitter;
-let fakeMembershipsFocusMap$: BehaviorSubject<
-  { membership: CallMembership; transport: LivekitTransport }[]
->;
+// let fakeMembershipsFocusMap$: BehaviorSubject<
+//   { membership: CallMembership; transport: LivekitTransport }[]
+// >;
 
 const livekitFocus: LivekitTransport = {
   livekit_alias: "!roomID:example.org",
@@ -74,9 +68,6 @@ function setupTest(): void {
     }),
     getDeviceId: vi.fn().mockReturnValue("ABCDEF"),
   } as unknown as OpenIDClientParts);
-  fakeMembershipsFocusMap$ = new BehaviorSubject<
-    { membership: CallMembership; transport: LivekitTransport }[]
-  >([]);
 
   localParticipantEventEmiter = new EventEmitter();
 
@@ -103,7 +94,7 @@ function setupTest(): void {
     disconnect: vi.fn(),
     remoteParticipants: new Map(),
     localParticipant: fakeLocalParticipant,
-    state: ConnectionState.Disconnected,
+    state: LivekitConnectionState.Disconnected,
     on: fakeRoomEventEmiter.on.bind(fakeRoomEventEmiter),
     off: fakeRoomEventEmiter.off.bind(fakeRoomEventEmiter),
     addListener: fakeRoomEventEmiter.addListener.bind(fakeRoomEventEmiter),
@@ -115,11 +106,10 @@ function setupTest(): void {
   } as unknown as LivekitRoom);
 }
 
-function setupRemoteConnection(): RemoteConnection {
+function setupRemoteConnection(): Connection {
   const opts: ConnectionOpts = {
     client: client,
     transport: livekitFocus,
-    remoteTransports$: fakeMembershipsFocusMap$,
     scope: testScope,
     livekitRoomFactory: () => fakeLivekitRoom,
   };
@@ -136,7 +126,7 @@ function setupRemoteConnection(): RemoteConnection {
 
   fakeLivekitRoom.connect.mockResolvedValue(undefined);
 
-  return new RemoteConnection(opts, undefined);
+  return new Connection(opts, logger);
 }
 
 afterEach(() => {
@@ -152,11 +142,10 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       transport: livekitFocus,
-      remoteTransports$: fakeMembershipsFocusMap$,
       scope: testScope,
       livekitRoomFactory: () => fakeLivekitRoom,
     };
-    const connection = new RemoteConnection(opts, undefined);
+    const connection = new Connection(opts, logger);
 
     expect(connection.state$.getValue().state).toEqual("Initialized");
   });
@@ -168,12 +157,11 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       transport: livekitFocus,
-      remoteTransports$: fakeMembershipsFocusMap$,
       scope: testScope,
       livekitRoomFactory: () => fakeLivekitRoom,
     };
 
-    const connection = new RemoteConnection(opts, undefined);
+    const connection = new Connection(opts, logger);
 
     const capturedStates: ConnectionState[] = [];
     const s = connection.state$.subscribe((value) => {
@@ -221,12 +209,11 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       transport: livekitFocus,
-      remoteTransports$: fakeMembershipsFocusMap$,
       scope: testScope,
       livekitRoomFactory: () => fakeLivekitRoom,
     };
 
-    const connection = new RemoteConnection(opts, undefined);
+    const connection = new Connection(opts, logger);
 
     const capturedStates: ConnectionState[] = [];
     const s = connection.state$.subscribe((value) => {
@@ -278,12 +265,11 @@ describe("Start connection states", () => {
     const opts: ConnectionOpts = {
       client: client,
       transport: livekitFocus,
-      remoteTransports$: fakeMembershipsFocusMap$,
       scope: testScope,
       livekitRoomFactory: () => fakeLivekitRoom,
     };
 
-    const connection = new RemoteConnection(opts, undefined);
+    const connection = new Connection(opts, logger);
 
     const capturedStates: ConnectionState[] = [];
     const s = connection.state$.subscribe((value) => {
@@ -376,21 +362,18 @@ describe("Start connection states", () => {
   });
 });
 
-function fakeRemoteLivekitParticipant(id: string): RemoteParticipant {
+function fakeRemoteLivekitParticipant(
+  id: string,
+  publications: number = 1,
+): RemoteParticipant {
   return {
     identity: id,
+    getTrackPublications: () => Array(publications),
   } as unknown as RemoteParticipant;
 }
 
-function fakeRtcMemberShip(userId: string, deviceId: string): CallMembership {
-  return {
-    userId,
-    deviceId,
-  } as unknown as CallMembership;
-}
-
 describe("Publishing participants observations", () => {
-  it("should emit the list of publishing participants", async () => {
+  it("should emit the list of publishing participants", () => {
     setupTest();
 
     const connection = setupRemoteConnection();
@@ -398,135 +381,53 @@ describe("Publishing participants observations", () => {
     const bobIsAPublisher = Promise.withResolvers<void>();
     const danIsAPublisher = Promise.withResolvers<void>();
     const observedPublishers: PublishingParticipant[][] = [];
-    const s = connection.allLivekitParticipants$.subscribe((publishers) => {
-      observedPublishers.push(publishers);
-      if (
-        publishers.some(
-          (p) => p.participant?.identity === "@bob:example.org:DEV111",
-        )
-      ) {
-        bobIsAPublisher.resolve();
-      }
-      if (
-        publishers.some(
-          (p) => p.participant?.identity === "@dan:example.org:DEV333",
-        )
-      ) {
-        danIsAPublisher.resolve();
-      }
-    });
+    const s = connection.remoteParticipantsWithTracks$.subscribe(
+      (publishers) => {
+        observedPublishers.push(publishers);
+        if (publishers.some((p) => p.identity === "@bob:example.org:DEV111")) {
+          bobIsAPublisher.resolve();
+        }
+        if (publishers.some((p) => p.identity === "@dan:example.org:DEV333")) {
+          danIsAPublisher.resolve();
+        }
+      },
+    );
     onTestFinished(() => s.unsubscribe());
     // The publishingParticipants$ observable is derived from the current members of the
     // livekitRoom and the rtc membership in order to publish the members that are publishing
     // on this connection.
 
     let participants: RemoteParticipant[] = [
-      fakeRemoteLivekitParticipant("@alice:example.org:DEV000"),
-      fakeRemoteLivekitParticipant("@bob:example.org:DEV111"),
-      fakeRemoteLivekitParticipant("@carol:example.org:DEV222"),
-      fakeRemoteLivekitParticipant("@dan:example.org:DEV333"),
+      fakeRemoteLivekitParticipant("@alice:example.org:DEV000", 0),
+      fakeRemoteLivekitParticipant("@bob:example.org:DEV111", 0),
+      fakeRemoteLivekitParticipant("@carol:example.org:DEV222", 0),
+      fakeRemoteLivekitParticipant("@dan:example.org:DEV333", 0),
     ];
 
     // Let's simulate 3 members on the livekitRoom
-    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockReturnValue(
-      new Map(participants.map((p) => [p.identity, p])),
+    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockImplementation(
+      () => new Map(participants.map((p) => [p.identity, p])),
     );
 
-    for (const participant of participants) {
-      fakeRoomEventEmiter.emit(RoomEvent.ParticipantConnected, participant);
-    }
+    participants.forEach((p) =>
+      fakeRoomEventEmiter.emit(RoomEvent.ParticipantConnected, p),
+    );
 
     // At this point there should be no publishers
     expect(observedPublishers.pop()!.length).toEqual(0);
 
-    const otherFocus: LivekitTransport = {
-      livekit_alias: "!roomID:example.org",
-      livekit_service_url: "https://other-matrix-rtc.example.org/livekit/jwt",
-      type: "livekit",
-    };
-
-    const rtcMemberships = [
-      // Say bob is on the same focus
-      {
-        membership: fakeRtcMemberShip("@bob:example.org", "DEV111"),
-        transport: livekitFocus,
-      },
-      // Alice and carol is on a different focus
-      {
-        membership: fakeRtcMemberShip("@alice:example.org", "DEV000"),
-        transport: otherFocus,
-      },
-      {
-        membership: fakeRtcMemberShip("@carol:example.org", "DEV222"),
-        transport: otherFocus,
-      },
-      // NO DAVE YET
+    participants = [
+      fakeRemoteLivekitParticipant("@alice:example.org:DEV000", 1),
+      fakeRemoteLivekitParticipant("@bob:example.org:DEV111", 1),
+      fakeRemoteLivekitParticipant("@carol:example.org:DEV222", 1),
+      fakeRemoteLivekitParticipant("@dan:example.org:DEV333", 2),
     ];
-    // signal this change in rtc memberships
-    fakeMembershipsFocusMap$.next(rtcMemberships);
-
-    // We should have bob has a publisher now
-    await bobIsAPublisher.promise;
-    const publishers = observedPublishers.pop();
-    expect(publishers?.length).toEqual(1);
-    expect(publishers?.[0].participant?.identity).toEqual(
-      "@bob:example.org:DEV111",
+    participants.forEach((p) =>
+      fakeRoomEventEmiter.emit(RoomEvent.ParticipantConnected, p),
     );
 
-    // Now let's make dan join the rtc memberships
-    rtcMemberships.push({
-      membership: fakeRtcMemberShip("@dan:example.org", "DEV333"),
-      transport: livekitFocus,
-    });
-    fakeMembershipsFocusMap$.next(rtcMemberships);
-
-    // We should have bob and dan has publishers now
-    await danIsAPublisher.promise;
-    const twoPublishers = observedPublishers.pop();
-    expect(twoPublishers?.length).toEqual(2);
-    expect(
-      twoPublishers?.some(
-        (p) => p.participant?.identity === "@bob:example.org:DEV111",
-      ),
-    ).toBeTruthy();
-    expect(
-      twoPublishers?.some(
-        (p) => p.participant?.identity === "@dan:example.org:DEV333",
-      ),
-    ).toBeTruthy();
-
-    // Now let's make bob leave the livekit room
-    participants = participants.filter(
-      (p) => p.identity !== "@bob:example.org:DEV111",
-    );
-    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockReturnValue(
-      new Map(participants.map((p) => [p.identity, p])),
-    );
-    fakeRoomEventEmiter.emit(
-      RoomEvent.ParticipantDisconnected,
-      fakeRemoteLivekitParticipant("@bob:example.org:DEV111"),
-    );
-
-    const updatedPublishers = observedPublishers.pop();
-    // Bob is not connected to the room but he is still in the rtc memberships declaring that
-    // he is using that focus to publish, so he should still appear as a publisher
-    expect(updatedPublishers?.length).toEqual(2);
-    const pp = updatedPublishers?.find(
-      (p) => p.membership.userId == "@bob:example.org",
-    );
-    expect(pp).toBeDefined();
-    expect(pp!.participant).not.toBeDefined();
-    expect(
-      updatedPublishers?.some(
-        (p) => p.participant?.identity === "@dan:example.org:DEV333",
-      ),
-    ).toBeTruthy();
-    // Now if bob is not in the rtc memberships, he should disappear
-    const noBob = rtcMemberships.filter(
-      ({ membership }) => membership.userId !== "@bob:example.org",
-    );
-    fakeMembershipsFocusMap$.next(noBob);
-    expect(observedPublishers.pop()?.length).toEqual(1);
+    // At this point there should be no publishers
+    expect(observedPublishers.pop()!.length).toEqual(4);
   });
 
   it("should be scoped to parent scope", (): void => {
@@ -535,18 +436,20 @@ describe("Publishing participants observations", () => {
     const connection = setupRemoteConnection();
 
     let observedPublishers: PublishingParticipant[][] = [];
-    const s = connection.allLivekitParticipants$.subscribe((publishers) => {
-      observedPublishers.push(publishers);
-    });
+    const s = connection.remoteParticipantsWithTracks$.subscribe(
+      (publishers) => {
+        observedPublishers.push(publishers);
+      },
+    );
     onTestFinished(() => s.unsubscribe());
 
     let participants: RemoteParticipant[] = [
-      fakeRemoteLivekitParticipant("@bob:example.org:DEV111"),
+      fakeRemoteLivekitParticipant("@bob:example.org:DEV111", 0),
     ];
 
     // Let's simulate 3 members on the livekitRoom
-    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockReturnValue(
-      new Map(participants.map((p) => [p.identity, p])),
+    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockImplementation(
+      () => new Map(participants.map((p) => [p.identity, p])),
     );
 
     for (const participant of participants) {
@@ -556,22 +459,16 @@ describe("Publishing participants observations", () => {
     // At this point there should be no publishers
     expect(observedPublishers.pop()!.length).toEqual(0);
 
-    const rtcMemberships = [
-      // Say bob is on the same focus
-      {
-        membership: fakeRtcMemberShip("@bob:example.org", "DEV111"),
-        transport: livekitFocus,
-      },
-    ];
-    // signal this change in rtc memberships
-    fakeMembershipsFocusMap$.next(rtcMemberships);
+    participants = [fakeRemoteLivekitParticipant("@bob:example.org:DEV111", 1)];
+
+    for (const participant of participants) {
+      fakeRoomEventEmiter.emit(RoomEvent.ParticipantConnected, participant);
+    }
 
     // We should have bob has a publisher now
     const publishers = observedPublishers.pop();
     expect(publishers?.length).toEqual(1);
-    expect(publishers?.[0].participant?.identity).toEqual(
-      "@bob:example.org:DEV111",
-    );
+    expect(publishers?.[0]?.identity).toEqual("@bob:example.org:DEV111");
 
     // end the parent scope
     testScope.end();
@@ -581,9 +478,7 @@ describe("Publishing participants observations", () => {
     participants = participants.filter(
       (p) => p.identity !== "@bob:example.org:DEV111",
     );
-    vi.spyOn(fakeLivekitRoom, "remoteParticipants", "get").mockReturnValue(
-      new Map(participants.map((p) => [p.identity, p])),
-    );
+
     fakeRoomEventEmiter.emit(
       RoomEvent.ParticipantDisconnected,
       fakeRemoteLivekitParticipant("@bob:example.org:DEV111"),
@@ -593,108 +488,112 @@ describe("Publishing participants observations", () => {
   });
 });
 
-describe("PublishConnection", () => {
-  // let fakeBlurProcessor: ProcessorWrapper<BackgroundOptions>;
-  let roomFactoryMock: Mock<() => LivekitRoom>;
-  let muteStates: MockedObject<MuteStates>;
+//
+// NOT USED ANYMORE ?
+//
+// This setup look like sth for the Publisher. Not a connection.
 
-  function setUpPublishConnection(): void {
-    setupTest();
+// describe("PublishConnection", () => {
+//   // let fakeBlurProcessor: ProcessorWrapper<BackgroundOptions>;
+//   let roomFactoryMock: Mock<() => LivekitRoom>;
+//   let muteStates: MockedObject<MuteStates>;
 
-    roomFactoryMock = vi.fn().mockReturnValue(fakeLivekitRoom);
+//   function setUpPublishConnection(): void {
+//     setupTest();
 
-    muteStates = mockMuteStates();
+//     roomFactoryMock = vi.fn().mockReturnValue(fakeLivekitRoom);
 
-    // fakeBlurProcessor = vi.mocked<ProcessorWrapper<BackgroundOptions>>({
-    //   name: "BackgroundBlur",
-    //   restart: vi.fn().mockResolvedValue(undefined),
-    //   setOptions: vi.fn().mockResolvedValue(undefined),
-    //   getOptions: vi.fn().mockReturnValue({ strength: 0.5 }),
-    //   isRunning: vi.fn().mockReturnValue(false)
-    // });
-  }
+//     muteStates = mockMuteStates();
 
-  describe("Livekit room creation", () => {
-    function createSetup(): void {
-      setUpPublishConnection();
+//     // fakeBlurProcessor = vi.mocked<ProcessorWrapper<BackgroundOptions>>({
+//     //   name: "BackgroundBlur",
+//     //   restart: vi.fn().mockResolvedValue(undefined),
+//     //   setOptions: vi.fn().mockResolvedValue(undefined),
+//     //   getOptions: vi.fn().mockReturnValue({ strength: 0.5 }),
+//     //   isRunning: vi.fn().mockReturnValue(false)
+//     // });
+//   }
 
-      const fakeTrackProcessorSubject$ = new BehaviorSubject<ProcessorState>({
-        supported: true,
-        processor: undefined,
-      });
+// describe("Livekit room creation", () => {
+//   function createSetup(): void {
+//     setUpPublishConnection();
 
-      const opts: ConnectionOpts = {
-        client: client,
-        transport: livekitFocus,
-        remoteTransports$: fakeMembershipsFocusMap$,
-        scope: testScope,
-        livekitRoomFactory: roomFactoryMock,
-      };
+//     const fakeTrackProcessorSubject$ = new BehaviorSubject<ProcessorState>({
+//       supported: true,
+//       processor: undefined,
+//     });
 
-      const audioInput = {
-        available$: of(new Map([["mic1", { id: "mic1" }]])),
-        selected$: new BehaviorSubject({ id: "mic1" }),
-        select(): void {},
-      };
+//     const opts: ConnectionOpts = {
+//       client: client,
+//       transport: livekitFocus,
+//       scope: testScope,
+//       livekitRoomFactory: roomFactoryMock,
+//     };
 
-      const videoInput = {
-        available$: of(new Map([["cam1", { id: "cam1" }]])),
-        selected$: new BehaviorSubject({ id: "cam1" }),
-        select(): void {},
-      };
+//     const audioInput = {
+//       available$: of(new Map([["mic1", { id: "mic1" }]])),
+//       selected$: new BehaviorSubject({ id: "mic1" }),
+//       select(): void {},
+//     };
 
-      const audioOutput = {
-        available$: of(new Map([["speaker", { id: "speaker" }]])),
-        selected$: new BehaviorSubject({ id: "speaker" }),
-        select(): void {},
-      };
+//     const videoInput = {
+//       available$: of(new Map([["cam1", { id: "cam1" }]])),
+//       selected$: new BehaviorSubject({ id: "cam1" }),
+//       select(): void {},
+//     };
 
-      // TODO understand what is wrong with our mocking that requires ts-expect-error
-      const fakeDevices = mockMediaDevices({
-        // @ts-expect-error Mocking only
-        audioInput,
-        // @ts-expect-error Mocking only
-        videoInput,
-        // @ts-expect-error Mocking only
-        audioOutput,
-      });
+//     const audioOutput = {
+//       available$: of(new Map([["speaker", { id: "speaker" }]])),
+//       selected$: new BehaviorSubject({ id: "speaker" }),
+//       select(): void {},
+//     };
 
-      new PublishConnection(
-        opts,
-        fakeDevices,
-        muteStates,
-        undefined,
-        fakeTrackProcessorSubject$,
-      );
-    }
+//     // TODO understand what is wrong with our mocking that requires ts-expect-error
+//     const fakeDevices = mockMediaDevices({
+//       // @ts-expect-error Mocking only
+//       audioInput,
+//       // @ts-expect-error Mocking only
+//       videoInput,
+//       // @ts-expect-error Mocking only
+//       audioOutput,
+//     });
 
-    it("should create room with proper initial audio and video settings", () => {
-      createSetup();
+//     new Connection(
+//       opts,
+//       fakeDevices,
+//       muteStates,
+//       undefined,
+//       fakeTrackProcessorSubject$,
+//     );
+//   }
 
-      expect(roomFactoryMock).toHaveBeenCalled();
+//   it("should create room with proper initial audio and video settings", () => {
+//     createSetup();
 
-      const lastCallArgs =
-        roomFactoryMock.mock.calls[roomFactoryMock.mock.calls.length - 1];
+//     expect(roomFactoryMock).toHaveBeenCalled();
 
-      const roomOptions = lastCallArgs.pop() as unknown as RoomOptions;
-      expect(roomOptions).toBeDefined();
+//     const lastCallArgs =
+//       roomFactoryMock.mock.calls[roomFactoryMock.mock.calls.length - 1];
 
-      expect(roomOptions!.videoCaptureDefaults?.deviceId).toEqual("cam1");
-      expect(roomOptions!.audioCaptureDefaults?.deviceId).toEqual("mic1");
-      expect(roomOptions!.audioOutput?.deviceId).toEqual("speaker");
-    });
+//     const roomOptions = lastCallArgs.pop() as unknown as RoomOptions;
+//     expect(roomOptions).toBeDefined();
 
-    it("respect controlledAudioDevices", () => {
-      // TODO: Refactor the code to make it testable.
-      // The UrlParams module is a singleton has a cache and is very hard to test.
-      // This breaks other tests as well if not handled properly.
-      // vi.mock(import("./../UrlParams"), () => {
-      //   return {
-      //     getUrlParams: vi.fn().mockReturnValue({
-      //       controlledAudioDevices: true
-      //     })
-      //   };
-      // });
-    });
-  });
-});
+//     expect(roomOptions!.videoCaptureDefaults?.deviceId).toEqual("cam1");
+//     expect(roomOptions!.audioCaptureDefaults?.deviceId).toEqual("mic1");
+//     expect(roomOptions!.audioOutput?.deviceId).toEqual("speaker");
+//   });
+
+//   it("respect controlledAudioDevices", () => {
+//     // TODO: Refactor the code to make it testable.
+//     // The UrlParams module is a singleton has a cache and is very hard to test.
+//     // This breaks other tests as well if not handled properly.
+//     // vi.mock(import("./../UrlParams"), () => {
+//     //   return {
+//     //     getUrlParams: vi.fn().mockReturnValue({
+//     //       controlledAudioDevices: true
+//     //     })
+//     //   };
+//     // });
+//   });
+// });
+// });

src/state/CallViewModel/remoteMembers/Connection.ts

@@ -18,20 +18,20 @@ import {
   RoomEvent,
 } from "livekit-client";
 import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
-import { BehaviorSubject, type Observable } from "rxjs";
+import { BehaviorSubject, map, type Observable } from "rxjs";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 
 import {
   getSFUConfigWithOpenID,
   type OpenIDClientParts,
   type SFUConfig,
-} from "../../livekit/openIDSFU.ts";
-import { type Behavior } from "../Behavior.ts";
-import { type ObservableScope } from "../ObservableScope.ts";
+} from "../../../livekit/openIDSFU.ts";
+import { type Behavior } from "../../Behavior.ts";
+import { type ObservableScope } from "../../ObservableScope.ts";
 import {
   InsufficientCapacityError,
   SFURoomCreationRestrictedError,
-} from "../../utils/errors.ts";
+} from "../../../utils/errors.ts";
 
 export type PublishingParticipant = LocalParticipant | RemoteParticipant;
 
@@ -98,6 +98,7 @@ export class Connection {
   // TODO dont make this throw and instead store a connection error state in this class?
   // TODO consider an autostart pattern...
   public async start(): Promise<void> {
+    this.logger.debug("Starting Connection");
     this.stopped = false;
     try {
       this._state$.next({
@@ -145,6 +146,7 @@ export class Connection {
         livekitConnectionState$: connectionStateObserver(this.livekitRoom),
       });
     } catch (error) {
+      this.logger.debug(`Failed to connect to LiveKit room: ${error}`);
       this._state$.next({
         state: "FailedToStart",
         error: error instanceof Error ? error : new Error(`${error}`),
@@ -169,6 +171,9 @@ export class Connection {
    * If the connection is already stopped, this is a no-op.
    */
   public async stop(): Promise<void> {
+    this.logger.debug(
+      `Stopping connection to ${this.transport.livekit_service_url}`,
+    );
     if (this.stopped) return;
     await this.livekitRoom.disconnect();
     this._state$.next({
@@ -179,12 +184,13 @@ export class Connection {
   }
 
   /**
-   * An observable of the participants that are publishing on this connection.
+   * An observable of the participants that are publishing on this connection. (Excluding our local participant)
    * This is derived from `participantsIncludingSubscribers$` and `remoteTransports$`.
    * It filters the participants to only those that are associated with a membership that claims to publish on this connection.
    */
-
-  public readonly participantsWithTrack$: Behavior<PublishingParticipant[]>;
+  public readonly remoteParticipantsWithTracks$: Behavior<
+    PublishingParticipant[]
+  >;
 
   /**
    * The media transport to connect to.
@@ -194,15 +200,18 @@ export class Connection {
   private readonly client: OpenIDClientParts;
   public readonly livekitRoom: LivekitRoom;
 
+  private readonly logger: Logger;
+
   /**
    * Creates a new connection to a matrix RTC LiveKit backend.
    *
-   * @param livekitRoom - LiveKit room instance to use.
    * @param opts - Connection options {@link ConnectionOpts}.
    *
+   * @param logger
    */
-  public constructor(opts: ConnectionOpts, logger?: Logger) {
-    logger?.info(
+  public constructor(opts: ConnectionOpts, logger: Logger) {
+    this.logger = logger.getChild("[Connection]");
+    this.logger.info(
       `[Connection] Creating new connection to ${opts.transport.livekit_service_url} ${opts.transport.livekit_alias}`,
     );
     const { transport, client, scope } = opts;
@@ -211,16 +220,28 @@ export class Connection {
     this.transport = transport;
     this.client = client;
 
-    this.participantsWithTrack$ = scope.behavior(
+    // REMOTE participants with track!!!
+    // this.remoteParticipantsWithTracks$
+    this.remoteParticipantsWithTracks$ = scope.behavior(
+      // only tracks remote participants
       connectedParticipantsObserver(this.livekitRoom, {
         additionalRoomEvents: [
           RoomEvent.TrackPublished,
           RoomEvent.TrackUnpublished,
         ],
-      }),
+      }).pipe(
+        map((participants) => {
+          return participants.filter(
+            (participant) => participant.getTrackPublications().length > 0,
+          );
+        }),
+      ),
       [],
     );
 
-    scope.onEnd(() => void this.stop());
+    scope.onEnd(() => {
+      this.logger.info(`Connection scope ended, stopping connection`);
+      void this.stop();
+    });
   }
 }

src/state/CallViewModel/remoteMembers/ConnectionFactory.ts

@@ -10,16 +10,18 @@ import {
   type E2EEOptions,
   Room as LivekitRoom,
   type RoomOptions,
+  type BaseKeyProvider,
 } from "livekit-client";
 import { type Logger } from "matrix-js-sdk/lib/logger";
+import E2EEWorker from "livekit-client/e2ee-worker?worker";
 
-import { type ObservableScope } from "../ObservableScope.ts";
+import { type ObservableScope } from "../../ObservableScope.ts";
 import { Connection } from "./Connection.ts";
-import type { OpenIDClientParts } from "../../livekit/openIDSFU.ts";
-import type { MediaDevices } from "../MediaDevices.ts";
-import type { Behavior } from "../Behavior.ts";
-import type { ProcessorState } from "../../livekit/TrackProcessorContext.tsx";
-import { defaultLiveKitOptions } from "../../livekit/options.ts";
+import type { OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
+import type { MediaDevices } from "../../MediaDevices.ts";
+import type { Behavior } from "../../Behavior.ts";
+import type { ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
+import { defaultLiveKitOptions } from "../../../livekit/options.ts";
 
 export interface ConnectionFactory {
   createConnection(
@@ -46,7 +48,7 @@ export class ECConnectionFactory implements ConnectionFactory {
     private client: OpenIDClientParts,
     private devices: MediaDevices,
     private processorState$: Behavior<ProcessorState>,
-    private e2eeLivekitOptions: E2EEOptions | undefined,
+    livekitKeyProvider: BaseKeyProvider | undefined,
     private controlledAudioDevices: boolean,
     livekitRoomFactory?: () => LivekitRoom,
   ) {
@@ -55,7 +57,12 @@ export class ECConnectionFactory implements ConnectionFactory {
         generateRoomOption(
           this.devices,
           this.processorState$.value,
-          this.e2eeLivekitOptions,
+          livekitKeyProvider && {
+            keyProvider: livekitKeyProvider,
+            // It's important that every room use a separate E2EE worker.
+            // They get confused if given streams from multiple rooms.
+            worker: new E2EEWorker(),
+          },
           this.controlledAudioDevices,
         ),
       );

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

@@ -0,0 +1,332 @@
+/*
+Copyright 2025 Element Creations Ltd.
+
+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, test, vi } from "vitest";
+import { BehaviorSubject } from "rxjs";
+import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+import { type Participant as LivekitParticipant } from "livekit-client";
+import { logger } from "matrix-js-sdk/lib/logger";
+
+import { Epoch, ObservableScope } from "../../ObservableScope.ts";
+import {
+  createConnectionManager$,
+  type ConnectionManagerData,
+} from "./ConnectionManager.ts";
+import { type ConnectionFactory } from "./ConnectionFactory.ts";
+import { type Connection } from "./Connection.ts";
+import { withTestScheduler } from "../../../utils/test.ts";
+import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
+import { type Behavior } from "../../Behavior.ts";
+
+// Some test constants
+
+const TRANSPORT_1: LivekitTransport = {
+  type: "livekit",
+  livekit_service_url: "https://lk.example.org",
+  livekit_alias: "!alias:example.org",
+};
+
+const TRANSPORT_2: LivekitTransport = {
+  type: "livekit",
+  livekit_service_url: "https://lk.sample.com",
+  livekit_alias: "!alias:sample.com",
+};
+
+let fakeConnectionFactory: ConnectionFactory;
+let testScope: ObservableScope;
+
+// Can be useful to track all created connections in tests, even the disposed ones
+let allCreatedConnections: Connection[];
+
+beforeEach(() => {
+  testScope = new ObservableScope();
+  allCreatedConnections = [];
+  fakeConnectionFactory = {} as unknown as ConnectionFactory;
+  vi.mocked(fakeConnectionFactory).createConnection = vi
+    .fn()
+    .mockImplementation(
+      (transport: LivekitTransport, scope: ObservableScope) => {
+        const mockConnection = {
+          transport,
+          remoteParticipantsWithTracks$: new BehaviorSubject([]),
+        } as unknown as Connection;
+        vi.mocked(mockConnection).start = vi.fn();
+        vi.mocked(mockConnection).stop = vi.fn();
+        // Tie the connection's lifecycle to the scope to test scope lifecycle management
+        scope.onEnd(() => {
+          void mockConnection.stop();
+        });
+        allCreatedConnections.push(mockConnection);
+        return mockConnection;
+      },
+    );
+});
+
+afterEach(() => {
+  testScope.end();
+});
+
+describe("connections$ stream", () => {
+  test("Should create and start new connections for each transports", () => {
+    withTestScheduler(({ behavior, expectObservable }) => {
+      const { connections$ } = createConnectionManager$({
+        scope: testScope,
+        connectionFactory: fakeConnectionFactory,
+        inputTransports$: behavior("a", {
+          a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
+        }),
+        logger: logger,
+      });
+
+      expectObservable(connections$).toBe("a", {
+        a: expect.toSatisfy((e: Epoch<Connection[]>) => {
+          const connections = e.value;
+          expect(connections.length).toBe(2);
+
+          expect(
+            vi.mocked(fakeConnectionFactory).createConnection,
+          ).toHaveBeenCalledTimes(2);
+
+          const conn1 = connections.find((c) =>
+            areLivekitTransportsEqual(c.transport, TRANSPORT_1),
+          );
+          expect(conn1).toBeDefined();
+          expect(conn1!.start).toHaveBeenCalled();
+
+          const conn2 = connections.find((c) =>
+            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
+          );
+          expect(conn2).toBeDefined();
+          expect(conn2!.start).toHaveBeenCalled();
+          return true;
+        }),
+      });
+    });
+  });
+
+  test("Should start connection only once", () => {
+    withTestScheduler(({ behavior, expectObservable }) => {
+      const { connections$ } = createConnectionManager$({
+        scope: testScope,
+        connectionFactory: fakeConnectionFactory,
+        inputTransports$: behavior("abcdef", {
+          a: new Epoch([TRANSPORT_1], 0),
+          b: new Epoch([TRANSPORT_1], 1),
+          c: new Epoch([TRANSPORT_1], 2),
+          d: new Epoch([TRANSPORT_1], 3),
+          e: new Epoch([TRANSPORT_1], 4),
+          f: new Epoch([TRANSPORT_1, TRANSPORT_2], 5),
+        }),
+        logger: logger,
+      });
+
+      expectObservable(connections$).toBe("xxxxxa", {
+        x: expect.anything(),
+        a: expect.toSatisfy((e: Epoch<Connection[]>) => {
+          const connections = e.value;
+
+          expect(connections.length).toBe(2);
+          expect(
+            vi.mocked(fakeConnectionFactory).createConnection,
+          ).toHaveBeenCalledTimes(2);
+
+          const conn2 = connections.find((c) =>
+            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
+          );
+          expect(conn2).toBeDefined();
+
+          const conn1 = connections.find((c) =>
+            areLivekitTransportsEqual(c.transport, TRANSPORT_1),
+          );
+          expect(conn1).toBeDefined();
+          expect(conn1!.start).toHaveBeenCalledOnce();
+
+          return true;
+        }),
+      });
+    });
+  });
+
+  test("Should cleanup connections when not needed anymore", () => {
+    withTestScheduler(({ behavior, expectObservable }) => {
+      const { connections$ } = createConnectionManager$({
+        scope: testScope,
+        connectionFactory: fakeConnectionFactory,
+        inputTransports$: behavior("abc", {
+          a: new Epoch([TRANSPORT_1], 0),
+          b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
+          c: new Epoch([TRANSPORT_1], 2),
+        }),
+        logger: logger,
+      });
+
+      expectObservable(connections$).toBe("xab", {
+        x: expect.anything(),
+        a: expect.toSatisfy((e: Epoch<Connection[]>) => {
+          const connections = e.value;
+          expect(connections.length).toBe(2);
+          return true;
+        }),
+        b: expect.toSatisfy((e: Epoch<Connection[]>) => {
+          const connections = e.value;
+
+          expect(connections.length).toBe(1);
+          // The second connection should have been stopped has it is no longer needed.
+          const connection2 = allCreatedConnections.find((c) =>
+            areLivekitTransportsEqual(c.transport, TRANSPORT_2),
+          );
+          expect(connection2).toBeDefined();
+          expect(connection2!.stop).toHaveBeenCalled();
+
+          // The first connection should still be active
+          const conn1 = connections[0];
+          expect(conn1.stop).not.toHaveBeenCalledOnce();
+
+          return true;
+        }),
+      });
+    });
+  });
+});
+
+describe("connectionManagerData$ stream", () => {
+  // Used in test to control fake connections' remoteParticipantsWithTracks$ streams
+  let fakePublishingParticipantsStreams: Map<
+    string,
+    Behavior<LivekitParticipant[]>
+  >;
+
+  function keyForTransport(transport: LivekitTransport): string {
+    return `${transport.livekit_service_url}|${transport.livekit_alias}`;
+  }
+
+  beforeEach(() => {
+    fakePublishingParticipantsStreams = new Map();
+
+    function getPublishingParticipantsFor(
+      transport: LivekitTransport,
+    ): Behavior<LivekitParticipant[]> {
+      return (
+        fakePublishingParticipantsStreams.get(keyForTransport(transport)) ??
+        new BehaviorSubject([])
+      );
+    }
+
+    // need a more advanced fake connection factory
+    vi.mocked(fakeConnectionFactory).createConnection = vi
+      .fn()
+      .mockImplementation(
+        (transport: LivekitTransport, scope: ObservableScope) => {
+          const fakePublishingParticipants$ = new BehaviorSubject<
+            LivekitParticipant[]
+          >([]);
+          const mockConnection = {
+            transport,
+            remoteParticipantsWithTracks$:
+              getPublishingParticipantsFor(transport),
+          } as unknown as Connection;
+          vi.mocked(mockConnection).start = vi.fn();
+          vi.mocked(mockConnection).stop = vi.fn();
+          // Tie the connection's lifecycle to the scope to test scope lifecycle management
+          scope.onEnd(() => {
+            void mockConnection.stop();
+          });
+
+          fakePublishingParticipantsStreams.set(
+            keyForTransport(transport),
+            fakePublishingParticipants$,
+          );
+          return mockConnection;
+        },
+      );
+  });
+
+  test("Should report connections with the publishing participants", () => {
+    withTestScheduler(({ expectObservable, schedule, behavior }) => {
+      // Setup the fake participants streams behavior
+      // ==============================
+      fakePublishingParticipantsStreams.set(
+        keyForTransport(TRANSPORT_1),
+        behavior("oa-b", {
+          o: [],
+          a: [{ identity: "user1A" } as LivekitParticipant],
+          b: [
+            { identity: "user1A" } as LivekitParticipant,
+            { identity: "user1B" } as LivekitParticipant,
+          ],
+        }),
+      );
+
+      fakePublishingParticipantsStreams.set(
+        keyForTransport(TRANSPORT_2),
+        behavior("o-a", {
+          o: [],
+          a: [{ identity: "user2A" } as LivekitParticipant],
+        }),
+      );
+      // ==============================
+
+      const { connectionManagerData$ } = createConnectionManager$({
+        scope: testScope,
+        connectionFactory: fakeConnectionFactory,
+        inputTransports$: behavior("a", {
+          a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
+        }),
+        logger,
+      });
+
+      expectObservable(connectionManagerData$).toBe("abcd", {
+        a: expect.toSatisfy((e) => {
+          const data: ConnectionManagerData = e.value;
+          expect(data.getConnections().length).toBe(2);
+          expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(0);
+          expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(0);
+          return true;
+        }),
+        b: expect.toSatisfy((e) => {
+          const data: ConnectionManagerData = e.value;
+          expect(data.getConnections().length).toBe(2);
+          expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(1);
+          expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(0);
+          expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
+            "user1A",
+          );
+          return true;
+        }),
+        c: expect.toSatisfy((e) => {
+          const data: ConnectionManagerData = e.value;
+          expect(data.getConnections().length).toBe(2);
+          expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(1);
+          expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(1);
+          expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
+            "user1A",
+          );
+          expect(data.getParticipantForTransport(TRANSPORT_2)[0].identity).toBe(
+            "user2A",
+          );
+          return true;
+        }),
+        d: expect.toSatisfy((e) => {
+          const data: ConnectionManagerData = e.value;
+          expect(data.getConnections().length).toBe(2);
+          expect(data.getParticipantForTransport(TRANSPORT_1).length).toBe(2);
+          expect(data.getParticipantForTransport(TRANSPORT_2).length).toBe(1);
+          expect(data.getParticipantForTransport(TRANSPORT_1)[0].identity).toBe(
+            "user1A",
+          );
+          expect(data.getParticipantForTransport(TRANSPORT_1)[1].identity).toBe(
+            "user1B",
+          );
+          expect(data.getParticipantForTransport(TRANSPORT_2)[0].identity).toBe(
+            "user2A",
+          );
+          return true;
+        }),
+      });
+    });
+  });
+});

src/state/CallViewModel/remoteMembers/ConnectionManager.ts

@@ -13,15 +13,15 @@ import {
   type LivekitTransport,
   type ParticipantId,
 } from "matrix-js-sdk/lib/matrixrtc";
-import { BehaviorSubject, combineLatest, map, switchMap } from "rxjs";
-import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
+import { BehaviorSubject, combineLatest, map, of, switchMap, tap } from "rxjs";
+import { type Logger } from "matrix-js-sdk/lib/logger";
 import { type LocalParticipant, type RemoteParticipant } from "livekit-client";
 
-import { type Behavior } from "../Behavior";
-import { type Connection } from "./Connection";
-import { type ObservableScope } from "../ObservableScope";
-import { generateKeyed$ } from "../../utils/observable";
-import { areLivekitTransportsEqual } from "./matrixLivekitMerger";
+import { type Behavior } from "../../Behavior.ts";
+import { type Connection } from "./Connection.ts";
+import { Epoch, type ObservableScope } from "../../ObservableScope.ts";
+import { generateItemsWithEpoch } from "../../../utils/observable.ts";
+import { areLivekitTransportsEqual } from "./MatrixLivekitMembers.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
 
 export class ConnectionManagerData {
@@ -55,8 +55,8 @@ export class ConnectionManagerData {
 
   public getConnectionForTransport(
     transport: LivekitTransport,
-  ): Connection | undefined {
-    return this.store.get(this.getKey(transport))?.[0];
+  ): Connection | null {
+    return this.store.get(this.getKey(transport))?.[0] ?? null;
   }
 
   public getParticipantForTransport(
@@ -90,10 +90,15 @@ export class ConnectionManagerData {
 interface Props {
   scope: ObservableScope;
   connectionFactory: ConnectionFactory;
-  inputTransports$: Behavior<LivekitTransport[]>;
+  inputTransports$: Behavior<Epoch<LivekitTransport[]>>;
+  logger: Logger;
 }
 // TODO - write test for scopes (do we really need to bind scope)
-
+export interface IConnectionManager {
+  transports$: Behavior<Epoch<LivekitTransport[]>>;
+  connectionManagerData$: Behavior<Epoch<ConnectionManagerData>>;
+  connections$: Behavior<Epoch<Connection[]>>;
+}
 /**
  * Crete a `ConnectionManager`
  * @param scope the observable scope used by this object.
@@ -112,12 +117,9 @@ export function createConnectionManager$({
   scope,
   connectionFactory,
   inputTransports$,
-}: Props): {
-  transports$: Behavior<LivekitTransport[]>;
-  connectionManagerData$: Behavior<ConnectionManagerData>;
-  connections$: Behavior<Connection[]>;
-} {
-  const logger = rootLogger.getChild("ConnectionManager");
+  logger: parentLogger,
+}: Props): IConnectionManager {
+  const logger = parentLogger.getChild("[ConnectionManager]");
 
   const running$ = new BehaviorSubject(true);
   scope.onEnd(() => running$.next(false));
@@ -133,8 +135,15 @@ export function createConnectionManager$({
    */
   const transports$ = scope.behavior(
     combineLatest([running$, inputTransports$]).pipe(
-      map(([running, transports]) => (running ? transports : [])),
-      map(removeDuplicateTransports),
+      map(([running, transports]) =>
+        transports.mapInner((transport) => (running ? transport : [])),
+      ),
+      map((transports) => transports.mapInner(removeDuplicateTransports)),
+      tap(({ value: transports }) => {
+        logger.trace(
+          `Managing transports: ${transports.map((t) => t.livekit_service_url).join(", ")}`,
+        );
+      }),
     ),
   );
 
@@ -142,66 +151,75 @@ export function createConnectionManager$({
    * Connections for each transport in use by one or more session members.
    */
   const connections$ = scope.behavior(
-    generateKeyed$<LivekitTransport[], Connection, Connection[]>(
-      transports$,
-      (transports, createOrGet) => {
-        const createConnection =
-          (
-            transport: LivekitTransport,
-          ): ((scope: ObservableScope) => Connection) =>
-          (scope) => {
-            const connection = connectionFactory.createConnection(
-              transport,
-              scope,
-              logger,
-            );
-            // Start the connection immediately
-            // Use connection state to track connection progress
-            void connection.start();
-            // TODO subscribe to connection state to retry or log issues?
-            return connection;
-          };
-
-        return transports.map((transport) => {
-          const key =
-            transport.livekit_service_url + "|" + transport.livekit_alias;
-          return createOrGet(key, createConnection(transport));
-        });
-      },
+    transports$.pipe(
+      generateItemsWithEpoch(
+        function* (transports) {
+          for (const transport of transports)
+            yield {
+              keys: [transport.livekit_service_url, transport.livekit_alias],
+              data: undefined,
+            };
+        },
+        (scope, _data$, serviceUrl, alias) => {
+          logger.debug(`Creating connection to ${serviceUrl} (${alias})`);
+          const connection = connectionFactory.createConnection(
+            {
+              type: "livekit",
+              livekit_service_url: serviceUrl,
+              livekit_alias: alias,
+            },
+            scope,
+            logger,
+          );
+          // Start the connection immediately
+          // Use connection state to track connection progress
+          void connection.start();
+          // TODO subscribe to connection state to retry or log issues?
+          return connection;
+        },
+      ),
     ),
   );
 
-  const connectionManagerData$: Behavior<ConnectionManagerData> =
-    scope.behavior(
-      connections$.pipe(
-        switchMap((connections) => {
-          // Map the connections to list of {connection, participants}[]
-          const listOfConnectionsWithPublishingParticipants = connections.map(
-            (connection) => {
-              return connection.participantsWithTrack$.pipe(
-                map((participants) => ({
-                  connection,
-                  participants,
-                })),
-              );
-            },
-          );
-          // combineLatest the several streams into a single stream with the ConnectionManagerData
-          return combineLatest(
-            listOfConnectionsWithPublishingParticipants,
-          ).pipe(
-            map((lists) =>
-              lists.reduce((data, { connection, participants }) => {
-                data.add(connection, participants);
-                return data;
-              }, new ConnectionManagerData()),
-            ),
-          );
-        }),
-      ),
-      // start empty
-      new ConnectionManagerData(),
-    );
+  const connectionManagerData$ = scope.behavior(
+    connections$.pipe(
+      switchMap((connections) => {
+        const epoch = connections.epoch;
+
+        // Map the connections to list of {connection, participants}[]
+        const listOfConnectionsWithPublishingParticipants =
+          connections.value.map((connection) => {
+            return connection.remoteParticipantsWithTracks$.pipe(
+              map((participants) => ({
+                connection,
+                participants,
+              })),
+            );
+          });
+
+        // probably not required
+        if (listOfConnectionsWithPublishingParticipants.length === 0) {
+          return of(new Epoch(new ConnectionManagerData(), epoch));
+        }
+
+        // combineLatest the several streams into a single stream with the ConnectionManagerData
+        return combineLatest(listOfConnectionsWithPublishingParticipants).pipe(
+          map(
+            (lists) =>
+              new Epoch(
+                lists.reduce((data, { connection, participants }) => {
+                  data.add(connection, participants);
+                  return data;
+                }, new ConnectionManagerData()),
+                epoch,
+              ),
+          ),
+        );
+      }),
+    ),
+    new Epoch(new ConnectionManagerData()),
+  );
+
   return { transports$, connectionManagerData$, connections$ };
 }
 

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

@@ -0,0 +1,384 @@
+/*
+Copyright 2025 Element Creations Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { describe, test, expect, beforeEach, afterEach } from "vitest";
+import {
+  type CallMembership,
+  type LivekitTransport,
+} from "matrix-js-sdk/lib/matrixrtc";
+import { getParticipantId } from "matrix-js-sdk/lib/matrixrtc/utils";
+import { combineLatest, map, type Observable } from "rxjs";
+
+import { type IConnectionManager } from "./ConnectionManager.ts";
+import {
+  type MatrixLivekitMember,
+  createMatrixLivekitMembers$,
+} from "./MatrixLivekitMembers.ts";
+import {
+  Epoch,
+  mapEpoch,
+  ObservableScope,
+  trackEpoch,
+} from "../../ObservableScope.ts";
+import { ConnectionManagerData } from "./ConnectionManager.ts";
+import {
+  mockCallMembership,
+  mockRemoteParticipant,
+  withTestScheduler,
+} from "../../../utils/test.ts";
+import { type Connection } from "./Connection.ts";
+
+let testScope: ObservableScope;
+
+const transportA: LivekitTransport = {
+  type: "livekit",
+  livekit_service_url: "https://lk.example.org",
+  livekit_alias: "!alias:example.org",
+};
+
+const transportB: LivekitTransport = {
+  type: "livekit",
+  livekit_service_url: "https://lk.sample.com",
+  livekit_alias: "!alias:sample.com",
+};
+
+const bobMembership = mockCallMembership(
+  "@bob:example.org",
+  "DEV000",
+  transportA,
+);
+const carlMembership = mockCallMembership(
+  "@carl:sample.com",
+  "DEV111",
+  transportB,
+);
+
+beforeEach(() => {
+  testScope = new ObservableScope();
+});
+
+afterEach(() => {
+  testScope.end();
+});
+
+function epochMeWith$<T, U>(
+  source$: Observable<Epoch<U>>,
+  me$: Observable<T>,
+): Observable<Epoch<T>> {
+  return combineLatest([source$, me$]).pipe(
+    map(([ep, cd]) => {
+      return new Epoch(cd, ep.epoch);
+    }),
+  );
+}
+
+test("should signal participant not yet connected to livekit", () => {
+  withTestScheduler(({ behavior, expectObservable }) => {
+    const { memberships$, membershipsWithTransport$ } = fromMemberships$(
+      behavior("a", {
+        a: [bobMembership],
+      }),
+    );
+
+    const connectionManagerData$ = epochMeWith$(
+      memberships$,
+      behavior("a", {
+        a: new ConnectionManagerData(),
+      }),
+    );
+
+    const matrixLivekitMember$ = createMatrixLivekitMembers$({
+      scope: testScope,
+      membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
+      connectionManager: {
+        connectionManagerData$: connectionManagerData$,
+      } as unknown as IConnectionManager,
+    });
+
+    expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
+      a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
+        expect(data.length).toEqual(1);
+        expectObservable(data[0].membership$).toBe("a", {
+          a: bobMembership,
+        });
+        expectObservable(data[0].participant$).toBe("a", {
+          a: null,
+        });
+        expectObservable(data[0].connection$).toBe("a", {
+          a: null,
+        });
+        return true;
+      }),
+    });
+  });
+});
+
+// Helper to create epoch'ed memberships$ and membershipsWithTransport$ from memberships observable.
+function fromMemberships$(m$: Observable<CallMembership[]>): {
+  memberships$: Observable<Epoch<CallMembership[]>>;
+  membershipsWithTransport$: Observable<
+    Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
+  >;
+} {
+  const memberships$ = m$.pipe(trackEpoch());
+  const membershipsWithTransport$ = memberships$.pipe(
+    mapEpoch((members) => {
+      return members.map((m) => {
+        const tr = m.getTransport(m);
+        return {
+          membership: m,
+          transport:
+            tr?.type === "livekit" ? (tr as LivekitTransport) : undefined,
+        };
+      });
+    }),
+  );
+  return {
+    memberships$,
+    membershipsWithTransport$,
+  };
+}
+
+test("should signal participant on a connection that is publishing", () => {
+  withTestScheduler(({ behavior, expectObservable }) => {
+    const bobParticipantId = getParticipantId(
+      bobMembership.userId,
+      bobMembership.deviceId,
+    );
+
+    const { memberships$, membershipsWithTransport$ } = fromMemberships$(
+      behavior("a", {
+        a: [bobMembership],
+      }),
+    );
+
+    const connection = {
+      transport: bobMembership.getTransport(bobMembership),
+    } as unknown as Connection;
+    const dataWithPublisher = new ConnectionManagerData();
+    dataWithPublisher.add(connection, [
+      mockRemoteParticipant({ identity: bobParticipantId }),
+    ]);
+
+    const connectionManagerData$ = epochMeWith$(
+      memberships$,
+      behavior("a", {
+        a: dataWithPublisher,
+      }),
+    );
+
+    const matrixLivekitMember$ = createMatrixLivekitMembers$({
+      scope: testScope,
+      membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
+      connectionManager: {
+        connectionManagerData$: connectionManagerData$,
+      } as unknown as IConnectionManager,
+    });
+
+    expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
+      a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
+        expect(data.length).toEqual(1);
+        expectObservable(data[0].membership$).toBe("a", {
+          a: bobMembership,
+        });
+        expectObservable(data[0].participant$).toBe("a", {
+          a: expect.toSatisfy((participant) => {
+            expect(participant).toBeDefined();
+            expect(participant!.identity).toEqual(bobParticipantId);
+            return true;
+          }),
+        });
+        expectObservable(data[0].connection$).toBe("a", {
+          a: connection,
+        });
+        return true;
+      }),
+    });
+  });
+});
+
+test("should signal participant on a connection that is not publishing", () => {
+  withTestScheduler(({ behavior, expectObservable }) => {
+    const { memberships$, membershipsWithTransport$ } = fromMemberships$(
+      behavior("a", {
+        a: [bobMembership],
+      }),
+    );
+
+    const connection = {
+      transport: bobMembership.getTransport(bobMembership),
+    } as unknown as Connection;
+    const dataWithPublisher = new ConnectionManagerData();
+    dataWithPublisher.add(connection, []);
+
+    const connectionManagerData$ = epochMeWith$(
+      memberships$,
+      behavior("a", {
+        a: dataWithPublisher,
+      }),
+    );
+
+    const matrixLivekitMember$ = createMatrixLivekitMembers$({
+      scope: testScope,
+      membershipsWithTransport$: testScope.behavior(membershipsWithTransport$),
+      connectionManager: {
+        connectionManagerData$: connectionManagerData$,
+      } as unknown as IConnectionManager,
+    });
+
+    expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe("a", {
+      a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
+        expect(data.length).toEqual(1);
+        expectObservable(data[0].membership$).toBe("a", {
+          a: bobMembership,
+        });
+        expectObservable(data[0].participant$).toBe("a", {
+          a: null,
+        });
+        expectObservable(data[0].connection$).toBe("a", {
+          a: connection,
+        });
+        return true;
+      }),
+    });
+  });
+});
+
+describe("Publication edge case", () => {
+  test("bob is publishing in several connections", () => {
+    withTestScheduler(({ behavior, expectObservable }) => {
+      const { memberships$, membershipsWithTransport$ } = fromMemberships$(
+        behavior("a", {
+          a: [bobMembership, carlMembership],
+        }),
+      );
+
+      const connectionWithPublisher = new ConnectionManagerData();
+      const bobParticipantId = getParticipantId(
+        bobMembership.userId,
+        bobMembership.deviceId,
+      );
+      const connectionA = {
+        transport: transportA,
+      } as unknown as Connection;
+      const connectionB = {
+        transport: transportB,
+      } as unknown as Connection;
+
+      connectionWithPublisher.add(connectionA, [
+        mockRemoteParticipant({ identity: bobParticipantId }),
+      ]);
+      connectionWithPublisher.add(connectionB, [
+        mockRemoteParticipant({ identity: bobParticipantId }),
+      ]);
+
+      const connectionManagerData$ = epochMeWith$(
+        memberships$,
+        behavior("a", {
+          a: connectionWithPublisher,
+        }),
+      );
+
+      const matrixLivekitMember$ = createMatrixLivekitMembers$({
+        scope: testScope,
+        membershipsWithTransport$: testScope.behavior(
+          membershipsWithTransport$,
+        ),
+        connectionManager: {
+          connectionManagerData$: connectionManagerData$,
+        } as unknown as IConnectionManager,
+      });
+
+      expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe(
+        "a",
+        {
+          a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
+            expect(data.length).toEqual(2);
+            expectObservable(data[0].membership$).toBe("a", {
+              a: bobMembership,
+            });
+            expectObservable(data[0].connection$).toBe("a", {
+              // The real connection should be from transportA as per the membership
+              a: connectionA,
+            });
+            expectObservable(data[0].participant$).toBe("a", {
+              a: expect.toSatisfy((participant) => {
+                expect(participant).toBeDefined();
+                expect(participant!.identity).toEqual(bobParticipantId);
+                return true;
+              }),
+            });
+            return true;
+          }),
+        },
+      );
+    });
+  });
+
+  test("bob is publishing in the wrong connection", () => {
+    withTestScheduler(({ behavior, expectObservable }) => {
+      const { memberships$, membershipsWithTransport$ } = fromMemberships$(
+        behavior("a", {
+          a: [bobMembership, carlMembership],
+        }),
+      );
+
+      const connectionWithPublisher = new ConnectionManagerData();
+      const bobParticipantId = getParticipantId(
+        bobMembership.userId,
+        bobMembership.deviceId,
+      );
+      const connectionA = { transport: transportA } as unknown as Connection;
+      const connectionB = { transport: transportB } as unknown as Connection;
+
+      // Bob is not publishing on A
+      connectionWithPublisher.add(connectionA, []);
+      // Bob is publishing on B but his membership says A
+      connectionWithPublisher.add(connectionB, [
+        mockRemoteParticipant({ identity: bobParticipantId }),
+      ]);
+
+      const connectionManagerData$ = epochMeWith$(
+        memberships$,
+        behavior("a", {
+          a: connectionWithPublisher,
+        }),
+      );
+
+      const matrixLivekitMember$ = createMatrixLivekitMembers$({
+        scope: testScope,
+        membershipsWithTransport$: testScope.behavior(
+          membershipsWithTransport$,
+        ),
+        connectionManager: {
+          connectionManagerData$: connectionManagerData$,
+        } as unknown as IConnectionManager,
+      });
+
+      expectObservable(matrixLivekitMember$.pipe(map((e) => e.value))).toBe(
+        "a",
+        {
+          a: expect.toSatisfy((data: MatrixLivekitMember[]) => {
+            expect(data.length).toEqual(2);
+            expectObservable(data[0].membership$).toBe("a", {
+              a: bobMembership,
+            });
+            expectObservable(data[0].connection$).toBe("a", {
+              // The real connection should be from transportA as per the membership
+              a: connectionA,
+            });
+            expectObservable(data[0].participant$).toBe("a", {
+              // No participant as Bob is not publishing on his membership transport
+              a: null,
+            });
+            return true;
+          }),
+        },
+      );
+    });
+  });
+});

src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts

@@ -0,0 +1,138 @@
+/*
+Copyright 2025 Element Creations Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import {
+  type LocalParticipant as LocalLivekitParticipant,
+  type RemoteParticipant as RemoteLivekitParticipant,
+} from "livekit-client";
+import {
+  type LivekitTransport,
+  type CallMembership,
+} from "matrix-js-sdk/lib/matrixrtc";
+import { combineLatest, filter, map } from "rxjs";
+import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
+
+import { type Behavior } from "../../Behavior";
+import { type IConnectionManager } from "./ConnectionManager";
+import { Epoch, type ObservableScope } from "../../ObservableScope";
+import { type Connection } from "./Connection";
+import { generateItemsWithEpoch } from "../../../utils/observable";
+
+const logger = rootLogger.getChild("[MatrixLivekitMembers]");
+
+/**
+ * Represents a Matrix call member and their associated LiveKit participation.
+ * `livekitParticipant` can be undefined if the member is not yet connected to the livekit room
+ * or if it has no livekit transport at all.
+ */
+export interface MatrixLivekitMember {
+  membership$: Behavior<CallMembership>;
+  participant$: Behavior<
+    LocalLivekitParticipant | RemoteLivekitParticipant | null
+  >;
+  connection$: Behavior<Connection | null>;
+  // participantId: string; We do not want a participantId here since it will be generated by the jwt
+  // TODO decide if we can also drop the userId. Its in the matrix membership anyways.
+  userId: string;
+}
+
+interface Props {
+  scope: ObservableScope;
+  membershipsWithTransport$: Behavior<
+    Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
+  >;
+  connectionManager: IConnectionManager;
+}
+/**
+ * Combines MatrixRTC and Livekit worlds.
+ *
+ * It has a small public interface:
+ *  - in (via constructor):
+ *    - an observable of CallMembership[] to track the call members (The matrix side)
+ *    - a `ConnectionManager` for the lk rooms (The livekit side)
+ *  - out (via public Observable):
+ *    - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data.
+ */
+export function createMatrixLivekitMembers$({
+  scope,
+  membershipsWithTransport$,
+  connectionManager,
+}: Props): Behavior<Epoch<MatrixLivekitMember[]>> {
+  /**
+   * Stream of all the call members and their associated livekit data (if available).
+   */
+
+  return scope.behavior(
+    combineLatest([
+      membershipsWithTransport$,
+      connectionManager.connectionManagerData$,
+    ]).pipe(
+      filter((values) =>
+        values.every((value) => value.epoch === values[0].epoch),
+      ),
+      map(
+        ([
+          { value: membershipsWithTransports, epoch },
+          { value: managerData },
+        ]) =>
+          new Epoch([membershipsWithTransports, managerData] as const, epoch),
+      ),
+      generateItemsWithEpoch(
+        // Generator function.
+        // creates an array of `{key, data}[]`
+        // Each change in the keys (new key, missing key) will result in a call to the factory function.
+        function* ([membershipsWithTransports, managerData]) {
+          for (const { membership, transport } of membershipsWithTransports) {
+            // TODO! cannot use membership.membershipID yet, Currently its hardcoded by the jwt service to
+            const participantId = /*membership.membershipID*/ `${membership.userId}:${membership.deviceId}`;
+
+            const participants = transport
+              ? managerData.getParticipantForTransport(transport)
+              : [];
+            const participant =
+              participants.find((p) => p.identity == participantId) ?? null;
+            const connection = transport
+              ? managerData.getConnectionForTransport(transport)
+              : null;
+
+            yield {
+              keys: [participantId, membership.userId],
+              data: { membership, participant, connection },
+            };
+          }
+        },
+        // Each update where the key of the generator array do not change will result in updates to the `data$` observable in the factory.
+        (scope, data$, participantId, userId) => {
+          logger.debug(
+            `Updating data$ for participantId: ${participantId}, userId: ${userId}`,
+          );
+          // will only get called once per `participantId, userId` pair.
+          // updates to data$ and as a result to displayName$ and mxcAvatarUrl$ are more frequent.
+          return {
+            participantId,
+            userId,
+            ...scope.splitBehavior(data$),
+          };
+        },
+      ),
+    ),
+  );
+}
+
+// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
+
+// TODO add this to the JS-SDK
+export function areLivekitTransportsEqual(
+  t1: LivekitTransport | null,
+  t2: LivekitTransport | null,
+): boolean {
+  if (t1 && t2) return t1.livekit_service_url === t2.livekit_service_url;
+  // In case we have different lk rooms in the same SFU (depends on the livekit authorization service)
+  // It is only needed in case the livekit authorization service is not behaving as expected (or custom implementation)
+  if (!t1 && !t2) return true;
+  return false;
+}

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

@@ -0,0 +1,611 @@
+/*
+Copyright 2025 Element Creations Ltd.
+
+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, vi } from "vitest";
+import {
+  type MatrixEvent,
+  type RoomMember,
+  type RoomState,
+  RoomStateEvent,
+} from "matrix-js-sdk";
+import EventEmitter from "events";
+import { it } from "vitest";
+
+import { ObservableScope } from "../../ObservableScope.ts";
+import type { Room as MatrixRoom } from "matrix-js-sdk/lib/models/room";
+import {
+  mockCallMembership,
+  mockMatrixRoomMember,
+  withTestScheduler,
+} from "../../../utils/test.ts";
+import {
+  createMatrixMemberMetadata$,
+  createRoomMembers$,
+} from "./MatrixMemberMetadata.ts";
+let testScope: ObservableScope;
+let mockMatrixRoom: MatrixRoom;
+
+describe("MatrixMemberMetadata", () => {
+  /*
+   * To be populated in the test setup.
+   * Maps userId to a partial/mock RoomMember object.
+   */
+  let fakeMembersMap: Map<string, Partial<RoomMember>>;
+
+  beforeEach(() => {
+    testScope = new ObservableScope();
+    fakeMembersMap = new Map<string, Partial<RoomMember>>();
+
+    const roomEmitter = new EventEmitter();
+    mockMatrixRoom = {
+      on: roomEmitter.on.bind(roomEmitter),
+      off: roomEmitter.off.bind(roomEmitter),
+      emit: roomEmitter.emit.bind(roomEmitter),
+      // addListener: roomEmitter.addListener.bind(roomEmitter),
+      // removeListener: roomEmitter.removeListener.bind(roomEmitter),
+      getMember: vi.fn().mockImplementation((userId: string) => {
+        const member = fakeMembersMap.get(userId);
+        if (member) {
+          return member as RoomMember;
+        }
+        return null;
+      }),
+      getMembers: vi.fn().mockImplementation(() => {
+        const members = Array.from(fakeMembersMap.values());
+        return members;
+      }),
+    } as unknown as MatrixRoom;
+  });
+
+  function fakeMemberWith(data: Partial<RoomMember>): void {
+    const userId = data.userId || "@alice:example.com";
+    const member: Partial<RoomMember> = {
+      userId: userId,
+      rawDisplayName: data.rawDisplayName ?? userId,
+      getMxcAvatarUrl:
+        data.getMxcAvatarUrl ||
+        vi.fn().mockImplementation(() => {
+          return `mxc://example.com/${userId}`;
+        }),
+      ...data,
+    } as unknown as RoomMember;
+    fakeMembersMap.set(userId, member);
+  }
+
+  afterEach(() => {
+    fakeMembersMap.clear();
+  });
+
+  describe("displayname", () => {
+    function updateDisplayName(
+      userId: `@${string}:${string}`,
+      newDisplayName: string,
+    ): void {
+      const member = fakeMembersMap.get(userId);
+      if (member) {
+        member.rawDisplayName = newDisplayName;
+        // Emit the event to notify listeners
+        mockMatrixRoom.emit(
+          RoomStateEvent.Members,
+          {} as unknown as MatrixEvent,
+          {} as unknown as RoomState,
+          member as RoomMember,
+        );
+      } else {
+        throw new Error(`No member found with userId: ${userId}`);
+      }
+    }
+
+    // TODO this is a regression, now there the own user is not always in the map. Ask Timo if fine
+    it("should show our own user if present in rtc session and room", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        fakeMemberWith({
+          userId: "@local:example.com",
+          rawDisplayName: "it's a me",
+        });
+        const memberships$ = behavior("a", {
+          a: [mockCallMembership("@local:example.com", "DEVICE1")],
+        });
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+        const dn$ =
+          metadataStore.createDisplayNameBehavior$("@local:example.com");
+
+        expectObservable(dn$).toBe("a", {
+          a: "it's a me",
+        });
+        expectObservable(metadataStore.displaynameMap$).toBe("a", {
+          a: new Map<string, string>([["@local:example.com", "it's a me"]]),
+        });
+      });
+    });
+
+    function setUpBasicRoom(): void {
+      fakeMemberWith({
+        userId: "@local:example.com",
+        rawDisplayName: "it's a me",
+      });
+      fakeMemberWith({ userId: "@alice:example.com", rawDisplayName: "Alice" });
+      fakeMemberWith({ userId: "@bob:example.com", rawDisplayName: "Bob" });
+      fakeMemberWith({ userId: "@carl:example.com", rawDisplayName: "Carl" });
+      fakeMemberWith({ userId: "@evil:example.com", rawDisplayName: "Carl" });
+      fakeMemberWith({ userId: "@bob:foo.bar", rawDisplayName: "Bob" });
+      fakeMemberWith({ userId: "@no-name:foo.bar" });
+    }
+
+    it("should get displayName for users", () => {
+      setUpBasicRoom();
+
+      withTestScheduler(({ behavior, expectObservable }) => {
+        const memberships$ = behavior("a", {
+          a: [
+            mockCallMembership("@alice:example.com", "DEVICE1"),
+            mockCallMembership("@bob:example.com", "DEVICE1"),
+          ],
+        });
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+        const aliceDispName$ =
+          metadataStore.createDisplayNameBehavior$("@alice:example.com");
+
+        expectObservable(aliceDispName$).toBe("a", {
+          a: "Alice",
+        });
+
+        expectObservable(metadataStore.displaynameMap$).toBe("a", {
+          a: new Map<string, string>([
+            ["@alice:example.com", "Alice"],
+            ["@bob:example.com", "Bob"],
+          ]),
+        });
+      });
+    });
+
+    it("should use userId if no display name", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        setUpBasicRoom();
+
+        const memberships$ = behavior("a", {
+          a: [mockCallMembership("@no-name:foo.bar", "D000")],
+        });
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        expectObservable(metadataStore.displaynameMap$).toBe("a", {
+          a: new Map<string, string>([
+            ["@no-name:foo.bar", "@no-name:foo.bar"],
+          ]),
+        });
+      });
+    });
+
+    it("should disambiguate users with same display name", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        setUpBasicRoom();
+
+        const memberships$ = behavior("a", {
+          a: [
+            mockCallMembership("@bob:example.com", "DEVICE1"),
+            mockCallMembership("@bob:example.com", "DEVICE2"),
+            mockCallMembership("@bob:foo.bar", "BOB000"),
+            mockCallMembership("@carl:example.com", "C000"),
+            mockCallMembership("@evil:example.com", "E000"),
+          ],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        expectObservable(metadataStore.displaynameMap$).toBe("a", {
+          a: new Map<string, string>([
+            // ["@local:example.com", "it's a me"],
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+            ["@bob:foo.bar", "Bob (@bob:foo.bar)"],
+            ["@carl:example.com", "Carl (@carl:example.com)"],
+            ["@evil:example.com", "Carl (@evil:example.com)"],
+          ]),
+        });
+      });
+    });
+
+    it("should start to disambiguate reactivly when needed", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        setUpBasicRoom();
+
+        const memberships$ = behavior("ab", {
+          a: [mockCallMembership("@bob:example.com", "DEVICE1")],
+          b: [
+            mockCallMembership("@bob:example.com", "DEVICE1"),
+            mockCallMembership("@bob:foo.bar", "BOB000"),
+          ],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        expectObservable(metadataStore.displaynameMap$).toBe("ab", {
+          a: new Map<string, string>([["@bob:example.com", "Bob"]]),
+          b: new Map<string, string>([
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+            ["@bob:foo.bar", "Bob (@bob:foo.bar)"],
+          ]),
+        });
+      });
+    });
+
+    it("should keep disambiguated name when other leave", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        setUpBasicRoom();
+
+        const memberships$ = behavior("ab", {
+          a: [
+            mockCallMembership("@bob:example.com", "DEVICE1"),
+            mockCallMembership("@bob:foo.bar", "BOB000"),
+          ],
+          b: [mockCallMembership("@bob:example.com", "DEVICE1")],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        expectObservable(metadataStore.displaynameMap$).toBe("ab", {
+          a: new Map<string, string>([
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+            ["@bob:foo.bar", "Bob (@bob:foo.bar)"],
+          ]),
+          b: new Map<string, string>([
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+          ]),
+        });
+      });
+    });
+
+    it("should disambiguate on name change", () => {
+      withTestScheduler(({ behavior, schedule, expectObservable }) => {
+        setUpBasicRoom();
+
+        const memberships$ = behavior("a", {
+          a: [
+            mockCallMembership("@bob:example.com", "B000"),
+            mockCallMembership("@carl:example.com", "C000"),
+          ],
+        });
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        schedule("-a", {
+          a: () => {
+            updateDisplayName("@carl:example.com", "Bob");
+          },
+        });
+
+        expectObservable(metadataStore.displaynameMap$).toBe("ab", {
+          a: new Map<string, string>([
+            ["@bob:example.com", "Bob"],
+            ["@carl:example.com", "Carl"],
+          ]),
+          b: new Map<string, string>([
+            ["@bob:example.com", "Bob (@bob:example.com)"],
+            ["@carl:example.com", "Bob (@carl:example.com)"],
+          ]),
+        });
+      });
+    });
+
+    it("should track individual member id with createDisplayNameBehavior", () => {
+      withTestScheduler(({ behavior, schedule, expectObservable }) => {
+        setUpBasicRoom();
+        const BOB = "@bob:example.com";
+        const CARL = "@carl:example.com";
+        // for this test we build a mock environment that does all possible changes:
+        // - memberships join/leave
+        // - room join/leave
+        // - disambiguate
+        const memberships$ = behavior("ab-d", {
+          a: [mockCallMembership(CARL, "C000")],
+          b: [
+            mockCallMembership(CARL, "C000"),
+            // bob joins
+            mockCallMembership(BOB, "B000"),
+          ],
+          // c carl gets renamed to BOB
+          d: [
+            // carl leaves
+            mockCallMembership(BOB, "B000"),
+          ],
+        });
+        schedule("--a-", {
+          a: () => {
+            // carl renames
+            updateDisplayName(CARL, "Bob");
+          },
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        const bob$ = metadataStore.createDisplayNameBehavior$(BOB);
+        const carl$ = metadataStore.createDisplayNameBehavior$(CARL);
+
+        expectObservable(bob$).toBe("abc-", {
+          a: undefined,
+          b: "Bob",
+          c: "Bob (@bob:example.com)",
+          // bob stays disambiguate even though carl left
+          // d: "Bob (@bob:example.com)",
+        });
+
+        expectObservable(carl$).toBe("a-cd", {
+          a: "Carl",
+          // b: "Carl",
+          // carl gets renamed and disambiguate
+          c: "Bob (@carl:example.com)",
+          d: undefined,
+        });
+      });
+    });
+
+    it("should disambiguate users with invisible characters", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        const bobRtcMember = mockCallMembership("@bob:example.org", "BBBB");
+        const bobZeroWidthSpaceRtcMember = mockCallMembership(
+          "@bob2:example.org",
+          "BBBB",
+        );
+        const bob = mockMatrixRoomMember(bobRtcMember, {
+          rawDisplayName: "Bob",
+        });
+        const bobZeroWidthSpace = mockMatrixRoomMember(
+          bobZeroWidthSpaceRtcMember,
+          {
+            rawDisplayName: "Bo\u200bb",
+          },
+        );
+        fakeMemberWith(bob);
+        fakeMemberWith(bobZeroWidthSpace);
+        fakeMemberWith({ userId: "@carol:example.org" });
+        const memberships$ = behavior("ab", {
+          a: [mockCallMembership("@carol:example.org", "1111"), bobRtcMember],
+          b: [
+            mockCallMembership("@carol:example.org", "1111"),
+            bobRtcMember,
+            bobZeroWidthSpaceRtcMember,
+          ],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        const bob$ =
+          metadataStore.createDisplayNameBehavior$("@bob:example.org");
+        const bob2$ =
+          metadataStore.createDisplayNameBehavior$("@bob2:example.org");
+        const carol$ =
+          metadataStore.createDisplayNameBehavior$("@carol:example.org");
+        expectObservable(bob$).toBe("ab", {
+          a: "Bob",
+          b: "Bob (@bob:example.org)",
+        });
+        expectObservable(bob2$).toBe("ab", {
+          a: undefined,
+          b: "Bo\u200bb (@bob2:example.org)",
+        });
+        expectObservable(carol$).toBe("a-", {
+          a: "@carol:example.org",
+        });
+
+        expectObservable(metadataStore.displaynameMap$).toBe("ab", {
+          // Carol has no displayname - So userId is used.
+          a: new Map([
+            ["@carol:example.org", "@carol:example.org"],
+            ["@bob:example.org", "Bob"],
+          ]),
+          // Other Bob joins, and should handle zero width hacks.
+          b: new Map([
+            ["@carol:example.org", "@carol:example.org"],
+            [bobRtcMember.userId, `Bob (@bob:example.org)`],
+            [
+              bobZeroWidthSpace.userId,
+              `${bobZeroWidthSpace.rawDisplayName} (${bobZeroWidthSpace.userId})`,
+            ],
+          ]),
+        });
+      });
+    });
+
+    it("should strip RTL characters from displayname", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        const daveRtcMember = mockCallMembership("@dave:example.org", "DDDD");
+        const daveRTLRtcMember = mockCallMembership(
+          "@dave2:example.org",
+          "DDDD",
+        );
+        const dave = mockMatrixRoomMember(daveRtcMember, {
+          rawDisplayName: "Dave",
+        });
+        const daveRTL = mockMatrixRoomMember(daveRTLRtcMember, {
+          rawDisplayName: "\u202eevaD",
+        });
+
+        fakeMemberWith({ userId: "@carol:example.org" });
+        fakeMemberWith(daveRTL);
+        fakeMemberWith(dave);
+        const memberships$ = behavior("ab", {
+          a: [mockCallMembership("@carol:example.org", "DDDD")],
+          b: [
+            mockCallMembership("@carol:example.org", "DDDD"),
+            daveRtcMember,
+            daveRTLRtcMember,
+          ],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        expectObservable(metadataStore.displaynameMap$).toBe("ab", {
+          // Carol has no displayname - So userId is used.
+          a: new Map([["@carol:example.org", "@carol:example.org"]]),
+          // Both Dave's join. Since after stripping
+          b: new Map([
+            ["@carol:example.org", "@carol:example.org"],
+            // Not disambiguated
+            ["@dave:example.org", "Dave"],
+            // This one is, since it's using RTL.
+            ["@dave2:example.org", "evaD (@dave2:example.org)"],
+          ]),
+        });
+      });
+    });
+  });
+
+  describe("avatarUrl", () => {
+    function updateAvatarUrl(
+      userId: `@${string}:${string}`,
+      avatarUrl: string,
+    ): void {
+      const member = fakeMembersMap.get(userId);
+      if (member) {
+        member.getMxcAvatarUrl = vi.fn().mockReturnValue(avatarUrl);
+        // Emit the event to notify listeners
+        mockMatrixRoom.emit(
+          RoomStateEvent.Members,
+          {} as unknown as MatrixEvent,
+          {} as unknown as RoomState,
+          member as RoomMember,
+        );
+      } else {
+        throw new Error(`No member found with userId: ${userId}`);
+      }
+    }
+
+    // TODO this is a regression, now there the own user is not always in the map. Ask Timo if fine
+    it("should use avatar url from room members", () => {
+      withTestScheduler(({ behavior, expectObservable }) => {
+        fakeMemberWith({
+          userId: "@local:example.com",
+        });
+        fakeMemberWith({
+          userId: "@alice:example.com",
+          getMxcAvatarUrl: vi.fn().mockReturnValue("mxc://custom.url/avatar"),
+        });
+        const memberships$ = behavior("a", {
+          a: [
+            mockCallMembership("@local:example.com", "DEVICE1"),
+            mockCallMembership("@alice:example.com", "DEVICE1"),
+          ],
+        });
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+        const local$ =
+          metadataStore.createAvatarUrlBehavior$("@local:example.com");
+
+        const alice$ =
+          metadataStore.createAvatarUrlBehavior$("@alice:example.com");
+
+        expectObservable(local$).toBe("a", {
+          a: "mxc://example.com/@local:example.com",
+        });
+        expectObservable(alice$).toBe("a", {
+          a: "mxc://custom.url/avatar",
+        });
+        expectObservable(metadataStore.avatarMap$).toBe("a", {
+          a: new Map<string, string>([
+            ["@local:example.com", "mxc://example.com/@local:example.com"],
+            ["@alice:example.com", "mxc://custom.url/avatar"],
+          ]),
+        });
+      });
+    });
+
+    it("should update on avatar change and user join/leave", () => {
+      withTestScheduler(({ behavior, schedule, expectObservable }) => {
+        fakeMemberWith({ userId: "@carl:example.com" });
+        fakeMemberWith({ userId: "@bob:example.com" });
+        const memberships$ = behavior("ab-d", {
+          a: [mockCallMembership("@bob:example.com", "B000")],
+          b: [
+            mockCallMembership("@bob:example.com", "B000"),
+            mockCallMembership("@carl:example.com", "C000"),
+          ],
+          d: [mockCallMembership("@carl:example.com", "C000")],
+        });
+
+        const metadataStore = createMatrixMemberMetadata$(
+          testScope,
+          memberships$,
+          createRoomMembers$(testScope, mockMatrixRoom),
+        );
+
+        schedule("--c-", {
+          c: () => {
+            updateAvatarUrl(
+              "@carl:example.com",
+              "mxc://updated.me/updatedAvatar",
+            );
+          },
+        });
+
+        const bob$ = metadataStore.createAvatarUrlBehavior$("@bob:example.com");
+        const carl$ =
+          metadataStore.createAvatarUrlBehavior$("@carl:example.com");
+        expectObservable(bob$).toBe("a---", {
+          a: "mxc://example.com/@bob:example.com",
+        });
+        expectObservable(carl$).toBe("a-c-", {
+          a: "mxc://example.com/@carl:example.com",
+
+          c: "mxc://updated.me/updatedAvatar",
+        });
+        expectObservable(metadataStore.avatarMap$).toBe("a-c-", {
+          a: new Map<string, string>([
+            ["@bob:example.com", "mxc://example.com/@bob:example.com"],
+            ["@carl:example.com", "mxc://example.com/@carl:example.com"],
+          ]),
+          // expect an update once we update the avatar URL
+          c: new Map<string, string>([
+            ["@bob:example.com", "mxc://example.com/@bob:example.com"],
+            ["@carl:example.com", "mxc://updated.me/updatedAvatar"],
+          ]),
+        });
+      });
+    });
+  });
+});

src/state/CallViewModel/remoteMembers/MatrixMemberMetadata.ts

@@ -0,0 +1,148 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { type RoomMember, RoomStateEvent } from "matrix-js-sdk";
+import { combineLatest, fromEvent, map } from "rxjs";
+import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
+import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
+import { type Room as MatrixRoom } from "matrix-js-sdk/lib/matrix";
+// eslint-disable-next-line rxjs/no-internal
+
+import { type ObservableScope } from "../../ObservableScope";
+import {
+  calculateDisplayName,
+  shouldDisambiguate,
+} from "../../../utils/displayname";
+import { type Behavior } from "../../Behavior";
+
+const logger = rootLogger.getChild("[MatrixMemberMetadata]");
+
+export type RoomMemberMap = Map<
+  string,
+  Pick<RoomMember, "userId" | "getMxcAvatarUrl" | "rawDisplayName">
+>;
+export function roomToMembersMap(matrixRoom: MatrixRoom): RoomMemberMap {
+  return matrixRoom.getMembers().reduce((acc, member) => {
+    acc.set(member.userId, {
+      userId: member.userId,
+      getMxcAvatarUrl: member.getMxcAvatarUrl.bind(member),
+      rawDisplayName: member.rawDisplayName,
+    });
+    return acc;
+  }, new Map());
+}
+
+export function createRoomMembers$(
+  scope: ObservableScope,
+  matrixRoom: MatrixRoom,
+): Behavior<RoomMemberMap> {
+  return scope.behavior(
+    fromEvent(matrixRoom, RoomStateEvent.Members).pipe(
+      map(() => roomToMembersMap(matrixRoom)),
+    ),
+    roomToMembersMap(matrixRoom),
+  );
+}
+/**
+ * Displayname for each member of the call. This will disambiguate
+ * any displayname that clashes with another member. Only members
+ * joined to the call are considered here.
+ *
+ * @returns Map<userId, displayname> uses the Matrix user ID as the key.
+ */
+// don't do this work more times than we need to. This is achieved by converting to a behavior:
+export const memberDisplaynames$ = (
+  scope: ObservableScope,
+  memberships$: Behavior<Pick<CallMembership, "userId">[]>,
+  roomMembers$: Behavior<RoomMemberMap>,
+): Behavior<Map<string, string>> => {
+  // This map tracks userIds that at some point needed disambiguation.
+  // This is a memory leak bound to the number of participants.
+  // A call application will always increase the memory if there have been more members in a call.
+  // Its capped by room member participants.
+  const shouldDisambiguateTrackerMap = new Set<string>();
+  return scope.behavior(
+    combineLatest([
+      // Handle call membership changes
+      memberships$,
+      // Additionally handle display name changes (implicitly reacting to them)
+      roomMembers$,
+      // TODO: do we need: pauseWhen(this.pretendToBeDisconnected$),
+    ]).pipe(
+      map(([memberships, roomMembers]) => {
+        const displaynameMap = new Map<string, string>();
+        // We only consider RTC members for disambiguation as they are the only visible members.
+        for (const rtcMember of memberships) {
+          const member = roomMembers.get(rtcMember.userId);
+          if (!member) {
+            logger.error(`Could not find member for user ${rtcMember.userId}`);
+            continue;
+          }
+          const disambiguateComputed = shouldDisambiguate(
+            member,
+            memberships,
+            roomMembers,
+          );
+
+          const disambiguate =
+            shouldDisambiguateTrackerMap.has(rtcMember.userId) ||
+            disambiguateComputed;
+          if (disambiguate) shouldDisambiguateTrackerMap.add(rtcMember.userId);
+          displaynameMap.set(
+            rtcMember.userId,
+            calculateDisplayName(member, disambiguate),
+          );
+        }
+        return displaynameMap;
+      }),
+    ),
+  );
+};
+
+export const createMatrixMemberMetadata$ = (
+  scope: ObservableScope,
+  memberships$: Behavior<Pick<CallMembership, "userId">[]>,
+  roomMembers$: Behavior<RoomMemberMap>,
+): {
+  createDisplayNameBehavior$: (userId: string) => Behavior<string | undefined>;
+  createAvatarUrlBehavior$: (userId: string) => Behavior<string | undefined>;
+  displaynameMap$: Behavior<Map<string, string>>;
+  avatarMap$: Behavior<Map<string, string | undefined>>;
+} => {
+  const displaynameMap$ = memberDisplaynames$(
+    scope,
+    memberships$,
+    roomMembers$,
+  );
+  const avatarMap$ = scope.behavior(
+    roomMembers$.pipe(
+      map((roomMembers) =>
+        Array.from(roomMembers.keys()).reduce((acc, key) => {
+          acc.set(key, roomMembers.get(key)?.getMxcAvatarUrl());
+          return acc;
+        }, new Map<string, string | undefined>()),
+      ),
+    ),
+  );
+  return {
+    createDisplayNameBehavior$: (userId: string) =>
+      scope.behavior(
+        displaynameMap$.pipe(
+          map((displaynameMap) => displaynameMap.get(userId)),
+        ),
+      ),
+    createAvatarUrlBehavior$: (userId: string) =>
+      scope.behavior(
+        roomMembers$.pipe(
+          map((roomMembers) => roomMembers.get(userId)?.getMxcAvatarUrl()),
+        ),
+      ),
+    // mostly for testing purposes
+    displaynameMap$,
+    avatarMap$,
+  };
+};

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

@@ -0,0 +1,228 @@
+/*
+Copyright 2025 Element Creations Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { test, vi, expect, beforeEach, afterEach } from "vitest";
+import { BehaviorSubject } from "rxjs";
+import { type Room as LivekitRoom } from "livekit-client";
+import EventEmitter from "events";
+import fetchMock from "fetch-mock";
+import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
+import { logger } from "matrix-js-sdk/lib/logger";
+
+import {
+  type Epoch,
+  ObservableScope,
+  trackEpoch,
+} from "../../ObservableScope.ts";
+import { ECConnectionFactory } from "./ConnectionFactory.ts";
+import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts";
+import {
+  mockCallMembership,
+  mockMediaDevices,
+  withTestScheduler,
+} from "../../../utils/test.ts";
+import { type ProcessorState } from "../../../livekit/TrackProcessorContext.tsx";
+import {
+  areLivekitTransportsEqual,
+  createMatrixLivekitMembers$,
+  type MatrixLivekitMember,
+} from "./MatrixLivekitMembers.ts";
+import { createConnectionManager$ } from "./ConnectionManager.ts";
+import { membershipsAndTransports$ } from "../../SessionBehaviors.ts";
+
+// Test the integration of ConnectionManager and MatrixLivekitMerger
+
+let testScope: ObservableScope;
+let ecConnectionFactory: ECConnectionFactory;
+let mockClient: OpenIDClientParts;
+let lkRoomFactory: () => LivekitRoom;
+
+const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
+
+beforeEach(() => {
+  testScope = new ObservableScope();
+  mockClient = {
+    getOpenIdToken: vi.fn().mockReturnValue(""),
+    getDeviceId: vi.fn().mockReturnValue("DEV000"),
+  };
+
+  lkRoomFactory = vi.fn().mockImplementation(() => {
+    const emitter = new EventEmitter();
+    const base = {
+      on: emitter.on.bind(emitter),
+      off: emitter.off.bind(emitter),
+      emit: emitter.emit.bind(emitter),
+      disconnect: vi.fn(),
+      remoteParticipants: new Map(),
+    } as unknown as LivekitRoom;
+
+    vi.mocked(base).connect = vi.fn().mockImplementation(({ url }) => {
+      createdMockLivekitRooms.set(url, base);
+    });
+    return base;
+  });
+
+  ecConnectionFactory = new ECConnectionFactory(
+    mockClient,
+    mockMediaDevices({}),
+    new BehaviorSubject<ProcessorState>({
+      supported: true,
+      processor: undefined,
+    }),
+    undefined,
+    false,
+    lkRoomFactory,
+  );
+
+  //TODO a bit annoying to have to do a http mock?
+  fetchMock.post(`path:/sfu/get`, (url) => {
+    const domain = new URL(url).hostname; // Extract the domain from the URL
+    return {
+      status: 200,
+      body: {
+        url: `wss://${domain}/livekit/sfu`,
+        jwt: "ATOKEN",
+      },
+    };
+  });
+});
+
+afterEach(() => {
+  testScope.end();
+  fetchMock.reset();
+});
+
+test("bob, carl, then bob joining no tracks yet", () => {
+  withTestScheduler(({ expectObservable, behavior, scope }) => {
+    const bobMembership = mockCallMembership("@bob:example.com", "BDEV000");
+    const carlMembership = mockCallMembership("@carl:example.com", "CDEV000");
+    const daveMembership = mockCallMembership("@dave:foo.bar", "DDEV000");
+
+    const eMarble = "abc";
+    const vMarble = "abc";
+    const memberships$ = scope.behavior(
+      behavior(eMarble, {
+        a: [bobMembership],
+        b: [bobMembership, carlMembership],
+        c: [bobMembership, carlMembership, daveMembership],
+      }).pipe(trackEpoch()),
+    );
+
+    const membershipsAndTransports = membershipsAndTransports$(
+      testScope,
+      memberships$,
+    );
+
+    const connectionManager = createConnectionManager$({
+      scope: testScope,
+      connectionFactory: ecConnectionFactory,
+      inputTransports$: membershipsAndTransports.transports$,
+      logger: logger,
+    });
+
+    const matrixLivekitItems$ = createMatrixLivekitMembers$({
+      scope: testScope,
+      membershipsWithTransport$:
+        membershipsAndTransports.membershipsWithTransport$,
+      connectionManager,
+    });
+
+    expectObservable(matrixLivekitItems$).toBe(vMarble, {
+      a: expect.toSatisfy((e: Epoch<MatrixLivekitMember[]>) => {
+        const items = e.value;
+        expect(items.length).toBe(1);
+        const item = items[0]!;
+        expectObservable(item.membership$).toBe("a", {
+          a: bobMembership,
+        });
+        expectObservable(item.connection$).toBe("a", {
+          a: expect.toSatisfy((co) =>
+            areLivekitTransportsEqual(
+              co.transport,
+              bobMembership.transports[0]! as LivekitTransport,
+            ),
+          ),
+        });
+        expectObservable(item.participant$).toBe("a", {
+          a: null,
+        });
+        return true;
+      }),
+      b: expect.toSatisfy((e: Epoch<MatrixLivekitMember[]>) => {
+        const items = e.value;
+        expect(items.length).toBe(2);
+
+        {
+          const item = items[0]!;
+          expectObservable(item.membership$).toBe("a", {
+            a: bobMembership,
+          });
+          expectObservable(item.participant$).toBe("a", {
+            a: null,
+          });
+        }
+
+        {
+          const item = items[1]!;
+
+          expectObservable(item.membership$).toBe("a", {
+            a: carlMembership,
+          });
+          expectObservable(item.participant$).toBe("a", {
+            a: null,
+          });
+          expectObservable(item.connection$).toBe("a", {
+            a: expect.toSatisfy((connection) => {
+              expect(
+                areLivekitTransportsEqual(
+                  connection.transport,
+                  carlMembership.transports[0]! as LivekitTransport,
+                ),
+              ).toBe(true);
+              return true;
+            }),
+          });
+        }
+        return true;
+      }),
+      c: expect.toSatisfy((e: Epoch<MatrixLivekitMember[]>) => {
+        const items = e.value;
+        expect(items.length).toBe(3);
+
+        expectObservable(items[0].membership$).toBe("a", {
+          a: bobMembership,
+        });
+        expectObservable(items[1].membership$).toBe("b", {
+          a: carlMembership,
+        });
+
+        {
+          const item = items[2]!;
+          expectObservable(item.membership$).toBe("a", {
+            a: daveMembership,
+          });
+          expectObservable(item.connection$).toBe("a", {
+            a: expect.toSatisfy((connection) => {
+              expect(
+                areLivekitTransportsEqual(
+                  connection.transport,
+                  daveMembership.transports[0]! as LivekitTransport,
+                ),
+              ).toBe(true);
+              return true;
+            }),
+          });
+          expectObservable(item.participant$).toBe("a", {
+            a: null,
+          });
+        }
+        return true;
+      }),
+      x: expect.anything(),
+    });
+  });
+});

src/state/CallViewModelWidget.test.ts

@@ -5,15 +5,18 @@ Copyright 2025 Element Creations Ltd.
 Please see LICENSE in the repository root for full details.
 */
 
-import { test, vi, expect } from "vitest";
+import { it, vi, expect } from "vitest";
 import EventEmitter from "events";
 
+// import * as ComponentsCore from "@livekit/components-core";
+import { withCallViewModel } from "./CallViewModel/CallViewModelTestUtils.ts";
+import { type CallViewModel } from "./CallViewModel/CallViewModel.ts";
 import { constant } from "./Behavior.ts";
-import { withCallViewModel } from "./CallViewModel.test.ts";
 import { aliceParticipant, localRtcMember } from "../utils/test-fixtures.ts";
 import { ElementWidgetActions, widget } from "../widget.ts";
 import { E2eeType } from "../e2ee/e2eeType.ts";
-import { type CallViewModel } from "./CallViewModel.ts";
+
+vi.mock("@livekit/components-core", { spy: true });
 
 vi.mock("../widget", () => ({
   ElementWidgetActions: {
@@ -31,7 +34,7 @@ vi.mock("../widget", () => ({
   },
 }));
 
-test("expect leave when ElementWidgetActions.HangupCall is called", async () => {
+it("expect leave when ElementWidgetActions.HangupCall is called", async () => {
   const pr = Promise.withResolvers<string>();
   withCallViewModel(
     {

src/state/MediaViewModel.ts

@@ -27,7 +27,6 @@ import {
   RoomEvent as LivekitRoomEvent,
   RemoteTrack,
 } from "livekit-client";
-import { type RoomMember } from "matrix-js-sdk";
 import { logger } from "matrix-js-sdk/lib/logger";
 import {
   BehaviorSubject,
@@ -44,6 +43,7 @@ import {
   startWith,
   switchMap,
   throttleTime,
+  distinctUntilChanged,
 } from "rxjs";
 
 import { alwaysShowSelf } from "../settings/settings";
@@ -180,29 +180,35 @@ function observeRemoteTrackReceivingOkay$(
 }
 
 function encryptionErrorObservable$(
-  room: LivekitRoom,
+  room$: Behavior<LivekitRoom | undefined>,
   participant: Participant,
   encryptionSystem: EncryptionSystem,
   criteria: string,
 ): Observable<boolean> {
-  return roomEventSelector(room, LivekitRoomEvent.EncryptionError).pipe(
-    map((e) => {
-      const [err] = e;
-      if (encryptionSystem.kind === E2eeType.PER_PARTICIPANT) {
-        return (
-          // Ideally we would pull the participant identity from the field on the error.
-          // However, it gets lost in the serialization process between workers.
-          // So, instead we do a string match
-          (err?.message.includes(participant.identity) &&
-            err?.message.includes(criteria)) ??
-          false
-        );
-      } else if (encryptionSystem.kind === E2eeType.SHARED_KEY) {
-        return !!err?.message.includes(criteria);
-      }
+  return room$.pipe(
+    switchMap((room) => {
+      if (room === undefined) return of(false);
+      return roomEventSelector(room, LivekitRoomEvent.EncryptionError).pipe(
+        map((e) => {
+          const [err] = e;
+          if (encryptionSystem.kind === E2eeType.PER_PARTICIPANT) {
+            return (
+              // Ideally we would pull the participant identity from the field on the error.
+              // However, it gets lost in the serialization process between workers.
+              // So, instead we do a string match
+              (err?.message.includes(participant.identity) &&
+                err?.message.includes(criteria)) ??
+              false
+            );
+          } else if (encryptionSystem.kind === E2eeType.SHARED_KEY) {
+            return !!err?.message.includes(criteria);
+          }
 
-      return false;
+          return false;
+        }),
+      );
     }),
+    distinctUntilChanged(),
     throttleTime(1000), // Throttle to avoid spamming the UI
     startWith(false),
   );
@@ -220,7 +226,7 @@ abstract class BaseMediaViewModel {
   /**
    * The LiveKit video track for this media.
    */
-  public readonly video$: Behavior<TrackReferenceOrPlaceholder | undefined>;
+  public readonly video$: Behavior<TrackReferenceOrPlaceholder | null>;
   /**
    * Whether there should be a warning that this media is unencrypted.
    */
@@ -235,12 +241,10 @@ abstract class BaseMediaViewModel {
 
   private observeTrackReference$(
     source: Track.Source,
-  ): Behavior<TrackReferenceOrPlaceholder | undefined> {
+  ): Behavior<TrackReferenceOrPlaceholder | null> {
     return this.scope.behavior(
       this.participant$.pipe(
-        switchMap((p) =>
-          p === undefined ? of(undefined) : observeTrackReference$(p, source),
-        ),
+        switchMap((p) => (!p ? of(null) : observeTrackReference$(p, source))),
       ),
     );
   }
@@ -252,23 +256,22 @@ abstract class BaseMediaViewModel {
      */
     public readonly id: string,
     /**
-     * The Matrix room member to which this media belongs.
+     * The Matrix user to which this media belongs.
      */
-    // TODO: Fully separate the data layer from the UI layer by keeping the
-    // member object internal
-    public readonly member: RoomMember,
+    public readonly userId: string,
     // We don't necessarily have a participant if a user connects via MatrixRTC but not (yet) through
     // livekit.
     protected readonly participant$: Observable<
-      LocalParticipant | RemoteParticipant | undefined
+      LocalParticipant | RemoteParticipant | null
     >,
 
     encryptionSystem: EncryptionSystem,
     audioSource: AudioSource,
     videoSource: VideoSource,
-    livekitRoom: LivekitRoom,
-    public readonly focusURL: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    public readonly focusUrl$: Behavior<string | undefined>,
     public readonly displayName$: Behavior<string>,
+    public readonly mxcAvatarUrl$: Behavior<string | undefined>,
   ) {
     const audio$ = this.observeTrackReference$(audioSource);
     this.video$ = this.observeTrackReference$(videoSource);
@@ -296,13 +299,13 @@ abstract class BaseMediaViewModel {
           } else if (encryptionSystem.kind === E2eeType.PER_PARTICIPANT) {
             return combineLatest([
               encryptionErrorObservable$(
-                livekitRoom,
+                livekitRoom$,
                 participant,
                 encryptionSystem,
                 "MissingKey",
               ),
               encryptionErrorObservable$(
-                livekitRoom,
+                livekitRoom$,
                 participant,
                 encryptionSystem,
                 "InvalidKey",
@@ -322,7 +325,7 @@ abstract class BaseMediaViewModel {
           } else {
             return combineLatest([
               encryptionErrorObservable$(
-                livekitRoom,
+                livekitRoom$,
                 participant,
                 encryptionSystem,
                 "InvalidKey",
@@ -404,26 +407,28 @@ abstract class BaseUserMediaViewModel extends BaseMediaViewModel {
   public constructor(
     scope: ObservableScope,
     id: string,
-    member: RoomMember,
-    participant$: Observable<LocalParticipant | RemoteParticipant | undefined>,
+    userId: string,
+    participant$: Observable<LocalParticipant | RemoteParticipant | null>,
     encryptionSystem: EncryptionSystem,
-    livekitRoom: LivekitRoom,
-    focusUrl: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    focusUrl$: Behavior<string | undefined>,
     displayName$: Behavior<string>,
+    mxcAvatarUrl$: Behavior<string | undefined>,
     public readonly handRaised$: Behavior<Date | null>,
     public readonly reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       scope,
       id,
-      member,
+      userId,
       participant$,
       encryptionSystem,
       Track.Source.Microphone,
       Track.Source.Camera,
-      livekitRoom,
-      focusUrl,
+      livekitRoom$,
+      focusUrl$,
       displayName$,
+      mxcAvatarUrl$,
     );
 
     const media$ = this.scope.behavior(
@@ -540,25 +545,27 @@ export class LocalUserMediaViewModel extends BaseUserMediaViewModel {
   public constructor(
     scope: ObservableScope,
     id: string,
-    member: RoomMember,
-    participant$: Behavior<LocalParticipant | undefined>,
+    userId: string,
+    participant$: Behavior<LocalParticipant | null>,
     encryptionSystem: EncryptionSystem,
-    livekitRoom: LivekitRoom,
-    focusURL: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    focusUrl$: Behavior<string | undefined>,
     private readonly mediaDevices: MediaDevices,
     displayName$: Behavior<string>,
+    mxcAvatarUrl$: Behavior<string | undefined>,
     handRaised$: Behavior<Date | null>,
     reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       scope,
       id,
-      member,
+      userId,
       participant$,
       encryptionSystem,
-      livekitRoom,
-      focusURL,
+      livekitRoom$,
+      focusUrl$,
       displayName$,
+      mxcAvatarUrl$,
       handRaised$,
       reaction$,
     );
@@ -650,25 +657,27 @@ export class RemoteUserMediaViewModel extends BaseUserMediaViewModel {
   public constructor(
     scope: ObservableScope,
     id: string,
-    member: RoomMember,
-    participant$: Observable<RemoteParticipant | undefined>,
+    userId: string,
+    participant$: Observable<RemoteParticipant | null>,
     encryptionSystem: EncryptionSystem,
-    livekitRoom: LivekitRoom,
-    focusUrl: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    focusUrl$: Behavior<string | undefined>,
     private readonly pretendToBeDisconnected$: Behavior<boolean>,
-    displayname$: Behavior<string>,
+    displayName$: Behavior<string>,
+    mxcAvatarUrl$: Behavior<string | undefined>,
     handRaised$: Behavior<Date | null>,
     reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       scope,
       id,
-      member,
+      userId,
       participant$,
       encryptionSystem,
-      livekitRoom,
-      focusUrl,
-      displayname$,
+      livekitRoom$,
+      focusUrl$,
+      displayName$,
+      mxcAvatarUrl$,
       handRaised$,
       reaction$,
     );
@@ -749,26 +758,28 @@ export class ScreenShareViewModel extends BaseMediaViewModel {
   public constructor(
     scope: ObservableScope,
     id: string,
-    member: RoomMember,
+    userId: string,
     participant$: Observable<LocalParticipant | RemoteParticipant>,
     encryptionSystem: EncryptionSystem,
-    livekitRoom: LivekitRoom,
-    focusUrl: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    focusUrl$: Behavior<string | undefined>,
     private readonly pretendToBeDisconnected$: Behavior<boolean>,
-    displayname$: Behavior<string>,
+    displayName$: Behavior<string>,
+    mxcAvatarUrl$: Behavior<string | undefined>,
     public readonly local: boolean,
   ) {
     super(
       scope,
       id,
-      member,
+      userId,
       participant$,
       encryptionSystem,
       Track.Source.ScreenShareAudio,
       Track.Source.ScreenShare,
-      livekitRoom,
-      focusUrl,
-      displayname$,
+      livekitRoom$,
+      focusUrl$,
+      displayName$,
+      mxcAvatarUrl$,
     );
   }
 }

src/state/ObservableScope.test.ts

@@ -0,0 +1,104 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { describe, expect, it } from "vitest";
+import { BehaviorSubject, combineLatest, Subject } from "rxjs";
+import { logger } from "matrix-js-sdk/lib/logger";
+
+import {
+  Epoch,
+  mapEpoch,
+  ObservableScope,
+  trackEpoch,
+} from "./ObservableScope";
+import { withTestScheduler } from "../utils/test";
+
+describe("Epoch", () => {
+  it("should map the value correctly", () => {
+    const epoch = new Epoch(1);
+    const mappedEpoch = epoch.mapInner((v) => v + 1);
+    expect(mappedEpoch.value).toBe(2);
+    expect(mappedEpoch.epoch).toBe(0);
+  });
+
+  it("should be tracked from an observable", () => {
+    withTestScheduler(({ expectObservable, behavior }) => {
+      const observable$ = behavior("abc", {
+        a: 1,
+        b: 2,
+        c: 3,
+      });
+      const epochObservable$ = observable$.pipe(trackEpoch());
+      expectObservable(epochObservable$).toBe("abc", {
+        a: expect.toSatisfy((e) => e.epoch === 0 && e.value === 1),
+        b: expect.toSatisfy((e) => e.epoch === 1 && e.value === 2),
+        c: expect.toSatisfy((e) => e.epoch === 2 && e.value === 3),
+      });
+    });
+  });
+
+  it("can be mapped without loosing epoch information", () => {
+    withTestScheduler(({ expectObservable, behavior }) => {
+      const observable$ = behavior("abc", {
+        a: "A",
+        b: "B",
+        c: "C",
+      });
+      const epochObservable$ = observable$.pipe(trackEpoch());
+      const derivedEpoch$ = epochObservable$.pipe(
+        mapEpoch((e) => e + "-mapped"),
+      );
+
+      expectObservable(derivedEpoch$).toBe("abc", {
+        a: new Epoch("A-mapped", 0),
+        b: new Epoch("B-mapped", 1),
+        c: new Epoch("C-mapped", 2),
+      });
+    });
+  });
+
+  it("diamonds emits in a predictable order", () => {
+    const sb$ = new BehaviorSubject("initial");
+    const root$ = sb$.pipe(trackEpoch());
+    const derivedA$ = root$.pipe(mapEpoch((e) => e + "-A"));
+    const derivedB$ = root$.pipe(mapEpoch((e) => e + "-B"));
+    combineLatest([root$, derivedB$, derivedA$]).subscribe(
+      ([root, derivedA, derivedB]) => {
+        logger.log(
+          "combined" +
+            root.epoch +
+            root.value +
+            "\n" +
+            derivedA.epoch +
+            derivedA.value +
+            "\n" +
+            derivedB.epoch +
+            derivedB.value,
+        );
+      },
+    );
+    sb$.next("updated");
+    sb$.next("ANOTERUPDATE");
+  });
+
+  it("behavior test", () => {
+    const scope = new ObservableScope();
+    const s$ = new Subject();
+    const behavior$ = scope.behavior(s$, 0);
+    behavior$.subscribe((value) => {
+      logger.log(`Received value: ${value}`);
+    });
+    s$.next(1);
+    s$.next(2);
+    s$.next(3);
+    s$.next(3);
+    s$.next(3);
+    s$.next(3);
+    s$.next(3);
+    s$.complete();
+  });
+});

src/state/ObservableScope.ts

@@ -12,7 +12,9 @@ import {
   EMPTY,
   endWith,
   filter,
+  map,
   type Observable,
+  type OperatorFunction,
   share,
   take,
   takeUntil,
@@ -22,6 +24,10 @@ import { type Behavior } from "./Behavior";
 
 type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
 
+type SplitBehavior<T> = keyof T extends string | number
+  ? { [K in keyof T as `${K}$`]: Behavior<T[K]> }
+  : never;
+
 const nothing = Symbol("nothing");
 
 /**
@@ -145,9 +151,132 @@ export class ObservableScope {
         })();
       });
   }
+
+  /**
+   * Splits a Behavior of objects with static properties into an object with
+   * Behavior properties.
+   *
+   * For example, splitting a `Behavior<{ name: string; age: number }>` results
+   * in an object of type `{ name$: Behavior<string>; age$: Behavior<number> }`.
+   */
+  public splitBehavior<T extends object>(
+    input$: Behavior<T>,
+  ): SplitBehavior<T> {
+    return Object.fromEntries(
+      Object.keys(input$.value).map((key) => [
+        `${key}$`,
+        this.behavior(input$.pipe(map((input) => input[key as keyof T]))),
+      ]),
+    ) as SplitBehavior<T>;
+  }
 }
 
 /**
  * The global scope, a scope which never ends.
  */
 export const globalScope = new ObservableScope();
+
+/**
+ * `Epoch`'s can be used to create `Behavior`s and `Observable`s which derivitives can be merged
+ * with `combinedLatest` without duplicated emissions.
+ *
+ * This is useful in the following example:
+ * ```
+ * const rootObs$ = of("red","green","blue");
+ * const derivedObs$ = rootObs$.pipe(
+ *   map((v)=> {red:"fire", green:"grass", blue:"water"}[v])
+ * );
+ * const otherDerivedObs$ = rootObs$.pipe(
+ *   map((v)=> {red:"tomatoes", green:"leaves", blue:"sky"}[v])
+ * );
+ * const mergedObs$ = combineLatest([rootObs$, derivedObs$, otherDerivedObs$]).pipe(
+ *   map(([color, a,b]) => color + " like " + a + " and " + b)
+ * );
+ *
+ * ```
+ * will result in 6 emissions with mismatching items like "red like fire and leaves"
+ *
+ * # Use Epoch
+ * ```
+ * const ancestorObs$ = of(1,2,3).pipe(trackEpoch());
+ * const derivedObs$ = ancestorObs$.pipe(
+ *   mapEpoch((v)=> "this number: " + v)
+ * );
+ * const otherDerivedObs$ = ancestorObs$.pipe(
+ *   mapEpoch((v)=> "multiplied by: " + v)
+ * );
+ * const mergedObs$ = combineLatest([derivedObs$, otherDerivedObs$]).pipe(
+ *   filter((values) => values.every((v) => v.epoch === values[0].v)),
+ *   map(([color, a, b]) => color + " like " + a + " and " + b)
+ * );
+ *
+ * ```
+ * will result in 3 emissions all matching (e.g. "blue like water and sky")
+ */
+export class Epoch<T> {
+  public readonly epoch: number;
+  public readonly value: T;
+
+  public constructor(value: T, epoch?: number) {
+    this.value = value;
+    this.epoch = epoch ?? 0;
+  }
+  /**
+   * Maps the value inside the epoch to a new value while keeping the epoch number.
+   * # usage
+   * ```
+   * const myEpoch$ = myObservable$.pipe(
+   *   map(trackEpoch()),
+   *   // this is the preferred way using mapEpoch
+   *   mapEpoch((v)=> v+1)
+   *   // This is how inner map can be used:
+   *   map((epoch) => epoch.innerMap((v)=> v+1))
+   *   // It is equivalent to:
+   *   map((epoch) => new Epoch(epoch.value + 1, epoch.epoch))
+   * )
+   * ```
+   * See also `Epoch<T>`
+   */
+  public mapInner<U>(map: (value: T) => U): Epoch<U> {
+    return new Epoch<U>(map(this.value), this.epoch);
+  }
+}
+
+/**
+ * A `pipe` compatible map oparator that keeps the epoch in tact but allows mapping the value.
+ * # usage
+ * ```
+ * const myEpoch$ = myObservable$.pipe(
+ *   map(trackEpoch()),
+ *   // this is the preferred way using mapEpoch
+ *   mapEpoch((v)=> v+1)
+ *   // This is how inner map can be used:
+ *   map((epoch) => epoch.innerMap((v)=> v+1))
+ *   // It is equivalent to:
+ *   map((epoch) => new Epoch(epoch.value + 1, epoch.epoch))
+ * )
+ * ```
+ * See also `Epoch<T>`
+ */
+export function mapEpoch<T, U>(
+  mapFn: (value: T) => U,
+): OperatorFunction<Epoch<T>, Epoch<U>> {
+  return map((e) => e.mapInner(mapFn));
+}
+
+/**
+ * # usage
+ * ```
+ * const myEpoch$ = myObservable$.pipe(
+ *   map(trackEpoch()),
+ *   map((epoch) => epoch.innerMap((v)=> v+1))
+ * )
+ * const derived = myEpoch$.pipe(
+ *   mapEpoch((v)=>v^2)
+ * )
+ * ```
+ * See also `Epoch<T>`
+ */
+export function trackEpoch<T>(): OperatorFunction<T, Epoch<T>> {
+  return map<T, Epoch<T>>((value, number) => new Epoch(value, number));
+}

src/state/ScreenShare.ts

@@ -4,7 +4,7 @@ Copyright 2025 New Vector Ltd.
 SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
-import { of, type Observable } from "rxjs";
+import { of } from "rxjs";
 import {
   type LocalParticipant,
   type RemoteParticipant,
@@ -13,7 +13,6 @@ import {
 
 import { type ObservableScope } from "./ObservableScope.ts";
 import { ScreenShareViewModel } from "./MediaViewModel.ts";
-import type { RoomMember } from "matrix-js-sdk";
 import type { EncryptionSystem } from "../e2ee/sharedKeyManagement.ts";
 import type { Behavior } from "./Behavior.ts";
 
@@ -28,24 +27,26 @@ export class ScreenShare {
   public constructor(
     private readonly scope: ObservableScope,
     id: string,
-    member: RoomMember,
+    userId: string,
     participant: LocalParticipant | RemoteParticipant,
     encryptionSystem: EncryptionSystem,
-    livekitRoom: LivekitRoom,
-    focusUrl: string,
+    livekitRoom$: Behavior<LivekitRoom | undefined>,
+    focusUrl$: Behavior<string | undefined>,
     pretendToBeDisconnected$: Behavior<boolean>,
-    displayName$: Observable<string>,
+    displayName$: Behavior<string>,
+    mxcAvatarUrl$: Behavior<string | undefined>,
   ) {
     this.vm = new ScreenShareViewModel(
       this.scope,
       id,
-      member,
+      userId,
       of(participant),
       encryptionSystem,
-      livekitRoom,
-      focusUrl,
+      livekitRoom$,
+      focusUrl$,
       pretendToBeDisconnected$,
-      this.scope.behavior(displayName$),
+      displayName$,
+      mxcAvatarUrl$,
       participant.isLocal,
     );
   }

src/state/SessionBehaviors.ts

@@ -12,33 +12,25 @@ import {
   type MatrixRTCSession,
   MatrixRTCSessionEvent,
 } from "matrix-js-sdk/lib/matrixrtc";
-import { fromEvent, map } from "rxjs";
+import { fromEvent } from "rxjs";
 
-import { type ObservableScope } from "./ObservableScope";
+import {
+  Epoch,
+  mapEpoch,
+  trackEpoch,
+  type ObservableScope,
+} from "./ObservableScope";
 import { type Behavior } from "./Behavior";
 
-interface Props {
-  scope: ObservableScope;
-  matrixRTCSession: MatrixRTCSession;
-}
-
-export const createSessionMembershipsAndTransports$ = ({
-  scope,
-  matrixRTCSession,
-}: Props): {
-  memberships$: Behavior<CallMembership[]>;
+export const membershipsAndTransports$ = (
+  scope: ObservableScope,
+  memberships$: Behavior<Epoch<CallMembership[]>>,
+): {
   membershipsWithTransport$: Behavior<
-    { membership: CallMembership; transport?: LivekitTransport }[]
+    Epoch<{ membership: CallMembership; transport?: LivekitTransport }[]>
   >;
-  transports$: Behavior<LivekitTransport[]>;
+  transports$: Behavior<Epoch<LivekitTransport[]>>;
 } => {
-  const memberships$ = scope.behavior(
-    fromEvent(
-      matrixRTCSession,
-      MatrixRTCSessionEvent.MembershipsChanged,
-      (_, memberships: CallMembership[]) => memberships,
-    ),
-  );
   /**
    * Lists the transports used by ourselves, plus all other MatrixRTC session
    * members. For completeness this also lists the preferred transport and
@@ -47,11 +39,9 @@ export const createSessionMembershipsAndTransports$ = ({
    * together when it might change together is what you have to do in RxJS to
    * avoid reading inconsistent state or observing too many changes.)
    */
-  const membershipsWithTransport$: Behavior<
-    { membership: CallMembership; transport?: LivekitTransport }[]
-  > = scope.behavior(
+  const membershipsWithTransport$ = scope.behavior(
     memberships$.pipe(
-      map((memberships) => {
+      mapEpoch((memberships) => {
         return memberships.map((membership) => {
           const oldestMembership = memberships[0] ?? membership;
           const transport = membership.getTransport(oldestMembership);
@@ -66,12 +56,26 @@ export const createSessionMembershipsAndTransports$ = ({
 
   const transports$ = scope.behavior(
     membershipsWithTransport$.pipe(
-      map((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
+      mapEpoch((mts) => mts.flatMap(({ transport: t }) => (t ? [t] : []))),
     ),
   );
+
   return {
-    memberships$,
     membershipsWithTransport$,
     transports$,
   };
 };
+
+export const createMemberships$ = (
+  scope: ObservableScope,
+  matrixRTCSession: MatrixRTCSession,
+): Behavior<Epoch<CallMembership[]>> => {
+  return scope.behavior(
+    fromEvent(
+      matrixRTCSession,
+      MatrixRTCSessionEvent.MembershipsChanged,
+      (_, memberships: CallMembership[]) => memberships,
+    ).pipe(trackEpoch()),
+    new Epoch(matrixRTCSession.memberships),
+  );
+};

src/state/TileStore.ts

@@ -14,7 +14,7 @@ import { fillGaps } from "../utils/iter";
 import { debugTileLayout } from "../settings/settings";
 
 function debugEntries(entries: GridTileData[]): string[] {
-  return entries.map((e) => e.media.member?.rawDisplayName ?? "[👻]");
+  return entries.map((e) => e.media.displayName$.value);
 }
 
 let DEBUG_ENABLED = false;
@@ -156,7 +156,7 @@ export class TileStoreBuilder {
   public registerSpotlight(media: MediaViewModel[], maximised: boolean): void {
     if (DEBUG_ENABLED)
       logger.debug(
-        `[TileStore, ${this.generation}] register spotlight: ${media.map((m) => m.member?.rawDisplayName ?? "[👻]")}`,
+        `[TileStore, ${this.generation}] register spotlight: ${media.map((m) => m.displayName$.value)}`,
       );
 
     if (this.spotlight !== null) throw new Error("Spotlight already set");
@@ -180,7 +180,7 @@ export class TileStoreBuilder {
   public registerGridTile(media: UserMediaViewModel): void {
     if (DEBUG_ENABLED)
       logger.debug(
-        `[TileStore, ${this.generation}] register grid tile: ${media.member?.rawDisplayName ?? "[👻]"}`,
+        `[TileStore, ${this.generation}] register grid tile: ${media.displayName$.value}`,
       );
 
     if (this.spotlight !== null) {
@@ -263,7 +263,7 @@ export class TileStoreBuilder {
   public registerPipTile(media: UserMediaViewModel): void {
     if (DEBUG_ENABLED)
       logger.debug(
-        `[TileStore, ${this.generation}] register PiP tile: ${media.member?.rawDisplayName ?? "[👻]"}`,
+        `[TileStore, ${this.generation}] register PiP tile: ${media.displayName$.value}`,
       );
 
     // If there is a single grid tile that we can reuse

src/state/UserMedia.ts

@@ -5,17 +5,9 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import {
-  BehaviorSubject,
-  combineLatest,
-  map,
-  type Observable,
-  of,
-  switchMap,
-} from "rxjs";
+import { combineLatest, map, type Observable, of, switchMap } from "rxjs";
 import {
   type LocalParticipant,
-  type Participant,
   ParticipantEvent,
   type RemoteParticipant,
   type Room as LivekitRoom,
@@ -29,11 +21,12 @@ import {
   type UserMediaViewModel,
 } from "./MediaViewModel.ts";
 import type { Behavior } from "./Behavior.ts";
-import type { RoomMember } from "matrix-js-sdk";
 import type { EncryptionSystem } from "../e2ee/sharedKeyManagement.ts";
 import type { MediaDevices } from "./MediaDevices.ts";
 import type { ReactionOption } from "../reactions";
 import { observeSpeaker$ } from "./observeSpeaker.ts";
+import { generateItems } from "../utils/observable.ts";
+import { ScreenShare } from "./ScreenShare.ts";
 
 /**
  * Sorting bins defining the order in which media tiles appear in the layout.
@@ -72,35 +65,35 @@ enum SortingBin {
 /**
  * A user media item to be presented in a tile. This is a thin wrapper around
  * UserMediaViewModel which additionally determines the media item's sorting bin
- * for inclusion in the call layout.
+ * for inclusion in the call layout and tracks associated screen shares.
  */
 export class UserMedia {
-  private readonly participant$ = new BehaviorSubject(this.initialParticipant);
-
   public readonly vm: UserMediaViewModel = this.participant$.value?.isLocal
     ? new LocalUserMediaViewModel(
         this.scope,
         this.id,
-        this.member,
-        this.participant$ as Behavior<LocalParticipant>,
+        this.userId,
+        this.participant$ as Behavior<LocalParticipant | null>,
         this.encryptionSystem,
-        this.livekitRoom,
-        this.focusURL,
+        this.livekitRoom$,
+        this.focusUrl$,
         this.mediaDevices,
-        this.scope.behavior(this.displayname$),
+        this.displayName$,
+        this.mxcAvatarUrl$,
         this.scope.behavior(this.handRaised$),
         this.scope.behavior(this.reaction$),
       )
     : new RemoteUserMediaViewModel(
         this.scope,
         this.id,
-        this.member,
-        this.participant$ as Observable<RemoteParticipant | undefined>,
+        this.userId,
+        this.participant$ as Behavior<RemoteParticipant | null>,
         this.encryptionSystem,
-        this.livekitRoom,
-        this.focusURL,
+        this.livekitRoom$,
+        this.focusUrl$,
         this.pretendToBeDisconnected$,
-        this.scope.behavior(this.displayname$),
+        this.displayName$,
+        this.mxcAvatarUrl$,
         this.scope.behavior(this.handRaised$),
         this.scope.behavior(this.reaction$),
       );
@@ -109,12 +102,55 @@ export class UserMedia {
     observeSpeaker$(this.vm.speaking$),
   );
 
-  private readonly presenter$ = this.scope.behavior(
+  /**
+   * All screen share media associated with this user media.
+   */
+  public readonly screenShares$ = this.scope.behavior(
     this.participant$.pipe(
-      switchMap((p) => (p === undefined ? of(false) : sharingScreen$(p))),
+      switchMap((p) =>
+        p === null
+          ? of([])
+          : observeParticipantEvents(
+              p,
+              ParticipantEvent.TrackPublished,
+              ParticipantEvent.TrackUnpublished,
+              ParticipantEvent.LocalTrackPublished,
+              ParticipantEvent.LocalTrackUnpublished,
+            ).pipe(
+              // Technically more than one screen share might be possible... our
+              // MediaViewModels don't support it though since they look for a unique
+              // track for the given source. So generateItems here is a bit overkill.
+              generateItems(
+                function* (p) {
+                  if (p.isScreenShareEnabled)
+                    yield {
+                      keys: ["screen-share"],
+                      data: undefined,
+                    };
+                },
+                (scope, _data$, key) =>
+                  new ScreenShare(
+                    scope,
+                    `${this.id}:${key}`,
+                    this.userId,
+                    p,
+                    this.encryptionSystem,
+                    this.livekitRoom$,
+                    this.focusUrl$,
+                    this.pretendToBeDisconnected$,
+                    this.displayName$,
+                    this.mxcAvatarUrl$,
+                  ),
+              ),
+            ),
+      ),
     ),
   );
 
+  private readonly presenter$ = this.scope.behavior(
+    this.screenShares$.pipe(map((screenShares) => screenShares.length > 0)),
+  );
+
   /**
    * Which sorting bin the media item should be placed in.
    */
@@ -147,37 +183,18 @@ export class UserMedia {
   public constructor(
     private readonly scope: ObservableScope,
     public readonly id: string,
-    private readonly member: RoomMember,
-    private readonly initialParticipant:
-      | LocalParticipant
-      | RemoteParticipant
-      | undefined,
+    private readonly userId: string,
+    private readonly participant$: Behavior<
+      LocalParticipant | RemoteParticipant | null
+    >,
     private readonly encryptionSystem: EncryptionSystem,
-    private readonly livekitRoom: LivekitRoom,
-    private readonly focusURL: string,
+    private readonly livekitRoom$: Behavior<LivekitRoom | undefined>,
+    private readonly focusUrl$: Behavior<string | undefined>,
     private readonly mediaDevices: MediaDevices,
     private readonly pretendToBeDisconnected$: Behavior<boolean>,
-    private readonly displayname$: Observable<string>,
+    private readonly displayName$: Behavior<string>,
+    private readonly mxcAvatarUrl$: Behavior<string | undefined>,
     private readonly handRaised$: Observable<Date | null>,
     private readonly reaction$: Observable<ReactionOption | null>,
   ) {}
-
-  public updateParticipant(
-    newParticipant: LocalParticipant | RemoteParticipant | undefined,
-  ): void {
-    if (this.participant$.value !== newParticipant) {
-      // Update the BehaviourSubject in the UserMedia.
-      this.participant$.next(newParticipant);
-    }
-  }
-}
-
-export function sharingScreen$(p: Participant): Observable<boolean> {
-  return observeParticipantEvents(
-    p,
-    ParticipantEvent.TrackPublished,
-    ParticipantEvent.TrackUnpublished,
-    ParticipantEvent.LocalTrackPublished,
-    ParticipantEvent.LocalTrackUnpublished,
-  ).pipe(map((p) => p.isScreenShareEnabled));
 }

src/state/localMember/LocalMembership.ts

@@ -1,422 +0,0 @@
-/*
-Copyright 2025 New Vector Ltd.
-
-SPDX-License-IdFentifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import { type LocalTrack, type E2EEOptions } from "livekit-client";
-import {
-  type LivekitTransport,
-  type MatrixRTCSession,
-  MembershipManagerEvent,
-  Status,
-} from "matrix-js-sdk/lib/matrixrtc";
-import { ClientEvent, SyncState, type Room as MatrixRoom } from "matrix-js-sdk";
-import {
-  BehaviorSubject,
-  combineLatest,
-  fromEvent,
-  map,
-  type Observable,
-  of,
-  scan,
-  startWith,
-  switchMap,
-  tap,
-} from "rxjs";
-import { logger } from "matrix-js-sdk/lib/logger";
-
-import { type Behavior } from "../Behavior";
-import { type createConnectionManager$ } from "../remoteMembers/ConnectionManager";
-import { ObservableScope } from "../ObservableScope";
-import { Publisher } from "./Publisher";
-import { type MuteStates } from "../MuteStates";
-import { type ProcessorState } from "../../livekit/TrackProcessorContext";
-import { type MediaDevices } from "../MediaDevices";
-import { and$ } from "../../utils/observable";
-import { areLivekitTransportsEqual } from "../remoteMembers/matrixLivekitMerger";
-import {
-  enterRTCSession,
-  type EnterRTCSessionOptions,
-} from "../../rtcSessionHelpers";
-import { type ElementCallError } from "../../utils/errors";
-import { ElementWidgetActions, type WidgetHelpers } from "../../widget";
-
-enum LivekitState {
-  UNINITIALIZED = "uninitialized",
-  CONNECTING = "connecting",
-  CONNECTED = "connected",
-  ERROR = "error",
-  DISCONNECTED = "disconnected",
-  DISCONNECTING = "disconnecting",
-}
-type LocalMemberLivekitState =
-  | { state: LivekitState.ERROR; error: string }
-  | { state: LivekitState.CONNECTED }
-  | { state: LivekitState.CONNECTING }
-  | { state: LivekitState.UNINITIALIZED }
-  | { state: LivekitState.DISCONNECTED }
-  | { state: LivekitState.DISCONNECTING };
-
-enum MatrixState {
-  CONNECTED = "connected",
-  DISCONNECTED = "disconnected",
-  CONNECTING = "connecting",
-}
-type LocalMemberMatrixState =
-  | { state: MatrixState.CONNECTED }
-  | { state: MatrixState.CONNECTING }
-  | { state: MatrixState.DISCONNECTED };
-
-export interface LocalMemberState {
-  livekit$: BehaviorSubject<LocalMemberLivekitState>;
-  matrix$: BehaviorSubject<LocalMemberMatrixState>;
-}
-
-/*
- * - get well known
- * - get oldest membership
- * - get transport to use
- * - get openId + jwt token
- * - wait for createTrack() call
- *    - create tracks
- * - wait for join() call
- *   - Publisher.publishTracks()
- *   - send join state/sticky event
- */
-interface Props {
-  options: Behavior<EnterRTCSessionOptions>;
-  scope: ObservableScope;
-  mediaDevices: MediaDevices;
-  muteStates: MuteStates;
-  connectionManager: ReturnType<typeof createConnectionManager$>;
-  matrixRTCSession: MatrixRTCSession;
-  matrixRoom: MatrixRoom;
-  localTransport$: Behavior<LivekitTransport | undefined>;
-  e2eeLivekitOptions: E2EEOptions | undefined;
-  trackProcessorState$: Behavior<ProcessorState>;
-  widget: WidgetHelpers | null;
-}
-
-/**
- * This class is responsible for managing the own membership in a room.
- * We want
- *  - a publisher
- *  -
- * @param param0
- * @returns
- *  - publisher: The handle to create tracks and publish them to the room.
- *  - connected$: the current connection state. Including matrix server and livekit server connection. (only the livekit server relevant for our own participation)
- *  - transport$: the transport object the ownMembership$ ended up using.
- *
- */
-export const createLocalMembership$ = ({
-  scope,
-  options,
-  muteStates,
-  mediaDevices,
-  connectionManager,
-  matrixRTCSession,
-  localTransport$,
-  matrixRoom,
-  e2eeLivekitOptions,
-  trackProcessorState$,
-  widget,
-}: Props): {
-  // publisher: Publisher
-  requestConnect: () => LocalMemberState;
-  startTracks: () => Behavior<LocalTrack[]>;
-  requestDisconnect: () => Observable<LocalMemberLivekitState> | null;
-  state: LocalMemberState; // TODO this is probably superseeded by joinState$
-  homeserverConnected$: Behavior<boolean>;
-  connected$: Behavior<boolean>;
-  reconnecting$: Behavior<boolean>;
-  configError$: Behavior<ElementCallError | null>;
-} => {
-  const state = {
-    livekit$: new BehaviorSubject<LocalMemberLivekitState>({
-      state: LivekitState.UNINITIALIZED,
-    }),
-    matrix$: new BehaviorSubject<LocalMemberMatrixState>({
-      state: MatrixState.DISCONNECTED,
-    }),
-  };
-
-  // This should be used in a combineLatest with publisher$ to connect.
-  // to make it possible to call startTracks before the preferredTransport$ has resolved.
-  const shouldStartTracks$ = new BehaviorSubject(false);
-
-  // This should be used in a combineLatest with publisher$ to connect.
-  const tracks$ = new BehaviorSubject<LocalTrack[]>([]);
-
-  const connection$ = scope.behavior(
-    combineLatest(
-      [connectionManager.connections$, localTransport$],
-      (connections, transport) => {
-        if (transport === undefined) return undefined;
-        return connections.find((connection) =>
-          areLivekitTransportsEqual(connection.transport, transport),
-        );
-      },
-    ),
-  );
-  /**
-   * Whether we are connected to the MatrixRTC session.
-   */
-  const homeserverConnected$ = scope.behavior(
-    // To consider ourselves connected to MatrixRTC, we check the following:
-    and$(
-      // The client is connected to the sync loop
-      (
-        fromEvent(matrixRoom.client, ClientEvent.Sync) as Observable<
-          [SyncState]
-        >
-      ).pipe(
-        startWith([matrixRoom.client.getSyncState()]),
-        map(([state]) => state === SyncState.Syncing),
-      ),
-      // Room state observed by session says we're connected
-      fromEvent(matrixRTCSession, MembershipManagerEvent.StatusChanged).pipe(
-        startWith(null),
-        map(() => matrixRTCSession.membershipStatus === Status.Connected),
-      ),
-      // Also watch out for warnings that we've likely hit a timeout and our
-      // delayed leave event is being sent (this condition is here because it
-      // provides an earlier warning than the sync loop timeout, and we wouldn't
-      // see the actual leave event until we reconnect to the sync loop)
-      fromEvent(matrixRTCSession, MembershipManagerEvent.ProbablyLeft).pipe(
-        startWith(null),
-        map(() => matrixRTCSession.probablyLeft !== true),
-      ),
-    ),
-  );
-
-  // /**
-  //  * Whether we are "fully" connected to the call. Accounts for both the
-  //  * connection to the MatrixRTC session and the LiveKit publish connection.
-  //  */
-  // // TODO use this in combination with the MemberState.
-  const connected$ = scope.behavior(
-    and$(
-      homeserverConnected$,
-      connection$.pipe(
-        switchMap((c) =>
-          c
-            ? c.state$.pipe(map((state) => state.state === "ConnectedToLkRoom"))
-            : of(false),
-        ),
-      ),
-    ),
-  );
-
-  const publisher$ = scope.behavior(
-    connection$.pipe(
-      map((connection) =>
-        connection
-          ? new Publisher(
-              scope,
-              connection,
-              mediaDevices,
-              muteStates,
-              e2eeLivekitOptions,
-              trackProcessorState$,
-            )
-          : null,
-      ),
-    ),
-  );
-
-  combineLatest(
-    [publisher$, shouldStartTracks$],
-    (publisher, shouldStartTracks) => {
-      if (publisher && shouldStartTracks) {
-        publisher
-          .createAndSetupTracks()
-          .then((tracks) => {
-            tracks$.next(tracks);
-          })
-          .catch((error) => {
-            logger.error("Error creating tracks:", error);
-          });
-      }
-    },
-  );
-
-  // MATRIX RELATED
-
-  // /**
-  //  * Whether we should tell the user that we're reconnecting to the call.
-  //  */
-  // DISCUSSION is there a better way to do this?
-  // sth that is more deriectly implied from the membership manager of the js sdk. (fromEvent(matrixRTCSession, Reconnecting)) ??? or similar
-  const reconnecting$ = scope.behavior(
-    connected$.pipe(
-      // We are reconnecting if we previously had some successful initial
-      // connection but are now disconnected
-      scan(
-        ({ connectedPreviously }, connectedNow) => ({
-          connectedPreviously: connectedPreviously || connectedNow,
-          reconnecting: connectedPreviously && !connectedNow,
-        }),
-        { connectedPreviously: false, reconnecting: false },
-      ),
-      map(({ reconnecting }) => reconnecting),
-    ),
-  );
-
-  const startTracks = (): Behavior<LocalTrack[]> => {
-    shouldStartTracks$.next(true);
-    return tracks$;
-  };
-
-  const requestConnect = (): LocalMemberState => {
-    if (state.livekit$.value === null) {
-      startTracks();
-      state.livekit$.next({ state: LivekitState.CONNECTING });
-      combineLatest([publisher$, tracks$], (publisher, tracks) => {
-        publisher
-          ?.startPublishing()
-          .then(() => {
-            state.livekit$.next({ state: LivekitState.CONNECTED });
-          })
-          .catch((error) => {
-            state.livekit$.next({ state: LivekitState.ERROR, error });
-          });
-      });
-    }
-    if (state.matrix$.value.state !== MatrixState.DISCONNECTED) {
-      state.matrix$.next({ state: MatrixState.CONNECTING });
-      localTransport$.pipe(
-        tap((transport) => {
-          if (transport !== undefined) {
-            enterRTCSession(matrixRTCSession, transport, options.value).catch(
-              (error) => {
-                logger.error(error);
-              },
-            );
-          } else {
-            logger.info("Waiting for transport to enter rtc session");
-          }
-        }),
-      );
-    }
-    return state;
-  };
-
-  const requestDisconnect = (): Behavior<LocalMemberLivekitState> | null => {
-    if (state.livekit$.value.state !== LivekitState.CONNECTED) return null;
-    state.livekit$.next({ state: LivekitState.DISCONNECTING });
-    combineLatest([publisher$, tracks$], (publisher, tracks) => {
-      publisher
-        ?.stopPublishing()
-        .then(() => {
-          tracks.forEach((track) => track.stop());
-          state.livekit$.next({ state: LivekitState.DISCONNECTED });
-        })
-        .catch((error) => {
-          state.livekit$.next({ state: LivekitState.ERROR, error });
-        });
-    });
-
-    return state.livekit$;
-  };
-
-  // Pause upstream of all local media tracks when we're disconnected from
-  // MatrixRTC, because it can be an unpleasant surprise for the app to say
-  // 'reconnecting' and yet still be transmitting your media to others.
-  // We use matrixConnected$ rather than reconnecting$ because we want to
-  // pause tracks during the initial joining sequence too until we're sure
-  // that our own media is displayed on screen.
-  combineLatest([connection$, homeserverConnected$])
-    .pipe(scope.bind())
-    .subscribe(([connection, connected]) => {
-      if (connection?.state$.value.state !== "ConnectedToLkRoom") return;
-      const publications =
-        connection.livekitRoom.localParticipant.trackPublications.values();
-      if (connected) {
-        for (const p of publications) {
-          if (p.track?.isUpstreamPaused === true) {
-            const kind = p.track.kind;
-            logger.log(`Resuming ${kind} track (MatrixRTC connection present)`);
-            p.track
-              .resumeUpstream()
-              .catch((e) =>
-                logger.error(
-                  `Failed to resume ${kind} track after MatrixRTC reconnection`,
-                  e,
-                ),
-              );
-          }
-        }
-      } else {
-        for (const p of publications) {
-          if (p.track?.isUpstreamPaused === false) {
-            const kind = p.track.kind;
-            logger.log(
-              `Pausing ${kind} track (uncertain MatrixRTC connection)`,
-            );
-            p.track
-              .pauseUpstream()
-              .catch((e) =>
-                logger.error(
-                  `Failed to pause ${kind} track after entering uncertain MatrixRTC connection`,
-                  e,
-                ),
-              );
-          }
-        }
-      }
-    });
-
-  const configError$ = new BehaviorSubject<ElementCallError | null>(null);
-  // TODO I do not fully understand what this does.
-  // Is it needed?
-  // Is this at the right place?
-  // Can this be simplified?
-  // Start and stop session membership as needed
-  scope.reconcile(localTransport$, async (advertised) => {
-    if (advertised !== null && advertised !== undefined) {
-      try {
-        configError$.next(null);
-        await enterRTCSession(matrixRTCSession, advertised, options.value);
-      } catch (e) {
-        logger.error("Error entering RTC session", e);
-      }
-
-      // Update our member event when our mute state changes.
-      const intentScope = new ObservableScope();
-      intentScope.reconcile(muteStates.video.enabled$, async (videoEnabled) =>
-        matrixRTCSession.updateCallIntent(videoEnabled ? "video" : "audio"),
-      );
-
-      return async (): Promise<void> => {
-        intentScope.end();
-        // Only sends Matrix leave event. The LiveKit session will disconnect
-        // as soon as either the stopConnection$ handler above gets to it or
-        // the view model is destroyed.
-        try {
-          await matrixRTCSession.leaveRoomSession();
-        } catch (e) {
-          logger.error("Error leaving RTC session", e);
-        }
-        try {
-          await widget?.api.transport.send(ElementWidgetActions.HangupCall, {});
-        } catch (e) {
-          logger.error("Failed to send hangup action", e);
-        }
-      };
-    }
-  });
-
-  return {
-    startTracks,
-    requestConnect,
-    requestDisconnect,
-    state,
-    homeserverConnected$,
-    connected$,
-    reconnecting$,
-    configError$,
-  };
-};

src/state/remoteMembers/ConnectionManager.test.ts

@@ -1,294 +0,0 @@
-/*
-Copyright 2025 Element Creations Ltd.
-
-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, test, vi } from "vitest";
-import { BehaviorSubject } from "rxjs";
-import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc";
-import { type Participant as LivekitParticipant } from "livekit-client";
-
-import { ObservableScope } from "../ObservableScope.ts";
-import { createConnectionManager$ } from "./ConnectionManager.ts";
-import { type ConnectionFactory } from "./ConnectionFactory.ts";
-import { type Connection } from "./Connection.ts";
-import { areLivekitTransportsEqual } from "./matrixLivekitMerger.ts";
-import { flushPromises, withTestScheduler } from "../../utils/test.ts";
-
-// Some test constants
-
-const TRANSPORT_1: LivekitTransport = {
-  type: "livekit",
-  livekit_service_url: "https://lk.example.org",
-  livekit_alias: "!alias:example.org",
-};
-
-const TRANSPORT_2: LivekitTransport = {
-  type: "livekit",
-  livekit_service_url: "https://lk.sample.com",
-  livekit_alias: "!alias:sample.com",
-};
-
-// const TRANSPORT_3: LivekitTransport = {
-//   type: "livekit",
-//   livekit_service_url: "https://lk-other.sample.com",
-//   livekit_alias: "!alias:sample.com",
-// };
-let fakeConnectionFactory: ConnectionFactory;
-let testScope: ObservableScope;
-let testTransportStream$: BehaviorSubject<LivekitTransport[]>;
-let connectionManagerInputs: {
-  scope: ObservableScope;
-  connectionFactory: ConnectionFactory;
-  inputTransports$: BehaviorSubject<LivekitTransport[]>;
-};
-let manager: ReturnType<typeof createConnectionManager$>;
-beforeEach(() => {
-  testScope = new ObservableScope();
-
-  fakeConnectionFactory = {} as unknown as ConnectionFactory;
-  vi.mocked(fakeConnectionFactory).createConnection = vi
-    .fn()
-    .mockImplementation(
-      (transport: LivekitTransport, scope: ObservableScope) => {
-        const mockConnection = {
-          transport,
-        } as unknown as Connection;
-        vi.mocked(mockConnection).start = vi.fn();
-        vi.mocked(mockConnection).stop = vi.fn();
-        // Tie the connection's lifecycle to the scope to test scope lifecycle management
-        scope.onEnd(() => {
-          void mockConnection.stop();
-        });
-        return mockConnection;
-      },
-    );
-
-  testTransportStream$ = new BehaviorSubject<LivekitTransport[]>([]);
-  connectionManagerInputs = {
-    scope: testScope,
-    connectionFactory: fakeConnectionFactory,
-    inputTransports$: testTransportStream$,
-  };
-  manager = createConnectionManager$(connectionManagerInputs);
-});
-
-afterEach(() => {
-  testScope.end();
-});
-
-describe("connections$ stream", () => {
-  test("Should create and start new connections for each transports", async () => {
-    const managedConnections = Promise.withResolvers<Connection[]>();
-    manager.connections$.subscribe((connections) => {
-      if (connections.length > 0) managedConnections.resolve(connections);
-    });
-
-    connectionManagerInputs.inputTransports$.next([TRANSPORT_1, TRANSPORT_2]);
-
-    const connections = await managedConnections.promise;
-
-    expect(connections.length).toBe(2);
-
-    expect(
-      vi.mocked(fakeConnectionFactory).createConnection,
-    ).toHaveBeenCalledTimes(2);
-
-    const conn1 = connections.find((c) =>
-      areLivekitTransportsEqual(c.transport, TRANSPORT_1),
-    );
-    expect(conn1).toBeDefined();
-    expect(conn1!.start).toHaveBeenCalled();
-
-    const conn2 = connections.find((c) =>
-      areLivekitTransportsEqual(c.transport, TRANSPORT_2),
-    );
-    expect(conn2).toBeDefined();
-    expect(conn2!.start).toHaveBeenCalled();
-  });
-
-  test("Should start connection only once", async () => {
-    const observedConnections: Connection[][] = [];
-    manager.connections$.subscribe((connections) => {
-      observedConnections.push(connections);
-    });
-
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1, TRANSPORT_2]);
-
-    await flushPromises();
-    const connections = observedConnections.pop()!;
-
-    expect(connections.length).toBe(2);
-    expect(
-      vi.mocked(fakeConnectionFactory).createConnection,
-    ).toHaveBeenCalledTimes(2);
-
-    const conn2 = connections.find((c) =>
-      areLivekitTransportsEqual(c.transport, TRANSPORT_2),
-    );
-    expect(conn2).toBeDefined();
-
-    const conn1 = connections.find((c) =>
-      areLivekitTransportsEqual(c.transport, TRANSPORT_1),
-    );
-    expect(conn1).toBeDefined();
-    expect(conn1!.start).toHaveBeenCalledOnce();
-  });
-
-  test("Should cleanup connections when not needed anymore", async () => {
-    const observedConnections: Connection[][] = [];
-    manager.connections$.subscribe((connections) => {
-      observedConnections.push(connections);
-    });
-
-    testTransportStream$.next([TRANSPORT_1]);
-    testTransportStream$.next([TRANSPORT_1, TRANSPORT_2]);
-
-    await flushPromises();
-
-    const conn2 = observedConnections
-      .pop()!
-      .find((c) => areLivekitTransportsEqual(c.transport, TRANSPORT_2))!;
-
-    testTransportStream$.next([TRANSPORT_1]);
-
-    await flushPromises();
-
-    // The second connection should have been stopped has it is no longer needed
-    expect(conn2.stop).toHaveBeenCalled();
-
-    // The first connection should still be active
-    const conn1 = observedConnections.pop()![0];
-    expect(conn1.stop).not.toHaveBeenCalledOnce();
-  });
-});
-
-describe("connectionManagerData$ stream", () => {
-  // Used in test to control fake connections' participantsWithTrack$ streams
-  let fakePublishingParticipantsStreams: Map<
-    string,
-    BehaviorSubject<LivekitParticipant[]>
-  >;
-
-  function keyForTransport(transport: LivekitTransport): string {
-    return `${transport.livekit_service_url}|${transport.livekit_alias}`;
-  }
-
-  beforeEach(() => {
-    fakePublishingParticipantsStreams = new Map();
-    // need a more advanced fake connection factory
-    vi.mocked(fakeConnectionFactory).createConnection = vi
-      .fn()
-      .mockImplementation(
-        (transport: LivekitTransport, scope: ObservableScope) => {
-          const fakePublishingParticipants$ = new BehaviorSubject<
-            LivekitParticipant[]
-          >([]);
-          const mockConnection = {
-            transport,
-            participantsWithTrack$: fakePublishingParticipants$,
-          } as unknown as Connection;
-          vi.mocked(mockConnection).start = vi.fn();
-          vi.mocked(mockConnection).stop = vi.fn();
-          // Tie the connection's lifecycle to the scope to test scope lifecycle management
-          scope.onEnd(() => {
-            void mockConnection.stop();
-          });
-
-          fakePublishingParticipantsStreams.set(
-            keyForTransport(transport),
-            fakePublishingParticipants$,
-          );
-          return mockConnection;
-        },
-      );
-  });
-
-  test("Should report connections with the publishing participants", () => {
-    withTestScheduler(({ expectObservable, schedule, behavior }) => {
-      manager = createConnectionManager$({
-        ...connectionManagerInputs,
-        inputTransports$: behavior("a", {
-          a: [TRANSPORT_1, TRANSPORT_2],
-        }),
-      });
-
-      const conn1Participants$ = fakePublishingParticipantsStreams.get(
-        keyForTransport(TRANSPORT_1),
-      )!;
-
-      schedule("-a-b", {
-        a: () => {
-          conn1Participants$.next([
-            { identity: "user1A" } as LivekitParticipant,
-          ]);
-        },
-        b: () => {
-          conn1Participants$.next([
-            { identity: "user1A" } as LivekitParticipant,
-            { identity: "user1B" } as LivekitParticipant,
-          ]);
-        },
-      });
-
-      const conn2Participants$ = fakePublishingParticipantsStreams.get(
-        keyForTransport(TRANSPORT_2),
-      )!;
-
-      schedule("--a", {
-        a: () => {
-          conn2Participants$.next([
-            { identity: "user2A" } as LivekitParticipant,
-          ]);
-        },
-      });
-
-      expectObservable(manager.connectionManagerData$).toBe("abcd", {
-        a: expect.toSatisfy((data) => {
-          return (
-            data.getConnections().length == 2 &&
-            data.getParticipantForTransport(TRANSPORT_1).length == 0 &&
-            data.getParticipantForTransport(TRANSPORT_2).length == 0
-          );
-        }),
-        b: expect.toSatisfy((data) => {
-          return (
-            data.getConnections().length == 2 &&
-            data.getParticipantForTransport(TRANSPORT_1).length == 1 &&
-            data.getParticipantForTransport(TRANSPORT_2).length == 0 &&
-            data.getParticipantForTransport(TRANSPORT_1)[0].identity == "user1A"
-          );
-        }),
-        c: expect.toSatisfy((data) => {
-          return (
-            data.getConnections().length == 2 &&
-            data.getParticipantForTransport(TRANSPORT_1).length == 1 &&
-            data.getParticipantForTransport(TRANSPORT_2).length == 1 &&
-            data.getParticipantForTransport(TRANSPORT_1)[0].identity ==
-              "user1A" &&
-            data.getParticipantForTransport(TRANSPORT_2)[0].identity == "user2A"
-          );
-        }),
-        d: expect.toSatisfy((data) => {
-          return (
-            data.getConnections().length == 2 &&
-            data.getParticipantForTransport(TRANSPORT_1).length == 2 &&
-            data.getParticipantForTransport(TRANSPORT_2).length == 1 &&
-            data.getParticipantForTransport(TRANSPORT_1)[0].identity ==
-              "user1A" &&
-            data.getParticipantForTransport(TRANSPORT_1)[1].identity ==
-              "user1B" &&
-            data.getParticipantForTransport(TRANSPORT_2)[0].identity == "user2A"
-          );
-        }),
-      });
-    });
-  });
-});

src/state/remoteMembers/MatrixLivekitMembers.ts

@@ -1,161 +0,0 @@
-/*
-Copyright 2025 Element Creations Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import {
-  type LocalParticipant as LocalLivekitParticipant,
-  type RemoteParticipant as RemoteLivekitParticipant,
-} from "livekit-client";
-import {
-  type LivekitTransport,
-  type CallMembership,
-} from "matrix-js-sdk/lib/matrixrtc";
-import { combineLatest, map, startWith, type Observable } from "rxjs";
-// eslint-disable-next-line rxjs/no-internal
-import { type NodeStyleEventEmitter } from "rxjs/internal/observable/fromEvent";
-import { type Room as MatrixRoom, type RoomMember } from "matrix-js-sdk";
-
-// import type { Logger } from "matrix-js-sdk/lib/logger";
-import { type Behavior } from "../Behavior";
-import { type ObservableScope } from "../ObservableScope";
-import { type createConnectionManager$ } from "./ConnectionManager";
-import { getRoomMemberFromRtcMember, memberDisplaynames$ } from "./displayname";
-import { type Connection } from "./Connection";
-
-/**
- * Represent a matrix call member and his associated livekit participation.
- * `livekitParticipant` can be undefined if the member is not yet connected to the livekit room
- * or if it has no livekit transport at all.
- */
-export interface MatrixLivekitMember {
-  membership: CallMembership;
-  displayName$: Behavior<string>;
-  participant?: LocalLivekitParticipant | RemoteLivekitParticipant;
-  connection?: Connection;
-  /**
-   * TODO Try to remove this! Its waaay to much information.
-   * Just get the member's avatar
-   * @deprecated
-   */
-  member: RoomMember;
-  mxcAvatarUrl?: string;
-  participantId: string;
-}
-
-interface Props {
-  scope: ObservableScope;
-  membershipsWithTransport$: Behavior<
-    { membership: CallMembership; transport?: LivekitTransport }[]
-  >;
-  connectionManager: ReturnType<typeof createConnectionManager$>;
-  // TODO this is too much information for that class,
-  // apparently needed to get a room member to later get the Avatar
-  // => Extract an AvatarService instead?
-  // Better with just `getMember`
-  matrixRoom: Pick<MatrixRoom, "getMember"> & NodeStyleEventEmitter;
-  userId: string;
-  deviceId: string;
-}
-// Alternative structure idea:
-// const livekitMatrixMember$ = (callMemberships$,connectionManager,scope): Observable<MatrixLivekitMember[]> => {
-
-/**
- * Combines MatrixRTC and Livekit worlds.
- *
- * It has a small public interface:
- *  - in (via constructor):
- *    - an observable of CallMembership[] to track the call members (The matrix side)
- *    - a `ConnectionManager` for the lk rooms (The livekit side)
- *  - out (via public Observable):
- *    - `remoteMatrixLivekitMember` an observable of MatrixLivekitMember[] to track the remote members and associated livekit data.
- */
-export function createMatrixLivekitMembers$({
-  scope,
-  membershipsWithTransport$,
-  connectionManager,
-  matrixRoom,
-  userId,
-  deviceId,
-}: Props): Behavior<MatrixLivekitMember[]> {
-  /**
-   * Stream of all the call members and their associated livekit data (if available).
-   */
-
-  function createMatrixLivekitMember$(): Observable<MatrixLivekitMember[]> {
-    const displaynameMap$ = memberDisplaynames$(
-      scope,
-      matrixRoom,
-      membershipsWithTransport$.pipe(map((v) => v.map((v) => v.membership))),
-      userId,
-      deviceId,
-    );
-
-    return combineLatest([
-      membershipsWithTransport$,
-      connectionManager.connectionManagerData$,
-    ]).pipe(
-      map(([memberships, managerData]) => {
-        const items: MatrixLivekitMember[] = memberships.map(
-          ({ membership, transport }) => {
-            // TODO! cannot use membership.membershipID yet, Currently its hardcoded by the jwt service to
-            const participantId = /*membership.membershipID*/ `${membership.userId}:${membership.deviceId}`;
-
-            const participants = transport
-              ? managerData.getParticipantForTransport(transport)
-              : [];
-            const participant = participants.find(
-              (p) => p.identity == participantId,
-            );
-            const member = getRoomMemberFromRtcMember(
-              membership,
-              matrixRoom,
-            )?.member;
-            const connection = transport
-              ? managerData.getConnectionForTransport(transport)
-              : undefined;
-            const displayName$ = scope.behavior(
-              displaynameMap$.pipe(
-                map(
-                  (displayNameMap) =>
-                    displayNameMap.get(membership.membershipID) ?? "---",
-                ),
-              ),
-            );
-            return {
-              participant,
-              membership,
-              connection,
-              // This makes sense to add the the js-sdk callMembership (we only need the avatar so probably the call memberhsip just should aquire the avatar)
-              // TODO Ugh this is hidign that it might be undefined!! best we remove the member entirely.
-              member: member as RoomMember,
-              displayName$,
-              mxcAvatarUrl: member?.getMxcAvatarUrl(),
-              participantId,
-            };
-          },
-        );
-        return items;
-      }),
-    );
-  }
-
-  return scope.behavior(createMatrixLivekitMember$().pipe(startWith([])));
-}
-
-// TODO add back in the callviewmodel pauseWhen(this.pretendToBeDisconnected$)
-
-// TODO add this to the JS-SDK
-export function areLivekitTransportsEqual(
-  t1: LivekitTransport,
-  t2: LivekitTransport,
-): boolean {
-  return (
-    t1.livekit_service_url === t2.livekit_service_url &&
-    // In case we have different lk rooms in the same SFU (depends on the livekit authorization service)
-    // It is only needed in case the livekit authorization service is not behaving as expected (or custom implementation)
-    t1.livekit_alias === t2.livekit_alias
-  );
-}

src/state/remoteMembers/MatrixLivekitMerger.test.ts

@@ -1,262 +0,0 @@
-/*
-Copyright 2025 Element Creations Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import {
-  describe,
-  test,
-  vi,
-  expect,
-  beforeEach,
-  afterEach,
-  type MockedObject,
-} from "vitest";
-import { BehaviorSubject, take } from "rxjs";
-import {
-  type CallMembership,
-  type LivekitTransport,
-} from "matrix-js-sdk/lib/matrixrtc";
-import { type Room as MatrixRoom } from "matrix-js-sdk";
-import { getParticipantId } from "matrix-js-sdk/lib/matrixrtc/utils";
-
-import {
-  type MatrixLivekitMember,
-  matrixLivekitMerger$,
-} from "./matrixLivekitMerger";
-import { ObservableScope } from "../ObservableScope";
-import {
-  type ConnectionManager,
-  ConnectionManagerData,
-} from "./ConnectionManager";
-import { aliceRtcMember } from "../../utils/test-fixtures";
-import { mockRemoteParticipant } from "../../utils/test.ts";
-import { type Connection } from "./Connection.ts";
-
-let testScope: ObservableScope;
-let fakeManagerData$: BehaviorSubject<ConnectionManagerData>;
-let fakeMemberships$: BehaviorSubject<CallMembership[]>;
-let mockConnectionManager: MockedObject<ConnectionManager>;
-let mockMatrixRoom: MatrixRoom;
-const userId = "@local:example.com";
-const deviceId = "DEVICE000";
-
-// The merger beeing tested
-let matrixLivekitMerger: matrixLivekitMerger$;
-
-beforeEach(() => {
-  testScope = new ObservableScope();
-  fakeMemberships$ = new BehaviorSubject<CallMembership[]>([]);
-  fakeManagerData$ = new BehaviorSubject<ConnectionManagerData>(
-    new ConnectionManagerData(),
-  );
-  mockConnectionManager = vi.mocked<ConnectionManager>({
-    registerTransports: vi.fn(),
-    connectionManagerData$: fakeManagerData$,
-  } as unknown as ConnectionManager);
-  mockMatrixRoom = vi.mocked<MatrixRoom>({
-    getMember: vi.fn().mockReturnValue(null),
-    addEventListener: vi.fn(),
-    removeEventListener: vi.fn(),
-  } as unknown as MatrixRoom);
-
-  matrixLivekitMerger = new matrixLivekitMerger$(
-    testScope,
-    fakeMemberships$,
-    mockConnectionManager,
-    mockMatrixRoom,
-    userId,
-    deviceId,
-  );
-});
-
-afterEach(() => {
-  testScope.end();
-});
-
-test("should signal participant not yet connected to livekit", () => {
-  fakeMemberships$.next([aliceRtcMember]);
-
-  let items: MatrixLivekitMember[] = [];
-  matrixLivekitMerger.matrixLivekitMember$
-    .pipe(take(1))
-    .subscribe((emitted) => {
-      items = emitted;
-    });
-
-  expect(items).toHaveLength(1);
-  const item = items[0];
-
-  // Assert the expected membership
-  expect(item.membership).toBe(aliceRtcMember);
-
-  // Assert participant & connection are absent (not just `undefined`)
-  expect(item.participant).not.toBeDefined();
-  expect(item.participant).not.toBeDefined();
-});
-
-test("should signal participant on a connection that is publishing", () => {
-  const fakeConnection = {
-    transport: aliceRtcMember.getTransport(aliceRtcMember) as LivekitTransport,
-  } as unknown as Connection;
-
-  fakeMemberships$.next([aliceRtcMember]);
-  const aliceParticipantId = getParticipantId(
-    aliceRtcMember.userId,
-    aliceRtcMember.deviceId,
-  );
-
-  const managerData: ConnectionManagerData = new ConnectionManagerData();
-  managerData.add(fakeConnection, [
-    mockRemoteParticipant({ identity: aliceParticipantId }),
-  ]);
-  fakeManagerData$.next(managerData);
-
-  let items: MatrixLivekitMember[] = [];
-  matrixLivekitMerger.matrixLivekitMember$
-    .pipe(take(1))
-    .subscribe((emitted) => {
-      items = emitted;
-    });
-  expect(items).toHaveLength(1);
-  const item = items[0];
-
-  // Assert the expected membership
-  expect(item.membership).toBe(aliceRtcMember);
-  expect(item.participant?.identity).toBe(aliceParticipantId);
-  expect(item.connection?.transport).toEqual(fakeConnection.transport);
-});
-
-test("should signal participant on a connection that is not publishing", () => {
-  const fakeConnection = {
-    transport: aliceRtcMember.getTransport(aliceRtcMember) as LivekitTransport,
-  } as unknown as Connection;
-
-  fakeMemberships$.next([aliceRtcMember]);
-
-  const managerData: ConnectionManagerData = new ConnectionManagerData();
-  managerData.add(fakeConnection, []);
-  fakeManagerData$.next(managerData);
-
-  matrixLivekitMerger.matrixLivekitMember$.pipe(take(1)).subscribe((items) => {
-    expect(items).toHaveLength(1);
-    const item = items[0];
-
-    // Assert the expected membership
-    expect(item.membership).toBe(aliceRtcMember);
-    expect(item.participant).not.toBeDefined();
-    // We have the connection
-    expect(item.connection?.transport).toEqual(fakeConnection.transport);
-  });
-});
-
-describe("Publication edge case", () => {
-  const connectionA = {
-    transport: {
-      type: "livekit",
-      livekit_service_url: "https://lk.example.org",
-      livekit_alias: "!alias:example.org",
-    },
-  } as unknown as Connection;
-
-  const connectionB = {
-    transport: {
-      type: "livekit",
-      livekit_service_url: "https://lk.sample.com",
-      livekit_alias: "!alias:sample.com",
-    },
-  } as unknown as Connection;
-
-  const bobMembership = {
-    userId: "@bob:example.org",
-    deviceId: "DEV000",
-    transports: [connectionA.transport],
-  } as unknown as CallMembership;
-
-  const bobParticipantId = getParticipantId(
-    bobMembership.userId,
-    bobMembership.deviceId,
-  );
-
-  test("bob is publishing in several connections", () => {
-    let lastMatrixLkItems: MatrixLivekitMember[] = [];
-    matrixLivekitMerger.matrixLivekitMember$.subscribe((items) => {
-      lastMatrixLkItems = items;
-    });
-
-    vi.mocked(bobMembership).getTransport = vi
-      .fn()
-      .mockReturnValue(connectionA.transport);
-
-    fakeMemberships$.next([bobMembership]);
-
-    const lkMap = new ConnectionManagerData();
-    lkMap.add(connectionA, [
-      mockRemoteParticipant({ identity: bobParticipantId }),
-    ]);
-    lkMap.add(connectionB, [
-      mockRemoteParticipant({ identity: bobParticipantId }),
-    ]);
-
-    fakeManagerData$.next(lkMap);
-
-    const items = lastMatrixLkItems;
-    expect(items).toHaveLength(1);
-    const item = items[0];
-
-    // Assert the expected membership
-    expect(item.membership.userId).toEqual(bobMembership.userId);
-    expect(item.membership.deviceId).toEqual(bobMembership.deviceId);
-
-    expect(item.participant?.identity).toEqual(bobParticipantId);
-
-    // The transport info should come from the membership transports and not only from the publishing connection
-    expect(item.connection?.transport?.livekit_service_url).toEqual(
-      bobMembership.transports[0]?.livekit_service_url,
-    );
-    expect(item.connection?.transport?.livekit_alias).toEqual(
-      bobMembership.transports[0]?.livekit_alias,
-    );
-  });
-
-  test("bob is publishing in the wrong connection", () => {
-    let lastMatrixLkItems: MatrixLivekitMember[] = [];
-    matrixLivekitMerger.matrixLivekitMember$.subscribe((items) => {
-      lastMatrixLkItems = items;
-    });
-
-    vi.mocked(bobMembership).getTransport = vi
-      .fn()
-      .mockReturnValue(connectionA.transport);
-
-    fakeMemberships$.next([bobMembership]);
-
-    const lkMap = new ConnectionManagerData();
-    lkMap.add(connectionA, []);
-    lkMap.add(connectionB, [
-      mockRemoteParticipant({ identity: bobParticipantId }),
-    ]);
-
-    fakeManagerData$.next(lkMap);
-
-    const items = lastMatrixLkItems;
-    expect(items).toHaveLength(1);
-    const item = items[0];
-
-    // Assert the expected membership
-    expect(item.membership.userId).toEqual(bobMembership.userId);
-    expect(item.membership.deviceId).toEqual(bobMembership.deviceId);
-
-    expect(item.participant).not.toBeDefined();
-
-    // The transport info should come from the membership transports and not only from the publishing connection
-    expect(item.connection?.transport?.livekit_service_url).toEqual(
-      bobMembership.transports[0]?.livekit_service_url,
-    );
-    expect(item.connection?.transport?.livekit_alias).toEqual(
-      bobMembership.transports[0]?.livekit_alias,
-    );
-  });
-});

src/state/remoteMembers/displayname.ts

@@ -1,87 +0,0 @@
-/*
-Copyright 2025 New Vector Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import { type RoomMember, RoomStateEvent } from "matrix-js-sdk";
-import { combineLatest, fromEvent, type Observable, startWith } from "rxjs";
-import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
-import { logger } from "matrix-js-sdk/lib/logger";
-import { type Room as MatrixRoom } from "matrix-js-sdk/lib/matrix";
-// eslint-disable-next-line rxjs/no-internal
-import { type NodeStyleEventEmitter } from "rxjs/internal/observable/fromEvent";
-
-import { type ObservableScope } from "../ObservableScope";
-import {
-  calculateDisplayName,
-  shouldDisambiguate,
-} from "../../utils/displayname";
-import { type Behavior } from "../Behavior";
-
-/**
- * Displayname for each member of the call. This will disambiguate
- * any displayname that clashes with another member. Only members
- * joined to the call are considered here.
- *
- * @returns Map<member.id, displayname> uses the rtc member idenitfier as the key.
- */
-// don't do this work more times than we need to. This is achieved by converting to a behavior:
-export const memberDisplaynames$ = (
-  scope: ObservableScope,
-  matrixRoom: Pick<MatrixRoom, "getMember"> & NodeStyleEventEmitter,
-  memberships$: Observable<CallMembership[]>,
-  userId: string,
-  deviceId: string,
-): Behavior<Map<string, string>> =>
-  scope.behavior(
-    combineLatest(
-      [
-        // Handle call membership changes
-        memberships$,
-        // Additionally handle display name changes (implicitly reacting to them)
-        fromEvent(matrixRoom, RoomStateEvent.Members).pipe(startWith(null)),
-        // TODO: do we need: pauseWhen(this.pretendToBeDisconnected$),
-      ],
-      (memberships, _displaynames) => {
-        const displaynameMap = new Map<string, string>([
-          [
-            `${userId}:${deviceId}`,
-            matrixRoom.getMember(userId)?.rawDisplayName ?? userId,
-          ],
-        ]);
-        const room = matrixRoom;
-
-        // We only consider RTC members for disambiguation as they are the only visible members.
-        for (const rtcMember of memberships) {
-          const matrixIdentifier = `${rtcMember.userId}:${rtcMember.deviceId}`;
-          const { member } = getRoomMemberFromRtcMember(rtcMember, room);
-          if (!member) {
-            logger.error(
-              "Could not find member for media id:",
-              matrixIdentifier,
-            );
-            continue;
-          }
-          const disambiguate = shouldDisambiguate(member, memberships, room);
-          displaynameMap.set(
-            matrixIdentifier,
-            calculateDisplayName(member, disambiguate),
-          );
-        }
-        return displaynameMap;
-      },
-    ),
-    new Map<string, string>(),
-  );
-
-export function getRoomMemberFromRtcMember(
-  rtcMember: CallMembership,
-  room: Pick<MatrixRoom, "getMember">,
-): { id: string; member: RoomMember | undefined } {
-  return {
-    id: rtcMember.userId + ":" + rtcMember.deviceId,
-    member: room.getMember(rtcMember.userId) ?? undefined,
-  };
-}

src/state/remoteMembers/integration.test.ts

@@ -1,157 +0,0 @@
-/*
-Copyright 2025 Element Creations Ltd.
-
-SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
-Please see LICENSE in the repository root for full details.
-*/
-
-import { test, vi, beforeEach, afterEach } from "vitest";
-import { BehaviorSubject, type Observable } from "rxjs";
-import { type Room as LivekitRoom } from "livekit-client";
-import { logger } from "matrix-js-sdk/lib/logger";
-import EventEmitter from "events";
-import fetchMock from "fetch-mock";
-
-import { ConnectionManager } from "./ConnectionManager.ts";
-import { ObservableScope } from "../ObservableScope.ts";
-import { ECConnectionFactory } from "./ConnectionFactory.ts";
-import { type OpenIDClientParts } from "../../livekit/openIDSFU.ts";
-import { mockMediaDevices, withTestScheduler } from "../../utils/test";
-import { type ProcessorState } from "../../livekit/TrackProcessorContext.tsx";
-import { matrixLivekitMerger$ } from "./matrixLivekitMerger.ts";
-import type { CallMembership, Transport } from "matrix-js-sdk/lib/matrixrtc";
-import { TRANSPORT_1 } from "./ConnectionManager.test.ts";
-
-// Test the integration of ConnectionManager and MatrixLivekitMerger
-
-let testScope: ObservableScope;
-let ecConnectionFactory: ECConnectionFactory;
-let mockClient: OpenIDClientParts;
-let lkRoomFactory: () => LivekitRoom;
-
-const createdMockLivekitRooms: Map<string, LivekitRoom> = new Map();
-
-// Main test input
-const memberships$ = new BehaviorSubject<CallMembership[]>([]);
-
-// under test
-let connectionManager: ConnectionManager;
-
-function createLkMerger(
-  memberships$: Observable<CallMembership[]>,
-): matrixLivekitMerger$ {
-  const mockRoomEmitter = new EventEmitter();
-  return new matrixLivekitMerger$(
-    testScope,
-    memberships$,
-    connectionManager,
-    {
-      on: mockRoomEmitter.on.bind(mockRoomEmitter),
-      off: mockRoomEmitter.off.bind(mockRoomEmitter),
-      getMember: vi.fn().mockReturnValue(undefined),
-    },
-    "@user:example.com",
-    "DEV000",
-  );
-}
-
-beforeEach(() => {
-  testScope = new ObservableScope();
-  mockClient = {
-    getOpenIdToken: vi.fn().mockReturnValue(""),
-    getDeviceId: vi.fn().mockReturnValue("DEV000"),
-  };
-
-  lkRoomFactory = vi.fn().mockImplementation(() => {
-    const emitter = new EventEmitter();
-    const base = {
-      on: emitter.on.bind(emitter),
-      off: emitter.off.bind(emitter),
-      emit: emitter.emit.bind(emitter),
-      disconnect: vi.fn(),
-      remoteParticipants: new Map(),
-    } as unknown as LivekitRoom;
-
-    vi.mocked(base).connect = vi.fn().mockImplementation(({ url }) => {
-      createdMockLivekitRooms.set(url, base);
-    });
-    return base;
-  });
-
-  ecConnectionFactory = new ECConnectionFactory(
-    mockClient,
-    mockMediaDevices({}),
-    new BehaviorSubject<ProcessorState>({
-      supported: true,
-      processor: undefined,
-    }),
-    undefined,
-    false,
-    lkRoomFactory,
-  );
-
-  connectionManager = new ConnectionManager(
-    testScope,
-    ecConnectionFactory,
-    logger,
-  );
-
-  //TODO a bit annoying to have to do a http mock?
-  fetchMock.post(`**/sfu/get`, (url) => {
-    const domain = new URL(url).hostname; // Extract the domain from the URL
-
-    return {
-      status: 200,
-      body: {
-        url: `wss://${domain}/livekit/sfu`,
-        jwt: "ATOKEN",
-      },
-    };
-  });
-});
-
-afterEach(() => {
-  testScope.end();
-  fetchMock.reset();
-});
-
-test("example test", () => {
-  withTestScheduler(({ schedule, expectObservable, cold }) => {
-    connectionManager.connections$.subscribe((connections) => {
-      // console.log(
-      //   "Connections updated:",
-      //   connections.map((c) => c.transport),
-      // );
-    });
-
-    const memberships$ = cold("-a-b-c", {
-      a: [mockCallmembership("@bob:example.com", "BDEV000")],
-      b: [
-        mockCallmembership("@bob:example.com", "BDEV000"),
-        mockCallmembership("@carl:example.com", "CDEV000"),
-      ],
-      c: [
-        mockCallmembership("@bob:example.com", "BDEV000"),
-        mockCallmembership("@carl:example.com", "CDEV000"),
-        mockCallmembership("@dave:foo.bar", "DDEV000"),
-      ],
-    });
-
-    // TODO IN PROGRESS
-    const merger = createLkMerger(memberships$);
-  });
-});
-
-function mockCallmembership(
-  userId: string,
-  deviceId: string,
-  transport?: Transport,
-): CallMembership {
-  const t = transport ?? TRANSPORT_1;
-  return {
-    userId: userId,
-    deviceId: deviceId,
-    getTransport: vi.fn().mockReturnValue(t),
-    transports: [t],
-  } as unknown as CallMembership;
-}

src/tile/GridTile.test.tsx

@@ -15,7 +15,7 @@ import { GridTile } from "./GridTile";
 import { mockRtcMembership, createRemoteMedia } from "../utils/test";
 import { GridTileViewModel } from "../state/TileViewModel";
 import { ReactionsSenderProvider } from "../reactions/useReactionsSender";
-import type { CallViewModel } from "../state/CallViewModel";
+import type { CallViewModel } from "../state/CallViewModel/CallViewModel";
 import { constant } from "../state/Behavior";
 
 global.IntersectionObserver = class MockIntersectionObserver {

src/tile/GridTile.tsx

@@ -58,7 +58,9 @@ interface TileProps {
   style?: ComponentProps<typeof animated.div>["style"];
   targetWidth: number;
   targetHeight: number;
+  focusUrl: string | undefined;
   displayName: string;
+  mxcAvatarUrl: string | undefined;
   showSpeakingIndicators: boolean;
   focusable: boolean;
 }
@@ -81,7 +83,9 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
   menuStart,
   menuEnd,
   className,
+  focusUrl,
   displayName,
+  mxcAvatarUrl,
   focusable,
   ...props
 }) => {
@@ -144,8 +148,8 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
   const tile = (
     <MediaView
       ref={ref}
-      video={video}
-      member={vm.member}
+      video={video ?? undefined}
+      userId={vm.userId}
       unencryptedWarning={unencryptedWarning}
       encryptionStatus={encryptionStatus}
       videoEnabled={videoEnabled}
@@ -164,6 +168,7 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
         />
       }
       displayName={displayName}
+      mxcAvatarUrl={mxcAvatarUrl}
       focusable={focusable}
       primaryButton={
         primaryButton ?? (
@@ -190,7 +195,7 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
       currentReaction={reaction ?? undefined}
       raisedHandOnClick={raisedHandOnClick}
       localParticipant={vm.local}
-      focusUrl={vm.focusURL}
+      focusUrl={focusUrl}
       audioStreamStats={audioStreamStats}
       videoStreamStats={videoStreamStats}
       {...props}
@@ -359,7 +364,9 @@ export const GridTile: FC<GridTileProps> = ({
   const ourRef = useRef<HTMLDivElement | null>(null);
   const ref = useMergedRefs(ourRef, theirRef);
   const media = useBehavior(vm.media$);
+  const focusUrl = useBehavior(media.focusUrl$);
   const displayName = useBehavior(media.displayName$);
+  const mxcAvatarUrl = useBehavior(media.mxcAvatarUrl$);
 
   if (media instanceof LocalUserMediaViewModel) {
     return (
@@ -367,7 +374,9 @@ export const GridTile: FC<GridTileProps> = ({
         ref={ref}
         vm={media}
         onOpenProfile={onOpenProfile}
+        focusUrl={focusUrl}
         displayName={displayName}
+        mxcAvatarUrl={mxcAvatarUrl}
         {...props}
       />
     );
@@ -376,7 +385,9 @@ export const GridTile: FC<GridTileProps> = ({
       <RemoteUserMediaTile
         ref={ref}
         vm={media}
+        focusUrl={focusUrl}
         displayName={displayName}
+        mxcAvatarUrl={mxcAvatarUrl}
         {...props}
       />
     );

src/tile/MediaView.test.tsx

@@ -5,7 +5,7 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { describe, expect, it, test, vi } from "vitest";
+import { describe, expect, it, test } from "vitest";
 import { render, screen } from "@testing-library/react";
 import { axe } from "vitest-axe";
 import { TooltipProvider } from "@vector-im/compound-web";
@@ -16,7 +16,6 @@ import {
 import { LocalTrackPublication, Track } from "livekit-client";
 import { TrackInfo } from "@livekit/protocol";
 import { type ComponentProps } from "react";
-import { type RoomMember } from "matrix-js-sdk";
 
 import { MediaView } from "./MediaView";
 import { EncryptionStatus } from "../state/MediaViewModel";
@@ -46,10 +45,8 @@ describe("MediaView", () => {
     mirror: false,
     unencryptedWarning: false,
     video: trackReference,
-    member: vi.mocked<RoomMember>({
-      userId: "@alice:example.com",
-      getMxcAvatarUrl: vi.fn().mockReturnValue(undefined),
-    } as unknown as RoomMember),
+    userId: "@alice:example.com",
+    mxcAvatarUrl: undefined,
     localParticipant: false,
     focusable: true,
   };

src/tile/MediaView.tsx

@@ -7,7 +7,6 @@ Please see LICENSE in the repository root for full details.
 
 import { type TrackReferenceOrPlaceholder } from "@livekit/components-core";
 import { animated } from "@react-spring/web";
-import { type RoomMember } from "matrix-js-sdk";
 import { type FC, type ComponentProps, type ReactNode } from "react";
 import { useTranslation } from "react-i18next";
 import classNames from "classnames";
@@ -32,12 +31,13 @@ interface Props extends ComponentProps<typeof animated.div> {
   video: TrackReferenceOrPlaceholder | undefined;
   videoFit: "cover" | "contain";
   mirror: boolean;
-  member: RoomMember;
+  userId: string;
   videoEnabled: boolean;
   unencryptedWarning: boolean;
   encryptionStatus: EncryptionStatus;
   nameTagLeadingIcon?: ReactNode;
   displayName: string;
+  mxcAvatarUrl: string | undefined;
   focusable: boolean;
   primaryButton?: ReactNode;
   raisedHandTime?: Date;
@@ -59,11 +59,12 @@ export const MediaView: FC<Props> = ({
   video,
   videoFit,
   mirror,
-  member,
+  userId,
   videoEnabled,
   unencryptedWarning,
   nameTagLeadingIcon,
   displayName,
+  mxcAvatarUrl,
   focusable,
   primaryButton,
   encryptionStatus,
@@ -94,10 +95,10 @@ export const MediaView: FC<Props> = ({
     >
       <div className={styles.bg}>
         <Avatar
-          id={member?.userId ?? displayName}
+          id={userId}
           name={displayName}
           size={avatarSize}
-          src={member?.getMxcAvatarUrl()}
+          src={mxcAvatarUrl}
           className={styles.avatar}
           style={{ display: video && videoEnabled ? "none" : "initial" }}
         />

src/tile/SpotlightTile.tsx

@@ -27,7 +27,6 @@ import { useObservableRef } from "observable-hooks";
 import { useTranslation } from "react-i18next";
 import classNames from "classnames";
 import { type TrackReferenceOrPlaceholder } from "@livekit/components-core";
-import { type RoomMember } from "matrix-js-sdk";
 
 import FullScreenMaximiseIcon from "../icons/FullScreenMaximise.svg?react";
 import FullScreenMinimiseIcon from "../icons/FullScreenMinimise.svg?react";
@@ -55,10 +54,12 @@ interface SpotlightItemBaseProps {
   targetHeight: number;
   video: TrackReferenceOrPlaceholder | undefined;
   videoEnabled: boolean;
-  member: RoomMember;
+  userId: string;
   unencryptedWarning: boolean;
   encryptionStatus: EncryptionStatus;
+  focusUrl: string | undefined;
   displayName: string;
+  mxcAvatarUrl: string | undefined;
   focusable: boolean;
   "aria-hidden"?: boolean;
   localParticipant: boolean;
@@ -78,7 +79,7 @@ const SpotlightLocalUserMediaItem: FC<SpotlightLocalUserMediaItemProps> = ({
   ...props
 }) => {
   const mirror = useBehavior(vm.mirror$);
-  return <MediaView mirror={mirror} focusUrl={vm.focusURL} {...props} />;
+  return <MediaView mirror={mirror} {...props} />;
 };
 
 SpotlightLocalUserMediaItem.displayName = "SpotlightLocalUserMediaItem";
@@ -134,7 +135,9 @@ const SpotlightItem: FC<SpotlightItemProps> = ({
 }) => {
   const ourRef = useRef<HTMLDivElement | null>(null);
   const ref = useMergedRefs(ourRef, theirRef);
+  const focusUrl = useBehavior(vm.focusUrl$);
   const displayName = useBehavior(vm.displayName$);
+  const mxcAvatarUrl = useBehavior(vm.mxcAvatarUrl$);
   const video = useBehavior(vm.video$);
   const videoEnabled = useBehavior(vm.videoEnabled$);
   const unencryptedWarning = useBehavior(vm.unencryptedWarning$);
@@ -161,11 +164,13 @@ const SpotlightItem: FC<SpotlightItemProps> = ({
     className: classNames(styles.item, { [styles.snap]: snap }),
     targetWidth,
     targetHeight,
-    video,
+    video: video ?? undefined,
     videoEnabled,
-    member: vm.member,
+    userId: vm.userId,
     unencryptedWarning,
+    focusUrl,
     displayName,
+    mxcAvatarUrl,
     focusable,
     encryptionStatus,
     "aria-hidden": ariaHidden,

src/utils/displayname-integration.test.ts

@@ -6,10 +6,10 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { afterEach, beforeAll, describe, expect, test, vi } from "vitest";
+import { type RoomMember } from "matrix-js-sdk";
 
 import { shouldDisambiguate } from "./displayname";
 import { alice } from "./test-fixtures";
-import { mockMatrixRoom } from "./test";
 
 // Ideally these tests would be in ./displayname.test.ts but I can't figure out how to
 // just spy on the removeHiddenChars() function without impacting the other tests.
@@ -29,7 +29,7 @@ describe("shouldDisambiguate", () => {
   });
 
   test("should only call removeHiddenChars once for a single displayname", () => {
-    const room = mockMatrixRoom({});
+    const room: Map<string, Pick<RoomMember, "userId">> = new Map([]);
     shouldDisambiguate(alice, [], room);
     expect(jsUtils.removeHiddenChars).toHaveBeenCalledTimes(1);
     for (let i = 0; i < 10; i++) {

src/utils/displayname.test.ts

@@ -20,62 +20,70 @@ import {
   daveRTL,
 } from "./test-fixtures";
 import { mockMatrixRoom } from "./test";
+import { roomToMembersMap } from "../state/CallViewModel/remoteMembers/MatrixMemberMetadata";
 
 describe("shouldDisambiguate", () => {
   test("should not disambiguate a solo member", () => {
-    const room = mockMatrixRoom({});
-    expect(shouldDisambiguate(alice, [], room)).toEqual(false);
+    const room = mockMatrixRoom({
+      getMembers: () => [],
+    });
+    expect(shouldDisambiguate(alice, [], roomToMembersMap(room))).toEqual(
+      false,
+    );
   });
   test("should not disambiguate a member with an empty displayname", () => {
     const room = mockMatrixRoom({
-      getMember: (u) =>
-        [alice, aliceDoppelganger].find((m) => m.userId === u) ?? null,
+      getMembers: () => [alice, aliceDoppelganger],
     });
     expect(
       shouldDisambiguate(
         { rawDisplayName: "", userId: alice.userId },
         [aliceRtcMember, aliceDoppelgangerRtcMember],
-        room,
+        roomToMembersMap(room),
       ),
     ).toEqual(false);
   });
   test("should disambiguate a member with RTL characters", () => {
-    const room = mockMatrixRoom({});
-    expect(shouldDisambiguate(daveRTL, [], room)).toEqual(true);
+    const room = mockMatrixRoom({ getMembers: () => [] });
+    expect(shouldDisambiguate(daveRTL, [], roomToMembersMap(room))).toEqual(
+      true,
+    );
   });
   test("should disambiguate a member with a matching displayname", () => {
     const room = mockMatrixRoom({
-      getMember: (u) =>
-        [alice, aliceDoppelganger].find((m) => m.userId === u) ?? null,
+      getMembers: () => [alice, aliceDoppelganger],
     });
     expect(
       shouldDisambiguate(
         alice,
         [aliceRtcMember, aliceDoppelgangerRtcMember],
-        room,
+        roomToMembersMap(room),
       ),
     ).toEqual(true);
     expect(
       shouldDisambiguate(
         aliceDoppelganger,
         [aliceRtcMember, aliceDoppelgangerRtcMember],
-        room,
+        roomToMembersMap(room),
       ),
     ).toEqual(true);
   });
   test("should disambiguate a member with a matching displayname with hidden spaces", () => {
     const room = mockMatrixRoom({
-      getMember: (u) =>
-        [bob, bobZeroWidthSpace].find((m) => m.userId === u) ?? null,
+      getMembers: () => [bob, bobZeroWidthSpace],
     });
     expect(
-      shouldDisambiguate(bob, [bobRtcMember, bobZeroWidthSpaceRtcMember], room),
+      shouldDisambiguate(
+        bob,
+        [bobRtcMember, bobZeroWidthSpaceRtcMember],
+        roomToMembersMap(room),
+      ),
     ).toEqual(true);
     expect(
       shouldDisambiguate(
         bobZeroWidthSpace,
         [bobRtcMember, bobZeroWidthSpaceRtcMember],
-        room,
+        roomToMembersMap(room),
       ),
     ).toEqual(true);
   });
@@ -83,11 +91,14 @@ describe("shouldDisambiguate", () => {
     "should disambiguate a member with a displayname containing a mxid-like string '%s'",
     (rawDisplayName) => {
       const room = mockMatrixRoom({
-        getMember: (u) =>
-          [alice, aliceDoppelganger].find((m) => m.userId === u) ?? null,
+        getMembers: () => [alice, aliceDoppelganger],
       });
       expect(
-        shouldDisambiguate({ rawDisplayName, userId: alice.userId }, [], room),
+        shouldDisambiguate(
+          { rawDisplayName, userId: alice.userId },
+          [],
+          roomToMembersMap(room),
+        ),
       ).toEqual(true);
     },
   );

src/utils/displayname.ts

@@ -10,7 +10,7 @@ import {
   removeHiddenChars as removeHiddenCharsUncached,
 } from "matrix-js-sdk/lib/utils";
 
-import type { Room } from "matrix-js-sdk";
+import type { RoomMember } from "matrix-js-sdk";
 import type { CallMembership } from "matrix-js-sdk/lib/matrixrtc";
 
 // Calling removeHiddenChars() can be slow on Safari, so we cache the results.
@@ -40,8 +40,8 @@ function removeHiddenChars(str: string): string {
 // Borrowed from https://github.com/matrix-org/matrix-js-sdk/blob/f10deb5ef2e8f061ff005af0476034382ea128ca/src/models/room-member.ts#L409
 export function shouldDisambiguate(
   member: { rawDisplayName?: string; userId: string },
-  memberships: CallMembership[],
-  room: Pick<Room, "getMember">,
+  memberships: Pick<CallMembership, "userId">[],
+  roomMembers: Map<string, Pick<RoomMember, "userId">>,
 ): boolean {
   const { rawDisplayName: displayName, userId } = member;
   if (!displayName || displayName === userId) return false;
@@ -65,7 +65,7 @@ export function shouldDisambiguate(
   // displayname, after hidden character removal.
   return (
     memberships
-      .map((m) => m.userId && room.getMember(m.userId))
+      .map((m) => m.userId && roomMembers.get(m.userId))
       // NOTE: We *should* have a room member for everyone.
       .filter((m) => !!m)
       .filter((m) => m.userId !== userId)

src/utils/observable.test.ts

@@ -9,7 +9,7 @@ import { test } from "vitest";
 import { Subject } from "rxjs";
 
 import { withTestScheduler } from "./test";
-import { generateKeyed$, pauseWhen } from "./observable";
+import { generateItems, pauseWhen } from "./observable";
 
 test("pauseWhen", () => {
   withTestScheduler(({ behavior, expectObservable }) => {
@@ -24,7 +24,7 @@ test("pauseWhen", () => {
   });
 });
 
-test("generateKeyed$ has the right output and ends scopes at the right times", () => {
+test("generateItems", () => {
   const scope1$ = new Subject<string>();
   const scope2$ = new Subject<string>();
   const scope3$ = new Subject<string>();
@@ -44,18 +44,27 @@ test("generateKeyed$ has the right output and ends scopes at the right times", (
     const scope4Marbles = "      ----yn";
 
     expectObservable(
-      generateKeyed$(hot<string>(inputMarbles), (input, createOrGet) => {
-        for (let i = 1; i <= +input; i++) {
-          createOrGet(i.toString(), (scope) => {
+      hot<string>(inputMarbles).pipe(
+        generateItems(
+          function* (input) {
+            for (let i = 1; i <= +input; i++) {
+              yield { keys: [i], data: undefined };
+            }
+          },
+          (scope, data$, i) => {
             scopeSubjects[i - 1].next("y");
             scope.onEnd(() => scopeSubjects[i - 1].next("n"));
             return i.toString();
-          });
-        }
-        return "abcd"[+input - 1];
-      }),
+          },
+        ),
+      ),
       subscriptionMarbles,
-    ).toBe(outputMarbles);
+    ).toBe(outputMarbles, {
+      a: ["1"],
+      b: ["1", "2"],
+      c: ["1", "2", "3"],
+      d: ["1", "2", "3", "4"],
+    });
 
     expectObservable(scope1$).toBe(scope1Marbles);
     expectObservable(scope2$).toBe(scope2Marbles);

src/utils/observable.ts

@@ -20,10 +20,12 @@ import {
   takeWhile,
   tap,
   withLatestFrom,
+  BehaviorSubject,
+  type OperatorFunction,
 } from "rxjs";
 
 import { type Behavior } from "../state/Behavior";
-import { ObservableScope } from "../state/ObservableScope";
+import { Epoch, ObservableScope } from "../state/ObservableScope";
 
 const nothing = Symbol("nothing");
 
@@ -119,70 +121,156 @@ export function pauseWhen<T>(pause$: Behavior<boolean>) {
     );
 }
 
+interface ItemHandle<Data, Item> {
+  scope: ObservableScope;
+  data$: BehaviorSubject<Data>;
+  item: Item;
+}
+
 /**
- * Maps a changing input value to an output value consisting of items that have
- * automatically generated ObservableScopes tied to a key. Items will be
- * automatically created when their key is requested for the first time, reused
- * when the same key is requested at a later time, and destroyed (have their
- * scope ended) when the key is no longer requested.
+ * Maps a changing input value to a collection of items that each capture some
+ * dynamic data and are tied to a key. Items will be automatically created when
+ * their key is requested for the first time, reused when the same key is
+ * requested at a later time, and destroyed (have their scope ended) when the
+ * key is no longer requested.
  *
  * @param input$ The input value to be mapped.
- * @param project A function mapping input values to output values. This
- *   function receives an additional callback `createOrGet` which can be used
- *   within the function body to request that an item be generated for a certain
- *   key. The caller provides a factory which will be used to create the item if
- *   it is being requested for the first time. Otherwise, the item previously
- *   existing under that key will be returned.
+ * @param generator A generator function yielding a tuple of keys and the
+ *   currently associated data for each item that it wants to exist.
+ * @param factory A function constructing an individual item, given the item's key,
+ *   dynamic data, and an automatically managed ObservableScope for the item.
  */
-export function generateKeyed$<In, Item, Out>(
-  input$: Observable<In>,
-  project: (
-    input: In,
-    createOrGet: (
-      key: string,
-      factory: (scope: ObservableScope) => Item,
-    ) => Item,
-  ) => Out,
-): Observable<Out> {
-  return input$.pipe(
-    // Keep track of the existing items over time, so we can reuse them
-    scan<
-      In,
-      {
-        items: Map<string, { item: Item; scope: ObservableScope }>;
-        output: Out;
-      },
-      { items: Map<string, { item: Item; scope: ObservableScope }> }
-    >(
-      (state, data) => {
-        const nextItems = new Map<
-          string,
-          { item: Item; scope: ObservableScope }
-        >();
+export function generateItems<
+  Input,
+  Keys extends [unknown, ...unknown[]],
+  Data,
+  Item,
+>(
+  generator: (
+    input: Input,
+  ) => Generator<{ keys: readonly [...Keys]; data: Data }, void, void>,
+  factory: (
+    scope: ObservableScope,
+    data$: Behavior<Data>,
+    ...keys: Keys
+  ) => Item,
+): OperatorFunction<Input, Item[]> {
+  return generateItemsInternal(generator, factory, (items) => items);
+}
 
-        const output = project(data, (key, factory) => {
-          let item = state.items.get(key);
-          if (item === undefined) {
-            // First time requesting the key; create the item
-            const scope = new ObservableScope();
-            item = { item: factory(scope), scope };
-          }
-          nextItems.set(key, item);
-          return item.item;
-        });
-
-        // Destroy all items that are no longer being requested
-        for (const [key, { scope }] of state.items)
-          if (!nextItems.has(key)) scope.end();
-
-        return { items: nextItems, output };
-      },
-      { items: new Map() },
-    ),
-    finalizeValue((state) => {
-      // Destroy all remaining items when no longer subscribed
-      for (const { scope } of state.items.values()) scope.end();
-    }),
-    map(({ output }) => output),
+/**
+ * Same as generateItems, but preserves epoch data.
+ */
+export function generateItemsWithEpoch<
+  Input,
+  Keys extends [unknown, ...unknown[]],
+  Data,
+  Item,
+>(
+  generator: (
+    input: Input,
+  ) => Generator<{ keys: readonly [...Keys]; data: Data }, void, void>,
+  factory: (
+    scope: ObservableScope,
+    data$: Behavior<Data>,
+    ...keys: Keys
+  ) => Item,
+): OperatorFunction<Epoch<Input>, Epoch<Item[]>> {
+  return generateItemsInternal(
+    function* (input) {
+      yield* generator(input.value);
+    },
+    factory,
+    (items, input) => new Epoch(items, input.epoch),
   );
 }
+
+function generateItemsInternal<
+  Input,
+  Keys extends [unknown, ...unknown[]],
+  Data,
+  Item,
+  Output,
+>(
+  generator: (
+    input: Input,
+  ) => Generator<{ keys: readonly [...Keys]; data: Data }, void, void>,
+  factory: (
+    scope: ObservableScope,
+    data$: Behavior<Data>,
+    ...keys: Keys
+  ) => Item,
+  project: (items: Item[], input: Input) => Output,
+): OperatorFunction<Input, Output> {
+  /* eslint-disable @typescript-eslint/no-explicit-any */
+  return (input$) =>
+    input$.pipe(
+      // Keep track of the existing items over time, so they can persist
+      scan<
+        Input,
+        {
+          map: Map<any, any>;
+          items: Set<ItemHandle<Data, Item>>;
+          input: Input;
+        },
+        { map: Map<any, any>; items: Set<ItemHandle<Data, Item>> }
+      >(
+        ({ map: prevMap, items: prevItems }, input) => {
+          const nextMap = new Map();
+          const nextItems = new Set<ItemHandle<Data, Item>>();
+
+          for (const { keys, data } of generator(input)) {
+            // Disable type checks for a second to grab the item out of a nested map
+            let i: any = prevMap;
+            for (const key of keys) i = i?.get(key);
+            let item = i as ItemHandle<Data, Item> | undefined;
+
+            if (item === undefined) {
+              // First time requesting the key; create the item
+              const scope = new ObservableScope();
+              const data$ = new BehaviorSubject(data);
+              item = { scope, data$, item: factory(scope, data$, ...keys) };
+            } else {
+              item.data$.next(data);
+            }
+
+            // Likewise, disable type checks to insert the item in the nested map
+            let m: Map<any, any> = nextMap;
+            for (let i = 0; i < keys.length - 1; i++) {
+              let inner = m.get(keys[i]);
+              if (inner === undefined) {
+                inner = new Map();
+                m.set(keys[i], inner);
+              }
+              m = inner;
+            }
+            const finalKey = keys[keys.length - 1];
+            if (m.has(finalKey))
+              throw new Error(
+                `Keys must be unique (tried to generate multiple items for key ${keys})`,
+              );
+            m.set(keys[keys.length - 1], item);
+            nextItems.add(item);
+          }
+
+          // Destroy all items that are no longer being requested
+          for (const item of prevItems)
+            if (!nextItems.has(item)) item.scope.end();
+
+          return { map: nextMap, items: nextItems, input };
+        },
+        { map: new Map(), items: new Set() },
+      ),
+      finalizeValue(({ items }) => {
+        // Destroy all remaining items when no longer subscribed
+        for (const { scope } of items) scope.end();
+      }),
+      map(({ items, input }) =>
+        project(
+          [...items].map(({ item }) => item),
+          input,
+        ),
+      ),
+    );
+  /* eslint-enable @typescript-eslint/no-explicit-any */
+}

src/utils/test-fixtures.ts

@@ -11,9 +11,9 @@ import {
   mockRemoteParticipant,
 } from "./test";
 
-export const localRtcMember = mockRtcMembership("@carol:example.org", "1111");
+export const localRtcMember = mockRtcMembership("@local:example.org", "1111");
 export const localRtcMemberDevice2 = mockRtcMembership(
-  "@carol:example.org",
+  "@local:example.org",
   "2222",
 );
 export const local = mockMatrixRoomMember(localRtcMember);
@@ -37,7 +37,6 @@ export const aliceDoppelganger = mockMatrixRoomMember(
     rawDisplayName: "Alice",
   },
 );
-export const aliceDoppelgangerId = `${aliceDoppelganger.userId}:${aliceDoppelgangerRtcMember.deviceId}`;
 
 export const bobRtcMember = mockRtcMembership("@bob:example.org", "BBBB");
 export const bob = mockMatrixRoomMember(bobRtcMember, {
@@ -55,10 +54,8 @@ export const bobZeroWidthSpace = mockMatrixRoomMember(
     rawDisplayName: "Bo\u200bb",
   },
 );
-export const bobZeroWidthSpaceId = `${bobZeroWidthSpace.userId}:${bobZeroWidthSpaceRtcMember.deviceId}`;
 
 export const daveRTLRtcMember = mockRtcMembership("@dave2:example.org", "DDDD");
 export const daveRTL = mockMatrixRoomMember(daveRTLRtcMember, {
   rawDisplayName: "\u202eevaD",
 });
-export const daveRTLId = `${daveRTL.userId}:${daveRTLRtcMember.deviceId}`;

src/utils/test-viewmodel.ts

@@ -6,7 +6,7 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
-import { BehaviorSubject, of } from "rxjs";
+import { BehaviorSubject } from "rxjs";
 import { vitest } from "vitest";
 import { type RelationsContainer } from "matrix-js-sdk/lib/models/relations-container";
 import EventEmitter from "events";
@@ -22,8 +22,9 @@ import { E2eeType } from "../e2ee/e2eeType";
 import {
   CallViewModel,
   type CallViewModelOptions,
-} from "../state/CallViewModel";
+} from "../state/CallViewModel/CallViewModel";
 import {
+  mockConfig,
   mockLivekitRoom,
   mockLocalParticipant,
   mockMatrixRoom,
@@ -36,6 +37,8 @@ import { aliceRtcMember, localRtcMember } from "./test-fixtures";
 import { type RaisedHandInfo, type ReactionInfo } from "../reactions";
 import { constant } from "../state/Behavior";
 
+mockConfig({ livekit: { livekit_service_url: "https://example.com" } });
+
 export function getBasicRTCSession(
   members: RoomMember[],
   initialRtcMemberships: CallMembership[] = [localRtcMember, aliceRtcMember],
@@ -57,6 +60,7 @@ export function getBasicRTCSession(
       getUserId: () => localRtcMember.userId,
       getDeviceId: () => localRtcMember.deviceId,
       getSyncState: () => SyncState.Syncing,
+      getDomain: () => null,
       sendEvent: vitest.fn().mockResolvedValue({ event_id: "$fake:event" }),
       redactEvent: vitest.fn().mockResolvedValue({ event_id: "$fake:event" }),
       decryptEventIfNeeded: vitest.fn().mockResolvedValue(undefined),
@@ -78,6 +82,7 @@ export function getBasicRTCSession(
         ),
     } as Partial<MatrixClient> as MatrixClient,
     getMember: (userId) => matrixRoomMembers.get(userId) ?? null,
+    getMembers: () => Array.from(matrixRoomMembers.values()),
     roomId: matrixRoomId,
     on: vitest
       .fn()
@@ -158,7 +163,7 @@ export function getBasicCallViewModelEnvironment(
     },
     handRaisedSubject$,
     reactionsSubject$,
-    of({ processor: undefined, supported: false }),
+    constant({ processor: undefined, supported: false }),
   );
   return {
     vm,

src/utils/test.ts

@@ -6,7 +6,14 @@ Please see LICENSE in the repository root for full details.
 */
 import { map, type Observable, of, type SchedulerLike } from "rxjs";
 import { type RunHelpers, TestScheduler } from "rxjs/testing";
-import { expect, type MockedObject, onTestFinished, vi, vitest } from "vitest";
+import {
+  expect,
+  type MockedObject,
+  type MockInstance,
+  onTestFinished,
+  vi,
+  vitest,
+} from "vitest";
 import {
   MatrixEvent,
   type Room as MatrixRoom,
@@ -78,11 +85,11 @@ export interface OurRunHelpers extends RunHelpers {
    * diagram.
    */
   schedule: (marbles: string, actions: Record<string, () => void>) => void;
-  behavior<T = string>(
+  behavior: <T>(
     marbles: string,
     values?: { [marble: string]: T },
     error?: unknown,
-  ): Behavior<T>;
+  ) => Behavior<T>;
   scope: ObservableScope;
 }
 
@@ -187,6 +194,29 @@ export const exampleTransport: LivekitTransport = {
   livekit_alias: "!alias:example.org",
 };
 
+export function mockCallMembership(
+  userId: string,
+  deviceId: string,
+  transport?: Transport,
+): CallMembership {
+  const t = transport ?? transportForUser(userId);
+  return {
+    userId: userId,
+    deviceId: deviceId,
+    getTransport: vi.fn().mockReturnValue(t),
+    transports: [t],
+  } as unknown as CallMembership;
+}
+
+function transportForUser(userId: string): Transport {
+  const domain = userId.split(":")[1];
+  return {
+    type: "livekit",
+    livekit_service_url: `https://lk.${domain}`,
+    livekit_alias: `!alias:${domain}`,
+  };
+}
+
 export function mockRtcMembership(
   user: string | RoomMember,
   deviceId: string,
@@ -246,6 +276,7 @@ export function mockLivekitRoom(
   }: { remoteParticipants$?: Observable<RemoteParticipant[]> } = {},
 ): LivekitRoom {
   const livekitRoom = {
+    options: {},
     ...mockEmitter(),
     ...room,
   } as Partial<LivekitRoom> as LivekitRoom;
@@ -268,6 +299,7 @@ export function mockLocalParticipant(
   return {
     isLocal: true,
     trackPublications: new Map(),
+    unpublishTracks: async () => Promise.resolve(),
     getTrackPublication: () =>
       ({}) as Partial<LocalTrackPublication> as LocalTrackPublication,
     ...mockEmitter(),
@@ -281,18 +313,20 @@ export function createLocalMedia(
   localParticipant: LocalParticipant,
   mediaDevices: MediaDevices,
 ): LocalUserMediaViewModel {
+  const member = mockMatrixRoomMember(localRtcMember, roomMember);
   return new LocalUserMediaViewModel(
     testScope(),
     "local",
-    mockMatrixRoomMember(localRtcMember, roomMember),
+    member.userId,
     constant(localParticipant),
     {
       kind: E2eeType.PER_PARTICIPANT,
     },
-    mockLivekitRoom({ localParticipant }),
-    "https://rtc-example.org",
+    constant(mockLivekitRoom({ localParticipant })),
+    constant("https://rtc-example.org"),
     mediaDevices,
-    constant(roomMember.rawDisplayName ?? "nodisplayname"),
+    constant(member.rawDisplayName ?? "nodisplayname"),
+    constant(member.getMxcAvatarUrl()),
     constant(null),
     constant(null),
   );
@@ -306,6 +340,8 @@ export function mockRemoteParticipant(
     setVolume() {},
     getTrackPublication: () =>
       ({}) as Partial<RemoteTrackPublication> as RemoteTrackPublication,
+    // this will only get used for `getTrackPublications().length`
+    getTrackPublications: () => [0],
     ...mockEmitter(),
     ...participant,
   } as RemoteParticipant;
@@ -316,31 +352,38 @@ export function createRemoteMedia(
   roomMember: Partial<RoomMember>,
   participant: Partial<RemoteParticipant>,
 ): RemoteUserMediaViewModel {
+  const member = mockMatrixRoomMember(localRtcMember, roomMember);
   const remoteParticipant = mockRemoteParticipant(participant);
   return new RemoteUserMediaViewModel(
     testScope(),
     "remote",
-    mockMatrixRoomMember(localRtcMember, roomMember),
+    member.userId,
     of(remoteParticipant),
     {
       kind: E2eeType.PER_PARTICIPANT,
     },
-    mockLivekitRoom({}, { remoteParticipants$: of([remoteParticipant]) }),
-    "https://rtc-example.org",
+    constant(
+      mockLivekitRoom({}, { remoteParticipants$: of([remoteParticipant]) }),
+    ),
+    constant("https://rtc-example.org"),
     constant(false),
-    constant(roomMember.rawDisplayName ?? "nodisplayname"),
+    constant(member.rawDisplayName ?? "nodisplayname"),
+    constant(member.getMxcAvatarUrl()),
     constant(null),
     constant(null),
   );
 }
 
-export function mockConfig(config: Partial<ResolvedConfigOptions> = {}): void {
-  vi.spyOn(Config, "get").mockReturnValue({
+export function mockConfig(
+  config: Partial<ResolvedConfigOptions> = {},
+): MockInstance<() => ResolvedConfigOptions> {
+  const spy = vi.spyOn(Config, "get").mockReturnValue({
     ...DEFAULT_CONFIG,
     ...config,
   });
   // simulate loading the config
   vi.spyOn(Config, "init").mockResolvedValue(void 0);
+  return spy;
 }
 
 export class MockRTCSession extends TypedEventEmitter<

tsconfig.json

@@ -50,5 +50,6 @@
 
     "plugins": [{ "name": "typescript-eslint-language-service" }]
   },
-  "include": ["./src/**/*.ts", "./src/**/*.tsx", "./playwright/**/*.ts"]
+  "include": ["./src/**/*.ts", "./src/**/*.tsx", "./playwright/**/*.ts"],
+  "exclude": ["**.test.ts"]
 }