fix watched queries

.gitignore

@@ -17,4 +17,5 @@ captures
 !*.xcodeproj/project.xcworkspace/
 !*.xcworkspace/contents.xcworkspacedata
 **/xcshareddata/WorkspaceSettings.xcsettings
-Pods/
+Pods/
+plugins/sonatype

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

@@ -19,6 +19,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.debounce
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import kotlinx.serialization.encodeToString
 
@@ -38,10 +39,17 @@ class PsInternalDatabase(val driver: PsSqlDriver, private val scope: CoroutineSc
 
     init {
         scope.launch {
-            tableUpdates().debounce(DEFAULT_WATCH_THROTTLE_MS).collect { tables ->
-                val dataTables = tables.map { toFriendlyTableName(it) }.filter { it.isNotBlank() }
-                driver.notifyListeners(queryKeys = dataTables.toTypedArray())
-            }
+            val accumulatedUpdates = mutableSetOf<String>();
+            tableUpdates()
+//               Debounce will discard any events which occur inside the debounce window
+//               This will accumulate those table updates
+                .onEach { tables -> accumulatedUpdates.addAll(tables) }
+                .debounce(DEFAULT_WATCH_THROTTLE_MS)
+                .collect {
+                    val dataTables = accumulatedUpdates.map { toFriendlyTableName(it) }.filter { it.isNotBlank() }
+                    driver.notifyListeners(queryKeys = dataTables.toTypedArray());
+                    accumulatedUpdates.clear();
+                }
         }
     }
 

dialect/bin/main/META-INF/services/app.cash.sqldelight.dialect.api.SqlDelightDialect

@@ -0,0 +1 @@
+com.powersync.sqlite.PowerSyncDialect

dialect/bin/main/com/powersync/sqlite/PowerSyncDialect.kt

@@ -0,0 +1,25 @@
+package com.powersync.sqlite
+
+import app.cash.sqldelight.dialect.api.IntermediateType
+import app.cash.sqldelight.dialect.api.PrimitiveType
+import app.cash.sqldelight.dialect.api.SqlDelightDialect
+import app.cash.sqldelight.dialect.api.TypeResolver
+import app.cash.sqldelight.dialects.sqlite_3_35.SqliteTypeResolver
+import app.cash.sqldelight.dialects.sqlite_3_38.SqliteDialect as Sqlite338Dialect
+import com.alecstrong.sql.psi.core.psi.SqlFunctionExpr
+
+class PowerSyncDialect : SqlDelightDialect by Sqlite338Dialect() {
+    override fun typeResolver(parentResolver: TypeResolver) = PowerSyncTypeResolver(parentResolver)
+}
+
+class PowerSyncTypeResolver(private val parentResolver: TypeResolver) :
+    TypeResolver by SqliteTypeResolver(parentResolver) {
+    override fun functionType(functionExpr: SqlFunctionExpr): IntermediateType? {
+        when (functionExpr.functionName.text) {
+            "sqlite_version", "powersync_rs_version", "powersync_replace_schema" -> return IntermediateType(
+                PrimitiveType.TEXT
+            )
+        }
+        return parentResolver.functionType(functionExpr)
+    }
+}