spb/zyquo-cloud-web Public MIT

1/*2 *  sse.ts3 *  Zyquo Cloud Web4 *5 *  Author: Simon-Pierre Boucher6 *  Mail: contact@spboucher.ai7 *8 *  fetch + ReadableStream SSE parsing, ported from the native SSEParser /9 *  StreamingService. One parser handles both stream shapes: OpenAI-style10 *  anonymous `data:` events and Anthropic's named `event:` blocks.11 */1213import type { Provider } from '../types'14import { ProviderError } from './types'1516/** One Server-Sent Event as parsed off the wire. */17export interface SSEEvent {18  /** The `event:` field, if the stream names its events (Anthropic does). */19  event: string | null20  /** Joined `data:` lines. */21  data: string22}2324/**25 * Incremental SSE parser. Feed it raw lines (without trailing newlines) and it26 * yields complete events at blank-line boundaries, ignoring `:` comment lines27 * (DeepSeek sends `: keep-alive`) and unknown fields.28 */29export class SSEParser {30  private currentEvent: string | null = null31  private currentData: string[] = []3233  /** Consumes one line. Returns a completed event at a blank separator, else null. */34  consume(line: string): SSEEvent | null {35    if (line === '') {36      if (this.currentData.length === 0 && this.currentEvent === null) return null37      const event: SSEEvent = { event: this.currentEvent, data: this.currentData.join('\n') }38      this.currentEvent = null39      this.currentData = []40      return event.data === '' && event.event === null ? null : event41    }42    if (line.startsWith(':')) return null // comment / keep-alive43    if (line.startsWith('event:')) {44      this.currentEvent = line.slice(6).trim()45    } else if (line.startsWith('data:')) {46      let value = line.slice(5)47      if (value.startsWith(' ')) value = value.slice(1)48      this.currentData.push(value)49    }50    // id:/retry:/unknown fields are ignored.51    return null52  }53}5455/**56 * POSTs `body` as JSON and yields the SSE events of the response.57 * Throws ProviderError on non-2xx status (reading the full error body).58 * Cancellation: abort the signal — surfaces as a `cancelled` ProviderError.59 */60export async function* sseEvents(61  url: string,62  init: { headers: Record<string, string>; body: string },63  provider: Provider,64  signal?: AbortSignal65): AsyncGenerator<SSEEvent> {66  let res: Response67  try {68    res = await fetch(url, {69      method: 'POST',70      headers: init.headers,71      body: init.body,72      ...(signal ? { signal } : {}),73    })74  } catch (err) {75    if (err instanceof DOMException && err.name === 'AbortError') throw ProviderError.cancelled()76    throw ProviderError.network(provider, err)77  }7879  if (!res.ok) {80    const body = await res.text().catch(() => '')81    const retryAfter = parseRetryAfter(res)82    throw ProviderError.from(res.status, body, provider, retryAfter)83  }84  if (!res.body) {85    throw ProviderError.invalidResponse(provider, 'response had no body')86  }8788  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader()89  const parser = new SSEParser()90  let buffer = ''9192  try {93    for (;;) {94      let chunk: ReadableStreamReadResult<string>95      try {96        chunk = await reader.read()97      } catch (err) {98        if (err instanceof DOMException && err.name === 'AbortError') throw ProviderError.cancelled()99        throw ProviderError.network(provider, err)100      }101      if (chunk.done) break102      buffer += chunk.value103      // Split on newlines, preserving blank lines (they are event separators).104      let newlineIndex: number105      while ((newlineIndex = buffer.indexOf('\n')) !== -1) {106        let line = buffer.slice(0, newlineIndex)107        buffer = buffer.slice(newlineIndex + 1)108        if (line.endsWith('\r')) line = line.slice(0, -1)109        const event = parser.consume(line)110        if (event) yield event111      }112    }113    // Flush a trailing line + event if the stream ended without a final114    // newline / blank separator.115    if (buffer !== '') {116      const event = parser.consume(buffer.endsWith('\r') ? buffer.slice(0, -1) : buffer)117      if (event) yield event118    }119    const finalEvent = parser.consume('')120    if (finalEvent) yield finalEvent121  } finally {122    reader.cancel().catch(() => {})123  }124}125126/**127 * Non-streaming JSON request with exponential backoff on 429/5xx (3 attempts,128 * honoring Retry-After). Returns the response body text.129 */130export async function requestJSON(131  url: string,132  init: { method: 'GET' | 'POST'; headers: Record<string, string>; body?: string },133  provider: Provider,134  signal?: AbortSignal135): Promise<string> {136  const maxAttempts = 3137  let lastError: ProviderError = ProviderError.invalidResponse(provider, 'no attempts made')138  for (let attempt = 1; attempt <= maxAttempts; attempt++) {139    let res: Response140    try {141      res = await fetch(url, {142        method: init.method,143        headers: init.headers,144        ...(init.body !== undefined ? { body: init.body } : {}),145        ...(signal ? { signal } : {}),146      })147    } catch (err) {148      if (err instanceof DOMException && err.name === 'AbortError') throw ProviderError.cancelled()149      throw ProviderError.network(provider, err)150    }151    const body = await res.text().catch(() => '')152    if (res.ok) return body153    const retryAfter = parseRetryAfter(res)154    const error = ProviderError.from(res.status, body, provider, retryAfter)155    if (attempt < maxAttempts && (res.status === 429 || res.status >= 500)) {156      lastError = error157      const delay = (retryAfter ?? 2 ** attempt * 2) * 1000 // 4s, 8s158      await sleep(delay, signal)159      continue160    }161    throw error162  }163  throw lastError164}165166function parseRetryAfter(res: Response): number | undefined {167  const header = res.headers.get('Retry-After')168  if (!header) return undefined169  const seconds = Number(header)170  return Number.isFinite(seconds) ? seconds : undefined171}172173function sleep(ms: number, signal?: AbortSignal): Promise<void> {174  return new Promise((resolve, reject) => {175    const timer = setTimeout(resolve, ms)176    signal?.addEventListener(177      'abort',178      () => {179        clearTimeout(timer)180        reject(ProviderError.cancelled())181      },182      { once: true }183    )184  })185}186187/** Joins a base URL and a path, preserving base path components. */188export function joinURL(base: string, path: string): string {189  return `${base.replace(/\/+$/, '')}/${path.replace(/^\/+/, '')}`190}191