Create SyncOrchestrator

appnav/src/main/kotlin/io/element/android/appnav/LoggedInFlowNode.kt

@@ -73,6 +73,7 @@ import io.element.android.libraries.matrix.api.core.RoomIdOrAlias
 import io.element.android.libraries.matrix.api.core.UserId
 import io.element.android.libraries.matrix.api.core.toRoomIdOrAlias
 import io.element.android.libraries.matrix.api.permalink.PermalinkData
+import io.element.android.libraries.matrix.api.sync.SyncOrchestratorProvider
 import io.element.android.libraries.matrix.api.verification.SessionVerificationRequestDetails
 import io.element.android.libraries.matrix.api.verification.SessionVerificationServiceListener
 import io.element.android.libraries.preferences.api.store.EnableNativeSlidingSyncUseCase
@@ -106,7 +107,7 @@ class LoggedInFlowNode @AssistedInject constructor(
     private val logoutEntryPoint: LogoutEntryPoint,
     private val incomingVerificationEntryPoint: IncomingVerificationEntryPoint,
     private val enableNativeSlidingSyncUseCase: EnableNativeSlidingSyncUseCase,
-    private val syncOrchestrator: SyncOrchestrator,
+    private val syncOrchestratorProvider: SyncOrchestratorProvider,
     snackbarDispatcher: SnackbarDispatcher,
 ) : BaseFlowNode<LoggedInFlowNode.NavTarget>(
     backstack = BackStack(
@@ -124,7 +125,6 @@ class LoggedInFlowNode @AssistedInject constructor(
         fun onOpenBugReport()
     }
 
-    private val syncService = matrixClient.syncService()
     private val loggedInFlowProcessor = LoggedInEventProcessor(
         snackbarDispatcher,
         matrixClient.roomMembershipObserver(),
@@ -139,7 +139,7 @@ class LoggedInFlowNode @AssistedInject constructor(
     override fun onBuilt() {
         super.onBuilt()
 
-        syncOrchestrator.start()
+        syncOrchestratorProvider.get(sessionId = matrixClient.sessionId)?.start()
 
         lifecycle.subscribe(
             onCreate = {

appnav/src/main/kotlin/io/element/android/appnav/SyncOrchestrator.kt → appnav/src/main/kotlin/io/element/android/appnav/di/DefaultSyncOrchestrator.kt

@@ -5,15 +5,16 @@
  * Please see LICENSE files in the repository root for full details.
  */
 
-package io.element.android.appnav
+package io.element.android.appnav.di
 
+import dagger.assisted.Assisted
+import dagger.assisted.AssistedFactory
+import dagger.assisted.AssistedInject
 import io.element.android.features.networkmonitor.api.NetworkMonitor
 import io.element.android.features.networkmonitor.api.NetworkStatus
 import io.element.android.libraries.core.coroutine.CoroutineDispatchers
-import io.element.android.libraries.di.SessionScope
-import io.element.android.libraries.di.SingleIn
-import io.element.android.libraries.di.annotations.SessionCoroutineScope
 import io.element.android.libraries.matrix.api.MatrixClient
+import io.element.android.libraries.matrix.api.sync.SyncOrchestrator
 import io.element.android.libraries.matrix.api.sync.SyncState
 import io.element.android.services.appnavstate.api.AppForegroundStateService
 import kotlinx.coroutines.CoroutineName
@@ -28,21 +29,22 @@
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import timber.log.Timber
-import javax.inject.Inject
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
-/**
- * Observes the app state and network state to start/stop the sync service.
- */
-@SingleIn(SessionScope::class)
-class SyncOrchestrator @Inject constructor(
-    matrixClient: MatrixClient,
+class DefaultSyncOrchestrator @AssistedInject constructor(
+    @Assisted matrixClient: MatrixClient,
+    private val baseCoroutineScope: CoroutineScope = matrixClient.sessionCoroutineScope,
     private val appForegroundStateService: AppForegroundStateService,
     private val networkMonitor: NetworkMonitor,
-    @SessionCoroutineScope private val sessionCoroutineScope: CoroutineScope,
     private val dispatchers: CoroutineDispatchers,
-) {
+) : SyncOrchestrator {
+    @AssistedFactory
+    interface Factory {
+        fun create(matrixClient: MatrixClient): DefaultSyncOrchestrator
+    }
+
     private val syncService = matrixClient.syncService()
 
     private val initialSyncMutex = Mutex()
@@ -51,18 +53,25 @@
 
     private val tag = "SyncOrchestrator"
 
+    private val started = AtomicBoolean(false)
+
     /**
      * Starting observing the app state and network state to start/stop the sync service.
      *
      * Before observing the state, a first attempt at starting the sync service will happen if it's not already running.
      */
     @OptIn(FlowPreview::class)
-    fun start() {
+    override fun start() {
+        if (!started.compareAndSet(false, true)) {
+            Timber.tag(tag).d("already started, exiting early")
+            return
+        }
+
         Timber.tag(tag).d("start observing the app and network state")
 
         if (syncService.syncState.value != SyncState.Running) {
             Timber.tag(tag).d("initial startSync")
-            sessionCoroutineScope.launch(dispatchers.io) {
+            baseCoroutineScope.launch(dispatchers.io) {
                 try {
                     initialSyncMutex.lock()
                     syncService.startSync()
@@ -75,7 +84,7 @@
             }
         }
 
-        coroutineScope = CoroutineScope(sessionCoroutineScope.coroutineContext + CoroutineName(tag) + dispatchers.io)
+        coroutineScope = CoroutineScope(baseCoroutineScope.coroutineContext + CoroutineName(tag) + dispatchers.io)
 
         coroutineScope?.launch {
             // Wait until the initial sync is done, either successfully or failing
@@ -96,23 +105,23 @@
                 if (syncState == SyncState.Running && (!isAppActive || !isNetworkAvailable)) {
                     // Don't stop the sync immediately, wait a bit to avoid starting/stopping the sync too often
                     delay(3.seconds)
-                    SyncObserverAction.StopSync
+                    SyncStateAction.StopSync
                 } else if (syncState != SyncState.Running && isAppActive && isNetworkAvailable) {
-                    SyncObserverAction.StartSync
+                    SyncStateAction.StartSync
                 } else {
-                    SyncObserverAction.NoOp
+                    SyncStateAction.NoOp
                 }
             }
                 .distinctUntilChanged()
                 .collect { action ->
                     when (action) {
-                        SyncObserverAction.StartSync -> {
+                        SyncStateAction.StartSync -> {
                             syncService.startSync()
                         }
-                        SyncObserverAction.StopSync -> {
+                        SyncStateAction.StopSync -> {
                             syncService.stopSync()
                         }
-                        SyncObserverAction.NoOp -> Unit
+                        SyncStateAction.NoOp -> Unit
                     }
                 }
         }
@@ -121,14 +130,18 @@
     /**
      * Stop observing the app state and network state.
      */
-    fun stop() {
+    override fun stop() {
+        if (!started.compareAndSet(true, false)) {
+            Timber.tag(tag).d("already stopped, exiting early")
+            return
+        }
         Timber.tag(tag).d("stop observing the app and network state")
         coroutineScope?.cancel()
         coroutineScope = null
     }
 }
 
-private enum class SyncObserverAction {
+private enum class SyncStateAction {
     StartSync,
     StopSync,
     NoOp,

appnav/src/main/kotlin/io/element/android/appnav/di/DefaultSyncOrchestratorProvider.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2025 New Vector Ltd.
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+ * Please see LICENSE files in the repository root for full details.
+ */
+
+package io.element.android.appnav.di
+
+import com.squareup.anvil.annotations.ContributesBinding
+import io.element.android.libraries.di.AppScope
+import io.element.android.libraries.di.SingleIn
+import io.element.android.libraries.matrix.api.core.SessionId
+import io.element.android.libraries.matrix.api.sync.SyncOrchestrator
+import io.element.android.libraries.matrix.api.sync.SyncOrchestratorProvider
+import javax.inject.Inject
+
+@SingleIn(AppScope::class)
+@ContributesBinding(AppScope::class)
+class DefaultSyncOrchestratorProvider @Inject constructor(
+    private val matrixClientsHolder: MatrixClientsHolder,
+) : SyncOrchestratorProvider {
+    override fun get(sessionId: SessionId): SyncOrchestrator? {
+        return matrixClientsHolder.getSyncOrchestratorOrNull(sessionId)
+    }
+}

appnav/src/main/kotlin/io/element/android/appnav/di/MatrixClientsHolder.kt

@@ -16,6 +16,7 @@ import io.element.android.libraries.matrix.api.MatrixClient
 import io.element.android.libraries.matrix.api.MatrixClientProvider
 import io.element.android.libraries.matrix.api.auth.MatrixAuthenticationService
 import io.element.android.libraries.matrix.api.core.SessionId
+import io.element.android.libraries.matrix.api.sync.SyncOrchestrator
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -29,13 +30,14 @@ private const val SAVE_INSTANCE_KEY = "io.element.android.x.di.MatrixClientsHold
 @ContributesBinding(AppScope::class)
 class MatrixClientsHolder @Inject constructor(
     private val authenticationService: MatrixAuthenticationService,
+    private val syncOrchestratorFactory: DefaultSyncOrchestrator.Factory,
 ) : MatrixClientProvider {
-    private val sessionIdsToMatrixClient = ConcurrentHashMap<SessionId, MatrixClient>()
+    private val sessionIdsToMatrixClient = ConcurrentHashMap<SessionId, Pair<MatrixClient, SyncOrchestrator>>()
     private val restoreMutex = Mutex()
 
     init {
         authenticationService.listenToNewMatrixClients { matrixClient ->
-            sessionIdsToMatrixClient[matrixClient.sessionId] = matrixClient
+            sessionIdsToMatrixClient[matrixClient.sessionId] = matrixClient to syncOrchestratorFactory.create(matrixClient)
         }
     }
 
@@ -48,18 +50,22 @@ class MatrixClientsHolder @Inject constructor(
     }
 
     override fun getOrNull(sessionId: SessionId): MatrixClient? {
-        return sessionIdsToMatrixClient[sessionId]
+        return sessionIdsToMatrixClient[sessionId]?.first
     }
 
     override suspend fun getOrRestore(sessionId: SessionId): Result<MatrixClient> {
         return restoreMutex.withLock {
-            when (val matrixClient = getOrNull(sessionId)) {
+            when (val cached = getOrNull(sessionId)) {
                 null -> restore(sessionId)
-                else -> Result.success(matrixClient)
+                else -> Result.success(cached)
             }
         }
     }
 
+    internal fun getSyncOrchestratorOrNull(sessionId: SessionId): SyncOrchestrator? {
+        return sessionIdsToMatrixClient[sessionId]?.second
+    }
+
     @Suppress("UNCHECKED_CAST")
     fun restoreWithSavedState(state: SavedStateMap?) {
         Timber.d("Restore state")
@@ -88,7 +94,7 @@ class MatrixClientsHolder @Inject constructor(
         Timber.d("Restore matrix session: $sessionId")
         return authenticationService.restoreSession(sessionId)
             .onSuccess { matrixClient ->
-                sessionIdsToMatrixClient[matrixClient.sessionId] = matrixClient
+                sessionIdsToMatrixClient[matrixClient.sessionId] = matrixClient to syncOrchestratorFactory.create(matrixClient)
             }
             .onFailure {
                 Timber.e(it, "Fail to restore session")

appnav/src/test/kotlin/io/element/android/appnav/SyncOrchestratorTest.kt → appnav/src/test/kotlin/io/element/android/appnav/DefaultSyncOrchestratorTest.kt

@@ -7,6 +7,7 @@
 
 package io.element.android.appnav
 
+import io.element.android.appnav.di.DefaultSyncOrchestrator
 import io.element.android.features.networkmonitor.api.NetworkStatus
 import io.element.android.features.networkmonitor.test.FakeNetworkMonitor
 import io.element.android.libraries.matrix.api.sync.SyncState
@@ -29,7 +30,7 @@ import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 @OptIn(ExperimentalCoroutinesApi::class)
-class SyncOrchestratorTest {
+class DefaultSyncOrchestratorTest {
     @get:Rule
     val warmUpRule = WarmUpRule()
 
@@ -322,11 +323,11 @@ class SyncOrchestratorTest {
         syncService: FakeSyncService = FakeSyncService(),
         networkMonitor: FakeNetworkMonitor = FakeNetworkMonitor(),
         appForegroundStateService: FakeAppForegroundStateService = FakeAppForegroundStateService(),
-    ) = SyncOrchestrator(
+    ) = DefaultSyncOrchestrator(
         matrixClient = FakeMatrixClient(syncService = syncService),
         networkMonitor = networkMonitor,
         appForegroundStateService = appForegroundStateService,
-        sessionCoroutineScope = CoroutineScope(coroutineContext + SupervisorJob()),
+        baseCoroutineScope = CoroutineScope(coroutineContext + SupervisorJob()),
         dispatchers = testCoroutineDispatchers(),
     )
 }

appnav/src/test/kotlin/io/element/android/appnav/di/MatrixClientsHolderTest.kt

@@ -9,24 +9,47 @@ package io.element.android.appnav.di
 
 import com.bumble.appyx.core.state.MutableSavedStateMapImpl
 import com.google.common.truth.Truth.assertThat
+import io.element.android.features.networkmonitor.test.FakeNetworkMonitor
+import io.element.android.libraries.matrix.api.MatrixClient
 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.auth.FakeMatrixAuthenticationService
+import io.element.android.services.appnavstate.test.FakeAppForegroundStateService
+import io.element.android.tests.testutils.testCoroutineDispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 
 class MatrixClientsHolderTest {
     @Test
-    fun `test getOrNull`() {
+    fun `test getOrNull`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         assertThat(matrixClientsHolder.getOrNull(A_SESSION_ID)).isNull()
     }
 
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `test getSyncOrchestratorOrNull`() = runTest {
+        val fakeAuthenticationService = FakeMatrixAuthenticationService()
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
+
+        // With no matrix client there is no sync orchestrator
+        assertThat(matrixClientsHolder.getOrNull(A_SESSION_ID)).isNull()
+        assertThat(matrixClientsHolder.getSyncOrchestratorOrNull(A_SESSION_ID)).isNull()
+
+        // But as soon as we receive a client, we can get the sync orchestrator
+        val fakeMatrixClient = FakeMatrixClient()
+        fakeAuthenticationService.givenMatrixClient(fakeMatrixClient)
+        assertThat(matrixClientsHolder.getOrRestore(A_SESSION_ID).getOrNull()).isEqualTo(fakeMatrixClient)
+        assertThat(matrixClientsHolder.getSyncOrchestratorOrNull(A_SESSION_ID)).isNotNull()
+    }
+
     @Test
     fun `test getOrRestore`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         val fakeMatrixClient = FakeMatrixClient()
         fakeAuthenticationService.givenMatrixClient(fakeMatrixClient)
         assertThat(matrixClientsHolder.getOrNull(A_SESSION_ID)).isNull()
@@ -39,7 +62,7 @@ class MatrixClientsHolderTest {
     @Test
     fun `test remove`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         val fakeMatrixClient = FakeMatrixClient()
         fakeAuthenticationService.givenMatrixClient(fakeMatrixClient)
         assertThat(matrixClientsHolder.getOrRestore(A_SESSION_ID).getOrNull()).isEqualTo(fakeMatrixClient)
@@ -52,7 +75,7 @@ class MatrixClientsHolderTest {
     @Test
     fun `test remove all`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         val fakeMatrixClient = FakeMatrixClient()
         fakeAuthenticationService.givenMatrixClient(fakeMatrixClient)
         assertThat(matrixClientsHolder.getOrRestore(A_SESSION_ID).getOrNull()).isEqualTo(fakeMatrixClient)
@@ -65,7 +88,7 @@ class MatrixClientsHolderTest {
     @Test
     fun `test save and restore`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         val fakeMatrixClient = FakeMatrixClient()
         fakeAuthenticationService.givenMatrixClient(fakeMatrixClient)
         matrixClientsHolder.getOrRestore(A_SESSION_ID)
@@ -85,7 +108,7 @@ class MatrixClientsHolderTest {
     @Test
     fun `test AuthenticationService listenToNewMatrixClients emits a Client value and we save it`() = runTest {
         val fakeAuthenticationService = FakeMatrixAuthenticationService()
-        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService)
+        val matrixClientsHolder = MatrixClientsHolder(fakeAuthenticationService, createSyncOrchestratorFactory())
         assertThat(matrixClientsHolder.getOrNull(A_SESSION_ID)).isNull()
 
         fakeAuthenticationService.givenMatrixClient(FakeMatrixClient(sessionId = A_SESSION_ID))
@@ -94,4 +117,16 @@ class MatrixClientsHolderTest {
         assertThat(loginSucceeded.isSuccess).isTrue()
         assertThat(matrixClientsHolder.getOrNull(A_SESSION_ID)).isNotNull()
     }
+
+    private fun TestScope.createSyncOrchestratorFactory() = object : DefaultSyncOrchestrator.Factory {
+        override fun create(matrixClient: MatrixClient): DefaultSyncOrchestrator {
+            return DefaultSyncOrchestrator(
+                matrixClient,
+                baseCoroutineScope = this@createSyncOrchestratorFactory,
+                appForegroundStateService = FakeAppForegroundStateService(),
+                networkMonitor = FakeNetworkMonitor(),
+                dispatchers = testCoroutineDispatchers(),
+            )
+        }
+    }
 }