serverSentEvents.gen.ts

// This file is auto-generated by @hey-api/openapi-ts

import type { Config } from "./types.gen.js"

export type ServerSentEventsOptions<TData = unknown> = Omit<RequestInit, "method"> &
  Pick<Config, "method" | "responseTransformer" | "responseValidator"> & {
    /**
     * Callback invoked when a network or parsing error occurs during streaming.
     *
     * This option applies only if the endpoint returns a stream of events.
     *
     * @param error The error that occurred.
     */
    onSseError?: (error: unknown) => void
    /**
     * Callback invoked when an event is streamed from the server.
     *
     * This option applies only if the endpoint returns a stream of events.
     *
     * @param event Event streamed from the server.
     * @returns Nothing (void).
     */
    onSseEvent?: (event: StreamEvent<TData>) => void
    /**
     * Default retry delay in milliseconds.
     *
     * This option applies only if the endpoint returns a stream of events.
     *
     * @default 3000
     */
    sseDefaultRetryDelay?: number
    /**
     * Maximum number of retry attempts before giving up.
     */
    sseMaxRetryAttempts?: number
    /**
     * Maximum retry delay in milliseconds.
     *
     * Applies only when exponential backoff is used.
     *
     * This option applies only if the endpoint returns a stream of events.
     *
     * @default 30000
     */
    sseMaxRetryDelay?: number
    /**
     * Optional sleep function for retry backoff.
     *
     * Defaults to using `setTimeout`.
     */
    sseSleepFn?: (ms: number) => Promise<void>
    url: string
  }

export interface StreamEvent<TData = unknown> {
  data: TData
  event?: string
  id?: string
  retry?: number
}

export type ServerSentEventsResult<TData = unknown, TReturn = void, TNext = unknown> = {
  stream: AsyncGenerator<TData extends Record<string, unknown> ? TData[keyof TData] : TData, TReturn, TNext>
}

export const createSseClient = <TData = unknown>({
  onSseError,
  onSseEvent,
  responseTransformer,
  responseValidator,
  sseDefaultRetryDelay,
  sseMaxRetryAttempts,
  sseMaxRetryDelay,
  sseSleepFn,
  url,
  ...options
}: ServerSentEventsOptions): ServerSentEventsResult<TData> => {
  let lastEventId: string | undefined

  const sleep = sseSleepFn ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)))

  const createStream = async function* () {
    let retryDelay: number = sseDefaultRetryDelay ?? 3000
    let attempt = 0
    const signal = options.signal ?? new AbortController().signal

    while (true) {
      if (signal.aborted) break

      attempt++

      const headers =
        options.headers instanceof Headers
          ? options.headers
          : new Headers(options.headers as Record<string, string> | undefined)

      if (lastEventId !== undefined) {
        headers.set("Last-Event-ID", lastEventId)
      }

      try {
        const response = await fetch(url, { ...options, headers, signal })

        if (!response.ok) throw new Error(`SSE failed: ${response.status} ${response.statusText}`)

        if (!response.body) throw new Error("No body in SSE response")

        const reader = response.body.pipeThrough(new TextDecoderStream()).getReader()

        let buffer = ""

        const abortHandler = () => {
          try {
            reader.cancel()
          } catch {
            // noop
          }
        }

        signal.addEventListener("abort", abortHandler)

        const MAX_BUFFER_SIZE = 10_485_760 // 10 MB
        try {
          while (true) {
            const { done, value } = await reader.read()
            if (done) break
            if (buffer.length + value.length > MAX_BUFFER_SIZE) {
              throw new Error(
                `SSE buffer overflow: accumulated ${buffer.length + value.length} bytes (limit ${MAX_BUFFER_SIZE})`,
              )
            }
            buffer += value

            const chunks = buffer.split("\n\n")
            buffer = chunks.pop() ?? ""

            for (const chunk of chunks) {
              const lines = chunk.split("\n")
              const dataLines: Array<string> = []
              let eventName: string | undefined

              for (const line of lines) {
                if (line.startsWith("data:")) {
                  dataLines.push(line.replace(/^data:\s*/, ""))
                } else if (line.startsWith("event:")) {
                  eventName = line.replace(/^event:\s*/, "")
                } else if (line.startsWith("id:")) {
                  lastEventId = line.replace(/^id:\s*/, "")
                } else if (line.startsWith("retry:")) {
                  const parsed = Number.parseInt(line.replace(/^retry:\s*/, ""), 10)
                  if (!Number.isNaN(parsed)) {
                    retryDelay = parsed
                  }
                }
              }

              let data: unknown
              let parsedJson = false

              if (dataLines.length) {
                const rawData = dataLines.join("\n")
                try {
                  data = JSON.parse(rawData)
                  parsedJson = true
                } catch {
                  data = rawData
                }
              }

              if (parsedJson) {
                if (responseValidator) {
                  await responseValidator(data)
                }

                if (responseTransformer) {
                  data = await responseTransformer(data)
                }
              }

              onSseEvent?.({
                data,
                event: eventName,
                id: lastEventId,
                retry: retryDelay,
              })

              if (dataLines.length) {
                yield data as any
              }
            }
          }
        } finally {
          signal.removeEventListener("abort", abortHandler)
          reader.releaseLock()
        }

        break // exit loop on normal completion
      } catch (error) {
        // connection failed or aborted; retry after delay
        onSseError?.(error)

        if (sseMaxRetryAttempts !== undefined && attempt >= sseMaxRetryAttempts) {
          break // stop after firing error
        }

        // exponential backoff: double retry each attempt, cap at 30s
        const backoff = Math.min(retryDelay * 2 ** (attempt - 1), sseMaxRetryDelay ?? 30000)
        await sleep(backoff)
      }
    }
  }

  const stream = createStream()

  return { stream }
}