src/lib/runtime/StreamManager.ts

/**
 * StreamManager.ts — Safari-safe streaming engine
 *
 * Features:
 * - cancel, resume (reconnect), retry
 * - normalize (strips SSE wrapper)
 * - buffer + token batching (reduces React re-renders)
 * - Safari iOS safe: handles broken pipes, readableStream drops
 */

export type StreamState = "idle" | "connecting" | "streaming" | "paused" | "done" | "error";

export interface StreamManagerOptions {
  onChunk: (text: string) => void;
  onStatus?: (state: StreamState) => void;
  onError?: (err: Error) => void;
  onDone?: () => void;
  batchIntervalMs?: number;
  maxRetries?: number;
  retryDelayMs?: number;
  timeoutMs?: number;
}

interface StreamSession {
  abortCtrl: AbortController;
  reader?: ReadableStreamDefaultReader<Uint8Array>;
}

export class StreamManager {
  private state: StreamState = "idle";
  private session: StreamSession | null = null;
  private buffer = "";
  private batchTimer: ReturnType<typeof setInterval> | null = null;
  private pendingChunks = "";
  private readonly opts: Required<StreamManagerOptions>;

  constructor(opts: StreamManagerOptions) {
    this.opts = {
      batchIntervalMs: 32,
      maxRetries: 2,
      retryDelayMs: 800,
      timeoutMs: 60_000,
      onStatus: () => {},
      onError: () => {},
      onDone: () => {},
      ...opts,
    };
  }

  private setState(s: StreamState) {
    this.state = s;
    this.opts.onStatus(s);
  }

  private startBatch() {
    if (this.batchTimer) return;
    this.batchTimer = setInterval(() => {
      if (this.pendingChunks) {
        this.opts.onChunk(this.pendingChunks);
        this.pendingChunks = "";
      }
    }, this.opts.batchIntervalMs);
  }

  private stopBatch() {
    if (this.batchTimer) {
      clearInterval(this.batchTimer);
      this.batchTimer = null;
    }
    if (this.pendingChunks) {
      this.opts.onChunk(this.pendingChunks);
      this.pendingChunks = "";
    }
  }

  private addChunk(text: string) {
    this.pendingChunks += text;
  }

  async stream(fetcher: () => Promise<Response>): Promise<void> {
    let attempt = 0;

    while (attempt <= this.opts.maxRetries) {
      attempt++;

      const ctrl = new AbortController();
      const timeoutId = setTimeout(() => ctrl.abort(), this.opts.timeoutMs);
      this.session = { abortCtrl: ctrl };
      this.setState("connecting");

      try {
        const response = await fetcher();

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

        if (!response.body) throw new Error("Risposta senza body");

        const reader = response.body.getReader();
        this.session.reader = reader;
        const decoder = new TextDecoder();
        this.setState("streaming");
        this.startBatch();

        while (true) {
          let done: boolean;
          let value: Uint8Array | undefined;

          try {
            ({ done, value } = await reader.read());
          } catch (readErr) {
            const e = readErr as Error;
            if (e.name === "AbortError" || ctrl.signal.aborted) break;
            throw e;
          }

          if (done) break;
          if (!value) continue;

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

        clearTimeout(timeoutId);
        this.stopBatch();
        this.setState("done");
        this.opts.onDone();
        return;
      } catch (err) {
        clearTimeout(timeoutId);
        this.stopBatch();

        const e = err instanceof Error ? err : new Error(String(err));

        if (this.state === "idle") return;

        if (e.name === "AbortError") {
          this.setState("done");
          return;
        }

        const isRetryable = attempt <= this.opts.maxRetries && !("status" in e && (e as {status:number}).status < 500);

        if (isRetryable) {
          this.setState("paused");
          await new Promise(r => setTimeout(r, this.opts.retryDelayMs * attempt));
          this.buffer = "";
          continue;
        }

        this.setState("error");
        this.opts.onError(e);
        return;
      }
    }
  }

  private processBuffer() {
    const lines = this.buffer.split("\n");
    this.buffer = lines[lines.length - 1];

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

      if (line.startsWith("data: ")) {
        const data = line.slice(6).trim();
        if (data === "[DONE]") {
          this.stopBatch();
          this.setState("done");
          this.opts.onDone();
          return;
        }
        const text = this.extractText(data);
        if (text) this.addChunk(text);
      }
    }
  }

  private extractText(data: string): string {
    try {
      const json = JSON.parse(data) as {
        choices?: Array<{ delta?: { content?: string }; text?: string }>;
        content?: string;
        text?: string;
      };
      return (
        json.choices?.[0]?.delta?.content ??
        json.choices?.[0]?.text ??
        json.content ??
        json.text ??
        ""
      );
    } catch {
      return data;
    }
  }

  cancel() {
    this.setState("idle");
    this.session?.abortCtrl.abort();
    this.session?.reader?.cancel().catch(() => {});
    this.session = null;
    this.stopBatch();
    this.buffer = "";
  }

  getState(): StreamState { return this.state; }
}