refactor(sync): allow concurrent sync start stop [WPB-15262]

app/src/main/kotlin/com/wire/android/WireApplication.kt

@@ -42,11 +42,11 @@ import com.wire.android.util.CurrentScreenManager
 import com.wire.android.util.DataDogLogger
 import com.wire.android.util.LogFileWriter
 import com.wire.android.util.getGitBuildId
-import com.wire.android.util.lifecycle.ConnectionPolicyManager
+import com.wire.android.util.lifecycle.SyncLifecycleManager
 import com.wire.android.workmanager.WireWorkerFactory
+import com.wire.kalium.common.logger.CoreLogger
 import com.wire.kalium.logger.KaliumLogLevel
 import com.wire.kalium.logger.KaliumLogger
-import com.wire.kalium.common.logger.CoreLogger
 import com.wire.kalium.logic.CoreLogic
 import com.wire.kalium.logic.feature.session.CurrentSessionResult
 import dagger.Lazy
@@ -72,7 +72,7 @@ class WireApplication : BaseApp() {
     lateinit var logFileWriter: Lazy<LogFileWriter>
 
     @Inject
-    lateinit var connectionPolicyManager: Lazy<ConnectionPolicyManager>
+    lateinit var syncLifecycleManager: Lazy<SyncLifecycleManager>
 
     @Inject
     lateinit var wireWorkerFactory: Lazy<WireWorkerFactory>
@@ -116,7 +116,9 @@ class WireApplication : BaseApp() {
             withContext(Dispatchers.Main) {
                 ProcessLifecycleOwner.get().lifecycle.addObserver(currentScreenManager)
             }
-            connectionPolicyManager.get().startObservingAppLifecycle()
+            launch {
+                syncLifecycleManager.get().observeAppLifecycle()
+            }
 
             appLogger.i("$TAG api version update")
             // TODO: Can be handled in one of Sync steps

app/src/main/kotlin/com/wire/android/notification/WireNotificationManager.kt

@@ -27,7 +27,7 @@ import com.wire.android.services.ServicesManager
 import com.wire.android.util.CurrentScreen
 import com.wire.android.util.CurrentScreenManager
 import com.wire.android.util.dispatchers.DispatcherProvider
-import com.wire.android.util.lifecycle.ConnectionPolicyManager
+import com.wire.android.util.lifecycle.SyncLifecycleManager
 import com.wire.android.util.logIfEmptyUserName
 import com.wire.kalium.logger.obfuscateId
 import com.wire.kalium.logic.CoreLogic
@@ -66,6 +66,7 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicReference
 import javax.inject.Inject
 import javax.inject.Singleton
+import kotlin.time.Duration.Companion.seconds
 
 @OptIn(ExperimentalCoroutinesApi::class)
 @Suppress("TooManyFunctions", "LongParameterList")
@@ -75,7 +76,7 @@ class WireNotificationManager @Inject constructor(
     private val currentScreenManager: CurrentScreenManager,
     private val messagesNotificationManager: MessageNotificationManager,
     private val callNotificationManager: CallNotificationManager,
-    private val connectionPolicyManager: ConnectionPolicyManager,
+    private val syncLifecycleManager: SyncLifecycleManager,
     private val servicesManager: ServicesManager,
     private val dispatcherProvider: DispatcherProvider,
     private val pingRinger: PingRinger
@@ -165,10 +166,7 @@ class WireNotificationManager @Inject constructor(
         val observeCallsJob = observeCallNotificationsOnceJob(userId)
 
         appLogger.d("$TAG start syncing")
-        connectionPolicyManager.handleConnectionOnPushNotification(
-            userId,
-            STAY_ALIVE_TIME_ON_PUSH_MS
-        )
+        syncLifecycleManager.syncTemporarily(userId, STAY_ALIVE_TIME_ON_PUSH_DURATION)
 
         observeMessagesJob?.cancel("$TAG checked the notifications once, canceling observing.")
         observeCallsJob?.cancel("$TAG checked the calls once, canceling observing.")
@@ -540,6 +538,6 @@ class WireNotificationManager @Inject constructor(
 
     companion object {
         private const val TAG = "WireNotificationManager"
-        private const val STAY_ALIVE_TIME_ON_PUSH_MS = 1000L
+        private val STAY_ALIVE_TIME_ON_PUSH_DURATION = 1.seconds
     }
 }

app/src/main/kotlin/com/wire/android/services/PersistentWebSocketService.kt

@@ -39,12 +39,15 @@ import com.wire.android.notification.WireNotificationManager
 import com.wire.android.notification.openAppPendingIntent
 import com.wire.android.util.dispatchers.DispatcherProvider
 import com.wire.kalium.logic.CoreLogic
-import com.wire.kalium.logic.data.sync.ConnectionPolicy
+import com.wire.kalium.logic.data.user.UserId
 import com.wire.kalium.logic.feature.user.webSocketStatus.ObservePersistentWebSocketConnectionStatusUseCase
 import dagger.hilt.android.AndroidEntryPoint
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.launch
 import javax.inject.Inject
 
@@ -96,23 +99,15 @@ class PersistentWebSocketService : Service() {
                     }
 
                     is ObservePersistentWebSocketConnectionStatusUseCase.Result.Success -> {
-                        result.persistentWebSocketStatusListFlow.collect { statuses ->
-
+                        result.persistentWebSocketStatusListFlow.collectLatest { statuses ->
                             val usersToObserve = statuses
                                 .filter { it.isPersistentWebSocketEnabled }
                                 .map { it.userId }
-
                             notificationManager.observeNotificationsAndCallsPersistently(
                                 usersToObserve,
                                 scope
                             )
-
-                            statuses.map { persistentWebSocketStatus ->
-                                if (persistentWebSocketStatus.isPersistentWebSocketEnabled) {
-                                    coreLogic.getSessionScope(persistentWebSocketStatus.userId)
-                                        .setConnectionPolicy(ConnectionPolicy.KEEP_ALIVE)
-                                }
-                            }
+                            keepRunningSyncForUsers(usersToObserve)
                         }
                     }
                 }
@@ -121,6 +116,18 @@ class PersistentWebSocketService : Service() {
         return START_STICKY
     }
 
+    private suspend fun keepRunningSyncForUsers(usersToObserve: List<UserId>) {
+        coroutineScope {
+            usersToObserve.forEach { userId ->
+                launch {
+                    coreLogic.getSessionScope(userId).syncExecutor.request {
+                        awaitCancellation()
+                    }
+                }
+            }
+        }
+    }
+
     private fun generateForegroundNotification() {
         notificationChannelsManager.createRegularChannel(WEB_SOCKET_CHANNEL_ID, WEB_SOCKET_CHANNEL_NAME)