crossmate

EngagementCoordinator.swift (15864B)

---

 import Foundation
 
 struct EngagementAddressee: Equatable, Sendable {
     var authorID: String
     var deviceID: String?
 
     var rawValue: String {
         if let deviceID, !deviceID.isEmpty {
             return "\(authorID):\(deviceID)"
         }
         return authorID
     }
 
     static func parse(_ rawValue: String?) -> EngagementAddressee? {
         guard let rawValue, !rawValue.isEmpty else { return nil }
         let parts = rawValue.split(separator: ":", maxSplits: 1, omittingEmptySubsequences: false)
         guard let author = parts.first, !author.isEmpty else { return nil }
         let device = parts.count == 2 ? String(parts[1]) : nil
         return EngagementAddressee(authorID: String(author), deviceID: device?.isEmpty == true ? nil : device)
     }
 
     func matches(authorID: String, deviceID: String) -> Bool {
         guard self.authorID == authorID else { return false }
         guard let targetDeviceID = self.deviceID else { return true }
         return targetDeviceID == deviceID
     }
 }
 
 struct EngagementRoomCredentials: Codable, Equatable, Sendable {
     var ver: Int
     var roomID: UUID
     var secret: String
     var createdAt: Date
     var expiresAt: Date
 
     init(
         roomID: UUID = UUID(),
         secret: String,
         createdAt: Date = Date(),
         expiresAt: Date,
         ver: Int = 2
     ) {
         self.ver = ver
         self.roomID = roomID
         self.secret = secret
         self.createdAt = createdAt
         self.expiresAt = expiresAt
     }
 
     func encoded() throws -> String {
         let data = try JSONEncoder().encode(self)
         guard let string = String(data: data, encoding: .utf8) else {
             throw EngagementCoordinatorError.invalidPayloadEncoding
         }
         return string
     }
 
     static func decode(_ string: String?) -> EngagementRoomCredentials? {
         guard let data = string?.data(using: .utf8) else { return nil }
         return try? JSONDecoder().decode(EngagementRoomCredentials.self, from: data)
     }
 
     static func fresh(now: Date = Date(), ttl: TimeInterval = 10 * 60) throws -> EngagementRoomCredentials {
         try EngagementRoomCredentials(
             secret: Data.secureRandom(count: 32).base64URLEncodedString(),
             createdAt: now,
             expiresAt: now.addingTimeInterval(ttl)
         )
     }
 }
 
 struct EngagementMessage: Codable, Equatable, Sendable {
     enum Kind: String, Codable, Sendable {
         case debugText
         case cellEdit
         case cellEditBatch
         case selection
     }
 
     var kind: Kind
     var text: String
     var cellEdit: RealtimeCellEdit?
     var cellEdits: [RealtimeCellEdit]?
     var selection: EngagementSelectionUpdate?
     var sentAt: Date
     var ver: Int
 
     init(
         kind: Kind = .debugText,
         text: String,
         cellEdit: RealtimeCellEdit? = nil,
         cellEdits: [RealtimeCellEdit]? = nil,
         selection: EngagementSelectionUpdate? = nil,
         sentAt: Date = Date(),
         ver: Int = 1
     ) {
         self.kind = kind
         self.text = text
         self.cellEdit = cellEdit
         self.cellEdits = cellEdits
         self.selection = selection
         self.sentAt = sentAt
         self.ver = ver
     }
 
     init(cellEdit: RealtimeCellEdit, sentAt: Date = Date(), ver: Int = 1) {
         self.kind = .cellEdit
         self.text = ""
         self.cellEdit = cellEdit
         self.cellEdits = nil
         self.selection = nil
         self.sentAt = sentAt
         self.ver = ver
     }
 
     /// Carries one bulk gesture (check/clear/multi-cell undo) as a single
     /// message. Peers that predate this kind fail to decode it and drop the
     /// live update; the same cells still arrive durably via the Moves/CloudKit
     /// path, so they degrade to "appears on sync" rather than breaking.
     init(cellEdits: [RealtimeCellEdit], sentAt: Date = Date(), ver: Int = 1) {
         self.kind = .cellEditBatch
         self.text = ""
         self.cellEdit = nil
         self.cellEdits = cellEdits
         self.selection = nil
         self.sentAt = sentAt
         self.ver = ver
     }
 
     init(selection: EngagementSelectionUpdate, sentAt: Date = Date(), ver: Int = 1) {
         self.kind = .selection
         self.text = ""
         self.cellEdit = nil
         self.cellEdits = nil
         self.selection = selection
         self.sentAt = sentAt
         self.ver = ver
     }
 
     func encodedData() throws -> Data {
         try JSONEncoder().encode(self)
     }
 
     static func decode(_ data: Data) -> EngagementMessage? {
         try? JSONDecoder().decode(EngagementMessage.self, from: data)
     }
 }
 
 enum EngagementReconcileOutcome: Equatable, Sendable {
     /// The reconcile ran; any connect failure is transient and covered by the
     /// normal retry backstops.
     case reconciled
     /// The worker refused the advertised room because it is registered under a
     /// different secret. Retrying with the same creds can never succeed; the
     /// caller should clear the advertised creds so a fresh room is minted.
     case roomRejected
 }
 
 @MainActor
 protocol EngagementTransporting: AnyObject, Sendable {
     func connect(
         engagementID: UUID,
         room: EngagementRoomCredentials,
         authorID: String,
         deviceID: String
     ) async throws
     func send(engagementID: UUID, message: Data) async throws
     func disconnect(engagementID: UUID)
 }
 
 actor EngagementCoordinator {
     typealias Log = @Sendable (_ message: String) async -> Void
 
     private enum State: Equatable {
         case idle
         case connecting(engagementID: UUID, room: EngagementRoomCredentials, at: Date)
         case live(engagementID: UUID, room: EngagementRoomCredentials)
 
         var engagementID: UUID? {
             switch self {
             case .idle:
                 nil
             case .connecting(let engagementID, _, _),
                  .live(let engagementID, _):
                 engagementID
             }
         }
 
         var roomID: UUID? {
             switch self {
             case .idle:
                 nil
             case .connecting(_, let room, _),
                  .live(_, let room):
                 room.roomID
             }
         }
     }
 
     private let host: any EngagementTransporting
     private let localAuthorID: @Sendable () async -> String?
     private let localDeviceID: String
     private let log: Log
     private let now: @Sendable () -> Date
     private let connectionTimeout: TimeInterval
     private var states: [UUID: State] = [:]
 
     init(
         host: any EngagementTransporting,
         localAuthorID: @escaping @Sendable () async -> String?,
         localDeviceID: String = RecordSerializer.localDeviceID,
         log: @escaping Log = { _ in },
         now: @escaping @Sendable () -> Date = Date.init,
         connectionTimeout: TimeInterval = 30
     ) {
         self.host = host
         self.localAuthorID = localAuthorID
         self.localDeviceID = localDeviceID
         self.log = log
         self.now = now
         self.connectionTimeout = connectionTimeout
     }
 
     /// Drives this game's live connection toward the room the shared Game
     /// record currently advertises. `creds` is the decoded `engagement` field
     /// (nil if none minted yet); `hasPeer` is whether any peer holds a live
     /// read lease. The desired state is: connected to `creds.roomID` when a
     /// peer is present and creds exist, disconnected otherwise.
     ///
     /// This subsumes connect, reconnect, migrate (a peer rotated the room), and
     /// teardown (peer left) — and the create race needs no arbiter: if two
     /// participants mint, the Game record's LWW picks one set of creds, and
     /// every device reconciles onto it, the loser migrating off its own room.
     @discardableResult
     func reconcile(gameID: UUID, creds: EngagementRoomCredentials?, hasPeer: Bool) async -> EngagementReconcileOutcome {
         await sweepStaleConnections()
         let current = state(for: gameID)
         guard hasPeer, let creds else {
             // No peer (or no creds) → disconnect. Leave state for `.channelClose`
             // to clear so the normal cleanup runs downstream.
             if let engagementID = current.engagementID {
                 await log("engagement: no present peer for \(gameID.uuidString), tearing down \(engagementID.uuidString)")
                 await host.disconnect(engagementID: engagementID)
             }
             return .reconciled
         }
         // Already connecting/live to the advertised room — nothing to do.
         if current.roomID == creds.roomID { return .reconciled }
         // Connect to the desired room first (so the old socket's `.channelClose`
         // sees a state that has already moved on and no-ops), then drop the old.
         let staleEngagementID = current.engagementID
         let outcome = await connect(gameID: gameID, room: creds)
         if let staleEngagementID {
             await host.disconnect(engagementID: staleEngagementID)
         }
         return outcome
     }
 
     private func sweepStaleConnections() async {
         let cutoff = now()
         var demoted: [(gameID: UUID, engagementID: UUID, age: TimeInterval)] = []
         for (gameID, state) in states {
             guard case .connecting(let engagementID, _, let at) = state else { continue }
             let age = cutoff.timeIntervalSince(at)
             if age > connectionTimeout {
                 demoted.append((gameID, engagementID, age))
                 states[gameID] = .idle
             }
         }
         for entry in demoted {
             await host.disconnect(engagementID: entry.engagementID)
             await log(
                 "engagement: connection timed out for \(entry.gameID.uuidString) " +
                 "after \(Int(entry.age))s, engagement \(entry.engagementID.uuidString)"
             )
         }
     }
 
     func teardown(gameID: UUID) async {
         let state = state(for: gameID)
         states[gameID] = .idle
         if let engagementID = state.engagementID {
             await host.disconnect(engagementID: engagementID)
         }
     }
 
     func channelOpened(engagementID: UUID) async -> UUID? {
         guard let (gameID, state) = stateEntry(for: engagementID) else { return nil }
         switch state {
         case .idle:
             return nil
         case .connecting(_, let room, _),
              .live(_, let room):
             states[gameID] = .live(engagementID: engagementID, room: room)
             return gameID
         }
     }
 
     func channelClosed(engagementID: UUID) async -> UUID? {
         guard let (gameID, state) = stateEntry(for: engagementID) else { return nil }
         if state.engagementID == engagementID {
             states[gameID] = .idle
             return gameID
         }
         return nil
     }
 
     func sendDebugMessage(gameID: UUID, text: String) async {
         guard case .live(let engagementID, _) = state(for: gameID) else {
             await log("engagement: test message skipped for \(gameID.uuidString), channel is not live")
             return
         }
         do {
             let message = EngagementMessage(text: text)
             try await host.send(engagementID: engagementID, message: message.encodedData())
             await log("engagement: sent test message \(engagementID.uuidString)")
         } catch {
             await log("engagement: test message failed \(engagementID.uuidString): \(error.localizedDescription)")
         }
     }
 
     func sendCellEdit(_ edit: RealtimeCellEdit) async {
         guard case .live(let engagementID, _) = state(for: edit.gameID) else { return }
         do {
             let message = EngagementMessage(cellEdit: edit)
             try await host.send(engagementID: engagementID, message: message.encodedData())
             await log(
                 "engagement: sent cellEdit \(engagementID.uuidString) " +
                 "r=\(edit.row) c=\(edit.col) device=\(edit.deviceID.prefix(8))"
             )
         } catch {
             await log("engagement: cell edit send failed \(engagementID.uuidString): \(error.localizedDescription)")
         }
     }
 
     /// Ships a bulk gesture as one batched message instead of one per cell, so
     /// a whole-grid action lands on the peer in a single frame. All edits in a
     /// batch share a game (they originate from one local gesture).
     func sendCellEdits(_ edits: [RealtimeCellEdit]) async {
         guard let first = edits.first else { return }
         guard case .live(let engagementID, _) = state(for: first.gameID) else { return }
         do {
             let message = EngagementMessage(cellEdits: edits)
             try await host.send(engagementID: engagementID, message: message.encodedData())
             await log(
                 "engagement: sent cellEditBatch \(engagementID.uuidString) " +
                 "count=\(edits.count) device=\(first.deviceID.prefix(8))"
             )
         } catch {
             await log("engagement: cell edit batch send failed \(engagementID.uuidString): \(error.localizedDescription)")
         }
     }
 
     func sendSelection(_ selection: EngagementSelectionUpdate) async {
         guard case .live(let engagementID, _) = state(for: selection.gameID) else { return }
         do {
             let message = EngagementMessage(selection: selection)
             try await host.send(engagementID: engagementID, message: message.encodedData())
             await log(
                 "engagement: sent selection \(engagementID.uuidString) " +
                 "r=\(selection.row) c=\(selection.col) device=\(selection.deviceID.prefix(8))"
             )
         } catch {
             await log("engagement: selection send failed \(engagementID.uuidString): \(error.localizedDescription)")
         }
     }
 
     /// Opens a socket to `room` and parks the game in `.connecting`; the
     /// `.channelOpen` callback promotes it to `.live`. Callers that are
     /// migrating off a previous room disconnect the stale engagement *after*
     /// this returns, so the old socket's close races against a state that has
     /// already moved on.
     private func connect(gameID: UUID, room: EngagementRoomCredentials) async -> EngagementReconcileOutcome {
         guard let localAuthorID = await localAuthorID(), !localAuthorID.isEmpty else {
             await log("engagement: connect skipped for \(gameID.uuidString), missing local author")
             return .reconciled
         }
         let engagementID = UUID()
         states[gameID] = .connecting(engagementID: engagementID, room: room, at: now())
         do {
             try await host.connect(
                 engagementID: engagementID,
                 room: room,
                 authorID: localAuthorID,
                 deviceID: localDeviceID
             )
             await log("engagement: connecting \(gameID.uuidString) to room \(room.roomID.uuidString)")
         } catch EngagementHostError.roomSecretMismatch {
             states[gameID] = .idle
             await log(
                 "engagement: room \(room.roomID.uuidString) rejected for \(gameID.uuidString): " +
                 "registered with a different secret"
             )
             return .roomRejected
         } catch {
             states[gameID] = .idle
             await log("engagement: connect failed for \(gameID.uuidString): \(error.localizedDescription)")
         }
         return .reconciled
     }
 
     private func state(for gameID: UUID) -> State {
         states[gameID] ?? .idle
     }
 
     private func stateEntry(for engagementID: UUID) -> (UUID, State)? {
         states.first { _, state in
             state.engagementID == engagementID
         }
     }
 }
 
 enum EngagementCoordinatorError: LocalizedError {
     case invalidPayloadEncoding
 
     var errorDescription: String? {
         switch self {
         case .invalidPayloadEncoding:
             "Unable to encode engagement room payload."
         }
     }
 }