src/lib/tools/streaming.ts

/**
 * tools/streaming.ts — Streaming utilities
 *
 * Gestisce:
 * - SSE parsing
 * - Chunk buffering
 * - Error recovery
 * - Abort handling
 */

export interface StreamOptions {
  onChunk?: (chunk: string) => void;
  onError?: (error: Error) => void;
  onComplete?: () => void;
  timeout?: number;
  abortSignal?: AbortSignal;
}

// ─────────────────────────────────────────────
// Stream response
// ─────────────────────────────────────────────

export async function streamResponse(
  response: Response,
  options: StreamOptions,
): Promise<void> {
  if (!response.body) {
    throw new Error("La risposta non contiene un corpo");
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let timeoutId: ReturnType<typeof setTimeout> | null =
    null;

  try {
    // Setup timeout
    if (options.timeout) {
      timeoutId = setTimeout(() => {
        reader.cancel();
        options.onError?.(
          new Error("Timeout dello stream"),
        );
      }, options.timeout);
    }

    // Setup abort listener
    if (options.abortSignal) {
      options.abortSignal.addEventListener(
        "abort",
        () => {
          reader.cancel();
        },
      );
    }

    // Read chunks
    while (true) {
      const { done, value } = await reader.read();

      if (done) break;

      buffer += decoder.decode(value, {
        stream: true,
      });

      // Process complete lines
      const lines = buffer.split("\n");
      buffer = lines[lines.length - 1];

      for (let i = 0; i < lines.length - 1; i++) {
        const line = lines[i].trim();

        if (!line) continue;

        // Handle SSE format
        if (line.startsWith("data: ")) {
          const data = line.slice(6);
          if (data === "[DONE]") {
            options.onComplete?.();
            return;
          }
          options.onChunk?.(data);
        } else if (line.startsWith("error: ")) {
          const error = line.slice(7);
          throw new Error(error);
        }
      }
    }

    // Process remaining buffer
    if (buffer.trim()) {
      if (buffer.startsWith("data: ")) {
        const data = buffer.slice(6);
        if (data !== "[DONE]") {
          options.onChunk?.(data);
        }
      }
    }

    options.onComplete?.();
  } catch (error) {
    options.onError?.(
      error instanceof Error
        ? error
        : new Error(String(error)),
    );
  } finally {
    if (timeoutId) clearTimeout(timeoutId);
    reader.cancel();
  }
}

// ─────────────────────────────────────────────
// Parse SSE chunk
// ─────────────────────────────────────────────

export function parseSSEChunk(
  chunk: string,
): unknown {
  try {
    return JSON.parse(chunk);
  } catch {
    return chunk;
  }
}

// ─────────────────────────────────────────────
// Build SSE request
// ─────────────────────────────────────────────

export function buildStreamRequest(
  url: string,
  options: {
    method?: string;
    headers?: Record<string, string>;
    body?: unknown;
    abortSignal?: AbortSignal;
  } = {},
): RequestInit {
  return {
    method: options.method || "POST",
    headers: {
      "Content-Type": "application/json",
      ...options.headers,
    },
    body:
      options.body instanceof FormData
        ? options.body
        : JSON.stringify(options.body),
    signal: options.abortSignal,
  };
}

// ─────────────────────────────────────────────
// Collect stream to string
// ─────────────────────────────────────────────

export async function collectStream(
  response: Response,
  options: { timeout?: number; abortSignal?: AbortSignal } = {},
): Promise<string> {
  return new Promise((resolve, reject) => {
    let result = "";

    streamResponse(response, {
      onChunk: (chunk) => {
        result += chunk;
      },
      onError: reject,
      onComplete: () => resolve(result),
      timeout: options.timeout,
      abortSignal: options.abortSignal,
    }).catch(reject);
  });
}

// ─────────────────────────────────────────────
// Stream with retry
// ─────────────────────────────────────────────

export async function streamWithRetry(
  fetcher: () => Promise<Response>,
  options: StreamOptions & { maxRetries?: number } = {},
): Promise<void> {
  const maxRetries = options.maxRetries ?? 2;
  let lastError: Error | undefined;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      const response = await fetcher();

      if (!response.ok) {
        throw new Error(
          `HTTP ${response.status}: ${response.statusText}`,
        );
      }

      await streamResponse(response, options);
      return;
    } catch (error) {
      lastError = error as Error;

      // Don't retry on abort
      if (
        lastError.name === "AbortError" ||
        lastError.message === "Richiesta interrotta"
      ) {
        throw lastError;
      }

      // Wait before retry
      if (attempt < maxRetries) {
        await new Promise((resolve) =>
          setTimeout(resolve, 1000 * (attempt + 1)),
        );
      }
    }
  }

  throw (
    lastError ||
    new Error("Stream non riuscito dopo i tentativi")
  );
}