Table updates from PowerSync side

Author: stevensJourney

Demo/GRDB Demo/GRDB Demo/GRDB_DemoApp.swift

@@ -33,9 +33,7 @@ func openDatabase()
         .appendingPathComponent("test.sqlite")
 
     var config = Configuration()
-
-    configurePowerSync(
-        config: &config,
+    config.configurePowerSync(
         schema: schema
     )
 

Sources/PowerSync/Kotlin/KotlinSQLiteConnectionPool.swift

@@ -1,20 +1,41 @@
 import PowerSyncKotlin
 
+class KotlinLeaseAdapter: PowerSyncKotlin.SwiftLeaseAdapter {
+    let pointer: UnsafeMutableRawPointer
+
+    init(
+        lease: SQLiteConnectionLease
+    ) {
+        pointer = UnsafeMutableRawPointer(lease.pointer)
+    }
+}
+
 final class SwiftSQLiteConnectionPoolAdapter: PowerSyncKotlin.SwiftPoolAdapter {
     let pool: SQLiteConnectionPoolProtocol
+    var updateTrackingTask: Task<Void, Never>?
 
     init(
         pool: SQLiteConnectionPoolProtocol
     ) {
         self.pool = pool
     }
 
-    func getPendingUpdates() -> Set<String> {
-        return pool.getPendingUpdates()
+    func linkUpdates(callback: any KotlinSuspendFunction1) {
+        updateTrackingTask = Task {
+            do {
+                for try await updates in pool.tableUpdates {
+                    _ = try await callback.invoke(p1: updates)
+                }
+            } catch {
+                // none of these calls should actually throw
+            }
+        }
     }
 
     func __closePool() async throws {
         do {
+            updateTrackingTask?.cancel()
+            updateTrackingTask = nil
             try pool.close()
         } catch {
             try? PowerSyncKotlin.throwPowerSyncException(
@@ -26,10 +47,22 @@ final class SwiftSQLiteConnectionPoolAdapter: PowerSyncKotlin.SwiftPoolAdapter {
         }
     }
 
-    func __leaseRead(callback: @escaping (Any) -> Void) async throws {
+    func __leaseRead(callback: any LeaseCallback) async throws {
         do {
-            try await pool.read { pointer in
-                callback(UInt(bitPattern: pointer))
+            var errorToThrow: Error?
+            try await pool.read { lease in
+                do {
+                    try callback.execute(
+                        lease: KotlinLeaseAdapter(
+                            lease: lease
+                        )
+                    )
+                } catch {
+                    errorToThrow = error
+                }
+            }
+            if let errorToThrow {
+                throw errorToThrow
             }
         } catch {
             try? PowerSyncKotlin.throwPowerSyncException(
@@ -41,10 +74,22 @@ final class SwiftSQLiteConnectionPoolAdapter: PowerSyncKotlin.SwiftPoolAdapter {
         }
     }
 
-    func __leaseWrite(callback: @escaping (Any) -> Void) async throws {
+    func __leaseWrite(callback: any LeaseCallback) async throws {
         do {
-            try await pool.write { pointer in
-                callback(UInt(bitPattern: pointer))
+            var errorToThrow: Error?
+            try await pool.write { lease in
+                do {
+                    try callback.execute(
+                        lease: KotlinLeaseAdapter(
+                            lease: lease
+                        )
+                    )
+                } catch {
+                    errorToThrow = error
+                }
+            }
+            if let errorToThrow {
+                throw errorToThrow
             }
         } catch {
             try? PowerSyncKotlin.throwPowerSyncException(
@@ -56,11 +101,16 @@ final class SwiftSQLiteConnectionPoolAdapter: PowerSyncKotlin.SwiftPoolAdapter {
         }
     }
 
-    func __leaseAll(callback: @escaping (Any, [Any]) -> Void) async throws {
+    func __leaseAll(callback: any AllLeaseCallback) async throws {
         // TODO, actually use all connections
         do {
-            try await pool.write { pointer in
-                callback(UInt(bitPattern: pointer), [])
+            try await pool.write { lease in
+                try? callback.execute(
+                    writeLease: KotlinLeaseAdapter(
+                        lease: lease
+                    ),
+                    readLeases: []
+                )
             }
         } catch {
             try? PowerSyncKotlin.throwPowerSyncException(

Sources/PowerSync/Protocol/SQLiteConnectionPool.swift

@@ -1,25 +1,29 @@
 import Foundation
 
+public protocol SQLiteConnectionLease {
+    var pointer: OpaquePointer { get }
+}
+
 /// 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>
+    var tableUpdates: AsyncStream<Set<String>> { get }
 
     /// Calls the callback with a read-only connection temporarily leased from the pool.
     func read(
-        onConnection: @Sendable @escaping (OpaquePointer) -> Void,
+        onConnection: @Sendable @escaping (SQLiteConnectionLease) -> Void,
     ) async throws
 
     /// Calls the callback with a read-write connection temporarily leased from the pool.
     func write(
-        onConnection: @Sendable @escaping (OpaquePointer) -> Void,
+        onConnection: @Sendable @escaping (SQLiteConnectionLease) -> Void,
     ) async throws
 
     /// Invokes the callback with all connections leased from the pool.
     func withAllConnections(
         onConnection: @Sendable @escaping (
-            _ writer: OpaquePointer,
-            _ readers: [OpaquePointer]
+            _ writer: SQLiteConnectionLease,
+            _ readers: [SQLiteConnectionLease]
         ) -> Void,
     ) async throws
 

Sources/PowerSyncGRDB/Config/Configuration+PowerSync.swift

@@ -0,0 +1,62 @@
+import Foundation
+import GRDB
+import PowerSync
+import SQLite3
+
+/// Extension for GRDB `Configuration` to add PowerSync support.
+///
+/// Call `configurePowerSync(schema:)` on your existing GRDB `Configuration` to:
+/// - Register the PowerSync SQLite core extension (required for PowerSync features).
+/// - Add PowerSync schema views to your database schema source.
+///
+/// This enables PowerSync replication and view management in your GRDB database.
+///
+/// Example usage:
+/// ```swift
+/// var config = Configuration()
+/// config.configurePowerSync(schema: mySchema)
+/// let dbQueue = try DatabaseQueue(path: dbPath, configuration: config)
+/// ```
+///
+/// - Parameter schema: The PowerSync `Schema` describing your sync views.
+public extension Configuration {
+    mutating func configurePowerSync(
+        schema: Schema
+    ) {
+        // Register the PowerSync core extension
+        prepareDatabase { database in
+            guard let bundle = Bundle(identifier: "co.powersync.sqlitecore") else {
+                throw PowerSyncGRDBError.coreBundleNotFound
+            }
+
+            // Construct the full path to the shared library inside the bundle
+            let fullPath = bundle.bundlePath + "/powersync-sqlite-core"
+
+            let extensionLoadResult = sqlite3_enable_load_extension(database.sqliteConnection, 1)
+            if extensionLoadResult != SQLITE_OK {
+                throw PowerSyncGRDBError.extensionLoadFailed("Could not enable extension loading")
+            }
+            var errorMsg: UnsafeMutablePointer<Int8>?
+            let loadResult = sqlite3_load_extension(database.sqliteConnection, fullPath, "sqlite3_powersync_init", &errorMsg)
+            if loadResult != SQLITE_OK {
+                if let errorMsg = errorMsg {
+                    let message = String(cString: errorMsg)
+                    sqlite3_free(errorMsg)
+                    throw PowerSyncGRDBError.extensionLoadFailed(message)
+                } else {
+                    throw PowerSyncGRDBError.unknownExtensionLoadError
+                }
+            }
+        }
+
+        // Supply the PowerSync views as a SchemaSource
+        let powerSyncSchemaSource = PowerSyncSchemaSource(
+            schema: schema
+        )
+        if let schemaSource = schemaSource {
+            self.schemaSource = schemaSource.then(powerSyncSchemaSource)
+        } else {
+            schemaSource = powerSyncSchemaSource
+        }
+    }
+}

Sources/PowerSyncGRDB/Config/PowerSyncSchemaSource.swift

@@ -0,0 +1,20 @@
+import GRDB
+import PowerSync
+
+/// A schema source used by GRDB to resolve primary keys for PowerSync views.
+///
+/// This struct allows GRDB to identify the primary key columns for tables/views
+/// defined in the PowerSync schema, enabling correct integration with GRDB's
+/// database observation and record management features.
+struct PowerSyncSchemaSource: DatabaseSchemaSource {
+    let schema: Schema
+
+    func columnsForPrimaryKey(_: Database, inView view: DatabaseObjectID) throws -> [String]? {
+        if schema.tables.first(where: { table in
+            table.viewName == view.name
+        }) != nil {
+            return ["id"]
+        }
+        return nil
+    }
+}

Sources/PowerSyncGRDB/Connections/GRDBConnecitonLease.swift

@@ -0,0 +1,18 @@
+import Foundation
+import GRDB
+import PowerSync
+
+/// Internal lease object that exposes the raw GRDB SQLite connection pointer.
+///
+/// This is used to bridge GRDB's managed database connection with the Kotlin SDK,
+/// allowing direct access to the underlying SQLite connection for PowerSync operations.
+final class GRDBConnectionLease: SQLiteConnectionLease {
+    var pointer: OpaquePointer
+
+    init(database: Database) throws {
+        guard let connection = database.sqliteConnection else {
+            throw PowerSyncGRDBError.connectionUnavailable
+        }
+        pointer = connection
+    }
+}

Sources/PowerSyncGRDB/Connections/GRDBConnectionPool.swift

@@ -0,0 +1,115 @@
+import Foundation
+import GRDB
+import PowerSync
+import SQLite3
+
+/// Adapts a GRDB `DatabasePool` for use with the PowerSync SDK.
+///
+/// This class implements `SQLiteConnectionPoolProtocol` and provides
+/// integration between GRDB's connection pool and PowerSync's requirements,
+/// including table update observation and direct access to SQLite connections.
+///
+/// - Provides async streams of table updates for replication.
+/// - Bridges GRDB's managed connections to PowerSync's lease abstraction.
+/// - Allows both read and write access to raw SQLite connections.
+public final class GRDBConnectionPool: SQLiteConnectionPoolProtocol {
+    let pool: DatabasePool
+
+    public private(set) var tableUpdates: AsyncStream<Set<String>>
+    private var tableUpdatesContinuation: AsyncStream<Set<String>>.Continuation?
+
+    public init(
+        pool: DatabasePool
+    ) {
+        self.pool = pool
+        // Cannot capture Self before initializing all properties
+        var tempContinuation: AsyncStream<Set<String>>.Continuation?
+        tableUpdates = AsyncStream { continuation in
+            tempContinuation = continuation
+            pool.add(
+                transactionObserver: PowerSyncTransactionObserver { updates in
+                    // push the update
+                    continuation.yield(updates)
+                },
+                extent: .databaseLifetime
+            )
+        }
+        tableUpdatesContinuation = tempContinuation
+    }
+
+    public func read(
+        onConnection: @Sendable @escaping (SQLiteConnectionLease) -> Void
+    ) async throws {
+        try await pool.read { database in
+            try onConnection(
+                GRDBConnectionLease(database: database)
+            )
+        }
+    }
+
+    public func write(
+        onConnection: @Sendable @escaping (SQLiteConnectionLease) -> Void
+    ) async throws {
+        // Don't start an explicit transaction, we do this internally
+        let updateBroker = UpdateBroker()
+        try await pool.writeWithoutTransaction { database in
+
+            let brokerPointer = Unmanaged.passUnretained(updateBroker).toOpaque()
+
+            /// GRDB only registers an update hook if it detects a requirement for one.
+            /// It also removes its own update hook if no longer needed.
+            /// We use the SQLite connection pointer directly, which sidesteps GRDB.
+            /// We can register our own temporary update hook here.
+            let previousParamPointer = sqlite3_update_hook(
+                database.sqliteConnection,
+                { brokerPointer, _, _, tableNameCString, _ in
+                    let broker = Unmanaged<UpdateBroker>.fromOpaque(brokerPointer!).takeUnretainedValue()
+                    broker.updates.insert(String(cString: tableNameCString!))
+                },
+                brokerPointer
+            )
+
+            // This should not be present
+            assert(previousParamPointer == nil, "A pre-existing update hook was already registered and has been overwritten.")
+
+            defer {
+                // Deregister our temporary hook
+                sqlite3_update_hook(database.sqliteConnection, nil, nil)
+            }
+
+            try onConnection(
+                GRDBConnectionLease(database: database)
+            )
+        }
+
+        // Notify GRDB consumers of updates
+        // Seems like we need to do this in a write transaction
+        try await pool.write { database in
+            for table in updateBroker.updates {
+                try database.notifyChanges(in: Table(table))
+                if table.hasPrefix("ps_data__") {
+                    let stripped = String(table.dropFirst("ps_data__".count))
+                    try database.notifyChanges(in: Table(stripped))
+                } else if table.hasPrefix("ps_data_local__") {
+                    let stripped = String(table.dropFirst("ps_data_local__".count))
+                    try database.notifyChanges(in: Table(stripped))
+                }
+            }
+        }
+        guard let pushUpdates = tableUpdatesContinuation else {
+            return
+        }
+        // Notify the PowerSync SDK consumers of updates
+        pushUpdates.yield(updateBroker.updates)
+    }
+
+    public func withAllConnections(
+        onConnection _: @escaping (SQLiteConnectionLease, [SQLiteConnectionLease]) -> Void
+    ) async throws {
+        // TODO:
+    }
+
+    public func close() throws {
+        try pool.close()
+    }
+}

Sources/PowerSyncGRDB/Errors.swift

@@ -0,0 +1,17 @@
+/// Errors thrown by the PowerSyncGRDB integration layer.
+///
+/// These errors represent issues encountered when bridging GRDB and PowerSync,
+/// such as missing extensions, failed extension loads, or unavailable connections.
+public enum PowerSyncGRDBError: Error {
+    /// The PowerSync SQLite core bundle could not be found.
+    case coreBundleNotFound
+
+    /// Failed to load the PowerSync SQLite extension, with an associated error message.
+    case extensionLoadFailed(String)
+
+    /// An unknown error occurred while loading the PowerSync SQLite extension.
+    case unknownExtensionLoadError
+
+    /// The underlying SQLite connection could not be obtained from GRDB.
+    case connectionUnavailable
+}