Function fetchSSE

web/lib/api/stream.ts:173–228

(
  url: string,
  opts?: {
    signal?: AbortSignal;
    headers?: Record<string, string>;
    maxReconnects?: number;
    onProgress?: (progress: StreamProgress) => void;
  }
)

 * Streams an async generator of StreamEvents with reconnection.
 */
export async function* fetchSSE(
  url: string,
  opts?: {
    signal?: AbortSignal;
    headers?: Record<string, string>;
    maxReconnects?: number;
    onProgress?: (progress: StreamProgress) => void;
  }
): AsyncGenerator<StreamEvent> {
  const maxReconnects = opts?.maxReconnects ?? 5;
  let attempt = 0;

  while (true) {
    if (opts?.signal?.aborted) break;

    try {
      const res = await fetch(url, {
        headers: {
          Accept: "text/event-stream",
          "Cache-Control": "no-cache",
          ...opts?.headers,
        },
        signal: opts?.signal,
      });

      if (!res.ok) {
        throw new ApiError(res.status, `SSE connection failed: ${res.status}`);
      }

      // Reset attempt counter on successful connection
      attempt = 0;
      yield* parseStream(res, { signal: opts?.signal, onProgress: opts?.onProgress });

      // Clean disconnect — don't reconnect
      break;
    } catch (err) {
      if (err instanceof ApiError && err.type === "abort") break;
      if (opts?.signal?.aborted) break;
      if (attempt >= maxReconnects) throw err;

      const delay = Math.min(1_000 * Math.pow(2, attempt), 30_000);
      attempt++;
      await new Promise<void>((resolve, reject) => {
        const timer = setTimeout(resolve, delay);
        opts?.signal?.addEventListener(
          "abort",
          () => {
            clearTimeout(timer);
            reject(new ApiError(0, "Cancelled", "abort"));
          },
          { once: true }
        );
      });
    }
  }
}