Sesame-Server/Sources/App/configure.swift

import Vapor
import Clairvoyant
import ClairvoyantVapor
import ClairvoyantBinaryCodable

var deviceManager: DeviceManager!

private var provider: VaporMetricProvider!

enum ServerError: Error {
    case invalidAuthenticationFileContent
    case invalidAuthenticationToken
}

private let dateFormatter: DateFormatter = {
    let df = DateFormatter()
    df.dateStyle = .short
    df.timeStyle = .short
    return df
}()

// configures your application
public func configure(_ app: Application) async throws {
    let storageFolder = URL(fileURLWithPath: app.directory.resourcesDirectory)
    let logFolder = storageFolder.appendingPathComponent("logs")
    
    let monitor = MetricObserver(logFileFolder: logFolder, logMetricId: "sesame.log")
    MetricObserver.standard = monitor

    let status = try await Metric<ServerStatus>("sesame.status")
    _ = try await status.update(.initializing)

    let configUrl = storageFolder.appendingPathComponent("config.json")
    let config = try Config(loadFrom: configUrl)

    app.http.server.configuration.port = config.port

    let keyFile = storageFolder.appendingPathComponent(config.keyFileName)

    let (deviceKey, remoteKey) = try loadKeys(at: keyFile)
    deviceManager = await DeviceManager(
        deviceKey: deviceKey,
        remoteKey: remoteKey,
        deviceTimeout: config.deviceTimeout)
    
    try routes(app)

    provider = .init(observer: monitor, accessManager: config.authenticationTokens)
    provider.registerRoutes(app)
    
    // Gracefully shut down by closing potentially open socket
    DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + .seconds(5)) {
        _ = app.server.onShutdown.always { _ in
            deviceManager.removeDeviceConnection()
        }
    }

    _ = try await status.update(.nominal)
    print("[\(dateFormatter.string(from: Date()))] Server started")
}

private func loadKeys(at url: URL) throws -> (deviceKey: Data, remoteKey: Data) {
    let authContent: [Data] = try String(contentsOf: url)
        .trimmingCharacters(in: .whitespacesAndNewlines)
        .components(separatedBy: "\n")
        .map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
        .map {
            guard let key = Data(fromHexEncodedString: $0) else {
                throw ServerError.invalidAuthenticationToken
            }
            guard key.count == SHA256.byteCount else {
                throw ServerError.invalidAuthenticationToken
            }
            return key
        }
    guard authContent.count == 2 else {
        throw ServerError.invalidAuthenticationFileContent
    }
    return (deviceKey: authContent[0], remoteKey: authContent[1])
}

func log(_ message: String) {
    guard let observer = MetricObserver.standard else {
        print(message)
        return
    }
    observer.log(message)
}

import CBORCoding

public func migrate(folder: URL) throws {
    try migrateMetric("sesame.log", containing: String.self, in: folder)
    try migrateMetric("sesame.status", containing: ServerStatus.self, in: folder)
    try migrateMetric("sesame.connected", containing: Bool.self, in: folder)
    try migrateMetric("sesame.messages", containing: Int.self, in: folder)
}

private func migrateMetric<T>(_ id: String, containing type: T.Type, in folder: URL) throws where T: MetricValue {
    print("Processing metric \(id)")
    let file = id.hashed()
    let url = folder.appendingPathComponent(file)

    let files = try FileManager.default.contentsOfDirectory(at: url, includingPropertiesForKeys: nil)
        .filter { Int($0.lastPathComponent) != nil }
    print("Found \(files.count) files for \(id)")

    let all: [Timestamped<T>] = try files.map(readElements(from:))
        .reduce([], +)
        .sorted { $0.timestamp < $1.timestamp }

    print("Found \(all.count) items for \(id)")

    try FileManager.default.removeItem(at: url)

    print("Removed log folder")

    // TODO: Write values back to disk
    let observer = MetricObserver(logFileFolder: folder, logMetricId: "sesame.migration")
    let metric: Metric<T> = observer.addMetric(id: id)
    private let semaphore = DispatchSemaphore(value: 0)
    Task {
        try await metric.update(all)
        print("Saved all values for metric \(id)")
        semaphore.signal()
    }
    semaphore.wait()
    print("Finished metric \(id)")
}

private func readElements<T>(from url: URL) throws -> [Timestamped<T>] where T: MetricValue {
    let data = try Data(contentsOf: url)
    let file = url.lastPathComponent
    print("File \(file): Loaded \(data.count) bytes")



    let decoder = CBORDecoder()
    let timestampLength = 9
    let byteCountLength = 2

    var result: [Timestamped<T>] = []
    var currentIndex = data.startIndex
    var skippedValues = 0
    while currentIndex < data.endIndex {
        let startIndexOfTimestamp = currentIndex + byteCountLength
        guard startIndexOfTimestamp <= data.endIndex else {
            print("File \(file): Only \(data.endIndex - currentIndex) bytes, needed \(byteCountLength) for byte count")
            throw MetricError.logFileCorrupted
        }
        guard let byteCount = UInt16(fromData: data[currentIndex..<startIndexOfTimestamp]) else {
            print("File \(file): Invalid byte count")
            throw MetricError.logFileCorrupted
        }
        let nextIndex = startIndexOfTimestamp + Int(byteCount)
        guard nextIndex <= data.endIndex else {
            print("File \(file): Needed \(byteCountLength + Int(byteCount)) for timestamped value, has \(data.endIndex - startIndexOfTimestamp)")
            throw MetricError.logFileCorrupted
        }
        guard byteCount >= timestampLength else {
            print("File \(file): Only \(byteCount) bytes, needed \(timestampLength) for timestamp")
            throw MetricError.logFileCorrupted
        }
        let timestampData = data[startIndexOfTimestamp..<startIndexOfTimestamp+timestampLength]
        let timestamp = try decoder.decode(Double.self, from: timestampData)
        let date = Date(timeIntervalSince1970: timestamp)
        let elementData = data[startIndexOfTimestamp+timestampLength..<nextIndex]
        do {
            let element: T = try decoder.decode(from: elementData)
            result.append(.init(value: element, timestamp: date))
        } catch {
            skippedValues += 1
        }
        currentIndex = nextIndex
        if result.count % 100 == 1 {
            print("File \(file): \(result.count) entries loaded (\(currentIndex)/\(data.endIndex) bytes)")
        }
    }
    print("Loaded \(result.count) data points (\(skippedValues) skipped)")
    return result
}

extension UInt16 {

    func toData() -> Data {
        Data([UInt8(self >> 8 & 0xFF), UInt8(self & 0xFF)])
    }

    init?<T: DataProtocol>(fromData data: T) {
        guard data.count == 2 else {
            return nil
        }
        let bytes = Array(data)
        self = UInt16(UInt32(bytes[0]) << 8 | UInt32(bytes[1]))
    }
}