Make sure declining a call stops observing the ringing call state (#5563)

* Add shared `removeCurrentCall` function to `DefaultActiveCallManager`

This centralises the shared call cancellation logic

* Add regression test for the issue

* Make sure the existing iterations of `flatMapLatest` in `observeRingingCall` get cancelled when the active call is null or not ringing anymore by passing null values, then filtering them out

Previously these kept running even if the `activeCall` was no longer valid

* Move the `timedOutCallJob` cancellation inside `removeCurrentCall` too

features/call/impl/src/main/kotlin/io/element/android/features/call/impl/utils/ActiveCallManager.kt

@@ -28,7 +28,9 @@ import io.element.android.libraries.core.extensions.runCatchingExceptions
 import io.element.android.libraries.di.annotations.AppCoroutineScope
 import io.element.android.libraries.di.annotations.ApplicationContext
 import io.element.android.libraries.matrix.api.MatrixClientProvider
+import io.element.android.libraries.matrix.api.core.EventId
 import io.element.android.libraries.matrix.api.core.SessionId
+import io.element.android.libraries.matrix.api.room.BaseRoom
 import io.element.android.libraries.matrix.ui.media.ImageLoaderHolder
 import io.element.android.libraries.push.api.notifications.ForegroundServiceType
 import io.element.android.libraries.push.api.notifications.NotificationIdProvider
@@ -39,16 +41,17 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.drop
 import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapLatest
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
@@ -180,13 +183,7 @@ class DefaultActiveCallManager(
 
         val previousActiveCall = activeCall.value ?: return
         val notificationData = (previousActiveCall.callState as? CallState.Ringing)?.notificationData ?: return
-        activeCall.value = null
+        removeCurrentCall()
-        if (activeWakeLock?.isHeld == true) {
-            Timber.tag(tag).d("Releasing partial wakelock after timeout")
-            activeWakeLock.release()
-        }
-
-        cancelIncomingCallNotification()
 
         if (displayMissedCallNotification) {
             displayMissedCallNotification(notificationData)
@@ -211,24 +208,16 @@ class DefaultActiveCallManager(
                 ?.declineCall(notificationData.eventId)
         }
 
-        cancelIncomingCallNotification()
+        removeCurrentCall()
-        if (activeWakeLock?.isHeld == true) {
-            Timber.tag(tag).d("Releasing partial wakelock after hang up")
-            activeWakeLock.release()
-        }
-        timedOutCallJob?.cancel()
-        activeCall.value = null
     }
 
+    /**
+     * Removes the current active call and any associated UI, cancelling the timeouts and wakelocks.
+     */
     override suspend fun joinedCall(callType: CallType) = mutex.withLock {
         Timber.tag(tag).d("Joined call: $callType")
 
-        cancelIncomingCallNotification()
+        removeCurrentCall()
-        if (activeWakeLock?.isHeld == true) {
-            Timber.tag(tag).d("Releasing partial wakelock after joining call")
-            activeWakeLock.release()
-        }
-        timedOutCallJob?.cancel()
 
         activeCall.value = ActiveCall(
             callType = callType,
@@ -236,6 +225,23 @@ class DefaultActiveCallManager(
         )
     }
 
+    @VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
+    internal fun removeCurrentCall() {
+        // Cancel and remove the timeout call job, if any
+        timedOutCallJob?.cancel()
+        timedOutCallJob = null
+
+        // Remove the active call and cancel the notification
+        activeCall.value = null
+        cancelIncomingCallNotification()
+
+        // Also remove any wake locks that may be held
+        if (activeWakeLock?.isHeld == true) {
+            Timber.tag(tag).d("Releasing partial wakelock after call declined from another session")
+            activeWakeLock.release()
+        }
+    }
+
     @SuppressLint("MissingPermission")
     private suspend fun showIncomingCallNotification(notificationData: CallNotificationData) {
         Timber.tag(tag).d("Displaying ringing call notification")
@@ -281,73 +287,75 @@ class DefaultActiveCallManager(
 
     @OptIn(ExperimentalCoroutinesApi::class)
     private fun observeRingingCall() {
-        activeCall
+        val roomForActiveCallFlow: Flow<Pair<BaseRoom, EventId>?> = activeCall.mapLatest { activeCall ->
-            .filterNotNull()
+            val callType = activeCall?.callType as? CallType.RoomCall ?: return@mapLatest null
-            .filter { it.callState is CallState.Ringing && it.callType is CallType.RoomCall }
+            val ringingInfo = activeCall.callState as? CallState.Ringing ?: return@mapLatest null
-            .flatMapLatest { activeCall ->
+            val client = matrixClientProvider.getOrRestore(callType.sessionId).getOrNull() ?: run {
-                val callType = activeCall.callType as CallType.RoomCall
+                Timber.tag(tag).d("Couldn't find session for incoming call: $activeCall")
-                val ringingInfo = activeCall.callState as CallState.Ringing
+                return@mapLatest null
-                val client = matrixClientProvider.getOrRestore(callType.sessionId).getOrNull() ?: run {
+            }
-                    Timber.tag(tag).d("Couldn't find session for incoming call: $activeCall")
+            val room = client.getRoom(callType.roomId) ?: run {
-                    return@flatMapLatest flowOf()
+                Timber.tag(tag).d("Couldn't find room for incoming call: $activeCall")
-                }
+                return@mapLatest null
-                val room = client.getRoom(callType.roomId) ?: run {
+            }
-                    Timber.tag(tag).d("Couldn't find room for incoming call: $activeCall")
-                    return@flatMapLatest flowOf()
-                }
 
-                Timber.tag(tag).d("Found room for ringing call: ${room.roomId}")
+            Timber.tag(tag).d("Found room for ringing call: ${room.roomId}")
+
+            val eventId = ringingInfo.notificationData.eventId
+            room to eventId
+        }
+
+        roomForActiveCallFlow
+            .flatMapLatest { pair ->
+                val (room, eventId) = pair
+                    // This will cancel the previous iteration of flatMapLatest if the active call is now null
+                    ?: return@flatMapLatest flowOf()
 
                 // If we have declined from another phone we want to stop ringing.
-                room.subscribeToCallDecline(ringingInfo.notificationData.eventId)
+                room.subscribeToCallDecline(eventId)
                     .filter { decliner ->
                         Timber.tag(tag).d("Call: $activeCall was declined by $decliner")
                         // only want to listen if the call was declined from another of my sessions,
                         // (we are ringing for an incoming call in a DM)
-                        decliner == client.sessionId
+                        decliner == room.sessionId
                     }
             }
             .onEach { decliner ->
                 Timber.tag(tag).d("Call: $activeCall was declined by user from another session")
-                // Remove the active call and cancel the notification
+                removeCurrentCall()
-                activeCall.value = null
-                if (activeWakeLock?.isHeld == true) {
-                    Timber.tag(tag).d("Releasing partial wakelock after call declined from another session")
-                    activeWakeLock.release()
-                }
-                cancelIncomingCallNotification()
             }
             .launchIn(coroutineScope)
+
         // This will observe ringing calls and ensure they're terminated if the room call is cancelled or if the user
         // has joined the call from another session.
-        activeCall
+        roomForActiveCallFlow
-            .filterNotNull()
+            .flatMapLatest { pair ->
-            .filter { it.callState is CallState.Ringing && it.callType is CallType.RoomCall }
+                val (room, _) = pair
-            .flatMapLatest { activeCall ->
+                // This will cancel the previous iteration of flatMapLatest if the active call is now null
-                val callType = activeCall.callType as CallType.RoomCall
+                    ?: return@flatMapLatest flowOf()
-                // Get a flow of updated `hasRoomCall` and `activeRoomCallParticipants` values for the room
+
-                val room = matrixClientProvider.getOrRestore(callType.sessionId).getOrNull()?.getRoom(callType.roomId) ?: run {
+                // We now observe the room info for changes to the active call state and the call participants
-                    Timber.tag(tag).d("Couldn't find room for incoming call: $activeCall")
-                    return@flatMapLatest flowOf()
-                }
                 room.roomInfoFlow.map {
-                    Timber.tag(tag).d("Has room call status changed for ringing call: ${it.hasRoomCall}")
+                    val participants = it.activeRoomCallParticipants
-                    it.hasRoomCall to (callType.sessionId in it.activeRoomCallParticipants)
+                    Timber.tag(tag).d("Room call status changed for ringing call | hasRoomCall: ${it.hasRoomCall} | participants: $participants")
+                    val userIsInTheCall = room.sessionId in participants
+                    it.hasRoomCall to userIsInTheCall
                 }
             }
-            // We only want to check if the room active call status changes
+            // Filter out duplicate values
             .distinctUntilChanged()
             // Skip the first one, we're not interested in it (if the check below passes, it had to be active anyway)
             .drop(1)
             .onEach { (roomHasActiveCall, userIsInTheCall) ->
                 if (!roomHasActiveCall) {
-                    // The call was cancelled
+                    val notificationData = (activeCall.value?.callState as? CallState.Ringing)?.notificationData
-                    timedOutCallJob?.cancel()
+                    removeCurrentCall()
-                    incomingCallTimedOut(displayMissedCallNotification = true)
+
+                    if (notificationData != null) {
+                        displayMissedCallNotification(notificationData)
+                    }
                 } else if (userIsInTheCall) {
-                    // The user joined the call from another session
+                    removeCurrentCall()
-                    timedOutCallJob?.cancel()
-                    incomingCallTimedOut(displayMissedCallNotification = false)
                 }
             }
             .launchIn(coroutineScope)

features/call/impl/src/test/kotlin/io/element/android/features/call/utils/DefaultActiveCallManagerTest.kt

@@ -28,6 +28,7 @@ import io.element.android.libraries.matrix.test.AN_EVENT_ID_2
 import io.element.android.libraries.matrix.test.A_ROOM_ID
 import io.element.android.libraries.matrix.test.A_ROOM_ID_2
 import io.element.android.libraries.matrix.test.A_SESSION_ID
+import io.element.android.libraries.matrix.test.A_USER_ID_2
 import io.element.android.libraries.matrix.test.FakeMatrixClient
 import io.element.android.libraries.matrix.test.FakeMatrixClientProvider
 import io.element.android.libraries.matrix.test.room.FakeBaseRoom
@@ -46,6 +47,7 @@ import io.element.android.tests.testutils.lambda.value
 import io.element.android.tests.testutils.plantTestTimber
 import io.mockk.coVerify
 import io.mockk.mockk
+import io.mockk.spyk
 import io.mockk.verify
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.TestScope
@@ -331,6 +333,49 @@ class DefaultActiveCallManagerTest {
         assertThat(manager.activeCall.value).isNull()
     }
 
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `observeRingingCalls - declining won't do anything if the call was already cancelled`() = runTest {
+        val room = FakeBaseRoom().apply {
+            givenRoomInfo(aRoomInfo())
+        }
+        val client = FakeMatrixClient().apply {
+            givenGetRoomResult(A_ROOM_ID, room)
+        }
+        val matrixClientProvider = FakeMatrixClientProvider(getClient = { Result.success(client) })
+        val notificationManagerCompat = mockk<NotificationManagerCompat>(relaxed = true)
+        val manager = spyk<DefaultActiveCallManager>(
+            createActiveCallManager(
+                matrixClientProvider = matrixClientProvider,
+                notificationManagerCompat = notificationManagerCompat,
+            )
+        )
+
+        manager.registerIncomingCall(aCallNotificationData())
+
+        // Call is active (the other user join the call)
+        room.givenRoomInfo(aRoomInfo(hasRoomCall = true))
+        advanceTimeBy(1)
+
+        // Call is cancelled by us, hanging up
+        manager.hungUpCall(CallType.RoomCall(A_SESSION_ID, A_ROOM_ID))
+        advanceTimeBy(1)
+
+        verify(exactly = 1) { notificationManagerCompat.cancel(any()) }
+        verify(exactly = 1) { manager.removeCurrentCall() }
+        assertThat(manager.activeCall.value).isNull()
+        assertThat(manager.activeWakeLock?.isHeld).isNull()
+
+        // Simulate that another user declined the call
+        room.givenDecliner(A_USER_ID_2, AN_EVENT_ID)
+        advanceTimeBy(1)
+
+        // Check everything stays the same, no extra call to cancelling notifications
+        verify(exactly = 1) { notificationManagerCompat.cancel(any()) }
+        verify(exactly = 1) { manager.removeCurrentCall() }
+        assertThat(manager.activeWakeLock?.isHeld).isNull()
+    }
+
     @OptIn(ExperimentalCoroutinesApi::class)
     @Test
     fun `observeRingingCalls - will do nothing if either the session or the room are not found`() = runTest {