add ability to throttle watched queries

Author: stevensJourney

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

@@ -139,11 +139,12 @@ internal class PowerSyncDatabaseImpl(
             }
         }
 
-        uploadJob =
+        uploadJob = 
             scope.launch {
-                internalDb.updatesOnTables(setOf(InternalTable.CRUD.toString())).debounce(crudThrottleMs).collect {
-                    syncStream!!.triggerCrudUpload()
-                }
+                internalDb.updatesOnTables(setOf(InternalTable.CRUD.toString()))
+                    .debounce(crudThrottleMs).collect {
+                        syncStream!!.triggerCrudUpload()
+                    }
             }
     }
 
@@ -210,7 +211,8 @@ internal class PowerSyncDatabaseImpl(
         }
     }
 
-    override suspend fun getPowerSyncVersion(): String = internalDb.queries.powerSyncVersion().executeAsOne()
+    override suspend fun getPowerSyncVersion(): String =
+        internalDb.queries.powerSyncVersion().executeAsOne()
 
     override suspend fun <RowType : Any> get(
         sql: String,
@@ -233,12 +235,15 @@ internal class PowerSyncDatabaseImpl(
     override fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>?,
+        throttleMs: Long?,
         mapper: (SqlCursor) -> RowType,
-    ): Flow<List<RowType>> = internalDb.watch(sql, parameters, mapper)
+    ): Flow<List<RowType>> = internalDb.watch(sql, parameters, throttleMs, mapper)
 
-    override suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R = internalDb.writeTransaction(callback)
+    override suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R =
+        internalDb.writeTransaction(callback)
 
-    override suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R = internalDb.writeTransaction(callback)
+    override suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R =
+        internalDb.writeTransaction(callback)
 
     override suspend fun execute(
         sql: String,
@@ -280,7 +285,11 @@ internal class PowerSyncDatabaseImpl(
             syncStream = null
         }
 
-        currentStatus.update(connected = false, connecting = false, lastSyncedAt = currentStatus.lastSyncedAt)
+        currentStatus.update(
+            connected = false,
+            connecting = false,
+            lastSyncedAt = currentStatus.lastSyncedAt
+        )
     }
 
     override suspend fun disconnectAndClear(clearLocal: Boolean) {

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

@@ -54,6 +54,10 @@ public interface Queries {
     public fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>? = listOf(),
+        /**
+         * Specify the minimum interval, in milliseconds, between queries.
+         */
+        throttleMs: Long? = null,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>>
 

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

@@ -51,13 +51,15 @@ internal class InternalDatabaseImpl(
                 sql: String,
                 parameters: List<Any?>?,
                 mapper: (SqlCursor) -> RowType,
-            ): List<RowType> = this@InternalDatabaseImpl.getAllSync(sql, parameters ?: emptyList(), mapper)
+            ): List<RowType> =
+                this@InternalDatabaseImpl.getAllSync(sql, parameters ?: emptyList(), mapper)
 
             override fun <RowType : Any> getOptional(
                 sql: String,
                 parameters: List<Any?>?,
                 mapper: (SqlCursor) -> RowType,
-            ): RowType? = this@InternalDatabaseImpl.getOptionalSync(sql, parameters ?: emptyList(), mapper)
+            ): RowType? =
+                this@InternalDatabaseImpl.getOptionalSync(sql, parameters ?: emptyList(), mapper)
         }
 
     companion object {
@@ -154,6 +156,7 @@ internal class InternalDatabaseImpl(
     override fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>?,
+        throttleMs: Long?,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>> =
         flow {
@@ -165,7 +168,7 @@ internal class InternalDatabaseImpl(
 
             emitAll(
                 updatesOnTables(tables)
-                    .debounce(DEFAULT_WATCH_THROTTLE_MS)
+                    .debounce(throttleMs ?: DEFAULT_WATCH_THROTTLE_MS)
                     .map {
                         getAll(sql, parameters = parameters, mapper = mapper)
                     }.onStart {
@@ -220,7 +223,8 @@ internal class InternalDatabaseImpl(
         }
 
     // Register callback for table updates on a specific table
-    override fun updatesOnTables(tableNames: Set<String>): Flow<Unit> = driver.updatesOnTables(tableNames)
+    override fun updatesOnTables(tableNames: Set<String>): Flow<Unit> =
+        driver.updatesOnTables(tableNames)
 
     private suspend fun getSourceTables(
         sql: String,