|
@@ -36,6 +36,10 @@ struct RemoteEvent: Decodable {
|
|
|
let textContent: String?
|
|
let textContent: String?
|
|
|
let voiceStartOffsetMs: Int64?
|
|
let voiceStartOffsetMs: Int64?
|
|
|
let voiceEndOffsetMs: Int64?
|
|
let voiceEndOffsetMs: Int64?
|
|
|
|
|
+ let locationName: String?
|
|
|
|
|
+ let locationAddress: String?
|
|
|
|
|
+ let latitude: Double?
|
|
|
|
|
+ let longitude: Double?
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
struct RemoteSession: Decodable {
|
|
struct RemoteSession: Decodable {
|
|
@@ -69,6 +73,10 @@ private struct EventUpload: Encodable {
|
|
|
let textContent: String?
|
|
let textContent: String?
|
|
|
let voiceStartOffsetMs: Int64?
|
|
let voiceStartOffsetMs: Int64?
|
|
|
let voiceEndOffsetMs: Int64?
|
|
let voiceEndOffsetMs: Int64?
|
|
|
|
|
+ let locationName: String?
|
|
|
|
|
+ let locationAddress: String?
|
|
|
|
|
+ let latitude: Double?
|
|
|
|
|
+ let longitude: Double?
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
private struct SyncCheckpoint: Decodable {
|
|
private struct SyncCheckpoint: Decodable {
|
|
@@ -151,7 +159,11 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
eventType: $0.eventType,
|
|
eventType: $0.eventType,
|
|
|
textContent: $0.textContent,
|
|
textContent: $0.textContent,
|
|
|
voiceStartOffsetMs: $0.voiceStartOffsetMs,
|
|
voiceStartOffsetMs: $0.voiceStartOffsetMs,
|
|
|
- voiceEndOffsetMs: $0.voiceEndOffsetMs
|
|
|
|
|
|
|
+ voiceEndOffsetMs: $0.voiceEndOffsetMs,
|
|
|
|
|
+ locationName: $0.locationName,
|
|
|
|
|
+ locationAddress: $0.locationAddress,
|
|
|
|
|
+ latitude: $0.latitude,
|
|
|
|
|
+ longitude: $0.longitude
|
|
|
)
|
|
)
|
|
|
},
|
|
},
|
|
|
baseRevision: session.serverRevision > 0 ? session.serverRevision : nil,
|
|
baseRevision: session.serverRevision > 0 ? session.serverRevision : nil,
|
|
@@ -162,6 +174,14 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
return try await client.request("sessions", method: .post, body: try encoder.encode(payload), authenticated: true)
|
|
return try await client.request("sessions", method: .post, body: try encoder.encode(payload), authenticated: true)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ func deleteSession(sessionID: String) async throws {
|
|
|
|
|
+ try await client.requestVoid(
|
|
|
|
|
+ "sessions/\(sessionID)",
|
|
|
|
|
+ method: .delete,
|
|
|
|
|
+ authenticated: true
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
func fetchStorageQuota() async throws -> StorageQuota {
|
|
func fetchStorageQuota() async throws -> StorageQuota {
|
|
|
try await client.request("storage/quota", authenticated: true)
|
|
try await client.request("storage/quota", authenticated: true)
|
|
|
}
|
|
}
|
|
@@ -230,32 +250,35 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
authenticated: true
|
|
authenticated: true
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
- let input = try FileHandle(forReadingFrom: fileURL)
|
|
|
|
|
- defer { try? input.close() }
|
|
|
|
|
for chunkIndex in 0..<upload.totalChunks {
|
|
for chunkIndex in 0..<upload.totalChunks {
|
|
|
let offset = UInt64(chunkIndex) * UInt64(upload.chunkSize)
|
|
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 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 +296,34 @@ final class RemoteNetworkService: NetworkServiceProtocol {
|
|
|
return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
|
|
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 {
|
|
private func fileSize(of url: URL) throws -> Int64 {
|
|
|
let values = try url.resourceValues(forKeys: [.fileSizeKey])
|
|
let values = try url.resourceValues(forKeys: [.fileSizeKey])
|
|
|
guard let size = values.fileSize else {
|
|
guard let size = values.fileSize else {
|
|
@@ -391,6 +442,8 @@ final class SyncManager: ObservableObject {
|
|
|
|
|
|
|
|
private let service: NetworkServiceProtocol
|
|
private let service: NetworkServiceProtocol
|
|
|
private var activeSyncTask: Task<Bool, Never>?
|
|
private var activeSyncTask: Task<Bool, Never>?
|
|
|
|
|
+ private var automaticSyncTasks: [UUID: Task<Void, Never>] = [:]
|
|
|
|
|
+ private var automaticSyncGenerations: [UUID: UUID] = [:]
|
|
|
|
|
|
|
|
init(service: NetworkServiceProtocol = RemoteNetworkService()) {
|
|
init(service: NetworkServiceProtocol = RemoteNetworkService()) {
|
|
|
self.service = service
|
|
self.service = service
|
|
@@ -412,12 +465,63 @@ 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(
|
|
func sync(
|
|
|
sessions: [CelestiaSession],
|
|
sessions: [CelestiaSession],
|
|
|
modelContext: ModelContext,
|
|
modelContext: ModelContext,
|
|
|
userID: String
|
|
userID: String
|
|
|
) async -> Bool {
|
|
) async -> Bool {
|
|
|
- guard !isSyncing else { return false }
|
|
|
|
|
|
|
+ guard !isSyncing else {
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "已有同步任务正在运行,忽略重复请求", level: .warning)
|
|
|
|
|
+ return false
|
|
|
|
|
+ }
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "开始同步,本地共 \(sessions.count) 条记录")
|
|
|
isSyncing = true
|
|
isSyncing = true
|
|
|
let task = Task { @MainActor [self] in
|
|
let task = Task { @MainActor [self] in
|
|
|
await performSync(
|
|
await performSync(
|
|
@@ -435,6 +539,11 @@ final class SyncManager: ObservableObject {
|
|
|
activeSyncTask = nil
|
|
activeSyncTask = nil
|
|
|
isSyncing = false
|
|
isSyncing = false
|
|
|
activeSessionID = nil
|
|
activeSessionID = nil
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ result ? "同步任务完成" : "同步任务未完成",
|
|
|
|
|
+ level: result ? .success : .warning
|
|
|
|
|
+ )
|
|
|
return result
|
|
return result
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -452,27 +561,48 @@ final class SyncManager: ObservableObject {
|
|
|
completedCount = 0
|
|
completedCount = 0
|
|
|
activeSessionID = nil
|
|
activeSessionID = nil
|
|
|
syncProgress = 0
|
|
syncProgress = 0
|
|
|
- totalCount = sessions.filter {
|
|
|
|
|
- ($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
|
|
|
|
|
- }.count
|
|
|
|
|
|
|
+ let queuedSessions = deduplicatedSessions(sessions.filter {
|
|
|
|
|
+ ($0.ownerUserID == nil || $0.ownerUserID == userID)
|
|
|
|
|
+ && $0.needsCloudSync
|
|
|
|
|
+ && $0.syncState != .conflict
|
|
|
|
|
+ })
|
|
|
|
|
+ totalCount = queuedSessions.count
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "待上传 \(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 {
|
|
do {
|
|
|
let remoteSessions = try await service.fetchSessions()
|
|
let remoteSessions = try await service.fetchSessions()
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "获取到 \(remoteSessions.count) 条云端记录", level: .success)
|
|
|
try Task.checkCancellation()
|
|
try Task.checkCancellation()
|
|
|
- let mergedSessions = merge(remoteSessions, into: sessions, modelContext: modelContext, userID: userID)
|
|
|
|
|
|
|
+ // 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 allLocalSessions = try modelContext.fetch(FetchDescriptor<CelestiaSession>())
|
|
|
|
|
+ let mergedSessions = try await merge(
|
|
|
|
|
+ remoteSessions,
|
|
|
|
|
+ into: allLocalSessions,
|
|
|
|
|
+ modelContext: modelContext,
|
|
|
|
|
+ userID: userID
|
|
|
|
|
+ )
|
|
|
for (remote, local) in mergedSessions {
|
|
for (remote, local) in mergedSessions {
|
|
|
try await downloadMissingAssets(for: local, remote: remote)
|
|
try await downloadMissingAssets(for: local, remote: remote)
|
|
|
try Task.checkCancellation()
|
|
try Task.checkCancellation()
|
|
|
|
|
+ await Task.yield()
|
|
|
}
|
|
}
|
|
|
try modelContext.save()
|
|
try modelContext.save()
|
|
|
|
|
|
|
|
- let eligible = sessions.filter {
|
|
|
|
|
|
|
+ let eligible = deduplicatedSessions(sessions.filter {
|
|
|
($0.ownerUserID == nil || $0.ownerUserID == userID)
|
|
($0.ownerUserID == nil || $0.ownerUserID == userID)
|
|
|
- && (!$0.isSynced || $0.cloudSessionId == nil)
|
|
|
|
|
|
|
+ && $0.needsCloudSync
|
|
|
&& $0.syncState != .conflict
|
|
&& $0.syncState != .conflict
|
|
|
- }
|
|
|
|
|
|
|
+ })
|
|
|
var failures: [String] = []
|
|
var failures: [String] = []
|
|
|
let eligibleCount = max(eligible.count, 1)
|
|
let eligibleCount = max(eligible.count, 1)
|
|
|
for (index, session) in eligible.enumerated() {
|
|
for (index, session) in eligible.enumerated() {
|
|
|
|
|
+ try Task.checkCancellation()
|
|
|
|
|
+ await Task.yield()
|
|
|
activeSessionID = session.id
|
|
activeSessionID = session.id
|
|
|
let sessionStartProgress = Double(index) / Double(eligibleCount)
|
|
let sessionStartProgress = Double(index) / Double(eligibleCount)
|
|
|
let sessionProgressSpan = 1.0 / Double(eligibleCount)
|
|
let sessionProgressSpan = 1.0 / Double(eligibleCount)
|
|
@@ -480,6 +610,10 @@ final class SyncManager: ObservableObject {
|
|
|
session.ownerUserID = userID
|
|
session.ownerUserID = userID
|
|
|
session.syncState = .syncing
|
|
session.syncState = .syncing
|
|
|
session.lastSyncError = nil
|
|
session.lastSyncError = nil
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "正在同步 \(index + 1)/\(eligible.count):\(session.title)"
|
|
|
|
|
+ )
|
|
|
do {
|
|
do {
|
|
|
let previousRemote = remoteSessions.first {
|
|
let previousRemote = remoteSessions.first {
|
|
|
$0.id == session.cloudSessionId
|
|
$0.id == session.cloudSessionId
|
|
@@ -509,6 +643,11 @@ final class SyncManager: ObservableObject {
|
|
|
session.lastSyncedAt = Date()
|
|
session.lastSyncedAt = Date()
|
|
|
completedCount += 1
|
|
completedCount += 1
|
|
|
syncProgress = sessionStartProgress + sessionProgressSpan
|
|
syncProgress = sessionStartProgress + sessionProgressSpan
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "\(session.title) 同步成功,云端版本 v\(session.serverRevision)",
|
|
|
|
|
+ level: .success
|
|
|
|
|
+ )
|
|
|
} catch where Task.isCancelled {
|
|
} catch where Task.isCancelled {
|
|
|
session.isSynced = false
|
|
session.isSynced = false
|
|
|
session.syncState = .pending
|
|
session.syncState = .pending
|
|
@@ -520,11 +659,21 @@ final class SyncManager: ObservableObject {
|
|
|
session.isSynced = false
|
|
session.isSynced = false
|
|
|
session.syncState = .conflict
|
|
session.syncState = .conflict
|
|
|
session.lastSyncError = "本地版本 \(session.serverRevision) 与云端版本不一致"
|
|
session.lastSyncError = "本地版本 \(session.serverRevision) 与云端版本不一致"
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "\(session.title) 发生版本冲突",
|
|
|
|
|
+ level: .error
|
|
|
|
|
+ )
|
|
|
} catch {
|
|
} catch {
|
|
|
session.isSynced = false
|
|
session.isSynced = false
|
|
|
session.syncState = .failed
|
|
session.syncState = .failed
|
|
|
session.lastSyncError = error.localizedDescription
|
|
session.lastSyncError = error.localizedDescription
|
|
|
failures.append("\(session.title):\(error.localizedDescription)")
|
|
failures.append("\(session.title):\(error.localizedDescription)")
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "\(session.title) 同步失败:\(error.localizedDescription)",
|
|
|
|
|
+ level: .error
|
|
|
|
|
+ )
|
|
|
}
|
|
}
|
|
|
try modelContext.save()
|
|
try modelContext.save()
|
|
|
}
|
|
}
|
|
@@ -536,18 +685,21 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
}
|
|
|
guard failures.isEmpty else {
|
|
guard failures.isEmpty else {
|
|
|
lastErrorMessage = failures.joined(separator: "\n")
|
|
lastErrorMessage = failures.joined(separator: "\n")
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "同步结束,存在 \(failures.count) 个问题", level: .error)
|
|
|
return false
|
|
return false
|
|
|
}
|
|
}
|
|
|
try await service.recordSyncCheckpoint()
|
|
try await service.recordSyncCheckpoint()
|
|
|
let now = Date()
|
|
let now = Date()
|
|
|
lastSyncDate = now
|
|
lastSyncDate = now
|
|
|
UserDefaults.standard.set(now, forKey: "com.celestia.trace.last_server_sync")
|
|
UserDefaults.standard.set(now, forKey: "com.celestia.trace.last_server_sync")
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "云端检查点已更新", level: .success)
|
|
|
return true
|
|
return true
|
|
|
} catch where Task.isCancelled {
|
|
} catch where Task.isCancelled {
|
|
|
syncProgress = 0
|
|
syncProgress = 0
|
|
|
return false
|
|
return false
|
|
|
} catch {
|
|
} catch {
|
|
|
lastErrorMessage = error.localizedDescription
|
|
lastErrorMessage = error.localizedDescription
|
|
|
|
|
+ DeveloperLogStore.log("现场记录同步", "同步中断:\(error.localizedDescription)", level: .error)
|
|
|
return false
|
|
return false
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -564,7 +716,7 @@ final class SyncManager: ObservableObject {
|
|
|
guard let url = AudioPathHelper.resolveURL(for: path) else {
|
|
guard let url = AudioPathHelper.resolveURL(for: path) else {
|
|
|
throw APIError.transport("本地录音文件不存在:\(path)")
|
|
throw APIError.transport("本地录音文件不存在:\(path)")
|
|
|
}
|
|
}
|
|
|
- let sha256 = try fileSHA256(of: url)
|
|
|
|
|
|
|
+ let sha256 = try await fileSHA256(of: url)
|
|
|
let hasMatchingAudio = (remote.assets ?? []).contains {
|
|
let hasMatchingAudio = (remote.assets ?? []).contains {
|
|
|
$0.kind.uppercased() == "AUDIO"
|
|
$0.kind.uppercased() == "AUDIO"
|
|
|
&& $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
|
|
&& $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
|
|
@@ -622,15 +774,23 @@ final class SyncManager: ObservableObject {
|
|
|
return Int64(values?.fileSize ?? 0)
|
|
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 {
|
|
private func downloadMissingAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
|
|
@@ -641,7 +801,12 @@ final class SyncManager: ObservableObject {
|
|
|
($0.createdAt ?? .distantPast) < ($1.createdAt ?? .distantPast)
|
|
($0.createdAt ?? .distantPast) < ($1.createdAt ?? .distantPast)
|
|
|
}) {
|
|
}) {
|
|
|
let localURL = AudioPathHelper.resolveURL(for: session.localAudioPath)
|
|
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
|
|
let needsDownload = localURL == nil
|
|
|
|| (session.isSynced
|
|
|| (session.isSynced
|
|
|
&& localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)
|
|
&& localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)
|
|
@@ -652,9 +817,24 @@ final class SyncManager: ObservableObject {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- let photoAssets = Dictionary(uniqueKeysWithValues: assets
|
|
|
|
|
- .filter { $0.kind.uppercased() == "PHOTO" }
|
|
|
|
|
- .map { ($0.clientId.lowercased(), $0) })
|
|
|
|
|
|
|
+ let photoAssets = Dictionary(
|
|
|
|
|
+ assets
|
|
|
|
|
+ .filter { $0.kind.uppercased() == "PHOTO" }
|
|
|
|
|
+ .map { ($0.clientId.lowercased(), $0) },
|
|
|
|
|
+ uniquingKeysWith: { current, candidate in
|
|
|
|
|
+ (candidate.createdAt ?? .distantPast) > (current.createdAt ?? .distantPast)
|
|
|
|
|
+ ? candidate
|
|
|
|
|
+ : current
|
|
|
|
|
+ }
|
|
|
|
|
+ )
|
|
|
|
|
+ let remotePhotoCount = assets.lazy.filter { $0.kind.uppercased() == "PHOTO" }.count
|
|
|
|
|
+ if photoAssets.count < remotePhotoCount {
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "云端照片资源存在重复 clientId,已选择最新资源继续同步",
|
|
|
|
|
+ level: .warning
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
for event in session.events where event.eventType == "PHOTO" {
|
|
for event in session.events where event.eventType == "PHOTO" {
|
|
|
guard event.localFilePath == nil || AudioPathHelper.resolveURL(for: event.localFilePath) == nil else { continue }
|
|
guard event.localFilePath == nil || AudioPathHelper.resolveURL(for: event.localFilePath) == nil else { continue }
|
|
|
let key = "\(event.id.uuidString)-photo".lowercased()
|
|
let key = "\(event.id.uuidString)-photo".lowercased()
|
|
@@ -677,14 +857,38 @@ final class SyncManager: ObservableObject {
|
|
|
into localSessions: [CelestiaSession],
|
|
into localSessions: [CelestiaSession],
|
|
|
modelContext: ModelContext,
|
|
modelContext: ModelContext,
|
|
|
userID: String
|
|
userID: String
|
|
|
- ) -> [(RemoteSession, CelestiaSession)] {
|
|
|
|
|
- var byClientID = Dictionary(uniqueKeysWithValues: localSessions.map { ($0.id.uuidString.lowercased(), $0) })
|
|
|
|
|
|
|
+ ) async throws -> [(RemoteSession, CelestiaSession)] {
|
|
|
|
|
+ var byClientID = Dictionary(
|
|
|
|
|
+ localSessions.map { ($0.id.uuidString.lowercased(), $0) },
|
|
|
|
|
+ uniquingKeysWith: preferredLocalSession
|
|
|
|
|
+ )
|
|
|
|
|
+ if byClientID.count < localSessions.count {
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "检测到 \(localSessions.count - byClientID.count) 条重复本地记录,已选择云端版本较新的记录继续同步",
|
|
|
|
|
+ level: .warning
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ let activeRemoteSessions = remoteSessions.filter { $0.deletedAt == nil }
|
|
|
|
|
+ let activeRemoteIDs = Set(activeRemoteSessions.map(\.id))
|
|
|
|
|
+ let activeRemoteClientIDs = Set(activeRemoteSessions.compactMap { $0.clientId?.lowercased() })
|
|
|
|
|
+ var deletedLocalObjects: Set<ObjectIdentifier> = []
|
|
|
|
|
+ var deletedFileURLs: Set<URL> = []
|
|
|
var merged: [(RemoteSession, CelestiaSession)] = []
|
|
var merged: [(RemoteSession, CelestiaSession)] = []
|
|
|
- for remote in remoteSessions {
|
|
|
|
|
|
|
+ for (index, remote) in remoteSessions.enumerated() {
|
|
|
|
|
+ if index.isMultiple(of: 20) {
|
|
|
|
|
+ await Task.yield()
|
|
|
|
|
+ }
|
|
|
guard let clientID = remote.clientId?.lowercased() else { continue }
|
|
guard let clientID = remote.clientId?.lowercased() else { continue }
|
|
|
if let local = byClientID[clientID] {
|
|
if let local = byClientID[clientID] {
|
|
|
- if remote.deletedAt != nil, local.isSynced {
|
|
|
|
|
|
|
+ if remote.deletedAt != nil, local.hasConfirmedCloudSync {
|
|
|
|
|
+ deletedFileURLs.formUnion(localFileURLs(for: local))
|
|
|
|
|
+ deletedLocalObjects.insert(ObjectIdentifier(local))
|
|
|
modelContext.delete(local)
|
|
modelContext.delete(local)
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "云端记录已删除,同步移除本地记录:\(local.title)"
|
|
|
|
|
+ )
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
if !local.isSynced,
|
|
if !local.isSynced,
|
|
@@ -694,7 +898,7 @@ final class SyncManager: ObservableObject {
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
if local.isSynced, remote.revision > local.serverRevision {
|
|
if local.isSynced, remote.revision > local.serverRevision {
|
|
|
- apply(remote, to: local, userID: userID)
|
|
|
|
|
|
|
+ await apply(remote, to: local, userID: userID)
|
|
|
}
|
|
}
|
|
|
if remote.deletedAt == nil {
|
|
if remote.deletedAt == nil {
|
|
|
merged.append((remote, local))
|
|
merged.append((remote, local))
|
|
@@ -702,17 +906,76 @@ final class SyncManager: ObservableObject {
|
|
|
} else if remote.deletedAt == nil {
|
|
} else if remote.deletedAt == nil {
|
|
|
let id = UUID(uuidString: clientID) ?? UUID()
|
|
let id = UUID(uuidString: clientID) ?? UUID()
|
|
|
let session = CelestiaSession(id: id, title: remote.title, startTime: remote.startTime)
|
|
let session = CelestiaSession(id: id, title: remote.title, startTime: remote.startTime)
|
|
|
- apply(remote, to: session, userID: userID)
|
|
|
|
|
|
|
+ await apply(remote, to: session, userID: userID)
|
|
|
modelContext.insert(session)
|
|
modelContext.insert(session)
|
|
|
byClientID[clientID] = session
|
|
byClientID[clientID] = session
|
|
|
merged.append((remote, session))
|
|
merged.append((remote, session))
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- try? modelContext.save()
|
|
|
|
|
|
|
+
|
|
|
|
|
+ // Some servers may permanently remove a record instead of returning a
|
|
|
|
|
+ // tombstone. A previously confirmed cloud record that is absent by both
|
|
|
|
|
+ // server ID and client ID should mirror that deletion locally. Records
|
|
|
|
|
+ // with pending/failed/conflicting local content are deliberately kept.
|
|
|
|
|
+ for (index, local) in localSessions.enumerated() {
|
|
|
|
|
+ if index.isMultiple(of: 20) {
|
|
|
|
|
+ await Task.yield()
|
|
|
|
|
+ }
|
|
|
|
|
+ guard !deletedLocalObjects.contains(ObjectIdentifier(local)),
|
|
|
|
|
+ local.ownerUserID == nil || local.ownerUserID == userID,
|
|
|
|
|
+ local.hasConfirmedCloudSync else { continue }
|
|
|
|
|
+ let existsRemotely =
|
|
|
|
|
+ local.cloudSessionId.map(activeRemoteIDs.contains) == true
|
|
|
|
|
+ || activeRemoteClientIDs.contains(local.id.uuidString.lowercased())
|
|
|
|
|
+ guard !existsRemotely else { continue }
|
|
|
|
|
+
|
|
|
|
|
+ deletedFileURLs.formUnion(localFileURLs(for: local))
|
|
|
|
|
+ deletedLocalObjects.insert(ObjectIdentifier(local))
|
|
|
|
|
+ modelContext.delete(local)
|
|
|
|
|
+ DeveloperLogStore.log(
|
|
|
|
|
+ "现场记录同步",
|
|
|
|
|
+ "云端不存在已同步记录,同步移除本地记录:\(local.title)"
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ try modelContext.save()
|
|
|
|
|
+ for fileURL in deletedFileURLs {
|
|
|
|
|
+ try? FileManager.default.removeItem(at: fileURL)
|
|
|
|
|
+ }
|
|
|
return merged
|
|
return merged
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- private func apply(_ remote: RemoteSession, to local: CelestiaSession, userID: String) {
|
|
|
|
|
|
|
+ private func localFileURLs(for session: CelestiaSession) -> Set<URL> {
|
|
|
|
|
+ let storedPaths =
|
|
|
|
|
+ [session.localAudioPath].compactMap { $0 }
|
|
|
|
|
+ + session.audioChunks.map(\.localFilePath)
|
|
|
|
|
+ + session.events.compactMap(\.localFilePath)
|
|
|
|
|
+ return Set(storedPaths.compactMap { AudioPathHelper.resolveURL(for: $0) })
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private func deduplicatedSessions(_ sessions: [CelestiaSession]) -> [CelestiaSession] {
|
|
|
|
|
+ Array(Dictionary(
|
|
|
|
|
+ sessions.map { ($0.id, $0) },
|
|
|
|
|
+ uniquingKeysWith: preferredLocalSession
|
|
|
|
|
+ ).values)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private func preferredLocalSession(
|
|
|
|
|
+ _ current: CelestiaSession,
|
|
|
|
|
+ _ candidate: CelestiaSession
|
|
|
|
|
+ ) -> CelestiaSession {
|
|
|
|
|
+ if (current.cloudSessionId == nil) != (candidate.cloudSessionId == nil) {
|
|
|
|
|
+ return current.cloudSessionId == nil ? candidate : current
|
|
|
|
|
+ }
|
|
|
|
|
+ if current.serverRevision != candidate.serverRevision {
|
|
|
|
|
+ return current.serverRevision > candidate.serverRevision ? current : candidate
|
|
|
|
|
+ }
|
|
|
|
|
+ return (current.lastSyncedAt ?? .distantPast) >= (candidate.lastSyncedAt ?? .distantPast)
|
|
|
|
|
+ ? current
|
|
|
|
|
+ : candidate
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private func apply(_ remote: RemoteSession, to local: CelestiaSession, userID: String) async {
|
|
|
local.title = remote.title
|
|
local.title = remote.title
|
|
|
local.startTime = remote.startTime
|
|
local.startTime = remote.startTime
|
|
|
local.endTime = remote.endTime
|
|
local.endTime = remote.endTime
|
|
@@ -723,12 +986,19 @@ final class SyncManager: ObservableObject {
|
|
|
local.syncState = .synced
|
|
local.syncState = .synced
|
|
|
local.lastSyncError = nil
|
|
local.lastSyncError = nil
|
|
|
local.events.removeAll()
|
|
local.events.removeAll()
|
|
|
- for remoteEvent in remote.events ?? [] {
|
|
|
|
|
|
|
+ for (index, remoteEvent) in (remote.events ?? []).enumerated() {
|
|
|
|
|
+ if index.isMultiple(of: 50) {
|
|
|
|
|
+ await Task.yield()
|
|
|
|
|
+ }
|
|
|
let id = remoteEvent.clientId.flatMap(UUID.init(uuidString:)) ?? UUID()
|
|
let id = remoteEvent.clientId.flatMap(UUID.init(uuidString:)) ?? UUID()
|
|
|
let event = CelestiaTimelineEvent(id: id, relativeTimeMs: remoteEvent.relativeTimeMs, eventType: remoteEvent.eventType)
|
|
let event = CelestiaTimelineEvent(id: id, relativeTimeMs: remoteEvent.relativeTimeMs, eventType: remoteEvent.eventType)
|
|
|
event.textContent = remoteEvent.textContent
|
|
event.textContent = remoteEvent.textContent
|
|
|
event.voiceStartOffsetMs = remoteEvent.voiceStartOffsetMs
|
|
event.voiceStartOffsetMs = remoteEvent.voiceStartOffsetMs
|
|
|
event.voiceEndOffsetMs = remoteEvent.voiceEndOffsetMs
|
|
event.voiceEndOffsetMs = remoteEvent.voiceEndOffsetMs
|
|
|
|
|
+ event.locationName = remoteEvent.locationName
|
|
|
|
|
+ event.locationAddress = remoteEvent.locationAddress
|
|
|
|
|
+ event.latitude = remoteEvent.latitude
|
|
|
|
|
+ event.longitude = remoteEvent.longitude
|
|
|
local.events.append(event)
|
|
local.events.append(event)
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|