Add opRepoPostCreateDelay

This is a new behavior which wakes a specific amount of time after
something is created (such as a User or Subscription) before make a
network call to get or update it.

The main motivation is that the OneSignal's server may return a 404 if
you attempt to GET or PATCH on something that was just created. This is
due fact that OneSignal's backend server replication sometimes has a
delay under load. This may be considered a bug in the backend, but on
the flip side the SDK is being inefficient if a follow up network
request is made any way, which was the 2nd motivation for this change.

Author: jkasten2

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

@@ -20,6 +20,12 @@ abstract class Operation(name: String) : Model() {
         this.name = name
     }
 
+    /**
+     * This is a unique id that points to a record this operation will affect.
+     * Example: If the operation is updating tags on a User this will be the onesignalId.
+     */
+    abstract val applyToRecordId: String
+
     /**
      * The key of this operation for when the starting operation has a [groupComparisonType]
      * of [GroupComparisonType.CREATE]

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

@@ -1,7 +1,6 @@
 package com.onesignal.core.internal.operations.impl
 
 import com.onesignal.common.threading.WaiterWithValue
-import com.onesignal.common.threading.suspendifyOnThread
 import com.onesignal.core.internal.config.ConfigModelStore
 import com.onesignal.core.internal.operations.ExecutionResult
 import com.onesignal.core.internal.operations.GroupComparisonType
@@ -12,7 +11,11 @@ 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.CoroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.coroutines.withTimeoutOrNull
 import java.util.UUID
 import kotlin.reflect.KClass
@@ -22,6 +25,7 @@ internal class OperationRepo(
     private val _operationModelStore: OperationModelStore,
     private val _configModelStore: ConfigModelStore,
     private val _time: ITime,
+    private val _newRecordState: NewRecordsState,
 ) : IOperationRepo, IStartableService {
     internal class OperationQueueItem(
         val operation: Operation,
@@ -38,6 +42,7 @@ internal class OperationRepo(
     private val queue = mutableListOf<OperationQueueItem>()
     private val waiter = WaiterWithValue<Boolean>()
     private var paused = false
+    private var coroutineScope = CoroutineScope(newSingleThreadContext(name = "OpRepo"))
 
     /** *** Buckets ***
      * Purpose: Bucketing is a pattern we are using to help save network
@@ -56,13 +61,11 @@ internal class OperationRepo(
      *       network calls.
      */
     private var enqueueIntoBucket = 0
-    private val executeBucket: Int get() {
-        return if (enqueueIntoBucket == 0) 0 else enqueueIntoBucket - 1
-    }
+    private val executeBucket get() =
+        if (enqueueIntoBucket == 0) 0 else enqueueIntoBucket - 1
 
     init {
         val executorsMap: MutableMap<String, IOperationExecutor> = mutableMapOf()
-
         for (executor in executors) {
             for (operation in executor.operations) {
                 executorsMap[operation] = executor
@@ -83,9 +86,7 @@ internal class OperationRepo(
 
     override fun start() {
         paused = false
-        suspendifyOnThread(name = "OpRepo") {
-            processQueueForever()
-        }
+        coroutineScope.launch { processQueueForever() }
     }
 
     override fun enqueue(
@@ -195,6 +196,18 @@ internal class OperationRepo(
                 synchronized(queue) {
                     queue.forEach { it.operation.translateIds(response.idTranslations) }
                 }
+                response.idTranslations.values.forEach { _newRecordState.add(it) }
+                coroutineScope.launch {
+                    delay(_configModelStore.model.opRepoPostCreateDelay)
+                    synchronized(queue) {
+                        // NOTE: Even if the queue is not empty we may wake
+                        // when not needed, as those operations may not have
+                        // depended on these ids. This however should be very
+                        // rare and the side-effect is only a bit less
+                        // batching.
+                        if (queue.isNotEmpty()) waiter.wake(false)
+                    }
+                }
             }
 
             when (response.result) {
@@ -278,7 +291,9 @@ internal class OperationRepo(
         return synchronized(queue) {
             val startingOp =
                 queue.firstOrNull {
-                    it.operation.canStartExecute && it.bucket <= bucketFilter
+                    it.operation.canStartExecute &&
+                        _newRecordState.canAccess(it.operation.applyToRecordId) &&
+                        it.bucket <= bucketFilter
                 }
 
             if (startingOp != null) {

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

@@ -25,6 +25,7 @@ import com.onesignal.user.internal.operations.impl.executors.UpdateUserOperation
 import com.onesignal.user.internal.operations.impl.listeners.IdentityModelStoreListener
 import com.onesignal.user.internal.operations.impl.listeners.PropertiesModelStoreListener
 import com.onesignal.user.internal.operations.impl.listeners.SubscriptionModelStoreListener
+import com.onesignal.user.internal.operations.impl.states.NewRecordsState
 import com.onesignal.user.internal.properties.PropertiesModelStore
 import com.onesignal.user.internal.service.UserRefreshService
 import com.onesignal.user.internal.subscriptions.ISubscriptionManager
@@ -68,5 +69,8 @@ internal class UserModule : IModule {
         builder.register<UserRefreshService>().provides<IStartableService>()
 
         builder.register<RecoverFromDroppedLoginBug>().provides<IStartableService>()
+
+        // Shared state between Executors
+        builder.register<NewRecordsState>().provides<NewRecordsState>()
     }
 }

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

@@ -86,6 +86,7 @@ class CreateSubscriptionOperation() : Operation(SubscriptionOperationExecutor.CR
     override val modifyComparisonKey: String get() = "$appId.User.$onesignalId.Subscription.$subscriptionId"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.ALTER
     override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId)
+    override val applyToRecordId: String get() = onesignalId
 
     constructor(appId: String, onesignalId: String, subscriptionId: String, type: SubscriptionType, enabled: Boolean, address: String, status: SubscriptionStatus) : this() {
         this.appId = appId

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

@@ -41,6 +41,7 @@ class DeleteAliasOperation() : Operation(IdentityOperationExecutor.DELETE_ALIAS)
     override val modifyComparisonKey: String get() = "$appId.User.$onesignalId.Alias.$label"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE
     override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId)
+    override val applyToRecordId: String get() = onesignalId
 
     constructor(appId: String, onesignalId: String, label: String) : this() {
         this.appId = appId

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

@@ -42,6 +42,7 @@ class DeleteSubscriptionOperation() : Operation(SubscriptionOperationExecutor.DE
     override val modifyComparisonKey: String get() = "$appId.User.$onesignalId.Subscription.$subscriptionId"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE
     override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(onesignalId)
+    override val applyToRecordId: String get() = subscriptionId
 
     constructor(appId: String, onesignalId: String, subscriptionId: String) : this() {
         this.appId = appId

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

@@ -42,6 +42,7 @@ class DeleteTagOperation() : Operation(UpdateUserOperationExecutor.DELETE_TAG) {
     override val modifyComparisonKey: String get() = "$appId.User.$onesignalId"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.ALTER
     override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId)
+    override val applyToRecordId: String get() = onesignalId
 
     constructor(appId: String, onesignalId: String, key: String) : this() {
         this.appId = appId

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

@@ -40,6 +40,7 @@ class LoginUserFromSubscriptionOperation() : Operation(LoginUserFromSubscription
     override val modifyComparisonKey: String get() = "$appId.Subscription.$subscriptionId.Login"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE
     override val canStartExecute: Boolean = true
+    override val applyToRecordId: String get() = subscriptionId
 
     constructor(appId: String, onesignalId: String, subscriptionId: String) : this() {
         this.appId = appId

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

@@ -56,6 +56,7 @@ class LoginUserOperation() : Operation(LoginUserOperationExecutor.LOGIN_USER) {
     override val modifyComparisonKey: String = ""
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.CREATE
     override val canStartExecute: Boolean get() = existingOnesignalId == null || !IDManager.isLocalId(existingOnesignalId!!)
+    override val applyToRecordId: String get() = onesignalId
 
     constructor(appId: String, onesignalId: String, externalId: String?, existingOneSignalId: String? = null) : this() {
         this.appId = appId

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

@@ -33,6 +33,7 @@ class RefreshUserOperation() : Operation(RefreshUserOperationExecutor.REFRESH_US
     override val modifyComparisonKey: String get() = "$appId.User.$onesignalId.Refresh"
     override val groupComparisonType: GroupComparisonType = GroupComparisonType.CREATE
     override val canStartExecute: Boolean get() = !IDManager.isLocalId(onesignalId)
+    override val applyToRecordId: String get() = onesignalId
 
     constructor(appId: String, onesignalId: String) : this() {
         this.appId = appId