chore: add transaction and watch error handling

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]
             )

Sources/PowerSync/Kotlin/KotlinPowerSyncDatabaseImpl.swift

@@ -194,8 +194,42 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
         }
     }
 
-    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 {
+        var err: Error? = nil
+        let result = try await kotlinDatabase.writeTransaction { transaction in
+            do {
+                let res = try callback(transaction)
+                return res as R
+            } catch {
+                err = error
+                return TransactionResponse.rollback
+            }
+        }
+
+        if(err != nil) {
+            throw err!
+        }
+
+        return result as! R
+    }
+
+    public func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) throws -> R) async throws -> R {
+        var err: Error? = nil
+        let result = try await kotlinDatabase.readTransaction { transaction in
+            do {
+                let res = try callback(transaction)
+                return res as R
+            } catch {
+                err = error
+                return TransactionResponse.rollback
+            }
+        }
+
+        if(err != nil) {
+            throw err!
+        }
+
+        return result as! R
     }
 
     public func readTransaction<R>(callback: @escaping (any PowerSyncTransaction) -> R) async throws -> R {

Sources/PowerSync/QueriesProtocol.swift

@@ -70,10 +70,10 @@ public protocol Queries {
     ) 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 {

Tests/PowerSyncTests/Kotlin/KotlinPowerSyncDatabaseImplTests.swift

@@ -26,7 +26,7 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         database = nil
         try await super.tearDown()
     }
-    
+
     func testExecuteError() async throws {
         do {
             _ = try await database.execute(
@@ -41,7 +41,7 @@ no such table: usersfail
 """)
         }
     }
-    
+
     func testInsertAndGet() async throws {
         _ = try await database.execute(
             sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
@@ -63,7 +63,7 @@ no such table: usersfail
         XCTAssertEqual(user.1, "Test User")
         XCTAssertEqual(user.2, "[email protected]")
     }
-    
+
     func testGetError() async throws {
         do {
             let _ = try await database.get(
@@ -109,7 +109,7 @@ no such table: usersfail
 
         XCTAssertEqual(existing, "Test User")
     }
-    
+
     func testGetOptionalError() async throws {
         do {
             let _ = try await database.getOptional(
@@ -153,7 +153,7 @@ no such table: usersfail
         XCTAssertEqual(users[1].0, "2")
         XCTAssertEqual(users[1].1, "User 2")
     }
-    
+
     func testGetAllError() async throws {
         do {
             let _ = try await database.getAll(
@@ -230,7 +230,7 @@ no such table: usersfail
         XCTAssertEqual(finalResults.count, 2)
         XCTAssertEqual(finalResults[1], ["User 1", "User 2"])
     }
-    
+
     func testWatchError() async throws {
         do {
             let stream = try database.watch(
@@ -239,12 +239,12 @@ no such table: usersfail
             ) { 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, """
@@ -256,12 +256,12 @@ 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", "[email protected]"]
             )
 
-            _ = transaction.execute(
+            _ = try transaction.execute(
                 sql: "INSERT INTO users (id, name, email) VALUES (?, ?, ?)",
                 parameters: ["2", "Test User 2", "[email protected]"]
             )
@@ -283,12 +283,12 @@ no such table: usersfail
 
         _ = 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)[email protected]"]
                 )
@@ -305,6 +305,51 @@ no such table: usersfail
         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", "[email protected]"]
+                )
+            }
+        } 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", "[email protected]"]
+                )
+
+                _ = try transaction.execute(
+                    sql: "INSERT INTO usersfail (id, name, email) VALUES (?, ?, ?)",
+                    parameters: ["2", "Test User 2", "[email protected]"]
+                )
+            }
+        } 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 (?, ?, ?)",
@@ -313,7 +358,7 @@ no such table: usersfail
 
 
         _ = try await database.readTransaction { transaction in
-            let result = transaction.get(
+            let result = try transaction.get(
                 sql: "SELECT COUNT(*) FROM users",
                 parameters: []
             ) { cursor in
@@ -323,4 +368,22 @@ no such table: usersfail
             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
+""")
+        }
+    }
 }