spb/zyquo-cloud Public MIT

1//2//  StreamingService.swift3//  Zyquo Cloud4//5//  Author: Simon-Pierre Boucher6//  Mail: contact@spboucher.ai7//89import Foundation1011/// One Server-Sent Event as parsed off the wire.12struct SSEEvent {13    /// The `event:` field, if the stream names its events (Anthropic does).14    var event: String?15    /// Joined `data:` lines.16    var data: String17}1819/// Incremental SSE parser. Feed it raw lines (or byte chunks split on newlines)20/// and it yields complete events at blank-line boundaries, ignoring `:` comment21/// lines (DeepSeek sends `: keep-alive`) and unknown fields.22struct SSEParser {23    private var currentEvent: String?24    private var currentData: [String] = []2526    /// Consumes one line (without its trailing newline). Returns a completed27    /// event when the line is the blank separator, else nil.28    mutating func consume(line: String) -> SSEEvent? {29        if line.isEmpty {30            guard !currentData.isEmpty || currentEvent != nil else { return nil }31            let event = SSEEvent(event: currentEvent, data: currentData.joined(separator: "\n"))32            currentEvent = nil33            currentData = []34            return event.data.isEmpty && event.event == nil ? nil : event35        }36        if line.hasPrefix(":") { return nil } // comment / keep-alive37        if line.hasPrefix("event:") {38            currentEvent = String(line.dropFirst(6)).trimmingCharacters(in: .whitespaces)39        } else if line.hasPrefix("data:") {40            var value = String(line.dropFirst(5))41            if value.hasPrefix(" ") { value.removeFirst() }42            currentData.append(value)43        }44        // id:/retry:/unknown fields are ignored.45        return nil46    }47}4849/// Shared networking for all provider clients: request construction helpers and50/// an SSE line stream over URLSession.51enum StreamingService {52    /// URLSession tuned for long-lived streaming responses.53    static let session: URLSession = {54        let config = URLSessionConfiguration.default55        config.timeoutIntervalForRequest = 12056        config.timeoutIntervalForResource = 90057        config.httpAdditionalHeaders = ["User-Agent": "ZyquoCloud/1.0 (macOS)"]58        return URLSession(configuration: config)59    }()6061    /// POSTs `body` as JSON and returns the SSE events of the response.62    /// Throws `ProviderError` on non-2xx status (reading the full error body).63    static func sseEvents(64        for request: URLRequest,65        provider: ProviderID66    ) -> AsyncThrowingStream<SSEEvent, Error> {67        AsyncThrowingStream { continuation in68            let task = Task {69                do {70                    let (bytes, response) = try await session.bytes(for: request)71                    guard let http = response as? HTTPURLResponse else {72                        throw ProviderError.invalidResponse(provider, detail: "not an HTTP response")73                    }74                    guard (200..<300).contains(http.statusCode) else {75                        var body = Data()76                        for try await byte in bytes { body.append(byte) }77                        throw ProviderError.from(status: http.statusCode, body: body, provider: provider)78                    }79                    // NOTE: AsyncBytes.lines skips empty lines, which are the80                    // SSE event separators — split manually to preserve them.81                    var parser = SSEParser()82                    var lineBuffer = Data()83                    for try await byte in bytes {84                        if Task.isCancelled { break }85                        if byte == 0x0A { // \n86                            if lineBuffer.last == 0x0D { lineBuffer.removeLast() } // \r\n87                            let line = String(decoding: lineBuffer, as: UTF8.self)88                            lineBuffer.removeAll(keepingCapacity: true)89                            if let event = parser.consume(line: line) {90                                continuation.yield(event)91                            }92                        } else {93                            lineBuffer.append(byte)94                        }95                    }96                    // Flush a trailing line + event if the stream ended97                    // without a final newline / blank separator.98                    if !lineBuffer.isEmpty {99                        let line = String(decoding: lineBuffer, as: UTF8.self)100                        if let event = parser.consume(line: line) {101                            continuation.yield(event)102                        }103                    }104                    if let event = parser.consume(line: "") {105                        continuation.yield(event)106                    }107                    continuation.finish()108                } catch is CancellationError {109                    continuation.finish(throwing: ProviderError.cancelled)110                } catch let error as ProviderError {111                    continuation.finish(throwing: error)112                } catch {113                    continuation.finish(throwing: ProviderError.networkError(underlying: error))114                }115            }116            continuation.onTermination = { _ in task.cancel() }117        }118    }119120    /// Non-streaming JSON POST with exponential backoff on 429/5xx (3 attempts).121    /// Returns the response body data.122    static func postJSON(123        _ request: URLRequest,124        provider: ProviderID125    ) async throws -> Data {126        let maxAttempts = 3127        var lastError: ProviderError = .invalidResponse(provider, detail: "no attempts made")128        for attempt in 1...maxAttempts {129            do {130                let (data, response) = try await session.data(for: request)131                guard let http = response as? HTTPURLResponse else {132                    throw ProviderError.invalidResponse(provider, detail: "not an HTTP response")133                }134                guard (200..<300).contains(http.statusCode) else {135                    let error = ProviderError.from(status: http.statusCode, body: data, provider: provider)136                    if attempt < maxAttempts, http.statusCode == 429 || http.statusCode >= 500 {137                        lastError = error138                        let retryAfter = (response as? HTTPURLResponse)?139                            .value(forHTTPHeaderField: "Retry-After").flatMap(Double.init)140                        let delay = retryAfter ?? pow(2, Double(attempt)) * 2 // 4s, 8s141                        try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))142                        continue143                    }144                    throw error145                }146                return data147            } catch let error as ProviderError {148                throw error149            } catch is CancellationError {150                throw ProviderError.cancelled151            } catch {152                throw ProviderError.networkError(underlying: error)153            }154        }155        throw lastError156    }157158    /// GET returning decoded JSON data, with the same error mapping.159    static func getJSON(160        _ request: URLRequest,161        provider: ProviderID162    ) async throws -> Data {163        try await postJSON(request, provider: provider)164    }165}166