Matt's minimalist GRDB CloudKit sync layer

App.swift

// fragment of main app file, to doing manual CloudKit fetches on appear, 
// because can't always rely on CloudKit subscription being timely

@main
struct LifeBalanceApp: App {
    @Environment(\.scenePhase) private var scenePhase
    
    var body: some Scene {
        WindowGroup {
            TodayTab().environmentObject(Session.highlander)
        }
        .onChange(of: scenePhase) { phase in
            switch phase {
            case .active:
                Task.init(priority: .utility) {
                    await Session.highlander.ckCoordinator.fetchChanges()
                }
            case .background:
                break
            case .inactive:
                break
            @unknown default:
                break
            }
        }
    }
}

CKCoordinator.swift

//
//  CKCoordinator.swift
//  LifeBalance
//
//  Created by Matt Greenfield on 21/1/22.
//

import GRDB
import Combine
import CloudKit
import WidgetKit

struct CKConfig {
    let container: CKContainer
    let zoneId: CKRecordZone.ID
    let subscriptionId: String
    var database: CKDatabase { return container.privateCloudDatabase }
}

public class CKCoordinator: TransactionObserver {
    
    let config: CKConfig
    var observers: [AnyCancellable] = []
    var haveTheZone = false
    var fetchingChanges = false

    // MARK: -
    
    init(config: CKConfig) {
        self.config = config
        startup()
    }
    
    // MARK: -
    
    func startup() {
        Task.init(priority: .utility) {
            Database.pool.add(transactionObserver: self)
            
            do {
                try await createTheZone()
                try await createTheSubscription()
                _ = await fetchChanges()
                
            } catch {
                logger.error(error, subsystem: .cloudkit)
            }
        }
    }
    
    // MARK: - Local -> Remote
    
    private var sendingQueuedRecords = false
    
    private func sendQueuedRecords() {
        guard haveTheZone else { return }
        if sendingQueuedRecords { return }

        Task.init(priority: .utility) {
            sendingQueuedRecords = true
            do {
                let queueEntries = try await Database.pool.read { try CKQueueEntry.fetchAll($0) }
                logger.info("sendQueuedRecords() queueEntries: \(queueEntries.count)", subsystem: .cloudkit)
                for entry in queueEntries {
                    var local: CKObject?
                    if entry.tableName == "Activity" {
                        local = try await Database.pool.read {
                            try Activity.filter(sql: "rowid = ?", arguments: [entry.localRowId]).fetchOne($0)
                        }
                    } else if entry.tableName == "LogEntry" {
                        local = try await Database.pool.read {
                            try LogEntry.filter(sql: "rowid = ?", arguments: [entry.localRowId]).fetchOne($0)
                        }
                    } else if entry.tableName == "Settings" {
                        local = try await Database.pool.read {
                            try Settings.filter(sql: "rowid = ?", arguments: [entry.localRowId]).fetchOne($0)
                        }
                    }
                    if let local = local {
                        // skip recently received from remote
                        if let incoming = fetchedChanges.first(where: { $0.ckRecord.recordID.recordName == local.ckRecordName }) {
                            fetchedChanges.remove(incoming)
                        } else { // send to remote
                            try await send(local)
                        }
                    }
                    _ = try await Database.pool.write { try entry.delete($0) }
                }
            } catch {
                logger.error(error, subsystem: .cloudkit)
            }
            sendingQueuedRecords = false
        }
    }
    
    private func send(_ local: CKObject) async throws {
        let remote: CKRecord
        let recordId = CKRecord.ID(recordName: local.ckRecordName, zoneID: config.zoneId)
        if let record = try? await config.database.record(for: recordId) { // existing CKRecord
            remote = record
        } else { // need new CKRecord
            let recordType = "\(type(of: local).self)"
            remote = CKRecord(recordType: recordType, recordID: recordId)
        }
        
        // update and save the CKRecord
        local.encode(to: remote)
        try await config.database.save(remote)
        logger.info("SENT \(type(of: local).self) (\(recordId.recordName))", subsystem: .cloudkit)
    }
    
    // MARK: - TransactionObserver
    
    public func observes(eventsOfKind eventKind: DatabaseEventKind) -> Bool {
        return eventKind.tableName != "CKQueueEntry"
    }
    
    public func databaseDidChange(with event: DatabaseEvent) {
        Task.init(priority: .utility) {
            let queueEntry = CKQueueEntry(tableName: event.tableName, localRowId: event.rowID)
            try await Database.pool.write { try queueEntry.save($0) }
            sendQueuedRecords()
        }
    }
    
    public func databaseDidCommit(_ db: GRDB.Database) {}
    
    public func databaseDidRollback(_ db: GRDB.Database) {}
    
    // MARK: - Remote -> Local
    
    private var fetchedChanges: Set<FetchedChange> = []

    func fetchChanges() async -> Result<Void, Error> {
        if fetchingChanges { return .success(()) }

        fetchingChanges = true

        do {
            logger.info("fetchChanges() lastChangeToken: \(String(describing: lastChangeToken))", subsystem: .cloudkit)
            
            let result = try await config.database.recordZoneChanges(inZoneWith: config.zoneId, since: lastChangeToken)
            let (changes, deletions, changeToken, moreComing) = result
            
            logger.info("fetchChanges() changes: \(changes.count), deletions: \(deletions.count), moreComing: \(moreComing)", subsystem: .cloudkit)
            
            for (_, result) in changes {
                switch result {
                case .success(let change):
                    fetchedChanges.insert(FetchedChange(ckRecord: change.record))
                case .failure(let error):
                    logger.error(error, subsystem: .cloudkit)
                }
            }
            
            lastChangeToken = changeToken
            fetchingChanges = false

            if moreComing {
                return await fetchChanges()
                
            } else {
                await processPendingCKRecords()
                return .success(())
            }
                
        } catch CKError.changeTokenExpired {
            logger.error("CKError.changeTokenExpired", subsystem: .cloudkit)
            lastChangeToken = nil
            return await fetchChanges()
            
        } catch {
            logger.error(error, subsystem: .cloudkit)
            return .failure(error)
        }
    }
    
    private func processPendingCKRecords() async {
        do {
            let todos = fetchedChanges.filter { $0.state == .received }
            for var change in todos {
                switch change.ckRecord.recordType {
                case "Activity":
                    change.localRecord = Activity()
                case "LogEntry":
                    change.localRecord = LogEntry(activityId: "", date: Date(), duration: 0)
                case "Settings":
                    change.localRecord = Settings()
                default: fatalError()
                }
                
                change.state = .willUpdate
                fetchedChanges.remove(change)
                fetchedChanges.insert(change)
                logger.info("RECEIVED \(change.ckRecord.recordType): \(change.ckRecord.recordID.recordName)", subsystem: .cloudkit)
            }
            
            // write them all to local db
            try await Database.pool.write { db in
                let todos = self.fetchedChanges.filter { $0.state == .willUpdate }
                for var change in todos {
                    change.localRecord?.update(from: change.ckRecord)
                    try change.localRecord?.save(db)
                    change.state = .didUpdate
                    self.fetchedChanges.remove(change)
                    self.fetchedChanges.insert(change)
                    logger.info("UPDATED \(change.ckRecord.recordType): \(change.ckRecord.recordID.recordName)", subsystem: .cloudkit)

                    // TODO: this should be handled inside Session (observer?) not here
                    if change.ckRecord.recordType == "Settings" {
                        Task { @MainActor in Session.highlander.loadSettings() }
                    }
                }
            }
            
            // widget needs update
            if !todos.isEmpty {
                WidgetCenter.shared.reloadAllTimelines()
            }
            
        } catch {
            logger.error(error, subsystem: .cloudkit)
        }
    }
    
    // MARK: -
    
    private var lastChangeToken: CKServerChangeToken? {
        get {
            guard let data = UserDefaults(suiteName: "group.LifeBalance")?.data(forKey: "lastChangeToken") else { return nil }
            return try? NSKeyedUnarchiver.unarchivedObject(ofClass: CKServerChangeToken.self, from: data)
        }
        
        set(newValue) {
            if let newValue = newValue {
                let data = try! NSKeyedArchiver.archivedData(withRootObject: newValue, requiringSecureCoding: true)
                UserDefaults(suiteName: "group.LifeBalance")?.set(data, forKey: "lastChangeToken")
            } else {
                UserDefaults(suiteName: "group.LifeBalance")?.set(nil, forKey: "lastChangeToken")
            }
        }
    }
    
    struct FetchedChange: Hashable {
        enum State { case received, willUpdate, didUpdate }
        var state: State = .received
        var ckRecord: CKRecord
        var localRecord: CKObject? = nil
        
        static func == (lhs: CKCoordinator.FetchedChange, rhs: CKCoordinator.FetchedChange) -> Bool {
            return lhs.ckRecord.recordID == rhs.ckRecord.recordID && lhs.ckRecord.recordChangeTag == rhs.ckRecord.recordChangeTag
        }
        
        func hash(into hasher: inout Hasher) {
            hasher.combine(ckRecord.recordID.recordName)
            hasher.combine(ckRecord.recordChangeTag)
        }
    }

    // MARK: - CloudKit user

    private var _userId: CKRecord.ID?

    func userId() async -> CKRecord.ID? {
        if let id = _userId { return id }
        do {
            _userId = try await config.container.userRecordID()
        } catch {
            logger.error(error, subsystem: .cloudkit)
        }
        return _userId
    }

    func userRecord() async -> CKRecord? {
        guard let userId = await userId() else { return nil }
        do {
            return try await config.database.record(for: userId)
        } catch {
            logger.error(error, subsystem: .cloudkit)
            return nil
        }
    }

    // MARK: - First time setup
    
    private func createTheZone() async throws {
        do {
            try await config.database.recordZone(for: config.zoneId)
            haveTheZone = true
            
        } catch {
            logger.info("saving the zone", subsystem: .cloudkit)
            try await config.database.save(CKRecordZone(zoneID: config.zoneId))
            logger.info("saved the zone", subsystem: .cloudkit)
            haveTheZone = true
        }
    }
    
    private func createTheSubscription() async throws {
        do {
            _ = try await config.database.subscription(for: config.subscriptionId)
        
        } catch {
            let subscription = CKRecordZoneSubscription(zoneID: config.zoneId, subscriptionID: config.subscriptionId)
            let noteInfo = CKSubscription.NotificationInfo()
            noteInfo.shouldSendContentAvailable = true
            subscription.notificationInfo = noteInfo
            
            logger.info("saving the subscription", subsystem: .cloudkit)
            try await config.database.save(subscription)
            logger.info("saved the subscription", subsystem: .cloudkit)
        }
    }
    // MARK: - Debug
    
    func debugResetLocal() {
        UserDefaults.standard.removeObject(forKey: "lastChangeToken")
    }
    
    func debugResetRemote() {
        config.database.delete(withRecordZoneID: config.zoneId) { zoneId, error in
            if let error = error {
                debugPrint("ERROR: \(error)")
            }
        }
    }
    
}

CKObject.swift

//
//  CKObject.swift
//  LifeBalance
//
//  Created by Matt Greenfield on 20/1/22.
//

import GRDB
import CloudKit

protocol CKObject: PersistableRecord {
    var ckRecordName: String { get set }
    
    func encode(to record: CKRecord)
    mutating func update(from record: CKRecord)
}

CKQueueEntry.swift

//
//  CKQueueEntry.swift
//  LifeBalance
//
//  Created by Matt Greenfield on 1/2/22.
//

import GRDB
import Foundation

struct CKQueueEntry: Identifiable, Codable, Equatable, TableRecord, FetchableRecord, PersistableRecord {
    var id: Int64?
    var date: Date = Date()
    var tableName: String
    var localRowId: Int64
}

Database.swift

// fragment of db migration code, for the pending changes table

migrator.registerMigration("v1.0") { db in
    try db.create(table: "CKQueueEntry") { table in
        table.autoIncrementedPrimaryKey("id")
        table.column("date", .datetime).notNull()
        table.column("TableName", .text).notNull()
        table.column("localRowId", .numeric).notNull()
    }
}