waitForNotificationAnswer

Signed-off-by: Timo K <toger5@hotmail.de>

src/state/CallViewModel.test.ts

@@ -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 { test, vi, onTestFinished, it } from "vitest";
+import { test, vi, onTestFinished, it, describe } from "vitest";
 import EventEmitter from "events";
 import {
   BehaviorSubject,
@@ -32,6 +32,9 @@ import {
   Status,
   type CallMembership,
   type MatrixRTCSession,
+  type IRTCNotificationContent,
+  type ICallNotifyContent,
+  MatrixRTCSessionEvent,
 } from "matrix-js-sdk/lib/matrixrtc";
 import { deepCompare } from "matrix-js-sdk/lib/utils";
 
@@ -1228,6 +1231,215 @@ test("autoLeaveWhenOthersLeft$ doesn't emits when autoLeaveWhenOthersLeft option
   });
 });
 
+describe("waitForNotificationAnswer$", () => {
+  test("unknown -> ringing -> timeout when notified and nobody joins", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      // No one ever joins (only local user)
+      withCallViewModel(
+        {
+          remoteParticipants$: scope.behavior(hot("a", { a: [] }), []),
+          rtcMembers$: scope.behavior(hot("a", { a: [localRtcMember] }), []),
+          connectionState$: of(ConnectionState.Connected),
+          speaking: new Map(),
+          mediaDevices: mockMediaDevices({}),
+        },
+        (vm, rtcSession) => {
+          // Fire a call notification at 10ms with lifetime 30ms
+          schedule("          10ms r", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { lifetime: 30 } as unknown as IRTCNotificationContent,
+                {} as unknown as ICallNotifyContent,
+              );
+            },
+          });
+
+          expectObservable(vm.waitForNotificationAnswer$).toBe(
+            "a 9ms b 29ms c",
+            { a: "unknown", b: "ringing", c: "timeout" },
+          );
+        },
+        {
+          waitForNotificationAnswer: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("ringing -> success if someone joins before timeout", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
+      const remote$ = scope.behavior(
+        hot("a--b", { a: [], b: [aliceParticipant] }),
+        [],
+      );
+      const rtc$ = scope.behavior(
+        hot("a--b", {
+          a: [localRtcMember],
+          b: [localRtcMember, aliceRtcMember],
+        }),
+        [],
+      );
+
+      withCallViewModel(
+        {
+          remoteParticipants$: remote$,
+          rtcMembers$: rtc$,
+          connectionState$: of(ConnectionState.Connected),
+          speaking: new Map(),
+          mediaDevices: mockMediaDevices({}),
+        },
+        (vm, rtcSession) => {
+          // Notify at 5ms so we enter ringing, then success at 20ms
+          schedule("          5ms r", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { lifetime: 100 } as unknown as IRTCNotificationContent,
+                {} as unknown as ICallNotifyContent,
+              );
+            },
+          });
+
+          expectObservable(vm.waitForNotificationAnswer$).toBe("a 2ms c", {
+            a: "unknown",
+            c: "success",
+          });
+        },
+        {
+          waitForNotificationAnswer: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("success when someone joins before we notify", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      // Join at 10ms, notify later at 20ms (state should stay success)
+      const remote$ = scope.behavior(
+        hot("a-b", { a: [], b: [aliceParticipant] }),
+        [],
+      );
+      const rtc$ = scope.behavior(
+        hot("a-b", {
+          a: [localRtcMember],
+          b: [localRtcMember, aliceRtcMember],
+        }),
+        [],
+      );
+
+      withCallViewModel(
+        {
+          remoteParticipants$: remote$,
+          rtcMembers$: rtc$,
+          connectionState$: of(ConnectionState.Connected),
+          speaking: new Map(),
+          mediaDevices: mockMediaDevices({}),
+        },
+        (vm, rtcSession) => {
+          schedule("          20ms r", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { lifetime: 50 } as unknown as IRTCNotificationContent,
+                {} as unknown as ICallNotifyContent,
+              );
+            },
+          });
+          expectObservable(vm.waitForNotificationAnswer$).toBe("a 1ms b", {
+            a: "unknown",
+            b: "success",
+          });
+        },
+        {
+          waitForNotificationAnswer: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("notify without lifetime -> immediate timeout", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      withCallViewModel(
+        {
+          remoteParticipants$: scope.behavior(hot("a", { a: [] }), []),
+          rtcMembers$: scope.behavior(hot("a", { a: [localRtcMember] }), []),
+          connectionState$: of(ConnectionState.Connected),
+          speaking: new Map(),
+          mediaDevices: mockMediaDevices({}),
+        },
+        (vm, rtcSession) => {
+          schedule("          10ms r", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { lifetime: 0 } as unknown as IRTCNotificationContent, // no lifetime
+                {} as unknown as ICallNotifyContent,
+              );
+            },
+          });
+          expectObservable(vm.waitForNotificationAnswer$).toBe("a 9ms b", {
+            a: "unknown",
+            b: "timeout",
+          });
+        },
+        {
+          waitForNotificationAnswer: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("stays null when waitForNotificationAnswer=false", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      const remote$ = scope.behavior(
+        hot("a--b", { a: [], b: [aliceParticipant] }),
+        [],
+      );
+      const rtc$ = scope.behavior(
+        hot("a--b", {
+          a: [localRtcMember],
+          b: [localRtcMember, aliceRtcMember],
+        }),
+        [],
+      );
+
+      withCallViewModel(
+        {
+          remoteParticipants$: remote$,
+          rtcMembers$: rtc$,
+          connectionState$: of(ConnectionState.Connected),
+          speaking: new Map(),
+          mediaDevices: mockMediaDevices({}),
+        },
+        (vm, rtcSession) => {
+          schedule("          5ms r", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { lifetime: 30 } as unknown as IRTCNotificationContent,
+                {} as unknown as ICallNotifyContent,
+              );
+            },
+          });
+          expectObservable(vm.waitForNotificationAnswer$).toBe("(n)", {
+            n: null,
+          });
+        },
+        {
+          waitForNotificationAnswer: false,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+});
+
 test("audio output changes when toggling earpiece mode", () => {
   withTestScheduler(({ schedule, expectObservable }) => {
     getUrlParams.mockReturnValue({ controlledAudioDevices: true });

src/state/CallViewModel.ts

@@ -53,6 +53,8 @@ import {
 import { logger } from "matrix-js-sdk/lib/logger";
 import {
   type CallMembership,
+  type ICallNotifyContent,
+  type IRTCNotificationContent,
   type MatrixRTCSession,
   MatrixRTCSessionEvent,
   MembershipManagerEvent,
@@ -110,7 +112,13 @@ import { type Behavior } from "./Behavior";
 export interface CallViewModelOptions {
   encryptionSystem: EncryptionSystem;
   autoLeaveWhenOthersLeft?: boolean;
+  /**
+   * If the call is started in a way where we want it to behave like a telephone usecase
+   * If we sent a notification event, we want the ui to show a ringing state
+   */
+  waitForNotificationAnswer?: boolean;
 }
+
 // How long we wait after a focus switch before showing the real participant
 // list again
 const POST_FOCUS_PARTICIPANT_UPDATE_DELAY_MS = 3000;
@@ -427,7 +435,14 @@ export class CallViewModel extends ViewModel {
         MembershipManagerEvent.StatusChanged,
       ).pipe(
         startWith(null),
-        map(() => this.matrixRTCSession.membershipStatus === Status.Connected),
+        map(
+          () =>
+            (
+              this.matrixRTCSession as unknown as {
+                membershipStatus?: Status;
+              }
+            ).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
@@ -438,7 +453,11 @@ export class CallViewModel extends ViewModel {
         MembershipManagerEvent.ProbablyLeft,
       ).pipe(
         startWith(null),
-        map(() => this.matrixRTCSession.probablyLeft !== true),
+        map(
+          () =>
+            (this.matrixRTCSession as unknown as { probablyLeft?: boolean })
+              .probablyLeft !== true,
+        ),
       ),
     ),
   );
@@ -825,49 +844,68 @@ export class CallViewModel extends ViewModel {
    *  - Each participant has a corresponding MatrixRTC membership state event
    *  - There can be multiple participants for one matrix user.
    */
-  public readonly participantChanges$ = this.userMedia$.pipe(
-    map((mediaItems) => mediaItems.map((m) => m.id)),
-    scan<string[], { ids: string[]; joined: string[]; left: string[] }>(
-      (prev, ids) => {
-        const left = prev.ids.filter((id) => !ids.includes(id));
-        const joined = ids.filter((id) => !prev.ids.includes(id));
-        return { ids, joined, left };
-      },
-      { ids: [], joined: [], left: [] },
+  public readonly participantChanges$ = this.scope.behavior(
+    this.userMedia$.pipe(
+      map((mediaItems) => mediaItems.map((m) => m.id)),
+      scan<string[], { ids: string[]; joined: string[]; left: string[] }>(
+        (prev, ids) => {
+          const left = prev.ids.filter((id) => !ids.includes(id));
+          const joined = ids.filter((id) => !prev.ids.includes(id));
+          return { ids, joined, left };
+        },
+        { ids: [], joined: [], left: [] },
+      ),
     ),
   );
 
+  /**
+   * The number of participants currently in the call.
+   *
+   *  - Each participant has one livekit connection
+   *  - Each participant has a corresponding MatrixRTC membership state event
+   *  - There can be multiple participants for one matrix user.
+   */
+  public readonly participantCount$ = this.scope.behavior(
+    this.participantChanges$.pipe(map(({ ids }) => ids.length)),
+  );
+
   /**
    * This observable tracks the matrix users that are currently in the call.
    * There can be just one matrix user with multiple participants (see also participantChanges$)
    */
-  public readonly matrixUserChanges$ = this.userMedia$.pipe(
-    map(
-      (mediaItems) =>
-        new Set(
-          mediaItems
-            .map((m) => m.vm.member?.userId)
-            .filter((id) => id !== undefined),
-        ),
-    ),
-    scan<
-      Set<string>,
-      {
-        userIds: Set<string>;
-        joinedUserIds: Set<string>;
-        leftUserIds: Set<string>;
-      }
-    >(
-      (prevState, userIds) => {
-        const left = new Set(
-          [...prevState.userIds].filter((id) => !userIds.has(id)),
-        );
-        const joined = new Set(
-          [...userIds].filter((id) => !prevState.userIds.has(id)),
-        );
-        return { userIds: userIds, joinedUserIds: joined, leftUserIds: left };
-      },
-      { userIds: new Set(), joinedUserIds: new Set(), leftUserIds: new Set() },
+  public readonly matrixUserChanges$ = this.scope.behavior(
+    this.userMedia$.pipe(
+      map(
+        (mediaItems) =>
+          new Set(
+            mediaItems
+              .map((m) => m.vm.member?.userId)
+              .filter((id) => id !== undefined),
+          ),
+      ),
+      scan<
+        Set<string>,
+        {
+          userIds: Set<string>;
+          joinedUserIds: Set<string>;
+          leftUserIds: Set<string>;
+        }
+      >(
+        (prevState, userIds) => {
+          const left = new Set(
+            [...prevState.userIds].filter((id) => !userIds.has(id)),
+          );
+          const joined = new Set(
+            [...userIds].filter((id) => !prevState.userIds.has(id)),
+          );
+          return { userIds: userIds, joinedUserIds: joined, leftUserIds: left };
+        },
+        {
+          userIds: new Set(),
+          joinedUserIds: new Set(),
+          leftUserIds: new Set(),
+        },
+      ),
     ),
   );
 
@@ -891,6 +929,84 @@ export class CallViewModel extends ViewModel {
     map(() => {}),
   );
 
+  /**
+   * "unknown": We don't know if the RTC session decides to send a notify event yet.
+   *   It will only be known once we sent our own membership and know we were the first one to join.
+   * "ringing": The notification event was sent.
+   * "ringEnded": The notification events lifetime has timed out -> ringing stopped on all receiving clients.
+   */
+  private readonly notificationEventIsRingingOthers$: Observable<
+    "unknown" | "ringing" | "ringEnded" | null
+  > = fromEvent<[IRTCNotificationContent, ICallNotifyContent]>(
+    this.matrixRTCSession,
+    MatrixRTCSessionEvent.DidSendCallNotification,
+  ).pipe(
+    switchMap(([notificationEvent]) => {
+      // event.lifetime is expected to be in ms
+      const lifetime = notificationEvent?.lifetime ?? 0;
+      if (lifetime > 0) {
+        // Emit true immediately, then false after lifetime ms
+        return concat(
+          of<"ringing" | null>("ringing"),
+          timer(lifetime).pipe(map((): "ringEnded" | null => "ringEnded")),
+        );
+      }
+      // If no lifetime, just emit true once
+      return of(null);
+    }),
+    startWith("unknown" as "unknown" | null),
+  );
+
+  /**
+   * If some other matrix user has joined the call. It can start with true if there are already multiple matrix users.
+   */
+  private readonly someoneElseJoined$ = this.matrixUserChanges$.pipe(
+    scan(
+      (someoneJoined, { joinedUserIds }) =>
+        someoneJoined || [...joinedUserIds].some((id) => id !== this.userId),
+      false,
+    ),
+    startWith(this.matrixUserChanges$.value.userIds.size > 1),
+  );
+
+  /**
+   * The current waiting for answer state of the call.
+   *  - "ringing": The call is ringing on other devices in this room (This client should give audiovisual feedback that this is happening).
+   *  - "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.
+   *  - "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. Stop audiovisual feedback.
+   *  - null: EC is configured to never show any waiting for answer state.
+   */
+  public readonly waitForNotificationAnswer$: Behavior<
+    "unknown" | "ringing" | "timeout" | "success" | null
+  > = this.scope.behavior(
+    combineLatest([
+      this.notificationEventIsRingingOthers$,
+      this.someoneElseJoined$,
+    ]).pipe(
+      map(([isRingingOthers, someoneJoined]) => {
+        // Never enter waiting for answer state if the app is not configured with waitingForAnswer.
+        if (!this.options.waitForNotificationAnswer) return null;
+        // As soon as someone joins, we can consider the call "wait for answer" successful
+        if (someoneJoined) return "success";
+
+        switch (isRingingOthers) {
+          case "unknown":
+            return "unknown";
+          case "ringing":
+            return "ringing";
+          case "ringEnded":
+            return "timeout";
+          default:
+            return "timeout";
+        }
+      }),
+      distinctUntilChanged(),
+    ),
+  );
+
   /**
    * List of MediaItems that we want to display, that are of type ScreenShare
    */