Merge remote-tracking branch 'origin/livekit' into hs/apply-hints

.github/workflows/publish-embedded-packages.yaml

@@ -44,6 +44,8 @@ jobs:
         run: |
           if [[ "${VERSION}" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
             echo "TAG=latest" >> "$GITHUB_OUTPUT"
+          elif [[ "${VERSION}" =~ ^v[0-9]+\.[0-9]+\.[0-9]+\-rc\.[0-9]+$ ]]; then
+            echo "TAG=rc" >> "$GITHUB_OUTPUT"
           else
             echo "TAG=other" >> "$GITHUB_OUTPUT"
           fi
@@ -163,6 +165,8 @@ jobs:
         run: |
           if [[ "${{ needs.versioning.outputs.TAG }}" == "latest" ]]; then
             echo "ARTIFACT_VERSION=${{ needs.versioning.outputs.UNPREFIXED_VERSION }}" >> "$GITHUB_ENV"
+          elif [[ "${{ needs.versioning.outputs.TAG }}" == "rc" ]]; then
+            echo "ARTIFACT_VERSION=${{ needs.versioning.outputs.UNPREFIXED_VERSION }}" >> "$GITHUB_ENV"
           else
             echo "ARTIFACT_VERSION=${{ needs.versioning.outputs.UNPREFIXED_VERSION }}-SNAPSHOT" >> "$GITHUB_ENV"
           fi

src/UrlParams.test.ts

@@ -5,12 +5,15 @@ 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 { describe, expect, it, onTestFinished, vi } from "vitest";
+import { logger } from "matrix-js-sdk/lib/logger";
 
+import * as PlatformMod from "../src/Platform";
 import {
   getRoomIdentifierFromUrl,
-  getUrlParams,
+  computeUrlParams,
   HeaderStyle,
+  getUrlParams,
 } from "../src/UrlParams";
 
 const ROOM_NAME = "roomNameHere";
@@ -103,16 +106,16 @@ describe("UrlParams", () => {
 
   describe("preload", () => {
     it("defaults to false", () => {
-      expect(getUrlParams().preload).toBe(false);
+      expect(computeUrlParams().preload).toBe(false);
     });
 
     it("ignored in SPA mode", () => {
-      expect(getUrlParams("?preload=true").preload).toBe(false);
+      expect(computeUrlParams("?preload=true").preload).toBe(false);
     });
 
     it("respected in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?preload=true&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).preload,
       ).toBe(true);
@@ -121,19 +124,20 @@ describe("UrlParams", () => {
 
   describe("returnToLobby", () => {
     it("is false in SPA mode", () => {
-      expect(getUrlParams("?returnToLobby=true").returnToLobby).toBe(false);
+      expect(computeUrlParams("?returnToLobby=true").returnToLobby).toBe(false);
     });
 
     it("defaults to false in widget mode", () => {
       expect(
-        getUrlParams("?widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo")
+        computeUrlParams(
-          .returnToLobby,
+          "?widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
+        ).returnToLobby,
       ).toBe(false);
     });
 
     it("respected in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?returnToLobby=true&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).returnToLobby,
       ).toBe(true);
@@ -142,12 +146,12 @@ describe("UrlParams", () => {
 
   describe("userId", () => {
     it("is ignored in SPA mode", () => {
-      expect(getUrlParams("?userId=asd").userId).toBe(null);
+      expect(computeUrlParams("?userId=asd").userId).toBe(null);
     });
 
     it("is parsed in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?userId=asd&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).userId,
       ).toBe("asd");
@@ -156,12 +160,12 @@ describe("UrlParams", () => {
 
   describe("deviceId", () => {
     it("is ignored in SPA mode", () => {
-      expect(getUrlParams("?deviceId=asd").deviceId).toBe(null);
+      expect(computeUrlParams("?deviceId=asd").deviceId).toBe(null);
     });
 
     it("is parsed in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?deviceId=asd&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).deviceId,
       ).toBe("asd");
@@ -170,12 +174,12 @@ describe("UrlParams", () => {
 
   describe("baseUrl", () => {
     it("is ignored in SPA mode", () => {
-      expect(getUrlParams("?baseUrl=asd").baseUrl).toBe(null);
+      expect(computeUrlParams("?baseUrl=asd").baseUrl).toBe(null);
     });
 
     it("is parsed in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?baseUrl=asd&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).baseUrl,
       ).toBe("asd");
@@ -185,28 +189,28 @@ describe("UrlParams", () => {
   describe("viaServers", () => {
     it("is ignored in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?viaServers=asd&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).viaServers,
       ).toBe(null);
     });
 
     it("is parsed in SPA mode", () => {
-      expect(getUrlParams("?viaServers=asd").viaServers).toBe("asd");
+      expect(computeUrlParams("?viaServers=asd").viaServers).toBe("asd");
     });
   });
 
   describe("homeserver", () => {
     it("is ignored in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?homeserver=asd&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).homeserver,
       ).toBe(null);
     });
 
     it("is parsed in SPA mode", () => {
-      expect(getUrlParams("?homeserver=asd").homeserver).toBe("asd");
+      expect(computeUrlParams("?homeserver=asd").homeserver).toBe("asd");
     });
   });
 
@@ -237,7 +241,7 @@ describe("UrlParams", () => {
       controlledAudioDevices: platform === "desktop" ? false : true,
       skipLobby: true,
       returnToLobby: false,
-      sendNotificationType: "notification",
+      sendNotificationType: platform === "desktop" ? "notification" : "ring",
     });
     const joinExistingCallDefaults = (platform: string): object => ({
       confineToRoom: true,
@@ -252,24 +256,55 @@ describe("UrlParams", () => {
       skipLobby: false,
       returnToLobby: false,
       sendNotificationType: "notification",
+      defaultAudioEnabled: true,
+      defaultVideoEnabled: true,
     });
     it("use no-intent-defaults with unknown intent", () => {
-      expect(getUrlParams()).toMatchObject(noIntentDefaults);
+      expect(computeUrlParams()).toMatchObject(noIntentDefaults);
     });
 
     it("ignores intent if it is not a valid value", () => {
-      expect(getUrlParams("?intent=foo")).toMatchObject(noIntentDefaults);
+      expect(computeUrlParams("?intent=foo")).toMatchObject(noIntentDefaults);
     });
 
     it("accepts start_call", () => {
       expect(
-        getUrlParams("?intent=start_call&widgetId=1234&parentUrl=parent.org"),
+        computeUrlParams(
+          "?intent=start_call&widgetId=1234&parentUrl=parent.org",
+        ),
       ).toMatchObject(startNewCallDefaults("desktop"));
     });
 
+    it("accepts start_call_dm mobile", () => {
+      vi.spyOn(PlatformMod, "platform", "get").mockReturnValue("android");
+      onTestFinished(() => {
+        vi.spyOn(PlatformMod, "platform", "get").mockReturnValue("desktop");
+      });
+      expect(
+        computeUrlParams(
+          "?intent=start_call_dm&widgetId=1234&parentUrl=parent.org",
+        ),
+      ).toMatchObject(startNewCallDefaults("android"));
+    });
+
+    it("accepts start_call_dm mobile and prioritizes overwritten params", () => {
+      vi.spyOn(PlatformMod, "platform", "get").mockReturnValue("android");
+      onTestFinished(() => {
+        vi.spyOn(PlatformMod, "platform", "get").mockReturnValue("desktop");
+      });
+      expect(
+        computeUrlParams(
+          "?intent=start_call_dm&widgetId=1234&parentUrl=parent.org&sendNotificationType=notification",
+        ),
+      ).toMatchObject({
+        ...startNewCallDefaults("android"),
+        sendNotificationType: "notification",
+      });
+    });
+
     it("accepts join_existing", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?intent=join_existing&widgetId=1234&parentUrl=parent.org",
         ),
       ).toMatchObject(joinExistingCallDefaults("desktop"));
@@ -278,31 +313,55 @@ describe("UrlParams", () => {
 
   describe("skipLobby", () => {
     it("defaults to false", () => {
-      expect(getUrlParams().skipLobby).toBe(false);
+      expect(computeUrlParams().skipLobby).toBe(false);
     });
 
     it("defaults to false if intent is start_call in SPA mode", () => {
-      expect(getUrlParams("?intent=start_call").skipLobby).toBe(false);
+      expect(computeUrlParams("?intent=start_call").skipLobby).toBe(false);
     });
 
     it("defaults to true if intent is start_call in widget mode", () => {
       expect(
-        getUrlParams(
+        computeUrlParams(
           "?intent=start_call&widgetId=12345&parentUrl=https%3A%2F%2Flocalhost%2Ffoo",
         ).skipLobby,
       ).toBe(true);
     });
 
     it("default to false if intent is join_existing", () => {
-      expect(getUrlParams("?intent=join_existing").skipLobby).toBe(false);
+      expect(computeUrlParams("?intent=join_existing").skipLobby).toBe(false);
     });
   });
   describe("header", () => {
     it("uses header if provided", () => {
-      expect(getUrlParams("?header=app_bar&hideHeader=true").header).toBe(
+      expect(computeUrlParams("?header=app_bar&hideHeader=true").header).toBe(
         "app_bar",
       );
-      expect(getUrlParams("?header=none&hideHeader=false").header).toBe("none");
+      expect(computeUrlParams("?header=none&hideHeader=false").header).toBe(
+        "none",
+      );
+    });
+  });
+  describe("getUrlParams", () => {
+    it("uses cached values", () => {
+      const spy = vi.spyOn(logger, "info");
+      // call get once
+      const params = getUrlParams("?header=app_bar&hideHeader=true", "");
+      // call get twice
+      expect(getUrlParams("?header=app_bar&hideHeader=true", "")).toBe(params);
+      // expect compute to only be called once
+      // it will only log when it is computing the values
+      expect(spy).toHaveBeenCalledExactlyOnceWith(
+        "UrlParams: final set of url params\n",
+        "intent:",
+        "unknown",
+        "\nproperties:",
+        expect.any(Object),
+        "configuration:",
+        expect.any(Object),
+        "intentAndPlatformDerivedConfiguration:",
+        {},
+      );
     });
   });
 });

src/UrlParams.ts

@@ -239,6 +239,10 @@ interface IntentAndPlatformDerivedConfiguration {
   defaultAudioEnabled?: boolean;
   defaultVideoEnabled?: boolean;
 }
+interface IntentAndPlatformDerivedConfiguration {
+  defaultAudioEnabled?: boolean;
+  defaultVideoEnabled?: boolean;
+}
 
 // If you need to add a new flag to this interface, prefer a name that describes
 // a specific behavior (such as 'confineToRoom'), rather than one that describes
@@ -329,8 +333,12 @@ let urlParamCache: {
   hash?: string;
   params?: UrlParams;
 } = {};
+<<<<<<< HEAD
+=======
+
+>>>>>>> origin/livekit
 /**
- * Gets the app parameters for the current URL.
+ * Gets the url params and loads them from a cache if already computed.
  * @param search The URL search string
  * @param hash The URL hash
  * @returns The app parameters encoded in the URL
@@ -341,15 +349,26 @@ export const getUrlParams = (
   /** Skipping the cache might be needed in tests, to allow recomputing based on mocked platform changes. */
   skipCache = false,
 ): UrlParams => {
-  // Only run the param configuration if we do not yet have it cached for this url.
   if (
     urlParamCache.search === search &&
     urlParamCache.hash === hash &&
-    urlParamCache.params &&
+    urlParamCache.params
-    !skipCache
   ) {
     return urlParamCache.params;
   }
+  const params = computeUrlParams(search, hash);
+  urlParamCache = { search, hash, params };
+
+  return params;
+};
+
+/**
+ * Gets the app parameters for the current URL.
+ * @param search The URL search string
+ * @param hash The URL hash
+ * @returns The app parameters encoded in the URL
+ */
+export const computeUrlParams = (search = "", hash = ""): UrlParams => {
   const parser = new ParamParser(search, hash);
 
   const fontScale = parseFloat(parser.getParam("fontScale") ?? "");
@@ -385,7 +404,7 @@ export const getUrlParams = (
     controlledAudioDevices: platform === "desktop" ? false : true,
     skipLobby: true,
     returnToLobby: false,
-    sendNotificationType: "notification" as RTCNotificationType,
+    sendNotificationType: "notification",
     autoLeaveWhenOthersLeft: false,
     waitForCallPickup: false,
   };
@@ -404,6 +423,7 @@ export const getUrlParams = (
     // Fall through
     case UserIntent.StartNewCallDM:
       intentPreset.skipLobby = true;
+      intentPreset.sendNotificationType = "ring";
       intentPreset.autoLeaveWhenOthersLeft = true;
       intentPreset.waitForCallPickup = true;
       intentPreset.callIntent = intentPreset.callIntent ?? "video";
@@ -526,7 +546,11 @@ export const getUrlParams = (
     intentAndPlatformDerivedConfiguration,
   );
 
+<<<<<<< HEAD
   const params = {
+=======
+  return {
+>>>>>>> origin/livekit
     ...properties,
     ...intentPreset,
     ...pickBy(configuration, (v?: unknown) => v !== undefined),

src/room/InCallView.tsx

@@ -125,6 +125,7 @@ import { prefetchSounds } from "../soundUtils";
 import { useAudioContext } from "../useAudioContext";
 import ringtoneMp3 from "../sound/ringtone.mp3?url";
 import ringtoneOgg from "../sound/ringtone.ogg?url";
+import { ObservableScope } from "../state/ObservableScope.ts";
 
 const canScreenshare = "getDisplayMedia" in (navigator.mediaDevices ?? {});
 
@@ -144,8 +145,13 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
     sfuConfig,
     props.e2eeSystem,
   );
-  const connStateObservable$ = useObservable(
+  const observableScope = useInitial(() => new ObservableScope());
-    (inputs$) => inputs$.pipe(map(([connState]) => connState)),
+  const connStateBehavior$ = useObservable(
+    (inputs$) =>
+      observableScope.behavior(
+        inputs$.pipe(map(([connState]) => connState)),
+        connState,
+      ),
     [connState],
   );
   const [vm, setVm] = useState<CallViewModel | null>(null);
@@ -188,7 +194,7 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
           waitForCallPickup:
             waitForCallPickup && sendNotificationType === "ring",
         },
-        connStateObservable$,
+        connStateBehavior$,
         reactionsReader.raisedHands$,
         reactionsReader.reactions$,
       );
@@ -204,7 +210,7 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
     livekitRoom,
     mediaDevices,
     props.e2eeSystem,
-    connStateObservable$,
+    connStateBehavior$,
     autoLeaveWhenOthersLeft,
     sendNotificationType,
     waitForCallPickup,

src/state/CallViewModel.test.ts

@@ -266,7 +266,7 @@ const mockLegacyRingEvent = {} as { event_id: string } & ICallNotifyContent;
 interface CallViewModelInputs {
   remoteParticipants$: Behavior<RemoteParticipant[]>;
   rtcMembers$: Behavior<Partial<CallMembership>[]>;
-  connectionState$: Observable<ECConnectionState>;
+  livekitConnectionState$: Behavior<ECConnectionState>;
   speaking: Map<Participant, Observable<boolean>>;
   mediaDevices: MediaDevices;
   initialSyncState: SyncState;
@@ -276,7 +276,9 @@ function withCallViewModel(
   {
     remoteParticipants$ = constant([]),
     rtcMembers$ = constant([localRtcMember]),
-    connectionState$ = of(ConnectionState.Connected),
+    livekitConnectionState$: connectionState$ = constant(
+      ConnectionState.Connected,
+    ),
     speaking = new Map(),
     mediaDevices = mockMediaDevices({}),
     initialSyncState = SyncState.Syncing,
@@ -384,7 +386,7 @@ test("participants are retained during a focus switch", () => {
           b: [],
         }),
         rtcMembers$: constant([localRtcMember, aliceRtcMember, bobRtcMember]),
-        connectionState$: behavior(connectionInputMarbles, {
+        livekitConnectionState$: behavior(connectionInputMarbles, {
           c: ConnectionState.Connected,
           s: ECAddonConnectionState.ECSwitchingFocus,
         }),
@@ -1251,6 +1253,41 @@ describe("waitForCallPickup$", () => {
     });
   });
 
+  test("regression test: does stop ringing in case livekitConnectionState$ emits after didSendCallNotification$ has already emitted", () => {
+    withTestScheduler(({ schedule, expectObservable, behavior }) => {
+      withCallViewModel(
+        {
+          livekitConnectionState$: behavior("d 9ms c", {
+            d: ConnectionState.Disconnected,
+            c: ConnectionState.Connected,
+          }),
+        },
+        (vm, rtcSession) => {
+          // Fire a call notification IMMEDIATELY (its important for this test, that this happens before the livekitConnectionState$ emits)
+          schedule("n", {
+            n: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                mockRingEvent("$notif1", 30),
+                mockLegacyRingEvent,
+              );
+            },
+          });
+
+          expectObservable(vm.callPickupState$).toBe("a 9ms b 19ms c", {
+            a: "unknown",
+            b: "ringing",
+            c: "timeout",
+          });
+        },
+        {
+          waitForCallPickup: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
   test("ringing -> success if someone joins before timeout", () => {
     withTestScheduler(({ behavior, schedule, expectObservable }) => {
       // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
@@ -1305,7 +1342,7 @@ describe("waitForCallPickup$", () => {
             a: [localRtcMember],
             b: [localRtcMember, aliceRtcMember],
           }),
-          connectionState$,
+          livekitConnectionState$: connectionState$,
         },
         (vm, rtcSession) => {
           // Notify at 5ms so we enter ringing, then get disconnected 5ms later

src/state/CallViewModel.ts

@@ -880,68 +880,77 @@ export class CallViewModel extends ViewModel {
     ? this.allOthersLeft$
     : NEVER;
 
+  private readonly didSendCallNotification$ = fromEvent(
+    this.matrixRTCSession,
+    MatrixRTCSessionEvent.DidSendCallNotification,
+  ) as Observable<
+    Parameters<
+      MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification]
+    >
+  >;
   /**
    * 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.
    */
-  private readonly ring$: Observable<
+  // This is a behavior since we need to store the latest state for when we subscribe to this after `didSendCallNotification$`
-    Observable<"ringing" | "timeout" | "decline">
+  // 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.
-    fromEvent(
+  // see also: callPickupState$ and in particular the line: `return this.ring$.pipe(mergeAll());` here we otherwise might get an EMPTY observable if
-      this.matrixRTCSession,
+  // `ring$` would not be a behavior.
-      MatrixRTCSessionEvent.DidSendCallNotification,
+  private readonly ring$: Behavior<"ringing" | "timeout" | "decline" | null> =
-    ) as Observable<
+    this.scope.behavior(
-      Parameters<
+      this.didSendCallNotification$.pipe(
-        MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification]
+        filter(
-      >
+          ([notificationEvent]) =>
-    >
+            notificationEvent.notification_type === "ring",
-  ).pipe(
-    filter(
-      ([notificationEvent]) => notificationEvent.notification_type === "ring",
-    ),
-    map(([notificationEvent]) => {
-      const lifetimeMs = notificationEvent?.lifetime ?? 0;
-      return concat(
-        lifetimeMs === 0
-          ? // If no lifetime, skip the ring state
-            EMPTY
-          : // 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),
-        NEVER,
-      ).pipe(
-        takeUntil(
-          (
-            fromEvent(this.matrixRoom, RoomEvent.Timeline) as Observable<
-              Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
-            >
-          ).pipe(
-            filter(
-              ([event]) =>
-                event.getType() === EventType.RTCDecline &&
-                event.getRelation()?.rel_type === "m.reference" &&
-                event.getRelation()?.event_id === notificationEvent.event_id &&
-                event.getSender() !== this.userId,
-            ),
-          ),
         ),
-        endWith("decline" as const),
+        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(
+              (
+                fromEvent(this.matrixRoom, RoomEvent.Timeline) as Observable<
+                  Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
+                >
+              ).pipe(
+                filter(
+                  ([event]) =>
+                    event.getType() === EventType.RTCDecline &&
+                    event.getRelation()?.rel_type === "m.reference" &&
+                    event.getRelation()?.event_id ===
+                      notificationEvent.event_id &&
+                    event.getSender() !== this.userId,
+                ),
+              ),
+            ),
+            endWith("decline" as const),
+          );
+        }),
+      ),
+      null,
+    );
 
   /**
    * Whether some Matrix user other than ourself is joined to the call.
    */
   private readonly someoneElseJoined$ = this.memberships$.pipe(
     map((ms) => ms.some((m) => m.sender !== this.userId)),
-  );
+  ) as Behavior<boolean>;
 
   /**
    * The current call pickup state of the call.
@@ -960,25 +969,19 @@ export class CallViewModel extends ViewModel {
     ? this.scope.behavior<
         "unknown" | "ringing" | "timeout" | "decline" | "success"
       >(
-        combineLatest([
+        combineLatest(
-          this.livekitConnectionState$,
+          [this.livekitConnectionState$, this.someoneElseJoined$, this.ring$],
-          this.someoneElseJoined$,
+          (livekitConnectionState, someoneElseJoined, ring) => {
-        ]).pipe(
-          switchMap(([livekitConnectionState, someoneElseJoined]) => {
             if (livekitConnectionState === ConnectionState.Disconnected) {
               // Do not ring until we're connected.
-              return of("unknown" as const);
+              return "unknown" as const;
             } else if (someoneElseJoined) {
-              return of("success" as const);
+              return "success" as const;
             }
             // Show the ringing state of the most recent ringing attempt.
-            return this.ring$.pipe(switchAll());
+            // as long as we have not yet sent an RTC notification event, ring will be null -> callPickupState$ = unknown.
-          }),
+            return ring ?? ("unknown" as const);
-          // The state starts as 'unknown' because we don't know if the RTC
+          },
-          // session will actually send a notify event yet. It will only be
-          // known once we send our own membership and see that we were the
-          // first one to join.
-          startWith("unknown" as const),
         ),
       )
     : constant(null);
@@ -1690,7 +1693,11 @@ export class CallViewModel extends ViewModel {
     private readonly livekitRoom: LivekitRoom,
     private readonly mediaDevices: MediaDevices,
     private readonly options: CallViewModelOptions,
+<<<<<<< HEAD
     public readonly livekitConnectionState$: Observable<ECConnectionState>,
+=======
+    public readonly livekitConnectionState$: Behavior<ECConnectionState>,
+>>>>>>> origin/livekit
     private readonly handsRaisedSubject$: Observable<
       Record<string, RaisedHandInfo>
     >,

src/utils/test-viewmodel.ts

@@ -39,6 +39,7 @@ import {
   localRtcMember,
 } from "./test-fixtures";
 import { type RaisedHandInfo, type ReactionInfo } from "../reactions";
+import { constant } from "../state/Behavior";
 
 export function getBasicRTCSession(
   members: RoomMember[],
@@ -154,7 +155,7 @@ export function getBasicCallViewModelEnvironment(
       encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
       ...callViewModelOptions,
     },
-    of(ConnectionState.Connected),
+    constant(ConnectionState.Connected),
     handRaisedSubject$,
     reactionsSubject$,
   );