byre/mac/Sources/Byre/EventStream.swift

import AppKit
import Foundation
import Network

private let staleAfter: TimeInterval = 40
private let retryAfter: Duration = .seconds(2)
private let checkEvery: Duration = .seconds(5)

private enum Frame: Sendable {
    case open
    case ping
    case ready
    case event(id: Int64?, ByreEvent)
}

@MainActor
final class EventStream {
    private let api: API
    private let lead: Bool
    private let onEvent: (ByreEvent) -> Void
    private let onState: (Bool) -> Void
    private let onReady: () -> Void
    private var last: Int64
    private var heard = Date()
    private var reading: Task<Void, Never>?
    private var retry: Task<Void, Never>?
    private var checker: Task<Void, Never>?
    private var generation = 0
    private let monitor = NWPathMonitor()
    private var online: Bool?
    private var activeObserver: NotificationCenter.ObservationToken?

    init(api: API, after: Int64, lead: Bool, onEvent: @escaping (ByreEvent) -> Void, onState: @escaping (Bool) -> Void, onReady: @escaping () -> Void) {
        self.api = api
        self.lead = lead
        self.onEvent = onEvent
        self.onState = onState
        self.onReady = onReady
        last = after
        checker = Task { [weak self] in
            while !Task.isCancelled {
                try? await Task.sleep(for: checkEvery)
                self?.check()
            }
        }
        monitor.pathUpdateHandler = { [weak self] path in
            let up = path.status == .satisfied
            Task { @MainActor in
                self?.pathChanged(up)
            }
        }
        monitor.start(queue: .main)
        activeObserver = NotificationCenter.default.addObserver(of: NSApplication.self, for: .didBecomeActive) { [weak self] _ in
            self?.check()
        }
        connect()
    }

    func stop() {
        generation += 1
        checker?.cancel()
        retry?.cancel()
        reading?.cancel()
        monitor.cancel()
        if let activeObserver {
            NotificationCenter.default.removeObserver(activeObserver)
        }
        activeObserver = nil
    }

    private func pathChanged(_ up: Bool) {
        let was = online
        online = up
        if was == false && up {
            connect()
        }
    }

    private func check() {
        if Date().timeIntervalSince(heard) < staleAfter {
            return
        }
        onState(false)
        connect()
    }

    private func connect() {
        retry?.cancel()
        reading?.cancel()
        generation += 1
        let gen = generation
        heard = Date()
        var query = [URLQueryItem(name: "after", value: String(last))]
        if lead {
            query.append(URLQueryItem(name: "lead", value: "true"))
        }
        var request = URLRequest(url: api.url("/api/events", query))
        request.setValue("text/event-stream", forHTTPHeaderField: "Accept")
        let session = api.session
        reading = Task.detached { [weak self] in
            do {
                try await Self.read(session, request) { frame in
                    self?.receive(frame, gen)
                }
                await self?.failed(gen)
            } catch {
                await self?.failed(gen)
            }
        }
    }

    private func receive(_ frame: Frame, _ gen: Int) {
        guard gen == generation else {
            return
        }
        heard = Date()
        switch frame {
        case .open:
            onState(true)
        case .ping:
            break
        case .ready:
            onReady()
        case .event(let id, let ev):
            if let id, id > last {
                last = id
            }
            if ev.seq > last {
                last = ev.seq
            }
            onEvent(ev)
        }
    }

    private func failed(_ gen: Int) {
        guard gen == generation else {
            return
        }
        onState(false)
        retry = Task { [weak self] in
            try? await Task.sleep(for: retryAfter)
            if !Task.isCancelled && self?.generation == gen {
                self?.connect()
            }
        }
    }

    nonisolated private static func read(_ session: URLSession, _ request: URLRequest, _ deliver: @escaping @MainActor @Sendable (Frame) -> Void) async throws {
        let (bytes, response) = try await session.bytes(for: request)
        let code = (response as! HTTPURLResponse).statusCode
        if code != 200 {
            throw ApiError(status: code, message: "\(code) \(HTTPURLResponse.localizedString(forStatusCode: code))")
        }
        await deliver(.open)
        let decoder = jsonDecoder()
        var line: [UInt8] = []
        var afterCR = false
        var kind = ""
        var data: [String] = []
        var id: Int64?
        for try await byte in bytes {
            if byte == 0x0A && afterCR {
                afterCR = false
                continue
            }
            afterCR = byte == 0x0D
            if byte != 0x0A && byte != 0x0D {
                line.append(byte)
                continue
            }
            let text = String(decoding: line, as: UTF8.self)
            line.removeAll(keepingCapacity: true)
            if !text.isEmpty {
                if text.hasPrefix(":") {
                    continue
                }
                let colon = text.firstIndex(of: ":") ?? text.endIndex
                let field = String(text[..<colon])
                var value = colon == text.endIndex ? "" : String(text[text.index(after: colon)...])
                if value.hasPrefix(" ") {
                    value.removeFirst()
                }
                switch field {
                case "event":
                    kind = value
                case "data":
                    data.append(value)
                case "id":
                    id = Int64(value)
                default:
                    break
                }
                continue
            }
            defer {
                kind = ""
                data = []
                id = nil
            }
            if data.isEmpty {
                continue
            }
            switch kind {
            case "ping":
                await deliver(.ping)
            case "ready":
                await deliver(.ready)
            case "", "message":
                do {
                    let ev = try decoder.decode(ByreEvent.self, from: Data(data.joined(separator: "\n").utf8))
                    await deliver(.event(id: id, ev))
                } catch {
                    log.error("decode event: \(String(describing: error), privacy: .public)")
                }
            default:
                break
            }
        }
    }
}