wip: table update hooks

Author: stevensJourney

Sources/PowerSync/Kotlin/KotlinSQLiteConnectionPool.swift

@@ -9,6 +9,10 @@ final class SwiftSQLiteConnectionPoolAdapter: PowerSyncKotlin.SwiftPoolAdapter {
         self.pool = pool
     }
 
+    func getPendingUpdates() -> Set<String> {
+        return pool.getPendingUpdates()
+    }
+
     func __closePool() async throws {
         do {
             try pool.close()

Sources/PowerSync/Protocol/SQLiteConnectionPool.swift

@@ -3,6 +3,8 @@ import Foundation
 /// An implementation of a connection pool providing asynchronous access to a single writer and multiple readers.
 /// This is the underlying pool implementation on which the higher-level PowerSync Swift SDK is built on.
 public protocol SQLiteConnectionPoolProtocol {
+    func getPendingUpdates() -> Set<String>
+
     /// Calls the callback with a read-only connection temporarily leased from the pool.
     func read(
         onConnection: @Sendable @escaping (OpaquePointer) -> Void,

Sources/PowerSyncGRDB/GRDBPool.swift

@@ -82,13 +82,58 @@ func configurePowerSync(
     }
 }
 
-class GRDBConnectionPool: SQLiteConnectionPoolProtocol {
+final class PowerSyncTransactionObserver: TransactionObserver {
+    let onChange: (_ tableName: String) -> Void
+
+    init(
+        onChange: @escaping (_ tableName: String) -> Void
+    ) {
+        self.onChange = onChange
+    }
+
+    func observes(eventsOfKind _: DatabaseEventKind) -> Bool {
+        // We want all the events for the PowerSync SDK
+        return true
+    }
+
+    func databaseDidChange(with event: DatabaseEvent) {
+        onChange(event.tableName)
+    }
+
+    func databaseDidCommit(_: GRDB.Database) {}
+
+    func databaseDidRollback(_: GRDB.Database) {}
+}
+
+final class GRDBConnectionPool: SQLiteConnectionPoolProtocol {
     let pool: DatabasePool
+    var pendingUpdates: Set<String>
+    private let pendingUpdatesQueue = DispatchQueue(
+        label: "co.powersync.pendingUpdatesQueue"
+    )
 
     init(
         pool: DatabasePool
     ) {
         self.pool = pool
+        self.pendingUpdates = Set()
+        pool.add(
+            transactionObserver: PowerSyncTransactionObserver { tableName in
+                // push the update
+                self.pendingUpdatesQueue.sync {
+                    self.pendingUpdates.insert(tableName)
+                }
+            },
+            extent: .databaseLifetime
+        )
+    }
+
+    func getPendingUpdates() -> Set<String> {
+        self.pendingUpdatesQueue.sync {
+            let copy = self.pendingUpdates
+            self.pendingUpdates.removeAll()
+            return copy
+        }
     }
 
     func read(

Tests/PowerSyncGRDBTests/BasicTest.swift

@@ -4,6 +4,17 @@
 
 import XCTest
 
+struct User: Codable, Identifiable, FetchableRecord, PersistableRecord {
+    var id: String
+    var name: String
+
+    static var databaseTableName = "users"
+
+    enum Columns {
+        static let name = Column(CodingKeys.name)
+    }
+}
+
 final class GRDBTests: XCTestCase {
     private var database: PowerSyncDatabaseProtocol!
     private var schema: Schema!
@@ -47,7 +58,7 @@ final class GRDBTests: XCTestCase {
         try await super.tearDown()
     }
 
-    func testValidValues() async throws {
+    func testBasicOperations() async throws {
         // Create users with the PowerSync SDK
         let initialUserName = "Bob"
 
@@ -66,17 +77,6 @@ final class GRDBTests: XCTestCase {
         XCTAssertTrue(initialUserNames.first == initialUserName)
 
         // Now define a GRDB struct for query purposes
-        struct User: Codable, Identifiable, FetchableRecord, PersistableRecord {
-            var id: String
-            var name: String
-
-            static var databaseTableName = "users"
-
-            enum Columns {
-                static let name = Column(CodingKeys.name)
-            }
-        }
-
         // Query the Users with GRDB, this should have the same result as with PowerSync
         let grdbUserNames = try await pool.read { database in
             try User.fetchAll(database)
@@ -98,4 +98,161 @@ final class GRDBTests: XCTestCase {
         XCTAssert(grdbUserNames2.count == 2)
         XCTAssert(grdbUserNames2[1].name == "another")
     }
+
+    func testPowerSyncUpdates() async throws {
+        let expectation = XCTestExpectation(description: "Watch changes")
+
+        // Create an actor to handle concurrent mutations
+        actor ResultsStore {
+            private var results: Set<String> = []
+
+            func append(_ names: [String]) {
+                results.formUnion(names)
+            }
+
+            func getResults() -> Set<String> {
+                results
+            }
+
+            func count() -> Int {
+                results.count
+            }
+        }
+
+        let resultsStore = ResultsStore()
+
+        let watchTask = Task {
+            let stream = try database.watch(
+                options: WatchOptions(
+                    sql: "SELECT name FROM users ORDER BY id",
+                    mapper: { cursor in
+                        try cursor.getString(index: 0)
+                    }
+                ))
+            for try await names in stream {
+                await resultsStore.append(names)
+                if await resultsStore.count() == 2 {
+                    expectation.fulfill()
+                }
+            }
+        }
+
+        try await database.execute(
+            sql: "INSERT INTO users(id, name) VALUES(uuid(), ?)",
+            parameters: ["one"]
+        )
+
+        try await database.execute(
+            sql: "INSERT INTO users(id, name) VALUES(uuid(), ?)",
+            parameters: ["two"]
+        )
+        await fulfillment(of: [expectation], timeout: 5)
+        watchTask.cancel()
+    }
+
+    func testPowerSyncUpdatesFromGRDB() async throws {
+        let expectation = XCTestExpectation(description: "Watch changes")
+
+        // Create an actor to handle concurrent mutations
+        actor ResultsStore {
+            private var results: Set<String> = []
+
+            func append(_ names: [String]) {
+                results.formUnion(names)
+            }
+
+            func getResults() -> Set<String> {
+                results
+            }
+
+            func count() -> Int {
+                results.count
+            }
+        }
+
+        let resultsStore = ResultsStore()
+
+        let watchTask = Task {
+            let stream = try database.watch(
+                options: WatchOptions(
+                    sql: "SELECT name FROM users ORDER BY id",
+                    mapper: { cursor in
+                        try cursor.getString(index: 0)
+                    }
+                ))
+            for try await names in stream {
+                await resultsStore.append(names)
+                if await resultsStore.count() == 2 {
+                    expectation.fulfill()
+                }
+            }
+        }
+
+        try await pool.write { database in
+            try User(
+                id: UUID().uuidString,
+                name: "one",
+            ).insert(database)
+        }
+
+        try await pool.write { database in
+            try User(
+                id: UUID().uuidString,
+                name: "two",
+            ).insert(database)
+        }
+
+        await fulfillment(of: [expectation], timeout: 5)
+        watchTask.cancel()
+    }
+
+    func testGRDBUpdatesFromPowerSync() async throws {
+        let expectation = XCTestExpectation(description: "Watch changes")
+
+        // Create an actor to handle concurrent mutations
+        actor ResultsStore {
+            private var results: Set<String> = []
+
+            func append(_ names: [String]) {
+                results.formUnion(names)
+            }
+
+            func getResults() -> Set<String> {
+                results
+            }
+
+            func count() -> Int {
+                results.count
+            }
+        }
+
+        let resultsStore = ResultsStore()
+
+        let watchTask = Task {
+            let observation = ValueObservation.tracking {
+                try User.order(User.Columns.name.asc).fetchAll($0)
+            }
+
+            for try await users in observation.values(in: pool) {
+                print("users \(users)")
+                await resultsStore.append(users.map { $0.name })
+                if await resultsStore.count() == 2 {
+                    expectation.fulfill()
+                }
+            }
+        }
+
+        try await database.execute(
+            sql: "INSERT INTO users(id, name) VALUES(uuid(), ?)",
+            parameters: ["one"]
+        )
+
+        try await database.execute(
+            sql: "INSERT INTO users(id, name) VALUES(uuid(), ?)",
+            parameters: ["two"]
+        )
+
+        await fulfillment(of: [expectation], timeout: 5)
+        watchTask.cancel()
+    }
 }