feat: add sync changes

Author: DominicGBauer

Sources/PowerSyncSwift/PowerSyncTransactionProtocol.swift

@@ -1,29 +1,56 @@
 public protocol PowerSyncTransactionProtocol {
     /// Execute a write query and return the number of affected rows
     func execute(
-        sql: String,
-        parameters: [Any]?
+        _ sql: String,
+        _ parameters: [Any]?
     ) async throws -> Int64
 
     /// Execute a read-only query and return a single optional result
     func getOptional<RowType>(
-        sql: String,
-        parameters: [Any]?,
+        _ sql: String,
+        _ parameters: [Any]?,
         mapper: @escaping (SqlCursor) -> RowType
     ) async throws -> RowType?
 
     /// Execute a read-only query and return all results
     func getAll<RowType>(
-        sql: String,
-        parameters: [Any]?,
+        _ sql: String,
+        _ parameters: [Any]?,
         mapper: @escaping (SqlCursor) -> RowType
     ) async throws -> [RowType]
 
     /// Execute a read-only query and return a single result
     /// Throws if no result is found
     func get<RowType>(
-        sql: String,
-        parameters: [Any]?,
+        _ sql: String,
+        _ parameters: [Any]?,
         mapper: @escaping (SqlCursor) -> RowType
     ) async throws -> RowType
 }
+
+extension PowerSyncTransactionProtocol {
+    public func execute(_ sql: String) async throws -> Int64 {
+        return try await execute(sql, [])
+    }
+    
+    public func get<RowType>(
+        _ sql: String,
+        mapper: @escaping (SqlCursor) -> RowType
+    ) async throws -> RowType {
+        return try await get(sql, [], mapper: mapper)
+    }
+    
+    public func getAll<RowType>(
+        _ sql: String,
+        mapper: @escaping (SqlCursor) -> RowType
+    ) async throws -> [RowType] {
+        return try await getAll(sql, [], mapper: mapper)
+    }
+    
+    public func getOptional<RowType>(
+        _ sql: String,
+        mapper: @escaping (SqlCursor) -> RowType
+    ) async throws -> RowType? {
+        return try await getOptional(sql, [], mapper: mapper)
+    }
+}

Sources/PowerSyncSwift/Sync/StreamingSyncCheckpointDiff.swift

@@ -0,0 +1,22 @@
+import Foundation
+
+struct StreamingSyncCheckpointDiff: Codable {
+    let lastOpId: String
+    let updatedBuckets: [BucketChecksum]
+    let removedBuckets: [String]
+    let writeCheckpoint: String?
+
+    enum CodingKeys: String, CodingKey {
+        case lastOpId = "last_op_id"
+        case updatedBuckets = "updated_buckets"
+        case removedBuckets = "removed_buckets"
+        case writeCheckpoint = "write_checkpoint"
+    }
+
+    init(lastOpId: String, updatedBuckets: [BucketChecksum], removedBuckets: [String], writeCheckpoint: String? = nil) {
+        self.lastOpId = lastOpId
+        self.updatedBuckets = updatedBuckets
+        self.removedBuckets = removedBuckets
+        self.writeCheckpoint = writeCheckpoint
+    }
+}

Sources/PowerSyncSwift/Sync/StreamingSyncRequest.swift

@@ -0,0 +1,48 @@
+import Foundation
+import AnyCodable
+
+struct StreamingSyncRequest: Codable {
+    let buckets: [BucketRequest]
+    let includeChecksum: Bool
+    let clientId: String
+    let parameters: [String: AnyCodable]
+    private let rawData: Bool = true
+
+    enum CodingKeys: String, CodingKey {
+        case buckets
+        case includeChecksum = "include_checksum"
+        case clientId = "client_id"
+        case parameters
+        case rawData = "raw_data"
+    }
+
+    init(from decoder: Decoder) throws {
+        let container = try decoder.container(keyedBy: CodingKeys.self)
+
+        self.buckets = try container.decode([BucketRequest].self, forKey: .buckets)
+        self.includeChecksum = try container.decode(Bool.self, forKey: .includeChecksum)
+        self.clientId = try container.decode(String.self, forKey: .clientId)
+        self.parameters = try container.decode([String: AnyCodable].self, forKey: .parameters)
+    }
+
+    func encode(to encoder: Encoder) throws {
+        var container = encoder.container(keyedBy: CodingKeys.self)
+
+        try container.encode(buckets, forKey: .buckets)
+        try container.encode(includeChecksum, forKey: .includeChecksum)
+        try container.encode(clientId, forKey: .clientId)
+        try container.encode(parameters, forKey: .parameters)
+        try container.encode(rawData, forKey: .rawData)
+    }
+
+    init(buckets: [BucketRequest],
+        includeChecksum: Bool = true,
+        clientId: String,
+        parameters: [String: AnyCodable] = [:]
+    ) {
+        self.buckets = buckets
+        self.includeChecksum = includeChecksum
+        self.clientId = clientId
+        self.parameters = parameters
+    }
+}

Sources/PowerSyncSwift/Sync/SyncDataBatch.swift

@@ -0,0 +1,5 @@
+import Foundation
+
+struct SyncDataBatch: Codable {
+    let buckets: [SyncDataBucket]
+}

Sources/PowerSyncSwift/Sync/SyncDataBucket.swift

@@ -0,0 +1,25 @@
+import Foundation
+
+struct SyncDataBucket: Codable {
+    let bucket: String
+    let data: [OplogEntry]
+    let hasMore: Bool
+    let after: String?
+    let nextAfter: String?
+
+    enum CodingKeys: String, CodingKey {
+        case bucket
+        case data
+        case hasMore = "has_more"
+        case after
+        case nextAfter = "next_after"
+    }
+
+    init(bucket: String, data: [OplogEntry], hasMore: Bool = false, after: String? = nil, nextAfter: String? = nil) {
+        self.bucket = bucket
+        self.data = data
+        self.hasMore = hasMore
+        self.after = after
+        self.nextAfter = nextAfter
+    }
+}

Sources/PowerSyncSwift/Sync/SyncLocalDatabaseResult.swift

@@ -0,0 +1,33 @@
+import Foundation
+
+struct SyncLocalDatabaseResult: Codable {
+    var ready: Bool = true
+    var checkpointValid: Bool = true
+    var checkpointFailures: [String]? = nil
+
+    enum CodingKeys: String, CodingKey {
+        case ready
+        case checkpointValid = "valid"
+        case checkpointFailures = "failed_buckets"
+    }
+    
+    init(ready: Bool = true, checkpointValid: Bool = true, checkpointFailures: [String]? = nil) {
+        self.ready = ready
+        self.checkpointValid = checkpointValid
+        self.checkpointFailures = checkpointFailures
+    }
+    
+    init(from decoder: Decoder) throws {
+        let container = try decoder.container(keyedBy: CodingKeys.self)
+        ready = try container.decodeIfPresent(Bool.self, forKey: .ready) ?? true
+        checkpointValid = try container.decode(Bool.self, forKey: .checkpointValid)
+        checkpointFailures = try container.decodeIfPresent([String].self, forKey: .checkpointFailures)
+    }
+}
+
+extension SyncLocalDatabaseResult: CustomStringConvertible {
+    var description: String {
+        return "SyncLocalDatabaseResult<ready=\(ready), checkpointValid=\(checkpointValid), failures=\(checkpointFailures ?? [])>"
+    }
+}
+

Sources/PowerSyncSwift/Sync/SyncStatus.swift

@@ -0,0 +1,145 @@
+import Foundation
+import Combine
+
+/// Protocol defining the sync status interface
+public protocol SyncStatusData {
+    /// true if currently connected.
+    ///
+    /// This means the PowerSync connection is ready to download, and PowerSyncBackendConnector.uploadData may be called for any local changes.
+    var connected: Bool { get }
+    
+    /// true if the PowerSync connection is busy connecting.
+    ///
+    /// During this stage, PowerSyncBackendConnector.uploadData may already be called, and uploading may be true.
+    var connecting: Bool { get }
+    
+    /// true if actively downloading changes.
+    ///
+    /// This is only true when connected is also true.
+    var downloading: Bool { get }
+    
+    /// true if uploading changes
+    var uploading: Bool { get }
+    
+    /// Time that a last sync has fully completed, if any.
+    ///
+    /// Currently this is reset to null after a restart.
+    var lastSyncedAt: Date? { get }
+    
+    /// Indicates whether there has been at least one full sync, if any.
+    ///
+    /// Is nil when unknown, for example when state is still being loaded from the database.
+    var hasSynced: Bool? { get }
+    
+    /// Error during uploading.
+    ///
+    /// Cleared on the next successful upload.
+    var uploadError: Any? { get }
+    
+    /// Error during downloading (including connecting).
+    ///
+    /// Cleared on the next successful data download.
+    var downloadError: Any? { get }
+    
+    /// Convenience getter for either the value of downloadError or uploadError
+    var anyError: Any? { get }
+}
+
+/// Internal container for sync status data
+struct SyncStatusDataContainer: SyncStatusData {
+    let connected: Bool
+    let connecting: Bool
+    let downloading: Bool
+    let uploading: Bool
+    let lastSyncedAt: Date?
+    let hasSynced: Bool?
+    let uploadError: Any?
+    let downloadError: Any?
+    
+    var anyError: Any? {
+        return downloadError ?? uploadError
+    }
+    
+    init(
+        connected: Bool = false,
+        connecting: Bool = false,
+        downloading: Bool = false,
+        uploading: Bool = false,
+        lastSyncedAt: Date? = nil,
+        hasSynced: Bool? = nil,
+        uploadError: Any? = nil,
+        downloadError: Any? = nil
+    ) {
+        self.connected = connected
+        self.connecting = connecting
+        self.downloading = downloading
+        self.uploading = uploading
+        self.lastSyncedAt = lastSyncedAt
+        self.hasSynced = hasSynced
+        self.uploadError = uploadError
+        self.downloadError = downloadError
+    }
+}
+
+/// Public sync status class
+public class SyncStatus: SyncStatusData {
+    private var data: SyncStatusDataContainer
+    private let stateSubject: CurrentValueSubject<SyncStatusDataContainer, Never>
+    
+    init(data: SyncStatusDataContainer = SyncStatusDataContainer()) {
+        self.data = data
+        self.stateSubject = CurrentValueSubject(data)
+    }
+    
+    /// Returns a publisher which emits whenever the sync status has changed
+    public func asPublisher() -> AnyPublisher<SyncStatusData, Never> {
+        return stateSubject
+            .map { $0 as SyncStatusData }
+            .eraseToAnyPublisher()
+    }
+    
+    /// Updates the internal sync status indicators and emits publisher updates
+    internal func update(
+        connected: Bool? = nil,
+        connecting: Bool? = nil,
+        downloading: Bool? = nil,
+        uploading: Bool? = nil,
+        hasSynced: Bool? = nil,
+        lastSyncedAt: Date? = nil,
+        uploadError: Any? = nil,
+        downloadError: Any? = nil,
+        clearUploadError: Bool = false,
+        clearDownloadError: Bool = false
+    ) {
+        data = SyncStatusDataContainer(
+            connected: connected ?? data.connected,
+            connecting: connecting ?? data.connecting,
+            downloading: downloading ?? data.downloading,
+            uploading: uploading ?? data.uploading,
+            lastSyncedAt: lastSyncedAt ?? data.lastSyncedAt,
+            hasSynced: hasSynced ?? data.hasSynced,
+            uploadError: clearUploadError ? nil : (uploadError ?? data.uploadError),
+            downloadError: clearDownloadError ? nil : (downloadError ?? data.downloadError)
+        )
+        stateSubject.send(data)
+    }
+    
+    public var connected: Bool { data.connected }
+    public var connecting: Bool { data.connecting }
+    public var downloading: Bool { data.downloading }
+    public var uploading: Bool { data.uploading }
+    public var lastSyncedAt: Date? { data.lastSyncedAt }
+    public var hasSynced: Bool? { data.hasSynced }
+    public var uploadError: Any? { data.uploadError }
+    public var downloadError: Any? { data.downloadError }
+    public var anyError: Any? { data.anyError }
+    
+    public var description: String {
+        return "SyncStatus(connected=\(connected), connecting=\(connecting), downloading=\(downloading), uploading=\(uploading), lastSyncedAt=\(String(describing: lastSyncedAt)), hasSynced=\(String(describing: hasSynced)), error=\(String(describing: anyError)))"
+    }
+    
+    /// Creates an empty sync status instance
+    public static func empty() -> SyncStatus {
+        return SyncStatus()
+    }
+}