import { randomUUID } from "node:crypto";
import {
  parseToolCallsFromContent,
  stripToolCallsFromContent,
  type ToolCall,
} from "./tool-call-parser";
import {
  REASONING_FIELDS,
  firstReasoningField,
  createThinkRewriter,
  thinkingTagPrefixIsPartial,
} from "./reasoning";

export interface StreamUsage {
  prompt_tokens: number;
  completion_tokens: number;
  reasoning_tokens?: number;
  cache_read_tokens?: number;
  cache_write_tokens?: number;
}

export interface ToolCallStreamOptions {
  bufferImplicitReasoningContent?: boolean;
}

export const createToolCallStream = (
  reader: ReadableStreamDefaultReader<Uint8Array>,
  onUsage?: (usage: StreamUsage) => void,
  onFirstToken?: () => void,
  options: ToolCallStreamOptions = {},
): ReadableStream<Uint8Array> => {
  const decoder = new TextDecoder();
  const encoder = new TextEncoder();
  let buffer = "";
  let pendingEventLines: string[] = [];
  let visibleContentBuffer = "";
  let toolCallsFound = false;
  let usageTracked = false;
  let emittedLines = 0;
  let downstreamClosed = false;
  let firstTokenTracked = false;
  const contentHistory = new Map<string, { text: string; snapshot: boolean }>();
  const reasoningHistory = new Map<string, { text: string; snapshot: boolean }>();
  const replayCursors = new Map<string, number>();
  const tearDownUpstream = async (): Promise<void> => {
    try {
      await reader.cancel();
    } catch {
      // upstream already torn down; ignore.
    }
  };
  const stripToolXmlDelta = (text: string): string => {
    return stripToolCallsFromContent(text);
  };

  const normalizeTextDelta = (
    history: Map<string, { text: string; snapshot: boolean }>,
    key: string,
    text: string,
    forceSnapshot = false,
  ): string => {
    if (!text) return text;
    const previous = history.get(key) ?? { text: "", snapshot: forceSnapshot };
    const replayCursor = replayCursors.get(key);
    if (replayCursor !== undefined) {
      const expected = previous.text.slice(replayCursor, replayCursor + text.length);
      if (expected === text) {
        const nextCursor = replayCursor + text.length;
        if (nextCursor >= previous.text.length) replayCursors.delete(key);
        else replayCursors.set(key, nextCursor);
        return "";
      }
      // The supposed replay diverged, so the prefix we suppressed was never a
      // replay — it was real content whose leading tokens happened to repeat an
      // earlier prefix (e.g. a new sentence starting with "The "/"Hello"/"\n").
      // Resurrect exactly what we withheld so no token is silently dropped.
      replayCursors.delete(key);
      const resurrected = previous.text.slice(0, replayCursor);
      const merged = resurrected + text;
      history.set(key, { text: previous.text + merged, snapshot: false });
      return merged;
    }
    // A cumulative snapshot is STRICTLY longer than what we've accumulated (it
    // adds new tokens). Using `>=` here misfired when a delta merely EQUALS the
    // accumulated text — e.g. a second "\n" right after a first "\n" (a blank
    // line / paragraph break, or the gap before a list). That equal "\n" was
    // treated as a cumulative snapshot, sliced to "" (dropped), AND flipped the
    // stream into snapshot mode, mangling everything after it — collapsing
    // newlines so a list rendered all on one line. Require strictly longer.
    const isCumulative =
      previous.text.length > 0 &&
      text.length > previous.text.length &&
      text.startsWith(previous.text);
    const shouldSlice = forceSnapshot || previous.snapshot || isCumulative;

    if (shouldSlice) {
      history.set(key, { text, snapshot: true });
      return isCumulative ? text.slice(previous.text.length) : text;
    }

    // A shorter NON-WHITESPACE delta that matches the start of the accumulated
    // text *might* be an upstream replaying from the top; speculatively suppress
    // it (the divergence branch above resurrects it if the replay never pans
    // out). A whitespace-only delta (a standalone "\n" between list rows or
    // paragraphs) is never a replay restart — suppressing it would drop the
    // newline — so always append it verbatim.
    if (
      text.trim() !== "" &&
      previous.text.length > text.length &&
      previous.text.startsWith(text)
    ) {
      replayCursors.set(key, text.length);
      return "";
    }

    history.set(key, { text: previous.text + text, snapshot: false });
    return text;
  };

  const contentThink = createThinkRewriter({
    bufferImplicitReasoningContent: Boolean(options.bufferImplicitReasoningContent),
  });
  const reasoningThink = createThinkRewriter();

  const enqueueLine = (
    controller: ReadableStreamDefaultController<Uint8Array>,
    line: string,
  ): void => {
    if (downstreamClosed) return;
    try {
      controller.enqueue(encoder.encode(`${line}\n`));
      emittedLines += 1;
    } catch {
      downstreamClosed = true;
      void tearDownUpstream();
    }
  };
  // Terminate each synthesized `data:` line with a blank line so the SSE parser
  // dispatches it as its own event; without it, an injected chunk followed by
  // `data: [DONE]` concatenates to `{...}\n[DONE]` and fails JSON.parse.
  const enqueueDataEvent = (
    controller: ReadableStreamDefaultController<Uint8Array>,
    dataLine: string,
  ): void => {
    enqueueLine(controller, dataLine);
    enqueueLine(controller, "");
  };

  const buildToolCallChunk = (toolCalls: ToolCall[]): string => {
    const payload = {
      id: `chatcmpl-${randomUUID().slice(0, 8)}`,
      choices: [
        {
          index: 0,
          delta: { tool_calls: toolCalls },
          finish_reason: "tool_calls",
        },
      ],
    };
    return `data: ${JSON.stringify(payload)}`;
  };

  const buildFlushChunk = (payload: {
    content?: string;
    reasoning_content?: string;
  }): string | null => {
    const content = payload.content ?? "";
    const reasoning = payload.reasoning_content ?? "";
    if (!content && !reasoning) return null;
    const delta: Record<string, string> = {};
    if (content) delta["content"] = content;
    if (reasoning) delta["reasoning_content"] = reasoning;
    return `data: ${JSON.stringify({ id: `chatcmpl-${randomUUID().slice(0, 8)}`, choices: [{ index: 0, delta }] })}`;
  };

  const emitVisibleContent = (
    controller: ReadableStreamDefaultController<Uint8Array>,
    content: string,
  ): void => {
    if (!content) return;
    visibleContentBuffer += content;
    const cleaned = stripToolXmlDelta(content);
    const chunk = buildFlushChunk({ content: cleaned });
    if (chunk) enqueueDataEvent(controller, chunk);
  };

  const flushThinkCarry = (controller: ReadableStreamDefaultController<Uint8Array>): void => {
    emitVisibleContent(controller, contentThink.drainPendingContent());
    const tail = contentThink.drainCarry();
    if (!tail) return;
    const carryLooksLikeThink = thinkingTagPrefixIsPartial(tail.trim());
    const chunk =
      contentThink.inThink() || carryLooksLikeThink
        ? buildFlushChunk({ reasoning_content: stripToolXmlDelta(tail) })
        : buildFlushChunk({ content: stripToolXmlDelta(tail) });
    if (chunk) enqueueDataEvent(controller, chunk);
  };

  const parseUsage = (data: Record<string, unknown>): void => {
    if (usageTracked || !onUsage) return;
    const usage = data["usage"] as Record<string, number> | undefined;
    if (usage && (usage["prompt_tokens"] || usage["completion_tokens"])) {
      onUsage({
        prompt_tokens: usage["prompt_tokens"] ?? 0,
        completion_tokens: usage["completion_tokens"] ?? 0,
        reasoning_tokens:
          (usage["reasoning_tokens"] as number | undefined) ??
          (usage["completion_tokens_details"] as Record<string, number> | undefined)?.[
            "reasoning_tokens"
          ] ??
          0,
        cache_read_tokens:
          (usage["prompt_tokens_details"] as Record<string, number> | undefined)?.[
            "cached_tokens"
          ] ?? 0,
        cache_write_tokens: 0,
      });
      usageTracked = true;
    }
  };

  const trackFirstToken = (): void => {
    if (firstTokenTracked) return;
    firstTokenTracked = true;
    onFirstToken?.();
  };

  const maybeInjectToolCalls = (controller: ReadableStreamDefaultController<Uint8Array>): void => {
    if (toolCallsFound || !visibleContentBuffer) return;
    const parsed = parseToolCallsFromContent(visibleContentBuffer);
    if (parsed.length > 0) {
      enqueueDataEvent(controller, buildToolCallChunk(parsed));
      toolCallsFound = true;
    }
  };

  type ReaderResult = { done: boolean; value?: Uint8Array | undefined };

  return new ReadableStream<Uint8Array>({
    async start(controller): Promise<void> {
      void controller;
    },
    async pull(controller): Promise<void> {
      const flushEvent = (lines: string[]): void => {
        if (lines.length === 0) return;

        const dataLines: string[] = [];
        const otherLines: string[] = [];
        for (const rawLine of lines) {
          const trimmedStart = rawLine.trimStart();
          if (trimmedStart.startsWith("data:")) {
            dataLines.push(trimmedStart.slice("data:".length).trimStart());
          } else if (rawLine.length > 0) {
            otherLines.push(rawLine);
          }
        }

        if (dataLines.length === 0) {
          for (const outLine of lines) {
            enqueueLine(controller, outLine);
          }
          return;
        }

        const data = dataLines.join("\n").trim();
        if (data === "[DONE]") {
          flushThinkCarry(controller);
          maybeInjectToolCalls(controller);
          for (const outLine of otherLines) {
            enqueueLine(controller, outLine);
          }
          enqueueDataEvent(controller, "data: [DONE]");
          return;
        }

        let parsed: Record<string, unknown> | null = null;
        try {
          parsed = JSON.parse(data) as Record<string, unknown>;
        } catch {
          parsed = null;
        }
        if (!parsed) {
          for (const outLine of lines) {
            enqueueLine(controller, outLine);
          }
          return;
        }

        parseUsage(parsed);
        const choices = parsed["choices"];
        if (Array.isArray(choices)) {
          for (const [choiceIndex, choice] of choices.entries()) {
            const choiceRecord = choice as Record<string, unknown>;
            const hasDelta = choiceRecord["delta"] && typeof choiceRecord["delta"] === "object";
            const delta = (hasDelta ? choiceRecord["delta"] : choiceRecord["message"]) as
              | Record<string, unknown>
              | undefined;
            if (!delta) continue;
            const toolCalls = delta["tool_calls"];
            const hasActiveToolCalls = Array.isArray(toolCalls) && toolCalls.length > 0;
            if (hasActiveToolCalls) {
              toolCallsFound = true;
              trackFirstToken();
            }
            const rawContent = typeof delta["content"] === "string" ? String(delta["content"]) : "";
            const content = normalizeTextDelta(
              contentHistory,
              `${choiceIndex}:content`,
              rawContent,
              !hasDelta,
            );
            const rawReasoning = firstReasoningField(delta);
            const reasoningRaw = rawReasoning
              ? normalizeTextDelta(
                  reasoningHistory,
                  `${choiceIndex}:reasoning`,
                  rawReasoning,
                  !hasDelta,
                )
              : "";
            if (content || reasoningRaw) trackFirstToken();
            let reasoning = "";
            let reasoningFromContent = "";
            if (content) {
              const rewritten = contentThink.rewrite(content, false);
              if (rewritten.content) {
                visibleContentBuffer += rewritten.content;
              }
              const cleanedContent = stripToolXmlDelta(rewritten.content);
              if (cleanedContent) {
                delta["content"] = cleanedContent;
              } else if ("content" in delta) {
                delete delta["content"];
              }
              reasoningFromContent = rewritten.reasoningAppend;
            } else if (rawContent && "content" in delta) {
              delete delta["content"];
            }

            if (reasoningRaw) {
              const rewrittenReasoning = reasoningThink.rewrite(reasoningRaw, true);
              reasoning = `${reasoning}${rewrittenReasoning.reasoningAppend}`;
            }

            if (reasoningFromContent) {
              reasoning = `${reasoning}${reasoningFromContent}`;
            }

            if (reasoning) {
              delta["reasoning_content"] = stripToolXmlDelta(reasoning);
            } else if (REASONING_FIELDS.some((field) => field in delta)) {
              delete delta["reasoning_content"];
            }
            delete delta["reasoning"];
            delete delta["reasoning_text"];
          }
        }

        for (const outLine of otherLines) {
          enqueueLine(controller, outLine);
        }
        enqueueDataEvent(controller, `data: ${JSON.stringify(parsed)}`);
      };

      if (downstreamClosed) {
        try {
          controller.close();
        } catch {}
        await tearDownUpstream();
        return;
      }

      const emittedBeforePull = emittedLines;

      try {
        while (!downstreamClosed && emittedLines === emittedBeforePull) {
          let result: ReaderResult;
          try {
            result = await reader.read();
          } catch {
            downstreamClosed = true;
            try {
              controller.close();
            } catch {}
            return;
          }
          if (result.done) {
            // Flush any bytes the streaming decoder is still holding (an
            // incomplete multibyte char at the final chunk boundary); without
            // this final decode the trailing character would be dropped.
            buffer += decoder.decode();
            if (buffer) {
              const trailing = buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer;
              if (trailing.length > 0) {
                pendingEventLines.push(trailing);
              }
              buffer = "";
            }
            if (pendingEventLines.length > 0) {
              flushEvent(pendingEventLines);
              pendingEventLines = [];
            }
            flushThinkCarry(controller);
            maybeInjectToolCalls(controller);
            try {
              controller.close();
            } catch {}
            return;
          }

          const chunk = result.value ?? new Uint8Array();
          buffer += decoder.decode(chunk, { stream: true });
          const lines = buffer.split("\n");
          buffer = lines.pop() ?? "";

          for (const line of lines) {
            const normalized = line.endsWith("\r") ? line.slice(0, -1) : line;
            if (normalized === "") {
              flushEvent(pendingEventLines);
              pendingEventLines = [];
              enqueueLine(controller, "");
              continue;
            }
            pendingEventLines.push(normalized);
          }
        }
      } catch {
        downstreamClosed = true;
        await tearDownUpstream();
        try {
          controller.close();
        } catch {}
      }
    },
    async cancel(): Promise<void> {
      downstreamClosed = true;
      await tearDownUpstream();
    },
  });
};