added tests for upserts to make sure only one emission

gradle/libs.versions.toml

@@ -8,6 +8,7 @@ kotlinx-serialization = { require = "1.7.3" }
 kotlinx-datetime = "0.6.1"
 paging = "3.3.0-alpha02-0.5.1"
 sqlDelight = "2.0.2"
+turbine = "1.2.0"
 
 [libraries]
 androidx-test-monitor = { module = "androidx.test:monitor", version.ref = "androidx-monitor" }
@@ -24,9 +25,10 @@ sqlDelight-driver-sqlite = { module = "app.cash.sqldelight:sqlite-driver", versi
 sqlDelight-driver-android = { module = "app.cash.sqldelight:android-driver", version.ref = "sqlDelight" }
 sqlDelight-coroutines = { module = "app.cash.sqldelight:coroutines-extensions", version.ref = "sqlDelight" }
 sqlite-requery-android = { module = "com.github.requery:sqlite-android", version = "3.45.0" }
-
 sqlDelight-driver-native = { module = "app.cash.sqldelight:native-driver", version.ref = "sqlDelight" }
 sqlDelight-driver-js = { module = "app.cash.sqldelight:web-worker-driver", version.ref = "sqlDelight" }
+turbine = { module = "app.cash.turbine:turbine", version.ref = "turbine" }
+
 
 [plugins]
 android-application = { id = "com.android.application", version.ref = "agp" }

library/build.gradle.kts

@@ -32,6 +32,7 @@ kotlin {
             implementation(libs.kotlinx.coroutines.test)
             implementation(kotlin("test"))
             implementation(libs.paging.testing)
+            implementation(libs.turbine)
         }
 
         androidMain.dependencies {

library/src/commonMain/kotlin/com/mercury/sqkon/db/EntityQueries.kt

@@ -5,11 +5,17 @@ import app.cash.sqldelight.SuspendingTransacterImpl
 import app.cash.sqldelight.db.QueryResult
 import app.cash.sqldelight.db.SqlCursor
 import app.cash.sqldelight.db.SqlDriver
+import kotlinx.coroutines.delay
+import org.jetbrains.annotations.VisibleForTesting
 
 class EntityQueries(
     driver: SqlDriver,
 ) : SuspendingTransacterImpl(driver) {
 
+    // Used to slow down insert/updates for testing
+    @VisibleForTesting
+    internal var slowWrite: Boolean = false
+
     suspend fun insertEntity(entity: Entity, ignoreIfExists: Boolean) {
         val identifier = identifier("insert", ignoreIfExists.toString())
         val orIgnore = if (ignoreIfExists) "OR IGNORE" else ""
@@ -32,6 +38,7 @@ class EntityQueries(
             emit("entity")
             emit("entity_${entity.entity_name}")
         }
+        if(slowWrite) delay(100)
     }
 
     suspend fun updateEntity(
@@ -59,6 +66,7 @@ class EntityQueries(
             emit("entity")
             emit("entity_${entityName}")
         }
+        if(slowWrite) delay(100)
     }
 
     fun select(
@@ -227,11 +235,11 @@ class EntityQueries(
         mapper: (SqlCursor) -> T,
     ) : Query<T>(mapper) {
 
-        override fun addListener(listener: Query.Listener) {
+        override fun addListener(listener: Listener) {
             driver.addListener("entity_$entityName", listener = listener)
         }
 
-        override fun removeListener(listener: Query.Listener) {
+        override fun removeListener(listener: Listener) {
             driver.removeListener("entity_$entityName", listener = listener)
         }
 

library/src/commonMain/kotlin/com/mercury/sqkon/db/KeyValueStorage.kt

@@ -5,7 +5,6 @@ import app.cash.sqldelight.SuspendingTransacter
 import app.cash.sqldelight.coroutines.asFlow
 import app.cash.sqldelight.coroutines.mapToList
 import app.cash.sqldelight.coroutines.mapToOne
-import app.cash.sqldelight.coroutines.mapToOneOrNull
 import com.mercury.sqkon.db.KeyValueStorage.Config.DeserializePolicy
 import com.mercury.sqkon.db.paging.OffsetQueryPagingSource
 import com.mercury.sqkon.db.serialization.KotlinSqkonSerializer
@@ -15,6 +14,7 @@ import com.mercury.sqkon.db.utils.nowMillis
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
@@ -110,11 +110,9 @@ open class KeyValueStorage<T : Any>(
      * @see insert
      * @see update
      */
-    suspend fun upsert(key: String, value: T) {
-        transaction {
-            update(key, value)
-            insert(key, value, ignoreIfExists = true)
-        }
+    suspend fun upsert(key: String, value: T) = transaction {
+        update(key, value)
+        insert(key, value, ignoreIfExists = true)
     }
 
     /**
@@ -125,12 +123,10 @@ open class KeyValueStorage<T : Any>(
      * @see insertAll
      * @see updateAll
      */
-    suspend fun upsertAll(values: Map<String, T>) {
-        transaction {
-            values.forEach { (key, value) ->
-                update(key, value)
-                insert(key, value, ignoreIfExists = true)
-            }
+    suspend fun upsertAll(values: Map<String, T>) = transaction {
+        values.forEach { (key, value) ->
+            update(key, value)
+            insert(key, value, ignoreIfExists = true)
         }
     }
 

library/src/commonTest/kotlin/com/mercury/sqkon/db/KeyValueStorageTest.kt

@@ -1,16 +1,21 @@
 package com.mercury.sqkon.db
 
+import app.cash.turbine.test
+import app.cash.turbine.turbineScope
 import com.mercury.sqkon.TestObject
 import com.mercury.sqkon.TestObjectChild
+import com.mercury.sqkon.TestValue
 import com.mercury.sqkon.until
 import kotlinx.coroutines.MainScope
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import kotlinx.datetime.Clock
 import org.junit.After
 import org.junit.Test
+import kotlin.test.BeforeTest
 import kotlin.test.assertEquals
 import kotlin.test.assertNull
 import kotlin.time.Duration.Companion.seconds
@@ -64,6 +69,23 @@ class KeyValueStorageTest {
         assertEquals(updated, actualUpdated)
     }
 
+    @Test
+    fun upsert() = runTest {
+        val inserted = TestObject()
+        testObjectStorage.upsert(inserted.id, inserted)
+        val actualInserted = testObjectStorage.selectAll().first().first()
+        assertEquals(inserted, actualInserted)
+        val updated = inserted.copy(
+            name = "Updated Name",
+            value = 12345,
+            description = "Updated Description",
+            child = inserted.child.copy(updatedAt = Clock.System.now())
+        )
+        testObjectStorage.upsert(updated.id, updated)
+        val actualUpdated = testObjectStorage.selectAll().first().first()
+        assertEquals(updated, actualUpdated)
+    }
+
     @Test
     fun selectAll_orderBy_EntityName() = runTest {
         val expected = (0..10).map { TestObject() }
@@ -428,5 +450,60 @@ class KeyValueStorageTest {
         assertEquals(expected = 0, results[2])
     }
 
+    @Test
+    fun selectCount_flowUpdatesOnUpsertOnce() = runTest {
+        entityQueries.slowWrite = true
+        val to1 = TestObject().also { testObjectStorage.upsert(it.id, it) }
+        testObjectStorage.selectAll(
+            orderBy = listOf(OrderBy(TestObject::child.then(TestObjectChild::createdAt)))
+        ).test {
+            assertEquals(listOf(to1), awaitItem())
+            // Insert new item
+            val to = TestObject()
+            testObjectStorage.upsert(to.id, to)
+            awaitItem().also {
+                assertEquals(2, it.size)
+                assertEquals(listOf(to1, to), it)
+            }
+            // Insert new item
+            TestObject().also { testObjectStorage.upsert(it.id, it) }
+            awaitItem()
+            // Update existing item
+            val new = to.copy(
+                name = "Updated Name",
+                value = 12345,
+                description = "Updated Description",
+                child = to.child.copy(updatedAt = Clock.System.now())
+            )
+            testObjectStorage.upsert(new.id, new)
+            awaitItem()
+            // Check only one update (not two) for the upsert
+            expectNoEvents()
+        }
+    }
+
+    @Test
+    fun selectCount_flowUpdatesOnUpsertAllOnce() = runTest {
+        entityQueries.slowWrite = true
+        val to1 = TestObject().also { testObjectStorage.upsert(it.id, it) }
+        val to2 = TestObject().also { testObjectStorage.upsert(it.id, it) }
+        testObjectStorage.selectAll(
+            orderBy = listOf(OrderBy(TestObject::child.then(TestObjectChild::createdAt)))
+        ).test {
+            assertEquals(listOf(to1, to2), awaitItem())
+            // Insert new item
+            val to3 = TestObject()
+            testObjectStorage.upsertAll(listOf(to3).associateBy { it.id })
+            awaitItem().also {
+                assertEquals(3, it.size)
+                assertEquals(listOf(to1, to2, to3), it)
+            }
+            expectNoEvents()
+            testObjectStorage.upsertAll(listOf(to1, to2, to3).associateBy { it.id })
+            awaitItem() // should only emit once for all the upserts
+            expectNoEvents()
+        }
+    }
+
 
 }