Commit 2ed92d773a
Unsigned
Layout: unified · split
DomainDig/CloudSyncService.swift +131 −109
| @@ -57,6 +57,10 @@ enum ShareableEntity: Identifiable, Hashable { | |||
| 57 | } | 57 | } |
| 58 | } | 58 | } |
| 59 | 59 | ||
| 60 | private enum CloudZone { | ||
| 61 | static let name = "DomainDigZone" | ||
| 62 | } | ||
| 63 | |||
| 60 | private enum CloudRecordType { | 64 | private enum CloudRecordType { |
| 61 | static let trackedDomain = "TrackedDomain" | 65 | static let trackedDomain = "TrackedDomain" |
| 62 | static let workflow = "DomainWorkflow" | 66 | static let workflow = "DomainWorkflow" |
| @@ -129,11 +133,6 @@ private enum SyncDatabaseScope { | |||
| 129 | case sharedDatabase | 133 | case sharedDatabase |
| 130 | } | 134 | } |
| 131 | 135 | ||
| 132 | private struct SharedRecord<Value> { | ||
| 133 | var record: CKRecord | ||
| 134 | var value: Value | ||
| 135 | } | ||
| 136 | |||
| 137 | private struct MergedSyncPayload { | 136 | private struct MergedSyncPayload { |
| 138 | var payload: SyncPayload | 137 | var payload: SyncPayload |
| 139 | var hadConflict: Bool | 138 | var hadConflict: Bool |
| @@ -154,6 +153,7 @@ final class CloudSyncService { | |||
| 154 | private var container: CKContainer? | 153 | private var container: CKContainer? |
| 155 | private var privateDatabase: CKDatabase? | 154 | private var privateDatabase: CKDatabase? |
| 156 | private var sharedDatabase: CKDatabase? | 155 | private var sharedDatabase: CKDatabase? |
| 156 | private var didEnsureZone = false | ||
| 157 | private let defaults: UserDefaults | 157 | private let defaults: UserDefaults |
| 158 | private let encoder: JSONEncoder | 158 | private let encoder: JSONEncoder |
| 159 | private let decoder: JSONDecoder | 159 | private let decoder: JSONDecoder |
| @@ -359,6 +359,8 @@ final class CloudSyncService { | |||
| 359 | throw CloudSyncRuntimeError.missingEntitlement | 359 | throw CloudSyncRuntimeError.missingEntitlement |
| 360 | } | 360 | } |
| 361 | 361 | ||
| 362 | try await ensureCustomZone(in: database) | ||
| 363 | |||
| 362 | switch entity { | 364 | switch entity { |
| 363 | case .trackedDomain(let domain): | 365 | case .trackedDomain(let domain): |
| 364 | let normalized = Self.normalizeDomain(domain) | 366 | let normalized = Self.normalizeDomain(domain) |
| @@ -543,31 +545,19 @@ final class CloudSyncService { | |||
| 543 | } | 545 | } |
| 544 | 546 | ||
| 545 | private func fetchRemotePayload() async throws -> SyncPayload { | 547 | private func fetchRemotePayload() async throws -> SyncPayload { |
| 546 | let privateTrackedDomains = try await fetchTrackedDomainRecords(in: .privateDatabase) | 548 | let privateRecords = try await fetchZoneRecords(in: .privateDatabase) |
| 547 | let sharedTrackedDomains = try await fetchTrackedDomainRecords(in: .sharedDatabase) | 549 | let sharedRecords = try await fetchZoneRecords(in: .sharedDatabase) |
| 548 | let privateWorkflows = try await fetchWorkflowRecords(in: .privateDatabase) | 550 | |
| 549 | let sharedWorkflows = try await fetchWorkflowRecords(in: .sharedDatabase) | 551 | let privateTrackedDomains = try await trackedDomains(from: privateRecords, in: .privateDatabase) |
| 550 | 552 | let sharedTrackedDomains = try await trackedDomains(from: sharedRecords, in: .sharedDatabase) | |
| 551 | let appSettingsRecords: [SyncedAppSettings] = try await fetchRecords( | 553 | let privateWorkflows = try await workflows(from: privateRecords, in: .privateDatabase) |
| 552 | ofType: CloudRecordType.appSettings, | 554 | let sharedWorkflows = try await workflows(from: sharedRecords, in: .sharedDatabase) |
| 553 | in: .privateDatabase | 555 | |
| 554 | ) | 556 | let appSettingsRecords: [SyncedAppSettings] = decodeRecords(privateRecords, ofType: CloudRecordType.appSettings) |
| 555 | let monitoringSettingsRecords: [SyncedMonitoringSettings] = try await fetchRecords( | 557 | let monitoringSettingsRecords: [SyncedMonitoringSettings] = decodeRecords(privateRecords, ofType: CloudRecordType.monitoringSettings) |
| 556 | ofType: CloudRecordType.monitoringSettings, | 558 | let domainNotes: [SyncedDomainNote] = decodeRecords(privateRecords, ofType: CloudRecordType.domainNote) |
| 557 | in: .privateDatabase | 559 | let historyMetadata: [SyncedHistoryMetadata] = decodeRecords(privateRecords, ofType: CloudRecordType.historyMetadata) |
| 558 | ) | 560 | let tombstones: [SyncTombstone] = decodeRecords(privateRecords, ofType: CloudRecordType.tombstone) |
| 559 | let domainNotes: [SyncedDomainNote] = try await fetchRecords( | ||
| 560 | ofType: CloudRecordType.domainNote, | ||
| 561 | in: .privateDatabase | ||
| 562 | ) | ||
| 563 | let historyMetadata: [SyncedHistoryMetadata] = try await fetchRecords( | ||
| 564 | ofType: CloudRecordType.historyMetadata, | ||
| 565 | in: .privateDatabase | ||
| 566 | ) | ||
| 567 | let tombstones: [SyncTombstone] = try await fetchRecords( | ||
| 568 | ofType: CloudRecordType.tombstone, | ||
| 569 | in: .privateDatabase | ||
| 570 | ) | ||
| 571 | 561 | ||
| 572 | let appSettings = appSettingsRecords.max(by: { $0.updatedAt < $1.updatedAt }) | 562 | let appSettings = appSettingsRecords.max(by: { $0.updatedAt < $1.updatedAt }) |
| 573 | ?? SyncedAppSettings(snapshot: DomainDataPortabilityService.loadAppSettings(), updatedAt: .distantPast) | 563 | ?? SyncedAppSettings(snapshot: DomainDataPortabilityService.loadAppSettings(), updatedAt: .distantPast) |
| @@ -590,6 +580,8 @@ final class CloudSyncService { | |||
| 590 | throw CloudSyncRuntimeError.missingEntitlement | 580 | throw CloudSyncRuntimeError.missingEntitlement |
| 591 | } | 581 | } |
| 592 | 582 | ||
| 583 | try await ensureCustomZone(in: privateDatabase) | ||
| 584 | |||
| 593 | var privateRecordsToSave: [CKRecord] = [] | 585 | var privateRecordsToSave: [CKRecord] = [] |
| 594 | var sharedRecordsToSave: [CKRecord] = [] | 586 | var sharedRecordsToSave: [CKRecord] = [] |
| 595 | var recordIDsToDelete: [CKRecord.ID] = [] | 587 | var recordIDsToDelete: [CKRecord.ID] = [] |
| @@ -626,11 +618,11 @@ final class CloudSyncService { | |||
| 626 | switch tombstone.entityType { | 618 | switch tombstone.entityType { |
| 627 | case .trackedDomain: | 619 | case .trackedDomain: |
| 628 | let identifier = tombstone.identifier | 620 | let identifier = tombstone.identifier |
| 629 | recordIDsToDelete.append(CKRecord.ID(recordName: trackedDomainRecordName(for: identifier))) | 621 | recordIDsToDelete.append(CKRecord.ID(recordName: trackedDomainRecordName(for: identifier), zoneID: customZoneID)) |
| 630 | recordIDsToDelete.append(CKRecord.ID(recordName: domainNoteRecordName(for: identifier))) | 622 | recordIDsToDelete.append(CKRecord.ID(recordName: domainNoteRecordName(for: identifier), zoneID: customZoneID)) |
| 631 | recordIDsToDelete.append(CKRecord.ID(recordName: historyMetadataRecordName(for: identifier))) | 623 | recordIDsToDelete.append(CKRecord.ID(recordName: historyMetadataRecordName(for: identifier), zoneID: customZoneID)) |
| 632 | case .workflow: | 624 | case .workflow: |
| 633 | recordIDsToDelete.append(CKRecord.ID(recordName: workflowRecordName(for: tombstone.identifier))) | 625 | recordIDsToDelete.append(CKRecord.ID(recordName: workflowRecordName(for: tombstone.identifier), zoneID: customZoneID)) |
| 634 | } | 626 | } |
| 635 | } | 627 | } |
| 636 | 628 | ||
| @@ -1051,7 +1043,12 @@ final class CloudSyncService { | |||
| 1051 | } | 1043 | } |
| 1052 | } | 1044 | } |
| 1053 | 1045 | ||
| 1054 | private func fetchRecords<T: Decodable>(ofType recordType: String, in scope: SyncDatabaseScope) async throws -> [T] { | 1046 | /// Fetches every record in `scope` via `CKFetchRecordZoneChangesOperation`, |
| 1047 | /// which — unlike a `CKQuery` — needs no queryable schema indexes and works | ||
| 1048 | /// on a brand-new zone. The private database holds the user's own records in | ||
| 1049 | /// a single custom zone; the shared database exposes one zone per accepted | ||
| 1050 | /// share, so every shared zone is fetched. | ||
| 1051 | private func fetchZoneRecords(in scope: SyncDatabaseScope) async throws -> [CKRecord] { | ||
| 1055 | guard let database = cloudKitDatabase(in: scope) else { | 1052 | guard let database = cloudKitDatabase(in: scope) else { |
| 1056 | if scope == .sharedDatabase { | 1053 | if scope == .sharedDatabase { |
| 1057 | return [] | 1054 | return [] |
| @@ -1059,99 +1056,108 @@ final class CloudSyncService { | |||
| 1059 | throw CloudSyncRuntimeError.missingEntitlement | 1056 | throw CloudSyncRuntimeError.missingEntitlement |
| 1060 | } | 1057 | } |
| 1061 | 1058 | ||
| 1062 | let query = CKQuery(recordType: recordType, predicate: NSPredicate(value: true)) | 1059 | let zoneIDs: [CKRecordZone.ID] |
| 1063 | var results: [T] = [] | 1060 | switch scope { |
| 1064 | var cursor: CKQueryOperation.Cursor? | 1061 | case .privateDatabase: |
| 1065 | 1062 | try await ensureCustomZone(in: database) | |
| 1066 | repeat { | 1063 | zoneIDs = [customZoneID] |
| 1067 | let batch: (matchResults: [(CKRecord.ID, Result<CKRecord, any Error>)], queryCursor: CKQueryOperation.Cursor?) | 1064 | case .sharedDatabase: |
| 1068 | if let cursor { | 1065 | zoneIDs = try await database.allRecordZones().map(\.zoneID) |
| 1069 | batch = try await database.records(continuingMatchFrom: cursor, desiredKeys: nil, resultsLimit: 200) | 1066 | } |
| 1070 | } else { | ||
| 1071 | batch = try await database.records(matching: query, inZoneWith: nil, desiredKeys: nil, resultsLimit: 200) | ||
| 1072 | } | ||
| 1073 | |||
| 1074 | for (_, result) in batch.matchResults { | ||
| 1075 | switch result { | ||
| 1076 | case .success(let record): | ||
| 1077 | guard let payload = record[CloudRecordKey.payload] as? Data else { continue } | ||
| 1078 | if let decoded = try? decoder.decode(T.self, from: payload) { | ||
| 1079 | results.append(decoded) | ||
| 1080 | } else { | ||
| 1081 | lastErrorMessage = "Some iCloud records were skipped because they could not be decoded." | ||
| 1082 | } | ||
| 1083 | case .failure: | ||
| 1084 | continue | ||
| 1085 | } | ||
| 1086 | } | ||
| 1087 | |||
| 1088 | cursor = batch.queryCursor | ||
| 1089 | } while cursor != nil | ||
| 1090 | 1067 | ||
| 1091 | return results | 1068 | guard !zoneIDs.isEmpty else { return [] } |
| 1069 | return try await fetchZoneChanges(in: database, zoneIDs: zoneIDs) | ||
| 1092 | } | 1070 | } |
| 1093 | 1071 | ||
| 1094 | private func fetchSharedRecords<T: Decodable>( | 1072 | private func fetchZoneChanges(in database: CKDatabase, zoneIDs: [CKRecordZone.ID]) async throws -> [CKRecord] { |
| 1095 | ofType recordType: String, | 1073 | // A nil change token requests every record in the zone. The server may |
| 1096 | in scope: SyncDatabaseScope | 1074 | // truncate a large response and set `moreComing`, so keep re-fetching |
| 1097 | ) async throws -> [SharedRecord<T>] { | 1075 | // the still-truncated zones with their latest tokens until all drain. |
| 1098 | guard let database = cloudKitDatabase(in: scope) else { | 1076 | var tokensByZone: [CKRecordZone.ID: CKServerChangeToken?] = [:] |
| 1099 | if scope == .sharedDatabase { | 1077 | for zoneID in zoneIDs { |
| 1100 | return [] | 1078 | tokensByZone[zoneID] = .some(nil) |
| 1101 | } | ||
| 1102 | throw CloudSyncRuntimeError.missingEntitlement | ||
| 1103 | } | 1079 | } |
| 1104 | 1080 | ||
| 1105 | let query = CKQuery(recordType: recordType, predicate: NSPredicate(value: true)) | 1081 | var records: [CKRecord] = [] |
| 1106 | var results: [SharedRecord<T>] = [] | 1082 | while !tokensByZone.isEmpty { |
| 1107 | var cursor: CKQueryOperation.Cursor? | 1083 | tokensByZone = try await withCheckedThrowingContinuation { continuation in |
| 1108 | 1084 | var moreByZone: [CKRecordZone.ID: CKServerChangeToken?] = [:] | |
| 1109 | repeat { | 1085 | let configurations = tokensByZone.mapValues { token -> CKFetchRecordZoneChangesOperation.ZoneConfiguration in |
| 1110 | let batch: (matchResults: [(CKRecord.ID, Result<CKRecord, any Error>)], queryCursor: CKQueryOperation.Cursor?) | 1086 | let configuration = CKFetchRecordZoneChangesOperation.ZoneConfiguration() |
| 1111 | if let cursor { | 1087 | configuration.previousServerChangeToken = token ?? nil |
| 1112 | batch = try await database.records(continuingMatchFrom: cursor, desiredKeys: nil, resultsLimit: 200) | 1088 | return configuration |
| 1113 | } else { | 1089 | } |
| 1114 | batch = try await database.records(matching: query, inZoneWith: nil, desiredKeys: nil, resultsLimit: 200) | ||
| 1115 | } | ||
| 1116 | 1090 | ||
| 1117 | for (_, result) in batch.matchResults { | 1091 | let operation = CKFetchRecordZoneChangesOperation( |
| 1118 | switch result { | 1092 | recordZoneIDs: Array(tokensByZone.keys), |
| 1119 | case .success(let record): | 1093 | configurationsByRecordZoneID: configurations |
| 1120 | guard let payload = record[CloudRecordKey.payload] as? Data else { continue } | 1094 | ) |
| 1121 | if let decoded = try? decoder.decode(T.self, from: payload) { | 1095 | operation.recordWasChangedBlock = { _, result in |
| 1122 | results.append(SharedRecord(record: record, value: decoded)) | 1096 | if case .success(let record) = result { |
| 1123 | } else { | 1097 | records.append(record) |
| 1124 | lastErrorMessage = "Some iCloud records were skipped because they could not be decoded." | ||
| 1125 | } | 1098 | } |
| 1126 | case .failure: | ||
| 1127 | continue | ||
| 1128 | } | 1099 | } |
| 1100 | operation.recordZoneFetchResultBlock = { zoneID, result in | ||
| 1101 | if case .success(let value) = result, value.moreComing { | ||
| 1102 | moreByZone[zoneID] = .some(value.serverChangeToken) | ||
| 1103 | } | ||
| 1104 | } | ||
| 1105 | operation.fetchRecordZoneChangesResultBlock = { result in | ||
| 1106 | switch result { | ||
| 1107 | case .success: | ||
| 1108 | continuation.resume(returning: moreByZone) | ||
| 1109 | case .failure(let error): | ||
| 1110 | continuation.resume(throwing: error) | ||
| 1111 | } | ||
| 1112 | } | ||
| 1113 | database.add(operation) | ||
| 1129 | } | 1114 | } |
| 1115 | } | ||
| 1130 | 1116 | ||
| 1131 | cursor = batch.queryCursor | 1117 | return records |
| 1132 | } while cursor != nil | 1118 | } |
| 1133 | 1119 | ||
| 1120 | private func decodeRecords<T: Decodable>(_ records: [CKRecord], ofType recordType: String) -> [T] { | ||
| 1121 | var results: [T] = [] | ||
| 1122 | for record in records where record.recordType == recordType { | ||
| 1123 | guard let payload = record[CloudRecordKey.payload] as? Data else { continue } | ||
| 1124 | if let decoded = try? decoder.decode(T.self, from: payload) { | ||
| 1125 | results.append(decoded) | ||
| 1126 | } else { | ||
| 1127 | lastErrorMessage = "Some iCloud records were skipped because they could not be decoded." | ||
| 1128 | } | ||
| 1129 | } | ||
| 1134 | return results | 1130 | return results |
| 1135 | } | 1131 | } |
| 1136 | 1132 | ||
| 1137 | private func fetchTrackedDomainRecords(in scope: SyncDatabaseScope) async throws -> [TrackedDomain] { | 1133 | private func trackedDomains(from records: [CKRecord], in scope: SyncDatabaseScope) async throws -> [TrackedDomain] { |
| 1138 | let records: [SharedRecord<TrackedDomain>] = try await fetchSharedRecords(ofType: CloudRecordType.trackedDomain, in: scope) | 1134 | let matching = records.filter { $0.recordType == CloudRecordType.trackedDomain } |
| 1139 | let shares = try await sharesByRecordID(for: records.map(\.record), in: scope) | 1135 | let shares = try await sharesByRecordID(for: matching, in: scope) |
| 1140 | 1136 | ||
| 1141 | return records.map { entry in | 1137 | return matching.compactMap { record in |
| 1142 | var trackedDomain = Self.payloadTrackedDomain(entry.value) | 1138 | guard let payload = record[CloudRecordKey.payload] as? Data, |
| 1143 | trackedDomain.collaboration = collaborationMetadata(for: entry.record, share: shares[entry.record.recordID], scope: scope) | 1139 | let decoded = try? decoder.decode(TrackedDomain.self, from: payload) else { |
| 1140 | lastErrorMessage = "Some iCloud records were skipped because they could not be decoded." | ||
| 1141 | return nil | ||
| 1142 | } | ||
| 1143 | var trackedDomain = Self.payloadTrackedDomain(decoded) | ||
| 1144 | trackedDomain.collaboration = collaborationMetadata(for: record, share: shares[record.recordID], scope: scope) | ||
| 1144 | return trackedDomain | 1145 | return trackedDomain |
| 1145 | } | 1146 | } |
| 1146 | } | 1147 | } |
| 1147 | 1148 | ||
| 1148 | private func fetchWorkflowRecords(in scope: SyncDatabaseScope) async throws -> [DomainWorkflow] { | 1149 | private func workflows(from records: [CKRecord], in scope: SyncDatabaseScope) async throws -> [DomainWorkflow] { |
| 1149 | let records: [SharedRecord<DomainWorkflow>] = try await fetchSharedRecords(ofType: CloudRecordType.workflow, in: scope) | 1150 | let matching = records.filter { $0.recordType == CloudRecordType.workflow } |
| 1150 | let shares = try await sharesByRecordID(for: records.map(\.record), in: scope) | 1151 | let shares = try await sharesByRecordID(for: matching, in: scope) |
| 1151 | 1152 | ||
| 1152 | return records.map { entry in | 1153 | return matching.compactMap { record in |
| 1153 | var workflow = normalizedWorkflow(entry.value) | 1154 | guard let payload = record[CloudRecordKey.payload] as? Data, |
| 1154 | workflow.collaboration = collaborationMetadata(for: entry.record, share: shares[entry.record.recordID], scope: scope) | 1155 | let decoded = try? decoder.decode(DomainWorkflow.self, from: payload) else { |
| 1156 | lastErrorMessage = "Some iCloud records were skipped because they could not be decoded." | ||
| 1157 | return nil | ||
| 1158 | } | ||
| 1159 | var workflow = normalizedWorkflow(decoded) | ||
| 1160 | workflow.collaboration = collaborationMetadata(for: record, share: shares[record.recordID], scope: scope) | ||
| 1155 | return workflow | 1161 | return workflow |
| 1156 | } | 1162 | } |
| 1157 | } | 1163 | } |
| @@ -1318,7 +1324,7 @@ final class CloudSyncService { | |||
| 1318 | payload: T, | 1324 | payload: T, |
| 1319 | extraFields: [String: CKRecordValue] | 1325 | extraFields: [String: CKRecordValue] |
| 1320 | ) -> CKRecord { | 1326 | ) -> CKRecord { |
| 1321 | let record = CKRecord(recordType: type, recordID: CKRecord.ID(recordName: recordName)) | 1327 | let record = CKRecord(recordType: type, recordID: CKRecord.ID(recordName: recordName, zoneID: customZoneID)) |
| 1322 | record[CloudRecordKey.payload] = (try? encoder.encode(payload)) as CKRecordValue? | 1328 | record[CloudRecordKey.payload] = (try? encoder.encode(payload)) as CKRecordValue? |
| 1323 | for (key, value) in extraFields { | 1329 | for (key, value) in extraFields { |
| 1324 | record[key] = value | 1330 | record[key] = value |
| @@ -1393,6 +1399,22 @@ final class CloudSyncService { | |||
| 1393 | return container | 1399 | return container |
| 1394 | } | 1400 | } |
| 1395 | 1401 | ||
| 1402 | /// The user's own records live in a single custom zone. A custom zone (as | ||
| 1403 | /// opposed to the default zone) is what lets the sync read every record with | ||
| 1404 | /// `CKFetchRecordZoneChangesOperation`, which needs no queryable schema | ||
| 1405 | /// indexes, and is also a prerequisite for CloudKit sharing. | ||
| 1406 | private var customZoneID: CKRecordZone.ID { | ||
| 1407 | CKRecordZone.ID(zoneName: CloudZone.name, ownerName: CKCurrentUserDefaultName) | ||
| 1408 | } | ||
| 1409 | |||
| 1410 | /// Creates the custom zone if it does not exist yet. Saving an existing zone | ||
| 1411 | /// is a no-op, so this is safe to call before every push and fetch. | ||
| 1412 | private func ensureCustomZone(in database: CKDatabase) async throws { | ||
| 1413 | if didEnsureZone { return } | ||
| 1414 | _ = try await database.modifyRecordZones(saving: [CKRecordZone(zoneID: customZoneID)], deleting: []) | ||
| 1415 | didEnsureZone = true | ||
| 1416 | } | ||
| 1417 | |||
| 1396 | private func cloudKitDatabase(in scope: SyncDatabaseScope) -> CKDatabase? { | 1418 | private func cloudKitDatabase(in scope: SyncDatabaseScope) -> CKDatabase? { |
| 1397 | switch scope { | 1419 | switch scope { |
| 1398 | case .privateDatabase: | 1420 | case .privateDatabase: |