ipodderx-app/ios/Sources/Events.swift

import Foundation

/// What the daemon is doing, as it does it.
///
/// `GET /api/events` is server-sent events carrying the same broadcast the control socket does,
/// so a scan started from anywhere shows up here. The stream is a broadcast: a client attached
/// to a busy daemon also sees that daemon's other work.
enum IPXEvent: Decodable, Sendable {
    case feedStart(feed: String)
    case feedSkip(feed: String, reason: String)
    case feedDone(feed: String, new: Int, downloaded: Int, failed: Int)
    case feedError(feed: String, msg: String)
    /// Carries the enclosure id, without which a UI cannot tell one download from another and
    /// ends up animating every pending row. Throttled to whole percents by the server.
    case progress(feed: String, enclosure: Int, file: String, done: Int, total: Int?)
    case downloadDone(feed: String, enclosure: Int, path: String, bytes: Int)
    case downloadError(feed: String, enclosure: Int, msg: String)
    case reaped(path: String, bytes: Int)
    case scanDone(feeds: Int)
    case reapDone(files: Int, bytes: Int)
    /// Anything this build does not know about. New events must not break an old app.
    case other(String)

    private enum Keys: String, CodingKey {
        case ev, feed, reason, new, downloaded, failed, msg, enclosure, file, done, total,
             path, bytes, feeds, files
    }

    init(from decoder: Decoder) throws {
        let c = try decoder.container(keyedBy: Keys.self)
        let ev = try c.decode(String.self, forKey: .ev)
        func str(_ k: Keys) -> String { (try? c.decode(String.self, forKey: k)) ?? "" }
        func int(_ k: Keys) -> Int { (try? c.decode(Int.self, forKey: k)) ?? 0 }
        switch ev {
        case "feed_start":     self = .feedStart(feed: str(.feed))
        case "feed_skip":      self = .feedSkip(feed: str(.feed), reason: str(.reason))
        case "feed_done":      self = .feedDone(feed: str(.feed), new: int(.new),
                                                downloaded: int(.downloaded), failed: int(.failed))
        case "feed_error":     self = .feedError(feed: str(.feed), msg: str(.msg))
        case "progress":       self = .progress(feed: str(.feed), enclosure: int(.enclosure),
                                                file: str(.file), done: int(.done),
                                                total: try? c.decode(Int.self, forKey: .total))
        case "download_done":  self = .downloadDone(feed: str(.feed), enclosure: int(.enclosure),
                                                    path: str(.path), bytes: int(.bytes))
        case "download_error": self = .downloadError(feed: str(.feed), enclosure: int(.enclosure),
                                                     msg: str(.msg))
        case "reaped":         self = .reaped(path: str(.path), bytes: int(.bytes))
        case "scan_done":      self = .scanDone(feeds: int(.feeds))
        case "reap_done":      self = .reapDone(files: int(.files), bytes: int(.bytes))
        default:               self = .other(ev)
        }
    }
}

/// Reads the event stream and keeps reading it.
///
/// URLSession's byte stream does the parsing work; there is no need for another dependency.
/// The connection drops -- a deploy, a sleeping laptop, a tunnel blinking -- so it reconnects,
/// backing off so a daemon that is down is not hammered.
@MainActor
final class Events {
    private(set) var isConnected = false
    private var task: Task<Void, Never>?
    private var handlers: [UUID: (IPXEvent) -> Void] = [:]

    /// Returns a token; hold it for as long as you want the events, and forget it to stop.
    @discardableResult
    func onEvent(_ handler: @escaping (IPXEvent) -> Void) -> UUID {
        let id = UUID()
        handlers[id] = handler
        return id
    }

    func remove(_ id: UUID) { handlers[id] = nil }

    func start() {
        guard task == nil else { return }
        task = Task { [weak self] in
            var backoff = 1.0
            while !Task.isCancelled {
                let ok = await self?.readOnce() ?? false
                if Task.isCancelled { return }
                // A clean end means the daemon closed it, which is normal on a deploy: try again
                // soon. A failure means it is down, so wait longer each time, up to half a minute.
                backoff = ok ? 1.0 : min(backoff * 2, 30)
                try? await Task.sleep(nanoseconds: UInt64(backoff * 1_000_000_000))
            }
        }
    }

    func stop() {
        task?.cancel()
        task = nil
        isConnected = false
    }

    /// One connection, for as long as it lasts. True if it ended rather than failed.
    private func readOnce() async -> Bool {
        guard let url = ServerSettings.url("/api/events") else { return false }
        var req = URLRequest(url: url)
        req.setValue("text/event-stream", forHTTPHeaderField: "Accept")
        // The stream is meant to stay open; the default timeout would cut it every minute.
        req.timeoutInterval = .infinity

        do {
            let (bytes, response) = try await URLSession.shared.bytes(for: req)
            guard (response as? HTTPURLResponse)?.statusCode == 200 else { return false }
            isConnected = true
            defer { isConnected = false }

            for try await line in bytes.lines {
                if Task.isCancelled { return true }
                // Every event ipx sends is unnamed, so the data lines are the whole of it.
                // Comments (": keep-alive") and blank separators are skipped.
                guard line.hasPrefix("data:") else { continue }
                let json = line.dropFirst(5).trimmingCharacters(in: .whitespaces)
                guard let data = json.data(using: .utf8),
                      let event = try? JSONDecoder().decode(IPXEvent.self, from: data) else { continue }
                for handler in handlers.values { handler(event) }
            }
            return true
        } catch {
            return false
        }
    }
}