feat: add analytics user profile - Part 2 (WPB-8978)

data/src/commonMain/kotlin/com/wire/kalium/logic/data/analytics/AnalyticsIdentifierResult.kt

@@ -0,0 +1,57 @@
+/*
+ * Wire
+ * Copyright (C) 2024 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+
+package com.wire.kalium.logic.data.analytics
+
+sealed interface AnalyticsIdentifierResult {
+
+    /**
+     * To be used when there is no user logged in or analytics settings is disabled.
+     */
+    data object Disabled : AnalyticsIdentifierResult
+
+    /**
+     * Wrapper: To be used when there is a user logged in and analytics settings is enabled.
+     */
+    sealed interface Enabled : AnalyticsIdentifierResult {
+        val identifier: String
+    }
+
+    /**
+     * To be used when user first login to device, generating a new identifier and sending over to other clients.
+     */
+    data class NonExistingIdentifier(
+        override val identifier: String
+    ) : Enabled
+
+    /**
+     * To be used when user already has a tracking identifier, meaning no migration will be done.
+     */
+    data class ExistingIdentifier(
+        override val identifier: String
+    ) : Enabled
+
+    /**
+     * To be used when user is already logged in and receive a new tracking identifier from another client,
+     * it needs to set received tracking identifier as current identifier with migration.
+     * (migrate old identifier events to new identifier)
+     */
+    data class MigrationIdentifier(
+        override val identifier: String
+    ) : Enabled
+}

logger/src/commonMain/kotlin/com/wire/kalium/logger/KaliumLogger.kt

@@ -258,7 +258,8 @@ class KaliumLogger(
         )
 
         enum class ApplicationFlow {
-            SYNC, EVENT_RECEIVER, CONVERSATIONS, CONNECTIONS, MESSAGES, SEARCH, SESSION, REGISTER, CLIENTS, CALLING, ASSETS, LOCAL_STORAGE
+            SYNC, EVENT_RECEIVER, CONVERSATIONS, CONNECTIONS, MESSAGES, SEARCH, SESSION, REGISTER,
+            CLIENTS, CALLING, ASSETS, LOCAL_STORAGE, ANALYTICS
         }
     }
 }

logic/src/commonMain/kotlin/com/wire/kalium/logic/configuration/UserConfigRepository.kt

@@ -136,7 +136,12 @@ interface UserConfigRepository {
     suspend fun clearE2EISettings()
     fun setShouldFetchE2EITrustAnchors(shouldFetch: Boolean)
     fun getShouldFetchE2EITrustAnchor(): Boolean
-    suspend fun setTrackingIdentifier(identifier: String)
+    suspend fun setCurrentTrackingIdentifier(newIdentifier: String)
+    suspend fun getCurrentTrackingIdentifier(): String?
+    suspend fun observeCurrentTrackingIdentifier(): Flow<Either<StorageFailure, String>>
+    suspend fun setPreviousTrackingIdentifier(identifier: String)
+    suspend fun getPreviousTrackingIdentifier(): String?
+    suspend fun deletePreviousTrackingIdentifier()
 }
 
 @Suppress("TooManyFunctions")
@@ -497,9 +502,30 @@ internal class UserConfigDataSource internal constructor(
 
     override fun getShouldFetchE2EITrustAnchor(): Boolean = userConfigStorage.getShouldFetchE2EITrustAnchorHasRun()
 
-    override suspend fun setTrackingIdentifier(identifier: String) {
+    override suspend fun setCurrentTrackingIdentifier(newIdentifier: String) {
         wrapStorageRequest {
-            userConfigDAO.setTrackingIdentifier(identifier = identifier)
+            userConfigDAO.setTrackingIdentifier(identifier = newIdentifier)
+        }
+    }
+
+    override suspend fun getCurrentTrackingIdentifier(): String? =
+        userConfigDAO.getTrackingIdentifier()
+
+    override suspend fun observeCurrentTrackingIdentifier(): Flow<Either<StorageFailure, String>> =
+        userConfigDAO.observeTrackingIdentifier().wrapStorageRequest()
+
+    override suspend fun setPreviousTrackingIdentifier(identifier: String) {
+        wrapStorageRequest {
+            userConfigDAO.setPreviousTrackingIdentifier(identifier = identifier)
+        }
+    }
+
+    override suspend fun getPreviousTrackingIdentifier(): String? =
+        userConfigDAO.getPreviousTrackingIdentifier()
+
+    override suspend fun deletePreviousTrackingIdentifier() {
+        wrapStorageRequest {
+            userConfigDAO.deletePreviousTrackingIdentifier()
         }
     }
 }

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/UserSessionScope.kt

@@ -152,6 +152,8 @@ import com.wire.kalium.logic.di.MapperProvider
 import com.wire.kalium.logic.di.PlatformUserStorageProperties
 import com.wire.kalium.logic.di.RootPathsProvider
 import com.wire.kalium.logic.di.UserStorageProvider
+import com.wire.kalium.logic.feature.analytics.AnalyticsIdentifierManager
+import com.wire.kalium.logic.feature.analytics.ObserveAnalyticsTrackingIdentifierStatusUseCase
 import com.wire.kalium.logic.feature.applock.AppLockTeamFeatureConfigObserver
 import com.wire.kalium.logic.feature.applock.AppLockTeamFeatureConfigObserverImpl
 import com.wire.kalium.logic.feature.applock.MarkTeamAppLockStatusAsNotifiedUseCase
@@ -1309,7 +1311,8 @@ class UserSessionScope internal constructor(
     private val dataTransferEventHandler: DataTransferEventHandler
         get() = DataTransferEventHandlerImpl(
             userId,
-            userConfigRepository
+            userConfigRepository,
+            userScopedLogger
         )
 
     private val applicationMessageHandler: ApplicationMessageHandler
@@ -1475,6 +1478,19 @@ class UserSessionScope internal constructor(
     val observeLegalHoldStateForUser: ObserveLegalHoldStateForUserUseCase
         get() = ObserveLegalHoldStateForUserUseCaseImpl(clientRepository)
 
+    val observeAnalyticsTrackingIdentifierStatus: ObserveAnalyticsTrackingIdentifierStatusUseCase
+        get() = ObserveAnalyticsTrackingIdentifierStatusUseCase(userConfigRepository, userScopedLogger)
+
+    val analyticsIdentifierManager: AnalyticsIdentifierManager
+        get() = AnalyticsIdentifierManager(
+            messages.messageSender,
+            userConfigRepository,
+            userId,
+            clientIdProvider,
+            selfConversationIdProvider,
+            userScopedLogger
+        )
+
     suspend fun observeIfE2EIRequiredDuringLogin(): Flow<Boolean?> = clientRepository.observeIsClientRegistrationBlockedByE2EI()
 
     val observeLegalHoldForSelfUser: ObserveLegalHoldStateForSelfUserUseCase

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/analytics/AnalyticsIdentifierManager.kt

@@ -0,0 +1,104 @@
+/*
+ * Wire
+ * Copyright (C) 2024 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+package com.wire.kalium.logic.feature.analytics
+
+import com.benasher44.uuid.uuid4
+import com.wire.kalium.logger.KaliumLogger
+import com.wire.kalium.logic.cache.SelfConversationIdProvider
+import com.wire.kalium.logic.configuration.UserConfigRepository
+import com.wire.kalium.logic.data.id.CurrentClientIdProvider
+import com.wire.kalium.logic.data.message.Message
+import com.wire.kalium.logic.data.message.MessageContent
+import com.wire.kalium.logic.data.message.MessageTarget
+import com.wire.kalium.logic.data.user.UserId
+import com.wire.kalium.logic.feature.message.MessageSender
+import com.wire.kalium.logic.functional.flatMap
+import com.wire.kalium.logic.functional.foldToEitherWhileRight
+import com.wire.kalium.logic.kaliumLogger
+import kotlinx.datetime.Clock
+
+interface AnalyticsIdentifierManager {
+
+    /**
+     * When doing a migration of tracking identifier (receive new identifier -> migrate new identifier),
+     * we should then after migration is complete, delete the previous tracking identifier.
+     *
+     * Previous tracking identifier is kept because in case migration or network failure, we still have both values
+     * to do the correct migration of tracking identifiers.
+     */
+    suspend fun onMigrationComplete()
+
+    /**
+     * When user first login, we generate a new tracking identifier, when this tracking identifier is set,
+     * we need to send a message to the other clients of the user, to ensure they also use this newly generated identifier.
+     */
+    suspend fun propagateTrackingIdentifier(identifier: String)
+}
+
+@Suppress("FunctionNaming", "LongParameterList")
+internal fun AnalyticsIdentifierManager(
+    messageSender: MessageSender,
+    userConfigRepository: UserConfigRepository,
+    selfUserId: UserId,
+    selfClientIdProvider: CurrentClientIdProvider,
+    selfConversationIdProvider: SelfConversationIdProvider,
+    defaultLogger: KaliumLogger = kaliumLogger
+) = object : AnalyticsIdentifierManager {
+
+    private val TAG = "AnalyticsIdentifierManager"
+    private val logger = defaultLogger.withFeatureId(KaliumLogger.Companion.ApplicationFlow.ANALYTICS)
+
+    override suspend fun onMigrationComplete() {
+        userConfigRepository.deletePreviousTrackingIdentifier()
+
+        logger.i("$TAG Previous Tracking Identifier deleted.")
+    }
+
+    override suspend fun propagateTrackingIdentifier(identifier: String) {
+        val messageContent = MessageContent.DataTransfer(
+            trackingIdentifier = MessageContent.DataTransfer.TrackingIdentifier(
+                identifier = identifier
+            )
+        )
+        selfClientIdProvider().flatMap { currentClientId ->
+            selfConversationIdProvider().flatMap { selfConversationIdList ->
+                selfConversationIdList.foldToEitherWhileRight(Unit) { selfConversationId, _ ->
+                    val date = Clock.System.now()
+                    val message = Message.Signaling(
+                        id = uuid4().toString(),
+                        content = messageContent,
+                        conversationId = selfConversationId,
+                        date = date,
+                        senderUserId = selfUserId,
+                        senderClientId = currentClientId,
+                        status = Message.Status.Sent,
+                        isSelfMessage = true,
+                        expirationData = null
+                    )
+
+                    messageSender.sendMessage(
+                        message = message,
+                        messageTarget = MessageTarget.Conversation()
+                    ).also {
+                        logger.i("$TAG Tracking Identifier propagated.")
+                    }
+                }
+            }
+        }
+    }
+}

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/analytics/ObserveAnalyticsTrackingIdentifierStatusUseCase.kt

@@ -0,0 +1,95 @@
+/*
+ * Wire
+ * Copyright (C) 2024 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+package com.wire.kalium.logic.feature.analytics
+
+import com.benasher44.uuid.uuid4
+import com.wire.kalium.logger.KaliumLogger
+import com.wire.kalium.logic.configuration.UserConfigRepository
+import com.wire.kalium.logic.data.analytics.AnalyticsIdentifierResult
+import com.wire.kalium.logic.functional.Either
+import com.wire.kalium.logic.functional.flatMapRightWithEither
+import com.wire.kalium.logic.functional.isRight
+import com.wire.kalium.logic.kaliumLogger
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
+
+/**
+ * Use case that allows observing if the analytics tracking identifier
+ * changes, due to receiving a new identifier from another client
+ * or when it's user's first interaction with analytics.
+ */
+interface ObserveAnalyticsTrackingIdentifierStatusUseCase {
+    /**
+     * Use case [ObserveAnalyticsTrackingIdentifierStatusUseCase] operation
+     *
+     * @return a [AnalyticsIdentifierResult]
+     */
+    suspend operator fun invoke(): Flow<AnalyticsIdentifierResult>
+}
+
+@Suppress("FunctionNaming")
+internal fun ObserveAnalyticsTrackingIdentifierStatusUseCase(
+    userConfigRepository: UserConfigRepository,
+    defaultLogger: KaliumLogger = kaliumLogger,
+) = object : ObserveAnalyticsTrackingIdentifierStatusUseCase {
+
+    private val TAG = "ObserveAnalyticsTrackingIdentifierStatusUseCase"
+    private val logger = defaultLogger.withFeatureId(KaliumLogger.Companion.ApplicationFlow.ANALYTICS)
+
+    override suspend fun invoke(): Flow<AnalyticsIdentifierResult> =
+        userConfigRepository
+            .observeCurrentTrackingIdentifier()
+            .distinctUntilChanged()
+            .flatMapRightWithEither { currentIdentifier: String ->
+                val result =
+                    userConfigRepository.getPreviousTrackingIdentifier()?.let {
+                        logger.i("$TAG Updating Tracking Identifier with migration value.")
+                        AnalyticsIdentifierResult.MigrationIdentifier(
+                            identifier = currentIdentifier
+                        )
+                    } ?: AnalyticsIdentifierResult.ExistingIdentifier(
+                        identifier = currentIdentifier
+                    ).also {
+                        logger.i("$TAG Updating Tracking Identifier with existing value.")
+                    }
+
+                flowOf(Either.Right(result))
+            }.map {
+                // it's needed, otherwise it will be detected as Flow<Any>
+                if (it.isRight()) it.value as AnalyticsIdentifierResult
+                else {
+                    userConfigRepository.getCurrentTrackingIdentifier()?.let { currentIdentifier: String ->
+                        logger.i("$TAG Updating Tracking Identifier with existing value.")
+                        AnalyticsIdentifierResult.ExistingIdentifier(
+                            identifier = currentIdentifier
+                        )
+                    } ?: uuid4().toString().let { trackingIdentifier: String ->
+                        logger.i("$TAG Generating new Tracking Identifier value.")
+                        userConfigRepository.setCurrentTrackingIdentifier(
+                            newIdentifier = trackingIdentifier
+                        )
+
+                        AnalyticsIdentifierResult.NonExistingIdentifier(
+                            identifier = trackingIdentifier
+                        )
+                    }
+                }
+            }
+}