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

* feat: add method to set/insert tracking identifier into user metadata table

* feat: add MessageContent for DataTransfer with TrackingIdentifier

* feat: add DataTransferEventHandler to handle new data transfer events and set tracking identifier to user config metadata

* chore: remove needless blank line

* feat: add get for current tracking identifier and get and set for previous tracking identifier

* feat: add usage and logic handling for current and previous tracking identifiers

* feat: add tests

* feat: add better handling of receiving already existing tracking id and tests

* feat: add new analytics logger to kalium logger

* feat: add usecase to observe analytics tracking identifier

* feat: add tests for observeAnalyticsTrackingIdentifierStatus

* feat: add none AnalyticsIdentifierResult to be used in AR and remove abstract attribute

* feat: add proper mapping to observer result and verification on Either Left is tracking identifier already exists

* feat: add tests for observer with new logic

* chore: add user scope logger

* chore: add `as` instantiation of right value for usecase result

* feat: move AnalyticsIdentifierResult to new data module and update its usages

* feat: add usecase to delete previous tracking identifier

* chore: add missing imports

* test: add test for DeletePreviousTrackingIdentifierUseCase

* chore: remove delete previous tracking identifier use case

* feat: add extra sealed interface for better handling of analytics identifier result

* feat: add AnalyticsIdentifierManager to handle migration complete and propagating tracking identifier

* chore: add missing extension from new sealed interface

* test: add tests for AnalyticsIdentifierManager

* chore: adjust detekt

* chore: remove unused import

* chore: rename user config current tracking identifier

* chore: add docs

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
+                        )
+                    }
+                }
+            }
+}