CrudBatch hasMore + onChange improvement (#182)

Author: stevensJourney

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## unreleased
+
+* Fixed `CrudBatch` `hasMore` always returning false.
+* Added `triggerImmediately` to `onChange` method.
+
 ## 1.0.0-BETA32
 
 * Added `onChange` method to the PowerSync client. This allows for observing table changes.

core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt

@@ -187,14 +187,22 @@ class DatabaseTest {
     fun testTableChangesUpdates() =
         databaseTest {
             turbineScope {
-                val query = database.onChange(tables = setOf("users")).testIn(this)
+                val query =
+                    database
+                        .onChange(
+                            tables = setOf("users"),
+                        ).testIn(this)
 
                 database.execute(
                     "INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)",
                     listOf("Test", "test@example.org"),
                 )
 
-                val changeSet = query.awaitItem()
+                var changeSet = query.awaitItem()
+                // The initial result
+                changeSet.count() shouldBe 0
+
+                changeSet = query.awaitItem()
                 changeSet.count() shouldBe 1
                 changeSet.contains("users") shouldBe true
 
@@ -418,4 +426,37 @@ class DatabaseTest {
 
             database.getNextCrudTransaction() shouldBe null
         }
+
+    @Test
+    fun testCrudBatch() =
+        databaseTest {
+            database.execute(
+                "INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)",
+                listOf("a", "a@example.org"),
+            )
+
+            database.writeTransaction {
+                it.execute(
+                    "INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)",
+                    listOf("b", "b@example.org"),
+                )
+                it.execute(
+                    "INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)",
+                    listOf("c", "c@example.org"),
+                )
+            }
+
+            // Purposely limit to less than the number of available ops
+            var batch = database.getCrudBatch(2) ?: error("Batch should not be null")
+            batch.hasMore shouldBe true
+            batch.crud shouldHaveSize 2
+            batch.complete(null)
+
+            batch = database.getCrudBatch(1000) ?: error("Batch should not be null")
+            batch.crud shouldHaveSize 1
+            batch.hasMore shouldBe false
+            batch.complete(null)
+
+            database.getCrudBatch() shouldBe null
+        }
 }

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -258,10 +258,10 @@ internal class PowerSyncDatabaseImpl(
             return null
         }
 
-        val entries =
+        var entries =
             internalDb.getAll(
                 "SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?",
-                listOf(limit.toLong()),
+                listOf(limit.toLong() + 1),
             ) {
                 CrudEntry.fromRow(
                     CrudRow(
@@ -278,7 +278,7 @@ internal class PowerSyncDatabaseImpl(
 
         val hasMore = entries.size > limit
         if (hasMore) {
-            entries.dropLast(entries.size - limit)
+            entries = entries.dropLast(1)
         }
 
         return CrudBatch(entries, hasMore, complete = { writeCheckpoint ->
@@ -351,11 +351,12 @@ internal class PowerSyncDatabaseImpl(
     override fun onChange(
         tables: Set<String>,
         throttleMs: Long,
+        triggerImmediately: Boolean,
     ): Flow<Set<String>> =
         flow {
             waitReady()
             emitAll(
-                internalDb.onChange(tables, throttleMs),
+                internalDb.onChange(tables, throttleMs, triggerImmediately),
             )
         }
 

core/src/commonMain/kotlin/com/powersync/db/Queries.kt

@@ -97,6 +97,7 @@ public interface Queries {
      *
      * @param tables The set of tables to monitor for changes.
      * @param throttleMs The minimum interval, in milliseconds, between emissions. Defaults to [DEFAULT_THROTTLE]. Table changes are accumulated while throttling is active. The accumulated set of tables will be emitted on the trailing edge of the throttle.
+     * @param triggerImmediately If true (default), the flow will immediately emit an empty set of tables when the flow is first collected. This can be useful for ensuring that the flow emits at least once, even if no changes occur to the monitored tables.
      * @return A [Flow] emitting the set of modified tables.
      * @throws PowerSyncException If a database error occurs.
      * @throws CancellationException If the operation is cancelled.
@@ -105,6 +106,7 @@ public interface Queries {
     public fun onChange(
         tables: Set<String>,
         throttleMs: Long = DEFAULT_THROTTLE.inWholeMilliseconds,
+        triggerImmediately: Boolean = true,
     ): Flow<Set<String>>
 
     /**

core/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt

@@ -106,6 +106,7 @@ internal class InternalDatabaseImpl(
     override fun onChange(
         tables: Set<String>,
         throttleMs: Long,
+        triggerImmediately: Boolean,
     ): Flow<Set<String>> =
         channelFlow {
             // Match all possible internal table combinations
@@ -116,7 +117,12 @@ internal class InternalDatabaseImpl(
             val batchedUpdates = AtomicMutableSet<String>()
 
             updatesOnTables()
-                .transform { updates ->
+                .onSubscription {
+                    if (triggerImmediately) {
+                        // Emit an initial event (if requested). No changes would be detected at this point
+                        send(setOf())
+                    }
+                }.transform { updates ->
                     val intersection = updates.intersect(watchedTables)
                     if (intersection.isNotEmpty()) {
                         // Transform table names using friendlyTableName