frontend/src/lib/stream.ts

import { StreamEventName } from "@/lib/types";

type SSEEvent = {
  event: StreamEventName;
  data: Record<string, unknown>;
};

function parseEventBlock(block: string): SSEEvent | null {
  const lines = block.split("\n").map((line) => line.trim()).filter(Boolean);
  if (!lines.length) {
    return null;
  }

  let eventName: StreamEventName = "status";
  const dataLines: string[] = [];

  for (const line of lines) {
    if (line.startsWith("event:")) {
      eventName = line.replace("event:", "").trim() as StreamEventName;
      continue;
    }
    if (line.startsWith("data:")) {
      dataLines.push(line.replace("data:", "").trim());
    }
  }

  if (!dataLines.length) {
    return null;
  }

  const raw = dataLines.join("\n");
  try {
    return {
      event: eventName,
      data: JSON.parse(raw) as Record<string, unknown>
    };
  } catch {
    return {
      event: eventName,
      data: { message: raw }
    };
  }
}

export async function* parseSSEStream(
  stream: ReadableStream<Uint8Array>
): AsyncGenerator<SSEEvent> {
  const reader = stream.getReader();
  const decoder = new TextDecoder();

  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) {
      break;
    }

    buffer += decoder.decode(value, { stream: true });
    const blocks = buffer.split("\n\n");
    buffer = blocks.pop() ?? "";

    for (const block of blocks) {
      const parsed = parseEventBlock(block);
      if (parsed) {
        yield parsed;
      }
    }
  }

  if (buffer.trim()) {
    const parsed = parseEventBlock(buffer);
    if (parsed) {
      yield parsed;
    }
  }
}