Add decline logic and tests

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

src/state/CallViewModel.test.ts

@@ -18,7 +18,14 @@ import {
   of,
   switchMap,
 } from "rxjs";
-import { ClientEvent, SyncState, type MatrixClient } from "matrix-js-sdk";
+import {
+  ClientEvent,
+  SyncState,
+  type MatrixClient,
+  RoomEvent as MatrixRoomEvent,
+  MatrixEvent,
+  type IRoomTimelineData,
+} from "matrix-js-sdk";
 import {
   ConnectionState,
   type LocalParticipant,
@@ -1249,10 +1256,12 @@ describe("shouldWaitForCallPickup$", () => {
             r: () => {
               rtcSession.emit(
                 MatrixRTCSessionEvent.DidSendCallNotification,
-                { lifetime: 30 } as unknown as {
+                { event_id: "$notif1", lifetime: 30 } as unknown as {
                   event_id: string;
                 } & IRTCNotificationContent,
-                {} as unknown as { event_id: string } & ICallNotifyContent,
+                { event_id: "$notif1" } as unknown as {
+                  event_id: string;
+                } & ICallNotifyContent,
               );
             },
           });
@@ -1300,10 +1309,10 @@ describe("shouldWaitForCallPickup$", () => {
             r: () => {
               rtcSession.emit(
                 MatrixRTCSessionEvent.DidSendCallNotification,
-                { lifetime: 100 } as unknown as {
+                { event_id: "$notif2", lifetime: 100 } as unknown as {
                   event_id: string;
                 } & IRTCNotificationContent,
-                {} as unknown as {
+                { event_id: "$notif2" } as unknown as {
                   event_id: string;
                 } & ICallNotifyContent,
               );
@@ -1351,10 +1360,10 @@ describe("shouldWaitForCallPickup$", () => {
             r: () => {
               rtcSession.emit(
                 MatrixRTCSessionEvent.DidSendCallNotification,
-                { lifetime: 50 } as unknown as {
+                { event_id: "$notif3", lifetime: 50 } as unknown as {
                   event_id: string;
                 } & IRTCNotificationContent,
-                {} as unknown as {
+                { event_id: "$notif3" } as unknown as {
                   event_id: string;
                 } & ICallNotifyContent,
               );
@@ -1388,10 +1397,10 @@ describe("shouldWaitForCallPickup$", () => {
             r: () => {
               rtcSession.emit(
                 MatrixRTCSessionEvent.DidSendCallNotification,
-                { lifetime: 0 } as unknown as {
+                { event_id: "$notif4", lifetime: 0 } as unknown as {
                   event_id: string;
                 } & IRTCNotificationContent, // no lifetime
-                {} as unknown as {
+                { event_id: "$notif4" } as unknown as {
                   event_id: string;
                 } & ICallNotifyContent,
               );
@@ -1437,10 +1446,10 @@ describe("shouldWaitForCallPickup$", () => {
             r: () => {
               rtcSession.emit(
                 MatrixRTCSessionEvent.DidSendCallNotification,
-                { lifetime: 30 } as unknown as {
+                { event_id: "$notif5", lifetime: 30 } as unknown as {
                   event_id: string;
                 } & IRTCNotificationContent,
-                {} as unknown as {
+                { event_id: "$notif5" } as unknown as {
                   event_id: string;
                 } & ICallNotifyContent,
               );
@@ -1457,6 +1466,149 @@ describe("shouldWaitForCallPickup$", () => {
       );
     });
   });
+
+  test("decline before timeout window ends -> decline", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      withCallViewModel(
+        {
+          remoteParticipants$: scope.behavior(hot("a", { a: [] }), []),
+          rtcMembers$: scope.behavior(hot("a", { a: [localRtcMember] }), []),
+          connectionState$: of(ConnectionState.Connected),
+        },
+        (vm, rtcSession) => {
+          // Notify at 10ms with 50ms lifetime, decline at 40ms with matching id
+          schedule("          10ms r 29ms d", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { event_id: "$decl1", lifetime: 50 } as unknown as {
+                  event_id: string;
+                } & IRTCNotificationContent,
+                { event_id: "$decl1" } as unknown as {
+                  event_id: string;
+                } & ICallNotifyContent,
+              );
+            },
+            d: () => {
+              // Emit decline timeline event with id matching the notification
+              rtcSession.room.emit(
+                MatrixRoomEvent.Timeline,
+                new MatrixEvent({ event_id: "$decl1", type: "m.rtc.decline" }),
+                rtcSession.room,
+                undefined,
+                false,
+                {} as IRoomTimelineData,
+              );
+            },
+          });
+          expectObservable(vm.callPickupState$).toBe("a 9ms b 29ms e", {
+            a: "unknown",
+            b: "ringing",
+            e: "decline",
+          });
+        },
+        {
+          shouldWaitForCallPickup: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("decline after timeout window ends -> stays timeout", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      withCallViewModel(
+        {
+          remoteParticipants$: scope.behavior(hot("a", { a: [] }), []),
+          rtcMembers$: scope.behavior(hot("a", { a: [localRtcMember] }), []),
+          connectionState$: of(ConnectionState.Connected),
+        },
+        (vm, rtcSession) => {
+          // Notify at 10ms with 20ms lifetime (timeout at 30ms), decline at 40ms
+          schedule("          10ms r 20ms t 10ms d", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { event_id: "$decl2", lifetime: 20 } as unknown as {
+                  event_id: string;
+                } & IRTCNotificationContent,
+                { event_id: "$decl2" } as unknown as {
+                  event_id: string;
+                } & ICallNotifyContent,
+              );
+            },
+            t: () => {},
+            d: () => {
+              rtcSession.room.emit(
+                MatrixRoomEvent.Timeline,
+                new MatrixEvent({ event_id: "$decl2", type: "m.rtc.decline" }),
+                rtcSession.room,
+                undefined,
+                false,
+                {} as IRoomTimelineData,
+              );
+            },
+          });
+          expectObservable(vm.callPickupState$).toBe("a 9ms b 19ms c", {
+            a: "unknown",
+            b: "ringing",
+            c: "timeout",
+          });
+        },
+        {
+          shouldWaitForCallPickup: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
+  test("decline with wrong id is ignored (stays ringing)", () => {
+    withTestScheduler(({ hot, schedule, expectObservable, scope }) => {
+      withCallViewModel(
+        {
+          remoteParticipants$: scope.behavior(hot("a", { a: [] }), []),
+          rtcMembers$: scope.behavior(hot("a", { a: [localRtcMember] }), []),
+          connectionState$: of(ConnectionState.Connected),
+        },
+        (vm, rtcSession) => {
+          // Notify at 10ms with id A, decline arrives at 20ms with id B
+          schedule("          10ms r 10ms d", {
+            r: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                { event_id: "$right", lifetime: 50 } as unknown as {
+                  event_id: string;
+                } & IRTCNotificationContent,
+                { event_id: "$right" } as unknown as {
+                  event_id: string;
+                } & ICallNotifyContent,
+              );
+            },
+            d: () => {
+              rtcSession.room.emit(
+                MatrixRoomEvent.Timeline,
+                new MatrixEvent({ event_id: "$wrong", type: "m.rtc.decline" }),
+                rtcSession.room,
+                undefined,
+                false,
+                {} as IRoomTimelineData,
+              );
+            },
+          });
+          // We assert up to 21ms to see the ringing at 10ms and no change at 20ms
+          expectObservable(vm.callPickupState$, "21ms !").toBe("a 9ms b", {
+            a: "unknown",
+            b: "ringing",
+          });
+        },
+        {
+          shouldWaitForCallPickup: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
 });
 
 test("audio output changes when toggling earpiece mode", () => {

src/state/CallViewModel.ts

@@ -19,6 +19,8 @@ import {
 } from "livekit-client";
 import {
   ClientEvent,
+  EventTimelineSetHandlerMap,
+  RoomEvent,
   RoomStateEvent,
   SyncState,
   type Room as MatrixRoom,
@@ -57,6 +59,7 @@ import {
   type IRTCNotificationContent,
   type MatrixRTCSession,
   MatrixRTCSessionEvent,
+  MatrixRTCSessionEventHandlerMap,
   MembershipManagerEvent,
   Status,
 } from "matrix-js-sdk/lib/matrixrtc";
@@ -935,26 +938,35 @@ export class CallViewModel extends ViewModel {
    * "ringing": The notification event was sent.
    * "ringEnded": The notification events lifetime has timed out -> ringing stopped on all receiving clients.
    */
-  private readonly notificationEventIsRingingOthers$: Observable<
+  private readonly rtcNotificationEventState$: Observable<
-    "unknown" | "ringing" | "ringEnded" | null
+    { state: "unknown" | "ringEnded" } | { state: "ringing"; event_id: string }
-  > = fromEvent<[IRTCNotificationContent, ICallNotifyContent]>(
+  > = fromEvent<
-    this.matrixRTCSession,
+    Parameters<
-    MatrixRTCSessionEvent.DidSendCallNotification,
+      MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification]
-  ).pipe(
+    >
+  >(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"),
+          of({
-          timer(lifetime).pipe(map((): "ringEnded" | null => "ringEnded")),
+            state: "ringing",
+            event_id: notificationEvent.event_id,
+          } as {
+            state: "ringing";
+            event_id: string;
+          }),
+          timer(lifetime).pipe(
+            map(() => ({ state: "ringEnded" }) as { state: "ringEnded" }),
+          ),
         );
       }
-      // If no lifetime, just emit true once
+      // If no lifetime, the notify event is basically invalid and we just stay in unknown state.
-      return of(null);
+      return of({ state: "unknown" } as { state: "unknown" });
     }),
-    startWith("unknown" as "unknown" | null),
+    startWith({ state: "unknown" } as { state: "unknown" }),
   );
 
   /**
@@ -980,22 +992,37 @@ export class CallViewModel extends ViewModel {
    *  - null: EC is configured to never show any waiting for answer state.
    */
   public readonly callPickupState$: Behavior<
-    "unknown" | "ringing" | "timeout" | "success" | null
+    "unknown" | "ringing" | "timeout" | "success" | "decline" | null
   > = this.scope.behavior(
     combineLatest([
-      this.notificationEventIsRingingOthers$,
+      this.rtcNotificationEventState$,
       this.someoneElseJoined$,
+      fromEvent<Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>>(
+        this.matrixRoom,
+        RoomEvent.Timeline,
+      ).pipe(
+        map(([event]) => {
+          // TODO use correct decline event type enum.
+          if (event.getType() === "m.rtc.decline") return event;
+          else return null;
+        }),
+        startWith(null),
+      ),
     ]).pipe(
-      map(([isRingingOthers, someoneJoined]) => {
+      map(([notificationEventState, someoneJoined, declineEvent]) => {
         // Never enter waiting for answer state if the app is not configured with waitingForAnswer.
         if (!this.options.shouldWaitForCallPickup) return null;
         // As soon as someone joins, we can consider the call "wait for answer" successful
         if (someoneJoined) return "success";
 
-        switch (isRingingOthers) {
+        switch (notificationEventState?.state) {
           case "unknown":
             return "unknown";
           case "ringing":
+            // Check if the decline event corresponds to the current notification event
+            if (declineEvent?.getId() === notificationEventState.event_id) {
+              return "decline";
+            }
             return "ringing";
           case "ringEnded":
             return "timeout";
@@ -1003,6 +1030,13 @@ export class CallViewModel extends ViewModel {
             return "timeout";
         }
       }),
+      // Once we reach a terminal state, keep it
+      scan((prev, next) => {
+        if (prev === "decline" || prev === "timeout" || prev === "success") {
+          return prev;
+        }
+        return next;
+      }),
       distinctUntilChanged(),
     ),
   );

src/utils/test.ts

@@ -38,6 +38,7 @@ import {
   type RoomAndToDeviceEventsHandlerMap,
 } from "matrix-js-sdk/lib/matrixrtc/RoomAndToDeviceKeyTransport";
 import { type TrackReference } from "@livekit/components-core";
+import EventEmitter from "events";
 
 import {
   LocalUserMediaViewModel,
@@ -143,27 +144,27 @@ export function withTestScheduler(
   scope.end();
 }
 
+
 interface EmitterMock<T> {
-  on: () => T;
+  on: (...args: unknown[]) => T;
-  off: () => T;
+  off: (...args: unknown[]) => T;
-  addListener: () => T;
+  addListener: (...args: unknown[]) => T;
-  removeListener: () => T;
+  removeListener: (...args: unknown[]) => T;
+  emit: (event: string | symbol, ...args: unknown[]) => boolean;
 }
 
 export function mockEmitter<T>(): EmitterMock<T> {
+  const ee = new EventEmitter();
   return {
-    on(): T {
+    on: ee.on.bind(ee) as unknown as (...args: unknown[]) => T,
-      return this as T;
+    off: ee.off.bind(ee) as unknown as (...args: unknown[]) => T,
-    },
+    addListener: ee.addListener.bind(ee) as unknown as (
-    off(): T {
+      ...args: unknown[]
-      return this as T;
+    ) => T,
-    },
+    removeListener: ee.removeListener.bind(ee) as unknown as (
-    addListener(): T {
+      ...args: unknown[]
-      return this as T;
+    ) => T,
-    },
+    emit: ee.emit.bind(ee),
-    removeListener(): T {
-      return this as T;
-    },
   };
 }