update watched query implementation

Author: stevensJourney

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+# 1.0.0-Beta.14
+
+- Removed references to the PowerSync Kotlin SDK from all public API protocols.
+- Improved the stability of watched queries. Watched queries were previously susceptible to runtime crashes if an exception was thrown in the update stream. Errors are now gracefully handled.
+
 # 1.0.0-Beta.13
 
 - Update `powersync-kotlin` dependency to version `1.0.0-BETA32`, which includes:

Package.resolved

@@ -17,7 +17,8 @@ let package = Package(
             targets: ["PowerSync"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/powersync-ja/powersync-kotlin.git", exact: "1.0.0-BETA32.0"),
+       // .package(url: "https://github.com/powersync-ja/powersync-kotlin.git", exact: "1.0.0-BETA32.0"),
+        .package(path: "/Users/stevenontong/Documents/platform_code/powersync/powersync-kotlin"),
         .package(url: "https://github.com/powersync-ja/powersync-sqlite-core-swift.git", "0.3.12"..<"0.4.0")
     ],
     targets: [

Package.swift

@@ -5,6 +5,7 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
     let logger: any LoggerProtocol
 
     private let kotlinDatabase: PowerSyncKotlin.PowerSyncDatabase
+    private let encoder = JSONEncoder()
     let currentStatus: SyncStatus
 
     init(
@@ -221,43 +222,30 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             // Create an outer task to monitor cancellation
             let task = Task {
                 do {
-                    var mapperError: Error?
-                    // HACK!
-                    // SKIEE doesn't support custom exceptions in Flows
-                    // Exceptions which occur in the Flow itself cause runtime crashes.
-                    // The most probable crash would be the internal EXPLAIN statement.
-                    // This attempts to EXPLAIN the query before passing it to Kotlin
-                    // We could introduce an onChange API in Kotlin which we use to implement watches here.
-                    // This would prevent most issues with exceptions.
-                    // EXPLAIN statement to prevent crashes in SKIEE
-                    _ = try await self.kotlinDatabase.getAll(
-                        sql: "EXPLAIN \(options.sql)",
-                        parameters: mapParameters(options.parameters),
-                        mapper: { _ in "" }
+                    let watchedTables = try await self.getQuerySourceTables(
+                        sql: options.sql,
+                        parameters: options.parameters
                     )
 
                     // Watching for changes in the database
-                    for try await values in try self.kotlinDatabase.watch(
-                        sql: options.sql,
-                        parameters: mapParameters(options.parameters),
+                    for try await _ in try self.kotlinDatabase.onChange(
+                        tables: Set(watchedTables),
                         throttleMs: options.throttleMs,
-                        mapper: { cursor in
-                            do {
-                                return try options.mapper(KotlinSqlCursor(base: cursor))
-                            } catch {
-                                mapperError = error
-                                return ()
-                            }
-                        }
+                        triggerImmediately: true // Allows emitting the first result even if there aren't changes
                     ) {
                         // Check if the outer task is cancelled
-                        try Task.checkCancellation() // This checks if the calling task was cancelled
-
-                        if mapperError != nil {
-                            throw mapperError!
-                        }
-
-                        try continuation.yield(safeCast(values, to: [RowType].self))
+                        try Task.checkCancellation()
+
+                        try continuation.yield(
+                            safeCast(
+                                await self.getAll(
+                                    sql: options.sql,
+                                    parameters: options.parameters,
+                                    mapper: options.mapper
+                                ),
+                                to: [RowType].self
+                            )
+                        )
                     }
 
                     continuation.finish()
@@ -352,13 +340,66 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             return try await handler()
         } catch {
             // Try and parse errors back from the Kotlin side
-
             if let mapperError = SqlCursorError.fromDescription(error.localizedDescription) {
                 throw mapperError
             }
 
-            // Throw remaining errors as-is
-            throw error
+            throw PowerSyncError.operationFailed(
+                underlyingError: error
+            )
+        }
+    }
+
+    private func getQuerySourceTables(
+        sql: String,
+        parameters: [Any?]
+    ) async throws -> Set<String> {
+        let rows = try await getAll(
+            sql: "EXPLAIN \(sql)",
+            parameters: parameters,
+            mapper: { cursor in
+                try ExplainQueryResult(
+                    addr: cursor.getString(index: 0),
+                    opcode: cursor.getString(index: 1),
+                    p1: cursor.getInt64(index: 2),
+                    p2: cursor.getInt64(index: 3),
+                    p3: cursor.getInt64(index: 4)
+                )
+            }
+        )
+
+        let rootPages = rows.compactMap { r in
+            if (r.opcode == "OpenRead" || r.opcode == "OpenWrite") &&
+                r.p3 == 0 && r.p2 != 0
+            {
+                return r.p2
+            }
+            return nil
+        }
+
+        do {
+            let pagesData = try encoder.encode(rootPages)
+            let tableRows = try await getAll(
+                sql: "SELECT tbl_name FROM sqlite_master WHERE rootpage IN (SELECT json_each.value FROM json_each(?))",
+                parameters: [
+                    String(data: pagesData, encoding: .utf8)
+                ]
+            ) { try $0.getString(index: 0) }
+
+            return Set(tableRows)
+        } catch {
+            throw PowerSyncError.operationFailed(
+                message: "Could not determine watched query tables",
+                underlyingError: error
+            )
         }
     }
 }
+
+private struct ExplainQueryResult {
+    let addr: String
+    let opcode: String
+    let p1: Int64
+    let p2: Int64
+    let p3: Int64
+}

Sources/PowerSync/Kotlin/KotlinPowerSyncDatabaseImpl.swift

@@ -0,0 +1,28 @@
+import Foundation
+
+/// Enum representing errors that can occur in the PowerSync system.
+public enum PowerSyncError: Error, LocalizedError {
+    
+    /// Represents a failure in an operation, potentially with a custom message and an underlying error.
+    case operationFailed(message: String? = nil, underlyingError: Error? = nil)
+    
+    /// A localized description of the error, providing details about the failure.
+    public var errorDescription: String? {
+        switch self {
+        case let .operationFailed(message, underlyingError):
+            // Combine message and underlying error description if both are available
+            if let message = message, let underlyingError = underlyingError {
+                return "\(message): \(underlyingError.localizedDescription)"
+            } else if let message = message {
+                // Return only the message if no underlying error is available
+                return message
+            } else if let underlyingError = underlyingError {
+                // Return only the underlying error description if no message is provided
+                return underlyingError.localizedDescription
+            } else {
+                // Fallback to a generic error description if neither message nor underlying error is provided
+                return "An unknown error occurred."
+            }
+        }
+    }
+}

Sources/PowerSync/PowerSyncError.swift

@@ -4,9 +4,7 @@ import Foundation
 
 /// A transaction of client-side changes.
 public protocol CrudBatch {
-    /// Unique transaction id.
-    ///
-    /// If nil, this contains a list of changes recorded without an explicit transaction associated.
+    /// Indicates if there are additional Crud items in the queue which are not included in this batch
     var hasMore: Bool { get }
 
     /// List of client-side changes.

Sources/PowerSync/protocol/db/CrudBatch.swift

@@ -69,16 +69,25 @@ final class CrudTests: XCTestCase {
         XCTAssert(fullBatch.hasMore == false)
         XCTAssert(fullBatch.crud.count == 100)
 
-        guard let txBatch = try await database.getNextCrudTransaction() else {
+        guard let nextTx = try await database.getNextCrudTransaction() else {
             return XCTFail("Failed to get transaction crud batch")
         }
 
-        XCTAssert(txBatch.crud.count == 100)
+        XCTAssert(nextTx.crud.count == 100)
+        
+        for r in nextTx.crud {
+            print(r)
+        }
 
         // Completing the transaction should clear the items
-        try await txBatch.complete()
+        try await nextTx.complete()
 
         let afterCompleteBatch = try await database.getNextCrudTransaction()
+        
+        for r in afterCompleteBatch?.crud ?? [] {
+            print(r)
+        }
+        
         XCTAssertNil(afterCompleteBatch)
     }
 }

Tests/PowerSyncTests/CrudTests.swift

@@ -96,7 +96,7 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
             sql: "SELECT name FROM users WHERE id = ?",
             parameters: ["999"]
         ) { cursor in
-            try cursor.getString(name: "")
+            try cursor.getString(name: "name")
         }
 
         XCTAssertNil(nonExistent)