|
|
@@ -1,6 +1,10 @@
|
|
|
import Foundation
|
|
|
import SwiftData
|
|
|
import UniformTypeIdentifiers
|
|
|
+import Network
|
|
|
+import CryptoKit
|
|
|
+
|
|
|
+// Remote contract: Docs/RemoteAPI.openapi.yaml
|
|
|
|
|
|
struct RemoteAsset: Decodable {
|
|
|
let id: String
|
|
|
@@ -10,6 +14,18 @@ struct RemoteAsset: Decodable {
|
|
|
let mimeType: String
|
|
|
let sizeBytes: Int64
|
|
|
let sha256: String
|
|
|
+ let createdAt: Date?
|
|
|
+}
|
|
|
+
|
|
|
+struct AssetUploadResult {
|
|
|
+ let asset: RemoteAsset
|
|
|
+ let sessionRevision: Int64?
|
|
|
+}
|
|
|
+
|
|
|
+struct StorageQuota: Decodable {
|
|
|
+ let totalBytes: Int64
|
|
|
+ let usedBytes: Int64
|
|
|
+ let remainingBytes: Int64
|
|
|
}
|
|
|
|
|
|
struct RemoteEvent: Decodable {
|
|
|
@@ -42,6 +58,8 @@ private struct SessionUpload: Encodable {
|
|
|
let endTime: Date?
|
|
|
let durationMs: Int64
|
|
|
let events: [EventUpload]
|
|
|
+ let baseRevision: Int64?
|
|
|
+ let deletedEventClientIds: [String]
|
|
|
}
|
|
|
|
|
|
private struct EventUpload: Encodable {
|
|
|
@@ -57,7 +75,55 @@ private struct SyncCheckpoint: Decodable {
|
|
|
let syncedAt: Date
|
|
|
}
|
|
|
|
|
|
+private struct AssetUploadPayload: Decodable {
|
|
|
+ let asset: RemoteAsset
|
|
|
+ let sessionRevision: Int64?
|
|
|
+
|
|
|
+ private enum CodingKeys: String, CodingKey {
|
|
|
+ case asset
|
|
|
+ case sessionRevision
|
|
|
+ }
|
|
|
+
|
|
|
+ init(from decoder: Decoder) throws {
|
|
|
+ if let container = try? decoder.container(keyedBy: CodingKeys.self),
|
|
|
+ container.contains(.asset) {
|
|
|
+ asset = try container.decode(RemoteAsset.self, forKey: .asset)
|
|
|
+ sessionRevision = try container.decodeIfPresent(Int64.self, forKey: .sessionRevision)
|
|
|
+ } else {
|
|
|
+ asset = try RemoteAsset(from: decoder)
|
|
|
+ sessionRevision = nil
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+private struct ChunkUploadInitBody: Encodable {
|
|
|
+ let clientId: String
|
|
|
+ let kind: String
|
|
|
+ let fileName: String
|
|
|
+ let mimeType: String
|
|
|
+ let fileSize: Int64
|
|
|
+ let chunkSize: Int
|
|
|
+}
|
|
|
+
|
|
|
+private struct ChunkUploadInitResponse: Decodable {
|
|
|
+ let uploadId: String
|
|
|
+ let totalChunks: Int
|
|
|
+ let chunkSize: Int
|
|
|
+}
|
|
|
+
|
|
|
+private struct ChunkUploadProgressResponse: Decodable {
|
|
|
+ let uploadedChunks: Int
|
|
|
+ let totalChunks: Int
|
|
|
+}
|
|
|
+
|
|
|
+private struct ChunkUploadCompleteBody: Encodable {
|
|
|
+ let uploadId: String
|
|
|
+ let totalChunks: Int
|
|
|
+}
|
|
|
+
|
|
|
final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
+ private static let chunkedUploadThreshold: Int64 = 16 * 1024 * 1024
|
|
|
+ private static let preferredChunkSize = 8 * 1024 * 1024
|
|
|
private let client: APIClient
|
|
|
|
|
|
init(client: APIClient = .shared) {
|
|
|
@@ -68,7 +134,10 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
try await client.request("sessions?includeDeleted=true", authenticated: true)
|
|
|
}
|
|
|
|
|
|
- func syncSession(_ session: CelestiaSession) async throws -> RemoteSession {
|
|
|
+ func syncSession(
|
|
|
+ _ session: CelestiaSession,
|
|
|
+ deletedEventClientIDs: [String]
|
|
|
+ ) async throws -> RemoteSession {
|
|
|
let payload = SessionUpload(
|
|
|
clientId: session.id.uuidString,
|
|
|
title: session.title,
|
|
|
@@ -84,22 +153,48 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
voiceStartOffsetMs: $0.voiceStartOffsetMs,
|
|
|
voiceEndOffsetMs: $0.voiceEndOffsetMs
|
|
|
)
|
|
|
- }
|
|
|
+ },
|
|
|
+ baseRevision: session.serverRevision > 0 ? session.serverRevision : nil,
|
|
|
+ deletedEventClientIds: deletedEventClientIDs
|
|
|
)
|
|
|
let encoder = JSONEncoder()
|
|
|
encoder.dateEncodingStrategy = .iso8601
|
|
|
return try await client.request("sessions", method: .post, body: try encoder.encode(payload), authenticated: true)
|
|
|
}
|
|
|
|
|
|
- func uploadAsset(sessionID: String, clientID: String, kind: String, fileURL: URL) async throws -> RemoteAsset {
|
|
|
+ func fetchStorageQuota() async throws -> StorageQuota {
|
|
|
+ try await client.request("storage/quota", authenticated: true)
|
|
|
+ }
|
|
|
+
|
|
|
+ func uploadAsset(
|
|
|
+ sessionID: String,
|
|
|
+ clientID: String,
|
|
|
+ kind: String,
|
|
|
+ fileURL: URL,
|
|
|
+ progress: @escaping @Sendable (Double) -> Void
|
|
|
+ ) async throws -> AssetUploadResult {
|
|
|
let mimeType = UTType(filenameExtension: fileURL.pathExtension)?.preferredMIMEType ?? "application/octet-stream"
|
|
|
- return try await client.upload(
|
|
|
+ let size = try fileSize(of: fileURL)
|
|
|
+ if size >= Self.chunkedUploadThreshold {
|
|
|
+ return try await uploadAssetInChunks(
|
|
|
+ sessionID: sessionID,
|
|
|
+ clientID: clientID,
|
|
|
+ kind: kind,
|
|
|
+ fileURL: fileURL,
|
|
|
+ fileSize: size,
|
|
|
+ mimeType: mimeType,
|
|
|
+ progress: progress
|
|
|
+ )
|
|
|
+ }
|
|
|
+ let payload: AssetUploadPayload = try await client.upload(
|
|
|
"sessions/\(sessionID)/assets",
|
|
|
fileURL: fileURL,
|
|
|
fileName: fileURL.lastPathComponent,
|
|
|
mimeType: mimeType,
|
|
|
- fields: ["clientId": clientID, "kind": kind]
|
|
|
+ fields: ["clientId": clientID, "kind": kind],
|
|
|
+ progress: progress
|
|
|
)
|
|
|
+ return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
|
|
|
}
|
|
|
|
|
|
func downloadAsset(sessionID: String, asset: RemoteAsset, destinationURL: URL) async throws {
|
|
|
@@ -109,6 +204,82 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
func recordSyncCheckpoint() async throws {
|
|
|
let _: SyncCheckpoint = try await client.request("sync/trigger", method: .post, authenticated: true)
|
|
|
}
|
|
|
+
|
|
|
+ private func uploadAssetInChunks(
|
|
|
+ sessionID: String,
|
|
|
+ clientID: String,
|
|
|
+ kind: String,
|
|
|
+ fileURL: URL,
|
|
|
+ fileSize: Int64,
|
|
|
+ mimeType: String,
|
|
|
+ progress: @escaping @Sendable (Double) -> Void
|
|
|
+ ) async throws -> AssetUploadResult {
|
|
|
+ let body = ChunkUploadInitBody(
|
|
|
+ clientId: clientID,
|
|
|
+ kind: kind,
|
|
|
+ fileName: fileURL.lastPathComponent,
|
|
|
+ mimeType: mimeType,
|
|
|
+ fileSize: fileSize,
|
|
|
+ chunkSize: Self.preferredChunkSize
|
|
|
+ )
|
|
|
+ let encoder = JSONEncoder()
|
|
|
+ let upload: ChunkUploadInitResponse = try await client.request(
|
|
|
+ "sessions/\(sessionID)/assets/init",
|
|
|
+ method: .post,
|
|
|
+ body: try encoder.encode(body),
|
|
|
+ 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 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 completeBody = ChunkUploadCompleteBody(
|
|
|
+ uploadId: upload.uploadId,
|
|
|
+ totalChunks: upload.totalChunks
|
|
|
+ )
|
|
|
+ let payload: AssetUploadPayload = try await client.request(
|
|
|
+ "sessions/\(sessionID)/assets/complete",
|
|
|
+ method: .post,
|
|
|
+ body: try encoder.encode(completeBody),
|
|
|
+ authenticated: true
|
|
|
+ )
|
|
|
+ progress(1)
|
|
|
+ return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
|
|
|
+ }
|
|
|
+
|
|
|
+ private func fileSize(of url: URL) throws -> Int64 {
|
|
|
+ let values = try url.resourceValues(forKeys: [.fileSizeKey])
|
|
|
+ guard let size = values.fileSize else {
|
|
|
+ throw APIError.transport("无法读取待上传文件大小")
|
|
|
+ }
|
|
|
+ return Int64(size)
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
struct RemoteBoundDevice: Decodable {
|
|
|
@@ -175,6 +346,37 @@ final class DeviceCloudService {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+final class NetworkStatusMonitor: ObservableObject, @unchecked Sendable {
|
|
|
+ static let shared = NetworkStatusMonitor()
|
|
|
+
|
|
|
+ @Published private(set) var isConnected = true
|
|
|
+ @Published private(set) var isWiFi = false
|
|
|
+ @Published private(set) var isCellular = false
|
|
|
+
|
|
|
+ private let monitor = NWPathMonitor()
|
|
|
+ private let queue = DispatchQueue(label: "com.celestia.trace.network-path")
|
|
|
+
|
|
|
+ private init() {
|
|
|
+ monitor.pathUpdateHandler = { [weak self] path in
|
|
|
+ let connected = path.status == .satisfied
|
|
|
+ let usesWiFi = path.usesInterfaceType(.wifi)
|
|
|
+ let usesCellular = path.usesInterfaceType(.cellular)
|
|
|
+ DispatchQueue.main.async {
|
|
|
+ self?.isConnected = connected
|
|
|
+ self?.isWiFi = usesWiFi
|
|
|
+ self?.isCellular = usesCellular
|
|
|
+ }
|
|
|
+ }
|
|
|
+ monitor.start(queue: queue)
|
|
|
+ }
|
|
|
+
|
|
|
+ var connectionName: String {
|
|
|
+ if isWiFi { return "Wi-Fi" }
|
|
|
+ if isCellular { return "蜂窝网络" }
|
|
|
+ return isConnected ? "当前网络" : "网络"
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
@MainActor
|
|
|
final class SyncManager: ObservableObject {
|
|
|
static let shared = SyncManager()
|
|
|
@@ -184,14 +386,32 @@ final class SyncManager: ObservableObject {
|
|
|
@Published private(set) var lastErrorMessage: String?
|
|
|
@Published private(set) var completedCount = 0
|
|
|
@Published private(set) var totalCount = 0
|
|
|
+ @Published private(set) var activeSessionID: UUID?
|
|
|
+ @Published private(set) var syncProgress: Double = 0
|
|
|
|
|
|
private let service: NetworkServiceProtocol
|
|
|
+ private var activeSyncTask: Task<Bool, Never>?
|
|
|
|
|
|
init(service: NetworkServiceProtocol = RemoteNetworkService()) {
|
|
|
self.service = service
|
|
|
lastSyncDate = UserDefaults.standard.object(forKey: "com.celestia.trace.last_server_sync") as? Date
|
|
|
}
|
|
|
|
|
|
+ func estimatedUploadBytes(for session: CelestiaSession) -> Int64 {
|
|
|
+ var urls: [URL] = []
|
|
|
+ if let url = AudioPathHelper.resolveURL(for: session.localAudioPath) {
|
|
|
+ urls.append(url)
|
|
|
+ }
|
|
|
+ urls.append(contentsOf: session.events.compactMap { event in
|
|
|
+ guard event.eventType == "PHOTO" else { return nil }
|
|
|
+ return AudioPathHelper.resolveURL(for: event.localFilePath)
|
|
|
+ })
|
|
|
+ return urls.reduce(into: 0) { total, url in
|
|
|
+ let values = try? url.resourceValues(forKeys: [.fileSizeKey])
|
|
|
+ total += Int64(values?.fileSize ?? 0)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
func sync(
|
|
|
sessions: [CelestiaSession],
|
|
|
modelContext: ModelContext,
|
|
|
@@ -199,38 +419,107 @@ final class SyncManager: ObservableObject {
|
|
|
) async -> Bool {
|
|
|
guard !isSyncing else { return false }
|
|
|
isSyncing = true
|
|
|
+ let task = Task { @MainActor [self] in
|
|
|
+ await performSync(
|
|
|
+ sessions: sessions,
|
|
|
+ modelContext: modelContext,
|
|
|
+ userID: userID
|
|
|
+ )
|
|
|
+ }
|
|
|
+ activeSyncTask = task
|
|
|
+ let result = await withTaskCancellationHandler {
|
|
|
+ await task.value
|
|
|
+ } onCancel: {
|
|
|
+ task.cancel()
|
|
|
+ }
|
|
|
+ activeSyncTask = nil
|
|
|
+ isSyncing = false
|
|
|
+ activeSessionID = nil
|
|
|
+ return result
|
|
|
+ }
|
|
|
+
|
|
|
+ func pauseSync(sessionID: UUID) {
|
|
|
+ guard isSyncing, activeSessionID == sessionID else { return }
|
|
|
+ activeSyncTask?.cancel()
|
|
|
+ }
|
|
|
+
|
|
|
+ private func performSync(
|
|
|
+ sessions: [CelestiaSession],
|
|
|
+ modelContext: ModelContext,
|
|
|
+ userID: String
|
|
|
+ ) async -> Bool {
|
|
|
lastErrorMessage = nil
|
|
|
completedCount = 0
|
|
|
+ activeSessionID = nil
|
|
|
+ syncProgress = 0
|
|
|
totalCount = sessions.filter {
|
|
|
($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
|
|
|
}.count
|
|
|
- defer { isSyncing = false }
|
|
|
-
|
|
|
do {
|
|
|
let remoteSessions = try await service.fetchSessions()
|
|
|
+ try Task.checkCancellation()
|
|
|
let mergedSessions = merge(remoteSessions, into: sessions, modelContext: modelContext, userID: userID)
|
|
|
for (remote, local) in mergedSessions {
|
|
|
try await downloadMissingAssets(for: local, remote: remote)
|
|
|
+ try Task.checkCancellation()
|
|
|
}
|
|
|
try modelContext.save()
|
|
|
|
|
|
let eligible = sessions.filter {
|
|
|
- ($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
|
|
|
+ ($0.ownerUserID == nil || $0.ownerUserID == userID)
|
|
|
+ && (!$0.isSynced || $0.cloudSessionId == nil)
|
|
|
+ && $0.syncState != .conflict
|
|
|
}
|
|
|
var failures: [String] = []
|
|
|
- for session in eligible {
|
|
|
+ let eligibleCount = max(eligible.count, 1)
|
|
|
+ for (index, session) in eligible.enumerated() {
|
|
|
+ activeSessionID = session.id
|
|
|
+ let sessionStartProgress = Double(index) / Double(eligibleCount)
|
|
|
+ let sessionProgressSpan = 1.0 / Double(eligibleCount)
|
|
|
+ syncProgress = sessionStartProgress
|
|
|
session.ownerUserID = userID
|
|
|
session.syncState = .syncing
|
|
|
session.lastSyncError = nil
|
|
|
do {
|
|
|
- let remote = try await service.syncSession(session)
|
|
|
+ let previousRemote = remoteSessions.first {
|
|
|
+ $0.id == session.cloudSessionId
|
|
|
+ || $0.clientId?.caseInsensitiveCompare(session.id.uuidString) == .orderedSame
|
|
|
+ }
|
|
|
+ let localEventIDs = Set(session.events.map { $0.id.uuidString.lowercased() })
|
|
|
+ let deletedEventClientIDs = previousRemote?.events?
|
|
|
+ .compactMap(\.clientId)
|
|
|
+ .filter { !localEventIDs.contains($0.lowercased()) } ?? []
|
|
|
+ let remote = try await service.syncSession(
|
|
|
+ session,
|
|
|
+ deletedEventClientIDs: deletedEventClientIDs
|
|
|
+ )
|
|
|
+ try Task.checkCancellation()
|
|
|
+ syncProgress = sessionStartProgress + sessionProgressSpan * 0.12
|
|
|
session.cloudSessionId = remote.id
|
|
|
session.serverRevision = remote.revision
|
|
|
- try await uploadLocalAssets(for: session, remote: remote)
|
|
|
+ try await uploadLocalAssets(for: session, remote: remote) { [weak self] assetProgress in
|
|
|
+ Task { @MainActor [weak self] in
|
|
|
+ self?.syncProgress = sessionStartProgress
|
|
|
+ + sessionProgressSpan * (0.12 + assetProgress * 0.83)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ try Task.checkCancellation()
|
|
|
session.isSynced = true
|
|
|
session.syncState = .synced
|
|
|
session.lastSyncedAt = Date()
|
|
|
completedCount += 1
|
|
|
+ syncProgress = sessionStartProgress + sessionProgressSpan
|
|
|
+ } catch where Task.isCancelled {
|
|
|
+ session.isSynced = false
|
|
|
+ session.syncState = .pending
|
|
|
+ session.lastSyncError = nil
|
|
|
+ syncProgress = 0
|
|
|
+ try? modelContext.save()
|
|
|
+ return false
|
|
|
+ } catch APIError.server(let code, _) where code == 409 {
|
|
|
+ session.isSynced = false
|
|
|
+ session.syncState = .conflict
|
|
|
+ session.lastSyncError = "本地版本 \(session.serverRevision) 与云端版本不一致"
|
|
|
} catch {
|
|
|
session.isSynced = false
|
|
|
session.syncState = .failed
|
|
|
@@ -239,6 +528,12 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
try modelContext.save()
|
|
|
}
|
|
|
+ let conflicts = sessions.filter { $0.syncState == .conflict }
|
|
|
+ if !conflicts.isEmpty {
|
|
|
+ failures.append(contentsOf: conflicts.map {
|
|
|
+ "\($0.title):本地和云端版本不一致"
|
|
|
+ })
|
|
|
+ }
|
|
|
guard failures.isEmpty else {
|
|
|
lastErrorMessage = failures.joined(separator: "\n")
|
|
|
return false
|
|
|
@@ -248,22 +543,35 @@ final class SyncManager: ObservableObject {
|
|
|
lastSyncDate = now
|
|
|
UserDefaults.standard.set(now, forKey: "com.celestia.trace.last_server_sync")
|
|
|
return true
|
|
|
+ } catch where Task.isCancelled {
|
|
|
+ syncProgress = 0
|
|
|
+ return false
|
|
|
} catch {
|
|
|
lastErrorMessage = error.localizedDescription
|
|
|
return false
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- private func uploadLocalAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
|
|
|
+ private func uploadLocalAssets(
|
|
|
+ for session: CelestiaSession,
|
|
|
+ remote: RemoteSession,
|
|
|
+ progress: @escaping @Sendable (Double) -> Void
|
|
|
+ ) async throws {
|
|
|
guard let cloudID = session.cloudSessionId else { throw APIError.missingData }
|
|
|
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 clientID = "\(session.id.uuidString)-audio"
|
|
|
- if !existingIDs.contains(clientID) {
|
|
|
- _ = try await service.uploadAsset(sessionID: cloudID, clientID: clientID, kind: "AUDIO", fileURL: url)
|
|
|
+ let sha256 = try fileSHA256(of: url)
|
|
|
+ let hasMatchingAudio = (remote.assets ?? []).contains {
|
|
|
+ $0.kind.uppercased() == "AUDIO"
|
|
|
+ && $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
|
|
|
+ }
|
|
|
+ if !hasMatchingAudio {
|
|
|
+ let clientID = "\(session.id.uuidString)-audio-\(sha256.prefix(16))"
|
|
|
+ uploads.append((clientID, "AUDIO", url, fileSize(of: url)))
|
|
|
}
|
|
|
}
|
|
|
for event in session.events where event.eventType == "PHOTO" {
|
|
|
@@ -273,18 +581,75 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
|
let clientID = "\(event.id.uuidString)-photo"
|
|
|
if !existingIDs.contains(clientID) {
|
|
|
- _ = try await service.uploadAsset(sessionID: cloudID, clientID: clientID, kind: "PHOTO", fileURL: url)
|
|
|
+ uploads.append((clientID, "PHOTO", url, fileSize(of: url)))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ let totalBytes = max(uploads.reduce(Int64(0)) { $0 + $1.size }, 1)
|
|
|
+ let requiredBytes = uploads.reduce(Int64(0)) { $0 + $1.size }
|
|
|
+ if requiredBytes > 0 {
|
|
|
+ let quota = try await service.fetchStorageQuota()
|
|
|
+ guard requiredBytes <= quota.remainingBytes else {
|
|
|
+ throw APIError.storageQuotaExceeded(
|
|
|
+ requiredBytes: requiredBytes,
|
|
|
+ remainingBytes: quota.remainingBytes
|
|
|
+ )
|
|
|
}
|
|
|
}
|
|
|
+ var completedBytes: Int64 = 0
|
|
|
+ progress(uploads.isEmpty ? 1 : 0)
|
|
|
+ for upload in uploads {
|
|
|
+ let bytesBeforeUpload = completedBytes
|
|
|
+ let result = try await service.uploadAsset(
|
|
|
+ sessionID: cloudID,
|
|
|
+ clientID: upload.clientID,
|
|
|
+ kind: upload.kind,
|
|
|
+ fileURL: upload.url
|
|
|
+ ) { fraction in
|
|
|
+ let sentBytes = Double(bytesBeforeUpload) + Double(upload.size) * fraction
|
|
|
+ progress(min(max(sentBytes / Double(totalBytes), 0), 1))
|
|
|
+ }
|
|
|
+ if let revision = result.sessionRevision {
|
|
|
+ session.serverRevision = revision
|
|
|
+ }
|
|
|
+ completedBytes += upload.size
|
|
|
+ progress(Double(completedBytes) / Double(totalBytes))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private func fileSize(of url: URL) -> Int64 {
|
|
|
+ let values = try? url.resourceValues(forKeys: [.fileSizeKey])
|
|
|
+ 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)
|
|
|
+ }
|
|
|
+ return hasher.finalize().map { String(format: "%02x", $0) }.joined()
|
|
|
}
|
|
|
|
|
|
private func downloadMissingAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
|
|
|
let assets = remote.assets ?? []
|
|
|
- if (session.localAudioPath == nil || AudioPathHelper.resolveURL(for: session.localAudioPath) == nil),
|
|
|
- let audio = assets.first(where: { $0.kind.uppercased() == "AUDIO" }) {
|
|
|
- let destination = downloadDestination(for: audio)
|
|
|
- try await service.downloadAsset(sessionID: remote.id, asset: audio, destinationURL: destination)
|
|
|
- session.localAudioPath = AudioPathHelper.relativePath(from: destination.path)
|
|
|
+ if let audio = assets
|
|
|
+ .filter({ $0.kind.uppercased() == "AUDIO" })
|
|
|
+ .max(by: {
|
|
|
+ ($0.createdAt ?? .distantPast) < ($1.createdAt ?? .distantPast)
|
|
|
+ }) {
|
|
|
+ let localURL = AudioPathHelper.resolveURL(for: session.localAudioPath)
|
|
|
+ let localHash = localURL.flatMap { try? fileSHA256(of: $0) }
|
|
|
+ let needsDownload = localURL == nil
|
|
|
+ || (session.isSynced
|
|
|
+ && 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)
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
let photoAssets = Dictionary(uniqueKeysWithValues: assets
|
|
|
@@ -322,6 +687,12 @@ final class SyncManager: ObservableObject {
|
|
|
modelContext.delete(local)
|
|
|
continue
|
|
|
}
|
|
|
+ if !local.isSynced,
|
|
|
+ remote.revision > local.serverRevision {
|
|
|
+ local.syncState = .conflict
|
|
|
+ local.lastSyncError = "本地版本 \(local.serverRevision) 与云端版本 \(remote.revision) 不一致"
|
|
|
+ continue
|
|
|
+ }
|
|
|
if local.isSynced, remote.revision > local.serverRevision {
|
|
|
apply(remote, to: local, userID: userID)
|
|
|
}
|