crossmate

RecordApplier.swift (25238B)

---

 import CloudKit
 import CoreData
 import Foundation
 
 struct BatchEffects {
     var movesUpdated = Set<UUID>()
     /// Games whose roster-relevant state changed in this batch — a Player
     /// record (name / selection / readAt), a Game record (share metadata), a
     /// deletion (participant removal), or a *new* contributor's first Moves
     /// row (a participant the roster can only discover from their moves; see
     /// `applyMovesRecord`'s `onNewAuthor`). Drives `.playerRosterShouldRefresh`.
     /// A repeat Moves row from a known contributor is excluded: it changes the
     /// grid, not the roster, and during a co-solve flurry those land at
     /// keystroke cadence — refreshing the roster (and re-evaluating the grid
     /// view) on each one was pure overhead.
     var rosterRelevant = Set<UUID>()
     var pings: [Ping] = []
     var playersUpdated = Set<UUID>()
     var playerPresenceChanged = Set<UUID>()
     var engagementChanged = Set<UUID>()
     /// Games whose inbound Game record changed the notification content key, so
     /// the caller re-mirrors the App Group key directory the NSE reads.
     var contentKeysChanged = Set<UUID>()
     var removed = Set<UUID>()
     /// Per-game incoming read cursor from one of *our own* devices: the
     /// `readAt` horizon plus the encoded per-peer "seen" baseline that sibling
     /// shipped on its `Player.sessionSnapshot` (nil on older writes). Drives
     /// cross-device baseline adoption.
     var readCursors: [(UUID, Date, Data?)] = []
     /// Games that just transitioned to completed via an inbound Game record, so
     /// this device uploads its journal for replay even though it didn't run the
     /// local completion path.
     var completedTransitions = Set<UUID>()
     /// Games for which an inbound `Journal` record landed — wakes a waiting
     /// finish-banner replay to re-check completeness.
     var journalsSynced = Set<UUID>()
     /// Account-level push address decisions seen in the private account zone.
     var accountPushAddresses: [String] = []
     /// Account-level push *secret* decisions seen in the private account zone,
     /// each with its generation. Drive re-derivation of every per-game push
     /// address (see `RecordSerializer.deriveGameAddress`); the version gates
     /// adoption so a stale inbound copy can't undo a rotation.
     var accountPushSecrets: [(secret: String, version: Int64)] = []
     /// Versions of *our own* name Decision echoed back by sync. Adopted into
     /// the local rename counter so the next rename supersedes the highest
     /// generation any of this account's devices has published.
     var selfNameVersions: [Int64] = []
     /// A `name` or `nickname` Decision changed a friend row in this batch —
     /// either side of an App Group nickname-directory entry, so the caller
     /// rebuilds the directory after the batch saves.
     var friendNamesChanged = false
     /// Diagnostics emitted while applying the batch inside `performAndWait` —
     /// chiefly Core Data fetch/save failures, which silently drop records (the
     /// engine's change token has already advanced, so they never redeliver).
     /// The batch context can't `await`, so messages accumulate here and the
     /// caller traces them afterwards — the same shape as the send path's
     /// failure messages. A bare `print` is invisible in Production, where the
     /// on-device diagnostics log is the only observability.
     var traces: [String] = []
 }
 
 extension SyncEngine {
     func applyDirectRecordZoneChanges(
         records: [CKRecord],
         deletions: [(CKRecord.ID, CKRecord.RecordType)],
         scopeValue: Int16
     ) async {
         guard !records.isEmpty || !deletions.isEmpty else { return }
         let ctx = persistence.container.newBackgroundContext()
         ctx.mergePolicy = NSMergePolicy.mergeByPropertyObjectTrump
         let localAuthorID = await currentLocalAuthorID()
         let effects: BatchEffects = ctx.performAndWait {
             var effects = BatchEffects()
             for record in records {
                 switch record.recordType {
                 case "Game":
                     let entity = RecordSerializer.applyGameRecord(
                         record,
                         to: ctx,
                         databaseScope: scopeValue,
                         onEngagementChange: { effects.engagementChanged.insert($0) },
                         onCompletedTransition: { effects.completedTransitions.insert($0) },
                         onContentKeyChange: { effects.contentKeysChanged.insert($0) }
                     )
                     if let id = entity.id { effects.rosterRelevant.insert(id) }
                 case "Moves":
                     if let value = RecordSerializer.parseMovesRecord(record) {
                         let cellsChanged = RecordSerializer.applyMovesRecord(
                             record,
                             value: value,
                             to: ctx,
                             localAuthorID: localAuthorID,
                             onNewAuthor: { _ in effects.rosterRelevant.insert(value.gameID) }
                         )
                         if cellsChanged { effects.movesUpdated.insert(value.gameID) }
                     }
                 case "Player":
                     if let (gameID, _) = RecordSerializer.parsePlayerRecordName(record.recordID.recordName) {
                         self.applyPlayerRecord(
                             record,
                             in: ctx,
                             localAuthorID: localAuthorID,
                             onFirstTime: { effects.playersUpdated.insert($0) },
                             onPresenceChange: { effects.playerPresenceChanged.insert($0) },
                             onReadCursor: { effects.readCursors.append(($0, $1, $2)) }
                         )
                         effects.rosterRelevant.insert(gameID)
                     }
                 case Archive.recordType:
                     if let id = self.applyArchiveRecord(record, in: ctx) {
                         effects.rosterRelevant.insert(id)
                     }
                 default:
                     break
                 }
             }
             for deletion in deletions {
                 self.applyDeletion(
                     recordID: deletion.0,
                     recordType: deletion.1,
                     in: ctx
                 )
                 if let id = self.gameID(fromRecordName: deletion.0.recordName) {
                     effects.rosterRelevant.insert(id)
                 }
             }
             for gameID in effects.movesUpdated {
                 effects.traces += self.replayCellCache(for: gameID, in: ctx)
             }
             if ctx.hasChanges {
                 do {
                     try ctx.save()
                 } catch {
                     let nsError = error as NSError
                     effects.traces.append(
                         "direct-push ctx.save FAILED " +
                         "— domain=\(nsError.domain) code=\(nsError.code) " +
                         "\(nsError.localizedDescription)"
                     )
                 }
             }
             // Re-mirror the App Group key directory once the batch is saved, so
             // a just-adopted content key is available to the NSE immediately.
             if !effects.contentKeysChanged.isEmpty {
                 GameEntity.rebuildContentKeyDirectory(in: ctx)
             }
             return effects
         }
 
         for message in effects.traces {
             await trace(message)
         }
         if let onRemoteMovesUpdated, !effects.movesUpdated.isEmpty {
             await onRemoteMovesUpdated(effects.movesUpdated)
         }
         if let onRemotePlayersUpdated, !effects.playersUpdated.isEmpty {
             await onRemotePlayersUpdated(effects.playersUpdated)
         }
         if let onRemotePlayerPresenceChanged, !effects.playerPresenceChanged.isEmpty {
             await onRemotePlayerPresenceChanged(effects.playerPresenceChanged)
         }
         if let onRemoteEngagementChanged, !effects.engagementChanged.isEmpty {
             await onRemoteEngagementChanged(effects.engagementChanged)
         }
         if let onIncomingReadCursor, !effects.readCursors.isEmpty {
             await onIncomingReadCursor(effects.readCursors)
         }
         // A game just learned it's complete via sync: upload this device's
         // journal (no-op if it logged nothing) so replay can converge. The
         // enqueue defers its CKSyncEngine drain via `sendChangesDetached`.
         if let localAuthorID, !localAuthorID.isEmpty {
             for id in effects.completedTransitions {
                 enqueueJournalUpload(gameID: id, authorID: localAuthorID)
             }
         }
         if let onGameCompleted {
             for id in effects.completedTransitions {
                 await onGameCompleted(id)
             }
         }
         let deletedPings = deletions.compactMap { deletion -> (recordName: String, gameID: UUID)? in
             let recordName = deletion.0.recordName
             guard recordName.hasPrefix("ping-"),
                   let gameID = gameID(fromRecordName: recordName)
             else { return nil }
             return (recordName, gameID)
         }
         if let onPingDeleted, !deletedPings.isEmpty {
             await onPingDeleted(deletedPings)
         }
         if !effects.rosterRelevant.isEmpty {
             NotificationCenter.default.post(
                 name: .playerRosterShouldRefresh,
                 object: nil,
                 userInfo: ["gameIDs": effects.rosterRelevant]
             )
         }
     }
 
     nonisolated func applyPlayerRecord(
         _ record: CKRecord,
         in ctx: NSManagedObjectContext,
         localAuthorID: String?,
         onFirstTime: (UUID) -> Void,
         onPresenceChange: (UUID) -> Void,
         onReadCursor: (UUID, Date, Data?) -> Void
     ) {
         let ckName = record.recordID.recordName
         guard let (gameID, authorID) = RecordSerializer.parsePlayerRecordName(ckName) else {
             return
         }
         guard let renderedName = record["name"] as? String else { return }
         let updatedAt = record["updatedAt"] as? Date
             ?? record.modificationDate
             ?? Date()
 
         let req = NSFetchRequest<PlayerEntity>(entityName: "PlayerEntity")
         req.predicate = NSPredicate(format: "ckRecordName == %@", ckName)
         req.fetchLimit = 1
 
         let entity: PlayerEntity
         let foundExisting: Bool
         if let existing = try? ctx.fetch(req).first {
             entity = existing
             foundExisting = true
         } else {
             let game = RecordSerializer.ensureGameEntity(
                 forGameID: gameID,
                 zoneID: record.recordID.zoneID,
                 in: ctx
             )
             entity = PlayerEntity(context: ctx)
             entity.game = game
             foundExisting = false
         }
 
         // Drop fetched snapshots older than what we already have. After a
         // successful push the writeback adopts the new etag; if a query
         // that started before the push lands later, applying its older
         // snapshot would downgrade our local etag and OpLock-fail the next
         // save (see `applyMovesRecord` for the same guard).
         if foundExisting,
            !RecordSerializer.incomingIsAtLeastAsFresh(record, existingFields: entity.ckSystemFields) {
             return
         }
 
         let oldSelection = (entity.selRow, entity.selCol, entity.selDir)
         let hadSelection = oldSelection.0 != nil && oldSelection.1 != nil && oldSelection.2 != nil
         let oldUpdatedAt = entity.updatedAt
 
         // Adopt the server's system fields — that's etag tracking and is
         // independent of which side has the freshest data.
         entity.ckRecordName = ckName
         entity.ckSystemFields = RecordSerializer.encodeSystemFields(of: record)
         entity.authorID = authorID
 
         // The read cursor and session snapshot are account-scoped convergence
         // state — "what this account has already seen" — not the live cursor, so
         // they are adopted ahead of (and independent of) the selection's
         // `updatedAt` LWW below. A sibling commits the catch-up baseline on leave
         // (`handlePuzzleLeft`), which advances the read cursor and writes the
         // snapshot but does *not* bump `updatedAt`; the outbound record therefore
         // ships a stale `updatedAt`. Gating these on it would let a device with a
         // fresher local cursor drop the baseline and re-report the same moves as
         // a duplicate catch-up banner. The etag guard above already rejects
         // genuinely stale fetches. Adopting the cursor here is *not* monotonic:
         // `noteIncomingReadCursor` adopts the inbound value directly under
         // last-writer-wins, so a leaving sibling's past horizon *can* pull the
         // account horizon back below another sibling's live presence lease.
         // That collapse is bounded and self-healing: the still-present device
         // re-asserts its lease as soon as it processes the inbound close
         // (AppServices' incoming-cursor drain re-runs `publishReadCursor`
         // ahead of the 5-min refresh floor), and a foreground device marks
         // inbound peer moves read on arrival regardless. Keeping "A left
         // while C is still here" representable without the dip would need
         // per-device Player rows, which don't exist (one row per author).
         let previousReadAt = entity.readAt
         let previousSessionSnapshot = entity.sessionSnapshot
         let incomingReadAt = RecordSerializer.parsePlayerReadAt(from: record)
         entity.readAt = incomingReadAt
         let incomingReadThrough = RecordSerializer.parsePlayerReadThrough(from: record)
         entity.readThrough = incomingReadThrough
         entity.sessionSnapshot = RecordSerializer.parsePlayerSessionSnapshot(from: record)
         // `timeLog` is the device-keyed solve-time log. Touch it only when the
         // fetched record actually carries the field. A partial fetch (CloudQuery
         // restricts `desiredKeys`) or an older record omits it, and because the
         // log only ever grows, an absent field never means "cleared" — adopting
         // nil there would erase a peer's or a sibling's intervals until the next
         // full sync re-delivered them. Applied outside the `updatedAt` freshness
         // guard below: like the read cursor, a sibling's leave-write can ship a
         // stale `updatedAt`.
         if record.allKeys().contains("timeLog") {
             let incomingTimeLog = RecordSerializer.parsePlayerTimeLog(from: record)
             if authorID == localAuthorID {
                 // Our own record echoing back from a sibling: merge by device,
                 // keeping this device's own slot (we are its sole writer, and the
                 // sibling's copy may have dropped a session we have open now).
                 var local = TimeLog.decode(entity.timeLog)
                 local.merge(
                     inbound: TimeLog.decode(incomingTimeLog),
                     preservingDevice: RecordSerializer.localDeviceID
                 )
                 entity.timeLog = local.devices.isEmpty ? nil : TimeLog.encode(local)
             } else {
                 entity.timeLog = incomingTimeLog
             }
         }
         // Only surface the cursor when the row actually changed. A re-application
         // of values we already hold — e.g. a catch-up query snapshot racing this
         // device's in-flight lease save shares the old etag, so the freshness
         // guard above admits it — carries no new account-horizon information,
         // and adopting it would rewind the just-minted lease and trigger a
         // redundant re-assert save (the open-time double mint).
         if authorID == localAuthorID, let readAt = incomingReadAt,
            readAt != previousReadAt || entity.sessionSnapshot != previousSessionSnapshot {
             onReadCursor(gameID, readAt, entity.sessionSnapshot)
         }
         // Our own record coming back from another of this account's devices
         // carries that device's read watermark. Adopt it monotonically onto the
         // shared game so the unread badge clears here once any device has read,
         // mirroring how the presence lease converges via `onReadCursor` above.
         if authorID == localAuthorID,
            let incomingReadThrough,
            let game = entity.game,
            (game.readThroughAt ?? .distantPast) < incomingReadThrough {
             game.readThroughAt = incomingReadThrough
         }
 
         // The remaining value fields are only adopted when the incoming record
         // is at least as new as what we have locally; otherwise a stale-but-
         // current server record (e.g. our own pending writes haven't landed yet)
         // would clobber the user's live selection on every fetch.
         let localUpdatedAt = entity.updatedAt
         let incomingIsFresher = localUpdatedAt.map { updatedAt >= $0 } ?? true
         guard incomingIsFresher else { return }
         // An empty `name` is what older builds shipped from the selection publisher
         // before the fix; treat it as "no information" rather than letting it
         // clobber a previously-resolved name.
         if !renderedName.isEmpty {
             entity.name = renderedName
         }
         entity.updatedAt = updatedAt
         if let selection = RecordSerializer.parsePlayerSelection(from: record) {
             entity.selRow = NSNumber(value: Int64(selection.row))
             entity.selCol = NSNumber(value: Int64(selection.col))
             entity.selDir = NSNumber(value: Int64(selection.direction.rawValue))
         } else {
             entity.selRow = nil
             entity.selCol = nil
             entity.selDir = nil
         }
         // Adopt the record's push address. For a peer this is how the sender
         // learns where to address pushes; for our own record synced from a
         // sibling device, it's how the account's devices converge on one
         // per-game address (the LWW winner of this record).
         entity.pushAddress = RecordSerializer.parsePlayerPushAddress(from: record)
         let isRemoteAuthor = authorID != localAuthorID && authorID != CKCurrentUserDefaultName
         let hasSelection = entity.selRow != nil && entity.selCol != nil && entity.selDir != nil
         if isRemoteAuthor,
            (hadSelection || hasSelection),
            oldUpdatedAt != entity.updatedAt ||
                oldSelection.0 != entity.selRow ||
                oldSelection.1 != entity.selCol ||
                oldSelection.2 != entity.selDir {
             onPresenceChange(gameID)
         }
         if !foundExisting {
             onFirstTime(gameID)
         }
     }
 
     /// Merges every device's `MovesEntity` row for `gameID` and reconciles the
     /// `CellEntity` cache against the resulting grid. Must be called inside a
     /// `performAndWait` block on the same context. Returns diagnostic messages
     /// for any fetch failure (normally empty) — the caller folds them into
     /// `BatchEffects.traces` so they reach the diagnostics log once the batch
     /// context unwinds.
     nonisolated func replayCellCache(
         for gameID: UUID,
         in ctx: NSManagedObjectContext
     ) -> [String] {
         let gameReq = NSFetchRequest<GameEntity>(entityName: "GameEntity")
         gameReq.predicate = NSPredicate(format: "id == %@", gameID as CVarArg)
         gameReq.fetchLimit = 1
         let game: GameEntity?
         do {
             game = try ctx.fetch(gameReq).first
         } catch {
             // CKSyncEngine commits the batch when the delegate returns
             // (see fetchedRecordZoneChanges save-failure note), so re-throwing
             // won't redeliver — surface the failure instead of silently
             // leaving the cell cache stale for this game.
             return [Self.syncErrorMessage("replayCellCache game fetch", gameID: gameID, error: error)]
         }
         guard let game else { return [] }
 
         let movesReq = NSFetchRequest<MovesEntity>(entityName: "MovesEntity")
         movesReq.predicate = NSPredicate(format: "game == %@", game)
         let movesEntities: [MovesEntity]
         do {
             movesEntities = try ctx.fetch(movesReq)
         } catch {
             return [Self.syncErrorMessage("replayCellCache moves fetch", gameID: gameID, error: error)]
         }
         let values: [MovesValue] = movesEntities.compactMap { Self.movesValue(from: $0) }
         let gridState = GridStateMerger.merge(values)
 
         let existingCells = (game.cells as? Set<CellEntity>) ?? []
         var byPosition: [GridPosition: CellEntity] = [:]
         for cell in existingCells {
             byPosition[GridPosition(row: Int(cell.row), col: Int(cell.col))] = cell
         }
 
         for (pos, gridCell) in gridState {
             let cell: CellEntity
             if let existing = byPosition[pos] {
                 cell = existing
             } else {
                 cell = CellEntity(context: ctx)
                 cell.game = game
                 cell.row = Int16(pos.row)
                 cell.col = Int16(pos.col)
             }
             cell.letter = gridCell.letter
             cell.markCode = gridCell.mark.code
             cell.letterAuthorID = gridCell.authorID
         }
 
         for (pos, cell) in byPosition where gridState[pos] == nil {
             cell.letter = ""
             cell.markCode = 0
             cell.letterAuthorID = nil
         }
         return []
     }
 
     /// Hydrates a `MovesValue` from a `MovesEntity`. Returns `nil` if the row
     /// is missing required fields (e.g. an unpopulated stub from a partial
     /// fetch).
     nonisolated static func movesValue(from entity: MovesEntity) -> MovesValue? {
         guard let gameID = entity.game?.id,
               let authorID = entity.authorID,
               let deviceID = entity.deviceID,
               let updatedAt = entity.updatedAt
         else { return nil }
         let cells = (entity.cells.flatMap { try? MovesCodec.decode($0) }) ?? [:]
         return MovesValue(
             gameID: gameID,
             authorID: authorID,
             deviceID: deviceID,
             cells: cells,
             updatedAt: updatedAt
         )
     }
 
     /// Formats a sync-context fetch/save failure for the diagnostics log. The
     /// engine's change token has already advanced by the time these helpers
     /// run inside performAndWait, so the only available remediation is making
     /// the drop visible — and visible means traced (the on-device diagnostics
     /// log), not printed: console output never reaches a collected log.
     nonisolated static func syncErrorMessage(_ label: String, gameID: UUID, error: Error) -> String {
         let nsError = error as NSError
         return "\(label) FAILED for \(gameID.uuidString) " +
             "— domain=\(nsError.domain) code=\(nsError.code) " +
             "\(nsError.localizedDescription)"
     }
 
     /// Applies an inbound `Archive` record. Inert while a live (non-revoked)
     /// copy of the original game still exists on this device — the device already
     /// holds the data, so surfacing a second row would duplicate it. On a device
     /// without the original (fresh install / reinstall), it hydrates the snapshot
     /// into a standalone completed, owned game. Returns the materialized game id
     /// when a row was created, else `nil`.
     @discardableResult
     nonisolated func applyArchiveRecord(
         _ record: CKRecord,
         in ctx: NSManagedObjectContext
     ) -> UUID? {
         guard let payload = Archive.payload(from: record) else { return nil }
 
         let liveReq = NSFetchRequest<GameEntity>(entityName: "GameEntity")
         liveReq.predicate = NSPredicate(
             format: "id == %@ AND isAccessRevoked == NO",
             payload.originalGameID as CVarArg
         )
         liveReq.fetchLimit = 1
         if (try? ctx.fetch(liveReq).first) != nil { return nil }
 
         let created = Archive.materialize(payload, in: ctx)
         return created?.id
     }
 
     nonisolated func applyDeletion(
         recordID: CKRecord.ID,
         recordType: CKRecord.RecordType,
         in ctx: NSManagedObjectContext
     ) {
         let name = recordID.recordName
         let entityName: String
         if name.hasPrefix("moves-") {
             entityName = "MovesEntity"
         } else if name.hasPrefix("player-") {
             entityName = "PlayerEntity"
         } else if name.hasPrefix("game-") {
             entityName = "GameEntity"
         } else {
             switch recordType {
             case "Moves": entityName = "MovesEntity"
             case "Player": entityName = "PlayerEntity"
             case "Game": entityName = "GameEntity"
             default: return
             }
         }
         let req = NSFetchRequest<NSManagedObject>(entityName: entityName)
         req.predicate = NSPredicate(format: "ckRecordName == %@", name)
         req.fetchLimit = 1
         if let obj = try? ctx.fetch(req).first {
             ctx.delete(obj)
         }
     }
 }