chore: make execute, watch and transaction functions throwable in swift (#21)

* chore: make execute and watch throwable in swift

* chore: add transaction and watch error handling

* chore: update package

Author: DominicGBauer

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## 1.0.0-Beta.6
+
+* BREAKING CHANGE: `watch` queries are now throwable and therefore will need to be accompanied by a `try` e.g.
+
+```swift
+try database.watch()
+```
+
+* BREAKING CHANGE: `transaction` functions are now throwable and therefore will need to be accompanied by a `try` e.g.
+
+```swift
+try await database.writeTransaction { transaction in
+  try transaction.execute(...)
+}
+```
+* Allow `execute` errors to be handled
+* `userId` is now set to `nil` by default and therefore it is no longer required to be set to `nil` when instantiating `PowerSyncCredentials` and can therefore be left out.
+
 ## 1.0.0-Beta.5
 
 * Implement improvements to errors originating in Kotlin so that they can be handled in Swift

Demo/PowerSyncExample.xcodeproj/project.xcworkspace/xcshareddata/swiftpm/Package.resolved

@@ -15,8 +15,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/powersync-ja/powersync-kotlin.git",
       "state" : {
-        "revision" : "61d195816585f30260181dcbd157bf1660c9ac4e",
-        "version" : "1.0.0-BETA22.0"
+        "revision" : "203db74889df8a20e3c6ac38aede6b0186d2e3b5",
+        "version" : "1.0.0-BETA23.0"
       }
     },
     {

Demo/PowerSyncExample/Components/TodoListView.swift

@@ -28,7 +28,7 @@ struct TodoListView: View {
             ForEach(todos) { todo in
                 TodoListRow(todo: todo) {
                     Task {
-                        await toggleCompletion(of: todo)
+                        try await toggleCompletion(of: todo)
                     }
                 }
             }

Demo/PowerSyncExample/PowerSync/SystemManager.swift

@@ -34,19 +34,23 @@ class SystemManager {
     }
 
     func watchLists(_ callback: @escaping (_ lists: [ListContent]) -> Void ) async {
-        for await lists in self.db.watch<[ListContent]>(
-            sql: "SELECT * FROM \(LISTS_TABLE)",
-            parameters: [],
-            mapper: { cursor in
-                ListContent(
-                    id: try cursor.getString(name: "id"),
-                    name: try cursor.getString(name: "name"),
-                    createdAt: try cursor.getString(name: "created_at"),
-                    ownerId: try cursor.getString(name: "owner_id")
-                )
+        do {
+            for try await lists in try self.db.watch<ListContent>(
+                sql: "SELECT * FROM \(LISTS_TABLE)",
+                parameters: [],
+                mapper: { cursor in
+                    try ListContent(
+                        id: cursor.getString(name: "id"),
+                        name: cursor.getString(name: "name"),
+                        createdAt: cursor.getString(name: "created_at"),
+                        ownerId: cursor.getString(name: "owner_id")
+                    )
+                }
+            ) {
+                callback(lists)
             }
-        ) {
-            callback(lists)
+        } catch {
+            print("Error in watch: \(error)")
         }
     }
 
@@ -59,11 +63,11 @@ class SystemManager {
 
     func deleteList(id: String) async throws {
         _ = try await db.writeTransaction(callback: { transaction in
-            _ = transaction.execute(
+            _ = try transaction.execute(
                 sql: "DELETE FROM \(LISTS_TABLE) WHERE id = ?",
                 parameters: [id]
             )
-            _ = transaction.execute(
+            _ = try transaction.execute(
                 sql: "DELETE FROM \(TODOS_TABLE) WHERE list_id = ?",
                 parameters: [id]
             )
@@ -72,24 +76,28 @@ class SystemManager {
     }
 
     func watchTodos(_ listId: String, _ callback: @escaping (_ todos: [Todo]) -> Void ) async {
-        for await todos in self.db.watch(
-            sql: "SELECT * FROM \(TODOS_TABLE) WHERE list_id = ?",
-            parameters: [listId],
-            mapper: { cursor in
-                return Todo(
-                    id: try cursor.getString(name: "id"),
-                    listId: try cursor.getString(name: "list_id"),
-                    photoId: try cursor.getStringOptional(name: "photo_id"),
-                    description: try cursor.getString(name: "description"),
-                    isComplete: try cursor.getBoolean(name: "completed"),
-                    createdAt: try cursor.getString(name: "created_at"),
-                    completedAt: try cursor.getStringOptional(name: "completed_at"),
-                    createdBy: try cursor.getStringOptional(name: "created_by"),
-                    completedBy: try cursor.getStringOptional(name: "completed_by")
-                )
+        do {
+            for try await todos in try self.db.watch(
+                sql: "SELECT * FROM \(TODOS_TABLE) WHERE list_id = ?",
+                parameters: [listId],
+                mapper: { cursor in
+                    try Todo(
+                        id: cursor.getString(name: "id"),
+                        listId: cursor.getString(name: "list_id"),
+                        photoId: cursor.getStringOptional(name: "photo_id"),
+                        description: cursor.getString(name: "description"),
+                        isComplete: cursor.getBoolean(name: "completed"),
+                        createdAt: cursor.getString(name: "created_at"),
+                        completedAt: cursor.getStringOptional(name: "completed_at"),
+                        createdBy: cursor.getStringOptional(name: "created_by"),
+                        completedBy: cursor.getStringOptional(name: "completed_by")
+                    )
+                }
+            ) {
+                callback(todos)
             }
-        ) {
-            callback(todos)
+        } catch {
+            print("Error in watch: \(error)")
         }
     }
 
@@ -117,7 +125,7 @@ class SystemManager {
 
     func deleteTodo(id: String) async throws {
         _ = try await db.writeTransaction(callback: { transaction in
-            transaction.execute(
+            try transaction.execute(
                 sql: "DELETE FROM \(TODOS_TABLE) WHERE id = ?",
                 parameters: [id]
             )

Package.resolved

@@ -5,8 +5,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/powersync-ja/powersync-kotlin.git",
       "state" : {
-        "revision" : "61d195816585f30260181dcbd157bf1660c9ac4e",
-        "version" : "1.0.0-BETA22.0"
+        "revision" : "203db74889df8a20e3c6ac38aede6b0186d2e3b5",
+        "version" : "1.0.0-BETA23.0"
       }
     },
     {

Package.swift

@@ -16,7 +16,7 @@ let package = Package(
             targets: ["PowerSync"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/powersync-ja/powersync-kotlin.git", exact: "1.0.0-BETA22.0"),
+        .package(url: "https://github.com/powersync-ja/powersync-kotlin.git", exact: "1.0.0-BETA23.0"),
         .package(url: "https://github.com/powersync-ja/powersync-sqlite-core-swift.git", "0.3.9"..<"0.4.0")
     ],
     targets: [

Sources/PowerSync/Kotlin/KotlinPowerSyncDatabaseImpl.swift

@@ -79,7 +79,7 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             mapper: mapper
         ) as! RowType
     }
-    
+
     func get<RowType>(
         sql: String,
         parameters: [Any]?,
@@ -93,7 +93,7 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             }
         ) as! RowType
     }
-    
+
     func getAll<RowType>(
         sql: String,
         parameters: [Any]?,
@@ -105,7 +105,7 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             mapper: mapper
         ) as! [RowType]
     }
-    
+
     func getAll<RowType>(
         sql: String,
         parameters: [Any]?,
@@ -131,7 +131,7 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
             mapper: mapper
         ) as! RowType?
     }
-    
+
     func getOptional<RowType>(
         sql: String,
         parameters: [Any]?,
@@ -150,48 +150,75 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
         sql: String,
         parameters: [Any]?,
         mapper: @escaping (SqlCursor) -> RowType
-    ) -> AsyncStream<[RowType]> {
-        AsyncStream { continuation in
+    ) throws -> AsyncThrowingStream<[RowType], Error> {
+        AsyncThrowingStream { continuation in
             Task {
-                for await values in try self.kotlinDatabase.watch(
-                    sql: sql,
-                    parameters: parameters,
-                    mapper: mapper
-                ) {
-                    continuation.yield(values as! [RowType])
+                do {
+                    for await values in try self.kotlinDatabase.watch(
+                        sql: sql,
+                        parameters: parameters,
+                        mapper: mapper
+                    ) {
+                        continuation.yield(values as! [RowType])
+                    }
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
                 }
-                continuation.finish()
             }
         }
     }
-    
+
     func watch<RowType>(
         sql: String,
         parameters: [Any]?,
         mapper: @escaping (SqlCursor) throws -> RowType
-    ) -> AsyncStream<[RowType]> {
-        AsyncStream { continuation in
+    ) throws -> AsyncThrowingStream<[RowType], Error> {
+        AsyncThrowingStream { continuation in
             Task {
-                for await values in try self.kotlinDatabase.watch(
-                    sql: sql,
-                    parameters: parameters,
-                    mapper: { cursor in
-                        try! mapper(cursor)
+                do {
+                    for await values in try self.kotlinDatabase.watch(
+                        sql: sql,
+                        parameters: parameters,
+                        mapper: { cursor in
+                            try! mapper(cursor)
+                        }
+                    ) {
+                        continuation.yield(values as! [RowType])
                     }
-                ) {
-                    continuation.yield(values as! [RowType])
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
                 }
-                continuation.finish()
             }
         }
     }
-    
-    public func writeTransaction<R>(callback: @escaping (any PowerSyncTransaction) -> R) async throws -> R {
-        return try await kotlinDatabase.writeTransaction(callback: callback) as! R
+
+    public func writeTransaction<R>(callback: @escaping (any PowerSyncTransaction) throws -> R) async throws -> R {
+        return try await kotlinDatabase.writeTransaction(callback: TransactionCallback(callback: callback)) as! R
+    }
+
+    public func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) throws -> R) async throws -> R {
+        return try await kotlinDatabase.readTransaction(callback: TransactionCallback(callback: callback)) as! R
     }
+}
+
+class TransactionCallback<R>: PowerSyncKotlin.ThrowableTransactionCallback {
+    let callback: (PowerSyncTransaction) throws -> R
 
-    public func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) -> R) async throws -> R {
-        return try await kotlinDatabase.readTransaction(callback: callback) as! R
+    init(callback: @escaping (PowerSyncTransaction) throws -> R) {
+        self.callback = callback
+    }
+
+    func execute(transaction: PowerSyncKotlin.PowerSyncTransaction) throws -> Any{
+        do {
+            return try callback(transaction)
+        } catch let error {
+            return PowerSyncKotlin.PowerSyncException(
+                message: error.localizedDescription,
+                cause: PowerSyncKotlin.KotlinThrowable(message: error.localizedDescription)
+            )
+        }
     }
 }
 

Sources/PowerSync/QueriesProtocol.swift

@@ -59,21 +59,21 @@ public protocol Queries {
         sql: String,
         parameters: [Any]?,
         mapper: @escaping (SqlCursor) -> RowType
-    ) -> AsyncStream<[RowType]>
+    ) throws -> AsyncThrowingStream<[RowType], Error>
 
     /// Execute a read-only (SELECT) query every time the source tables are modified
     /// and return the results as an array in a Publisher.
     func watch<RowType>(
         sql: String,
         parameters: [Any]?,
         mapper: @escaping (SqlCursor) throws -> RowType
-    ) -> AsyncStream<[RowType]>
+    ) throws -> AsyncThrowingStream<[RowType], Error>
 
     /// Execute a write transaction with the given callback
-    func writeTransaction<R>(callback: @escaping (any PowerSyncTransaction) -> R) async throws -> R
+    func writeTransaction<R>(callback: @escaping (any PowerSyncTransaction) throws -> R) async throws -> R
 
     /// Execute a read transaction with the given callback
-    func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) -> R) async throws -> R
+    func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) throws -> R) async throws -> R
 }
 
 extension Queries {
@@ -105,7 +105,7 @@ extension Queries {
     public func watch<RowType>(
         _ sql: String,
         mapper: @escaping (SqlCursor) -> RowType
-    ) -> AsyncStream<[RowType]> {
-        return watch(sql: sql, parameters: [], mapper: mapper)
+    ) throws -> AsyncThrowingStream<[RowType], Error> {
+        return try watch(sql: sql, parameters: [], mapper: mapper)
     }
 }

Tests/PowerSyncTests/Kotlin/KotlinPowerSyncDatabaseImplTests.swift

@@ -27,6 +27,21 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         try await super.tearDown()
     }
 
+    func testExecuteError() async throws {
+        do {
+            _ = try await database.execute(
+                sql: "INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)",
+                parameters: ["1", "Test User", "test@example.com"]
+            )
+            XCTFail("Expected an error to be thrown")
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)
+no such table: usersfail
+""")
+        }
+    }
+
     func testInsertAndGet() async throws {
         _ = try await database.execute(
             sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
@@ -49,6 +64,27 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(user.2, "test@example.com")
     }
 
+    func testGetError() async throws {
+        do {
+            let _ = try await database.get(
+                sql: "SELECT id, name, email FROM usersfail WHERE id = ?",
+                parameters: ["1"]
+            ) { cursor in
+                (
+                    try cursor.getString(name: "id"),
+                    try cursor.getString(name: "name"),
+                    try cursor.getString(name: "email")
+                )
+            }
+            XCTFail("Expected an error to be thrown")
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: SELECT id, name, email FROM usersfail WHERE id = ?
+no such table: usersfail
+""")
+        }
+    }
+
     func testGetOptional() async throws {
         let nonExistent: String? = try await database.getOptional(
             sql: "SELECT name FROM users WHERE id = ?",
@@ -74,6 +110,27 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(existing, "Test User")
     }
 
+    func testGetOptionalError() async throws {
+        do {
+            let _ = try await database.getOptional(
+                sql: "SELECT id, name, email FROM usersfail WHERE id = ?",
+                parameters: ["1"]
+            ) { cursor in
+                (
+                    try cursor.getString(name: "id"),
+                    try cursor.getString(name: "name"),
+                    try cursor.getString(name: "email")
+                )
+            }
+            XCTFail("Expected an error to be thrown")
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: SELECT id, name, email FROM usersfail WHERE id = ?
+no such table: usersfail
+""")
+        }
+    }
+
     func testGetAll() async throws {
         _ = try await database.execute(
             sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?), (?, ?, ?)",
@@ -97,6 +154,27 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(users[1].1, "User 2")
     }
 
+    func testGetAllError() async throws {
+        do {
+            let _ = try await database.getAll(
+                sql: "SELECT id, name, email FROM usersfail WHERE id = ?",
+                parameters: ["1"]
+            ) { cursor in
+                (
+                    try cursor.getString(name: "id"),
+                    try cursor.getString(name: "name"),
+                    try cursor.getString(name: "email")
+                )
+            }
+            XCTFail("Expected an error to be thrown")
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: SELECT id, name, email FROM usersfail WHERE id = ?
+no such table: usersfail
+""")
+        }
+    }
+
     func testWatchTableChanges() async throws {
         let expectation = XCTestExpectation(description: "Watch changes")
 
@@ -119,15 +197,15 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
 
         let resultsStore = ResultsStore()
 
-        let stream = database.watch(
+        let stream = try database.watch(
             sql: "SELECT name FROM users ORDER BY id",
             parameters: nil
         ) { cursor in
             cursor.getString(index: 0)!
         }
 
         let watchTask = Task {
-            for await names in stream {
+            for try await names in stream {
                 await resultsStore.append(names)
                 if await resultsStore.count() == 2 {
                     expectation.fulfill()
@@ -153,14 +231,37 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(finalResults[1], ["User 1", "User 2"])
     }
 
+    func testWatchError() async throws {
+        do {
+            let stream = try database.watch(
+                sql: "SELECT name FROM usersfail ORDER BY id",
+                parameters: nil
+            ) { cursor in
+                cursor.getString(index: 0)!
+            }
+
+            // Actually consume the stream to trigger the error
+            for try await _ in stream {
+                XCTFail("Should not receive any values")
+            }
+
+            XCTFail("Expected an error to be thrown")
+        } catch {
+        XCTAssertEqual(error.localizedDescription, """
+error while compiling: EXPLAIN SELECT name FROM usersfail ORDER BY id
+no such table: usersfail
+""")
+        }
+    }
+
     func testWriteTransaction() async throws {
         _ = try await database.writeTransaction { transaction in
-            _ = transaction.execute(
+            _ = try transaction.execute(
                 sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
                 parameters: ["1", "Test User", "test@example.com"]
             )
 
-            _ = transaction.execute(
+            _ = try transaction.execute(
                 sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
                 parameters: ["2", "Test User 2", "test2@example.com"]
             )
@@ -182,12 +283,12 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
 
         _ = try await database.writeTransaction { transaction in
             for i in 1...loopCount {
-                _ = transaction.execute(
+                _ = try transaction.execute(
                     sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
                     parameters: [String(i), "Test User \(i)", "test\(i)@example.com"]
                 )
 
-                _ = transaction.execute(
+                _ = try transaction.execute(
                     sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
                     parameters: [String(i*10000), "Test User \(i)-2", "test\(i)-2@example.com"]
                 )
@@ -204,6 +305,51 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(result as! Int, 2 * loopCount)
     }
 
+    func testWriteTransactionError() async throws {
+        do {
+            _ = try await database.writeTransaction { transaction in
+                _ = try transaction.execute(
+                    sql: "INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)",
+                    parameters: ["2", "Test User 2", "test2@example.com"]
+                )
+            }
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)
+no such table: usersfail
+""")
+        }
+    }
+
+    func testWriteTransactionErrorPerformsRollBack() async throws {
+        do {
+            _ = try await database.writeTransaction { transaction in
+                _ = try transaction.execute(
+                    sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
+                    parameters: ["1", "Test User", "test@example.com"]
+                )
+
+                _ = try transaction.execute(
+                    sql: "INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)",
+                    parameters: ["2", "Test User 2", "test2@example.com"]
+                )
+            }
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)
+no such table: usersfail
+""")
+        }
+
+        let result = try await database.getOptional(
+            sql: "SELECT COUNT(*) FROM users",
+            parameters: []
+        ) { cursor in try cursor.getLong(index: 0)
+        }
+
+        XCTAssertEqual(result, 0)
+    }
+
     func testReadTransaction() async throws {
         _ = try await database.execute(
             sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
@@ -212,7 +358,7 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
 
 
         _ = try await database.readTransaction { transaction in
-            let result = transaction.get(
+            let result = try transaction.get(
                 sql: "SELECT COUNT(*) FROM users",
                 parameters: []
             ) { cursor in
@@ -222,4 +368,22 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
             XCTAssertEqual(result as! Int, 1)
         }
     }
+
+    func testReadTransactionError() async throws {
+        do {
+            _ = try await database.readTransaction { transaction in
+                let result = try transaction.get(
+                    sql: "SELECT COUNT(*) FROM usersfail",
+                    parameters: []
+                ) { cursor in
+                    cursor.getLong(index: 0)
+                }
+            }
+        } catch {
+            XCTAssertEqual(error.localizedDescription, """
+error while compiling: SELECT COUNT(*) FROM usersfail
+no such table: usersfail
+""")
+        }
+    }
 }