EngagementCoordinator.swift (20408B)
1 import Foundation 2 3 struct EngagementAddressee: Equatable, Sendable { 4 var authorID: String 5 var deviceID: String? 6 7 var rawValue: String { 8 if let deviceID, !deviceID.isEmpty { 9 return "\(authorID):\(deviceID)" 10 } 11 return authorID 12 } 13 14 static func parse(_ rawValue: String?) -> EngagementAddressee? { 15 guard let rawValue, !rawValue.isEmpty else { return nil } 16 let parts = rawValue.split(separator: ":", maxSplits: 1, omittingEmptySubsequences: false) 17 guard let author = parts.first, !author.isEmpty else { return nil } 18 let device = parts.count == 2 ? String(parts[1]) : nil 19 return EngagementAddressee(authorID: String(author), deviceID: device?.isEmpty == true ? nil : device) 20 } 21 22 func matches(authorID: String, deviceID: String) -> Bool { 23 guard self.authorID == authorID else { return false } 24 guard let targetDeviceID = self.deviceID else { return true } 25 return targetDeviceID == deviceID 26 } 27 } 28 29 struct EngagementRoomCredentials: Codable, Equatable, Sendable { 30 var ver: Int 31 var roomID: UUID 32 var secret: String 33 var createdAt: Date 34 var expiresAt: Date 35 36 init( 37 roomID: UUID = UUID(), 38 secret: String, 39 createdAt: Date = Date(), 40 expiresAt: Date, 41 ver: Int = 2 42 ) { 43 self.ver = ver 44 self.roomID = roomID 45 self.secret = secret 46 self.createdAt = createdAt 47 self.expiresAt = expiresAt 48 } 49 50 func encoded() throws -> String { 51 let data = try JSONEncoder().encode(self) 52 guard let string = String(data: data, encoding: .utf8) else { 53 throw EngagementCoordinatorError.invalidPayloadEncoding 54 } 55 return string 56 } 57 58 static func decode(_ string: String?) -> EngagementRoomCredentials? { 59 guard let data = string?.data(using: .utf8) else { return nil } 60 return try? JSONDecoder().decode(EngagementRoomCredentials.self, from: data) 61 } 62 63 static func fresh(now: Date = Date(), ttl: TimeInterval = 10 * 60) throws -> EngagementRoomCredentials { 64 try EngagementRoomCredentials( 65 secret: Data.secureRandom(count: 32).base64URLEncodedString(), 66 createdAt: now, 67 expiresAt: now.addingTimeInterval(ttl) 68 ) 69 } 70 } 71 72 struct EngagementMessage: Codable, Equatable, Sendable { 73 enum Kind: String, Codable, Sendable { 74 case debugText 75 case cellEdit 76 case cellEditBatch 77 case selection 78 } 79 80 var kind: Kind 81 var text: String 82 var cellEdit: RealtimeCellEdit? 83 var cellEdits: [RealtimeCellEdit]? 84 var selection: EngagementSelectionUpdate? 85 var sentAt: Date 86 var ver: Int 87 /// The connection's claimed author. The relay is a blind pass-through, so 88 /// this is only trustworthy once the matching `auth` tag verifies. 89 var senderAuthorID: String 90 /// Per-recipient HMAC tags, keyed by recipient authorID (base64). The 91 /// receiver applies the frame only if the tag addressed to it verifies 92 /// against `key(for: senderAuthorID)`. See `EngagementMessageAuthenticator`. 93 var auth: [String: String] 94 95 init( 96 kind: Kind = .debugText, 97 text: String, 98 cellEdit: RealtimeCellEdit? = nil, 99 cellEdits: [RealtimeCellEdit]? = nil, 100 selection: EngagementSelectionUpdate? = nil, 101 sentAt: Date = Date(), 102 senderAuthorID: String = "", 103 auth: [String: String] = [:], 104 ver: Int = 1 105 ) { 106 self.kind = kind 107 self.text = text 108 self.cellEdit = cellEdit 109 self.cellEdits = cellEdits 110 self.selection = selection 111 self.sentAt = sentAt 112 self.ver = ver 113 self.senderAuthorID = senderAuthorID 114 self.auth = auth 115 } 116 117 init( 118 cellEdit: RealtimeCellEdit, 119 sentAt: Date = Date(), 120 senderAuthorID: String = "", 121 auth: [String: String] = [:], 122 ver: Int = 1 123 ) { 124 self.kind = .cellEdit 125 self.text = "" 126 self.cellEdit = cellEdit 127 self.cellEdits = nil 128 self.selection = nil 129 self.sentAt = sentAt 130 self.ver = ver 131 self.senderAuthorID = senderAuthorID 132 self.auth = auth 133 } 134 135 /// Carries one bulk gesture (check/clear/multi-cell undo) as a single 136 /// message. Peers that predate this kind fail to decode it and drop the 137 /// live update; the same cells still arrive durably via the Moves/CloudKit 138 /// path, so they degrade to "appears on sync" rather than breaking. 139 init( 140 cellEdits: [RealtimeCellEdit], 141 sentAt: Date = Date(), 142 senderAuthorID: String = "", 143 auth: [String: String] = [:], 144 ver: Int = 1 145 ) { 146 self.kind = .cellEditBatch 147 self.text = "" 148 self.cellEdit = nil 149 self.cellEdits = cellEdits 150 self.selection = nil 151 self.sentAt = sentAt 152 self.ver = ver 153 self.senderAuthorID = senderAuthorID 154 self.auth = auth 155 } 156 157 init( 158 selection: EngagementSelectionUpdate, 159 sentAt: Date = Date(), 160 senderAuthorID: String = "", 161 auth: [String: String] = [:], 162 ver: Int = 1 163 ) { 164 self.kind = .selection 165 self.text = "" 166 self.cellEdit = nil 167 self.cellEdits = nil 168 self.selection = selection 169 self.sentAt = sentAt 170 self.ver = ver 171 self.senderAuthorID = senderAuthorID 172 self.auth = auth 173 } 174 175 func encodedData() throws -> Data { 176 try JSONEncoder().encode(self) 177 } 178 179 /// Ingress bounds for the live channel. The relay is a blind pass-through, 180 /// so a hostile participant controls frame size and batch shape; every 181 /// limit here is enforced *before* the frame reaches authentication or 182 /// Core Data. The worker mirrors `maxEncodedFrameBytes` (see 183 /// `ROOM_MAX_FRAME_BYTES` in room-worker.js) and the sender chunks bulk 184 /// gestures to `maxBatchEdits` (`EngagementLifecycle.sendLocalCellEdits`), 185 /// so a legitimate peer never trips them. 186 /// 187 /// Sizing: a `maxBatchEdits` batch encodes to ~300 KB, plus one ~120-byte 188 /// auth tag per recipient (CloudKit shares top out at 100 participants), 189 /// comfortably inside `maxEncodedFrameBytes`. Cloudflare's own WebSocket 190 /// cap is 1 MiB, so the worker limit is meaningfully tighter than the 191 /// platform's. 192 static let maxEncodedFrameBytes = 512 * 1024 193 static let maxBatchEdits = 1024 194 static let maxAuthTags = 128 195 static let maxIdentifierLength = 128 196 static let maxLetterLength = 16 197 static let maxTextLength = 512 198 199 static func decode(_ data: Data) -> EngagementMessage? { 200 guard data.count <= maxEncodedFrameBytes else { return nil } 201 guard let message = try? JSONDecoder().decode(EngagementMessage.self, from: data), 202 message.isWithinBounds 203 else { return nil } 204 return message 205 } 206 207 /// Splits a bulk gesture into sendable batches. Order is preserved, but 208 /// chunks travel as independent frames, so per-cell last-writer-wins is 209 /// what guarantees convergence — same as any other frame reordering. 210 static func batchChunks(_ edits: [RealtimeCellEdit]) -> [[RealtimeCellEdit]] { 211 stride(from: 0, to: edits.count, by: maxBatchEdits).map { start in 212 Array(edits[start..<min(start + maxBatchEdits, edits.count)]) 213 } 214 } 215 216 private var isWithinBounds: Bool { 217 func boundedID(_ id: String?) -> Bool { 218 (id ?? "").count <= Self.maxIdentifierLength 219 } 220 func bounded(_ edit: RealtimeCellEdit) -> Bool { 221 boundedID(edit.authorID) && boundedID(edit.deviceID) 222 && boundedID(edit.cellAuthorID) 223 && edit.letter.count <= Self.maxLetterLength 224 } 225 guard boundedID(senderAuthorID), text.count <= Self.maxTextLength else { return false } 226 guard auth.count <= Self.maxAuthTags, 227 auth.allSatisfy({ boundedID($0.key) && boundedID($0.value) }) 228 else { return false } 229 if let cellEdit, !bounded(cellEdit) { return false } 230 if let cellEdits { 231 guard cellEdits.count <= Self.maxBatchEdits, cellEdits.allSatisfy(bounded) else { 232 return false 233 } 234 } 235 if let selection { 236 guard boundedID(selection.authorID), boundedID(selection.deviceID) else { return false } 237 } 238 return true 239 } 240 } 241 242 enum EngagementReconcileOutcome: Equatable, Sendable { 243 /// The reconcile ran; any connect failure is transient and covered by the 244 /// normal retry backstops. 245 case reconciled 246 /// The worker refused the advertised room because it is registered under a 247 /// different secret. Retrying with the same creds can never succeed; the 248 /// caller should clear the advertised creds so a fresh room is minted. 249 case roomRejected 250 } 251 252 @MainActor 253 protocol EngagementTransporting: AnyObject, Sendable { 254 func connect( 255 engagementID: UUID, 256 room: EngagementRoomCredentials, 257 authorID: String, 258 deviceID: String 259 ) async throws 260 func send(engagementID: UUID, message: Data) async throws 261 func disconnect(engagementID: UUID) 262 } 263 264 actor EngagementCoordinator { 265 typealias Log = @Sendable (_ message: String) async -> Void 266 267 private enum State: Equatable { 268 case idle 269 case connecting(engagementID: UUID, room: EngagementRoomCredentials, at: Date) 270 case live(engagementID: UUID, room: EngagementRoomCredentials) 271 272 var engagementID: UUID? { 273 switch self { 274 case .idle: 275 nil 276 case .connecting(let engagementID, _, _), 277 .live(let engagementID, _): 278 engagementID 279 } 280 } 281 282 var roomID: UUID? { 283 switch self { 284 case .idle: 285 nil 286 case .connecting(_, let room, _), 287 .live(_, let room): 288 room.roomID 289 } 290 } 291 } 292 293 private let host: any EngagementTransporting 294 private let localAuthorID: @Sendable () async -> String? 295 private let localDeviceID: String 296 private let log: Log 297 private let now: @Sendable () -> Date 298 private let connectionTimeout: TimeInterval 299 private var states: [UUID: State] = [:] 300 301 init( 302 host: any EngagementTransporting, 303 localAuthorID: @escaping @Sendable () async -> String?, 304 localDeviceID: String = RecordSerializer.localDeviceID, 305 log: @escaping Log = { _ in }, 306 now: @escaping @Sendable () -> Date = Date.init, 307 connectionTimeout: TimeInterval = 30 308 ) { 309 self.host = host 310 self.localAuthorID = localAuthorID 311 self.localDeviceID = localDeviceID 312 self.log = log 313 self.now = now 314 self.connectionTimeout = connectionTimeout 315 } 316 317 /// Drives this game's live connection toward the room the shared Game 318 /// record currently advertises. `creds` is the decoded `engagement` field 319 /// (nil if none minted yet); `hasPeer` is whether any peer holds a live 320 /// read lease. The desired state is: connected to `creds.roomID` when a 321 /// peer is present and creds exist, disconnected otherwise. 322 /// 323 /// This subsumes connect, reconnect, migrate (a peer rotated the room), and 324 /// teardown (peer left) — and the create race needs no arbiter: if two 325 /// participants mint, the Game record's LWW picks one set of creds, and 326 /// every device reconciles onto it, the loser migrating off its own room. 327 @discardableResult 328 func reconcile(gameID: UUID, creds: EngagementRoomCredentials?, hasPeer: Bool) async -> EngagementReconcileOutcome { 329 await sweepStaleConnections() 330 let current = state(for: gameID) 331 guard hasPeer, let creds else { 332 // No peer (or no creds) → disconnect. Leave state for `.channelClose` 333 // to clear so the normal cleanup runs downstream. 334 if let engagementID = current.engagementID { 335 await log("engagement: no present peer for \(gameID.uuidString), tearing down \(engagementID.uuidString)") 336 await host.disconnect(engagementID: engagementID) 337 } 338 return .reconciled 339 } 340 // Already connecting/live to the advertised room — nothing to do. 341 if current.roomID == creds.roomID { return .reconciled } 342 // Connect to the desired room first (so the old socket's `.channelClose` 343 // sees a state that has already moved on and no-ops), then drop the old. 344 let staleEngagementID = current.engagementID 345 let outcome = await connect(gameID: gameID, room: creds) 346 if let staleEngagementID { 347 await host.disconnect(engagementID: staleEngagementID) 348 } 349 return outcome 350 } 351 352 private func sweepStaleConnections() async { 353 let cutoff = now() 354 var demoted: [(gameID: UUID, engagementID: UUID, age: TimeInterval)] = [] 355 for (gameID, state) in states { 356 guard case .connecting(let engagementID, _, let at) = state else { continue } 357 let age = cutoff.timeIntervalSince(at) 358 if age > connectionTimeout { 359 demoted.append((gameID, engagementID, age)) 360 states[gameID] = .idle 361 } 362 } 363 for entry in demoted { 364 await host.disconnect(engagementID: entry.engagementID) 365 await log( 366 "engagement: connection timed out for \(entry.gameID.uuidString) " + 367 "after \(Int(entry.age))s, engagement \(entry.engagementID.uuidString)" 368 ) 369 } 370 } 371 372 func teardown(gameID: UUID) async { 373 let state = state(for: gameID) 374 states[gameID] = .idle 375 if let engagementID = state.engagementID { 376 await host.disconnect(engagementID: engagementID) 377 } 378 } 379 380 func channelOpened(engagementID: UUID) async -> UUID? { 381 guard let (gameID, state) = stateEntry(for: engagementID) else { return nil } 382 switch state { 383 case .idle: 384 return nil 385 case .connecting(_, let room, _), 386 .live(_, let room): 387 states[gameID] = .live(engagementID: engagementID, room: room) 388 return gameID 389 } 390 } 391 392 /// The game a channel is bound to (connecting or live), or nil for an 393 /// unknown or stale engagement. An inbound frame may only be applied to 394 /// this game; a frame arriving on a channel that no longer resolves is 395 /// dropped and converges via Moves sync instead. 396 func gameID(for engagementID: UUID) -> UUID? { 397 stateEntry(for: engagementID)?.0 398 } 399 400 func channelClosed(engagementID: UUID) async -> UUID? { 401 guard let (gameID, state) = stateEntry(for: engagementID) else { return nil } 402 if state.engagementID == engagementID { 403 states[gameID] = .idle 404 return gameID 405 } 406 return nil 407 } 408 409 func sendDebugMessage(gameID: UUID, text: String) async { 410 guard case .live(let engagementID, _) = state(for: gameID) else { 411 await log("engagement: test message skipped for \(gameID.uuidString), channel is not live") 412 return 413 } 414 do { 415 let message = EngagementMessage(text: text) 416 try await host.send(engagementID: engagementID, message: message.encodedData()) 417 await log("engagement: sent test message \(engagementID.uuidString)") 418 } catch { 419 await log("engagement: test message failed \(engagementID.uuidString): \(error.localizedDescription)") 420 } 421 } 422 423 func sendCellEdit(_ edit: RealtimeCellEdit, senderAuthorID: String, auth: [String: String]) async { 424 guard case .live(let engagementID, _) = state(for: edit.gameID) else { return } 425 do { 426 let message = EngagementMessage(cellEdit: edit, senderAuthorID: senderAuthorID, auth: auth) 427 try await host.send(engagementID: engagementID, message: message.encodedData()) 428 await log( 429 "engagement: sent cellEdit \(engagementID.uuidString) " + 430 "r=\(edit.row) c=\(edit.col) device=\(edit.deviceID.prefix(8))" 431 ) 432 } catch { 433 await log("engagement: cell edit send failed \(engagementID.uuidString): \(error.localizedDescription)") 434 } 435 } 436 437 /// Ships a bulk gesture as one batched message instead of one per cell, so 438 /// a whole-grid action lands on the peer in a single frame. All edits in a 439 /// batch share a game (they originate from one local gesture). 440 func sendCellEdits(_ edits: [RealtimeCellEdit], senderAuthorID: String, auth: [String: String]) async { 441 guard let first = edits.first else { return } 442 guard case .live(let engagementID, _) = state(for: first.gameID) else { return } 443 do { 444 let message = EngagementMessage(cellEdits: edits, senderAuthorID: senderAuthorID, auth: auth) 445 try await host.send(engagementID: engagementID, message: message.encodedData()) 446 await log( 447 "engagement: sent cellEditBatch \(engagementID.uuidString) " + 448 "count=\(edits.count) device=\(first.deviceID.prefix(8))" 449 ) 450 } catch { 451 await log("engagement: cell edit batch send failed \(engagementID.uuidString): \(error.localizedDescription)") 452 } 453 } 454 455 func sendSelection(_ selection: EngagementSelectionUpdate, senderAuthorID: String, auth: [String: String]) async { 456 guard case .live(let engagementID, _) = state(for: selection.gameID) else { return } 457 do { 458 let message = EngagementMessage(selection: selection, senderAuthorID: senderAuthorID, auth: auth) 459 try await host.send(engagementID: engagementID, message: message.encodedData()) 460 await log( 461 "engagement: sent selection \(engagementID.uuidString) " + 462 "r=\(selection.row) c=\(selection.col) device=\(selection.deviceID.prefix(8))" 463 ) 464 } catch { 465 await log("engagement: selection send failed \(engagementID.uuidString): \(error.localizedDescription)") 466 } 467 } 468 469 /// Opens a socket to `room` and parks the game in `.connecting`; the 470 /// `.channelOpen` callback promotes it to `.live`. Callers that are 471 /// migrating off a previous room disconnect the stale engagement *after* 472 /// this returns, so the old socket's close races against a state that has 473 /// already moved on. 474 private func connect(gameID: UUID, room: EngagementRoomCredentials) async -> EngagementReconcileOutcome { 475 guard let localAuthorID = await localAuthorID(), !localAuthorID.isEmpty else { 476 await log("engagement: connect skipped for \(gameID.uuidString), missing local author") 477 return .reconciled 478 } 479 let engagementID = UUID() 480 states[gameID] = .connecting(engagementID: engagementID, room: room, at: now()) 481 do { 482 try await host.connect( 483 engagementID: engagementID, 484 room: room, 485 authorID: localAuthorID, 486 deviceID: localDeviceID 487 ) 488 await log("engagement: connecting \(gameID.uuidString) to room \(room.roomID.uuidString)") 489 } catch EngagementHostError.roomSecretMismatch { 490 states[gameID] = .idle 491 await log( 492 "engagement: room \(room.roomID.uuidString) rejected for \(gameID.uuidString): " + 493 "registered with a different secret" 494 ) 495 return .roomRejected 496 } catch { 497 states[gameID] = .idle 498 await log("engagement: connect failed for \(gameID.uuidString): \(error.localizedDescription)") 499 } 500 return .reconciled 501 } 502 503 private func state(for gameID: UUID) -> State { 504 states[gameID] ?? .idle 505 } 506 507 private func stateEntry(for engagementID: UUID) -> (UUID, State)? { 508 states.first { _, state in 509 state.engagementID == engagementID 510 } 511 } 512 } 513 514 enum EngagementCoordinatorError: LocalizedError { 515 case invalidPayloadEncoding 516 517 var errorDescription: String? { 518 switch self { 519 case .invalidPayloadEncoding: 520 "Unable to encode engagement room payload." 521 } 522 } 523 }