/** * Low-level HTTP + SSE helpers for the provider layer. * * This module provides: * - `readSSEEvents`: parse a `text/event-stream` `ReadableStream` into * concatenated `data:` payloads (one per event). * - `fetchWithRetry`: a `fetch` wrapper that retries on 429/5xx with * exponential backoff and surfaces auth errors immediately. * * It is intentionally dependency-free and browser-safe (uses the global * `fetch`/`TextDecoder`/`ReadableStream`). See CHUNK#8.md ยง3. */ import { ProviderAuthError, ProviderError, ProviderRateLimitError, } from "./errors.js"; /** Read a `text/event-stream` body and yield each event's `data:` payload. */ export async function* readSSEEvents( stream: ReadableStream>, ): AsyncGenerator { const reader = stream.getReader(); const decoder = new TextDecoder(); let buffer = ""; const dataLines: string[] = []; const flush = (): string | null => { if (dataLines.length === 0) { return null; } const value = dataLines.join("\n"); dataLines.length = 0; return value; }; for (;;) { const { done, value } = await reader.read(); if (value) { buffer += decoder.decode(value, { stream: true }); } let newlineIndex: number; while ((newlineIndex = buffer.indexOf("\n")) >= 0) { let line = buffer.slice(0, newlineIndex); buffer = buffer.slice(newlineIndex + 1); if (line.endsWith("\r")) { line = line.slice(0, -1); } if (line === "") { // Blank line terminates an SSE event. const payload = flush(); if (payload !== null) { yield payload; } } else if (line.startsWith("data:")) { dataLines.push(line.slice(5).replace(/^ /, "")); } // Ignore `event:`, `id:`, `retry:` lines. } if (done) { break; } } // Flush any trailing data without a terminating blank line. if (buffer.length > 0) { if (buffer.startsWith("data:")) { dataLines.push(buffer.slice(5).replace(/^ /, "")); } const payload = flush(); if (payload !== null) { yield payload; } } } /** Parse a JSON SSE event, returning `null` for malformed input. */ export function parseSSEEvent(event: string): T | null { try { return JSON.parse(event) as T; } catch { return null; } } /** Sleep for `ms` milliseconds. */ export function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } /** Build the retry backoff delay for a given attempt (0-based). */ function backoffDelay(attempt: number, maxMs: number): number { const base = 250 * 2 ** attempt; return Math.min(base, maxMs); } /** * `fetch` wrapper that retries on 429/5xx with exponential backoff. Auth * errors (401/403) are thrown immediately; other non-retryable errors are * surfaced as-is. */ export async function fetchWithRetry( url: string, init: RequestInit, options: { retries?: number; maxDelayMs?: number; signal?: AbortSignal } = {}, ): Promise { const retries = options.retries ?? 2; const maxDelayMs = options.maxDelayMs ?? 5000; let lastError: unknown; for (let attempt = 0; attempt <= retries; attempt++) { if (attempt > 0) { await sleep(backoffDelay(attempt - 1, maxDelayMs)); } try { const response = await fetch(url, init); if (response.ok) { return response; } const detail = await safeText(response); if (response.status === 401 || response.status === 403) { throw new ProviderAuthError( `Provider rejected credentials (${response.status}).`, ); } if (response.status === 429 || response.status >= 500) { lastError = new ProviderRateLimitError( `Provider rate-limited or overloaded (${response.status}).`, ); if (attempt < retries) { continue; } throw lastError; } throw new ProviderError( `Provider request failed (${response.status}): ${detail}`, ); } catch (error) { if (error instanceof Error && error.name === "AbortError") { throw error; } lastError = error; if (attempt >= retries) { throw error; } } } // Unreachable, but satisfies the type checker. throw lastError instanceof Error ? lastError : new ProviderError("Request failed after retries."); } /** Read a response body as text, tolerating errors. */ async function safeText(response: Response): Promise { try { return await response.text(); } catch { return "(no body)"; } }