refactor(model/sync): 重构 CelestiaSession 同步状态判定与 SyncManager 自动后台队列引擎

CelestiaTrace/Models/CelestiaSession.swift

@@ -58,10 +58,27 @@ final class CelestiaSession {
         set { syncStateRaw = newValue.rawValue }
     }
 
+    /// A server-confirmed sync requires both the legacy completion flag and a
+    /// remote identifier. This remains compatible with records created before
+    /// `syncStateRaw` was added, whose migrated state defaults to `localOnly`.
+    var hasConfirmedCloudSync: Bool {
+        isSynced && cloudSessionId != nil
+    }
+
+    var needsCloudSync: Bool {
+        !hasConfirmedCloudSync
+    }
+
+    func markContentModified() {
+        isSynced = false
+        syncState = .pending
+        lastSyncError = nil
+    }
+
     /// The revision represented by the current local content.
     /// Unsynced edits form the next revision based on the last known server revision.
     var localRevision: Int64 {
-        if isSynced && syncState == .synced {
+        if hasConfirmedCloudSync {
             return max(serverRevision, 1)
         }
         return max(serverRevision + 1, 1)

CelestiaTrace/Services/Network/RemoteNetworkService.swift

@@ -230,32 +230,35 @@ final class RemoteNetworkService: NetworkServiceProtocol {
             authenticated: true
         )
 
-        let input = try FileHandle(forReadingFrom: fileURL)
-        defer { try? input.close() }
         for chunkIndex in 0..<upload.totalChunks {
             let offset = UInt64(chunkIndex) * UInt64(upload.chunkSize)
-            try input.seek(toOffset: offset)
-            guard let data = try input.read(upToCount: upload.chunkSize), !data.isEmpty else {
-                throw APIError.transport("读取上传分片失败")
-            }
-            let temporaryURL = FileManager.default.temporaryDirectory
-                .appendingPathComponent("celestia-\(upload.uploadId)-\(chunkIndex).chunk")
-            try data.write(to: temporaryURL, options: .atomic)
-            defer { try? FileManager.default.removeItem(at: temporaryURL) }
+            let preparedChunk = try await prepareUploadChunk(
+                fileURL: fileURL,
+                uploadID: upload.uploadId,
+                chunkIndex: chunkIndex,
+                chunkSize: upload.chunkSize,
+                offset: offset
+            )
 
             let chunkStart = Double(offset) / Double(fileSize)
-            let chunkSpan = Double(data.count) / Double(fileSize)
-            let _: ChunkUploadProgressResponse = try await client.upload(
-                "sessions/\(sessionID)/assets/chunk",
-                fileURL: temporaryURL,
-                fileName: "chunk-\(chunkIndex)",
-                mimeType: "application/octet-stream",
-                fields: [
-                    "uploadId": upload.uploadId,
-                    "chunkIndex": String(chunkIndex)
-                ]
-            ) { fraction in
-                progress(min(max(chunkStart + chunkSpan * fraction, 0), 1))
+            let chunkSpan = Double(preparedChunk.byteCount) / Double(fileSize)
+            do {
+                let _: ChunkUploadProgressResponse = try await client.upload(
+                    "sessions/\(sessionID)/assets/chunk",
+                    fileURL: preparedChunk.url,
+                    fileName: "chunk-\(chunkIndex)",
+                    mimeType: "application/octet-stream",
+                    fields: [
+                        "uploadId": upload.uploadId,
+                        "chunkIndex": String(chunkIndex)
+                    ]
+                ) { fraction in
+                    progress(min(max(chunkStart + chunkSpan * fraction, 0), 1))
+                }
+                try? FileManager.default.removeItem(at: preparedChunk.url)
+            } catch {
+                try? FileManager.default.removeItem(at: preparedChunk.url)
+                throw error
             }
         }
 
@@ -273,6 +276,34 @@ final class RemoteNetworkService: NetworkServiceProtocol {
         return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
     }
 
+    private func prepareUploadChunk(
+        fileURL: URL,
+        uploadID: String,
+        chunkIndex: Int,
+        chunkSize: Int,
+        offset: UInt64
+    ) async throws -> (url: URL, byteCount: Int) {
+        let task = Task.detached(priority: .utility) {
+            try Task.checkCancellation()
+            let input = try FileHandle(forReadingFrom: fileURL)
+            defer { try? input.close() }
+            try input.seek(toOffset: offset)
+            guard let data = try input.read(upToCount: chunkSize), !data.isEmpty else {
+                throw APIError.transport("读取上传分片失败")
+            }
+            try Task.checkCancellation()
+            let temporaryURL = FileManager.default.temporaryDirectory
+                .appendingPathComponent("celestia-\(uploadID)-\(chunkIndex).chunk")
+            try data.write(to: temporaryURL, options: .atomic)
+            return (temporaryURL, data.count)
+        }
+        return try await withTaskCancellationHandler {
+            try await task.value
+        } onCancel: {
+            task.cancel()
+        }
+    }
+
     private func fileSize(of url: URL) throws -> Int64 {
         let values = try url.resourceValues(forKeys: [.fileSizeKey])
         guard let size = values.fileSize else {
@@ -391,6 +422,8 @@ final class SyncManager: ObservableObject {
 
     private let service: NetworkServiceProtocol
     private var activeSyncTask: Task<Bool, Never>?
+    private var automaticSyncTasks: [UUID: Task<Void, Never>] = [:]
+    private var automaticSyncGenerations: [UUID: UUID] = [:]
 
     init(service: NetworkServiceProtocol = RemoteNetworkService()) {
         self.service = service
@@ -412,6 +445,53 @@ final class SyncManager: ObservableObject {
         }
     }
 
+    func scheduleAutomaticSync(
+        for session: CelestiaSession,
+        modelContext: ModelContext,
+        delay: Duration = .seconds(1.5)
+    ) {
+        guard session.endTime != nil,
+              session.syncState != .conflict,
+              AuthManager.shared.currentUser?.id != nil else { return }
+
+        let sessionID = session.id
+        let generation = UUID()
+        automaticSyncGenerations[sessionID] = generation
+        automaticSyncTasks[sessionID]?.cancel()
+        if isSyncing, activeSessionID == sessionID {
+            activeSyncTask?.cancel()
+        }
+        automaticSyncTasks[sessionID] = Task { @MainActor [weak self] in
+            do {
+                try await Task.sleep(for: delay)
+                while let self,
+                      (self.isSyncing || !NetworkStatusMonitor.shared.isConnected) {
+                    try Task.checkCancellation()
+                    try await Task.sleep(for: .seconds(0.75))
+                }
+                guard let self,
+                      !Task.isCancelled,
+                      self.automaticSyncGenerations[sessionID] == generation,
+                      session.needsCloudSync,
+                      let userID = AuthManager.shared.currentUser?.id else { return }
+                _ = await self.sync(
+                    sessions: [session],
+                    modelContext: modelContext,
+                    userID: userID
+                )
+                if self.automaticSyncGenerations[sessionID] == generation {
+                    self.automaticSyncTasks[sessionID] = nil
+                    self.automaticSyncGenerations[sessionID] = nil
+                }
+            } catch {
+                guard let self,
+                      self.automaticSyncGenerations[sessionID] == generation else { return }
+                self.automaticSyncTasks[sessionID] = nil
+                self.automaticSyncGenerations[sessionID] = nil
+            }
+        }
+    }
+
     func sync(
         sessions: [CelestiaSession],
         modelContext: ModelContext,
@@ -452,9 +532,15 @@ final class SyncManager: ObservableObject {
         completedCount = 0
         activeSessionID = nil
         syncProgress = 0
-        totalCount = sessions.filter {
-            ($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
-        }.count
+        let queuedSessions = sessions.filter {
+            ($0.ownerUserID == nil || $0.ownerUserID == userID)
+                && $0.needsCloudSync
+                && $0.syncState != .conflict
+        }
+        totalCount = queuedSessions.count
+        // Give the UI immediate feedback while the remote index is being fetched.
+        // The persisted state is changed only when this session actually begins syncing.
+        activeSessionID = queuedSessions.first?.id
         do {
             let remoteSessions = try await service.fetchSessions()
             try Task.checkCancellation()
@@ -467,7 +553,7 @@ final class SyncManager: ObservableObject {
 
             let eligible = sessions.filter {
                 ($0.ownerUserID == nil || $0.ownerUserID == userID)
-                    && (!$0.isSynced || $0.cloudSessionId == nil)
+                    && $0.needsCloudSync
                     && $0.syncState != .conflict
             }
             var failures: [String] = []
@@ -564,7 +650,7 @@ final class SyncManager: ObservableObject {
             guard let url = AudioPathHelper.resolveURL(for: path) else {
                 throw APIError.transport("本地录音文件不存在：\(path)")
             }
-            let sha256 = try fileSHA256(of: url)
+            let sha256 = try await fileSHA256(of: url)
             let hasMatchingAudio = (remote.assets ?? []).contains {
                 $0.kind.uppercased() == "AUDIO"
                     && $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
@@ -622,15 +708,23 @@ final class SyncManager: ObservableObject {
         return Int64(values?.fileSize ?? 0)
     }
 
-    private func fileSHA256(of url: URL) throws -> String {
-        let input = try FileHandle(forReadingFrom: url)
-        defer { try? input.close() }
-        var hasher = SHA256()
-        while let data = try input.read(upToCount: 1_024 * 1_024),
-              !data.isEmpty {
-            hasher.update(data: data)
+    private func fileSHA256(of url: URL) async throws -> String {
+        let task = Task.detached(priority: .utility) {
+            let input = try FileHandle(forReadingFrom: url)
+            defer { try? input.close() }
+            var hasher = SHA256()
+            while let data = try input.read(upToCount: 1_024 * 1_024),
+                  !data.isEmpty {
+                try Task.checkCancellation()
+                hasher.update(data: data)
+            }
+            return hasher.finalize().map { String(format: "%02x", $0) }.joined()
+        }
+        return try await withTaskCancellationHandler {
+            try await task.value
+        } onCancel: {
+            task.cancel()
         }
-        return hasher.finalize().map { String(format: "%02x", $0) }.joined()
     }
 
     private func downloadMissingAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
@@ -641,7 +735,12 @@ final class SyncManager: ObservableObject {
                 ($0.createdAt ?? .distantPast) < ($1.createdAt ?? .distantPast)
             }) {
             let localURL = AudioPathHelper.resolveURL(for: session.localAudioPath)
-            let localHash = localURL.flatMap { try? fileSHA256(of: $0) }
+            let localHash: String?
+            if let localURL {
+                localHash = try? await fileSHA256(of: localURL)
+            } else {
+                localHash = nil
+            }
             let needsDownload = localURL == nil
                 || (session.isSynced
                     && localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)