refactor(iv): drop jwt-invalidated buffer/replay; pure pub/sub matches iOS

The buffer-and-consume design (pendingJwtInvalidatedExternalId +
jwtInvalidatedLock + consume-on-first-subscribe replay) was added to
handle cold-start 401s when no listener was subscribed yet. iOS doesn't
do this — it fires only to currently-subscribed listeners, late
subscribers miss earlier events. Match iOS: drop the buffer entirely.

Reverts the structure introduced by d137481 ("align jwt-invalidated
replay with #2613 buffer-and-consume"). The simpler design here matches
reference branch #2599.

Side effects:
- onModelReplaced becomes a no-op (no buffer to clear). Kept the
  override since ISingletonModelStoreChangeHandler requires it.
- fireJwtInvalidated now uses OneSignalDispatchers.launchOnDefault per
  maintainer request (#3184062053) instead of a custom
  CoroutineScope(SupervisorJob() + Dispatchers.Default).
- KDoc on IOneSignal.addUserJwtInvalidatedListener and the OneSignal.kt
  facade now spell out the pure-pub/sub semantics: "Subscribe early
  (e.g. in Application.onCreate) to avoid missing cold-start 401s."

Test changes: drop the 4 buffer/replay-specific tests
(listener-replay-buffered-event, consume-on-first-subscribe,
fire-with-subscribers-no-buffer, onModelReplaced-clears-buffer).
Replace with one test that confirms late subscribers don't receive
earlier events.

Author: nan-li

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

@@ -151,9 +151,8 @@ interface IOneSignal {
      * a 401 from the OneSignal backend). Apps should respond by fetching a fresh JWT from
      * their backend and supplying it via [updateUserJwt].
      *
-     * Listener replay: if an invalidation has already occurred before this listener is
-     * registered, the most recent invalidation is delivered to the new listener so apps
-     * that subscribe late don't miss the signal.
+     * Pure pub/sub: only listeners subscribed at the time of the invalidation receive the
+     * event. Subscribe early (e.g. in `Application.onCreate`) to avoid missing events.
      */
     fun addUserJwtInvalidatedListener(listener: IUserJwtInvalidatedListener)
 

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

@@ -364,9 +364,8 @@ object OneSignal {
      * a 401 from the OneSignal backend). Apps should respond by fetching a fresh JWT from
      * their backend and supplying it via [updateUserJwt].
      *
-     * Listener replay: if an invalidation has already occurred before this listener is
-     * registered, the most recent invalidation is delivered to the new listener so apps
-     * that subscribe late don't miss the signal.
+     * Pure pub/sub: only listeners subscribed at the time of the invalidation receive the
+     * event. Subscribe early (e.g. in `Application.onCreate`) to avoid missing events.
      */
     @JvmStatic
     fun addUserJwtInvalidatedListener(listener: IUserJwtInvalidatedListener) =

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

@@ -25,11 +25,8 @@ import com.onesignal.user.internal.subscriptions.SubscriptionList
 import com.onesignal.user.state.IUserStateObserver
 import com.onesignal.user.state.UserChangedState
 import com.onesignal.user.state.UserState
+import com.onesignal.common.threading.OneSignalDispatchers
 import com.onesignal.user.subscriptions.IPushSubscription
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.launch
 
 internal open class UserManager(
     private val _subscriptionManager: ISubscriptionManager,
@@ -55,22 +52,6 @@ internal open class UserManager(
 
     private val jwtInvalidatedNotifier = EventProducer<IUserJwtInvalidatedListener>()
 
-    /**
-     * 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.
-     */
-    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)
-
     override val pushSubscription: IPushSubscription
         get() = _subscriptionManager.subscriptions.push
 
@@ -91,47 +72,25 @@ internal open class UserManager(
         _jwtTokenStore.subscribe(this)
     }
 
-    /**
-     * Subscribe a developer-facing listener for JWT-invalidated events. If an invalidation
-     * 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) {
-        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) }
-        }
+        jwtInvalidatedNotifier.subscribe(listener)
     }
 
     fun removeJwtInvalidatedListener(listener: IUserJwtInvalidatedListener) {
         jwtInvalidatedNotifier.unsubscribe(listener)
     }
 
     /**
-     * Fire [IUserJwtInvalidatedListener.onUserJwtInvalidated] to all subscribed listeners on
-     * 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).
+     * Fire [IUserJwtInvalidatedListener.onUserJwtInvalidated] to all currently-subscribed
+     * listeners on a background dispatcher (so HYDRATE / op-repo paths that synchronously
+     * trigger invalidation don't run app code on the SDK's internal thread). Pure pub/sub —
+     * matches iOS: late subscribers don't get a replay of earlier events.
      */
     fun fireJwtInvalidated(externalId: String) {
-        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) }
-                    }
-                }
-            } else {
-                pendingJwtInvalidatedExternalId = externalId
+        OneSignalDispatchers.launchOnDefault {
+            jwtInvalidatedNotifier.fire { listener ->
+                runCatching { listener.onUserJwtInvalidated(UserJwtInvalidatedEvent(externalId)) }
+                    .onFailure { ex -> Logging.warn("UserManager: jwt-invalidated listener threw", ex) }
             }
         }
     }
@@ -342,13 +301,7 @@ 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,8 +227,9 @@ class UserManagerTests : FunSpec({
         firedExternalId shouldBe "alice"
     }
 
-    test("listener replay: first subscriber after a no-listener fire receives the buffered event") {
-        // Given
+    test("late subscriber does not receive earlier events — pure pub/sub, no replay") {
+        // Matches iOS: only listeners subscribed at the time of the fire receive the event.
+        // No buffering of earlier invalidations for late subscribers.
         val jwtTokenStore = JwtTokenStore(MockPreferencesService())
         val userManager =
             UserManager(
@@ -239,86 +240,15 @@ class UserManagerTests : FunSpec({
                 MockHelper.languageContext(),
                 jwtTokenStore,
             )
-        // Fire BEFORE any listener is registered → buffered.
-        userManager.fireJwtInvalidated("alice")
 
-        // When: listener subscribes after the fire.
-        var lateExternalId: String? = null
-        userManager.addJwtInvalidatedListener { event -> lateExternalId = event.externalId }
-
-        // Then: replay delivers synchronously on subscribe.
-        lateExternalId shouldBe "alice"
-    }
-
-    test("buffered event is consumed by the first subscriber; second subscriber gets nothing") {
-        val jwtTokenStore = JwtTokenStore(MockPreferencesService())
-        val userManager =
-            UserManager(
-                mockk<ISubscriptionManager>(),
-                MockHelper.identityModelStore(),
-                MockHelper.propertiesModelStore(),
-                MockHelper.customEventController(),
-                MockHelper.languageContext(),
-                jwtTokenStore,
-            )
-        userManager.fireJwtInvalidated("alice")
-
-        // 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 */ }
+        // Fire before any listener is registered.
         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
-    }
-
-    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, "")
-
-        // Late subscriber must NOT receive the (now-cleared) buffered event.
+        // Late subscriber must not receive the earlier event.
         var lateFired = false
         userManager.addJwtInvalidatedListener { lateFired = true }
+        Thread.sleep(50)
         lateFired shouldBe false
     }