chore(source): update cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/PayCraft.kt cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/core/PayCraftBillingManager.kt cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/network/PayCraftRealtime.kt

Author: therajanmaurya

cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/PayCraft.kt

@@ -539,6 +539,23 @@ object PayCraft {
         realtime.ensureConfigChannel(tenantId) {
             configFetchJob = applicationScope.launch { runCatching { prefetchProducts() } }
         }
+        refreshRealtimeIdentity()
+    }
+
+    /**
+     * (Re)bind the realtime ENTITLEMENT channel to the CURRENT buyer identity
+     * (`email ?: deviceId`). Called on every config apply AND by [BillingManager]
+     * whenever the identity changes — login (device-id → email) and logout
+     * (email → device-id) — so an entitlement push always reaches the buyer's
+     * channel. The realtime client removes the prior channel before subscribing
+     * the new one, so the old user's pings stop after logout. No-op until a
+     * SuiteConfig (hence tenant_id) has landed. Best-effort; failures leave the
+     * TTL/foreground sync as the fallback.
+     */
+    internal fun refreshRealtimeIdentity() {
+        val tenantId = suiteConfig?.tenantId ?: return
+        val koin = KoinPlatform.getKoinOrNull() ?: return
+        val realtime = koin.getOrNull<PayCraftRealtime>() ?: return
         applicationScope.launch {
             val email = runCatching { koin.getOrNull<PayCraftStore>()?.getEmail() }.getOrNull()
             val appUserId = email?.trim()?.lowercase()?.ifBlank { null } ?: deviceId

cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/core/PayCraftBillingManager.kt

@@ -138,6 +138,7 @@ class PayCraftBillingManager(
         _billingState.value = BillingState.Loading
         scope.launch {
             store.saveEmail(normalized)
+            PayCraft.refreshRealtimeIdentity() // re-bind entitlement channel to the new email (audit H3)
             performRegisterAndLogin(normalized)
         }
     }
@@ -336,15 +337,19 @@ class PayCraftBillingManager(
         val currentState = _billingState.value
         PayCraftLogger.onRefreshStatus(email)
 
-        // Don't refresh while a conflict/verification/transfer flow is active —
-        // the concurrent re-register would overwrite OwnershipVerified with DeviceConflict.
-        if (currentState is BillingState.DeviceConflict ||
-            currentState is BillingState.OwnershipVerified ||
-            currentState is BillingState.Loading
-        ) {
+        // Never refresh over a conflict/verification/transfer flow — a concurrent
+        // re-register would overwrite OwnershipVerified with DeviceConflict. These
+        // stay protected even under force.
+        // A `Loading` window is only skipped when NOT forced: a realtime entitlement
+        // push (force=true) MUST still land during an in-flight refresh, otherwise the
+        // live update is silently dropped and falls back to the slow TTL (audit H5).
+        val protectedFlow = currentState is BillingState.DeviceConflict ||
+            currentState is BillingState.OwnershipVerified
+        val loadingSkip = currentState is BillingState.Loading && !force
+        if (protectedFlow || loadingSkip) {
             PayCraftLogger.onFlow(
                 "refreshStatus",
-                "SKIPPED — active flow in progress (state=${currentState::class.simpleName})",
+                "SKIPPED — active flow in progress (state=${currentState::class.simpleName}, force=$force)",
             )
             return
         }
@@ -400,7 +405,7 @@ class PayCraftBillingManager(
 
         val normalized = email.trim().lowercase()
         _userEmail.value = normalized
-        scope.launch { store.saveEmail(normalized) }
+        scope.launch { store.saveEmail(normalized); PayCraft.refreshRealtimeIdentity() }
 
         // If there's an active conflict and the verified email matches → ownership proven.
         val pendingToken = DeviceTokenStore.getToken()
@@ -573,7 +578,7 @@ class PayCraftBillingManager(
         _billingState.value = BillingState.Free
         lastConflict = null
         store.clearCache()
-        scope.launch { store.clearEmail() }
+        scope.launch { store.clearEmail(); PayCraft.refreshRealtimeIdentity() } // drop old user's entitlement channel (audit M1)
     }
 
     // ─── Store5 entitlement gating (Phase 4 — cache-first + revalidate) ───────

cmp-paycraft/src/commonMain/kotlin/com/mobilebytelabs/paycraft/network/PayCraftRealtime.kt

@@ -12,6 +12,8 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.serialization.json.JsonObject
 
 /**
@@ -42,67 +44,78 @@ class PayCraftRealtime(private val supabase: SupabaseClient) {
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
 
+    // All channel state is read/written ONLY inside [mutex] — this makes the
+    // check-then-subscribe atomic (no duplicate channel when startRealtime is
+    // called twice per config fetch) and the fields thread-safe across the
+    // Default-dispatcher coroutines + the calling thread (JVM + Kotlin/Native).
+    private val mutex = Mutex()
     private var configChannel: RealtimeChannel? = null
     private var entitlementChannel: RealtimeChannel? = null
     private var configTenant: String? = null
     private var entitlementKey: String? = null // "tenant:appUserId"
 
     /** Subscribe to `config:{tenantId}`; invoke [onChanged] on every config ping. */
     fun ensureConfigChannel(tenantId: String, onChanged: () -> Unit) {
-        if (configTenant == tenantId && configChannel != null) return
         scope.launch {
-            runCatching {
-                configChannel?.let { supabase.realtime.removeChannel(it) }
-                val ch = supabase.channel("config:$tenantId")
-                ch.broadcastFlow<JsonObject>(event = "config_changed")
-                    .onEach {
-                        PayCraftLogger.onFlow("realtime", "config ping → refetching /config")
-                        onChanged()
-                    }
-                    .launchIn(scope)
-                ch.subscribe()
-                configChannel = ch
-                configTenant = tenantId
-                PayCraftLogger.onFlow("realtime", "subscribed config:$tenantId")
-            }.onFailure {
-                PayCraftLogger.onFlow("realtime", "config subscribe failed (TTL fallback stays): ${it.message}")
+            mutex.withLock {
+                if (configTenant == tenantId && configChannel != null) return@withLock
+                runCatching {
+                    configChannel?.let { supabase.realtime.removeChannel(it) }
+                    val ch = supabase.channel("config:$tenantId")
+                    ch.broadcastFlow<JsonObject>(event = "config_changed")
+                        .onEach {
+                            PayCraftLogger.onFlow("realtime", "config ping → refetching /config")
+                            onChanged()
+                        }
+                        .launchIn(scope)
+                    ch.subscribe()
+                    configChannel = ch
+                    configTenant = tenantId
+                    PayCraftLogger.onFlow("realtime", "subscribed config:$tenantId")
+                }.onFailure {
+                    PayCraftLogger.onFlow("realtime", "config subscribe failed (TTL fallback stays): ${it.message}")
+                }
             }
         }
     }
 
     /** Subscribe to `entitlement:{tenantId}:{appUserId}`; re-subscribes on identity change. */
     fun ensureEntitlementChannel(tenantId: String, appUserId: String, onChanged: () -> Unit) {
         val key = "$tenantId:$appUserId"
-        if (entitlementKey == key && entitlementChannel != null) return
         scope.launch {
-            runCatching {
-                entitlementChannel?.let { supabase.realtime.removeChannel(it) }
-                val ch = supabase.channel("entitlement:$tenantId:$appUserId")
-                ch.broadcastFlow<JsonObject>(event = "entitlement_changed")
-                    .onEach {
-                        PayCraftLogger.onFlow("realtime", "entitlement ping → force refresh")
-                        onChanged()
-                    }
-                    .launchIn(scope)
-                ch.subscribe()
-                entitlementChannel = ch
-                entitlementKey = key
-                PayCraftLogger.onFlow("realtime", "subscribed entitlement:$tenantId:***")
-            }.onFailure {
-                PayCraftLogger.onFlow("realtime", "entitlement subscribe failed (TTL fallback stays): ${it.message}")
+            mutex.withLock {
+                if (entitlementKey == key && entitlementChannel != null) return@withLock
+                runCatching {
+                    entitlementChannel?.let { supabase.realtime.removeChannel(it) }
+                    val ch = supabase.channel("entitlement:$tenantId:$appUserId")
+                    ch.broadcastFlow<JsonObject>(event = "entitlement_changed")
+                        .onEach {
+                            PayCraftLogger.onFlow("realtime", "entitlement ping → force refresh")
+                            onChanged()
+                        }
+                        .launchIn(scope)
+                    ch.subscribe()
+                    entitlementChannel = ch
+                    entitlementKey = key
+                    PayCraftLogger.onFlow("realtime", "subscribed entitlement:$tenantId:***")
+                }.onFailure {
+                    PayCraftLogger.onFlow("realtime", "entitlement subscribe failed (TTL fallback stays): ${it.message}")
+                }
             }
         }
     }
 
     /** Tear down both channels (call on logout / SDK teardown). */
     fun stop() {
         scope.launch {
-            configChannel?.let { runCatching { supabase.realtime.removeChannel(it) } }
-            entitlementChannel?.let { runCatching { supabase.realtime.removeChannel(it) } }
-            configChannel = null
-            entitlementChannel = null
-            configTenant = null
-            entitlementKey = null
+            mutex.withLock {
+                configChannel?.let { runCatching { supabase.realtime.removeChannel(it) } }
+                entitlementChannel?.let { runCatching { supabase.realtime.removeChannel(it) } }
+                configChannel = null
+                entitlementChannel = null
+                configTenant = null
+                entitlementKey = null
+            }
         }
     }
 }