refactor(iv): align jwt-invalidated replay with #2613 buffer-and-consume

Rewrite the listener replay to match the reference design from #2613
commit 89cca43:
- Lock-protected pendingJwtInvalidatedExternalId (consume-on-first-subscribe)
  replaces the @volatile lastJwtInvalidatedExternalId (always-replay)
- fireJwtInvalidated buffers ONLY when no subscribers exist; otherwise
  schedules an async fire. Closes the double-fire race.
- onModelReplaced clears the buffer on IdentityModel replacement (login or
  logout switch via UserSwitcher.createAndSwitchToNewUser → replace).
- Drop the explicit clearLastJwtInvalidated() method and its calls in
  OneSignalImp.logout() / logoutSuspend() — clear is now automatic.

Match #2599 conventions:
- Drop the isInitialized throw in addUserJwtInvalidatedListener /
  removeUserJwtInvalidatedListener.
- Add updateUserJwtSuspend.

Update IUserJwtInvalidatedListener docstring to clarify replay is
synchronous on the caller's thread; regular fire is async.

Tests rewritten for the new semantics: first-subscriber-replay,
consume-on-first, fire-with-subscribers-doesn't-buffer, onModelReplaced
clears.

Author: nan-li

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/IOneSignal.kt

@@ -255,4 +255,16 @@ interface IOneSignal {
      * Logout the current user (suspend version).
      */
     suspend fun logoutSuspend()
+
+    /**
+     * Update the JWT bearer token associated with [externalId] (suspend version). Suspends
+     * until SDK initialization is complete, then stores the JWT and wakes the operation queue.
+     *
+     * @param externalId The external ID the JWT belongs to.
+     * @param token The new JWT bearer token issued by your backend.
+     */
+    suspend fun updateUserJwtSuspend(
+        externalId: String,
+        token: String,
+    )
 }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/IUserJwtInvalidatedListener.kt

@@ -6,7 +6,11 @@ package com.onesignal
  * detected that the JWT for a user is no longer valid (typically a 401 from
  * the OneSignal backend on a request signed with that JWT).
  *
- * Callbacks are delivered on a background thread.
+ * Threading: regular fire delivery happens on a background dispatcher. Replay
+ * delivery (when an invalidation occurred before any listener was subscribed)
+ * happens synchronously on the thread that calls
+ * [IOneSignal.addUserJwtInvalidatedListener]. Implementations should not assume
+ * a specific thread.
  */
 fun interface IUserJwtInvalidatedListener {
     /**

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt

@@ -420,10 +420,6 @@ internal class OneSignalImp(
             }
         }
 
-        // Clear the replay cache so a stale jwt-invalidated event from the previous user
-        // doesn't fire for a listener registered after logout switches us to a new user.
-        services.getServiceOrNull<UserManager>()?.clearLastJwtInvalidated()
-
         val context = logoutHelper.switchUser() ?: return
 
         if (isBackgroundThreadingEnabled) {
@@ -457,16 +453,10 @@ internal class OneSignalImp(
     }
 
     override fun addUserJwtInvalidatedListener(listener: IUserJwtInvalidatedListener) {
-        if (!isInitialized) {
-            throw IllegalStateException("Must call 'initWithContext' before 'addUserJwtInvalidatedListener'")
-        }
         services.getService<UserManager>().addJwtInvalidatedListener(listener)
     }
 
     override fun removeUserJwtInvalidatedListener(listener: IUserJwtInvalidatedListener) {
-        if (!isInitialized) {
-            throw IllegalStateException("Must call 'initWithContext' before 'removeUserJwtInvalidatedListener'")
-        }
         services.getService<UserManager>().removeJwtInvalidatedListener(listener)
     }
 
@@ -711,6 +701,22 @@ internal class OneSignalImp(
         loginHelper.enqueueLogin(context)
     }
 
+    override suspend fun updateUserJwtSuspend(
+        externalId: String,
+        token: String,
+    ) = withContext(runtimeIoDispatcher) {
+        Logging.log(LogLevel.DEBUG, "updateUserJwtSuspend(externalId: $externalId, token: ...${token.takeLast(8)})")
+
+        suspendUntilInit(operationName = "updateUserJwt")
+
+        if (!isInitialized) {
+            throw IllegalStateException("'initWithContext failed' before 'updateUserJwt'")
+        }
+
+        jwtTokenStore.putJwt(externalId, token)
+        operationRepo.forceExecuteOperations()
+    }
+
     override suspend fun logoutSuspend() =
         withContext(runtimeIoDispatcher) {
             Logging.log(LogLevel.DEBUG, "logoutSuspend()")
@@ -721,8 +727,6 @@ internal class OneSignalImp(
                 throw IllegalStateException("'initWithContext failed' before 'logout'")
             }
 
-            services.getServiceOrNull<UserManager>()?.clearLastJwtInvalidated()
-
             val context = logoutHelper.switchUser() ?: return@withContext
             logoutHelper.enqueueLogout(context)
         }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/internal/UserManager.kt

@@ -56,16 +56,18 @@ internal open class UserManager(
     private val jwtInvalidatedNotifier = EventProducer<IUserJwtInvalidatedListener>()
 
     /**
-     * Most recent invalidation event, replayed to listeners that subscribe after the fact.
-     * Cleared on logout (via [clearLastJwtInvalidated]) so a stale event from a previous
-     * user doesn't fire for an unrelated session.
+     * Buffers a fired invalidation when no listeners are subscribed yet (e.g. SDK init / cold-start
+     * 401 before app code wires up its listener). Consumed-on-first-subscribe by the next
+     * [addJwtInvalidatedListener] call. Cleared automatically when the IdentityModel is replaced
+     * (login or logout) so a stale event doesn't leak across users.
      */
-    @Volatile
-    private var lastJwtInvalidatedExternalId: String? = null
+    private val jwtInvalidatedLock = Any()
+    private var pendingJwtInvalidatedExternalId: String? = null
 
     /**
      * Async dispatch of [IUserJwtInvalidatedListener] callbacks so HYDRATE / op-repo paths
      * that synchronously trigger invalidation don't run app code on the SDK's internal thread.
+     * Replay (synchronous, on the calling thread) bypasses this scope.
      */
     private val jwtInvalidatedDispatchScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
 
@@ -91,17 +93,22 @@ internal open class UserManager(
 
     /**
      * Subscribe a developer-facing listener for JWT-invalidated events. If an invalidation
-     * has already fired before this listener subscribed, the most recent event is replayed
-     * so apps that subscribe late don't miss the signal.
+     * has already fired before any listener was subscribed (e.g. early-startup 401), the
+     * buffered event is delivered to this listener synchronously and consumed — subsequent
+     * subscribers do not receive that same event.
      */
     fun addJwtInvalidatedListener(listener: IUserJwtInvalidatedListener) {
-        jwtInvalidatedNotifier.subscribe(listener)
-        val replayId = lastJwtInvalidatedExternalId
-        if (replayId != null) {
-            jwtInvalidatedDispatchScope.launch {
-                runCatching { listener.onUserJwtInvalidated(UserJwtInvalidatedEvent(replayId)) }
-                    .onFailure { Logging.warn("UserManager: replayed jwt-invalidated listener threw: ${it.message}") }
-            }
+        val pendingExternalId: String?
+        synchronized(jwtInvalidatedLock) {
+            jwtInvalidatedNotifier.subscribe(listener)
+            pendingExternalId = pendingJwtInvalidatedExternalId
+            pendingJwtInvalidatedExternalId = null
+        }
+        // Deliver the replay outside the lock so a slow listener doesn't block other
+        // subscribe/unsubscribe/fire calls. Replay runs on the caller's thread (sync).
+        pendingExternalId?.let {
+            runCatching { listener.onUserJwtInvalidated(UserJwtInvalidatedEvent(it)) }
+                .onFailure { ex -> Logging.warn("UserManager: replayed jwt-invalidated listener threw: ${ex.message}") }
         }
     }
 
@@ -111,25 +118,24 @@ internal open class UserManager(
 
     /**
      * Fire [IUserJwtInvalidatedListener.onUserJwtInvalidated] to all subscribed listeners on
-     * a background dispatcher. Caches the event for late-subscribing listeners (replay).
+     * a background dispatcher. If no listener is currently subscribed, buffer the externalId
+     * to be delivered to the next listener that subscribes (consume-on-first-subscribe).
      */
     fun fireJwtInvalidated(externalId: String) {
-        lastJwtInvalidatedExternalId = externalId
-        jwtInvalidatedDispatchScope.launch {
-            runCatching {
-                jwtInvalidatedNotifier.fire { listener ->
-                    runCatching { listener.onUserJwtInvalidated(UserJwtInvalidatedEvent(externalId)) }
-                        .onFailure { Logging.warn("UserManager: jwt-invalidated listener threw: ${it.message}") }
+        synchronized(jwtInvalidatedLock) {
+            if (jwtInvalidatedNotifier.hasSubscribers) {
+                jwtInvalidatedDispatchScope.launch {
+                    jwtInvalidatedNotifier.fire { listener ->
+                        runCatching { listener.onUserJwtInvalidated(UserJwtInvalidatedEvent(externalId)) }
+                            .onFailure { ex -> Logging.warn("UserManager: jwt-invalidated listener threw: ${ex.message}") }
+                    }
                 }
-            }.onFailure { Logging.warn("Failed to deliver JWT invalidated event for externalId=$externalId: ${it.message}") }
+            } else {
+                pendingJwtInvalidatedExternalId = externalId
+            }
         }
     }
 
-    /** Clear the replay cache. Called on logout so stale events don't fire for the next user. */
-    fun clearLastJwtInvalidated() {
-        lastJwtInvalidatedExternalId = null
-    }
-
     // IJwtUpdateListener — JwtTokenStore -> developer-facing event bridge.
     override fun onJwtInvalidated(externalId: String) {
         fireJwtInvalidated(externalId)
@@ -336,7 +342,13 @@ internal open class UserManager(
     override fun onModelReplaced(
         model: IdentityModel,
         tag: String,
-    ) { }
+    ) {
+        // IdentityModel replacement = login or logout switch. Clear any buffered invalidation
+        // so the next user's listener doesn't replay the previous user's stale event.
+        synchronized(jwtInvalidatedLock) {
+            pendingJwtInvalidatedExternalId = null
+        }
+    }
 
     override fun onModelUpdated(
         args: ModelChangedArgs,

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/UserManagerTests.kt

@@ -227,7 +227,7 @@ class UserManagerTests : FunSpec({
         firedExternalId shouldBe "alice"
     }
 
-    test("listener replay: subscribers added after fire receive the most recent event") {
+    test("listener replay: first subscriber after a no-listener fire receives the buffered event") {
         // Given
         val jwtTokenStore = JwtTokenStore(MockPreferencesService())
         val userManager =
@@ -239,24 +239,18 @@ class UserManagerTests : FunSpec({
                 MockHelper.languageContext(),
                 jwtTokenStore,
             )
-        // Fire an invalidation BEFORE any listener is registered.
+        // Fire BEFORE any listener is registered → buffered.
         userManager.fireJwtInvalidated("alice")
 
-        var lateExternalId: String? = null
-        val waiter = com.onesignal.common.threading.Waiter()
-
         // When: listener subscribes after the fire.
-        userManager.addJwtInvalidatedListener { event ->
-            lateExternalId = event.externalId
-            waiter.wake()
-        }
-        waiter.waitForWake()
+        var lateExternalId: String? = null
+        userManager.addJwtInvalidatedListener { event -> lateExternalId = event.externalId }
 
-        // Then: late subscriber receives the cached event.
+        // Then: replay delivers synchronously on subscribe.
         lateExternalId shouldBe "alice"
     }
 
-    test("clearLastJwtInvalidated stops replay for new subscribers") {
+    test("buffered event is consumed by the first subscriber; second subscriber gets nothing") {
         val jwtTokenStore = JwtTokenStore(MockPreferencesService())
         val userManager =
             UserManager(
@@ -269,16 +263,62 @@ class UserManagerTests : FunSpec({
             )
         userManager.fireJwtInvalidated("alice")
 
-        // When: cache cleared (e.g. on logout)
-        userManager.clearLastJwtInvalidated()
+        // First subscriber consumes the buffered event.
+        var firstFired: String? = null
+        userManager.addJwtInvalidatedListener { event -> firstFired = event.externalId }
+        firstFired shouldBe "alice"
+
+        // Second subscriber must NOT receive a replay (buffer was already consumed).
+        var secondFired = false
+        userManager.addJwtInvalidatedListener { secondFired = true }
+        secondFired shouldBe false
+    }
 
+    test("fire when subscribers exist does NOT buffer for late subscribers") {
+        val jwtTokenStore = JwtTokenStore(MockPreferencesService())
+        val userManager =
+            UserManager(
+                mockk<ISubscriptionManager>(),
+                MockHelper.identityModelStore(),
+                MockHelper.propertiesModelStore(),
+                MockHelper.customEventController(),
+                MockHelper.languageContext(),
+                jwtTokenStore,
+            )
+
+        // Existing subscriber at the time of fire.
+        userManager.addJwtInvalidatedListener { /* no-op */ }
+        userManager.fireJwtInvalidated("alice")
+        Thread.sleep(50) // allow async fire to dispatch
+
+        // Late subscriber must NOT receive a replay (event was not buffered, since
+        // there was already a listener at fire time).
         var lateFired = false
         userManager.addJwtInvalidatedListener { lateFired = true }
+        lateFired shouldBe false
+    }
 
-        // Give async dispatcher a chance to run if it would have.
-        Thread.sleep(50)
+    test("onModelReplaced clears any buffered invalidation event (login/logout switch)") {
+        val identityModelStore = MockHelper.identityModelStore()
+        val jwtTokenStore = JwtTokenStore(MockPreferencesService())
+        val userManager =
+            UserManager(
+                mockk<ISubscriptionManager>(),
+                identityModelStore,
+                MockHelper.propertiesModelStore(),
+                MockHelper.customEventController(),
+                MockHelper.languageContext(),
+                jwtTokenStore,
+            )
+        userManager.fireJwtInvalidated("alice")
+
+        // Simulate a user-switch: IdentityModelStore replaces the model and notifies
+        // subscribers (UserManager subscribes itself in init).
+        userManager.onModelReplaced(identityModelStore.model, "")
 
-        // Then: no replay fired.
+        // Late subscriber must NOT receive the (now-cleared) buffered event.
+        var lateFired = false
+        userManager.addJwtInvalidatedListener { lateFired = true }
         lateFired shouldBe false
     }