feat(session): gate remaining session observers PR 7

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

@@ -23,6 +23,8 @@ import com.wire.android.di.KaliumCoreLogic
 import com.wire.android.notification.NotificationChannelsManager
 import com.wire.android.notification.WireNotificationManager
 import com.wire.android.services.SendPendingMessagesAfterForegroundSyncUseCase
+import com.wire.android.session.AppUserSessionPreparationResult
+import com.wire.android.session.UserSessionPreparationGate
 import com.wire.android.util.CurrentScreenManager
 import com.wire.android.util.dispatchers.DispatcherProvider
 import com.wire.kalium.logic.CoreLogic
@@ -69,14 +71,17 @@ class GlobalObserversManager @Inject constructor(
 ) {
     // TODO(tests): refactor so scope/dispatcher can be injected and properly stopped
     private val scope = CoroutineScope(SupervisorJob() + dispatcherProvider.io())
+    private val userSessionPreparationGate by lazy { UserSessionPreparationGate(coreLogic) }
 
     fun observe() {
         scope.launch { setUpNotifications() }
         scope.launch {
             coreLogic.getGlobalScope().observeValidAccounts().distinctUntilChanged().collectLatest {
                 coroutineScope {
                     it.forEach {
-                        launch { coreLogic.getSessionScope(it.first.id).calls.endCallOnConversationChange() }
+                        launch {
+                            preparedSessionScope(it.first.id)?.calls?.endCallOnConversationChange()
+                        }
                     }
                 }
             }
@@ -165,7 +170,7 @@ class GlobalObserversManager @Inject constructor(
                         emptyFlow()
                     }
                 }
-                .collect { userId -> coreLogic.getSessionScope(userId).messages.deleteEphemeralMessageEndDate() }
+                .collect { userId -> preparedSessionScope(userId)?.messages?.deleteEphemeralMessageEndDate() }
         }
     }
 
@@ -200,4 +205,13 @@ class GlobalObserversManager @Inject constructor(
     private companion object {
         private const val TAG = "GlobalObserversManager"
     }
+
+    private suspend fun preparedSessionScope(userId: UserId) =
+        when (val result = userSessionPreparationGate.prepare(userId)) {
+            is AppUserSessionPreparationResult.Ready -> result.sessionScope
+            is AppUserSessionPreparationResult.Failed -> {
+                appLogger.w("$TAG skipping database observer for ${userId.toLogString()}: ${result.reason}")
+                null
+            }
+        }
 }

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

@@ -42,6 +42,8 @@ import com.wire.android.feature.analytics.AnonymousAnalyticsRecorderImpl
 import com.wire.android.feature.analytics.globalAnalyticsManager
 import com.wire.android.feature.analytics.model.AnalyticsEvent
 import com.wire.android.feature.analytics.model.AnalyticsSettings
+import com.wire.android.session.AppUserSessionPreparationResult
+import com.wire.android.session.UserSessionPreparationGate
 import com.wire.android.util.AppNameUtil
 import com.wire.android.util.CurrentScreenManager
 import com.wire.android.util.DataDogLogger
@@ -54,21 +56,26 @@ import com.wire.kalium.common.logger.CoreLogger
 import com.wire.kalium.logger.KaliumLogLevel
 import com.wire.kalium.logger.KaliumLogger
 import com.wire.kalium.logic.CoreLogic
+import com.wire.kalium.logic.data.user.UserId
+import com.wire.kalium.logic.feature.UserSessionScope
 import com.wire.kalium.logic.feature.session.CurrentSessionResult
 import com.wire.kalium.logic.feature.session.GetAllSessionsResult
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
 import dev.zacsweers.metro.Inject
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.collections.filter
 
@@ -113,6 +120,8 @@ class WireApplication : BaseApp() {
     @Inject
     lateinit var analyticsManager: Lazy<AnonymousAnalyticsManager>
 
+    private val userSessionPreparationGate by lazy { UserSessionPreparationGate(coreLogic.value) }
+
     @Inject
     lateinit var workManager: WorkManager
 
@@ -192,7 +201,7 @@ class WireApplication : BaseApp() {
             ::Pair
         ).collect { (isAppVisible, validSessions) ->
             validSessions.forEach {
-                coreLogic.value.getSessionScope(it.userId).calls.setBackground(!isAppVisible)
+                preparedSessionScope(it.userId)?.calls?.setBackground(!isAppVisible)
             }
         }
     }
@@ -201,7 +210,11 @@ class WireApplication : BaseApp() {
         coreLogic.value.getGlobalScope().session.currentSessionFlow().filterIsInstance(CurrentSessionResult.Success::class)
             .filter { session -> session.accountInfo.isValid() }
             .flatMapLatest { session ->
-                coreLogic.value.getSessionScope(session.accountInfo.userId).calls.observeRecentlyEndedCallMetadata()
+                flow {
+                    preparedSessionScope(session.accountInfo.userId)?.let { sessionScope ->
+                        emitAll(sessionScope.calls.observeRecentlyEndedCallMetadata())
+                    }
+                }
             }
             .collect { metadata ->
                 analyticsManager.value.sendEvent(AnalyticsEvent.RecentlyEndedCallEvent(metadata))
@@ -213,7 +226,11 @@ class WireApplication : BaseApp() {
             .filterIsInstance<CurrentSessionResult.Success>()
             .map { it.accountInfo.userId }
             .flatMapLatest {
-                coreLogic.value.getSessionScope(it).messages.observeAssetUploadState()
+                flow {
+                    preparedSessionScope(it)?.let { sessionScope ->
+                        emitAll(sessionScope.messages.observeAssetUploadState())
+                    }
+                }
             }
             .collect { uploadInProgress ->
                 if (uploadInProgress) {
@@ -360,6 +377,7 @@ class WireApplication : BaseApp() {
         initializeAnonymousAnalytics()
     }
 
+    @Suppress("LongMethod")
     private fun initializeAnonymousAnalytics() {
         if (!BuildConfig.ANALYTICS_ENABLED) return
 
@@ -370,22 +388,29 @@ class WireApplication : BaseApp() {
             enableDebugLogging = BuildConfig.DEBUG
         )
 
+        val analyticsSessionScopes = ConcurrentHashMap<UserId, UserSessionScope>()
         val analyticsResultFlow = ObserveCurrentSessionAnalyticsUseCase(
             currentSessionFlow = coreLogic.value.getGlobalScope().session.currentSessionFlow(),
             getAnalyticsContactsData = { userId ->
-                coreLogic.value.getSessionScope(userId).getAnalyticsContactsData()
+                checkNotNull(analyticsSessionScopes[userId]).getAnalyticsContactsData()
             },
             observeAnalyticsTrackingIdentifierStatusFlow = { userId ->
-                coreLogic.value.getSessionScope(userId).observeAnalyticsTrackingIdentifierStatus()
+                checkNotNull(analyticsSessionScopes[userId]).observeAnalyticsTrackingIdentifierStatus()
             },
             analyticsIdentifierManagerProvider = { userId ->
-                coreLogic.value.getSessionScope(userId).analyticsIdentifierManager
+                checkNotNull(analyticsSessionScopes[userId]).analyticsIdentifierManager
             },
             userDataStoreProvider = userDataStoreProvider.value,
             globalDataStore = globalDataStore.value,
             currentBackend = { userId ->
-                coreLogic.value.getSessionScope(userId).users.serverLinks()
-            }
+                checkNotNull(analyticsSessionScopes[userId]).users.serverLinks()
+            },
+            prepareSession = { userId ->
+                preparedSessionScope(userId)?.let { sessionScope ->
+                    analyticsSessionScopes[userId] = sessionScope
+                    true
+                } ?: false
+            },
         ).invoke()
 
         AnonymousAnalyticsManagerImpl.init(
@@ -411,7 +436,7 @@ class WireApplication : BaseApp() {
                 .collect {
                     val currentSessionResult = coreLogic.value.getGlobalScope().session.currentSessionFlow().first()
                     val isTeamMember = if (currentSessionResult is CurrentSessionResult.Success) {
-                        coreLogic.value.getSessionScope(currentSessionResult.accountInfo.userId).team.isSelfATeamMember()
+                        preparedSessionScope(currentSessionResult.accountInfo.userId)?.team?.isSelfATeamMember()
                     } else {
                         null
                     }
@@ -440,6 +465,15 @@ class WireApplication : BaseApp() {
         Log.i(TAG, "startup:$event$elapsed")
     }
 
+    private suspend fun preparedSessionScope(userId: UserId) =
+        when (val result = userSessionPreparationGate.prepare(userId)) {
+            is AppUserSessionPreparationResult.Ready -> result.sessionScope
+            is AppUserSessionPreparationResult.Failed -> {
+                appLogger.w("Skipping application database observer for ${userId.toLogString()}: ${result.reason}")
+                null
+            }
+        }
+
     override fun onTrimMemory(level: Int) {
         super.onTrimMemory(level)
         appLogger.w(

app/src/main/kotlin/com/wire/android/analytics/ObserveCurrentSessionAnalyticsUseCase.kt

@@ -52,10 +52,11 @@ fun ObserveCurrentSessionAnalyticsUseCase(
     currentSessionFlow: Flow<CurrentSessionResult>,
     getAnalyticsContactsData: suspend (UserId) -> AnalyticsContactsData,
     observeAnalyticsTrackingIdentifierStatusFlow: suspend (UserId) -> Flow<AnalyticsIdentifierResult>,
-    analyticsIdentifierManagerProvider: (UserId) -> AnalyticsIdentifierManager,
+    analyticsIdentifierManagerProvider: suspend (UserId) -> AnalyticsIdentifierManager,
     userDataStoreProvider: UserDataStoreProvider,
     globalDataStore: GlobalDataStore,
-    currentBackend: suspend (UserId) -> SelfServerConfigUseCase.Result
+    currentBackend: suspend (UserId) -> SelfServerConfigUseCase.Result,
+    prepareSession: suspend (UserId) -> Boolean = { true },
 ) = object : ObserveCurrentSessionAnalyticsUseCase {
 
     private var previousAnalyticsResult: AnalyticsIdentifierResult? = null
@@ -89,6 +90,9 @@ fun ObserveCurrentSessionAnalyticsUseCase(
 
             if (currentSession is CurrentSessionResult.Success && currentSession.accountInfo.isValid()) {
                 val userId = currentSession.accountInfo.userId
+                if (!prepareSession(userId)) {
+                    return@flatMapLatest flowOf(disabledAnalyticsResult())
+                }
                 val analyticsIdentifierManager = analyticsIdentifierManagerProvider(userId)
                 combine(
                     observeAnalyticsTrackingIdentifierStatusFlow(userId)
@@ -131,22 +135,22 @@ fun ObserveCurrentSessionAnalyticsUseCase(
                     )
                 }
             } else {
-                flowOf(
-                    AnalyticsResult<AnalyticsIdentifierManager>(
-                        identifierResult = AnalyticsIdentifierResult.Disabled,
-                        profileProperties = {
-                            AnalyticsProfileProperties(
-                                isTeamMember = false,
-                                teamId = null,
-                                contactsAmount = null,
-                                teamMembersAmount = null,
-                                isEnterprise = null
-                            )
-                        },
-                        manager = null
-                    )
-                )
+                flowOf(disabledAnalyticsResult())
             }
         }.distinctUntilChanged()
     }
 }
+
+private fun disabledAnalyticsResult() = AnalyticsResult<AnalyticsIdentifierManager>(
+    identifierResult = AnalyticsIdentifierResult.Disabled,
+    profileProperties = {
+        AnalyticsProfileProperties(
+            isTeamMember = false,
+            teamId = null,
+            contactsAmount = null,
+            teamMembersAmount = null,
+            isEnterprise = null,
+        )
+    },
+    manager = null,
+)

app/src/main/kotlin/com/wire/android/media/audiomessage/ConversationAudioMessagePlayer.kt

@@ -24,10 +24,13 @@ import android.net.Uri
 import com.wire.android.di.ApplicationScope
 import com.wire.android.di.KaliumCoreLogic
 import com.wire.android.services.ServicesManager
+import com.wire.android.session.AppUserSessionPreparationResult
+import com.wire.android.session.UserSessionPreparationGate
 import com.wire.android.ui.common.R as commonR
 import com.wire.android.util.dispatchers.DispatcherProvider
 import com.wire.android.util.extension.intervalFlow
 import com.wire.android.util.ui.UIText
+import com.wire.kalium.common.error.CoreFailure
 import com.wire.kalium.logic.CoreLogic
 import com.wire.kalium.logic.data.id.ConversationId
 import com.wire.kalium.logic.data.user.UserId
@@ -73,6 +76,8 @@ class ConversationAudioMessagePlayer
     @ApplicationScope private val scope: CoroutineScope,
     private val dispatchers: DispatcherProvider,
 ) {
+    private val userSessionPreparationGate by lazy { UserSessionPreparationGate(coreLogic) }
+
     private companion object {
         const val UPDATE_POSITION_INTERVAL_IN_MS = 1000L
     }
@@ -398,13 +403,21 @@ class ConversationAudioMessagePlayer
         conversationId: ConversationId,
         messageId: String,
     ): MessageAssetResult = withContext(dispatchers.io()) {
+        val preparation = userSessionPreparationGate.prepare(userId)
+        val sessionScope = when (preparation) {
+            is AppUserSessionPreparationResult.Ready -> preparation.sessionScope
+            is AppUserSessionPreparationResult.Failed -> return@withContext MessageAssetResult.Failure(
+                CoreFailure.Unknown(IllegalStateException("User session preparation failed: ${preparation.reason}")),
+                preparation.canRetry,
+            )
+        }
         val key = GetAssetMessageKey(userId, conversationId, messageId)
         getAssetMessageMutex.withLock {
             // keep deferred in the map to prevent multiple calls to the same asset at the same time, instead just reuse the existing one
             val deferredResult = getAssetMessageDeferredMap[key]
             // if no deferred exists or the existing one is already completed with failure, create a new one
             if (deferredResult == null || (deferredResult.isCompleted && deferredResult.getCompleted() is MessageAssetResult.Failure)) {
-                coreLogic.getSessionScope(userId).messages.getAssetMessage(conversationId, messageId).also {
+                sessionScope.messages.getAssetMessage(conversationId, messageId).also {
                     getAssetMessageDeferredMap[key] = it
                 }
             } else {
@@ -415,7 +428,7 @@ class ConversationAudioMessagePlayer
             // this is to handle the case when the file has been uploaded and the file name has changed from temporary to proper one
             if (result is MessageAssetResult.Success && !result.decodedAssetPath.toFile().exists()) {
                 getAssetMessageMutex.withLock {
-                    coreLogic.getSessionScope(userId).messages.getAssetMessage(conversationId, messageId).also {
+                    sessionScope.messages.getAssetMessage(conversationId, messageId).also {
                         getAssetMessageDeferredMap[key] = it
                     }
                 }.await()
@@ -463,30 +476,37 @@ class ConversationAudioMessagePlayer
         _audioSpeed.emit(currentSpeed)
     }
 
+    @Suppress("NestedBlockDepth")
     private suspend fun tryToPlayNextAudio(currentMessageIdWrapper: MessageIdWrapper): Boolean {
         val (conversationId, currentMessageId) = currentMessageIdWrapper
 
         val currentAccountResult = coreLogic.getGlobalScope().session.currentSession()
         if (currentAccountResult is CurrentSessionResult.Success) {
-            coreLogic
-                .getSessionScope((currentAccountResult).accountInfo.userId)
-                .messages
-                .getNextAudioMessageInConversation(conversationId, currentMessageId).let { nextAudio ->
-                    if (nextAudio is GetNextAudioMessageInConversationUseCase.Result.Success) {
-                        playAudio(conversationId, nextAudio.messageId)
-                        return true
+            val preparation = userSessionPreparationGate.prepare(currentAccountResult.accountInfo.userId)
+            if (preparation is AppUserSessionPreparationResult.Ready) {
+                preparation.sessionScope
+                    .messages
+                    .getNextAudioMessageInConversation(conversationId, currentMessageId).let { nextAudio ->
+                        if (nextAudio is GetNextAudioMessageInConversationUseCase.Result.Success) {
+                            playAudio(conversationId, nextAudio.messageId)
+                            return true
+                        }
                     }
-                }
+            }
         }
         return false
     }
 
+    @Suppress("ReturnCount")
     private suspend fun getSenderNameByMessageId(conversationId: ConversationId, messageId: String): String? {
         val currentAccountResult = coreLogic.getGlobalScope().session.currentSession()
         if (currentAccountResult is CurrentSessionResult.Failure) return null
 
-        val senderNameResult = coreLogic
-            .getSessionScope((currentAccountResult as CurrentSessionResult.Success).accountInfo.userId)
+        val preparation = userSessionPreparationGate.prepare(
+            (currentAccountResult as CurrentSessionResult.Success).accountInfo.userId
+        )
+        if (preparation !is AppUserSessionPreparationResult.Ready) return null
+        val senderNameResult = preparation.sessionScope
             .messages
             .getSenderNameByMessageId(conversationId, messageId)