refactor: CloudSyncContainer
This commit is contained in:
1 parent
60d8822df9
commit
aeedc538a9
14 files changed
+321
-478
No files matched your search
@@ -343,7 +343,7 @@ private final class HistoryFixture {
|
||||
let device: HistoryEnvelope
|
||||
var scope: CloudSyncScope {
|
||||
CloudSyncScope(
|
||||
container: HistoryRecordSyncCoordinator.containerIdentifier, environment: "Test", account: "current-user",
|
||||
container: CloudSyncContainer.identifier, environment: "Test", account: "current-user",
|
||||
zone: HistoryRecordSyncCoordinator.zoneName, owner: CKCurrentUserDefaultName)
|
||||
}
|
||||
init() throws {
|
||||
|
||||
@@ -122,7 +122,10 @@ struct SFTPSettingsSyncTests {
|
||||
record["payload"] = bytes as NSData
|
||||
record["payloadHash"] = SHA256.hash(data: bytes).map { String(format: "%02x", $0) }.joined() as NSString
|
||||
#expect(throws: SettingsSyncPersistenceError.self) { try SettingsSyncCoordinator.envelope(from: record) }
|
||||
#expect(try SettingsSyncCoordinator.payloadData(from: record) == bytes)
|
||||
#expect(
|
||||
try CloudSyncRecordCoding.payloadData(
|
||||
from: record, invalidEnvelope: SettingsSyncPersistenceError.invalidEnvelope(recordID.recordName))
|
||||
== bytes)
|
||||
#expect(throws: SettingsSyncPersistenceError.self) { try fixture.store.applyRemote([invalid]) }
|
||||
#expect(try fixture.store.loadSFTP() == SFTPPreferences())
|
||||
#expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 30)
|
||||
@@ -134,7 +137,7 @@ struct SFTPSettingsSyncTests {
|
||||
let wrong = CKRecord(recordType: SettingsSyncCoordinator.recordType, recordID: recordID)
|
||||
for field in record.allKeys() { wrong[field] = record[field] }
|
||||
#expect(throws: SettingsSyncPersistenceError.self) { try SettingsSyncCoordinator.envelope(from: wrong) }
|
||||
let fields = try SettingsSyncCoordinator.systemFields(of: wrong)
|
||||
let fields = try CloudSyncRecordCoding.systemFields(of: wrong)
|
||||
#expect(throws: SettingsSyncPersistenceError.self) {
|
||||
try SettingsSyncCoordinator.makeRecord(
|
||||
.init(envelope: envelope, systemFieldsData: fields), recordID: recordID)
|
||||
|
||||
@@ -9,19 +9,11 @@ struct CloudSyncSentChanges {
|
||||
|
||||
var savedRecords: [CKRecord]
|
||||
var failedRecordSaves: [FailedSave]
|
||||
var deletedRecordIDs: [CKRecord.ID]
|
||||
var failedRecordDeletes: [CKRecord.ID: CKError]
|
||||
|
||||
init(
|
||||
savedRecords: [CKRecord], failedRecordSaves: [FailedSave], deletedRecordIDs: [CKRecord.ID] = [],
|
||||
failedRecordDeletes: [CKRecord.ID: CKError] = [:]
|
||||
) {
|
||||
self.savedRecords = savedRecords
|
||||
self.failedRecordSaves = failedRecordSaves
|
||||
self.deletedRecordIDs = deletedRecordIDs
|
||||
self.failedRecordDeletes = failedRecordDeletes
|
||||
}
|
||||
var deletedRecordIDs: [CKRecord.ID] = []
|
||||
var failedRecordDeletes: [CKRecord.ID: CKError] = [:]
|
||||
}
|
||||
|
||||
extension CloudSyncSentChanges {
|
||||
init(_ changes: CKSyncEngine.Event.SentRecordZoneChanges) {
|
||||
self.init(
|
||||
savedRecords: changes.savedRecords,
|
||||
@@ -30,14 +22,14 @@ struct CloudSyncSentChanges {
|
||||
}
|
||||
}
|
||||
|
||||
/// A constructible projection of the CKSyncEngine events used by writable sync categories.
|
||||
/// Keeping this value separate from CloudKit's event constructors gives the shared dispatcher a deterministic test seam.
|
||||
enum CloudSyncZoneDeletionReason: Sendable {
|
||||
case deleted
|
||||
case purged
|
||||
case encryptedDataReset
|
||||
}
|
||||
|
||||
/// A constructible projection of the CKSyncEngine events used by writable sync categories.
|
||||
/// Keeping this value separate from CloudKit's event constructors gives the shared dispatcher a deterministic test seam.
|
||||
enum CloudSyncEventInput {
|
||||
case stateUpdate(CKSyncEngine.State.Serialization)
|
||||
case accountChange(CKSyncEngine.Event.AccountChange.ChangeType)
|
||||
@@ -159,29 +151,28 @@ extension CloudSyncLifecycle {
|
||||
guard let scope else { throw CancellationError() }
|
||||
// A save result proves that the ID exists after any previously handed-off delete.
|
||||
noteRecordsObserved(changes.savedRecords)
|
||||
for id in changes.deletedRecordIDs where id.zoneID == zoneID {
|
||||
guard let mutation = takeDeletionAcknowledgement(id) else { continue }
|
||||
guard try !store.safety.isUploadBlocked(scope, id.recordName) else { continue }
|
||||
// A delete of an already missing record succeeded as far as this replica is concerned.
|
||||
let deleted =
|
||||
changes.deletedRecordIDs
|
||||
+ changes.failedRecordDeletes.filter { $0.value.code == .unknownItem }.map(\.key)
|
||||
for id in deleted where id.zoneID == zoneID {
|
||||
guard let mutation = takeDeletionAcknowledgement(id),
|
||||
try !store.safety.isUploadBlocked(scope, id.recordName)
|
||||
else { continue }
|
||||
try acknowledge(mutation)
|
||||
}
|
||||
for (id, error) in changes.failedRecordDeletes where id.zoneID == zoneID {
|
||||
if error.code == .unknownItem {
|
||||
guard let mutation = takeDeletionAcknowledgement(id) else { continue }
|
||||
guard try !store.safety.isUploadBlocked(scope, id.recordName) else { continue }
|
||||
try acknowledge(mutation)
|
||||
for (id, error) in changes.failedRecordDeletes where id.zoneID == zoneID && error.code != .unknownItem {
|
||||
let mutation = takeDeletionAcknowledgement(id)
|
||||
if error.code == .zoneNotFound {
|
||||
try handleZoneDeletion()
|
||||
} else if Self.isRetryable(error) || Self.isCancellation(error) {
|
||||
// The server may have committed an ambiguous delete. Release only
|
||||
// the in-memory causal slot so it can retry; keep the durable fence.
|
||||
} else {
|
||||
let mutation = takeDeletionAcknowledgement(id)
|
||||
if error.code == .zoneNotFound {
|
||||
try handleZoneDeletion()
|
||||
} else if Self.isRetryable(error) || Self.isCancellation(error) {
|
||||
// The server may have committed an ambiguous delete. Release only
|
||||
// the in-memory causal slot so it can retry; keep the durable fence.
|
||||
} else {
|
||||
if let mutation { try cancelDeletionHandoff(mutation) }
|
||||
logFailure(error, stage: "delete")
|
||||
terminalError = error.localizedDescription
|
||||
status = .failed(error.localizedDescription)
|
||||
}
|
||||
if let mutation { try cancelDeletionHandoff(mutation) }
|
||||
logFailure(error, stage: "delete")
|
||||
terminalError = error.localizedDescription
|
||||
status = .failed(error.localizedDescription)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
import CloudKit
|
||||
import Foundation
|
||||
import TermindDomain
|
||||
import TermindPersistence
|
||||
|
||||
nonisolated protocol CloudSyncRecordKey: Hashable {
|
||||
var recordName: String { get }
|
||||
}
|
||||
|
||||
nonisolated protocol CloudSyncEnvelope: Equatable {
|
||||
associatedtype Key: CloudSyncRecordKey
|
||||
var key: Key { get }
|
||||
var version: VaultVersion { get }
|
||||
var isDeleted: Bool { get }
|
||||
}
|
||||
|
||||
/// The engine binding every category journal persists beside its records.
|
||||
protocol CloudSyncJournalState: AnyObject {
|
||||
var engineStateData: Data? { get set }
|
||||
var boundUserRecordName: String? { get set }
|
||||
var pendingUserRecordName: String? { get set }
|
||||
var lastSuccessfulSync: Date? { get set }
|
||||
}
|
||||
|
||||
/// The journal surface shared by the category-owned sync coordinators.
|
||||
@MainActor
|
||||
protocol CloudSyncJournal: AnyObject {
|
||||
associatedtype Envelope: CloudSyncEnvelope
|
||||
associatedtype State: CloudSyncJournalState
|
||||
typealias Key = Envelope.Key
|
||||
|
||||
func state() throws -> State
|
||||
func cloudRecords() throws -> [CloudSyncJournalRecord]
|
||||
func envelope(for key: Key) throws -> Envelope?
|
||||
func markNeedsUpload(_ keys: Set<Key>) throws
|
||||
func markUploaded(_ key: Key, version: VaultVersion) throws
|
||||
func updateSystemFields(_ data: Data?, for key: Key) throws
|
||||
func acknowledgeDeletion(_ mutation: CloudSyncMutation, scope: CloudSyncScope) throws
|
||||
}
|
||||
|
||||
extension CloudSyncJournal {
|
||||
func syncState() throws -> CloudSyncState {
|
||||
let state = try state()
|
||||
return CloudSyncState(
|
||||
engineStateData: state.engineStateData, boundUserRecordName: state.boundUserRecordName,
|
||||
pendingUserRecordName: state.pendingUserRecordName, lastSuccessfulSync: state.lastSuccessfulSync)
|
||||
}
|
||||
|
||||
func updateSyncState(_ value: CloudSyncState) throws {
|
||||
let state = try state()
|
||||
state.engineStateData = value.engineStateData
|
||||
state.boundUserRecordName = value.boundUserRecordName
|
||||
state.pendingUserRecordName = value.pendingUserRecordName
|
||||
state.lastSuccessfulSync = value.lastSuccessfulSync
|
||||
}
|
||||
|
||||
/// Records the cloud metadata of a remote value inside the merge transaction.
|
||||
func noteRemoteRecord(
|
||||
_ record: CKRecord, envelope: Envelope, cloud: CloudSyncRecordStateStore, safety: CloudSyncSafetyStore
|
||||
) throws {
|
||||
let recordName = record.recordID.recordName
|
||||
try cloud.observed(recordName, version: envelope.version, current: cloudRecord(named: recordName))
|
||||
try updateSystemFields(CloudSyncRecordCoding.systemFields(of: record), for: envelope.key)
|
||||
if try self.envelope(for: envelope.key) == envelope {
|
||||
try markUploaded(envelope.key, version: envelope.version)
|
||||
}
|
||||
try safety.clearQuarantine(recordName: recordName)
|
||||
}
|
||||
|
||||
/// A remote deletion completes a pending local tombstone for the same key.
|
||||
func acknowledgeRemoteDeletions(_ keys: Set<Key>, scope: CloudSyncScope) throws {
|
||||
for key in keys {
|
||||
guard let envelope = try envelope(for: key), envelope.isDeleted else { continue }
|
||||
try acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
}
|
||||
}
|
||||
|
||||
func acknowledgeSavedRecord(_ record: CKRecord, envelope: Envelope, cloud: CloudSyncRecordStateStore) throws {
|
||||
let recordName = record.recordID.recordName
|
||||
let current = try self.envelope(for: envelope.key)
|
||||
try cloud.acknowledgeSave(
|
||||
.init(recordName: recordName, operation: .save, version: envelope.version),
|
||||
current: cloudRecord(named: recordName))
|
||||
if current?.isDeleted == true {
|
||||
// An in-flight save may finish after a local/remote deletion. Delete its server copy again.
|
||||
try markNeedsUpload([envelope.key])
|
||||
} else {
|
||||
try markUploaded(envelope.key, version: envelope.version)
|
||||
try updateSystemFields(CloudSyncRecordCoding.systemFields(of: record), for: envelope.key)
|
||||
}
|
||||
}
|
||||
|
||||
private func cloudRecord(named recordName: String) throws -> CloudSyncJournalRecord? {
|
||||
try cloudRecords().first { $0.mutation.recordName == recordName }
|
||||
}
|
||||
}
|
||||
|
||||
extension VaultRecordKey: CloudSyncRecordKey {}
|
||||
extension SettingsRecordKey: CloudSyncRecordKey {}
|
||||
extension MetricsRecordKey: CloudSyncRecordKey {}
|
||||
extension HistoryRecordKey: CloudSyncRecordKey {}
|
||||
|
||||
extension VaultEnvelope: CloudSyncEnvelope {}
|
||||
extension SettingsEnvelope: CloudSyncEnvelope {}
|
||||
extension MetricsEnvelope: CloudSyncEnvelope {}
|
||||
extension HistoryEnvelope: CloudSyncEnvelope {}
|
||||
|
||||
extension PersistentVaultSyncState: CloudSyncJournalState {}
|
||||
extension PersistentAIConfigurationSyncState: CloudSyncJournalState {}
|
||||
extension PersistentMetricsSyncState: CloudSyncJournalState {}
|
||||
extension PersistentHistorySyncState: CloudSyncJournalState {}
|
||||
|
||||
extension VaultSyncJournal: CloudSyncJournal {}
|
||||
extension SettingsSyncJournal: CloudSyncJournal {}
|
||||
extension MetricsSyncJournal: CloudSyncJournal {}
|
||||
extension HistorySyncJournal: CloudSyncJournal {}
|
||||
@@ -254,7 +254,7 @@ final class CloudSyncLifecycle {
|
||||
func noteLocalChanges(_ recordNames: [String]) {
|
||||
guard !recordNames.isEmpty else { return }
|
||||
do {
|
||||
try enqueue(recordNames)
|
||||
try reconcilePendingChanges()
|
||||
guard isEnabled else { return }
|
||||
if engine == nil {
|
||||
start()
|
||||
@@ -347,7 +347,6 @@ final class CloudSyncLifecycle {
|
||||
let requiresRefetch = try store.safety.prepare(scope, legacyScope(for: state))
|
||||
state.boundUserRecordName = currentUser
|
||||
state.pendingUserRecordName = nil
|
||||
if requiresRefetch { state.engineStateData = nil }
|
||||
try store.updateState(state)
|
||||
if defaults.string(forKey: environmentKey) != environmentIdentifier {
|
||||
try store.resetForEnvironment()
|
||||
@@ -431,7 +430,6 @@ final class CloudSyncLifecycle {
|
||||
try store.safety.setRecoveryRequired(scope, required)
|
||||
try store.save()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
extension CloudSyncLifecycle {
|
||||
@@ -447,7 +445,7 @@ extension CloudSyncLifecycle {
|
||||
let result = try replica.reconcileDeletions(scope, Array(deletedNames))
|
||||
if !result.recreated.isEmpty {
|
||||
try store.save()
|
||||
try enqueue(result.recreated)
|
||||
try reconcilePendingChanges()
|
||||
}
|
||||
let accepted = Set(result.deleted)
|
||||
return (
|
||||
@@ -492,10 +490,8 @@ extension CloudSyncLifecycle {
|
||||
fetchFailed = fetchFailed || !succeeded
|
||||
if fetching == 0 && !fetchFailed, let scope {
|
||||
if inventoryPending {
|
||||
let missing =
|
||||
try store.replica?.missing(scope, inventorySeen)
|
||||
?? .init(deleted: [], recreated: [])
|
||||
try receiveSnapshot([], missing.deleted.map { CKRecord.ID(recordName: $0, zoneID: zoneID) })
|
||||
let deleted = try store.replica?.missing(scope, inventorySeen).deleted ?? []
|
||||
try receiveSnapshot([], deleted.map { CKRecord.ID(recordName: $0, zoneID: zoneID) })
|
||||
try store.replica?.complete(scope)
|
||||
try store.save()
|
||||
inventoryPending = false
|
||||
@@ -524,11 +520,6 @@ extension CloudSyncLifecycle {
|
||||
try setRecoveryRequired(!succeeded || fetching > 0 || sending > 0 || fetchFailed)
|
||||
}
|
||||
|
||||
func enqueue(_ recordNames: [String]) throws {
|
||||
guard !recordNames.isEmpty else { return }
|
||||
try reconcilePendingChanges()
|
||||
}
|
||||
|
||||
func reconcilePendingChanges() throws {
|
||||
guard let engine, let scope else { return }
|
||||
_ = try reconcilePendingChanges(engine: engine, scope: scope)
|
||||
@@ -777,9 +768,6 @@ extension CloudSyncLifecycle {
|
||||
return .failed(error.localizedDescription)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
extension CloudSyncLifecycle {
|
||||
private func status(for accountStatus: CKAccountStatus) -> VaultSyncStatus {
|
||||
switch accountStatus {
|
||||
case .noAccount: .signInRequired
|
||||
|
||||
+33
-43
@@ -15,16 +15,14 @@ struct CloudSyncLifecycleSafety {
|
||||
|
||||
/// The lifecycle commits both legacy imports and recovery state before creating an engine.
|
||||
static func persistent(
|
||||
context: ModelContext, capabilityID: String? = nil, capabilityProvider: (() -> String)? = nil,
|
||||
importLegacyQuarantine: @escaping (CloudSyncScope) throws -> Void
|
||||
context: ModelContext, capabilityID: String? = nil,
|
||||
importLegacyQuarantine: @escaping (CloudSyncScope) throws -> Void = { _ in }
|
||||
) -> Self {
|
||||
Self(
|
||||
prepare: { scope, legacyScope in
|
||||
try importLegacyQuarantine(legacyScope)
|
||||
let safety = CloudSyncSafetyStore(context: context, scope: scope)
|
||||
if let expected = capabilityProvider?() ?? capabilityID,
|
||||
try safety.capability().capabilityID != expected
|
||||
{
|
||||
if let capabilityID, try safety.capability().capabilityID != capabilityID {
|
||||
return true
|
||||
}
|
||||
return try safety.recovery().requiresRefetch
|
||||
@@ -42,23 +40,32 @@ struct CloudSyncLifecycleSafety {
|
||||
try CloudSyncSafetyStore(context: context, scope: scope).requiresNewerClient()
|
||||
},
|
||||
didCompleteFetch: { scope in
|
||||
if let expected = capabilityProvider?() ?? capabilityID {
|
||||
try CloudSyncSafetyStore(context: context, scope: scope).capability().capabilityID = expected
|
||||
if let capabilityID {
|
||||
try CloudSyncSafetyStore(context: context, scope: scope).capability().capabilityID = capabilityID
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// History and devices share a zone, so each kind reports only its own blocked records.
|
||||
static func persistentHistory(context: ModelContext, kind: HistoryRecordKind) -> Self {
|
||||
var safety = persistent(
|
||||
context: context, capabilityID: SyncRecordFormats.history.capabilityID + ";kind:" + kind.rawValue)
|
||||
safety.blockedRecordNames = { scope in
|
||||
try CloudSyncSafetyStore(context: context, scope: scope).blockedRecordNames().filter {
|
||||
HistoryRecordKey(recordName: $0).map { $0.kind == kind } ?? true
|
||||
}
|
||||
}
|
||||
safety.requiresNewerClient = { [blockedRecordNames = safety.blockedRecordNames] scope in
|
||||
let store = CloudSyncSafetyStore(context: context, scope: scope)
|
||||
return try blockedRecordNames(scope).contains {
|
||||
try store.entry(recordName: $0)?.reasonCode == CloudSyncQuarantineReason.unsupportedFormat.rawValue
|
||||
}
|
||||
}
|
||||
return safety
|
||||
}
|
||||
}
|
||||
|
||||
extension CloudSyncRecordCoding {
|
||||
nonisolated static func validateFormat(in data: Data, supported: Int) throws {
|
||||
struct Header: Decodable { var formatVersion: Int }
|
||||
let header = try JSONDecoder().decode(Header.self, from: data)
|
||||
if header.formatVersion > supported {
|
||||
throw CloudSyncCompatibilityError.unsupportedFormat(header.formatVersion)
|
||||
}
|
||||
guard header.formatVersion == supported else { throw CocoaError(.coderReadCorrupt) }
|
||||
}
|
||||
|
||||
@MainActor
|
||||
static func quarantine(
|
||||
_ record: CKRecord, error: any Error, in safety: CloudSyncSafetyStore, nativeFields: Bool = false
|
||||
@@ -81,33 +88,16 @@ extension CloudSyncRecordCoding {
|
||||
systemFieldsData: try systemFields(of: record),
|
||||
reason: error.localizedDescription, reasonCode: reasonCode)
|
||||
}
|
||||
}
|
||||
|
||||
extension CloudSyncLifecycleSafety {
|
||||
static func persistentHistory(context: ModelContext, kind: HistoryRecordKind) -> Self {
|
||||
var safety = persistent(
|
||||
context: context,
|
||||
capabilityID: SyncRecordFormats.history.capabilityID + ";kind:" + kind.rawValue,
|
||||
importLegacyQuarantine: { _ in })
|
||||
let allBlocked = safety.blockedRecordNames
|
||||
safety.blockedRecordNames = { scope in
|
||||
Set(
|
||||
try allBlocked(scope).filter {
|
||||
HistoryRecordKey(recordName: $0).map { $0.kind == kind } ?? true
|
||||
})
|
||||
}
|
||||
safety.requiresNewerClient = { scope in
|
||||
let store = CloudSyncSafetyStore(context: context, scope: scope)
|
||||
for name in try store.blockedRecordNames()
|
||||
where HistoryRecordKey(recordName: name).map({ $0.kind == kind }) ?? true {
|
||||
if try store.entry(recordName: name)?.reasonCode
|
||||
== CloudSyncQuarantineReason.unsupportedFormat.rawValue
|
||||
{
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
return safety
|
||||
/// Finishes an asset cache purge that a committed cache purge marked as pending.
|
||||
@MainActor
|
||||
static func completePendingAssetPurge(
|
||||
directoryName: String, context: ModelContext, cacheRoot: URL?, removeItem: (URL) throws -> Void
|
||||
) throws {
|
||||
let cleanup = CloudSyncAssetCleanupStore(context: context)
|
||||
guard try cleanup.isPending(cacheKey: directoryName) else { return }
|
||||
try purgeAssetDirectory(directoryName: directoryName, cacheRoot: cacheRoot, removeItem: removeItem)
|
||||
try cleanup.complete(cacheKey: directoryName)
|
||||
try context.save()
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,16 @@
|
||||
import CloudKit
|
||||
import CryptoKit
|
||||
import Foundation
|
||||
import TermindDomain
|
||||
|
||||
/// The CloudKit container shared by every sync category.
|
||||
public nonisolated enum CloudSyncContainer {
|
||||
public static let identifier = "iCloud.com.akinokaede.termind"
|
||||
|
||||
public static var environmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String ?? "Unknown"
|
||||
}
|
||||
}
|
||||
|
||||
/// CloudKit transport mechanics shared by the independently owned sync coordinators.
|
||||
nonisolated enum CloudSyncRecordCoding {
|
||||
@@ -16,6 +26,20 @@ nonisolated enum CloudSyncRecordCoding {
|
||||
return CKRecord(recordType: type, recordID: id)
|
||||
}
|
||||
|
||||
/// Queryable metadata written beside the JSON payload of Vault, Settings, and Metrics records.
|
||||
static func setMetadata(
|
||||
on record: CKRecord, formatVersion: Int, kind: String, entityID: String, version: VaultVersion,
|
||||
isDeleted: Bool
|
||||
) {
|
||||
record["formatVersion"] = NSNumber(value: formatVersion)
|
||||
record["kind"] = kind as NSString
|
||||
record["entityID"] = entityID as NSString
|
||||
record["modifiedAt"] = version.modifiedAt as NSDate
|
||||
record["counter"] = NSNumber(value: Int64(clamping: version.counter))
|
||||
record["writerID"] = version.writerID.uuidString.lowercased() as NSString
|
||||
record["isDeleted"] = NSNumber(value: isDeleted)
|
||||
}
|
||||
|
||||
static func setPayload(
|
||||
_ payload: Data, on record: CKRecord, assetURL: @autoclosure () throws -> URL
|
||||
) throws {
|
||||
@@ -39,6 +63,15 @@ nonisolated enum CloudSyncRecordCoding {
|
||||
return data
|
||||
}
|
||||
|
||||
static func validateFormat(in data: Data, supported: Int) throws {
|
||||
struct Header: Decodable { var formatVersion: Int }
|
||||
let header = try JSONDecoder().decode(Header.self, from: data)
|
||||
if header.formatVersion > supported {
|
||||
throw CloudSyncCompatibilityError.unsupportedFormat(header.formatVersion)
|
||||
}
|
||||
guard header.formatVersion == supported else { throw CocoaError(.coderReadCorrupt) }
|
||||
}
|
||||
|
||||
/// Raw bytes remain available to quarantine even when the payload hash is invalid.
|
||||
static func payloadData(
|
||||
from record: CKRecord, invalidEnvelope: @autoclosure () -> any Error
|
||||
|
||||
@@ -5,10 +5,7 @@ import TermindPersistence
|
||||
|
||||
@MainActor
|
||||
struct CloudSyncReplicaStore {
|
||||
struct MissingRecords {
|
||||
var deleted: [String]
|
||||
var recreated: [String]
|
||||
}
|
||||
typealias MissingRecords = (deleted: [String], recreated: [String])
|
||||
|
||||
var prepare: (CloudSyncScope) throws -> Bool
|
||||
var handlesRecordName: (String) -> Bool
|
||||
@@ -25,32 +22,30 @@ struct CloudSyncReplicaStore {
|
||||
handlesRecordName: @escaping (String) -> Bool = { _ in true },
|
||||
preservesNewerIncarnation: @escaping (String) -> Bool = { _ in true }
|
||||
) -> Self {
|
||||
Self(
|
||||
prepare: { try CloudSyncRecordStateStore(context: context, scope: $0).prepare(records()) },
|
||||
func store(_ scope: CloudSyncScope) -> CloudSyncRecordStateStore {
|
||||
CloudSyncRecordStateStore(context: context, scope: scope)
|
||||
}
|
||||
// A recreated record is a new incarnation; its old system fields would target the deleted record.
|
||||
func clearingRecreated(_ result: MissingRecords) throws -> MissingRecords {
|
||||
for name in result.recreated { try clearSystemFields(name) }
|
||||
return result
|
||||
}
|
||||
return Self(
|
||||
prepare: { try store($0).prepare(records()) },
|
||||
handlesRecordName: handlesRecordName,
|
||||
missing: { scope, seen in
|
||||
let result = try CloudSyncRecordStateStore(context: context, scope: scope)
|
||||
.missingRecordNames(
|
||||
seen: seen, records: records(),
|
||||
preservesNewerIncarnation: preservesNewerIncarnation)
|
||||
for name in result.recreated { try clearSystemFields(name) }
|
||||
return MissingRecords(deleted: result.deleted, recreated: result.recreated)
|
||||
try clearingRecreated(
|
||||
store(scope).missingRecordNames(
|
||||
seen: seen, records: records(), preservesNewerIncarnation: preservesNewerIncarnation))
|
||||
},
|
||||
reconcileDeletions: { scope, names in
|
||||
let result = try CloudSyncRecordStateStore(context: context, scope: scope)
|
||||
.reconcileDeletions(
|
||||
names, records: records(),
|
||||
preservesNewerIncarnation: preservesNewerIncarnation)
|
||||
for name in result.recreated { try clearSystemFields(name) }
|
||||
return MissingRecords(deleted: result.deleted, recreated: result.recreated)
|
||||
try clearingRecreated(
|
||||
store(scope).reconcileDeletions(
|
||||
names, records: records(), preservesNewerIncarnation: preservesNewerIncarnation))
|
||||
},
|
||||
recordDeletionHandoff: {
|
||||
try CloudSyncRecordStateStore(context: context, scope: $0).recordDeletionHandoff($1)
|
||||
},
|
||||
cancelDeletionHandoff: {
|
||||
try CloudSyncRecordStateStore(context: context, scope: $0).cancelDeletionHandoff($1)
|
||||
},
|
||||
complete: { try CloudSyncRecordStateStore(context: context, scope: $0).completeReconciliation() },
|
||||
reset: { try CloudSyncRecordStateStore(context: context, scope: $0).resetZone() })
|
||||
recordDeletionHandoff: { try store($0).recordDeletionHandoff($1) },
|
||||
cancelDeletionHandoff: { try store($0).cancelDeletionHandoff($1) },
|
||||
complete: { try store($0).completeReconciliation() },
|
||||
reset: { try store($0).resetZone() })
|
||||
}
|
||||
}
|
||||
@@ -7,10 +7,8 @@ import TermindPersistence
|
||||
/// Independently synchronizes connection events under the History selection.
|
||||
@MainActor
|
||||
public final class ConnectionHistorySyncCoordinator {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
// Preserve the deployed zone so fresh installations can read previously uploaded records.
|
||||
public nonisolated static let zoneName = "TermindConnectionHistoryV1"
|
||||
public nonisolated static let recordType = HistoryRecordKind.connectionHistory.recordType
|
||||
public nonisolated static let zoneName = HistoryRecordSyncCoordinator.zoneName
|
||||
public nonisolated static let subscriptionID = "termind-connection-history-v1-changes"
|
||||
|
||||
let core: HistoryRecordSyncCoordinator
|
||||
@@ -23,9 +21,9 @@ public final class ConnectionHistorySyncCoordinator {
|
||||
|
||||
public convenience init(
|
||||
persistence: HistorySyncPersistence,
|
||||
container: CKContainer = CKContainer(identifier: ConnectionHistorySyncCoordinator.containerIdentifier),
|
||||
container: CKContainer = CKContainer(identifier: CloudSyncContainer.identifier),
|
||||
defaults: UserDefaults = .standard,
|
||||
environmentIdentifier: String = ConnectionHistorySyncCoordinator.runtimeEnvironmentIdentifier
|
||||
environmentIdentifier: String = CloudSyncContainer.environmentIdentifier
|
||||
) {
|
||||
self.init(
|
||||
persistence: persistence, defaults: defaults, environmentIdentifier: environmentIdentifier,
|
||||
@@ -43,10 +41,6 @@ public final class ConnectionHistorySyncCoordinator {
|
||||
environmentIdentifier: environmentIdentifier, makeTransport: makeTransport)
|
||||
}
|
||||
|
||||
public nonisolated static var runtimeEnvironmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String ?? "Unknown"
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) { core.configureLogging(logging) }
|
||||
public func start() { core.start() }
|
||||
public func setEnabled(_ enabled: Bool) { core.setEnabled(enabled) }
|
||||
|
||||
@@ -7,10 +7,8 @@ import TermindPersistence
|
||||
/// Independently synchronizes the device directory under the master iCloud selection.
|
||||
@MainActor
|
||||
public final class DeviceSyncCoordinator {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
// Preserve the deployed zone so fresh installations can read previously uploaded records.
|
||||
public nonisolated static let zoneName = "TermindConnectionHistoryV1"
|
||||
public nonisolated static let recordType = HistoryRecordKind.device.recordType
|
||||
public nonisolated static let zoneName = HistoryRecordSyncCoordinator.zoneName
|
||||
public nonisolated static let subscriptionID = "termind-device-v1-changes"
|
||||
|
||||
let core: HistoryRecordSyncCoordinator
|
||||
@@ -23,9 +21,9 @@ public final class DeviceSyncCoordinator {
|
||||
|
||||
public convenience init(
|
||||
persistence: HistorySyncPersistence,
|
||||
container: CKContainer = CKContainer(identifier: DeviceSyncCoordinator.containerIdentifier),
|
||||
container: CKContainer = CKContainer(identifier: CloudSyncContainer.identifier),
|
||||
defaults: UserDefaults = .standard,
|
||||
environmentIdentifier: String = DeviceSyncCoordinator.runtimeEnvironmentIdentifier
|
||||
environmentIdentifier: String = CloudSyncContainer.environmentIdentifier
|
||||
) {
|
||||
self.init(
|
||||
persistence: persistence, defaults: defaults, environmentIdentifier: environmentIdentifier,
|
||||
@@ -43,13 +41,8 @@ public final class DeviceSyncCoordinator {
|
||||
environmentIdentifier: environmentIdentifier, makeTransport: makeTransport)
|
||||
}
|
||||
|
||||
public nonisolated static var runtimeEnvironmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String ?? "Unknown"
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) { core.configureLogging(logging) }
|
||||
public func start() { core.start() }
|
||||
|
||||
public func pause() { core.pause() }
|
||||
public func retry() { core.retry() }
|
||||
public func confirmAccountSwitchMerge() { core.confirmAccountSwitchMerge() }
|
||||
|
||||
@@ -10,9 +10,8 @@ import TermindPersistence
|
||||
@MainActor
|
||||
@Observable
|
||||
final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchecked Sendable {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
public nonisolated static let zoneName = "TermindConnectionHistoryV1"
|
||||
public nonisolated static let recordTypes = Set(HistoryRecordKind.allCases.map(\.recordType))
|
||||
nonisolated static let zoneName = "TermindConnectionHistoryV1"
|
||||
nonisolated static let recordTypes = Set(HistoryRecordKind.allCases.map(\.recordType))
|
||||
|
||||
private let context: ModelContext
|
||||
private let persistence: HistorySyncPersistence
|
||||
@@ -25,13 +24,9 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
|
||||
@ObservationIgnored private(set) lazy var lifecycle = makeLifecycle()
|
||||
|
||||
public private(set) var status: VaultSyncStatus {
|
||||
get { lifecycle.status }
|
||||
set { lifecycle.status = newValue }
|
||||
}
|
||||
public var didApplyRemote: ((Set<HistoryRecordKey>) -> Void)?
|
||||
|
||||
public var isEnabled: Bool { lifecycle.isEnabled }
|
||||
var status: VaultSyncStatus { lifecycle.status }
|
||||
var didApplyRemote: ((Set<HistoryRecordKey>) -> Void)?
|
||||
var isEnabled: Bool { lifecycle.isEnabled }
|
||||
private func includes(_ key: HistoryRecordKey) -> Bool { key.kind == kind }
|
||||
|
||||
init(
|
||||
@@ -49,23 +44,17 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
super.init()
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) {
|
||||
func configureLogging(_ logging: any ApplicationLogManaging) {
|
||||
lifecycle.logger = logging.makeLogger(label: "termind.cloud-sync")
|
||||
}
|
||||
|
||||
public func start() {
|
||||
lifecycle.start()
|
||||
}
|
||||
public func setEnabled(_ enabled: Bool) { lifecycle.setEnabled(enabled) }
|
||||
public func pause() {
|
||||
lifecycle.pause()
|
||||
}
|
||||
public func retry() { lifecycle.retry() }
|
||||
public func confirmAccountSwitchMerge() {
|
||||
lifecycle.confirmAccountSwitchMerge()
|
||||
}
|
||||
func start() { lifecycle.start() }
|
||||
func setEnabled(_ enabled: Bool) { lifecycle.setEnabled(enabled) }
|
||||
func pause() { lifecycle.pause() }
|
||||
func retry() { lifecycle.retry() }
|
||||
func confirmAccountSwitchMerge() { lifecycle.confirmAccountSwitchMerge() }
|
||||
|
||||
public func noteLocalChanges(_ mutations: HistoryMutationSet) {
|
||||
func noteLocalChanges(_ mutations: HistoryMutationSet) {
|
||||
lifecycle.noteLocalChanges(mutations.keys.filter(includes).map(\.recordName))
|
||||
}
|
||||
|
||||
@@ -75,20 +64,7 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
let kind = kind
|
||||
return CloudSyncLifecycle(
|
||||
store: CloudSyncLifecycleStore(
|
||||
state: {
|
||||
let state = try journal.state()
|
||||
return CloudSyncState(
|
||||
engineStateData: state.engineStateData, boundUserRecordName: state.boundUserRecordName,
|
||||
pendingUserRecordName: state.pendingUserRecordName, lastSuccessfulSync: state.lastSuccessfulSync
|
||||
)
|
||||
},
|
||||
updateState: { value in
|
||||
let state = try journal.state()
|
||||
state.engineStateData = value.engineStateData
|
||||
state.boundUserRecordName = value.boundUserRecordName
|
||||
state.pendingUserRecordName = value.pendingUserRecordName
|
||||
state.lastSuccessfulSync = value.lastSuccessfulSync
|
||||
},
|
||||
state: journal.syncState, updateState: journal.updateSyncState,
|
||||
save: { try context.save() }, rollback: { context.rollback() },
|
||||
pendingChanges: {
|
||||
try CloudSyncRecordStateStore(context: context, scope: $0).pendingChanges(journal.cloudRecords())
|
||||
@@ -112,7 +88,7 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
kind == .device && HistoryRecordKey(recordName: recordName)?.kind == .device
|
||||
})),
|
||||
transport: makeTransport(zoneID, self),
|
||||
zoneID: zoneID, containerIdentifier: Self.containerIdentifier, defaults: defaults,
|
||||
zoneID: zoneID, containerIdentifier: CloudSyncContainer.identifier, defaults: defaults,
|
||||
enabledKey: kind == .connectionHistory ? SettingsKey.historySyncEnabled : nil,
|
||||
defaultEnabled: SettingsKey.defaultHistorySyncEnabled,
|
||||
environmentKey: kind == .connectionHistory
|
||||
@@ -122,20 +98,20 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
receiveSnapshot: { [weak self] in try self?.receiveRemoteRecords($0, deletions: $1) })
|
||||
}
|
||||
|
||||
public nonisolated func nextFetchChangesOptions(
|
||||
nonisolated func nextFetchChangesOptions(
|
||||
_ context: CKSyncEngine.FetchChangesContext, syncEngine: CKSyncEngine
|
||||
) async -> CKSyncEngine.FetchChangesOptions {
|
||||
.init(scope: .zoneIDs([CKRecordZone.ID(zoneName: Self.zoneName, ownerName: CKCurrentUserDefaultName)]))
|
||||
}
|
||||
|
||||
public nonisolated func handleEvent(_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine) async {
|
||||
nonisolated func handleEvent(_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine) async {
|
||||
await lifecycle.handle(
|
||||
CloudSyncEventInput(event), engine: syncEngine,
|
||||
receiveRemoteRecords: { try receiveRemoteRecords($0, deletions: $1) },
|
||||
handleSentChanges: handleSentChanges)
|
||||
}
|
||||
|
||||
public nonisolated func nextRecordZoneChangeBatch(
|
||||
nonisolated func nextRecordZoneChangeBatch(
|
||||
_ context: CKSyncEngine.SendChangesContext, syncEngine: CKSyncEngine
|
||||
) async -> CKSyncEngine.RecordZoneChangeBatch? {
|
||||
await makeChangeBatch(context, engine: syncEngine)
|
||||
@@ -208,22 +184,9 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
decoded.map { $0.1 }, deletions: deletedKeys,
|
||||
beforeSave: {
|
||||
for (record, envelope) in decoded {
|
||||
let current = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.observed(
|
||||
record.recordID.recordName, version: envelope.version, current: current)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
if try journal.envelope(for: envelope.key) == envelope {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
}
|
||||
try safety.clearQuarantine(recordName: record.recordID.recordName)
|
||||
}
|
||||
for key in deletedKeys {
|
||||
guard let envelope = try journal.envelope(for: key), envelope.isDeleted else { continue }
|
||||
try journal.acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
try journal.noteRemoteRecord(record, envelope: envelope, cloud: cloud, safety: safety)
|
||||
}
|
||||
try journal.acknowledgeRemoteDeletions(deletedKeys, scope: scope)
|
||||
})
|
||||
try lifecycle.reconcilePendingChanges()
|
||||
if !changed.isEmpty { didApplyRemote?(changed) }
|
||||
@@ -231,27 +194,14 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
|
||||
func handleSentChanges(_ changes: CloudSyncSentChanges) throws {
|
||||
guard let scope = lifecycle.scope else { throw CancellationError() }
|
||||
let safety = CloudSyncSafetyStore(context: context, scope: scope)
|
||||
let cloud = CloudSyncRecordStateStore(context: context, scope: scope)
|
||||
for record in changes.savedRecords
|
||||
where record.recordID.zoneID == zoneID
|
||||
&& HistoryRecordKey(recordName: record.recordID.recordName).map(includes) == true
|
||||
{
|
||||
guard try !lifecycle.store.safety.isUploadBlocked(scope, record.recordID.recordName) else { continue }
|
||||
let envelope = try Self.envelope(from: record)
|
||||
let current = try journal.envelope(for: envelope.key)
|
||||
let currentRecord = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.acknowledgeSave(
|
||||
.init(recordName: record.recordID.recordName, operation: .save, version: envelope.version),
|
||||
current: currentRecord)
|
||||
if current?.isDeleted == true {
|
||||
// An in-flight save may finish after a local/remote deletion. Delete its server copy again.
|
||||
try journal.markNeedsUpload([envelope.key])
|
||||
} else {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
}
|
||||
guard try !safety.isUploadBlocked(recordName: record.recordID.recordName) else { continue }
|
||||
try journal.acknowledgeSavedRecord(record, envelope: Self.envelope(from: record), cloud: cloud)
|
||||
}
|
||||
for failure in changes.failedRecordSaves where failure.record.recordID.zoneID == zoneID {
|
||||
if failure.error.code == .unknownItem,
|
||||
@@ -275,7 +225,7 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
}
|
||||
}
|
||||
|
||||
public nonisolated static func makeRecord(
|
||||
nonisolated static func makeRecord(
|
||||
_ outgoing: HistoryOutgoingEntry, recordID: CKRecord.ID
|
||||
) throws -> CKRecord {
|
||||
let envelope = outgoing.envelope
|
||||
@@ -285,11 +235,7 @@ final class HistoryRecordSyncCoordinator: NSObject, CKSyncEngineDelegate, @unche
|
||||
return record
|
||||
}
|
||||
|
||||
public nonisolated static func envelope(from record: CKRecord) throws -> HistoryEnvelope {
|
||||
nonisolated static func envelope(from record: CKRecord) throws -> HistoryEnvelope {
|
||||
try HistoryCloudRecordV1.read(record)
|
||||
}
|
||||
|
||||
public nonisolated static func systemFields(of record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.systemFields(of: record)
|
||||
}
|
||||
}
|
||||
@@ -10,11 +10,9 @@ import TermindPersistence
|
||||
@MainActor
|
||||
@Observable
|
||||
public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchecked Sendable {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
public nonisolated static let zoneName = "TermindMetricsV1"
|
||||
public nonisolated static let recordType = "MetricsItem"
|
||||
public nonisolated static let subscriptionID = "termind-metrics-v1-changes"
|
||||
public nonisolated static let inlinePayloadLimit = CloudSyncRecordCoding.inlinePayloadLimit
|
||||
nonisolated static let assetDirectoryName = "Termind/MetricsSyncAssets"
|
||||
|
||||
private let context: ModelContext
|
||||
@@ -29,19 +27,16 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
|
||||
@ObservationIgnored private(set) lazy var lifecycle = makeLifecycle()
|
||||
|
||||
public private(set) var status: VaultSyncStatus {
|
||||
get { lifecycle.status }
|
||||
set { lifecycle.status = newValue }
|
||||
}
|
||||
public var status: VaultSyncStatus { lifecycle.status }
|
||||
public var didApplyRemote: ((MetricsSyncSnapshot, Set<MetricsRecordKey>) -> Void)?
|
||||
|
||||
public var isEnabled: Bool { lifecycle.isEnabled }
|
||||
|
||||
public convenience init(
|
||||
persistence: MetricsSyncPersistence,
|
||||
container: CKContainer = CKContainer(identifier: MetricsSyncCoordinator.containerIdentifier),
|
||||
container: CKContainer = CKContainer(identifier: CloudSyncContainer.identifier),
|
||||
defaults: UserDefaults = .standard,
|
||||
environmentIdentifier: String = MetricsSyncCoordinator.runtimeEnvironmentIdentifier
|
||||
environmentIdentifier: String = CloudSyncContainer.environmentIdentifier
|
||||
) {
|
||||
self.init(
|
||||
persistence: persistence, defaults: defaults, environmentIdentifier: environmentIdentifier,
|
||||
@@ -69,10 +64,6 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
persistence.didRecordChanges = { [weak self] mutations in self?.noteLocalChanges(mutations) }
|
||||
}
|
||||
|
||||
public nonisolated static var runtimeEnvironmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String ?? "Unknown"
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) {
|
||||
lifecycle.logger = logging.makeLogger(label: "termind.cloud-sync")
|
||||
}
|
||||
@@ -92,20 +83,7 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
let context = context
|
||||
return CloudSyncLifecycle(
|
||||
store: CloudSyncLifecycleStore(
|
||||
state: {
|
||||
let state = try journal.state()
|
||||
return CloudSyncState(
|
||||
engineStateData: state.engineStateData, boundUserRecordName: state.boundUserRecordName,
|
||||
pendingUserRecordName: state.pendingUserRecordName, lastSuccessfulSync: state.lastSuccessfulSync
|
||||
)
|
||||
},
|
||||
updateState: { value in
|
||||
let state = try journal.state()
|
||||
state.engineStateData = value.engineStateData
|
||||
state.boundUserRecordName = value.boundUserRecordName
|
||||
state.pendingUserRecordName = value.pendingUserRecordName
|
||||
state.lastSuccessfulSync = value.lastSuccessfulSync
|
||||
},
|
||||
state: journal.syncState, updateState: journal.updateSyncState,
|
||||
save: { try context.save() }, rollback: { context.rollback() },
|
||||
pendingChanges: {
|
||||
try CloudSyncRecordStateStore(context: context, scope: $0).pendingChanges(journal.cloudRecords())
|
||||
@@ -130,20 +108,16 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
clearSystemFields: journal.clearSystemFields,
|
||||
handlesRecordName: { MetricsRecordKey(recordName: $0) != nil })),
|
||||
transport: makeTransport(zoneID, self),
|
||||
zoneID: zoneID, containerIdentifier: Self.containerIdentifier, defaults: defaults,
|
||||
zoneID: zoneID, containerIdentifier: CloudSyncContainer.identifier, defaults: defaults,
|
||||
enabledKey: SettingsKey.metricsSyncEnabled, defaultEnabled: SettingsKey.defaultMetricsSyncEnabled,
|
||||
environmentKey: SettingsKey.metricsSyncEnvironment, environmentIdentifier: environmentIdentifier,
|
||||
receiveSnapshot: { [weak self] in try self?.receiveRemoteRecords($0, deletions: $1) })
|
||||
}
|
||||
|
||||
private func completePendingAssetPurge() throws {
|
||||
let cleanup = CloudSyncAssetCleanupStore(context: context)
|
||||
guard try cleanup.isPending(cacheKey: Self.assetDirectoryName) else { return }
|
||||
try CloudSyncRecordCoding.purgeAssetDirectory(
|
||||
directoryName: Self.assetDirectoryName, cacheRoot: assetCacheRoot,
|
||||
try CloudSyncRecordCoding.completePendingAssetPurge(
|
||||
directoryName: Self.assetDirectoryName, context: context, cacheRoot: assetCacheRoot,
|
||||
removeItem: removeAssetDirectory)
|
||||
try cleanup.complete(cacheKey: Self.assetDirectoryName)
|
||||
try context.save()
|
||||
}
|
||||
|
||||
public nonisolated func nextFetchChangesOptions(
|
||||
@@ -217,22 +191,9 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
decoded.map { $0.1 }, deletions: deletedKeys,
|
||||
beforeSave: {
|
||||
for (record, envelope) in decoded {
|
||||
let current = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.observed(
|
||||
record.recordID.recordName, version: envelope.version, current: current)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
if try journal.envelope(for: envelope.key) == envelope {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
}
|
||||
try safety.clearQuarantine(recordName: record.recordID.recordName)
|
||||
}
|
||||
for key in deletedKeys {
|
||||
guard let envelope = try journal.envelope(for: key), envelope.isDeleted else { continue }
|
||||
try journal.acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
try journal.noteRemoteRecord(record, envelope: envelope, cloud: cloud, safety: safety)
|
||||
}
|
||||
try journal.acknowledgeRemoteDeletions(deletedKeys, scope: scope)
|
||||
})
|
||||
try lifecycle.reconcilePendingChanges()
|
||||
if !result.acceptedKeys.isEmpty { didApplyRemote?(result.snapshot, result.acceptedKeys) }
|
||||
@@ -245,20 +206,7 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
for record in changes.savedRecords where record.recordID.zoneID == zoneID {
|
||||
guard try !safety.isUploadBlocked(recordName: record.recordID.recordName) else { continue }
|
||||
let envelope = try Self.envelope(from: record)
|
||||
let current = try journal.envelope(for: envelope.key)
|
||||
let currentRecord = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.acknowledgeSave(
|
||||
.init(recordName: record.recordID.recordName, operation: .save, version: envelope.version),
|
||||
current: currentRecord)
|
||||
if current?.isDeleted == true {
|
||||
// An in-flight save may finish after a local/remote deletion. Delete its server copy again.
|
||||
try journal.markNeedsUpload([envelope.key])
|
||||
} else {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
}
|
||||
try journal.acknowledgeSavedRecord(record, envelope: envelope, cloud: cloud)
|
||||
try? FileManager.default.removeItem(at: Self.assetURL(for: envelope.key))
|
||||
}
|
||||
for failure in changes.failedRecordSaves where failure.record.recordID.zoneID == zoneID {
|
||||
@@ -290,13 +238,10 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
let record = try CloudSyncRecordCoding.makeRecord(
|
||||
type: recordType, id: recordID, systemFields: outgoing.systemFieldsData)
|
||||
let payload = try MetricsSyncCodec.encode(envelope)
|
||||
record["formatVersion"] = NSNumber(value: envelope.formatVersion)
|
||||
record["kind"] = envelope.key.kind.rawValue as NSString
|
||||
record["entityID"] = envelope.key.id.uuidString.lowercased() as NSString
|
||||
record["modifiedAt"] = envelope.version.modifiedAt as NSDate
|
||||
record["counter"] = NSNumber(value: Int64(clamping: envelope.version.counter))
|
||||
record["writerID"] = envelope.version.writerID.uuidString.lowercased() as NSString
|
||||
record["isDeleted"] = NSNumber(value: envelope.isDeleted)
|
||||
CloudSyncRecordCoding.setMetadata(
|
||||
on: record, formatVersion: envelope.formatVersion, kind: envelope.key.kind.rawValue,
|
||||
entityID: envelope.key.id.uuidString.lowercased(), version: envelope.version,
|
||||
isDeleted: envelope.isDeleted)
|
||||
try CloudSyncRecordCoding.setPayload(payload, on: record, assetURL: assetURL(for: envelope.key))
|
||||
return record
|
||||
}
|
||||
@@ -313,17 +258,7 @@ public final class MetricsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unch
|
||||
return envelope
|
||||
}
|
||||
|
||||
public nonisolated static func payloadData(from record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.payloadData(
|
||||
from: record, invalidEnvelope: MetricsSyncPersistenceError.invalidEnvelope(record.recordID.recordName))
|
||||
}
|
||||
|
||||
public nonisolated static func systemFields(of record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.systemFields(of: record)
|
||||
}
|
||||
|
||||
private nonisolated static func assetURL(for key: MetricsRecordKey) throws -> URL {
|
||||
try CloudSyncRecordCoding.assetURL(recordName: key.recordName, directoryName: assetDirectoryName)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -9,7 +9,6 @@ import TermindPersistence
|
||||
@MainActor
|
||||
@Observable
|
||||
public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchecked Sendable {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
// Keep the initial AI-sync identifiers so existing replicas remain reachable.
|
||||
public nonisolated static let zoneName = "TermindAIConfigurationV1"
|
||||
public nonisolated static let recordType = "AIConfigurationItem"
|
||||
@@ -18,7 +17,6 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
public nonisolated static let sftpRecordType = "SFTPPreferencesItem"
|
||||
public nonisolated static let workspaceRecordType = "WorkspacePreferencesItem"
|
||||
public nonisolated static let subscriptionID = "termind-ai-configuration-v1-changes"
|
||||
public nonisolated static let inlinePayloadLimit = CloudSyncRecordCoding.inlinePayloadLimit
|
||||
nonisolated static let assetDirectoryName = "Termind/SettingsSyncAssets"
|
||||
|
||||
private let context: ModelContext
|
||||
@@ -34,10 +32,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
|
||||
@ObservationIgnored private(set) lazy var lifecycle = makeLifecycle()
|
||||
|
||||
public private(set) var status: VaultSyncStatus {
|
||||
get { lifecycle.status }
|
||||
set { lifecycle.status = newValue }
|
||||
}
|
||||
public var status: VaultSyncStatus { lifecycle.status }
|
||||
public var didApplyRemote: ((SettingsSyncSnapshot, Set<SettingsRecordKey>) -> Void)?
|
||||
|
||||
public var isEnabled: Bool { lifecycle.isEnabled }
|
||||
@@ -45,9 +40,9 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
public convenience init(
|
||||
persistence: SettingsSyncPersistence,
|
||||
secrets: AISecretStore,
|
||||
container: CKContainer = CKContainer(identifier: SettingsSyncCoordinator.containerIdentifier),
|
||||
container: CKContainer = CKContainer(identifier: CloudSyncContainer.identifier),
|
||||
defaults: UserDefaults = .standard,
|
||||
environmentIdentifier: String = SettingsSyncCoordinator.runtimeEnvironmentIdentifier
|
||||
environmentIdentifier: String = CloudSyncContainer.environmentIdentifier
|
||||
) {
|
||||
self.init(
|
||||
persistence: persistence, secrets: secrets, defaults: defaults,
|
||||
@@ -78,11 +73,6 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
persistence.didRecordChanges = { [weak self] mutations in self?.noteLocalChanges(mutations) }
|
||||
}
|
||||
|
||||
public nonisolated static var runtimeEnvironmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String
|
||||
?? "Unknown"
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) {
|
||||
lifecycle.logger = logging.makeLogger(label: "termind.cloud-sync")
|
||||
}
|
||||
@@ -102,20 +92,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
let context = context
|
||||
return CloudSyncLifecycle(
|
||||
store: CloudSyncLifecycleStore(
|
||||
state: {
|
||||
let state = try journal.state()
|
||||
return CloudSyncState(
|
||||
engineStateData: state.engineStateData, boundUserRecordName: state.boundUserRecordName,
|
||||
pendingUserRecordName: state.pendingUserRecordName, lastSuccessfulSync: state.lastSuccessfulSync
|
||||
)
|
||||
},
|
||||
updateState: { value in
|
||||
let state = try journal.state()
|
||||
state.engineStateData = value.engineStateData
|
||||
state.boundUserRecordName = value.boundUserRecordName
|
||||
state.pendingUserRecordName = value.pendingUserRecordName
|
||||
state.lastSuccessfulSync = value.lastSuccessfulSync
|
||||
},
|
||||
state: journal.syncState, updateState: journal.updateSyncState,
|
||||
save: { try context.save() }, rollback: { context.rollback() },
|
||||
pendingChanges: {
|
||||
try CloudSyncRecordStateStore(context: context, scope: $0).pendingChanges(journal.cloudRecords())
|
||||
@@ -140,7 +117,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
clearSystemFields: journal.clearSystemFields,
|
||||
handlesRecordName: { SettingsRecordKey(recordName: $0) != nil })),
|
||||
transport: makeTransport(zoneID, self),
|
||||
zoneID: zoneID, containerIdentifier: Self.containerIdentifier, defaults: defaults,
|
||||
zoneID: zoneID, containerIdentifier: CloudSyncContainer.identifier, defaults: defaults,
|
||||
enabledKey: SettingsKey.settingsSyncEnabled, defaultEnabled: SettingsKey.defaultSettingsSyncEnabled,
|
||||
environmentKey: SettingsKey.settingsSyncEnvironment, environmentIdentifier: environmentIdentifier,
|
||||
authorizeUploads: { [secrets] in secrets.uploadsAuthorized = $0 },
|
||||
@@ -149,13 +126,9 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
}
|
||||
|
||||
private func completePendingAssetPurge() throws {
|
||||
let cleanup = CloudSyncAssetCleanupStore(context: context)
|
||||
guard try cleanup.isPending(cacheKey: Self.assetDirectoryName) else { return }
|
||||
try CloudSyncRecordCoding.purgeAssetDirectory(
|
||||
directoryName: Self.assetDirectoryName, cacheRoot: assetCacheRoot,
|
||||
try CloudSyncRecordCoding.completePendingAssetPurge(
|
||||
directoryName: Self.assetDirectoryName, context: context, cacheRoot: assetCacheRoot,
|
||||
removeItem: removeAssetDirectory)
|
||||
try cleanup.complete(cacheKey: Self.assetDirectoryName)
|
||||
try context.save()
|
||||
}
|
||||
|
||||
private func migrateSecrets() throws {
|
||||
@@ -166,14 +139,10 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
+ config.mcpServers.map(\.credentialRef)))
|
||||
}
|
||||
|
||||
/// Accepted remote credential references become eligible for Keychain synchronization.
|
||||
private func applyRemote(
|
||||
_ envelopes: [SettingsEnvelope], deletions: Set<SettingsRecordKey> = [], beforeSave: () throws -> Void
|
||||
) throws
|
||||
-> (
|
||||
snapshot: SettingsSyncSnapshot, acceptedKeys: Set<SettingsRecordKey>,
|
||||
needsUpload: Set<SettingsRecordKey>
|
||||
)
|
||||
{
|
||||
_ envelopes: [SettingsEnvelope], deletions: Set<SettingsRecordKey>, beforeSave: () throws -> Void
|
||||
) throws -> (snapshot: SettingsSyncSnapshot, acceptedKeys: Set<SettingsRecordKey>) {
|
||||
let result = try persistence.applyRemote(envelopes, deletions: deletions, beforeSave: beforeSave)
|
||||
let references = envelopes.filter { result.acceptedKeys.contains($0.key) }.flatMap { envelope -> [String] in
|
||||
switch envelope.payload {
|
||||
@@ -183,7 +152,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
}
|
||||
}
|
||||
try secrets.acceptRemoteReferences(Set(references))
|
||||
return result
|
||||
return (result.snapshot, result.acceptedKeys)
|
||||
}
|
||||
|
||||
public nonisolated func nextFetchChangesOptions(
|
||||
@@ -192,9 +161,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
.init(scope: .zoneIDs([CKRecordZone.ID(zoneName: Self.zoneName, ownerName: CKCurrentUserDefaultName)]))
|
||||
}
|
||||
|
||||
public nonisolated func handleEvent(
|
||||
_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine
|
||||
) async {
|
||||
public nonisolated func handleEvent(_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine) async {
|
||||
await lifecycle.handle(
|
||||
CloudSyncEventInput(event), engine: syncEngine,
|
||||
receiveRemoteRecords: { try receiveRemoteRecords($0, deletions: $1) },
|
||||
@@ -207,7 +174,6 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
await makeChangeBatch(context, engine: syncEngine)
|
||||
}
|
||||
|
||||
@MainActor
|
||||
private func makeChangeBatch(
|
||||
_ context: CKSyncEngine.SendChangesContext, engine: CKSyncEngine
|
||||
) async -> CKSyncEngine.RecordZoneChangeBatch? {
|
||||
@@ -248,11 +214,11 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
lifecycle.noteRecordsObserved(records, excluding: deletions)
|
||||
let allDeletedKeys = Set(
|
||||
deletions.compactMap { id -> SettingsRecordKey? in
|
||||
guard id.zoneID == zoneID, let key = SettingsRecordKey(recordName: id.recordName) else { return nil }
|
||||
return key
|
||||
guard id.zoneID == zoneID else { return nil }
|
||||
return SettingsRecordKey(recordName: id.recordName)
|
||||
})
|
||||
let deletedKeys = Set(allDeletedKeys.filter { $0.kind.participatesInSync && !$0.isLegacyGuardianRole })
|
||||
let cleanupDeletedKeys = allDeletedKeys.subtracting(deletedKeys)
|
||||
// Historical and legacy Guardian deletions only retire cleanup tombstones; they never delete local data.
|
||||
let deletedKeys = allDeletedKeys.filter { $0.kind.participatesInSync && !$0.isLegacyGuardianRole }
|
||||
let deletedNames = Set(allDeletedKeys.map(\.recordName))
|
||||
let cloud = CloudSyncRecordStateStore(context: context, scope: scope)
|
||||
let safety = CloudSyncSafetyStore(context: context, scope: scope)
|
||||
@@ -287,7 +253,8 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
if record.recordID.recordName == envelope.key.recordName,
|
||||
envelope.key.kind.participatesInSync
|
||||
{
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
try journal.updateSystemFields(
|
||||
CloudSyncRecordCoding.systemFields(of: record), for: envelope.key)
|
||||
if try journal.envelope(for: envelope.key) == envelope {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
}
|
||||
@@ -297,16 +264,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
}
|
||||
try safety.clearQuarantine(recordName: record.recordID.recordName)
|
||||
}
|
||||
for key in deletedKeys {
|
||||
guard let envelope = try journal.envelope(for: key), envelope.isDeleted else { continue }
|
||||
try journal.acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
}
|
||||
for key in cleanupDeletedKeys {
|
||||
guard let envelope = try journal.envelope(for: key), envelope.isDeleted else { continue }
|
||||
try journal.acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
}
|
||||
try journal.acknowledgeRemoteDeletions(allDeletedKeys, scope: scope)
|
||||
})
|
||||
try lifecycle.reconcilePendingChanges()
|
||||
if !result.acceptedKeys.isEmpty { didApplyRemote?(result.snapshot, result.acceptedKeys) }
|
||||
@@ -319,20 +277,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
for record in changes.savedRecords where record.recordID.zoneID == zoneID {
|
||||
guard try !safety.isUploadBlocked(recordName: record.recordID.recordName) else { continue }
|
||||
let envelope = try Self.envelope(from: record)
|
||||
let current = try journal.envelope(for: envelope.key)
|
||||
let currentRecord = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.acknowledgeSave(
|
||||
.init(recordName: record.recordID.recordName, operation: .save, version: envelope.version),
|
||||
current: currentRecord)
|
||||
if current?.isDeleted == true {
|
||||
// An in-flight save may finish after a local/remote deletion. Delete its server copy again.
|
||||
try journal.markNeedsUpload([envelope.key])
|
||||
} else {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
}
|
||||
try journal.acknowledgeSavedRecord(record, envelope: envelope, cloud: cloud)
|
||||
try? FileManager.default.removeItem(at: Self.assetURL(for: envelope.key))
|
||||
}
|
||||
for failure in changes.failedRecordSaves where failure.record.recordID.zoneID == zoneID {
|
||||
@@ -380,13 +325,9 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
throw SettingsSyncPersistenceError.invalidEnvelope(recordID.recordName)
|
||||
}
|
||||
let payload = try SettingsSyncCodec.encode(envelope)
|
||||
record["formatVersion"] = NSNumber(value: envelope.formatVersion)
|
||||
record["kind"] = envelope.key.kind.rawValue as NSString
|
||||
record["entityID"] = envelope.key.id as NSString
|
||||
record["modifiedAt"] = envelope.version.modifiedAt as NSDate
|
||||
record["counter"] = NSNumber(value: Int64(clamping: envelope.version.counter))
|
||||
record["writerID"] = envelope.version.writerID.uuidString.lowercased() as NSString
|
||||
record["isDeleted"] = NSNumber(value: envelope.isDeleted)
|
||||
CloudSyncRecordCoding.setMetadata(
|
||||
on: record, formatVersion: envelope.formatVersion, kind: envelope.key.kind.rawValue,
|
||||
entityID: envelope.key.id, version: envelope.version, isDeleted: envelope.isDeleted)
|
||||
try CloudSyncRecordCoding.setPayload(payload, on: record, assetURL: assetURL(for: envelope.key))
|
||||
return record
|
||||
}
|
||||
@@ -409,17 +350,7 @@ public final class SettingsSyncCoordinator: NSObject, CKSyncEngineDelegate, @unc
|
||||
return envelope
|
||||
}
|
||||
|
||||
public nonisolated static func payloadData(from record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.payloadData(
|
||||
from: record, invalidEnvelope: SettingsSyncPersistenceError.invalidEnvelope(record.recordID.recordName))
|
||||
}
|
||||
|
||||
public nonisolated static func systemFields(of record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.systemFields(of: record)
|
||||
}
|
||||
|
||||
private nonisolated static func assetURL(for key: SettingsRecordKey) throws -> URL {
|
||||
try CloudSyncRecordCoding.assetURL(recordName: key.recordName, directoryName: assetDirectoryName)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -8,11 +8,9 @@ import TermindPersistence
|
||||
@MainActor
|
||||
@Observable
|
||||
public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchecked Sendable {
|
||||
public nonisolated static let containerIdentifier = "iCloud.com.akinokaede.termind"
|
||||
public nonisolated static let zoneName = "TermindVaultV1"
|
||||
public nonisolated static let recordType = "VaultItem"
|
||||
public nonisolated static let subscriptionID = "termind-vault-v1-changes"
|
||||
public nonisolated static let inlinePayloadLimit = CloudSyncRecordCoding.inlinePayloadLimit
|
||||
nonisolated static let assetDirectoryName = "Termind/VaultSyncAssets"
|
||||
|
||||
private let context: ModelContext
|
||||
@@ -27,10 +25,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
|
||||
@ObservationIgnored private(set) lazy var lifecycle = makeLifecycle()
|
||||
|
||||
public private(set) var status: VaultSyncStatus {
|
||||
get { lifecycle.status }
|
||||
set { lifecycle.status = newValue }
|
||||
}
|
||||
public var status: VaultSyncStatus { lifecycle.status }
|
||||
public var didApplyRemote: ((VaultSnapshot, Set<VaultRecordKey>) -> Void)?
|
||||
public var beforeSend: (() throws -> Void)?
|
||||
|
||||
@@ -38,9 +33,9 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
|
||||
public convenience init(
|
||||
context: ModelContext,
|
||||
container: CKContainer = CKContainer(identifier: VaultSyncCoordinator.containerIdentifier),
|
||||
container: CKContainer = CKContainer(identifier: CloudSyncContainer.identifier),
|
||||
defaults: UserDefaults = .standard,
|
||||
environmentIdentifier: String = VaultSyncCoordinator.runtimeEnvironmentIdentifier
|
||||
environmentIdentifier: String = CloudSyncContainer.environmentIdentifier
|
||||
) {
|
||||
self.init(
|
||||
context: context, defaults: defaults, environmentIdentifier: environmentIdentifier,
|
||||
@@ -67,11 +62,6 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
super.init()
|
||||
}
|
||||
|
||||
public nonisolated static var runtimeEnvironmentIdentifier: String {
|
||||
Bundle.main.object(forInfoDictionaryKey: "TermindCloudKitEnvironment") as? String
|
||||
?? "Unknown"
|
||||
}
|
||||
|
||||
public func configureLogging(_ logging: any ApplicationLogManaging) {
|
||||
lifecycle.logger = logging.makeLogger(label: "termind.cloud-sync")
|
||||
}
|
||||
@@ -90,20 +80,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
let context = context
|
||||
return CloudSyncLifecycle(
|
||||
store: CloudSyncLifecycleStore(
|
||||
state: {
|
||||
let state = try journal.state()
|
||||
return CloudSyncState(
|
||||
engineStateData: state.engineStateData, boundUserRecordName: state.boundUserRecordName,
|
||||
pendingUserRecordName: state.pendingUserRecordName, lastSuccessfulSync: state.lastSuccessfulSync
|
||||
)
|
||||
},
|
||||
updateState: { value in
|
||||
let state = try journal.state()
|
||||
state.engineStateData = value.engineStateData
|
||||
state.boundUserRecordName = value.boundUserRecordName
|
||||
state.pendingUserRecordName = value.pendingUserRecordName
|
||||
state.lastSuccessfulSync = value.lastSuccessfulSync
|
||||
},
|
||||
state: journal.syncState, updateState: journal.updateSyncState,
|
||||
save: { try context.save() }, rollback: { context.rollback() },
|
||||
pendingChanges: {
|
||||
let cloud = CloudSyncRecordStateStore(context: context, scope: $0)
|
||||
@@ -135,7 +112,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
&& HistoryRecordKey(recordName: $0) == nil
|
||||
})),
|
||||
transport: makeTransport(zoneID, self),
|
||||
zoneID: zoneID, containerIdentifier: Self.containerIdentifier, defaults: defaults,
|
||||
zoneID: zoneID, containerIdentifier: CloudSyncContainer.identifier, defaults: defaults,
|
||||
enabledKey: nil, defaultEnabled: SettingsKey.defaultVaultSyncEnabled,
|
||||
environmentKey: SettingsKey.vaultSyncEnvironment, environmentIdentifier: environmentIdentifier,
|
||||
beforeSend: { [weak self] in try self?.beforeSend?() },
|
||||
@@ -143,13 +120,9 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
}
|
||||
|
||||
private func completePendingAssetPurge() throws {
|
||||
let cleanup = CloudSyncAssetCleanupStore(context: context)
|
||||
guard try cleanup.isPending(cacheKey: Self.assetDirectoryName) else { return }
|
||||
try CloudSyncRecordCoding.purgeAssetDirectory(
|
||||
directoryName: Self.assetDirectoryName, cacheRoot: assetCacheRoot,
|
||||
try CloudSyncRecordCoding.completePendingAssetPurge(
|
||||
directoryName: Self.assetDirectoryName, context: context, cacheRoot: assetCacheRoot,
|
||||
removeItem: removeAssetDirectory)
|
||||
try cleanup.complete(cacheKey: Self.assetDirectoryName)
|
||||
try context.save()
|
||||
}
|
||||
|
||||
public nonisolated func nextFetchChangesOptions(
|
||||
@@ -158,11 +131,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
.init(scope: .zoneIDs([CKRecordZone.ID(zoneName: Self.zoneName, ownerName: CKCurrentUserDefaultName)]))
|
||||
}
|
||||
|
||||
// MARK: CKSyncEngineDelegate
|
||||
|
||||
public nonisolated func handleEvent(
|
||||
_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine
|
||||
) async {
|
||||
public nonisolated func handleEvent(_ event: CKSyncEngine.Event, syncEngine: CKSyncEngine) async {
|
||||
await lifecycle.handle(
|
||||
CloudSyncEventInput(event), engine: syncEngine,
|
||||
receiveRemoteRecords: { try receiveRemoteRecords($0, deletions: $1) },
|
||||
@@ -175,7 +144,6 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
await makeChangeBatch(context, engine: syncEngine)
|
||||
}
|
||||
|
||||
@MainActor
|
||||
private func makeChangeBatch(
|
||||
_ context: CKSyncEngine.SendChangesContext, engine: CKSyncEngine
|
||||
) async -> CKSyncEngine.RecordZoneChangeBatch? {
|
||||
@@ -263,22 +231,9 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
// Only a sent-delete acknowledgement retires their cleanup task (unknownItem is also success).
|
||||
// New-zone deletion increments are handled by HistoryRecordSyncCoordinator as normal user deletions.
|
||||
for (record, envelope) in decoded {
|
||||
let current = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.observed(
|
||||
record.recordID.recordName, version: envelope.version, current: current)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
if try journal.envelope(for: envelope.key) == envelope {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
}
|
||||
try safety.clearQuarantine(recordName: record.recordID.recordName)
|
||||
}
|
||||
for key in deletedKeys {
|
||||
guard let envelope = try journal.envelope(for: key), envelope.isDeleted else { continue }
|
||||
try journal.acknowledgeDeletion(
|
||||
.init(recordName: key.recordName, operation: .delete, version: envelope.version), scope: scope)
|
||||
try journal.noteRemoteRecord(record, envelope: envelope, cloud: cloud, safety: safety)
|
||||
}
|
||||
try journal.acknowledgeRemoteDeletions(deletedKeys, scope: scope)
|
||||
})
|
||||
try lifecycle.reconcilePendingChanges()
|
||||
if !result.acceptedKeys.isEmpty { didApplyRemote?(result.snapshot, result.acceptedKeys) }
|
||||
@@ -294,20 +249,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
{
|
||||
guard try !safety.isUploadBlocked(recordName: record.recordID.recordName) else { continue }
|
||||
let envelope = try Self.envelope(from: record)
|
||||
let current = try journal.envelope(for: envelope.key)
|
||||
let currentRecord = try journal.cloudRecords().first {
|
||||
$0.mutation.recordName == record.recordID.recordName
|
||||
}
|
||||
try cloud.acknowledgeSave(
|
||||
.init(recordName: record.recordID.recordName, operation: .save, version: envelope.version),
|
||||
current: currentRecord)
|
||||
if current?.isDeleted == true {
|
||||
// An in-flight save may finish after a local/remote deletion. Delete its server copy again.
|
||||
try journal.markNeedsUpload([envelope.key])
|
||||
} else {
|
||||
try journal.markUploaded(envelope.key, version: envelope.version)
|
||||
try journal.updateSystemFields(Self.systemFields(of: record), for: envelope.key)
|
||||
}
|
||||
try journal.acknowledgeSavedRecord(record, envelope: envelope, cloud: cloud)
|
||||
try? FileManager.default.removeItem(at: Self.assetURL(for: envelope.key))
|
||||
}
|
||||
for failure in changes.failedRecordSaves where failure.record.recordID.zoneID == zoneID {
|
||||
@@ -333,8 +275,6 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: Record coding
|
||||
|
||||
public nonisolated static func makeRecord(
|
||||
_ outgoing: VaultOutgoingEntry, recordID: CKRecord.ID
|
||||
) throws -> CKRecord {
|
||||
@@ -342,13 +282,10 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
let record = try CloudSyncRecordCoding.makeRecord(
|
||||
type: recordType, id: recordID, systemFields: outgoing.systemFieldsData)
|
||||
let payload = try VaultSyncCodec.encode(envelope)
|
||||
record["formatVersion"] = NSNumber(value: envelope.formatVersion)
|
||||
record["kind"] = envelope.key.kind.rawValue as NSString
|
||||
record["entityID"] = envelope.key.id.uuidString.lowercased() as NSString
|
||||
record["modifiedAt"] = envelope.version.modifiedAt as NSDate
|
||||
record["counter"] = NSNumber(value: Int64(clamping: envelope.version.counter))
|
||||
record["writerID"] = envelope.version.writerID.uuidString.lowercased() as NSString
|
||||
record["isDeleted"] = NSNumber(value: envelope.isDeleted)
|
||||
CloudSyncRecordCoding.setMetadata(
|
||||
on: record, formatVersion: envelope.formatVersion, kind: envelope.key.kind.rawValue,
|
||||
entityID: envelope.key.id.uuidString.lowercased(), version: envelope.version,
|
||||
isDeleted: envelope.isDeleted)
|
||||
try CloudSyncRecordCoding.setPayload(payload, on: record, assetURL: assetURL(for: envelope.key))
|
||||
return record
|
||||
}
|
||||
@@ -364,17 +301,7 @@ public final class VaultSyncCoordinator: NSObject, CKSyncEngineDelegate, @unchec
|
||||
return envelope
|
||||
}
|
||||
|
||||
public nonisolated static func payloadData(from record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.payloadData(
|
||||
from: record, invalidEnvelope: VaultSyncPersistenceError.invalidEnvelope(record.recordID.recordName))
|
||||
}
|
||||
|
||||
public nonisolated static func systemFields(of record: CKRecord) throws -> Data {
|
||||
try CloudSyncRecordCoding.systemFields(of: record)
|
||||
}
|
||||
|
||||
private nonisolated static func assetURL(for key: VaultRecordKey) throws -> URL {
|
||||
try CloudSyncRecordCoding.assetURL(recordName: key.recordName, directoryName: assetDirectoryName)
|
||||
}
|
||||
|
||||
}
|
||||
Reference in new issue
Block a user