swap EventProducer for CompletableDeferred

This change fixes a rare bug where RecoverFromDroppedLoginBug may not
run. While very rare this could happen if start was called on
OperationRepo first and it finished before
RecoverFromDroppedLoginBug.start was called.

We swapped out EventProducer for a CompletableDeferred to address this
issue. CompletableDeferred is a better fit as we simply need to wait
until something is initialized, and it will never fire more than once.

We also refactored RecoverFromDroppedLoginBugTests to account for the
change, but improved the testing to ensure the operation itself is
correct.

Author: jkasten2

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepo.kt

@@ -39,7 +39,7 @@ interface IOperationRepo {
      */
     fun <T : Operation> containsInstanceOf(type: KClass<T>): Boolean
 
-    fun addOperationLoadedListener(handler: IOperationRepoLoadedListener)
+    suspend fun awaitInitialized()
 }
 
 // Extension function so the syntax containsInstanceOf<Operation>() can be used over

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/IOperationRepoLoadedListener.kt

@@ -1,20 +1,18 @@
 package com.onesignal.core.internal.operations.impl
 
-import com.onesignal.common.events.EventProducer
-import com.onesignal.common.events.IEventNotifier
 import com.onesignal.common.threading.WaiterWithValue
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.ExecutionResult
 import com.onesignal.core.internal.operations.GroupComparisonType
 import com.onesignal.core.internal.operations.IOperationExecutor
 import com.onesignal.core.internal.operations.IOperationRepo
-import com.onesignal.core.internal.operations.IOperationRepoLoadedListener
 import com.onesignal.core.internal.operations.Operation
 import com.onesignal.core.internal.startup.IStartableService
 import com.onesignal.core.internal.time.ITime
 import com.onesignal.debug.LogLevel
 import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.user.internal.operations.impl.states.NewRecordsState
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -30,7 +28,7 @@ internal class OperationRepo(
     private val _configModelStore: ConfigModelStore,
     private val _time: ITime,
     private val _newRecordState: NewRecordsState,
-) : IOperationRepo, IStartableService, IEventNotifier<IOperationRepoLoadedListener> {
+) : IOperationRepo, IStartableService {
     internal class OperationQueueItem(
         val operation: Operation,
         val waiter: WaiterWithValue<Boolean>? = null,
@@ -52,17 +50,10 @@ internal class OperationRepo(
     private val waiter = WaiterWithValue<LoopWaiterMessage>()
     private var paused = false
     private var coroutineScope = CoroutineScope(newSingleThreadContext(name = "OpRepo"))
-    private val loadedSubscription: EventProducer<IOperationRepoLoadedListener> = EventProducer()
+    private val initialized = CompletableDeferred<Unit>()
 
-    override val hasSubscribers: Boolean
-        get() = loadedSubscription.hasSubscribers
-
-    override fun unsubscribe(handler: IOperationRepoLoadedListener) {
-        loadedSubscription.unsubscribe(handler)
-    }
-
-    override fun subscribe(handler: IOperationRepoLoadedListener) {
-        loadedSubscription.subscribe(handler)
+    override suspend fun awaitInitialized() {
+        initialized.await()
     }
 
     /** *** Buckets ***
@@ -101,10 +92,6 @@ internal class OperationRepo(
         }
     }
 
-    override fun addOperationLoadedListener(handler: IOperationRepoLoadedListener) {
-        subscribe(handler)
-    }
-
     override fun start() {
         paused = false
         coroutineScope.launch {
@@ -431,6 +418,6 @@ internal class OperationRepo(
                 )
             if (successful) successfulIndex++
         }
-        loadedSubscription.fire { it.onOperationRepoLoaded() }
+        initialized.complete(Unit)
     }
 }

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt

@@ -3,12 +3,14 @@ package com.onesignal.user.internal.migrations
 import com.onesignal.common.IDManager
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.IOperationRepo
-import com.onesignal.core.internal.operations.IOperationRepoLoadedListener
 import com.onesignal.core.internal.operations.containsInstanceOf
 import com.onesignal.core.internal.startup.IStartableService
 import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.user.internal.identity.IdentityModelStore
 import com.onesignal.user.internal.operations.LoginUserOperation
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.launch
 
 /**
  * Purpose: Automatically recovers a stalled User in the OperationRepo due
@@ -31,21 +33,20 @@ class RecoverFromDroppedLoginBug(
     private val _operationRepo: IOperationRepo,
     private val _identityModelStore: IdentityModelStore,
     private val _configModelStore: ConfigModelStore,
-) : IStartableService, IOperationRepoLoadedListener {
+) : IStartableService {
     override fun start() {
-        _operationRepo.addOperationLoadedListener(this)
-    }
-
-    override fun onOperationRepoLoaded() {
-        if (isInBadState()) {
-            Logging.warn(
-                "User with externalId:" +
-                    "${_identityModelStore.model.externalId} " +
-                    "was in a bad state, causing it to not update on OneSignal's " +
-                    "backend! We are recovering and replaying all unsent " +
-                    "operations now.",
-            )
-            recoverByAddingBackDroppedLoginOperation()
+        GlobalScope.launch(Dispatchers.IO) {
+            _operationRepo.awaitInitialized()
+            if (isInBadState()) {
+                Logging.warn(
+                    "User with externalId:" +
+                        "${_identityModelStore.model.externalId} " +
+                        "was in a bad state, causing it to not update on OneSignal's " +
+                        "backend! We are recovering and replaying all unsent " +
+                        "operations now.",
+                )
+                recoverByAddingBackDroppedLoginOperation()
+            }
         }
     }
 

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

@@ -25,6 +25,8 @@ import io.mockk.spyk
 import io.mockk.verify
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.time.withTimeout
+import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.coroutines.yield
 import java.util.UUID
@@ -599,24 +601,17 @@ class OperationRepoTests : FunSpec({
         result shouldBe null
     }
 
-    test("ensure onOperationRepoLoaded is called once loading is completed") {
+    test("ensure awaitInitialized() unsuspends") {
         // Given
         val mocks = Mocks()
-        val spyListener = spyk<IOperationRepoLoadedListener>()
 
         // When
-        mocks.operationRepo.addOperationLoadedListener(spyListener)
         mocks.operationRepo.start()
         // enqueueAndWait used to know we are fully loaded.
         mocks.operationRepo.enqueueAndWait(mockOperation())
 
         // Then
-        mocks.operationRepo.hasSubscribers shouldBe true
-        coVerifyOrder {
-            mocks.operationRepo.subscribe(any())
-            mocks.operationModelStore.loadOperations()
-            spyListener.onOperationRepoLoaded()
-        }
+        withTimeout(1_000) { mocks.operationRepo.awaitInitialized() }
     }
 
     test("ensure loadSavedOperations doesn't duplicate existing OperationItems") {

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt

@@ -1,14 +1,15 @@
 package com.onesignal.user.internal.migrations
 
-import com.onesignal.common.threading.Waiter
-import com.onesignal.core.internal.config.ConfigModelStore
-import com.onesignal.core.internal.operations.IOperationRepoLoadedListener
 import com.onesignal.core.internal.operations.impl.OperationModelStore
 import com.onesignal.core.internal.operations.impl.OperationRepo
 import com.onesignal.core.internal.time.impl.Time
+import com.onesignal.debug.LogLevel
+import com.onesignal.debug.internal.logging.Logging
 import com.onesignal.mocks.MockHelper
 import com.onesignal.user.internal.operations.ExecutorMocks
+import com.onesignal.user.internal.operations.LoginUserOperation
 import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
 import io.mockk.every
 import io.mockk.just
 import io.mockk.mockk
@@ -24,9 +25,11 @@ private class Mocks {
             val mockOperationModelStore = mockk<OperationModelStore>()
             every { mockOperationModelStore.loadOperations() } just runs
             every { mockOperationModelStore.list() } returns listOf()
+            every { mockOperationModelStore.add(any()) } just runs
+            every { mockOperationModelStore.remove(any()) } just runs
             mockOperationModelStore
         }
-    val configModelStore = mockk<ConfigModelStore>()
+    val configModelStore = MockHelper.configModelStore()
     val operationRepo =
         spyk(
             OperationRepo(
@@ -38,36 +41,59 @@ private class Mocks {
             ),
         )
 
-    val recovery = spyk(RecoverFromDroppedLoginBug(operationRepo, MockHelper.identityModelStore(), configModelStore))
+    var oneSignalId = "local-id"
+    val identityModelStore by lazy {
+        MockHelper.identityModelStore {
+            it.onesignalId = oneSignalId
+            it.externalId = "myExtId"
+        }
+    }
+    val recovery = spyk(RecoverFromDroppedLoginBug(operationRepo, identityModelStore, configModelStore))
+
+    val expectedOperation by lazy {
+        LoginUserOperation(
+            configModelStore.model.appId,
+            identityModelStore.model.onesignalId,
+            identityModelStore.model.externalId,
+            null,
+        )
+    }
+
+    fun verifyExpectedLoginOperation(expectedOp: LoginUserOperation = expectedOperation) {
+        verify(exactly = 1) {
+            operationRepo.enqueue(
+                withArg {
+                    (it is LoginUserOperation) shouldBe true
+                    val op = it as LoginUserOperation
+                    op.appId shouldBe expectedOp.appId
+                    op.externalId shouldBe expectedOp.externalId
+                    op.existingOnesignalId shouldBe expectedOp.existingOnesignalId
+                    op.onesignalId shouldBe expectedOp.onesignalId
+                },
+            )
+        }
+    }
 }
 
 class RecoverFromDroppedLoginBugTests : FunSpec({
-    test("ensure onOperationRepoLoaded callback fires from operationRepo") {
+    beforeAny {
+        Logging.logLevel = LogLevel.NONE
+    }
+
+    test("ensure it adds missing operation") {
         // Given
         val mocks = Mocks()
 
         // When
         mocks.recovery.start()
-        val waiter = Waiter()
-        mocks.operationRepo.addOperationLoadedListener(
-            object : IOperationRepoLoadedListener {
-                override fun onOperationRepoLoaded() {
-                    waiter.wake()
-                }
-            },
-        )
         mocks.operationRepo.start()
-        // Waiting here ensures recovery.onOperationRepoLoaded() is called consistently
-        waiter.waitForWake()
+        mocks.operationRepo.awaitInitialized()
 
         // Then
-        verify(exactly = 1) {
-            mocks.operationRepo.subscribe(mocks.recovery)
-            mocks.recovery.onOperationRepoLoaded()
-        }
+        mocks.verifyExpectedLoginOperation()
     }
 
-    test("ensure onOperationRepoLoaded callback fires from operationRepo, even if started first") {
+    test("ensure it adds missing operation, even if operationRepo is already initialized") {
         // Given
         val mocks = Mocks()
 
@@ -77,22 +103,9 @@ class RecoverFromDroppedLoginBugTests : FunSpec({
         delay(200)
 
         mocks.recovery.start()
-
-        val waiter = Waiter()
-        mocks.operationRepo.addOperationLoadedListener(
-            object : IOperationRepoLoadedListener {
-                override fun onOperationRepoLoaded() {
-                    waiter.wake()
-                }
-            },
-        )
-        // Waiting here ensures recovery.onOperationRepoLoaded() is called consistently
-        withTimeout(1_000) { waiter.waitForWake() }
+        withTimeout(1_000) { mocks.operationRepo.awaitInitialized() }
 
         // Then
-        verify(exactly = 1) {
-            mocks.operationRepo.subscribe(mocks.recovery)
-            mocks.recovery.onOperationRepoLoaded()
-        }
+        mocks.verifyExpectedLoginOperation()
     }
 })