import Foundation public struct LogCenterService: Sendable { private struct ProbeOutcome: Sendable { var status: ConnectionStatus var outcome: ConnectionOutcome var state: String var errorCode: String? var httpStatus: Int? var durationMs: Int var message: String var version: String? var platform: String? } private struct InstanceCollection: Sendable { var runtime: [RuntimeLogEntry] = [] var connections: [ConnectionLogEntry] = [] var diagnostic: InstanceDiagnostic? } private let client: any SimAdminClientProtocol private let sessions: InstanceSessionStore private let credentials: (any CredentialStore)? private let store: any LogStore public init( client: any SimAdminClientProtocol, sessions: InstanceSessionStore, credentials: (any CredentialStore)? = nil, store: any LogStore ) { self.client = client self.sessions = sessions self.credentials = credentials self.store = store } public func refresh(_ instances: [Instance]) async throws -> LogSnapshot { var collected: [InstanceCollection] = [] await withTaskGroup(of: InstanceCollection.self) { group in for instance in instances { group.addTask { await self.collect(instance) } } for await value in group { collected.append(value) } } collected.sort { ($0.diagnostic?.name ?? "") < ($1.diagnostic?.name ?? "") } let runtime = collected.flatMap(\.runtime).sorted { $0.occurredAt > $1.occurredAt } let connections = collected.flatMap(\.connections).sorted { $0.observedAt > $1.observedAt } let diagnostics = collected.compactMap(\.diagnostic) try await store.append(runtime: runtime, connections: connections) var counts: [RuntimeLogSource: Int] = [:] for source in RuntimeLogSource.allCases { counts[source] = runtime.filter { $0.source == source }.count } return LogSnapshot( observedAt: Date(), runtime: runtime, counts: counts, connections: connections, diagnostics: diagnostics ) } private func collect(_ instance: Instance) async -> InstanceCollection { var result = InstanceCollection() let probe = await probe(instance) var modules: [String: Result] = [ "connection": .success(true), "device": await moduleResult(instance, .device) ] if probe.outcome == .failed { modules["connection"] = .failure(UpstreamError.notSupported) modules["device"] = .failure(UpstreamError.notSupported) } result.connections.append(ConnectionLogEntry( id: "probe-\(instance.id.uuidString)-\(Int(Date().timeIntervalSince1970 * 1000))", instanceID: instance.id, instanceName: instance.name, outcome: probe.outcome, state: probe.state, errorCode: probe.errorCode, httpStatus: probe.httpStatus, durationMs: probe.durationMs, observedAt: Date() )) if probe.outcome == .failed || probe.outcome == .stale { result.runtime.append(RuntimeLogEntry( id: "connection-\(instance.id.uuidString)-\(Int(Date().timeIntervalSince1970 * 1000))", source: .delivery, level: probe.outcome == .failed ? .error : .warning, message: probe.message, instanceID: instance.id, instanceName: instance.name, occurredAt: Date(), code: probe.errorCode, durationMs: probe.durationMs )) } let deliveryEntries = probe.outcome == .failed ? [] : await deliveryLogs(instance) result.runtime.append(contentsOf: deliveryEntries) if probe.outcome == .failed { modules["delivery"] = .failure(UpstreamError.notSupported) } else if deliveryEntries.contains(where: { $0.code == "UNAUTHORIZED" }) { modules["delivery"] = .failure(UpstreamError.unauthorized) } else if deliveryEntries.contains(where: { $0.code == "UPSTREAM" }) { modules["delivery"] = .failure(UpstreamError.invalidResponse) } else { modules["delivery"] = .success(true) } result.diagnostic = diagnostic( instance, probe: probe, modules: modules, connections: result.connections ) return result } private func probe(_ instance: Instance) async -> ProbeOutcome { let startedAt = Date() do { let response = try await client.send(.health, baseURL: instance.origin, cookie: nil) let duration = elapsed(startedAt) let root = (try? JSONSerialization.jsonObject(with: response.data)) as? [String: Any] ?? [:] return ProbeOutcome( status: .online, outcome: .success, state: "online", errorCode: nil, httpStatus: response.status, durationMs: duration, message: "节点连接成功", version: Self.text(root["version"]), platform: Self.text(root["platform"]) ) } catch UpstreamError.unauthorized, UpstreamError.forbidden { return ProbeOutcome( status: .authRequired, outcome: .stale, state: "auth_required", errorCode: "UNAUTHORIZED", httpStatus: 401, durationMs: elapsed(startedAt), message: "节点需要登录", version: nil, platform: nil ) } catch UpstreamError.timeout { return ProbeOutcome( status: .offline, outcome: .failed, state: "unreachable", errorCode: "TIMEOUT", httpStatus: nil, durationMs: elapsed(startedAt), message: "节点连接超时", version: nil, platform: nil ) } catch { return ProbeOutcome( status: .degraded, outcome: .failed, state: "unreachable", errorCode: "TRANSPORT", httpStatus: nil, durationMs: elapsed(startedAt), message: "节点连接失败", version: nil, platform: nil ) } } private func moduleResult( _ instance: Instance, _ route: UpstreamRoute ) async -> Result { do { _ = try await request(route, instance) return .success(true) } catch { return .failure(error) } } private func deliveryLogs(_ instance: Instance) async -> [RuntimeLogEntry] { do { let response = try await request(.notificationLogs(limit: 50, offset: 0), instance) let entries = Self.deliveryEntries(response, instance) if entries.isEmpty { throw UpstreamError.invalidResponse } return entries } catch UpstreamError.unauthorized, UpstreamError.forbidden { return [RuntimeLogEntry( id: "delivery-\(instance.id.uuidString)-auth", source: .delivery, level: .warning, message: "通知日志未同步:需要登录", instanceID: instance.id, instanceName: instance.name, occurredAt: Date(), code: "UNAUTHORIZED", durationMs: nil )] } catch { return [RuntimeLogEntry( id: "delivery-\(instance.id.uuidString)-failed", source: .delivery, level: .warning, message: "通知日志未同步:接口异常", instanceID: instance.id, instanceName: instance.name, occurredAt: Date(), code: "UPSTREAM", durationMs: nil )] } } private func diagnostic( _ instance: Instance, probe: ProbeOutcome, modules: [String: Result], connections: [ConnectionLogEntry] ) -> InstanceDiagnostic { let featureStatus = modules .sorted { $0.key < $1.key } .map { key, result -> FeatureStatus in let name = key == "connection" ? "连接" : key == "device" ? "设备" : "通知" switch result { case .success: return FeatureStatus(name: name, state: .ready) case .failure(UpstreamError.unauthorized), .failure(UpstreamError.forbidden): return FeatureStatus(name: name, state: .requiresLogin) case .failure: return FeatureStatus(name: name, state: .unavailable) } } let success = connections.filter { $0.outcome == .success }.count let failed = connections.filter { $0.outcome == .failed }.count let average = connections.isEmpty ? 0 : connections.map(\.durationMs).reduce(0, +) / connections.count return InstanceDiagnostic( id: instance.id, name: instance.name, origin: instance.origin.absoluteString, tags: instance.tags, state: probe.status, healthStatus: Self.healthLabel(probe), version: probe.version, platform: probe.platform, observedAt: Date(), featureStatus: featureStatus, connection: ConnectionSummary( total: connections.count, success: success, failed: failed, averageDurationMs: average, availabilityPercent: connections.isEmpty ? 0 : success * 100 / connections.count ) ) } static func deliveryEntries( _ response: UpstreamResponse, _ instance: Instance ) -> [RuntimeLogEntry] { guard let root = try? JSONSerialization.jsonObject(with: response.data) as? [String: Any], let data = root["data"] as? [String: Any], let raw = data["logs"] as? [[String: Any]] else { return [] } return raw.compactMap { item in guard let id = Self.text(item["id"]), let status = Self.text(item["status"]), let eventType = Self.text(item["event_type"]) else { return nil } let level: RuntimeLogLevel = status == "failed" ? .error : .info return RuntimeLogEntry( id: "delivery-\(instance.id.uuidString)-\(id)", source: .delivery, level: level, message: Self.text(item["message"]) ?? "\(eventType) · \(status)", eventType: eventType, instanceID: instance.id, instanceName: instance.name, occurredAt: Self.date(item["created_at"]), code: status, durationMs: nil ) } } private func request(_ route: UpstreamRoute, _ instance: Instance) async throws -> UpstreamResponse { var cookie = await sessions.cookie(for: instance.id, origin: instance.origin) do { return try await client.send(route, baseURL: instance.origin, cookie: cookie) } catch UpstreamError.unauthorized, UpstreamError.forbidden { cookie = try await login(instance) return try await client.send(route, baseURL: instance.origin, cookie: cookie) } } private func login(_ instance: Instance) async throws -> String { guard let credentials, let password = try credentials.get(instanceID: instance.id), !password.isEmpty else { throw UpstreamError.unauthorized } let service = InstanceSessionService(store: sessions, client: client) let result = try await service.login(instance, password: password) guard result.authenticated, let cookie = await sessions.cookie(for: instance.id, origin: instance.origin) else { throw UpstreamError.unauthorized } return cookie } private func elapsed(_ startedAt: Date) -> Int { max(0, Int(Date().timeIntervalSince(startedAt) * 1000)) } private static func healthLabel(_ probe: ProbeOutcome) -> String { switch probe.state { case "online", "UP", "ok": return "在线" case "auth_required": return "需登录" case "unreachable": return probe.errorCode == "TIMEOUT" ? "超时" : "不可达" default: return probe.state } } private static func text(_ value: Any?) -> String? { guard let value = value as? String, !value.isEmpty, value.count <= 512 else { return nil } return value } private static func date(_ value: Any?) -> Date { guard let text = value as? String else { return Date() } let formatter = ISO8601DateFormatter() formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds] if let parsed = formatter.date(from: text) { return parsed } formatter.formatOptions = [.withInternetDateTime] return formatter.date(from: text) ?? Date() } }