140 lines
4.5 KiB
Swift
140 lines
4.5 KiB
Swift
import Foundation
|
|
|
|
private let encoder: JSONEncoder = {
|
|
let e = JSONEncoder()
|
|
e.dateEncodingStrategy = .secondsSince1970
|
|
return e
|
|
}()
|
|
|
|
private let decoder: JSONDecoder = {
|
|
let d = JSONDecoder()
|
|
d.dateDecodingStrategy = .secondsSince1970
|
|
return d
|
|
}()
|
|
|
|
final class EventCache: @unchecked Sendable {
|
|
private let dir: URL
|
|
private let queue = DispatchQueue(label: "run.fc.byre.events")
|
|
private var lastSeqs: [String: Int64] = [:]
|
|
private var generation = 0
|
|
|
|
init(host: String) {
|
|
dir = URL.applicationSupportDirectory.appending(path: "run.fc.byre.mac/events/\(host)", directoryHint: .isDirectory)
|
|
}
|
|
|
|
func open(schema: String) {
|
|
queue.async {
|
|
self.generation += 1
|
|
let fm = FileManager.default
|
|
let marker = self.dir.appending(path: "schema")
|
|
if (try? String(contentsOf: marker, encoding: .utf8)) == schema {
|
|
return
|
|
}
|
|
self.lastSeqs = [:]
|
|
try? fm.removeItem(at: self.dir)
|
|
do {
|
|
try fm.createDirectory(at: self.dir, withIntermediateDirectories: true)
|
|
try schema.write(to: marker, atomically: true, encoding: .utf8)
|
|
} catch {
|
|
log.error("open event cache: \(errorText(error), privacy: .public)")
|
|
}
|
|
}
|
|
}
|
|
|
|
func load(_ taskId: String) async -> (events: [ByreEvent], generation: Int) {
|
|
await withCheckedContinuation { c in
|
|
queue.async {
|
|
let events = self.read(taskId)
|
|
self.lastSeqs[taskId] = events.last?.seq ?? 0
|
|
c.resume(returning: (events, self.generation))
|
|
}
|
|
}
|
|
}
|
|
|
|
func append(_ taskId: String, _ events: [ByreEvent], generation: Int) {
|
|
queue.async {
|
|
guard generation == self.generation, let last = self.lastSeqs[taskId] else {
|
|
return
|
|
}
|
|
let fresh = events.filter { $0.seq > last }
|
|
guard let newest = fresh.last else {
|
|
return
|
|
}
|
|
do {
|
|
try self.write(taskId, fresh)
|
|
self.lastSeqs[taskId] = newest.seq
|
|
} catch {
|
|
log.error("cache events of \(taskId, privacy: .public): \(errorText(error), privacy: .public)")
|
|
self.forget(taskId)
|
|
}
|
|
}
|
|
}
|
|
|
|
func drop(_ taskId: String) {
|
|
queue.async {
|
|
self.forget(taskId)
|
|
}
|
|
}
|
|
|
|
func keep(_ taskIds: Set<String>) {
|
|
queue.async {
|
|
let files = (try? FileManager.default.contentsOfDirectory(at: self.dir, includingPropertiesForKeys: nil)) ?? []
|
|
for url in files where url.pathExtension == "ndjson" {
|
|
let taskId = url.deletingPathExtension().lastPathComponent
|
|
if !taskIds.contains(taskId) {
|
|
self.forget(taskId)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private func file(_ taskId: String) -> URL {
|
|
dir.appending(path: "\(taskId).ndjson")
|
|
}
|
|
|
|
private func forget(_ taskId: String) {
|
|
lastSeqs[taskId] = nil
|
|
try? FileManager.default.removeItem(at: file(taskId))
|
|
}
|
|
|
|
private func read(_ taskId: String) -> [ByreEvent] {
|
|
let url = file(taskId)
|
|
guard let data = try? Data(contentsOf: url) else {
|
|
return []
|
|
}
|
|
var events: [ByreEvent] = []
|
|
var kept = 0
|
|
for line in data.split(separator: 0x0a, omittingEmptySubsequences: false) where !line.isEmpty {
|
|
guard let ev = try? decoder.decode(ByreEvent.self, from: line) else {
|
|
break
|
|
}
|
|
events.append(ev)
|
|
kept = min(line.endIndex + 1, data.endIndex) - data.startIndex
|
|
}
|
|
if kept < data.count, let h = try? FileHandle(forWritingTo: url) {
|
|
try? h.truncate(atOffset: UInt64(kept))
|
|
try? h.close()
|
|
}
|
|
return events
|
|
}
|
|
|
|
private func write(_ taskId: String, _ events: [ByreEvent]) throws {
|
|
var data = Data()
|
|
for ev in events {
|
|
data.append(try encoder.encode(ev))
|
|
data.append(0x0a)
|
|
}
|
|
let url = file(taskId)
|
|
let fm = FileManager.default
|
|
if !fm.fileExists(atPath: url.path), !fm.createFile(atPath: url.path, contents: nil) {
|
|
throw ApiError(status: 0, message: "create \(url.path)")
|
|
}
|
|
let h = try FileHandle(forWritingTo: url)
|
|
defer {
|
|
try? h.close()
|
|
}
|
|
try h.seekToEnd()
|
|
try h.write(contentsOf: data)
|
|
}
|
|
}
|