Implement CRUD transactions in Swift instead of Kotlin

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 1.14.0 (unreleased)
+
+* Add `opDataTyped` and `previousValuesTyped` to `CrudEntry`, providing typed values instead of strings.
+* Make `CrudBatch`, `CrudEntry` and `CrudTransaction` a concrete struct. Note that these can no longer be created in user code.
+
 ## 1.13.1
 
 * Don't attempt to create WebSocket connections on watchOS.

Sources/PowerSync/Kotlin/KotlinPowerSyncDatabaseImpl.swift

@@ -72,19 +72,6 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol,
         )
     }
 
-    func getCrudBatch(limit: Int32 = 100) async throws -> CrudBatch? {
-        guard let base = try await kotlinDatabase.getCrudBatch(limit: limit) else {
-            return nil
-        }
-        return try KotlinCrudBatch(
-            batch: base
-        )
-    }
-
-    func getCrudTransactions() -> any CrudTransactions {
-        return KotlinCrudTransactions(db: kotlinDatabase)
-    }
-
     func getPowerSyncVersion() async throws -> String {
         try await kotlinDatabase.getPowerSyncVersion()
     }

Sources/PowerSync/Protocol/PowerSyncDatabaseProtocol.swift

@@ -191,40 +191,6 @@ public protocol PowerSyncDatabaseProtocol: Queries, Sendable {
         options: ConnectOptions?
     ) async throws
 
-    /// Get a batch of crud data to upload.
-    ///
-    /// Returns nil if there is no data to upload.
-    ///
-    /// Use this from the `PowerSyncBackendConnector.uploadData` callback.
-    ///
-    /// Once the data have been successfully uploaded, call `CrudBatch.complete` before
-    /// requesting the next batch.
-    ///
-    /// - Parameter limit: Maximum number of updates to return in a single batch. Default is 100.
-    ///
-    /// This method does include transaction ids in the result, but does not group
-    /// data by transaction. One batch may contain data from multiple transactions,
-    /// and a single transaction may be split over multiple batches.
-    func getCrudBatch(limit: Int32) async throws -> CrudBatch?
-
-    /// Obtains an async iterator of completed transactions with local writes against the database.
-    ///
-    /// This is typically used from the ``PowerSyncBackendConnectorProtocol/uploadData(database:)`` callback.
-    /// Each entry emitted by teh returned flow is a full transaction containing all local writes made while that transaction was
-    /// active.
-    ///
-    /// Unlike ``getNextCrudTransaction()``, which always returns the oldest transaction that hasn't been
-    /// ``CrudTransaction/complete()``d yet, this iterator can be used to upload multiple transactions.
-    /// Calling ``CrudTransaction/complete()`` will mark that and all prior transactions returned by this iterator as
-    /// completed.
-    ///
-    /// This can be used to upload multiple transactions in a single batch, e.g. with
-    ///
-    /// ```Swift
-    ///
-    /// ```
-    func getCrudTransactions() -> any CrudTransactions
-
     /// Convenience method to get the current version of PowerSync.
     func getPowerSyncVersion() async throws -> String
 
@@ -343,9 +309,60 @@ public extension PowerSyncDatabaseProtocol {
         try await disconnectAndClear(clearLocal: true, soft: soft)
     }
 
+    /// Get a batch of crud data to upload.
+    ///
+    /// Returns nil if there is no data to upload.
+    ///
+    /// Use this from the `PowerSyncBackendConnector.uploadData` callback.
+    ///
+    /// Once the data have been successfully uploaded, call `CrudBatch.complete` before
+    /// requesting the next batch.
+    ///
+    /// - Parameter limit: Maximum number of updates to return in a single batch. Default is 100.
+    ///
+    /// This method does include transaction ids in the result, but does not group
+    /// data by transaction. One batch may contain data from multiple transactions,
+    /// and a single transaction may be split over multiple batches.
     func getCrudBatch(limit: Int32 = 100) async throws -> CrudBatch? {
-        try await getCrudBatch(
-            limit: limit
+        var entries = try await getAll(
+            sql: "SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?",
+            parameters: [Int64(limit + 1)],
+            mapper: CrudEntry.fromCursor
         )
+
+        if entries.isEmpty {
+            return nil
+        }
+
+        let hasMore = entries.count > limit
+        if hasMore {
+            entries.removeLast()
+        }
+
+        return CrudBatch(
+            hasMore: hasMore,
+            crud: entries,
+            db: self
+        )
+    }
+
+    /// Obtains an async iterator of completed transactions with local writes against the database.
+    ///
+    /// This is typically used from the ``PowerSyncBackendConnectorProtocol/uploadData(database:)`` callback.
+    /// Each entry emitted by teh returned flow is a full transaction containing all local writes made while that transaction was
+    /// active.
+    ///
+    /// Unlike ``getNextCrudTransaction()``, which always returns the oldest transaction that hasn't been
+    /// ``CrudTransaction/complete()``d yet, this iterator can be used to upload multiple transactions.
+    /// Calling ``CrudTransaction/complete()`` will mark that and all prior transactions returned by this iterator as
+    /// completed.
+    ///
+    /// This can be used to upload multiple transactions in a single batch, e.g. with
+    ///
+    /// ```Swift
+    ///
+    /// ```
+    func getCrudTransactions() -> CrudTransactions {
+        CrudTransactions(db: self)
     }
 }

Sources/PowerSync/Protocol/db/CrudBatch.swift

@@ -1,24 +1,40 @@
 import Foundation
 
-/// A transaction of client-side changes.
-public protocol CrudBatch: Sendable {
+/// A collection of client-side changes.
+public struct CrudBatch: Sendable {
     /// Indicates if there are additional Crud items in the queue which are not included in this batch
-    var hasMore: Bool { get }
+    public let hasMore: Bool
 
     /// List of client-side changes.
-    var crud: [any CrudEntry] { get }
+    public let crud: [CrudEntry]
+
+    private let db: PowerSyncDatabaseProtocol
+
+    internal init(hasMore: Bool, crud: [CrudEntry], db: PowerSyncDatabaseProtocol) {
+        self.hasMore = hasMore
+        self.crud = crud
+        self.db = db
+    }
 
     /// Call to remove the changes from the local queue, once successfully uploaded.
     ///
     /// `writeCheckpoint` is optional.
-    func complete(writeCheckpoint: String?) async throws
+    public func complete(writeCheckpoint: String? = nil) async throws {
+        let lastId = crud.last!.clientId
+        try await completeCrudItems(self.db, lastId, writeCheckpoint: writeCheckpoint)
+    }
 }
 
-public extension CrudBatch {
-    /// Call to remove the changes from the local queue, once successfully uploaded.
-    func complete() async throws {
-        try await self.complete(
-            writeCheckpoint: nil
-        )
+internal func completeCrudItems(_ db: any PowerSyncDatabaseProtocol, _ lastItemId: Int64, writeCheckpoint: String? = nil) async throws {
+    return try await db.writeTransaction { tx in
+        try tx.execute(sql: "DELETE FROM ps_crud WHERE id <= ?", parameters: [lastItemId])
+        if writeCheckpoint != nil {
+            let hasCrud = (try tx.getOptional(sql: "SELECT 1 FROM ps_crud", parameters: nil) { cursor in () }) != nil
+            if !hasCrud {
+                try tx.execute(sql: "UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name = '$local'", parameters: [writeCheckpoint])
+                return
+            }
+        }
+        try tx.execute(sql: "UPDATE ps_buckets SET target_op = 9223372036854775807 WHERE name = '$local'", parameters: nil)
     }
 }