recover from a stalled user in the OperationRepo

Commit 7fe2010 fixed a bug with login
operations being drop in some scenarios, however it doesn't fix those
already in the bad state.

This comment get us out of the bad state by generating a replacement
LoginUserOperation. See the comments in the code for more details.

This comment was manually tested, ensuring it gets us out of the bad
state, and updates works afterwords.

Author: jkasten2

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

@@ -1,5 +1,7 @@
 package com.onesignal.core.internal.operations
 
+import kotlin.reflect.KClass
+
 /**
  * The operation queue provides a mechanism to queue one or more [Operation] with the promise
  * it will be executed in a background thread at some point in the future.  Operations are
@@ -31,4 +33,13 @@ interface IOperationRepo {
         operation: Operation,
         flush: Boolean = false,
     ): Boolean
+
+    /**
+     * Check if the queue contains a specific operation type
+     */
+    fun <T : Operation> containsInstanceOf(type: KClass<T>): Boolean
 }
+
+// Extension function so the syntax containsInstanceOf<Operation>() can be used over
+// containsInstanceOf(Operation::class)
+inline fun <reified T : Operation> IOperationRepo.containsInstanceOf(): Boolean = containsInstanceOf(T::class)

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

@@ -15,6 +15,7 @@ import com.onesignal.debug.internal.logging.Logging
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.UUID
+import kotlin.reflect.KClass
 
 internal class OperationRepo(
     executors: List<IOperationExecutor>,
@@ -52,6 +53,12 @@ internal class OperationRepo(
         }
     }
 
+    override fun <T : Operation> containsInstanceOf(type: KClass<T>): Boolean {
+        synchronized(queue) {
+            return queue.any { type.isInstance(it.operation) }
+        }
+    }
+
     override fun start() {
         paused = false
         suspendifyOnThread(name = "OpRepo") {

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/user/UserModule.kt

@@ -15,6 +15,7 @@ import com.onesignal.user.internal.backend.impl.UserBackendService
 import com.onesignal.user.internal.builduser.IRebuildUserService
 import com.onesignal.user.internal.builduser.impl.RebuildUserService
 import com.onesignal.user.internal.identity.IdentityModelStore
+import com.onesignal.user.internal.migrations.RecoverFromDroppedLoginBug
 import com.onesignal.user.internal.operations.impl.executors.IdentityOperationExecutor
 import com.onesignal.user.internal.operations.impl.executors.LoginUserFromSubscriptionOperationExecutor
 import com.onesignal.user.internal.operations.impl.executors.LoginUserOperationExecutor
@@ -65,5 +66,7 @@ internal class UserModule : IModule {
         builder.register<UserManager>().provides<IUserManager>()
 
         builder.register<UserRefreshService>().provides<IStartableService>()
+
+        builder.register<RecoverFromDroppedLoginBug>().provides<IStartableService>()
     }
 }

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

@@ -0,0 +1,85 @@
+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.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
+
+/**
+ * Purpose: Automatically recovers a stalled User in the OperationRepo due
+ * to a bug in the SDK from 5.0.0 to 5.1.7.
+ *
+ * Issue: Some calls to OneSignal.login() would not be reflected on the
+ * backend and would stall the the queue for that User. This would result
+ * in User and Subscription operations to not be processed by
+ * OperationRepo.
+ * See PR #2046 for more details.
+ *
+ * Even if the developer called OneSignal.login() again with the same
+ * externalId it would not correct the stalled User.
+ *   - Only calling logout() or login() with different externalId would
+ *     have un-stalled the OperationRepo. And then only after logging
+ *     back to the stalled user would it have recover all the unsent
+ *     operations they may exist.
+ */
+class RecoverFromDroppedLoginBug(
+    private val _operationRepo: IOperationRepo,
+    private val _identityModelStore: IdentityModelStore,
+    private val _configModelStore: ConfigModelStore,
+) : IStartableService {
+    override fun start() {
+        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()
+        }
+    }
+
+    // We are in the bad state if ALL are true:
+    // 1. externalId is set (because OneSignal.login was called)
+    // 2. We don't have a real yet onesignalId
+    //   - We haven't made a successful user create call yet.
+    // 3. There is no attempt to create the User left in the
+    //    OperationRepo's queue.
+    private fun isInBadState(): Boolean {
+        val externalId = _identityModelStore.model.externalId
+        val onesignalId = _identityModelStore.model.onesignalId
+
+        // NOTE: We are not accounting a more rare (and less important case)
+        // where a previously logged in User was never created.
+        // That is, if another user already logged in successfully, but
+        // the last user still has stuck pending operations due to the
+        // User never being created on the OneSignal's backend.
+        return externalId != null &&
+            IDManager.isLocalId(onesignalId) &&
+            !_operationRepo.containsInstanceOf<LoginUserOperation>()
+    }
+
+    private fun recoverByAddingBackDroppedLoginOperation() {
+        // This is the operation that was dropped by mistake in some cases,
+        // once it is added to the queue all and it gets executed, all
+        // operations waiting on it will be sent.
+
+        // This enqueues at the end, however this is ok, since
+        // the OperationRepo is designed find the first operation that is
+        // executable.
+        _operationRepo.enqueue(
+            LoginUserOperation(
+                _configModelStore.model.appId,
+                _identityModelStore.model.onesignalId,
+                _identityModelStore.model.externalId,
+                null,
+            ),
+            true,
+        )
+    }
+}

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

@@ -62,6 +62,28 @@ class OperationRepoTests : FunSpec({
         Logging.logLevel = LogLevel.NONE
     }
 
+    test("containsInstanceOf") {
+        // Given
+        val operationRepo = Mocks().operationRepo
+
+        open class MyOperation : Operation("MyOp") {
+            override val createComparisonKey = ""
+            override val modifyComparisonKey = ""
+            override val groupComparisonType = GroupComparisonType.NONE
+            override val canStartExecute = false
+        }
+
+        class MyOperation2 : MyOperation()
+
+        // When
+        operationRepo.start()
+        operationRepo.enqueue(MyOperation())
+
+        // Then
+        operationRepo.containsInstanceOf<MyOperation>() shouldBe true
+        operationRepo.containsInstanceOf<MyOperation2>() shouldBe false
+    }
+
     // Ensures we are not continuously waking the CPU
     test("ensure processQueueForever suspends when queue is empty") {
         // Given