Merge pull request #3260 from element-hq/feature/fga/start_sync_on_push

Feature/fga/start sync on push

features/call/impl/src/test/kotlin/io/element/android/features/call/ui/CallScreenPresenterTest.kt

@@ -36,6 +36,7 @@ import io.element.android.libraries.matrix.test.A_SESSION_ID
 import io.element.android.libraries.matrix.test.FakeMatrixClient
 import io.element.android.libraries.matrix.test.FakeMatrixClientProvider
 import io.element.android.libraries.matrix.test.room.FakeMatrixRoom
+import io.element.android.libraries.matrix.test.sync.FakeSyncService
 import io.element.android.libraries.matrix.test.widget.FakeMatrixWidgetDriver
 import io.element.android.libraries.network.useragent.UserAgentProvider
 import io.element.android.services.analytics.api.ScreenTracker
@@ -43,11 +44,13 @@ import io.element.android.services.analytics.test.FakeScreenTracker
 import io.element.android.services.toolbox.api.systemclock.SystemClock
 import io.element.android.tests.testutils.WarmUpRule
 import io.element.android.tests.testutils.consumeItemsUntilTimeout
+import io.element.android.tests.testutils.lambda.assert
 import io.element.android.tests.testutils.lambda.lambdaRecorder
 import io.element.android.tests.testutils.lambda.value
 import io.element.android.tests.testutils.testCoroutineDispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.test.TestScope
@@ -86,8 +89,9 @@ class CallScreenPresenterTest {
     @Test
     fun `present - with CallType RoomCall sets call as active, loads URL, runs WidgetDriver and notifies the other clients a call started`() = runTest {
         val sendCallNotificationIfNeededLambda = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+        val syncService = FakeSyncService(MutableStateFlow(SyncState.Running))
         val fakeRoom = FakeMatrixRoom(sendCallNotificationIfNeededResult = sendCallNotificationIfNeededLambda)
-        val client = FakeMatrixClient().apply {
+        val client = FakeMatrixClient(syncService = syncService).apply {
             givenGetRoomResult(A_ROOM_ID, fakeRoom)
         }
         val widgetDriver = FakeMatrixWidgetDriver()
@@ -216,7 +220,12 @@ class CallScreenPresenterTest {
     fun `present - automatically starts the Matrix client sync when on RoomCall`() = runTest {
         val navigator = FakeCallScreenNavigator()
         val widgetDriver = FakeMatrixWidgetDriver()
-        val matrixClient = FakeMatrixClient()
+        val syncStateFlow = MutableStateFlow(SyncState.Idle)
+        val startSyncLambda = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+        val syncService = FakeSyncService(syncStateFlow = syncStateFlow).apply {
+            this.startSyncLambda = startSyncLambda
+        }
+        val matrixClient = FakeMatrixClient(syncService = syncService)
         val presenter = createCallScreenPresenter(
             callType = CallType.RoomCall(A_SESSION_ID, A_ROOM_ID),
             widgetDriver = widgetDriver,
@@ -230,7 +239,7 @@ class CallScreenPresenterTest {
         }.test {
             consumeItemsUntilTimeout()
 
-            assertThat(matrixClient.syncService().syncState.value).isEqualTo(SyncState.Running)
+            assert(startSyncLambda).isCalledOnce()
 
             cancelAndIgnoreRemainingEvents()
         }
@@ -240,7 +249,12 @@ class CallScreenPresenterTest {
     fun `present - automatically stops the Matrix client sync on dispose`() = runTest {
         val navigator = FakeCallScreenNavigator()
         val widgetDriver = FakeMatrixWidgetDriver()
-        val matrixClient = FakeMatrixClient()
+        val syncStateFlow = MutableStateFlow(SyncState.Running)
+        val stopSyncLambda = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+        val syncService = FakeSyncService(syncStateFlow = syncStateFlow).apply {
+            this.stopSyncLambda = stopSyncLambda
+        }
+        val matrixClient = FakeMatrixClient(syncService = syncService)
         val presenter = createCallScreenPresenter(
             callType = CallType.RoomCall(A_SESSION_ID, A_ROOM_ID),
             widgetDriver = widgetDriver,
@@ -262,7 +276,7 @@ class CallScreenPresenterTest {
 
         job.cancelAndJoin()
 
-        assertThat(matrixClient.syncService().syncState.value).isEqualTo(SyncState.Terminated)
+        assert(stopSyncLambda).isCalledOnce()
     }
 
     private fun TestScope.createCallScreenPresenter(

features/roomlist/impl/src/test/kotlin/io/element/android/features/roomlist/impl/RoomListPresenterTest.kt

@@ -95,6 +95,7 @@ import io.element.android.tests.testutils.testCoroutineDispatchers
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
 import org.junit.Rule
@@ -219,7 +220,7 @@ class RoomListPresenterTest {
         val encryptionService = FakeEncryptionService().apply {
             emitRecoveryState(RecoveryState.INCOMPLETE)
         }
-        val syncService = FakeSyncService(initialState = SyncState.Running)
+        val syncService = FakeSyncService(MutableStateFlow(SyncState.Running))
         val presenter = createRoomListPresenter(
             client = FakeMatrixClient(roomListService = roomListService, encryptionService = encryptionService, syncService = syncService),
             coroutineScope = scope,
@@ -250,7 +251,7 @@ class RoomListPresenterTest {
             sessionVerificationService = FakeSessionVerificationService().apply {
                 givenNeedsSessionVerification(false)
             },
-            syncService = FakeSyncService(initialState = SyncState.Running)
+            syncService = FakeSyncService(MutableStateFlow(SyncState.Running))
         )
         val scope = CoroutineScope(context = coroutineContext + SupervisorJob())
         val presenter = createRoomListPresenter(

libraries/matrix/test/src/main/kotlin/io/element/android/libraries/matrix/test/room/FakeMatrixRoom.kt

@@ -139,6 +139,7 @@ class FakeMatrixRoom(
     private val saveComposerDraftLambda: (ComposerDraft) -> Result<Unit> = { _: ComposerDraft -> Result.success(Unit) },
     private val loadComposerDraftLambda: () -> Result<ComposerDraft?> = { Result.success<ComposerDraft?>(null) },
     private val clearComposerDraftLambda: () -> Result<Unit> = { Result.success(Unit) },
+    private val subscribeToSyncLambda: () -> Unit = { lambdaError() },
 ) : MatrixRoom {
     private val _roomInfoFlow: MutableSharedFlow<MatrixRoomInfo> = MutableSharedFlow(replay = 1)
     override val roomInfoFlow: Flow<MatrixRoomInfo> = _roomInfoFlow
@@ -181,7 +182,9 @@ class FakeMatrixRoom(
         timelineFocusedOnEventResult(eventId)
     }
 
-    override suspend fun subscribeToSync() = Unit
+    override suspend fun subscribeToSync() {
+        subscribeToSyncLambda()
+    }
 
     override suspend fun powerLevels(): Result<MatrixRoomPowerLevels> {
         return powerLevelsResult()

libraries/matrix/test/src/main/kotlin/io/element/android/libraries/matrix/test/sync/FakeSyncService.kt

@@ -22,22 +22,16 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 
 class FakeSyncService(
-    initialState: SyncState = SyncState.Idle
+    syncStateFlow: MutableStateFlow<SyncState> = MutableStateFlow(SyncState.Idle)
 ) : SyncService {
-    private val syncStateFlow = MutableStateFlow(initialState)
+    var startSyncLambda: () -> Result<Unit> = { Result.success(Unit) }
-
-    fun simulateError() {
-        syncStateFlow.value = SyncState.Error
-    }
-
     override suspend fun startSync(): Result<Unit> {
-        syncStateFlow.value = SyncState.Running
+        return startSyncLambda()
-        return Result.success(Unit)
     }
 
+    var stopSyncLambda: () -> Result<Unit> = { Result.success(Unit) }
     override suspend fun stopSync(): Result<Unit> {
-        syncStateFlow.value = SyncState.Terminated
+        return stopSyncLambda()
-        return Result.success(Unit)
     }
 
     override val syncState: StateFlow<SyncState> = syncStateFlow

libraries/push/impl/build.gradle.kts

@@ -82,4 +82,6 @@ dependencies {
     testImplementation(projects.services.appnavstate.test)
     testImplementation(projects.services.toolbox.impl)
     testImplementation(projects.services.toolbox.test)
+    testImplementation(projects.libraries.featureflag.test)
+    testImplementation(libs.kotlinx.collections.immutable)
 }

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/push/OnNotifiableEventReceived.kt

@@ -18,9 +18,6 @@ package io.element.android.libraries.push.impl.push
 
 import com.squareup.anvil.annotations.ContributesBinding
 import io.element.android.libraries.di.AppScope
-import io.element.android.libraries.featureflag.api.FeatureFlagService
-import io.element.android.libraries.featureflag.api.FeatureFlags
-import io.element.android.libraries.matrix.api.MatrixClientProvider
 import io.element.android.libraries.push.impl.notifications.DefaultNotificationDrawerManager
 import io.element.android.libraries.push.impl.notifications.model.NotifiableEvent
 import kotlinx.coroutines.CoroutineScope
@@ -35,23 +32,12 @@ interface OnNotifiableEventReceived {
 class DefaultOnNotifiableEventReceived @Inject constructor(
     private val defaultNotificationDrawerManager: DefaultNotificationDrawerManager,
     private val coroutineScope: CoroutineScope,
-    private val matrixClientProvider: MatrixClientProvider,
+    private val syncOnNotifiableEvent: SyncOnNotifiableEvent,
-    private val featureFlagService: FeatureFlagService,
 ) : OnNotifiableEventReceived {
     override fun onNotifiableEventReceived(notifiableEvent: NotifiableEvent) {
         coroutineScope.launch {
-            subscribeToRoomIfNeeded(notifiableEvent)
+            launch { syncOnNotifiableEvent(notifiableEvent) }
             defaultNotificationDrawerManager.onNotifiableEventReceived(notifiableEvent)
         }
     }
-
-    private fun CoroutineScope.subscribeToRoomIfNeeded(notifiableEvent: NotifiableEvent) = launch {
-        if (!featureFlagService.isFeatureEnabled(FeatureFlags.SyncOnPush)) {
-            return@launch
-        }
-        val client = matrixClientProvider.getOrRestore(notifiableEvent.sessionId).getOrNull() ?: return@launch
-        client.getRoom(notifiableEvent.roomId)?.use { room ->
-            room.subscribeToSync()
-        }
-    }
 }

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/push/SyncOnNotifiableEvent.kt

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2024 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.element.android.libraries.push.impl.push
+
+import io.element.android.libraries.core.coroutine.CoroutineDispatchers
+import io.element.android.libraries.featureflag.api.FeatureFlagService
+import io.element.android.libraries.featureflag.api.FeatureFlags
+import io.element.android.libraries.matrix.api.MatrixClientProvider
+import io.element.android.libraries.matrix.api.core.EventId
+import io.element.android.libraries.matrix.api.room.MatrixRoom
+import io.element.android.libraries.matrix.api.sync.SyncService
+import io.element.android.libraries.matrix.api.timeline.MatrixTimelineItem
+import io.element.android.libraries.push.impl.notifications.model.NotifiableEvent
+import io.element.android.services.appnavstate.api.AppForegroundStateService
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.concurrent.atomic.AtomicInteger
+import javax.inject.Inject
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+
+class SyncOnNotifiableEvent @Inject constructor(
+    private val matrixClientProvider: MatrixClientProvider,
+    private val featureFlagService: FeatureFlagService,
+    private val appForegroundStateService: AppForegroundStateService,
+    private val dispatchers: CoroutineDispatchers,
+) {
+    private var syncCounter = AtomicInteger(0)
+
+    suspend operator fun invoke(notifiableEvent: NotifiableEvent) = withContext(dispatchers.io) {
+        if (!featureFlagService.isFeatureEnabled(FeatureFlags.SyncOnPush)) {
+            return@withContext
+        }
+        val client = matrixClientProvider.getOrRestore(notifiableEvent.sessionId).getOrNull() ?: return@withContext
+        client.getRoom(notifiableEvent.roomId)?.use { room ->
+            room.subscribeToSync()
+
+            // If the app is in foreground, sync is already running, so just add the subscription.
+            if (!appForegroundStateService.isInForeground.value) {
+                val syncService = client.syncService()
+                syncService.startSyncIfNeeded()
+                room.waitsUntilEventIsKnown(eventId = notifiableEvent.eventId, timeout = 10.seconds)
+                syncService.stopSyncIfNeeded()
+            }
+        }
+    }
+
+    private suspend fun MatrixRoom.waitsUntilEventIsKnown(eventId: EventId, timeout: Duration) {
+        withTimeoutOrNull(timeout) {
+            liveTimeline.timelineItems.first { timelineItems ->
+                timelineItems.any { timelineItem ->
+                    when (timelineItem) {
+                        is MatrixTimelineItem.Event -> timelineItem.eventId == eventId
+                        else -> false
+                    }
+                }
+            }
+        }
+    }
+
+    private suspend fun SyncService.startSyncIfNeeded() {
+        if (syncCounter.getAndIncrement() == 0) {
+            startSync()
+        }
+    }
+
+    private suspend fun SyncService.stopSyncIfNeeded() {
+        if (syncCounter.decrementAndGet() == 0 && !appForegroundStateService.isInForeground.value) {
+            stopSync()
+        }
+    }
+}

libraries/push/impl/src/test/kotlin/io/element/android/libraries/push/impl/push/SyncOnNotifiableEventTest.kt

@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2024 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.element.android.libraries.push.impl.push
+
+import io.element.android.libraries.featureflag.api.FeatureFlags
+import io.element.android.libraries.featureflag.test.FakeFeatureFlagService
+import io.element.android.libraries.matrix.api.MatrixClient
+import io.element.android.libraries.matrix.api.sync.SyncState
+import io.element.android.libraries.matrix.api.timeline.MatrixTimelineItem
+import io.element.android.libraries.matrix.test.A_ROOM_ID
+import io.element.android.libraries.matrix.test.A_UNIQUE_ID
+import io.element.android.libraries.matrix.test.FakeMatrixClient
+import io.element.android.libraries.matrix.test.FakeMatrixClientProvider
+import io.element.android.libraries.matrix.test.room.FakeMatrixRoom
+import io.element.android.libraries.matrix.test.sync.FakeSyncService
+import io.element.android.libraries.matrix.test.timeline.FakeTimeline
+import io.element.android.libraries.matrix.test.timeline.anEventTimelineItem
+import io.element.android.libraries.push.impl.notifications.fixtures.aNotifiableMessageEvent
+import io.element.android.services.appnavstate.test.FakeAppForegroundStateService
+import io.element.android.tests.testutils.lambda.assert
+import io.element.android.tests.testutils.lambda.lambdaRecorder
+import io.element.android.tests.testutils.testCoroutineDispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+
+class SyncOnNotifiableEventTest {
+    private val timelineItems = MutableStateFlow<List<MatrixTimelineItem>>(emptyList())
+    private val syncStateFlow = MutableStateFlow(SyncState.Idle)
+    private val startSyncLambda = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+    private val stopSyncLambda = lambdaRecorder<Result<Unit>> { Result.success(Unit) }
+    private val subscribeToSyncLambda = lambdaRecorder<Unit> { }
+
+    private val liveTimeline = FakeTimeline(
+        timelineItems = timelineItems,
+    )
+    private val room = FakeMatrixRoom(
+        roomId = A_ROOM_ID,
+        liveTimeline = liveTimeline,
+        subscribeToSyncLambda = subscribeToSyncLambda
+    )
+    private val syncService = FakeSyncService(syncStateFlow).also {
+        it.startSyncLambda = startSyncLambda
+        it.stopSyncLambda = stopSyncLambda
+    }
+
+    private val client = FakeMatrixClient(
+        syncService = syncService,
+    ).apply {
+        givenGetRoomResult(A_ROOM_ID, room)
+    }
+
+    private val notifiableEvent = aNotifiableMessageEvent()
+
+    @Test
+    fun `when feature flag is disabled, nothing happens`() = runTest {
+        val sut = createSyncOnNotifiableEvent(client = client, isSyncOnPushEnabled = false)
+
+        sut(notifiableEvent)
+
+        assert(startSyncLambda).isNeverCalled()
+        assert(stopSyncLambda).isNeverCalled()
+        assert(subscribeToSyncLambda).isNeverCalled()
+    }
+
+    @Test
+    fun `when feature flag is enabled and app is in foreground, sync is not started`() = runTest {
+        val sut = createSyncOnNotifiableEvent(client = client, isAppInForeground = true, isSyncOnPushEnabled = true)
+
+        sut(notifiableEvent)
+
+        assert(startSyncLambda).isNeverCalled()
+        assert(stopSyncLambda).isNeverCalled()
+        assert(subscribeToSyncLambda).isCalledOnce()
+    }
+
+    @Test
+    fun `when feature flag is enabled and app is in background, sync is started and stopped`() = runTest {
+        val sut = createSyncOnNotifiableEvent(client = client, isAppInForeground = false, isSyncOnPushEnabled = true)
+
+        timelineItems.emit(
+            listOf(MatrixTimelineItem.Event(A_UNIQUE_ID, anEventTimelineItem()))
+        )
+        syncStateFlow.emit(SyncState.Running)
+        sut(notifiableEvent)
+
+        assert(startSyncLambda).isCalledOnce()
+        assert(stopSyncLambda).isCalledOnce()
+        assert(subscribeToSyncLambda).isCalledOnce()
+    }
+
+    @Test
+    fun `when feature flag is enabled and app is in background, running multiple time only call once`() = runTest {
+        val sut = createSyncOnNotifiableEvent(client = client, isAppInForeground = false, isSyncOnPushEnabled = true)
+
+        coroutineScope {
+            launch { sut(notifiableEvent) }
+            launch { sut(notifiableEvent) }
+            launch {
+                delay(1)
+                timelineItems.emit(
+                    listOf(MatrixTimelineItem.Event(A_UNIQUE_ID, anEventTimelineItem()))
+                )
+            }
+        }
+
+        assert(startSyncLambda).isCalledOnce()
+        assert(stopSyncLambda).isCalledOnce()
+        assert(subscribeToSyncLambda).isCalledExactly(2)
+    }
+
+    private fun TestScope.createSyncOnNotifiableEvent(
+        client: MatrixClient = FakeMatrixClient(),
+        isSyncOnPushEnabled: Boolean = true,
+        isAppInForeground: Boolean = true,
+    ): SyncOnNotifiableEvent {
+        val featureFlagService = FakeFeatureFlagService(
+            initialState = mapOf(
+                FeatureFlags.SyncOnPush.key to isSyncOnPushEnabled
+            )
+        )
+        val appForegroundStateService = FakeAppForegroundStateService(
+            initialValue = isAppInForeground
+        )
+        val matrixClientProvider = FakeMatrixClientProvider { Result.success(client) }
+        return SyncOnNotifiableEvent(
+            matrixClientProvider = matrixClientProvider,
+            featureFlagService = featureFlagService,
+            appForegroundStateService = appForegroundStateService,
+            dispatchers = testCoroutineDispatchers(),
+        )
+    }
+}

services/appnavstate/impl/src/test/kotlin/io/element/android/services/appnavstate/impl/DefaultNavigationStateServiceTest.kt

@@ -31,6 +31,7 @@ import io.element.android.services.appnavstate.test.A_ROOM_OWNER
 import io.element.android.services.appnavstate.test.A_SESSION_OWNER
 import io.element.android.services.appnavstate.test.A_SPACE_OWNER
 import io.element.android.services.appnavstate.test.A_THREAD_OWNER
+import io.element.android.services.appnavstate.test.FakeAppForegroundStateService
 import io.element.android.tests.testutils.runCancellableScopeTest
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.first

services/appnavstate/test/src/main/kotlin/io/element/android/services/appnavstate/test/FakeAppForegroundStateService.kt

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.element.android.services.appnavstate.impl
+package io.element.android.services.appnavstate.test
 
 import io.element.android.services.appnavstate.api.AppForegroundStateService
 import kotlinx.coroutines.flow.MutableStateFlow