ConsistencyManager interface & implementation

Manages read-your-write tokens. The manager works based on conditions & tokens.

Tokens are stored in a nested map indexed by a unique id (e.g. `onesignalId`) and a token key (e.g. `USER`).

This allows us to track tokens on a per-user basis (e.g. handle switching users).

Conditions work by creating a blocking mechanism with customizable token retrieval until a pre-defined condition is met (e.g. at least two specific tokens are available).

Also allows extensibility for future applications to control offset blocking mechanism in consistency use-cases.

Author: Rodrigo Gomez Palacio

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/consistency/impl/ConsistencyManager.kt

@@ -0,0 +1,92 @@
+package com.onesignal.common.consistency.impl
+
+import com.onesignal.common.consistency.models.ICondition
+import com.onesignal.common.consistency.models.IConsistencyKeyEnum
+import com.onesignal.common.consistency.models.IConsistencyManager
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/**
+ * Manages read-your-write tokens for more accurate segment membership
+ * calculation. Uses customizable conditions that block retrieval of the newest token until met.
+ *
+ * Usage:
+ *  val consistencyManager = ConsistencyManager<MyEnum>()
+ *  val updateConditionDeferred = consistencyManager.registerCondition(MyCustomCondition())
+ *  val rywToken = updateConditionDeferred.await()
+ */
+class ConsistencyManager : IConsistencyManager {
+    private val mutex = Mutex()
+    private val indexedTokens: MutableMap<String, MutableMap<IConsistencyKeyEnum, String>> = mutableMapOf()
+    private val conditions: MutableList<Pair<ICondition, CompletableDeferred<String?>>> =
+        mutableListOf()
+
+    /**
+     * Set method to update the token based on the key.
+     *  Params:
+     *      id: String - the index of the token map (e.g. onesignalId)
+     *      key: K - corresponds to the operation for which we have a read-your-write token
+     *      value: String? - the token (read-your-write token)
+     */
+    override suspend fun setRywToken(
+        id: String,
+        key: IConsistencyKeyEnum,
+        value: String,
+    ) {
+        mutex.withLock {
+            val rywTokens = indexedTokens.getOrPut(id) { mutableMapOf() }
+            rywTokens[key] = value
+            checkConditionsAndComplete()
+        }
+    }
+
+    /**
+     * Register a condition with its corresponding deferred action. Returns a deferred condition.
+     */
+    override suspend fun registerCondition(condition: ICondition): CompletableDeferred<String?> {
+        mutex.withLock {
+            val deferred = CompletableDeferred<String?>()
+            val pair = Pair(condition, deferred)
+            conditions.add(pair)
+            checkConditionsAndComplete()
+            return deferred
+        }
+    }
+
+    override suspend fun resolveConditionsWithID(id: String) {
+        val completedConditions = mutableListOf<Pair<ICondition, CompletableDeferred<String?>>>()
+
+        for ((condition, deferred) in conditions) {
+            if (condition.id == id) {
+                if (!deferred.isCompleted) {
+                    deferred.complete(null)
+                }
+            }
+            completedConditions.add(Pair(condition, deferred))
+        }
+
+        // Remove completed conditions from the list
+        conditions.removeAll(completedConditions)
+    }
+
+    /**
+     * IMPORTANT: calling code should be protected by mutex to avoid potential inconsistencies
+     */
+    private fun checkConditionsAndComplete() {
+        val completedConditions = mutableListOf<Pair<ICondition, CompletableDeferred<String?>>>()
+
+        for ((condition, deferred) in conditions) {
+            if (condition.isMet(indexedTokens)) {
+                val newestToken = condition.getNewestToken(indexedTokens)
+                if (!deferred.isCompleted) {
+                    deferred.complete(newestToken)
+                }
+                completedConditions.add(Pair(condition, deferred))
+            }
+        }
+
+        // Remove completed conditions from the list
+        conditions.removeAll(completedConditions)
+    }
+}

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/consistency/models/ICondition.kt

@@ -0,0 +1,21 @@
+package com.onesignal.common.consistency.models
+
+interface ICondition {
+    /**
+     * Every implementation should define a unique ID & make available via a companion object for
+     * ease of use
+     */
+    val id: String
+
+    /**
+     * Define a condition that "unblocks" execution
+     * e.g. we have token (A && B) || A
+     */
+    fun isMet(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String>>): Boolean
+
+    /**
+     * Used to process tokens according to their format & return the newest token.
+     * e.g. numeric strings would be compared differently from JWT tokens
+     */
+    fun getNewestToken(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String?>>): String?
+}

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/consistency/models/IConsistencyKeyEnum.kt

@@ -0,0 +1,3 @@
+package com.onesignal.common.consistency.models
+
+interface IConsistencyKeyEnum

OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/consistency/models/IConsistencyManager.kt

@@ -0,0 +1,31 @@
+package com.onesignal.common.consistency.models
+
+import kotlinx.coroutines.CompletableDeferred
+
+interface IConsistencyManager {
+    /**
+     * Set method to update the RYW token based on the key.
+     * Params:
+     *  id: String - the index of the RYW token map (e.g., onesignalId)
+     *  key: IConsistencyKeyEnum - corresponds to the operation for which we have a read-your-write token
+     *  value: String? - the read-your-write token
+     */
+    suspend fun setRywToken(
+        id: String,
+        key: IConsistencyKeyEnum,
+        value: String,
+    )
+
+    /**
+     * Register a condition with its corresponding deferred action. Returns a deferred condition.
+     * Params:
+     *  condition: ICondition - the condition to be registered
+     * Returns: CompletableDeferred<String?> - a deferred action that completes when the condition is met
+     */
+    suspend fun registerCondition(condition: ICondition): CompletableDeferred<String?>
+
+    /**
+     * Resolve all conditions with a specific ID
+     */
+    suspend fun resolveConditionsWithID(id: String)
+}

OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/consistency/ConsistencyManagerTests.kt

@@ -0,0 +1,111 @@
+package com.onesignal.common.consistency.impl
+
+import com.onesignal.common.consistency.enums.IamFetchRywTokenKey
+import com.onesignal.common.consistency.models.ICondition
+import com.onesignal.common.consistency.models.IConsistencyKeyEnum
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.test.runTest
+
+class ConsistencyManagerTests : FunSpec({
+
+    lateinit var consistencyManager: ConsistencyManager
+
+    beforeAny {
+        consistencyManager = ConsistencyManager()
+    }
+
+    test("setRywToken updates the token correctly") {
+        runTest {
+            // Given
+            val id = "test_id"
+            val key = IamFetchRywTokenKey.USER
+            val value = "123"
+
+            consistencyManager.setRywToken(id, key, value)
+
+            val condition = TestMetCondition(mapOf(id to mapOf(key to value)))
+            val deferred = consistencyManager.registerCondition(condition)
+            val result = deferred.await()
+
+            result shouldBe value
+        }
+    }
+
+    test("registerCondition completes when condition is met") {
+        runTest {
+            // Given
+            val id = "test_id"
+            val key = IamFetchRywTokenKey.USER
+            val value = "123"
+
+            // Set a token to meet the condition
+            consistencyManager.setRywToken(id, key, value)
+
+            val condition = TestMetCondition(mapOf(id to mapOf(key to value)))
+            val deferred = consistencyManager.registerCondition(condition)
+
+            deferred.await()
+            deferred.isCompleted shouldBe true
+        }
+    }
+
+    test("registerCondition does not complete when condition is not met") {
+        runTest {
+            val condition = TestUnmetCondition()
+            val deferred = consistencyManager.registerCondition(condition)
+
+            consistencyManager.setRywToken("id", IamFetchRywTokenKey.USER, "123")
+            deferred.isCompleted shouldBe false
+        }
+    }
+
+    test("resolveConditionsWithID resolves conditions based on ID") {
+        runTest {
+            val condition = TestUnmetCondition()
+            val deferred = consistencyManager.registerCondition(condition)
+            consistencyManager.resolveConditionsWithID(TestUnmetCondition.ID)
+            deferred.await()
+
+            deferred.isCompleted shouldBe true
+        }
+    }
+}) {
+    // Mock implementation of ICondition that simulates a condition that isn't met
+    private class TestUnmetCondition : ICondition {
+        companion object {
+            const val ID = "TestUnmetCondition"
+        }
+
+        override val id: String
+            get() = ID
+
+        override fun isMet(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String>>): Boolean {
+            return false // Always returns false to simulate an unmet condition
+        }
+
+        override fun getNewestToken(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String?>>): String? {
+            return null
+        }
+    }
+
+    // Mock implementation of ICondition for cases where the condition is met
+    private class TestMetCondition(
+        private val expectedRywTokens: Map<String, Map<IConsistencyKeyEnum, String?>>,
+    ) : ICondition {
+        companion object {
+            const val ID = "TestMetCondition"
+        }
+
+        override val id: String
+            get() = ID
+
+        override fun isMet(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String>>): Boolean {
+            return indexedTokens == expectedRywTokens
+        }
+
+        override fun getNewestToken(indexedTokens: Map<String, Map<IConsistencyKeyEnum, String?>>): String? {
+            return expectedRywTokens.values.firstOrNull()?.values?.firstOrNull()
+        }
+    }
+}