fix(live-notifications): recover the opt-in set in a cold process

core/src/main/kotlin/io/customer/sdk/core/pipeline/DataPipeline.kt

@@ -12,6 +12,11 @@ import io.customer.base.internal.InternalCustomerIOApi
  */
 @InternalCustomerIOApi
 interface DataPipeline {
+    /**
+     * Whether a user is currently identified. Updated synchronously on the caller's thread
+     * during identify()/clearIdentify(), so it is accurate the instant those calls return —
+     * a consumer may gate on it immediately after identify() without racing async propagation.
+     */
     val isUserIdentified: Boolean
     fun track(name: String, properties: Map<String, Any?>)
 }

datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt

@@ -118,6 +118,16 @@ class CustomerIO private constructor(
     // Reset on clearIdentify so a subsequent identify of the same userId is not deduped.
     private var lastIdentifiedUserIdThisSession: String? = null
 
+    // Caller-thread mirror of whether a user is currently identified, backing the synchronous
+    // [isUserIdentified]. analytics.userId() is updated asynchronously and lags a synchronous
+    // identify(), so a consumer that gates on it right after login (e.g. a live-notification
+    // start() on the next line) would otherwise read stale state and drop the event. Set on the
+    // caller's thread during identify()/clearIdentify() so it is accurate the instant they return.
+    // null = no identify/reset has happened this process yet, so fall back to the (possibly
+    // restored-from-persistence) analytics userId; true = identified; false = anonymous.
+    @Volatile
+    private var syncUserIdentified: Boolean? = null
+
     init {
         // Set analytics logger and debug logs based on SDK logger configuration
         Analytics.debugLogsEnabled = logger.logLevel == CioLogLevel.DEBUG
@@ -296,6 +306,12 @@ class CustomerIO private constructor(
             traits = traits,
             serializationStrategy = serializationStrategy
         )
+        // Reflect identity synchronously on the caller's thread so isUserIdentified is correct
+        // immediately, before analytics.userId() catches up (see syncUserIdentified). Must be set
+        // before publishUserChanged: the mirror takes precedence over the analytics fallback, so a
+        // subscriber that gates on isUserIdentified would otherwise read a stale `false` left by an
+        // earlier clearIdentify() even though analytics.userId() is already correct.
+        syncUserIdentified = true
         // Publish for other modules. Must come after analytics.identify() so analytics.userId()
         // returns the new userId for subscribers that gate on it (e.g. location resync).
         publishUserChanged(userId)
@@ -344,6 +360,9 @@ class CustomerIO private constructor(
 
         // Reset the dedup marker so a subsequent identify of the same userId is not deduped.
         lastIdentifiedUserIdThisSession = null
+        // Reflect logout synchronously so isUserIdentified reads false immediately, before
+        // analytics.reset() propagates (see syncUserIdentified).
+        syncUserIdentified = false
 
         logger.debug("deleting device token to remove device from user profile")
 
@@ -377,7 +396,10 @@ class CustomerIO private constructor(
         get() = analytics.userId()
 
     override val isUserIdentified: Boolean
-        get() = !analytics.userId().isNullOrEmpty()
+        // Prefer the caller-thread mirror so this is correct synchronously right after
+        // identify()/clearIdentify(); fall back to analytics for a session restored from
+        // persistence, where neither was called this process. See [syncUserIdentified].
+        get() = syncUserIdentified ?: !analytics.userId().isNullOrEmpty()
 
     @Deprecated("Use setDeviceAttributes() function instead")
     @set:JvmName("setDeviceAttributesDeprecated")

datapipelines/src/test/java/io/customer/datapipelines/IsUserIdentifiedSyncTests.kt

@@ -0,0 +1,47 @@
+package io.customer.datapipelines
+
+import io.customer.commontest.extensions.random
+import io.customer.datapipelines.testutils.core.JUnitTest
+import org.amshove.kluent.shouldBeFalse
+import org.amshove.kluent.shouldBeTrue
+import org.junit.jupiter.api.Test
+
+/**
+ * Direct coverage for the synchronous `CustomerIO.isUserIdentified` mirror at the layer that owns
+ * it. The messagingpush live-notification tests mock `isUserIdentified`, so they don't prove the
+ * contract itself: it must flip true on the caller thread the instant `identify()` returns (before
+ * the async `analytics.userId()` catches up), flip false the instant `clearIdentify()` returns, and
+ * still reflect a session restored from persistence where neither was called this process.
+ */
+class IsUserIdentifiedSyncTests : JUnitTest() {
+
+    @Test
+    fun isUserIdentified_isTrueSynchronouslyAfterIdentify() {
+        sdkInstance.isUserIdentified.shouldBeFalse()
+
+        sdkInstance.identify(String.random)
+
+        // No async wait: the mirror is set on the caller thread inside identify(), so a consumer
+        // gating on it immediately after login (e.g. a live-notification start()) is not dropped.
+        sdkInstance.isUserIdentified.shouldBeTrue()
+    }
+
+    @Test
+    fun isUserIdentified_isFalseSynchronouslyAfterClearIdentify() {
+        sdkInstance.identify(String.random)
+        sdkInstance.isUserIdentified.shouldBeTrue()
+
+        sdkInstance.clearIdentify()
+
+        sdkInstance.isUserIdentified.shouldBeFalse()
+    }
+
+    @Test
+    fun isUserIdentified_fallsBackToAnalyticsForRestoredSession() {
+        // Cold launch: no identify() this process (the mirror stays unset), but analytics restored a
+        // userId from persistence. isUserIdentified must reflect the restored user via the fallback.
+        analytics.identify(String.random)
+
+        sdkInstance.isUserIdentified.shouldBeTrue()
+    }
+}