ios-celestia-trace/CelestiaTrace/Services/Network/RemoteNetworkService.swift

import Foundation
import SwiftData
import UniformTypeIdentifiers
import Network
import CryptoKit

// Remote contract: Docs/RemoteAPI.openapi.yaml

struct RemoteAsset: Decodable, Sendable {
    let id: String
    let clientId: String
    let kind: String
    let fileName: String
    let mimeType: String
    let sizeBytes: Int64
    let sha256: String
    let createdAt: Date?
}

struct AssetUploadResult: Sendable {
    let asset: RemoteAsset
    let sessionRevision: Int64?
}

struct StorageQuota: Decodable, Sendable {
    let totalBytes: Int64
    let usedBytes: Int64
    let remainingBytes: Int64
}

struct RemoteEvent: Decodable, Sendable {
    let id: String
    let clientId: String?
    let relativeTimeMs: Int64
    let eventType: String
    let textContent: String?
    let voiceStartOffsetMs: Int64?
    let voiceEndOffsetMs: Int64?
    let locationName: String?
    let locationAddress: String?
    let latitude: Double?
    let longitude: Double?
}

struct RemoteSessionIndex: Decodable, Sendable {
    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, Sendable {
    let id: String
    let clientId: String?
    let title: String
    let startTime: Date
    let endTime: Date?
    let durationMs: Int64
    let revision: Int64
    let deletedAt: Date?
    let updatedAt: Date
    let events: [RemoteEvent]?
    let assets: [RemoteAsset]?
}

private struct SessionUpload: Encodable {
    let clientId: String
    let title: String
    let startTime: Date
    let endTime: Date?
    let durationMs: Int64
    let events: [EventUpload]
    let baseRevision: Int64?
    let deletedEventClientIds: [String]
}

private struct EventUpload: Encodable {
    let clientId: String
    let relativeTimeMs: Int64
    let eventType: String
    let textContent: String?
    let voiceStartOffsetMs: Int64?
    let voiceEndOffsetMs: Int64?
    let locationName: String?
    let locationAddress: String?
    let latitude: Double?
    let longitude: Double?
}

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) {
        self.client = client
    }

    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: SyncSessionSnapshot,
        deletedEventClientIDs: [String]
    ) async throws -> RemoteSession {
        let payload = SessionUpload(
            clientId: session.id.uuidString,
            title: session.title,
            startTime: session.startTime,
            endTime: session.endTime,
            durationMs: session.durationMs,
            events: session.events.map {
                EventUpload(
                    clientId: $0.id.uuidString,
                    relativeTimeMs: $0.relativeTimeMs,
                    eventType: $0.eventType,
                    textContent: $0.textContent,
                    voiceStartOffsetMs: $0.voiceStartOffsetMs,
                    voiceEndOffsetMs: $0.voiceEndOffsetMs,
                    locationName: $0.locationName,
                    locationAddress: $0.locationAddress,
                    latitude: $0.latitude,
                    longitude: $0.longitude
                )
            },
            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 deleteSession(sessionID: String) async throws {
        try await client.requestVoid(
            "sessions/\(sessionID)",
            method: .delete,
            authenticated: true
        )
    }

    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"
        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],
            progress: progress
        )
        return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
    }

    func downloadAsset(sessionID: String, asset: RemoteAsset, destinationURL: URL) async throws {
        try await client.download("sessions/\(sessionID)/assets/\(asset.id)", to: destinationURL)
    }

    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
        )

        for chunkIndex in 0..<upload.totalChunks {
            let offset = UInt64(chunkIndex) * UInt64(upload.chunkSize)
            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(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
            }
        }

        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 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 {
            throw APIError.transport("无法读取待上传文件大小")
        }
        return Int64(size)
    }
}

struct RemoteBoundDevice: Decodable {
    let id: String
    let name: String
    let peripheralUUID: String
}

private struct BindDeviceBody: Encodable {
    let name: String
    let peripheralUUID: String
    let hardwareMAC: String?
    let firmwareVersion: String?
    let batteryLevel: Int?
    let freeStorageMB: Int?
    let totalStorageMB: Int?
}

private struct UpdateDeviceBody: Encodable {
    let name: String?
    let batteryLevel: Int?
    let isConnected: Bool?
    let firmwareVersion: String?
    let freeStorageMB: Int?
    let totalStorageMB: Int?
}

final class DeviceCloudService {
    static let shared = DeviceCloudService()
    private let client: APIClient

    init(client: APIClient = .shared) {
        self.client = client
    }

    func register(_ device: BoundDevice) async throws -> RemoteBoundDevice {
        let body = BindDeviceBody(
            name: device.name,
            peripheralUUID: device.peripheralUUID,
            hardwareMAC: device.hardwareMAC,
            firmwareVersion: device.firmwareVersion,
            batteryLevel: device.batteryLevel,
            freeStorageMB: device.freeStorageMB,
            totalStorageMB: device.totalStorageMB
        )
        return try await client.request("devices", method: .post, body: try JSONEncoder().encode(body), authenticated: true)
    }

    func update(_ device: BoundDevice) async throws -> RemoteBoundDevice {
        guard let cloudID = device.cloudID else { return try await register(device) }
        let body = UpdateDeviceBody(
            name: device.name,
            batteryLevel: device.batteryLevel,
            isConnected: device.isConnected,
            firmwareVersion: device.firmwareVersion,
            freeStorageMB: device.freeStorageMB,
            totalStorageMB: device.totalStorageMB
        )
        return try await client.request("devices/\(cloudID)", method: .put, body: try JSONEncoder().encode(body), authenticated: true)
    }

    func remove(cloudID: String) async throws {
        try await client.requestVoid("devices/\(cloudID)", method: .delete, authenticated: true)
    }
}

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 ? "当前网络" : "网络"
    }
}

struct SyncEventSnapshot: Sendable {
    let id: UUID
    let relativeTimeMs: Int64
    let eventType: String
    let textContent: String?
    let localFilePath: String?
    let voiceStartOffsetMs: Int64?
    let voiceEndOffsetMs: Int64?
    let locationName: String?
    let locationAddress: String?
    let latitude: Double?
    let longitude: Double?
}

struct SyncSessionSnapshot: Sendable {
    let id: UUID
    let title: String
    let startTime: Date
    let endTime: Date?
    let durationMs: Int64
    let localAudioPath: String?
    let cloudSessionId: String?
    let serverRevision: Int64
    let isSynced: Bool
    let events: [SyncEventSnapshot]
}

private struct SyncIndexMatch: Sendable {
    let remote: RemoteSessionIndex
    let localSessionID: UUID
    let title: String
    let needsDetail: Bool
}

private struct SyncIndexPreparation: Sendable {
    let localSessionCount: Int
    let matches: [SyncIndexMatch]
}

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
}

private enum SyncPersistedFailure: Sendable {
    case pending
    case conflict
    case failed(String)
    case remoteDownloadFailed(String)
}

@ModelActor
private actor SyncBackgroundActor {
    func mergeIndex(
        _ remoteSessions: [RemoteSessionIndex],
        userID: String
    ) async throws -> SyncIndexPreparation {
        let localSessions = try modelContext.fetch(FetchDescriptor<CelestiaSession>())
        let scopedLocalSessions = localSessions.filter {
            $0.ownerUserID == nil || $0.ownerUserID == userID
        }
        var byClientID = Dictionary(
            scopedLocalSessions.map { ($0.id.uuidString.lowercased(), $0) },
            uniquingKeysWith: preferredLocalSession
        )
        var byCloudID = Dictionary(
            scopedLocalSessions.compactMap { session in
                session.cloudSessionId.map { ($0, session) }
            },
            uniquingKeysWith: preferredLocalSession
        )
        if byClientID.count < scopedLocalSessions.count {
            DeveloperLogStore.log(
                "现场记录同步",
                "检测到 \(scopedLocalSessions.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 matches: [SyncIndexMatch] = []

        for (index, remote) in remoteSessions.enumerated() {
            if index.isMultiple(of: 20) {
                try Task.checkCancellation()
                await Task.yield()
            }
            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))
                    modelContext.delete(local)
                    DeveloperLogStore.log(
                        "现场记录同步",
                        "云端记录已删除，同步移除本地记录：\(local.title)"
                    )
                    continue
                }
                if !local.isSynced, remote.revision > local.serverRevision {
                    local.syncState = .conflict
                    local.lastSyncError = "本地版本 \(local.serverRevision) 与云端版本 \(remote.revision) 不一致"
                    continue
                }
                if remote.deletedAt == nil {
                    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
                    }
                    matches.append(
                        SyncIndexMatch(
                            remote: remote,
                            localSessionID: local.id,
                            title: local.title,
                            needsDetail: needsDetail
                        )
                    )
                }
            } else if remote.deletedAt == nil {
                let id = clientID.flatMap(UUID.init(uuidString:)) ?? UUID()
                let session = CelestiaSession(id: id, title: remote.title, startTime: remote.startTime)
                session.endTime = remote.endTime
                session.cloudSessionId = remote.id
                session.ownerUserID = userID
                session.serverRevision = remote.revision
                session.isSynced = true
                session.syncState = .syncing
                modelContext.insert(session)
                if let clientID {
                    byClientID[clientID] = session
                }
                byCloudID[remote.id] = session
                matches.append(
                    SyncIndexMatch(
                        remote: remote,
                        localSessionID: session.id,
                        title: session.title,
                        needsDetail: true
                    )
                )
            }
        }

        for (index, local) in localSessions.enumerated() {
            if index.isMultiple(of: 20) {
                try Task.checkCancellation()
                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 SyncIndexPreparation(
            localSessionCount: localSessions.count,
            matches: matches
        )
    }

    func applyRemoteDetail(
        _ remote: RemoteSession,
        to sessionID: UUID,
        userID: String
    ) async throws -> SyncSessionSnapshot {
        guard let local = try session(withID: sessionID) else {
            throw APIError.missingData
        }
        apply(remote, to: local, userID: userID, syncState: .syncing)
        try modelContext.save()
        return snapshot(of: local)
    }

    func finishRemoteDetail(
        sessionID: UUID,
        remoteUpdatedAt: Date,
        downloadedAssets: [AssetDownloadJob]
    ) throws {
        guard let local = try session(withID: sessionID) else {
            throw APIError.missingData
        }
        for job in downloadedAssets {
            let relativePath = AudioPathHelper.relativePath(from: job.destination.path)
            switch job.target {
            case .audio:
                local.localAudioPath = relativePath
            case .photo(let eventID):
                local.events.first(where: { $0.id == eventID })?.localFilePath = relativePath
            }
        }
        local.serverUpdatedAt = remoteUpdatedAt
        local.syncState = .synced
        local.lastSyncError = nil
        local.lastSyncedAt = Date()
        try modelContext.save()
    }

    func prepareUploads(
        requestedSessionIDs: [UUID],
        userID: String
    ) throws -> [SyncSessionSnapshot] {
        let requestedIDs = Set(requestedSessionIDs)
        return try modelContext.fetch(FetchDescriptor<CelestiaSession>())
            .filter {
                requestedIDs.contains($0.id)
                    && ($0.ownerUserID == nil || $0.ownerUserID == userID)
                    && $0.needsCloudSync
                    && $0.syncState != .conflict
            }
            .reduce(into: [UUID: CelestiaSession]()) { result, session in
                if let current = result[session.id] {
                    result[session.id] = preferredLocalSession(current, session)
                } else {
                    result[session.id] = session
                }
            }
            .values
            .sorted { lhs, rhs in
                let lhsIndex = requestedSessionIDs.firstIndex(of: lhs.id) ?? .max
                let rhsIndex = requestedSessionIDs.firstIndex(of: rhs.id) ?? .max
                return lhsIndex < rhsIndex
            }
            .map(snapshot)
    }

    func markUploadStarted(sessionID: UUID, userID: String) throws -> SyncSessionSnapshot {
        guard let local = try session(withID: sessionID) else {
            throw APIError.missingData
        }
        local.ownerUserID = userID
        local.syncState = .syncing
        local.lastSyncError = nil
        try modelContext.save()
        return snapshot(of: local)
    }

    func markUploadSucceeded(sessionID: UUID, remote: RemoteSession) throws -> Int64 {
        guard let local = try session(withID: sessionID) else {
            throw APIError.missingData
        }
        local.cloudSessionId = remote.id
        local.serverRevision = remote.revision
        local.serverUpdatedAt = remote.updatedAt
        local.isSynced = true
        local.syncState = .synced
        local.lastSyncError = nil
        local.lastSyncedAt = Date()
        try modelContext.save()
        return local.serverRevision
    }

    func markFailure(sessionID: UUID, failure: SyncPersistedFailure) throws {
        guard let local = try session(withID: sessionID) else { return }
        switch failure {
        case .pending:
            local.isSynced = false
            local.syncState = .pending
            local.lastSyncError = nil
        case .conflict:
            local.isSynced = false
            local.syncState = .conflict
            local.lastSyncError = "本地版本 \(local.serverRevision) 与云端版本不一致"
        case .failed(let message):
            local.isSynced = false
            local.syncState = .failed
            local.lastSyncError = message
        case .remoteDownloadFailed(let message):
            guard local.hasConfirmedCloudSync, local.syncState == .syncing else { return }
            local.syncState = .failed
            local.lastSyncError = message
        }
        try modelContext.save()
    }

    func conflictMessages(for requestedSessionIDs: [UUID]) throws -> [String] {
        let requestedIDs = Set(requestedSessionIDs)
        return try modelContext.fetch(FetchDescriptor<CelestiaSession>())
            .filter { requestedIDs.contains($0.id) && $0.syncState == .conflict }
            .map { "\($0.title)：本地和云端版本不一致" }
    }

    private func session(withID id: UUID) throws -> CelestiaSession? {
        try modelContext.fetch(FetchDescriptor<CelestiaSession>())
            .first(where: { $0.id == id })
    }

    private func snapshot(of session: CelestiaSession) -> SyncSessionSnapshot {
        SyncSessionSnapshot(
            id: session.id,
            title: session.title,
            startTime: session.startTime,
            endTime: session.endTime,
            durationMs: session.durationMs,
            localAudioPath: session.localAudioPath,
            cloudSessionId: session.cloudSessionId,
            serverRevision: session.serverRevision,
            isSynced: session.isSynced,
            events: session.events.map {
                SyncEventSnapshot(
                    id: $0.id,
                    relativeTimeMs: $0.relativeTimeMs,
                    eventType: $0.eventType,
                    textContent: $0.textContent,
                    localFilePath: $0.localFilePath,
                    voiceStartOffsetMs: $0.voiceStartOffsetMs,
                    voiceEndOffsetMs: $0.voiceEndOffsetMs,
                    locationName: $0.locationName,
                    locationAddress: $0.locationAddress,
                    latitude: $0.latitude,
                    longitude: $0.longitude
                )
            }
        )
    }

    private func apply(
        _ remote: RemoteSession,
        to local: CelestiaSession,
        userID: String,
        syncState: SessionSyncState
    ) {
        local.title = remote.title
        local.startTime = remote.startTime
        local.endTime = remote.endTime
        local.cloudSessionId = remote.id
        local.ownerUserID = userID
        local.serverRevision = remote.revision
        local.isSynced = true
        local.syncState = syncState
        local.lastSyncError = nil
        local.events.removeAll()
        for remoteEvent in remote.events ?? [] {
            let id = remoteEvent.clientId.flatMap(UUID.init(uuidString:)) ?? UUID()
            let event = CelestiaTimelineEvent(
                id: id,
                relativeTimeMs: remoteEvent.relativeTimeMs,
                eventType: remoteEvent.eventType
            )
            event.textContent = remoteEvent.textContent
            event.voiceStartOffsetMs = remoteEvent.voiceStartOffsetMs
            event.voiceEndOffsetMs = remoteEvent.voiceEndOffsetMs
            event.locationName = remoteEvent.locationName
            event.locationAddress = remoteEvent.locationAddress
            event.latitude = remoteEvent.latitude
            event.longitude = remoteEvent.longitude
            local.events.append(event)
        }
    }

    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 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
    }
}

@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?
    @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>?
    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 sessionIDs: [UUID]
        let modelContainer: ModelContainer
        let userID: String
    }

    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 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,
        userID: String
    ) async -> Bool {
        let sessionIDs = deduplicatedSessions(sessions).map(\.id)
        guard !isSyncing else {
            queuedSyncRequest = SyncRequest(
                sessionIDs: sessionIDs,
                modelContainer: modelContext.container,
                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(
                sessionIDs: sessionIDs,
                modelContainer: modelContext.container,
                userID: userID
            )
        }
        activeSyncTask = task
        let result = await withTaskCancellationHandler {
            await task.value
        } onCancel: {
            task.cancel()
        }
        activeSyncTask = nil
        isSyncing = false
        activeSessionID = nil
        DeveloperLogStore.log(
            "现场记录同步",
            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(
                    sessionIDs: queuedRequest.sessionIDs,
                    modelContainer: queuedRequest.modelContainer,
                    userID: queuedRequest.userID
                )
            }
        }
        return result
    }

    private func sync(
        sessionIDs: [UUID],
        modelContainer: ModelContainer,
        userID: String
    ) async -> Bool {
        guard !isSyncing else {
            queuedSyncRequest = SyncRequest(
                sessionIDs: sessionIDs,
                modelContainer: modelContainer,
                userID: userID
            )
            return false
        }
        isSyncing = true
        let task = Task { @MainActor [self] in
            await performSync(
                sessionIDs: sessionIDs,
                modelContainer: modelContainer,
                userID: userID
            )
        }
        activeSyncTask = task
        let result = await withTaskCancellationHandler {
            await task.value
        } onCancel: {
            task.cancel()
        }
        activeSyncTask = nil
        isSyncing = false
        activeSessionID = nil
        if let queuedRequest = queuedSyncRequest {
            queuedSyncRequest = nil
            Task { @MainActor [weak self] in
                guard AuthManager.shared.currentUser?.id == queuedRequest.userID else { return }
                _ = await self?.sync(
                    sessionIDs: queuedRequest.sessionIDs,
                    modelContainer: queuedRequest.modelContainer,
                    userID: queuedRequest.userID
                )
            }
        }
        return result
    }

    func pauseSync(sessionID: UUID) {
        guard isSyncing, activeSessionID == sessionID else { return }
        queuedSyncRequest = nil
        activeSyncTask?.cancel()
    }

    private func performSync(
        sessionIDs: [UUID],
        modelContainer: ModelContainer,
        userID: String
    ) async -> Bool {
        lastErrorMessage = nil
        completedCount = 0
        activeSessionID = nil
        syncProgress = 0
        let worker = SyncBackgroundActor(modelContainer: modelContainer)
        totalCount = sessionIDs.count
        DeveloperLogStore.log("现场记录同步", "待检查 \(sessionIDs.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 = sessionIDs.first
        var pendingRemoteDownloadIDs: [UUID] = []
        do {
            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()
            let mergeStartedAt = diagnosticsNow
            let indexPreparation = try await worker.mergeIndex(
                remoteIndex,
                userID: userID
            )
            diagnosticsLog(
                "阶段 2/5 完成：后台读取并合并本地记录=\(indexPreparation.localSessionCount)，索引匹配=\(indexPreparation.matches.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 = indexPreparation.matches.filter(\.needsDetail)
            pendingRemoteDownloadIDs = sessionsRequiringDetail.map(\.localSessionID)
            activeSessionID = sessionsRequiringDetail.first?.localSessionID
            await Task.yield()

            let detailStageStartedAt = diagnosticsNow
            diagnosticsLog("阶段 3/5 开始：需要拉取详情=\(sessionsRequiringDetail.count)")
            for (detailIndex, match) in sessionsRequiringDetail.enumerated() {
                let recordLabel = diagnosticsRecordLabel(
                    sessionID: match.localSessionID,
                    position: detailIndex + 1,
                    total: sessionsRequiringDetail.count
                )
                let recordStartedAt = diagnosticsNow
                do {
                    activeSessionID = match.localSessionID
                    let detailStartedAt = diagnosticsNow
                    let remote = try await service.fetchSession(sessionID: match.remote.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
                    let snapshot = try await worker.applyRemoteDetail(
                        remote,
                        to: match.localSessionID,
                        userID: userID
                    )
                    diagnosticsLog(
                        "\(recordLabel) 详情由后台 Actor 写入本地：耗时 \(diagnosticsElapsed(since: applyStartedAt))"
                    )
                    await Task.yield()

                    let assetStartedAt = diagnosticsNow
                    let downloadedAssets = try await downloadMissingAssets(
                        for: snapshot,
                        remote: remote
                    )
                    try Task.checkCancellation()
                    let finalSaveStartedAt = diagnosticsNow
                    try await worker.finishRemoteDetail(
                        sessionID: match.localSessionID,
                        remoteUpdatedAt: remote.updatedAt,
                        downloadedAssets: downloadedAssets
                    )
                    for job in downloadedAssets {
                        if case .audio = job.target {
                            Task.detached(priority: .utility) {
                                await SilenceDetector.warmCache(for: job.destination)
                            }
                        }
                    }
                    diagnosticsLog(
                        "\(recordLabel) 资源阶段：\(downloadedAssets.isEmpty ? "无需下载" : "下载 \(downloadedAssets.count) 个文件")，耗时 \(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
                    )
                    try? await worker.markFailure(
                        sessionID: match.localSessionID,
                        failure: .remoteDownloadFailed("已暂停从云端下载")
                    )
                    throw CancellationError()
                } catch {
                    try? await worker.markFailure(
                        sessionID: match.localSessionID,
                        failure: .remoteDownloadFailed("从云端下载失败：\(error.localizedDescription)")
                    )
                    diagnosticsLog(
                        "\(recordLabel) 云端拉取失败：耗时 \(diagnosticsElapsed(since: recordStartedAt))；\(error.localizedDescription)",
                        level: .error
                    )
                    failures.append("\(match.title)：\(error.localizedDescription)")
                    DeveloperLogStore.log(
                        "现场记录同步",
                        "\(match.title) 从云端下载失败：\(error.localizedDescription)",
                        level: .error
                    )
                }
            }
            diagnosticsLog(
                "阶段 3/5 完成：详情记录=\(sessionsRequiringDetail.count)，总耗时 \(diagnosticsElapsed(since: detailStageStartedAt))"
            )

            let eligible = try await worker.prepareUploads(
                requestedSessionIDs: sessionIDs,
                userID: userID
            )
            let eligibleCount = max(eligible.count, 1)
            totalCount = eligible.count
            let uploadStageStartedAt = diagnosticsNow
            diagnosticsLog("阶段 4/5 开始：需要上传=\(eligible.count)")
            for (index, preparedSession) in eligible.enumerated() {
                try Task.checkCancellation()
                await Task.yield()
                let session = try await worker.markUploadStarted(
                    sessionID: preparedSession.id,
                    userID: userID
                )
                activeSessionID = session.id
                let sessionStartProgress = Double(index) / Double(eligibleCount)
                let sessionProgressSpan = 1.0 / Double(eligibleCount)
                syncProgress = sessionStartProgress
                DeveloperLogStore.log(
                    "现场记录同步",
                    "正在同步 \(index + 1)/\(eligible.count)：\(session.title)"
                )
                let recordLabel = diagnosticsRecordLabel(
                    sessionID: session.id,
                    position: index + 1,
                    total: eligible.count
                )
                let recordStartedAt = diagnosticsNow
                do {
                    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
                    try await uploadLocalAssets(
                        for: session,
                        cloudSessionID: remote.id,
                        remote: remote
                    ) { [weak self] assetProgress in
                        Task { @MainActor [weak self] in
                            self?.syncProgress = sessionStartProgress
                                + sessionProgressSpan * (0.12 + assetProgress * 0.83)
                        }
                    }
                    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
                    let finalRevision = try await worker.markUploadSucceeded(
                        sessionID: session.id,
                        remote: finalRemote
                    )
                    completedCount += 1
                    syncProgress = sessionStartProgress + sessionProgressSpan
                    DeveloperLogStore.log(
                        "现场记录同步",
                        "\(session.title) 同步成功，云端版本 v\(finalRevision)",
                        level: .success
                    )
                    diagnosticsLog(
                        "\(recordLabel) 上传链路完成：总耗时 \(diagnosticsElapsed(since: recordStartedAt))",
                        level: .success
                    )
                } catch where Task.isCancelled {
                    diagnosticsLog(
                        "\(recordLabel) 上传链路取消：已运行 \(diagnosticsElapsed(since: recordStartedAt))",
                        level: .warning
                    )
                    try? await worker.markFailure(sessionID: session.id, failure: .pending)
                    syncProgress = 0
                    return false
                } catch APIError.server(let code, _) where code == 409 {
                    diagnosticsLog(
                        "\(recordLabel) 上传链路冲突：耗时 \(diagnosticsElapsed(since: recordStartedAt))",
                        level: .error
                    )
                    try? await worker.markFailure(sessionID: session.id, failure: .conflict)
                    DeveloperLogStore.log(
                        "现场记录同步",
                        "\(session.title) 发生版本冲突",
                        level: .error
                    )
                } catch {
                    diagnosticsLog(
                        "\(recordLabel) 上传链路失败：耗时 \(diagnosticsElapsed(since: recordStartedAt))；\(error.localizedDescription)",
                        level: .error
                    )
                    try? await worker.markFailure(
                        sessionID: session.id,
                        failure: .failed(error.localizedDescription)
                    )
                    failures.append("\(session.title)：\(error.localizedDescription)")
                    DeveloperLogStore.log(
                        "现场记录同步",
                        "\(session.title) 同步失败：\(error.localizedDescription)",
                        level: .error
                    )
                }
            }
            diagnosticsLog(
                "阶段 4/5 完成：上传记录=\(eligible.count)，总耗时 \(diagnosticsElapsed(since: uploadStageStartedAt))"
            )
            failures.append(contentsOf: try await worker.conflictMessages(for: sessionIDs))
            guard failures.isEmpty else {
                lastErrorMessage = failures.joined(separator: "\n")
                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 sessionID in pendingRemoteDownloadIDs {
                try? await worker.markFailure(
                    sessionID: sessionID,
                    failure: .remoteDownloadFailed("已暂停从云端下载")
                )
            }
            syncProgress = 0
            return false
        } catch {
            for sessionID in pendingRemoteDownloadIDs {
                try? await worker.markFailure(
                    sessionID: sessionID,
                    failure: .remoteDownloadFailed("从云端下载中断：\(error.localizedDescription)")
                )
            }
            lastErrorMessage = error.localizedDescription
            DeveloperLogStore.log("现场记录同步", "同步中断：\(error.localizedDescription)", level: .error)
            return false
        }
    }

    private func uploadLocalAssets(
        for session: SyncSessionSnapshot,
        cloudSessionID: String,
        remote: RemoteSession,
        progress: @escaping @Sendable (Double) -> Void
    ) async throws {
        let stageStartedAt = diagnosticsNow
        let recordLabel = diagnosticsRecordLabel(sessionID: session.id)
        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,
                purpose: "\(recordLabel) 音频上传校验"
            )
            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" {
            guard let path = event.localFilePath else { continue }
            guard let url = AudioPathHelper.resolveURL(for: path) else {
                throw APIError.transport("照片文件不存在：\(path)")
            }
            let clientID = "\(event.id.uuidString)-photo"
            if !existingIDs.contains(clientID) {
                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 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,
                    remainingBytes: quota.remainingBytes
                )
            }
        }
        var completedBytes: Int64 = 0
        progress(uploads.isEmpty ? 1 : 0)
        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
            )
            _ = try await service.uploadAsset(
                sessionID: cloudSessionID,
                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))
            }
            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
            )
        }
    }

    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, 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() }
            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()
        }
        let digest = try await withTaskCancellationHandler {
            try await task.value
        } onCancel: {
            task.cancel()
        }
        diagnosticsLog("\(purpose) 完成：耗时 \(diagnosticsElapsed(since: startedAt))")
        return digest
    }

    private func downloadMissingAssets(
        for session: SyncSessionSnapshot,
        remote: RemoteSession
    ) async throws -> [AssetDownloadJob] {
        let assets = remote.assets ?? []
        var jobs: [AssetDownloadJob] = []
        var downloadedAssets: [AssetDownloadJob] = []
        let recordLabel = diagnosticsRecordLabel(sessionID: session.id)
        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: String?
            if let localURL {
                localHash = try? await fileSHA256(
                    of: localURL,
                    purpose: "\(recordLabel) 音频本地校验"
                )
            } else {
                localHash = nil
            }
            let needsDownload = localURL == nil
                || (session.isSynced
                    && localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)
            if needsDownload {
                let destination = downloadDestination(for: audio)
                jobs.append(
                    AssetDownloadJob(
                        sessionID: remote.id,
                        asset: audio,
                        destination: destination,
                        target: .audio,
                        displayName: "音频"
                    )
                )
            }
        }

        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" {
            guard event.localFilePath == nil || AudioPathHelper.resolveURL(for: event.localFilePath) == nil else { continue }
            let key = "\(event.id.uuidString)-photo".lowercased()
            guard let asset = photoAssets[key] else { continue }
            let destination = downloadDestination(for: asset)
            jobs.append(
                AssetDownloadJob(
                    sessionID: remote.id,
                    asset: asset,
                    destination: destination,
                    target: .photo(eventID: event.id),
                    displayName: "照片"
                )
            )
        }

        guard !jobs.isEmpty else { return [] }
        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])

            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
                        downloadedAssets.append(completedJob)
                    }
                } 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 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(
        sessionID: UUID,
        position: Int? = nil,
        total: Int? = nil
    ) -> String {
        let shortID = sessionID.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 {
        let documents = FileManager.default.urls(for: .documentDirectory, in: .userDomainMask)[0]
        let fileExtension = (asset.fileName as NSString).pathExtension
        let suffix = fileExtension.isEmpty ? "" : ".\(fileExtension.lowercased())"
        return documents.appendingPathComponent("cloud_\(asset.id)\(suffix)")
    }

    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
    }

}