byre/mac/Sources/Byre/ClientRun.swift

import ByreAPI
import Foundation

private let maxOutput = 64 << 10

private let runsDir = FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask)[0].appending(path: "run.fc.byre.mac/runs", directoryHint: .isDirectory)

@MainActor
private var watching: Set<String> = []

@MainActor
func runOnThisMac(_ api: API, _ item: InboxItem) async throws {
    let dir = runsDir.appending(path: UUID().uuidString, directoryHint: .isDirectory)
    watching.insert(dir.path)
    do {
        try await start(api, item, dir)
    } catch {
        try? FileManager.default.removeItem(at: dir)
        watching.remove(dir.path)
        throw error
    }
    poll(api, dir)
}

@MainActor
private func start(_ api: API, _ item: InboxItem, _ dir: URL) async throws {
    let files = dir.appending(path: "files", directoryHint: .isDirectory)
    try FileManager.default.createDirectory(at: files, withIntermediateDirectories: true)
    try item.id.write(to: dir.appending(path: "item"), atomically: true, encoding: .utf8)
    try api.origin.absoluteString.write(to: dir.appending(path: "server"), atomically: true, encoding: .utf8)
    try str(item.payload.value, "command")!.write(to: dir.appending(path: "command.sh"), atomically: true, encoding: .utf8)

    let (bundle, response) = try await api.session.download(from: api.url("/api/inbox/\(item.id)/bundle"))
    let code = (response as! HTTPURLResponse).statusCode
    if code != 200 {
        let text = (try? String(contentsOf: bundle, encoding: .utf8)) ?? ""
        throw ApiError(status: code, message: "download bundle: \(code) \(text)")
    }
    try await untar(bundle, into: files, errors: dir.appending(path: "untar.log"))
    try? FileManager.default.removeItem(at: bundle)

    let p = Process()
    p.executableURL = URL(filePath: "/usr/bin/lockf")
    p.arguments = ["-k", "lock", "/bin/zsh", "-c", "cd files && /bin/zsh -l ../command.sh >../output.log 2>&1; echo $? >../exit.tmp; mv ../exit.tmp ../exit"]
    p.currentDirectoryURL = dir
    p.standardInput = FileHandle.nullDevice
    p.standardOutput = FileHandle.nullDevice
    p.standardError = FileHandle.nullDevice
    try p.run()
    let lock = dir.appending(path: "lock")
    while !FileManager.default.fileExists(atPath: lock.path) {
        if !p.isRunning {
            throw ApiError(status: 0, message: "start command: lockf exited \(p.terminationStatus) before taking its lock")
        }
        try await Task.sleep(for: .milliseconds(20))
    }
    log.info("client run started: item \(item.id, privacy: .public) in \(dir.path, privacy: .public)")
}

private func untar(_ bundle: URL, into dir: URL, errors: URL) async throws {
    FileManager.default.createFile(atPath: errors.path, contents: nil)
    let errFile = try FileHandle(forWritingTo: errors)
    defer {
        try? errFile.close()
    }
    let p = Process()
    p.executableURL = URL(filePath: "/usr/bin/tar")
    p.arguments = ["-xzf", bundle.path, "-C", dir.path]
    p.standardError = errFile
    p.standardOutput = FileHandle.nullDevice
    try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, any Error>) in
        p.terminationHandler = { _ in
            c.resume()
        }
        do {
            try p.run()
        } catch {
            c.resume(throwing: error)
        }
    }
    if p.terminationStatus != 0 {
        let text = tail(errors)
        throw ApiError(status: 0, message: "unpack bundle: tar exited \(p.terminationStatus): \(text)")
    }
}

@MainActor
func resumeClientRuns(_ api: API) {
    let dirs = (try? FileManager.default.contentsOfDirectory(at: runsDir, includingPropertiesForKeys: nil)) ?? []
    for dir in dirs where !watching.contains(dir.path) && (try? String(contentsOf: dir.appending(path: "server"), encoding: .utf8)) == api.origin.absoluteString {
        if !FileManager.default.fileExists(atPath: dir.appending(path: "lock").path) {
            log.info("client run never started; removing \(dir.path, privacy: .public)")
            try? FileManager.default.removeItem(at: dir)
            continue
        }
        watching.insert(dir.path)
        poll(api, dir)
    }
}

@MainActor
private func poll(_ api: API, _ dir: URL) {
    Task {
        let exitFile = dir.appending(path: "exit").path
        var interrupted = false
        while !FileManager.default.fileExists(atPath: exitFile) {
            if lockFree(dir.appending(path: "lock").path) {
                interrupted = !FileManager.default.fileExists(atPath: exitFile)
                break
            }
            try? await Task.sleep(for: .seconds(1))
        }
        await report(api, dir, interrupted: interrupted)
        watching.remove(dir.path)
    }
}

private func lockFree(_ path: String) -> Bool {
    guard let h = FileHandle(forReadingAtPath: path) else {
        return false
    }
    defer {
        try? h.close()
    }
    let fd = h.fileDescriptor
    if flock(fd, LOCK_EX | LOCK_NB) != 0 {
        return false
    }
    flock(fd, LOCK_UN)
    return true
}

@MainActor
private func report(_ api: API, _ dir: URL, interrupted: Bool) async {
    do {
        let itemId = trimmed(try String(contentsOf: dir.appending(path: "item"), encoding: .utf8))
        var exit = -1
        var output = tail(dir.appending(path: "output.log"))
        if interrupted {
            output += "\n[byre: the command stopped without finishing, for example because the Mac restarted]"
        } else {
            exit = Int(trimmed(try String(contentsOf: dir.appending(path: "exit"), encoding: .utf8))) ?? -1
        }
        let client = Host.current().localizedName ?? ProcessInfo.processInfo.hostName
        _ = try await api.client.reportRun(path: .init(itemId: itemId), body: .json(.init(runId: dir.lastPathComponent, client: client, exitCode: exit, output: output))).ok
        log.info("client run reported: item \(itemId, privacy: .public) exit \(exit)")
    } catch {
        log.error("report client run \(dir.path, privacy: .public): \(errorText(error), privacy: .public)")
        guard let e = rootError(error) as? ApiError, (400..<500).contains(e.status) else {
            return
        }
    }
    try? FileManager.default.removeItem(at: dir)
}

private func tail(_ file: URL) -> String {
    guard let h = try? FileHandle(forReadingFrom: file) else {
        return ""
    }
    defer {
        try? h.close()
    }
    let size = (try? h.seekToEnd()) ?? 0
    try? h.seek(toOffset: size > UInt64(maxOutput) ? size - UInt64(maxOutput) : 0)
    return String(decoding: (try? h.readToEnd()) ?? Data(), as: UTF8.self)
}