Merge pull request #8 from powersync-ja/fix-watched-queries

[Fix] Watched queries + Schema indexes

.gitignore

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

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();
+                }
         }
     }
 

core/src/commonMain/kotlin/com/powersync/db/schema/IndexedColumn.kt

@@ -1,27 +1,45 @@
 package com.powersync.db.schema
 
+import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 
 /**
  * Describes an indexed column.
  */
 @Serializable
-data class IndexedColumn (
+data class IndexedColumn(
     /**
      * Name of the column to index.
      */
+    @SerialName("name")
     val column: String,
 
     /**
      * Whether this column is stored in ascending order in the index.
      */
-    private val ascending: Boolean = true
+    private val ascending: Boolean = true,
+
+    private var columnDefinition: Column? = null,
+
+    /**
+     * The column definition type
+     */
+    var type: ColumnType? = null
 ) {
     companion object {
         fun ascending(column: String) = IndexedColumn(column, true)
         fun descending(column: String) = IndexedColumn(column, false)
     }
 
+    /**
+     * Sets the parent column definition. The column definition's type
+     * is required for the serialized JSON payload of powersync_replace_schema
+     */
+    fun setColumnDefinition(column: Column) {
+        this.type = column.type;
+        this.columnDefinition = column;
+    }
+
     fun toSql(table: Table): String {
         val fullColumn = table[column] // errors if not found
         return fullColumn.let {

core/src/commonMain/kotlin/com/powersync/db/schema/Table.kt

@@ -2,13 +2,14 @@ package com.powersync.db.schema
 
 import com.powersync.invalidSqliteCharacters
 import kotlinx.serialization.Serializable
+import kotlin.math.log
 
 /**
  * A single table in the schema.
  */
 @Serializable
 
-data class Table  constructor(
+data class Table constructor(
     /**
      * The synced table name, matching sync rules.
      */
@@ -34,6 +35,21 @@ data class Table  constructor(
      */
     private val _viewNameOverride: String? = null
 ) {
+
+    init {
+        /**
+         * Need to set the column definition for each index column.
+         * This is required for serialization
+         */
+        indexes.forEach { index ->
+            index.columns.forEach {
+                val matchingColumn = columns.find { c -> c.name == it.column }
+                    ?: throw AssertionError("Could not find column definition for index ${index.name}:${it.column}")
+                it.setColumnDefinition(column = matchingColumn)
+            }
+        }
+    }
+
     companion object {
         /**
          * Create a table that only exists locally.

demos/hello-powersync/composeApp/src/commonMain/kotlin/com/powersync/demos/AppSchema.kt

@@ -1,16 +1,21 @@
 package com.powersync.demos
 
 import com.powersync.db.schema.Column
+import com.powersync.db.schema.Index
+import com.powersync.db.schema.IndexedColumn
 import com.powersync.db.schema.Schema
 import com.powersync.db.schema.Table
 
 val AppSchema: Schema = Schema(
     listOf(
         Table(
-            "customers",
-            listOf(
+            name = "customers",
+            columns = listOf(
                 Column.text("name"),
                 Column.text("email")
+            ),
+            indexes = listOf(
+                Index("name", listOf(IndexedColumn.descending("name")))
             )
         )
     )

gradle.properties

@@ -17,7 +17,7 @@ development=true
 RELEASE_SIGNING_ENABLED=true
 # Library config
 GROUP=com.powersync
-LIBRARY_VERSION=0.0.1-ALPHA4
+LIBRARY_VERSION=0.0.1-ALPHA5
 GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
 # POM
 POM_URL=https://github.com/powersync-ja/powersync-kotlin/