chore: make prefs operation async

buildSrc/src/main/kotlin/io.customer/android/Dependencies.kt

@@ -41,6 +41,7 @@ object Dependencies {
     const val kotlinBinaryValidator =
         "org.jetbrains.kotlinx:binary-compatibility-validator:${Versions.KOTLIN_BINARY_VALIDATOR}"
     const val kotlinxSerializationJson = "org.jetbrains.kotlinx:kotlinx-serialization-json:${Versions.KOTLINX_SERIALIZATION_JSON}"
+    const val dataStore = "androidx.datastore:datastore-preferences:${Versions.DATASTORE}"
     const val kluentJava = "org.amshove.kluent:kluent:${Versions.KLUENT}"
     const val kotlinGradlePlugin = "org.jetbrains.kotlin:kotlin-gradle-plugin:${Versions.KOTLIN}"
     const val kotlinSerialization = "org.jetbrains.kotlin:kotlin-serialization:${Versions.KOTLIN}"

buildSrc/src/main/kotlin/io.customer/android/Versions.kt

@@ -30,6 +30,7 @@ object Versions {
     internal const val KLUENT = "1.72"
     internal const val KOTLIN_BINARY_VALIDATOR = "0.14.0"
     internal const val KOTLINX_SERIALIZATION_JSON = "1.5.1"
+    internal const val DATASTORE = "1.1.7"
     internal const val KOTLIN = "1.8.21"
     internal const val MATERIAL_COMPONENTS = "1.4.0"
     internal const val MOCKITO_KOTLIN = "4.0.0"

common-test/src/main/java/io/customer/commontest/extensions/MockKExtensions.kt

@@ -2,6 +2,7 @@ package io.customer.commontest.extensions
 
 import io.mockk.Called
 import io.mockk.MockKVerificationScope
+import io.mockk.coVerify
 import io.mockk.verify
 
 /**
@@ -18,6 +19,20 @@ fun assertCalledNever(
     verifyBlock: MockKVerificationScope.() -> Unit
 ) = verify(exactly = 0, verifyBlock = verifyBlock)
 
+/**
+ * Extension for MockK coVerify function to verify a suspend block of code is called exactly once.
+ */
+fun assertCoCalledOnce(
+    verifyBlock: suspend MockKVerificationScope.() -> Unit
+) = coVerify(exactly = 1, verifyBlock = verifyBlock)
+
+/**
+ * Extension for MockK coVerify function to verify a suspend block of code was never called.
+ */
+fun assertCoCalledNever(
+    verifyBlock: suspend MockKVerificationScope.() -> Unit
+) = coVerify(exactly = 0, verifyBlock = verifyBlock)
+
 /**
  * Extension for MockK verify function to verify that no interactions were made with provided mocks.
  */

common-test/src/main/java/io/customer/commontest/util/DeviceTokenManagerStub.kt

@@ -0,0 +1,44 @@
+package io.customer.commontest.util
+
+import io.customer.sdk.data.store.DeviceTokenManager
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asStateFlow
+
+class DeviceTokenManagerStub : DeviceTokenManager {
+    private val _deviceTokenFlow = MutableStateFlow<String?>(null)
+
+    override val deviceToken: String?
+        get() = _deviceTokenFlow.value
+
+    override val deviceTokenFlow: Flow<String?>
+        get() = _deviceTokenFlow.asStateFlow()
+
+    override fun setDeviceToken(token: String?) {
+        _deviceTokenFlow.value = token
+    }
+
+    override fun clearDeviceToken() {
+        _deviceTokenFlow.value = null
+    }
+
+    override fun replaceToken(newToken: String?, onOldTokenDelete: (String) -> Unit) {
+        val currentToken = _deviceTokenFlow.value
+
+        // If there's an existing token and it's different from the new one, notify about deletion
+        if (currentToken != null && currentToken != newToken) {
+            onOldTokenDelete(currentToken)
+        }
+
+        // Set the new token
+        _deviceTokenFlow.value = newToken
+    }
+
+    override fun cleanup() {
+        // No-op for stub
+    }
+
+    fun reset() {
+        _deviceTokenFlow.value = null
+    }
+}

common-test/src/main/java/io/customer/commontest/util/ScopeProviderStub.kt

@@ -9,7 +9,8 @@ import kotlinx.coroutines.test.UnconfinedTestDispatcher
 class ScopeProviderStub private constructor(
     override val eventBusScope: TestScope,
     override val lifecycleListenerScope: TestScope,
-    override val inAppLifecycleScope: TestScope
+    override val inAppLifecycleScope: TestScope,
+    override val backgroundScope: TestScope
 ) : ScopeProvider {
 
     @Suppress("FunctionName")
@@ -18,13 +19,15 @@ class ScopeProviderStub private constructor(
         fun Unconfined(): ScopeProviderStub = ScopeProviderStub(
             eventBusScope = TestScope(UnconfinedTestDispatcher()),
             lifecycleListenerScope = TestScope(UnconfinedTestDispatcher()),
-            inAppLifecycleScope = TestScope(UnconfinedTestDispatcher())
+            inAppLifecycleScope = TestScope(UnconfinedTestDispatcher()),
+            backgroundScope = TestScope(UnconfinedTestDispatcher())
         )
 
         fun Standard(): ScopeProviderStub = ScopeProviderStub(
             eventBusScope = TestScope(StandardTestDispatcher()),
             lifecycleListenerScope = TestScope(StandardTestDispatcher()),
-            inAppLifecycleScope = TestScope(StandardTestDispatcher())
+            inAppLifecycleScope = TestScope(StandardTestDispatcher()),
+            backgroundScope = TestScope(StandardTestDispatcher())
         )
     }
 }

core/api/core.api

@@ -117,6 +117,7 @@ public abstract class io/customer/sdk/core/di/AndroidSDKComponent : io/customer/
 	public abstract fun getBuildStore ()Lio/customer/sdk/data/store/BuildStore;
 	public abstract fun getClient ()Lio/customer/sdk/data/store/Client;
 	public abstract fun getDeviceStore ()Lio/customer/sdk/data/store/DeviceStore;
+	public abstract fun getDeviceTokenManager ()Lio/customer/sdk/data/store/DeviceTokenManager;
 	public abstract fun getGlobalPreferenceStore ()Lio/customer/sdk/data/store/GlobalPreferenceStore;
 }
 
@@ -128,6 +129,7 @@ public final class io/customer/sdk/core/di/AndroidSDKComponentImpl : io/customer
 	public fun getBuildStore ()Lio/customer/sdk/data/store/BuildStore;
 	public fun getClient ()Lio/customer/sdk/data/store/Client;
 	public fun getDeviceStore ()Lio/customer/sdk/data/store/DeviceStore;
+	public fun getDeviceTokenManager ()Lio/customer/sdk/data/store/DeviceTokenManager;
 	public fun getGlobalPreferenceStore ()Lio/customer/sdk/data/store/GlobalPreferenceStore;
 	public fun reset ()V
 }
@@ -220,6 +222,7 @@ public final class io/customer/sdk/core/util/Logger$DefaultImpls {
 }
 
 public abstract interface class io/customer/sdk/core/util/ScopeProvider {
+	public abstract fun getBackgroundScope ()Lkotlinx/coroutines/CoroutineScope;
 	public abstract fun getEventBusScope ()Lkotlinx/coroutines/CoroutineScope;
 	public abstract fun getInAppLifecycleScope ()Lkotlinx/coroutines/CoroutineScope;
 	public abstract fun getLifecycleListenerScope ()Lkotlinx/coroutines/CoroutineScope;
@@ -234,6 +237,7 @@ public final class io/customer/sdk/core/util/SdkDispatchers : io/customer/sdk/co
 
 public final class io/customer/sdk/core/util/SdkScopeProvider : io/customer/sdk/core/util/ScopeProvider {
 	public fun <init> (Lio/customer/sdk/core/util/DispatchersProvider;)V
+	public fun getBackgroundScope ()Lkotlinx/coroutines/CoroutineScope;
 	public fun getEventBusScope ()Lkotlinx/coroutines/CoroutineScope;
 	public fun getInAppLifecycleScope ()Lkotlinx/coroutines/CoroutineScope;
 	public fun getLifecycleListenerScope ()Lkotlinx/coroutines/CoroutineScope;

core/build.gradle

@@ -39,6 +39,7 @@ dependencies {
     api project(":base")
     api Dependencies.androidxCoreKtx
     implementation Dependencies.coroutinesAndroid
+    implementation Dependencies.dataStore
     // Use this as API so customers can provide objects serializations without
     // needing to add it as a dependency to their app
     api(Dependencies.kotlinxSerializationJson)

core/src/main/kotlin/io/customer/sdk/core/di/AndroidSDKComponent.kt

@@ -10,6 +10,8 @@
 import io.customer.sdk.data.store.Client
 import io.customer.sdk.data.store.DeviceStore
 import io.customer.sdk.data.store.DeviceStoreImpl
+import io.customer.sdk.data.store.DeviceTokenManager
+import io.customer.sdk.data.store.DeviceTokenManagerImpl
 import io.customer.sdk.data.store.GlobalPreferenceStore
 import io.customer.sdk.data.store.GlobalPreferenceStoreImpl
 
@@ -21,6 +23,7 @@
     abstract val applicationStore: ApplicationStore
     abstract val deviceStore: DeviceStore
     abstract val globalPreferenceStore: GlobalPreferenceStore
+    abstract val deviceTokenManager: DeviceTokenManager
 }
 
 /**
@@ -57,8 +60,16 @@
         }
     override val globalPreferenceStore: GlobalPreferenceStore
         get() = singleton<GlobalPreferenceStore> { GlobalPreferenceStoreImpl(applicationContext) }
+    override val deviceTokenManager: DeviceTokenManager
+        get() = singleton<DeviceTokenManager> {
+            DeviceTokenManagerImpl(
+                globalPreferenceStore = globalPreferenceStore
+            )
+        }
 
     override fun reset() {
+        deviceTokenManager.cleanup()
+
         super.reset()
 
         SDKComponent.activityLifecycleCallbacks.unregister(application)

core/src/main/kotlin/io/customer/sdk/core/extensions/DataStoreExtensions.kt

@@ -0,0 +1,36 @@
+package io.customer.sdk.core.extensions
+
+import androidx.datastore.core.DataStore
+import androidx.datastore.preferences.core.Preferences
+import androidx.datastore.preferences.core.emptyPreferences
+import io.customer.sdk.core.di.SDKComponent
+import io.customer.sdk.core.util.Logger
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.map
+
+/**
+ * Extension function to safely read from DataStore with consistent error handling.
+ *
+ * As an SDK, we must never crash the host app. All exceptions are caught and logged,
+ * with graceful fallback to empty preferences to ensure app stability.
+ */
+internal fun DataStore<Preferences>.safeData(): Flow<Preferences> {
+    val logger: Logger = SDKComponent.logger
+
+    return data.catch { exception ->
+        logger.error("DataStore error: ${exception.message}")
+        emit(emptyPreferences())
+    }
+}
+
+/**
+ * Safely reads a string value from DataStore with error handling.
+ * Returns null if the key doesn't exist or if there are any errors.
+ *
+ * As an SDK, we prioritize app stability over data completeness.
+ */
+internal suspend fun DataStore<Preferences>.safeGetString(key: Preferences.Key<String>): String? {
+    return safeData().map { it[key] }.first()
+}

core/src/main/kotlin/io/customer/sdk/core/util/ScopeProvider.kt

@@ -7,6 +7,7 @@ interface ScopeProvider {
     val eventBusScope: CoroutineScope
     val lifecycleListenerScope: CoroutineScope
     val inAppLifecycleScope: CoroutineScope
+    val backgroundScope: CoroutineScope
 }
 
 class SdkScopeProvider(private val dispatchers: DispatchersProvider) : ScopeProvider {
@@ -16,4 +17,6 @@ class SdkScopeProvider(private val dispatchers: DispatchersProvider) : ScopeProv
         get() = CoroutineScope(dispatchers.default + SupervisorJob())
     override val inAppLifecycleScope: CoroutineScope
         get() = CoroutineScope(dispatchers.default + SupervisorJob())
+    override val backgroundScope: CoroutineScope
+        get() = CoroutineScope(dispatchers.background + SupervisorJob())
 }

core/src/main/kotlin/io/customer/sdk/data/store/DeviceTokenManager.kt

@@ -0,0 +1,141 @@
+package io.customer.sdk.data.store
+
+import io.customer.sdk.core.di.SDKComponent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.launch
+
+/**
+ * Manages device token with automatic synchronization between memory and persistent storage.
+ *
+ * This class provides:
+ * - Immediate synchronous access to the current device token
+ * - Automatic background persistence to storage
+ * - Single source of truth for device token state
+ * - Reactive updates via Flow
+ * - Automatic initialization from existing storage
+ */
+interface DeviceTokenManager {
+    /**
+     * Current device token value. Always returns immediately from memory.
+     */
+    val deviceToken: String?
+
+    /**
+     * Flow of device token changes for reactive programming.
+     */
+    val deviceTokenFlow: Flow<String?>
+
+    /**
+     * Updates the device token. Immediately updates memory and async saves to storage.
+     */
+    fun setDeviceToken(token: String?)
+
+    /**
+     * Clears the device token. Immediately clears memory and async removes from storage.
+     */
+    fun clearDeviceToken()
+
+    /**
+     * Replaces the current token with a new one, handling the transition properly.
+     * If there was an existing token different from the new one, calls onOldTokenDelete.
+     *
+     * @param newToken The new device token to set
+     * @param onOldTokenDelete Callback invoked with the old token if a replacement occurred
+     */
+    fun replaceToken(newToken: String?, onOldTokenDelete: (String) -> Unit)
+
+    /**
+     * Cancels any pending operations and cleans up resources.
+     * Should be called when the manager is no longer needed.
+     */
+    fun cleanup()
+}
+
+internal class DeviceTokenManagerImpl(
+    private val globalPreferenceStore: GlobalPreferenceStore
+) : DeviceTokenManager {
+
+    private val _deviceTokenFlow = MutableStateFlow<String?>(null)
+    private val backgroundScope: CoroutineScope = SDKComponent.scopeProvider.backgroundScope
+    private val logger = SDKComponent.logger
+
+    // Single-threaded dispatcher for all token operations to eliminate race conditions
+    @OptIn(ExperimentalCoroutinesApi::class)
+    private val tokenOperationScope = CoroutineScope(
+        backgroundScope.coroutineContext +
+            Dispatchers.Default.limitedParallelism(1) +
+            SupervisorJob()
+    )
+
+    init {
+        tokenOperationScope.launch {
+            try {
+                val existingToken = globalPreferenceStore.getDeviceToken()
+                _deviceTokenFlow.value = existingToken
+            } catch (e: Exception) {
+                logger.error("Error during DeviceTokenManager initialization: ${e.message}")
+                _deviceTokenFlow.value = null
+            }
+        }
+    }
+
+    override val deviceToken: String?
+        get() = _deviceTokenFlow.value
+
+    override val deviceTokenFlow: Flow<String?>
+        get() = _deviceTokenFlow.asStateFlow()
+
+    override fun setDeviceToken(token: String?) {
+        tokenOperationScope.launch {
+            _deviceTokenFlow.value = token
+
+            try {
+                if (token != null) {
+                    globalPreferenceStore.saveDeviceToken(token)
+                } else {
+                    globalPreferenceStore.removeDeviceToken()
+                }
+            } catch (e: Exception) {
+                logger.error("Error saving device token: ${e.message}")
+            }
+        }
+    }
+
+    override fun clearDeviceToken() {
+        setDeviceToken(null)
+    }
+
+    override fun replaceToken(newToken: String?, onOldTokenDelete: (String) -> Unit) {
+        tokenOperationScope.launch {
+            val currentToken = _deviceTokenFlow.value
+
+            // If there's an existing token and it's different from the new one, notify about deletion
+            if (currentToken != null && currentToken != newToken) {
+                onOldTokenDelete(currentToken)
+            }
+
+            _deviceTokenFlow.value = newToken
+
+            try {
+                if (newToken != null) {
+                    globalPreferenceStore.saveDeviceToken(newToken)
+                } else {
+                    globalPreferenceStore.removeDeviceToken()
+                }
+            } catch (e: Exception) {
+                logger.error("Error saving replaced device token: ${e.message}")
+            }
+        }
+    }
+
+    override fun cleanup() {
+        tokenOperationScope.cancel()
+    }
+}