|
|
@@ -6,7 +6,7 @@ import CryptoKit
|
|
|
|
|
|
// Remote contract: Docs/RemoteAPI.openapi.yaml
|
|
|
|
|
|
-struct RemoteAsset: Decodable {
|
|
|
+struct RemoteAsset: Decodable, Sendable {
|
|
|
let id: String
|
|
|
let clientId: String
|
|
|
let kind: String
|
|
|
@@ -42,6 +42,20 @@ struct RemoteEvent: Decodable {
|
|
|
let longitude: Double?
|
|
|
}
|
|
|
|
|
|
+struct RemoteSessionIndex: Decodable {
|
|
|
+ let id: String
|
|
|
+ let clientId: String?
|
|
|
+ let title: String
|
|
|
+ let startTime: Date
|
|
|
+ let endTime: Date?
|
|
|
+ let durationMs: Int64
|
|
|
+ let photoCount: Int
|
|
|
+ let noteCount: Int
|
|
|
+ let revision: Int64
|
|
|
+ let deletedAt: Date?
|
|
|
+ let updatedAt: Date
|
|
|
+}
|
|
|
+
|
|
|
struct RemoteSession: Decodable {
|
|
|
let id: String
|
|
|
let clientId: String?
|
|
|
@@ -51,6 +65,7 @@ struct RemoteSession: Decodable {
|
|
|
let durationMs: Int64
|
|
|
let revision: Int64
|
|
|
let deletedAt: Date?
|
|
|
+ let updatedAt: Date
|
|
|
let events: [RemoteEvent]?
|
|
|
let assets: [RemoteAsset]?
|
|
|
}
|
|
|
@@ -138,10 +153,14 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
self.client = client
|
|
|
}
|
|
|
|
|
|
- func fetchSessions() async throws -> [RemoteSession] {
|
|
|
+ func fetchSessionIndex() async throws -> [RemoteSessionIndex] {
|
|
|
try await client.request("sessions?includeDeleted=true", authenticated: true)
|
|
|
}
|
|
|
|
|
|
+ func fetchSession(sessionID: String) async throws -> RemoteSession {
|
|
|
+ try await client.request("sessions/\(sessionID)", authenticated: true)
|
|
|
+ }
|
|
|
+
|
|
|
func syncSession(
|
|
|
_ session: CelestiaSession,
|
|
|
deletedEventClientIDs: [String]
|
|
|
@@ -431,6 +450,7 @@ final class NetworkStatusMonitor: ObservableObject, @unchecked Sendable {
|
|
|
@MainActor
|
|
|
final class SyncManager: ObservableObject {
|
|
|
static let shared = SyncManager()
|
|
|
+ private static let maxConcurrentAssetDownloads = 4
|
|
|
|
|
|
@Published private(set) var isSyncing = false
|
|
|
@Published private(set) var lastSyncDate: Date?
|
|
|
@@ -444,6 +464,33 @@ final class SyncManager: ObservableObject {
|
|
|
private var activeSyncTask: Task<Bool, Never>?
|
|
|
private var automaticSyncTasks: [UUID: Task<Void, Never>] = [:]
|
|
|
private var automaticSyncGenerations: [UUID: UUID] = [:]
|
|
|
+ private var queuedSyncRequest: SyncRequest?
|
|
|
+ private static let diagnosticsByteFormatter: ByteCountFormatter = {
|
|
|
+ let formatter = ByteCountFormatter()
|
|
|
+ formatter.countStyle = .file
|
|
|
+ formatter.allowedUnits = [.useKB, .useMB, .useGB]
|
|
|
+ formatter.isAdaptive = true
|
|
|
+ return formatter
|
|
|
+ }()
|
|
|
+
|
|
|
+ private struct SyncRequest {
|
|
|
+ let sessions: [CelestiaSession]
|
|
|
+ let modelContext: ModelContext
|
|
|
+ let userID: String
|
|
|
+ }
|
|
|
+
|
|
|
+ private enum AssetDownloadTarget: Sendable {
|
|
|
+ case audio
|
|
|
+ case photo(eventID: UUID)
|
|
|
+ }
|
|
|
+
|
|
|
+ private struct AssetDownloadJob: Sendable {
|
|
|
+ let sessionID: String
|
|
|
+ let asset: RemoteAsset
|
|
|
+ let destination: URL
|
|
|
+ let target: AssetDownloadTarget
|
|
|
+ let displayName: String
|
|
|
+ }
|
|
|
|
|
|
init(service: NetworkServiceProtocol = RemoteNetworkService()) {
|
|
|
self.service = service
|
|
|
@@ -518,10 +565,17 @@ final class SyncManager: ObservableObject {
|
|
|
userID: String
|
|
|
) async -> Bool {
|
|
|
guard !isSyncing else {
|
|
|
- DeveloperLogStore.log("现场记录同步", "已有同步任务正在运行,忽略重复请求", level: .warning)
|
|
|
+ queuedSyncRequest = SyncRequest(
|
|
|
+ sessions: sessions,
|
|
|
+ modelContext: modelContext,
|
|
|
+ userID: userID
|
|
|
+ )
|
|
|
+ DeveloperLogStore.log("现场记录同步", "已有同步任务正在运行,已排队再次检查时间戳")
|
|
|
return false
|
|
|
}
|
|
|
DeveloperLogStore.log("现场记录同步", "开始同步,本地共 \(sessions.count) 条记录")
|
|
|
+ let syncStartedAt = diagnosticsNow
|
|
|
+ diagnosticsLog("整轮开始:本地记录=\(sessions.count)")
|
|
|
isSyncing = true
|
|
|
let task = Task { @MainActor [self] in
|
|
|
await performSync(
|
|
|
@@ -544,11 +598,27 @@ final class SyncManager: ObservableObject {
|
|
|
result ? "同步任务完成" : "同步任务未完成",
|
|
|
level: result ? .success : .warning
|
|
|
)
|
|
|
+ diagnosticsLog(
|
|
|
+ "整轮\(result ? "完成" : "未完成"):总耗时 \(diagnosticsElapsed(since: syncStartedAt))",
|
|
|
+ level: result ? .success : .warning
|
|
|
+ )
|
|
|
+ if let queuedRequest = queuedSyncRequest {
|
|
|
+ queuedSyncRequest = nil
|
|
|
+ Task { @MainActor [weak self] in
|
|
|
+ guard AuthManager.shared.currentUser?.id == queuedRequest.userID else { return }
|
|
|
+ _ = await self?.sync(
|
|
|
+ sessions: queuedRequest.sessions,
|
|
|
+ modelContext: queuedRequest.modelContext,
|
|
|
+ userID: queuedRequest.userID
|
|
|
+ )
|
|
|
+ }
|
|
|
+ }
|
|
|
return result
|
|
|
}
|
|
|
|
|
|
func pauseSync(sessionID: UUID) {
|
|
|
guard isSyncing, activeSessionID == sessionID else { return }
|
|
|
+ queuedSyncRequest = nil
|
|
|
activeSyncTask?.cancel()
|
|
|
}
|
|
|
|
|
|
@@ -571,35 +641,144 @@ final class SyncManager: ObservableObject {
|
|
|
// 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
|
|
|
+ var pendingRemoteDownloads: [CelestiaSession] = []
|
|
|
do {
|
|
|
- let remoteSessions = try await service.fetchSessions()
|
|
|
- DeveloperLogStore.log("现场记录同步", "获取到 \(remoteSessions.count) 条云端记录", level: .success)
|
|
|
+ let indexStartedAt = diagnosticsNow
|
|
|
+ diagnosticsLog("阶段 1/5 开始:请求服务器时间戳索引", level: .receive)
|
|
|
+ let remoteIndex = try await service.fetchSessionIndex()
|
|
|
+ diagnosticsLog(
|
|
|
+ "阶段 1/5 完成:云端记录=\(remoteIndex.count),耗时 \(diagnosticsElapsed(since: indexStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
+ DeveloperLogStore.log("现场记录同步", "获取到 \(remoteIndex.count) 条云端时间戳", level: .success)
|
|
|
try Task.checkCancellation()
|
|
|
- // Detail and post-recording syncs may upload only one session, while
|
|
|
- // fetchSessions always returns the complete cloud history. Use the
|
|
|
- // complete local store as the merge baseline to avoid inserting the
|
|
|
- // other cloud sessions again with duplicate client IDs.
|
|
|
+ let mergeStartedAt = diagnosticsNow
|
|
|
let allLocalSessions = try modelContext.fetch(FetchDescriptor<CelestiaSession>())
|
|
|
- let mergedSessions = try await merge(
|
|
|
- remoteSessions,
|
|
|
+ let indexedSessions = try await mergeIndex(
|
|
|
+ remoteIndex,
|
|
|
into: allLocalSessions,
|
|
|
modelContext: modelContext,
|
|
|
userID: userID
|
|
|
)
|
|
|
- for (remote, local) in mergedSessions {
|
|
|
- try await downloadMissingAssets(for: local, remote: remote)
|
|
|
- try Task.checkCancellation()
|
|
|
- await Task.yield()
|
|
|
+ diagnosticsLog(
|
|
|
+ "阶段 2/5 完成:读取并合并本地记录=\(allLocalSessions.count),索引匹配=\(indexedSessions.count),耗时 \(diagnosticsElapsed(since: mergeStartedAt))"
|
|
|
+ )
|
|
|
+ var failures: [String] = []
|
|
|
+ var remoteDetailsByID: [String: RemoteSession] = [:]
|
|
|
+
|
|
|
+ // mergeIndex has already persisted list metadata. Only timestamp
|
|
|
+ // changes proceed to the second phase that fetches concrete content.
|
|
|
+ let sessionsRequiringDetail = indexedSessions.filter { $0.needsDetail }
|
|
|
+ pendingRemoteDownloads = sessionsRequiringDetail.map(\.local)
|
|
|
+ if let firstDownload = sessionsRequiringDetail.first?.local {
|
|
|
+ activeSessionID = firstDownload.id
|
|
|
+ }
|
|
|
+ await Task.yield()
|
|
|
+
|
|
|
+ let detailStageStartedAt = diagnosticsNow
|
|
|
+ diagnosticsLog("阶段 3/5 开始:需要拉取详情=\(sessionsRequiringDetail.count)")
|
|
|
+ for (detailIndex, match) in sessionsRequiringDetail.enumerated() {
|
|
|
+ let local = match.local
|
|
|
+ let recordLabel = diagnosticsRecordLabel(
|
|
|
+ local,
|
|
|
+ position: detailIndex + 1,
|
|
|
+ total: sessionsRequiringDetail.count
|
|
|
+ )
|
|
|
+ let recordStartedAt = diagnosticsNow
|
|
|
+ do {
|
|
|
+ activeSessionID = local.id
|
|
|
+ let detailStartedAt = diagnosticsNow
|
|
|
+ let remote = try await service.fetchSession(sessionID: match.index.id)
|
|
|
+ let remoteAssets = remote.assets ?? []
|
|
|
+ let remoteBytes = remoteAssets.reduce(Int64(0)) { $0 + $1.sizeBytes }
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 详情响应:事件=\(remote.events?.count ?? 0),资源=\(remoteAssets.count)(\(diagnosticsBytes(remoteBytes))),耗时 \(diagnosticsElapsed(since: detailStartedAt))",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
+ remoteDetailsByID[remote.id] = remote
|
|
|
+ let applyStartedAt = diagnosticsNow
|
|
|
+ await apply(
|
|
|
+ remote,
|
|
|
+ to: local,
|
|
|
+ userID: userID,
|
|
|
+ syncState: .syncing
|
|
|
+ )
|
|
|
+ try modelContext.save()
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 详情写入本地:耗时 \(diagnosticsElapsed(since: applyStartedAt))"
|
|
|
+ )
|
|
|
+ await Task.yield()
|
|
|
+
|
|
|
+ let assetStartedAt = diagnosticsNow
|
|
|
+ let downloadedAssets = try await downloadMissingAssets(
|
|
|
+ for: local,
|
|
|
+ remote: remote
|
|
|
+ ) {
|
|
|
+ guard local.hasConfirmedCloudSync else { return }
|
|
|
+ activeSessionID = local.id
|
|
|
+ local.syncState = .syncing
|
|
|
+ local.lastSyncError = nil
|
|
|
+ try? modelContext.save()
|
|
|
+ }
|
|
|
+ try Task.checkCancellation()
|
|
|
+ local.serverUpdatedAt = remote.updatedAt
|
|
|
+ local.syncState = .synced
|
|
|
+ local.lastSyncError = nil
|
|
|
+ local.lastSyncedAt = Date()
|
|
|
+ let finalSaveStartedAt = diagnosticsNow
|
|
|
+ try modelContext.save()
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 资源阶段:\(downloadedAssets ? "发生下载" : "无需下载"),耗时 \(diagnosticsElapsed(since: assetStartedAt));最终保存 \(diagnosticsElapsed(since: finalSaveStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 云端拉取完成:总耗时 \(diagnosticsElapsed(since: recordStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
+ await Task.yield()
|
|
|
+ } catch where Task.isCancelled {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 云端拉取取消:已运行 \(diagnosticsElapsed(since: recordStartedAt))",
|
|
|
+ level: .warning
|
|
|
+ )
|
|
|
+ if local.hasConfirmedCloudSync,
|
|
|
+ local.syncState == .syncing {
|
|
|
+ local.syncState = .failed
|
|
|
+ local.lastSyncError = "已暂停从云端下载"
|
|
|
+ try? modelContext.save()
|
|
|
+ }
|
|
|
+ throw CancellationError()
|
|
|
+ } catch {
|
|
|
+ if local.hasConfirmedCloudSync {
|
|
|
+ local.syncState = .failed
|
|
|
+ local.lastSyncError = "从云端下载失败:\(error.localizedDescription)"
|
|
|
+ try? modelContext.save()
|
|
|
+ }
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 云端拉取失败:耗时 \(diagnosticsElapsed(since: recordStartedAt));\(error.localizedDescription)",
|
|
|
+ level: .error
|
|
|
+ )
|
|
|
+ failures.append("\(local.title):\(error.localizedDescription)")
|
|
|
+ DeveloperLogStore.log(
|
|
|
+ "现场记录同步",
|
|
|
+ "\(local.title) 从云端下载失败:\(error.localizedDescription)",
|
|
|
+ level: .error
|
|
|
+ )
|
|
|
+ }
|
|
|
}
|
|
|
try modelContext.save()
|
|
|
+ diagnosticsLog(
|
|
|
+ "阶段 3/5 完成:详情记录=\(sessionsRequiringDetail.count),总耗时 \(diagnosticsElapsed(since: detailStageStartedAt))"
|
|
|
+ )
|
|
|
|
|
|
let eligible = deduplicatedSessions(sessions.filter {
|
|
|
($0.ownerUserID == nil || $0.ownerUserID == userID)
|
|
|
&& $0.needsCloudSync
|
|
|
&& $0.syncState != .conflict
|
|
|
})
|
|
|
- var failures: [String] = []
|
|
|
let eligibleCount = max(eligible.count, 1)
|
|
|
+ let uploadStageStartedAt = diagnosticsNow
|
|
|
+ diagnosticsLog("阶段 4/5 开始:需要上传=\(eligible.count)")
|
|
|
for (index, session) in eligible.enumerated() {
|
|
|
try Task.checkCancellation()
|
|
|
await Task.yield()
|
|
|
@@ -614,23 +793,46 @@ final class SyncManager: ObservableObject {
|
|
|
"现场记录同步",
|
|
|
"正在同步 \(index + 1)/\(eligible.count):\(session.title)"
|
|
|
)
|
|
|
+ let recordLabel = diagnosticsRecordLabel(
|
|
|
+ session,
|
|
|
+ position: index + 1,
|
|
|
+ total: eligible.count
|
|
|
+ )
|
|
|
+ let recordStartedAt = diagnosticsNow
|
|
|
do {
|
|
|
- let previousRemote = remoteSessions.first {
|
|
|
- $0.id == session.cloudSessionId
|
|
|
- || $0.clientId?.caseInsensitiveCompare(session.id.uuidString) == .orderedSame
|
|
|
+ var previousRemote: RemoteSession?
|
|
|
+ if let cloudID = session.cloudSessionId {
|
|
|
+ if let cached = remoteDetailsByID[cloudID] {
|
|
|
+ previousRemote = cached
|
|
|
+ } else if remoteIndex.contains(where: { $0.id == cloudID }) {
|
|
|
+ let previousDetailStartedAt = diagnosticsNow
|
|
|
+ let fetched = try await service.fetchSession(sessionID: cloudID)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传前详情:耗时 \(diagnosticsElapsed(since: previousDetailStartedAt))",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
+ remoteDetailsByID[cloudID] = fetched
|
|
|
+ previousRemote = fetched
|
|
|
+ }
|
|
|
}
|
|
|
let localEventIDs = Set(session.events.map { $0.id.uuidString.lowercased() })
|
|
|
let deletedEventClientIDs = previousRemote?.events?
|
|
|
.compactMap(\.clientId)
|
|
|
.filter { !localEventIDs.contains($0.lowercased()) } ?? []
|
|
|
+ let metadataStartedAt = diagnosticsNow
|
|
|
let remote = try await service.syncSession(
|
|
|
session,
|
|
|
deletedEventClientIDs: deletedEventClientIDs
|
|
|
)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 元数据上传:耗时 \(diagnosticsElapsed(since: metadataStartedAt))",
|
|
|
+ level: .transmit
|
|
|
+ )
|
|
|
try Task.checkCancellation()
|
|
|
syncProgress = sessionStartProgress + sessionProgressSpan * 0.12
|
|
|
session.cloudSessionId = remote.id
|
|
|
session.serverRevision = remote.revision
|
|
|
+ session.serverUpdatedAt = remote.updatedAt
|
|
|
try await uploadLocalAssets(for: session, remote: remote) { [weak self] assetProgress in
|
|
|
Task { @MainActor [weak self] in
|
|
|
self?.syncProgress = sessionStartProgress
|
|
|
@@ -638,6 +840,15 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
}
|
|
|
try Task.checkCancellation()
|
|
|
+ let finalDetailStartedAt = diagnosticsNow
|
|
|
+ let finalRemote = try await service.fetchSession(sessionID: remote.id)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传后详情确认:耗时 \(diagnosticsElapsed(since: finalDetailStartedAt))",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
+ remoteDetailsByID[finalRemote.id] = finalRemote
|
|
|
+ session.serverRevision = finalRemote.revision
|
|
|
+ session.serverUpdatedAt = finalRemote.updatedAt
|
|
|
session.isSynced = true
|
|
|
session.syncState = .synced
|
|
|
session.lastSyncedAt = Date()
|
|
|
@@ -648,7 +859,15 @@ final class SyncManager: ObservableObject {
|
|
|
"\(session.title) 同步成功,云端版本 v\(session.serverRevision)",
|
|
|
level: .success
|
|
|
)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传链路完成:总耗时 \(diagnosticsElapsed(since: recordStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
} catch where Task.isCancelled {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传链路取消:已运行 \(diagnosticsElapsed(since: recordStartedAt))",
|
|
|
+ level: .warning
|
|
|
+ )
|
|
|
session.isSynced = false
|
|
|
session.syncState = .pending
|
|
|
session.lastSyncError = nil
|
|
|
@@ -656,6 +875,10 @@ final class SyncManager: ObservableObject {
|
|
|
try? modelContext.save()
|
|
|
return false
|
|
|
} catch APIError.server(let code, _) where code == 409 {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传链路冲突:耗时 \(diagnosticsElapsed(since: recordStartedAt))",
|
|
|
+ level: .error
|
|
|
+ )
|
|
|
session.isSynced = false
|
|
|
session.syncState = .conflict
|
|
|
session.lastSyncError = "本地版本 \(session.serverRevision) 与云端版本不一致"
|
|
|
@@ -665,6 +888,10 @@ final class SyncManager: ObservableObject {
|
|
|
level: .error
|
|
|
)
|
|
|
} catch {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 上传链路失败:耗时 \(diagnosticsElapsed(since: recordStartedAt));\(error.localizedDescription)",
|
|
|
+ level: .error
|
|
|
+ )
|
|
|
session.isSynced = false
|
|
|
session.syncState = .failed
|
|
|
session.lastSyncError = error.localizedDescription
|
|
|
@@ -677,6 +904,9 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
try modelContext.save()
|
|
|
}
|
|
|
+ diagnosticsLog(
|
|
|
+ "阶段 4/5 完成:上传记录=\(eligible.count),总耗时 \(diagnosticsElapsed(since: uploadStageStartedAt))"
|
|
|
+ )
|
|
|
let conflicts = sessions.filter { $0.syncState == .conflict }
|
|
|
if !conflicts.isEmpty {
|
|
|
failures.append(contentsOf: conflicts.map {
|
|
|
@@ -688,16 +918,31 @@ final class SyncManager: ObservableObject {
|
|
|
DeveloperLogStore.log("现场记录同步", "同步结束,存在 \(failures.count) 个问题", level: .error)
|
|
|
return false
|
|
|
}
|
|
|
+ let checkpointStartedAt = diagnosticsNow
|
|
|
try await service.recordSyncCheckpoint()
|
|
|
+ diagnosticsLog(
|
|
|
+ "阶段 5/5 完成:提交同步检查点,耗时 \(diagnosticsElapsed(since: checkpointStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
let now = Date()
|
|
|
lastSyncDate = now
|
|
|
UserDefaults.standard.set(now, forKey: "com.celestia.trace.last_server_sync")
|
|
|
DeveloperLogStore.log("现场记录同步", "云端检查点已更新", level: .success)
|
|
|
return true
|
|
|
} catch where Task.isCancelled {
|
|
|
+ for session in pendingRemoteDownloads where session.syncState == .syncing {
|
|
|
+ session.syncState = .failed
|
|
|
+ session.lastSyncError = "已暂停从云端下载"
|
|
|
+ }
|
|
|
+ try? modelContext.save()
|
|
|
syncProgress = 0
|
|
|
return false
|
|
|
} catch {
|
|
|
+ for session in pendingRemoteDownloads where session.syncState == .syncing {
|
|
|
+ session.syncState = .failed
|
|
|
+ session.lastSyncError = "从云端下载中断:\(error.localizedDescription)"
|
|
|
+ }
|
|
|
+ try? modelContext.save()
|
|
|
lastErrorMessage = error.localizedDescription
|
|
|
DeveloperLogStore.log("现场记录同步", "同步中断:\(error.localizedDescription)", level: .error)
|
|
|
return false
|
|
|
@@ -710,13 +955,18 @@ final class SyncManager: ObservableObject {
|
|
|
progress: @escaping @Sendable (Double) -> Void
|
|
|
) async throws {
|
|
|
guard let cloudID = session.cloudSessionId else { throw APIError.missingData }
|
|
|
+ let stageStartedAt = diagnosticsNow
|
|
|
+ let recordLabel = diagnosticsRecordLabel(session)
|
|
|
let existingIDs = Set((remote.assets ?? []).map(\.clientId))
|
|
|
var uploads: [(clientID: String, kind: String, url: URL, size: Int64)] = []
|
|
|
if let path = session.localAudioPath {
|
|
|
guard let url = AudioPathHelper.resolveURL(for: path) else {
|
|
|
throw APIError.transport("本地录音文件不存在:\(path)")
|
|
|
}
|
|
|
- let sha256 = try await fileSHA256(of: url)
|
|
|
+ let sha256 = try await fileSHA256(
|
|
|
+ of: url,
|
|
|
+ purpose: "\(recordLabel) 音频上传校验"
|
|
|
+ )
|
|
|
let hasMatchingAudio = (remote.assets ?? []).contains {
|
|
|
$0.kind.uppercased() == "AUDIO"
|
|
|
&& $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
|
|
|
@@ -740,7 +990,12 @@ final class SyncManager: ObservableObject {
|
|
|
let totalBytes = max(uploads.reduce(Int64(0)) { $0 + $1.size }, 1)
|
|
|
let requiredBytes = uploads.reduce(Int64(0)) { $0 + $1.size }
|
|
|
if requiredBytes > 0 {
|
|
|
+ let quotaStartedAt = diagnosticsNow
|
|
|
let quota = try await service.fetchStorageQuota()
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 存储配额查询:耗时 \(diagnosticsElapsed(since: quotaStartedAt))",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
guard requiredBytes <= quota.remainingBytes else {
|
|
|
throw APIError.storageQuotaExceeded(
|
|
|
requiredBytes: requiredBytes,
|
|
|
@@ -750,8 +1005,19 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
var completedBytes: Int64 = 0
|
|
|
progress(uploads.isEmpty ? 1 : 0)
|
|
|
- for upload in uploads {
|
|
|
+ if uploads.isEmpty {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 本地资源上传:无需上传,检查耗时 \(diagnosticsElapsed(since: stageStartedAt))"
|
|
|
+ )
|
|
|
+ }
|
|
|
+ for (uploadIndex, upload) in uploads.enumerated() {
|
|
|
let bytesBeforeUpload = completedBytes
|
|
|
+ let uploadStartedAt = diagnosticsNow
|
|
|
+ let assetKind = upload.kind.uppercased() == "AUDIO" ? "音频" : "照片"
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) \(assetKind) \(uploadIndex + 1)/\(uploads.count) 上传开始:\(diagnosticsBytes(upload.size))",
|
|
|
+ level: .transmit
|
|
|
+ )
|
|
|
let result = try await service.uploadAsset(
|
|
|
sessionID: cloudID,
|
|
|
clientID: upload.clientID,
|
|
|
@@ -766,6 +1032,17 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
completedBytes += upload.size
|
|
|
progress(Double(completedBytes) / Double(totalBytes))
|
|
|
+ let uploadDuration = diagnosticsDuration(since: uploadStartedAt)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) \(assetKind) \(uploadIndex + 1)/\(uploads.count) 上传完成:耗时 \(diagnosticsDurationText(uploadDuration)),平均 \(diagnosticsRate(bytes: upload.size, duration: uploadDuration))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
+ }
|
|
|
+ if !uploads.isEmpty {
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 本地资源上传完成:文件=\(uploads.count),总大小=\(diagnosticsBytes(requiredBytes)),总耗时 \(diagnosticsElapsed(since: stageStartedAt))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -774,7 +1051,10 @@ final class SyncManager: ObservableObject {
|
|
|
return Int64(values?.fileSize ?? 0)
|
|
|
}
|
|
|
|
|
|
- private func fileSHA256(of url: URL) async throws -> String {
|
|
|
+ private func fileSHA256(of url: URL, purpose: String) async throws -> String {
|
|
|
+ let sizeBytes = fileSize(of: url)
|
|
|
+ let startedAt = diagnosticsNow
|
|
|
+ diagnosticsLog("\(purpose) 开始:\(diagnosticsBytes(sizeBytes))")
|
|
|
let task = Task.detached(priority: .utility) {
|
|
|
let input = try FileHandle(forReadingFrom: url)
|
|
|
defer { try? input.close() }
|
|
|
@@ -786,15 +1066,24 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
return hasher.finalize().map { String(format: "%02x", $0) }.joined()
|
|
|
}
|
|
|
- return try await withTaskCancellationHandler {
|
|
|
+ let digest = try await withTaskCancellationHandler {
|
|
|
try await task.value
|
|
|
} onCancel: {
|
|
|
task.cancel()
|
|
|
}
|
|
|
+ diagnosticsLog("\(purpose) 完成:耗时 \(diagnosticsElapsed(since: startedAt))")
|
|
|
+ return digest
|
|
|
}
|
|
|
|
|
|
- private func downloadMissingAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
|
|
|
+ private func downloadMissingAssets(
|
|
|
+ for session: CelestiaSession,
|
|
|
+ remote: RemoteSession,
|
|
|
+ onDownloadStarted: () -> Void
|
|
|
+ ) async throws -> Bool {
|
|
|
let assets = remote.assets ?? []
|
|
|
+ var jobs: [AssetDownloadJob] = []
|
|
|
+ var downloadedAssets = false
|
|
|
+ let recordLabel = diagnosticsRecordLabel(session)
|
|
|
if let audio = assets
|
|
|
.filter({ $0.kind.uppercased() == "AUDIO" })
|
|
|
.max(by: {
|
|
|
@@ -803,7 +1092,10 @@ final class SyncManager: ObservableObject {
|
|
|
let localURL = AudioPathHelper.resolveURL(for: session.localAudioPath)
|
|
|
let localHash: String?
|
|
|
if let localURL {
|
|
|
- localHash = try? await fileSHA256(of: localURL)
|
|
|
+ localHash = try? await fileSHA256(
|
|
|
+ of: localURL,
|
|
|
+ purpose: "\(recordLabel) 音频本地校验"
|
|
|
+ )
|
|
|
} else {
|
|
|
localHash = nil
|
|
|
}
|
|
|
@@ -812,8 +1104,15 @@ final class SyncManager: ObservableObject {
|
|
|
&& localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)
|
|
|
if needsDownload {
|
|
|
let destination = downloadDestination(for: audio)
|
|
|
- try await service.downloadAsset(sessionID: remote.id, asset: audio, destinationURL: destination)
|
|
|
- session.localAudioPath = AudioPathHelper.relativePath(from: destination.path)
|
|
|
+ jobs.append(
|
|
|
+ AssetDownloadJob(
|
|
|
+ sessionID: remote.id,
|
|
|
+ asset: audio,
|
|
|
+ destination: destination,
|
|
|
+ target: .audio,
|
|
|
+ displayName: "音频"
|
|
|
+ )
|
|
|
+ )
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -840,9 +1139,141 @@ final class SyncManager: ObservableObject {
|
|
|
let key = "\(event.id.uuidString)-photo".lowercased()
|
|
|
guard let asset = photoAssets[key] else { continue }
|
|
|
let destination = downloadDestination(for: asset)
|
|
|
- try await service.downloadAsset(sessionID: remote.id, asset: asset, destinationURL: destination)
|
|
|
- event.localFilePath = AudioPathHelper.relativePath(from: destination.path)
|
|
|
+ jobs.append(
|
|
|
+ AssetDownloadJob(
|
|
|
+ sessionID: remote.id,
|
|
|
+ asset: asset,
|
|
|
+ destination: destination,
|
|
|
+ target: .photo(eventID: event.id),
|
|
|
+ displayName: "照片"
|
|
|
+ )
|
|
|
+ )
|
|
|
+ }
|
|
|
+
|
|
|
+ guard !jobs.isEmpty else { return false }
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) 资源并发下载:文件=\(jobs.count),并发上限=\(Self.maxConcurrentAssetDownloads)",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
+
|
|
|
+ for batchStart in stride(
|
|
|
+ from: 0,
|
|
|
+ to: jobs.count,
|
|
|
+ by: Self.maxConcurrentAssetDownloads
|
|
|
+ ) {
|
|
|
+ try Task.checkCancellation()
|
|
|
+ let batchEnd = min(batchStart + Self.maxConcurrentAssetDownloads, jobs.count)
|
|
|
+ let batch = Array(jobs[batchStart..<batchEnd])
|
|
|
+ batch.forEach { _ in onDownloadStarted() }
|
|
|
+
|
|
|
+ let tasks = batch.map { job in
|
|
|
+ Task { @MainActor [self] in
|
|
|
+ try await downloadAssetJob(job, recordLabel: recordLabel)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ do {
|
|
|
+ try await withTaskCancellationHandler {
|
|
|
+ for task in tasks {
|
|
|
+ let completedJob = try await task.value
|
|
|
+ applyDownloadedAsset(completedJob, to: session)
|
|
|
+ downloadedAssets = true
|
|
|
+ }
|
|
|
+ } onCancel: {
|
|
|
+ tasks.forEach { $0.cancel() }
|
|
|
+ }
|
|
|
+ } catch {
|
|
|
+ tasks.forEach { $0.cancel() }
|
|
|
+ throw error
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return downloadedAssets
|
|
|
+ }
|
|
|
+
|
|
|
+ private func downloadAssetJob(
|
|
|
+ _ job: AssetDownloadJob,
|
|
|
+ recordLabel: String
|
|
|
+ ) async throws -> AssetDownloadJob {
|
|
|
+ try Task.checkCancellation()
|
|
|
+ let downloadStartedAt = diagnosticsNow
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) \(job.displayName)下载开始:\(diagnosticsBytes(job.asset.sizeBytes))",
|
|
|
+ level: .receive
|
|
|
+ )
|
|
|
+ try await service.downloadAsset(
|
|
|
+ sessionID: job.sessionID,
|
|
|
+ asset: job.asset,
|
|
|
+ destinationURL: job.destination
|
|
|
+ )
|
|
|
+ try Task.checkCancellation()
|
|
|
+ let downloadDuration = diagnosticsDuration(since: downloadStartedAt)
|
|
|
+ diagnosticsLog(
|
|
|
+ "\(recordLabel) \(job.displayName)下载完成:耗时 \(diagnosticsDurationText(downloadDuration)),平均 \(diagnosticsRate(bytes: job.asset.sizeBytes, duration: downloadDuration))",
|
|
|
+ level: .success
|
|
|
+ )
|
|
|
+ return job
|
|
|
+ }
|
|
|
+
|
|
|
+ private func applyDownloadedAsset(
|
|
|
+ _ job: AssetDownloadJob,
|
|
|
+ to session: CelestiaSession
|
|
|
+ ) {
|
|
|
+ let relativePath = AudioPathHelper.relativePath(from: job.destination.path)
|
|
|
+ switch job.target {
|
|
|
+ case .audio:
|
|
|
+ session.localAudioPath = relativePath
|
|
|
+ Task.detached(priority: .utility) {
|
|
|
+ await SilenceDetector.warmCache(for: job.destination)
|
|
|
+ }
|
|
|
+ case .photo(let eventID):
|
|
|
+ session.events.first(where: { $0.id == eventID })?.localFilePath = relativePath
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private var diagnosticsNow: TimeInterval {
|
|
|
+ ProcessInfo.processInfo.systemUptime
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsDuration(since startedAt: TimeInterval) -> TimeInterval {
|
|
|
+ max(0, diagnosticsNow - startedAt)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsElapsed(since startedAt: TimeInterval) -> String {
|
|
|
+ diagnosticsDurationText(diagnosticsDuration(since: startedAt))
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsDurationText(_ duration: TimeInterval) -> String {
|
|
|
+ if duration < 1 {
|
|
|
+ return "\(Int((duration * 1_000).rounded())) 毫秒"
|
|
|
+ }
|
|
|
+ return String(format: "%.2f 秒", duration)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsBytes(_ bytes: Int64) -> String {
|
|
|
+ Self.diagnosticsByteFormatter.string(fromByteCount: max(0, bytes))
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsRate(bytes: Int64, duration: TimeInterval) -> String {
|
|
|
+ guard duration > 0 else { return "—" }
|
|
|
+ return "\(diagnosticsBytes(Int64(Double(bytes) / duration)))/秒"
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsRecordLabel(
|
|
|
+ _ session: CelestiaSession,
|
|
|
+ position: Int? = nil,
|
|
|
+ total: Int? = nil
|
|
|
+ ) -> String {
|
|
|
+ let shortID = session.id.uuidString.lowercased().prefix(8)
|
|
|
+ if let position, let total {
|
|
|
+ return "记录 \(position)/\(total) [\(shortID)]"
|
|
|
}
|
|
|
+ return "记录 [\(shortID)]"
|
|
|
+ }
|
|
|
+
|
|
|
+ private func diagnosticsLog(
|
|
|
+ _ message: String,
|
|
|
+ level: DeveloperLogEntry.Level = .info
|
|
|
+ ) {
|
|
|
+ DeveloperLogStore.log("同步性能", message, level: level)
|
|
|
}
|
|
|
|
|
|
private func downloadDestination(for asset: RemoteAsset) -> URL {
|
|
|
@@ -852,20 +1283,35 @@ final class SyncManager: ObservableObject {
|
|
|
return documents.appendingPathComponent("cloud_\(asset.id)\(suffix)")
|
|
|
}
|
|
|
|
|
|
- private func merge(
|
|
|
- _ remoteSessions: [RemoteSession],
|
|
|
+ private struct IndexedSessionMatch {
|
|
|
+ let index: RemoteSessionIndex
|
|
|
+ let local: CelestiaSession
|
|
|
+ let needsDetail: Bool
|
|
|
+ }
|
|
|
+
|
|
|
+ private func mergeIndex(
|
|
|
+ _ remoteSessions: [RemoteSessionIndex],
|
|
|
into localSessions: [CelestiaSession],
|
|
|
modelContext: ModelContext,
|
|
|
userID: String
|
|
|
- ) async throws -> [(RemoteSession, CelestiaSession)] {
|
|
|
+ ) async throws -> [IndexedSessionMatch] {
|
|
|
+ let scopedLocalSessions = localSessions.filter {
|
|
|
+ $0.ownerUserID == nil || $0.ownerUserID == userID
|
|
|
+ }
|
|
|
var byClientID = Dictionary(
|
|
|
- localSessions.map { ($0.id.uuidString.lowercased(), $0) },
|
|
|
+ scopedLocalSessions.map { ($0.id.uuidString.lowercased(), $0) },
|
|
|
uniquingKeysWith: preferredLocalSession
|
|
|
)
|
|
|
- if byClientID.count < localSessions.count {
|
|
|
+ var byCloudID = Dictionary(
|
|
|
+ scopedLocalSessions.compactMap { session in
|
|
|
+ session.cloudSessionId.map { ($0, session) }
|
|
|
+ },
|
|
|
+ uniquingKeysWith: preferredLocalSession
|
|
|
+ )
|
|
|
+ if byClientID.count < scopedLocalSessions.count {
|
|
|
DeveloperLogStore.log(
|
|
|
"现场记录同步",
|
|
|
- "检测到 \(localSessions.count - byClientID.count) 条重复本地记录,已选择云端版本较新的记录继续同步",
|
|
|
+ "检测到 \(scopedLocalSessions.count - byClientID.count) 条重复本地记录,已选择云端版本较新的记录继续同步",
|
|
|
level: .warning
|
|
|
)
|
|
|
}
|
|
|
@@ -874,13 +1320,14 @@ final class SyncManager: ObservableObject {
|
|
|
let activeRemoteClientIDs = Set(activeRemoteSessions.compactMap { $0.clientId?.lowercased() })
|
|
|
var deletedLocalObjects: Set<ObjectIdentifier> = []
|
|
|
var deletedFileURLs: Set<URL> = []
|
|
|
- var merged: [(RemoteSession, CelestiaSession)] = []
|
|
|
+ var merged: [IndexedSessionMatch] = []
|
|
|
for (index, remote) in remoteSessions.enumerated() {
|
|
|
if index.isMultiple(of: 20) {
|
|
|
await Task.yield()
|
|
|
}
|
|
|
- guard let clientID = remote.clientId?.lowercased() else { continue }
|
|
|
- if let local = byClientID[clientID] {
|
|
|
+ let clientID = remote.clientId?.lowercased()
|
|
|
+ let existingLocal = clientID.flatMap { byClientID[$0] } ?? byCloudID[remote.id]
|
|
|
+ if let local = existingLocal {
|
|
|
if remote.deletedAt != nil, local.hasConfirmedCloudSync {
|
|
|
deletedFileURLs.formUnion(localFileURLs(for: local))
|
|
|
deletedLocalObjects.insert(ObjectIdentifier(local))
|
|
|
@@ -897,19 +1344,50 @@ final class SyncManager: ObservableObject {
|
|
|
local.lastSyncError = "本地版本 \(local.serverRevision) 与云端版本 \(remote.revision) 不一致"
|
|
|
continue
|
|
|
}
|
|
|
- if local.isSynced, remote.revision > local.serverRevision {
|
|
|
- await apply(remote, to: local, userID: userID)
|
|
|
- }
|
|
|
if remote.deletedAt == nil {
|
|
|
- merged.append((remote, local))
|
|
|
+ let needsDetail =
|
|
|
+ local.isSynced
|
|
|
+ && (local.serverUpdatedAt.map { remote.updatedAt > $0 } ?? true
|
|
|
+ || remote.revision > local.serverRevision
|
|
|
+ || local.syncState == .failed
|
|
|
+ || local.syncState == .syncing)
|
|
|
+ local.cloudSessionId = remote.id
|
|
|
+ local.ownerUserID = userID
|
|
|
+ if local.isSynced {
|
|
|
+ local.title = remote.title
|
|
|
+ local.startTime = remote.startTime
|
|
|
+ local.endTime = remote.endTime
|
|
|
+ }
|
|
|
+ if needsDetail {
|
|
|
+ local.serverRevision = remote.revision
|
|
|
+ local.syncState = .syncing
|
|
|
+ local.lastSyncError = nil
|
|
|
+ }
|
|
|
+ merged.append(IndexedSessionMatch(
|
|
|
+ index: remote,
|
|
|
+ local: local,
|
|
|
+ needsDetail: needsDetail
|
|
|
+ ))
|
|
|
}
|
|
|
} else if remote.deletedAt == nil {
|
|
|
- let id = UUID(uuidString: clientID) ?? UUID()
|
|
|
+ let id = clientID.flatMap(UUID.init(uuidString:)) ?? UUID()
|
|
|
let session = CelestiaSession(id: id, title: remote.title, startTime: remote.startTime)
|
|
|
- await apply(remote, to: session, userID: userID)
|
|
|
+ session.endTime = remote.endTime
|
|
|
+ session.cloudSessionId = remote.id
|
|
|
+ session.ownerUserID = userID
|
|
|
+ session.serverRevision = remote.revision
|
|
|
+ session.isSynced = true
|
|
|
+ session.syncState = .syncing
|
|
|
modelContext.insert(session)
|
|
|
- byClientID[clientID] = session
|
|
|
- merged.append((remote, session))
|
|
|
+ if let clientID {
|
|
|
+ byClientID[clientID] = session
|
|
|
+ }
|
|
|
+ byCloudID[remote.id] = session
|
|
|
+ merged.append(IndexedSessionMatch(
|
|
|
+ index: remote,
|
|
|
+ local: session,
|
|
|
+ needsDetail: true
|
|
|
+ ))
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -975,7 +1453,12 @@ final class SyncManager: ObservableObject {
|
|
|
: candidate
|
|
|
}
|
|
|
|
|
|
- private func apply(_ remote: RemoteSession, to local: CelestiaSession, userID: String) async {
|
|
|
+ private func apply(
|
|
|
+ _ remote: RemoteSession,
|
|
|
+ to local: CelestiaSession,
|
|
|
+ userID: String,
|
|
|
+ syncState: SessionSyncState = .synced
|
|
|
+ ) async {
|
|
|
local.title = remote.title
|
|
|
local.startTime = remote.startTime
|
|
|
local.endTime = remote.endTime
|
|
|
@@ -983,7 +1466,7 @@ final class SyncManager: ObservableObject {
|
|
|
local.ownerUserID = userID
|
|
|
local.serverRevision = remote.revision
|
|
|
local.isSynced = true
|
|
|
- local.syncState = .synced
|
|
|
+ local.syncState = syncState
|
|
|
local.lastSyncError = nil
|
|
|
local.events.removeAll()
|
|
|
for (index, remoteEvent) in (remote.events ?? []).enumerated() {
|