Xây dựng UI chat AI streaming mà vẫn mượt: SSE, ReadableStream, markdown tăng dần và nút Stop

export type ChatEvent =
  | { type: "status"; value: "thinking" | "writing" }
  | { type: "text-delta"; text: string }
  | { type: "tool-call"; id: string; name: string; input: unknown }
  | { type: "tool-result"; id: string; output: unknown; isError: boolean }
  | { type: "error"; message: string; retryable: boolean }
  | { type: "done"; reason: "end_turn" | "max_tokens" | "aborted" | "other" };

const encoder = new TextEncoder();

/** Mã hoá một event theo định dạng text/event-stream. */
export function encodeEvent(event: ChatEvent): Uint8Array {
  // JSON.stringify không sinh ký tự xuống dòng thô, nên một dòng data: là đủ
  return encoder.encode(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`);
}

/** Dòng comment — client bỏ qua, nhưng giữ kết nối không bị proxy cắt. */
export const HEARTBEAT = encoder.encode(": ping\n\n");

import Anthropic from "@anthropic-ai/sdk";
import { encodeEvent, HEARTBEAT, type ChatEvent } from "@/lib/chat-protocol";
import { SUPPORT_TOOLS, runSupportTool } from "@/lib/support-tools";
import { requireSession } from "@/lib/auth";

export const maxDuration = 60;

const client = new Anthropic(); // đọc ANTHROPIC_API_KEY — chỉ tồn tại trên server
const MODEL = process.env.CHAT_MODEL;
const MAX_TOOL_ROUNDS = 3;

type ClientMessage = { role: "user" | "assistant"; text: string };

function isRetryable(err: unknown): boolean {
  if (!(err instanceof Anthropic.APIError)) return false;
  // status undefined: lỗi kết nối hoặc event error giữa stream (vd. overloaded)
  return err.status === undefined || err.status === 429 || err.status >= 500;
}

export async function POST(req: Request) {
  const session = await requireSession(req); // ném 401 nếu chưa đăng nhập
  if (!MODEL) return Response.json({ error: "CHAT_MODEL is not set" }, { status: 500 });

  const { messages } = (await req.json()) as { messages: ClientMessage[] };
  // Chỉ nhận role user/assistant và text — không bao giờ để client gửi role "system"
  const history: Anthropic.MessageParam[] = messages
    .filter((m) => m.role === "user" || m.role === "assistant")
    .map((m) => ({ role: m.role, content: String(m.text).slice(0, 8_000) }));

  // Hai cách biết client đã bỏ đi: req.signal, hoặc response stream bị cancel
  const upstream = new AbortController();
  req.signal.addEventListener("abort", () => upstream.abort(), { once: true });
  let heartbeat: ReturnType<typeof setInterval> | undefined;

  const body = new ReadableStream<Uint8Array>({
    async start(controller) {
      const send = (event: ChatEvent) => {
        if (!upstream.signal.aborted) controller.enqueue(encodeEvent(event));
      };
      heartbeat = setInterval(() => {
        if (!upstream.signal.aborted) controller.enqueue(HEARTBEAT);
      }, 15_000);

      try {
        for (let round = 0; round < MAX_TOOL_ROUNDS; round++) {
          const stream = client.messages.stream(
            { model: MODEL, max_tokens: 4096, system: SYSTEM_PROMPT, tools: SUPPORT_TOOLS, messages: history },
            { signal: upstream.signal },
          );

          for await (const event of stream) {
            if (event.type === "content_block_start") {
              if (event.content_block.type === "thinking") send({ type: "status", value: "thinking" });
              if (event.content_block.type === "text") send({ type: "status", value: "writing" });
            } else if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
              send({ type: "text-delta", text: event.delta.text });
            }
          }

          const message = await stream.finalMessage();
          const calls = message.content.filter((b): b is Anthropic.ToolUseBlock => b.type === "tool_use");
          if (message.stop_reason !== "tool_use" || calls.length === 0) {
            const r = message.stop_reason;
            send({ type: "done", reason: r === "end_turn" || r === "max_tokens" ? r : "other" });
            return;
          }

          history.push({ role: "assistant", content: message.content });
          const results: Anthropic.ToolResultBlockParam[] = [];
          for (const call of calls) {
            send({ type: "tool-call", id: call.id, name: call.name, input: call.input });
            // Tool chạy với quyền của session, không tin input của model về "ai là chủ đơn"
            const { output, isError } = await runSupportTool(call, session, upstream.signal);
            send({ type: "tool-result", id: call.id, output, isError });
            results.push({ type: "tool_result", tool_use_id: call.id, content: JSON.stringify(output), is_error: isError });
          }
          history.push({ role: "user", content: results });
        }
        send({ type: "error", message: "Too many tool steps", retryable: false });
      } catch (err) {
        if (upstream.signal.aborted) return; // người dùng bấm Stop — không phải lỗi
        send({ type: "error", message: "The assistant is unavailable", retryable: isRetryable(err) });
      } finally {
        clearInterval(heartbeat);
        try { controller.close(); } catch { /* stream đã bị cancel */ }
      }
    },
    cancel() {
      clearInterval(heartbeat);
      upstream.abort();
    },
  });

  return new Response(body, {
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      "X-Accel-Buffering": "no",
    },
  });
}

const SYSTEM_PROMPT =
  "You are the support assistant for our store. Use lookup_order for order questions. Answer in the user's language.";

{
  "functions": {
    "app/api/**/*": { "supportsCancellation": true }
  }
}

import type { ChatEvent } from "./chat-protocol";

export type SSEMessage = { event: string; data: string };

export function createSSEParser(): TransformStream<string, SSEMessage> {
  let buffer = "";
  let pendingCR = false; // chunk trước kết thúc bằng \r — có thể \n đang ở chunk sau
  let eventName = "";
  let data: string[] = [];

  const processLine = (line: string, out: TransformStreamDefaultController<SSEMessage>) => {
    if (line === "") {
      if (data.length > 0) out.enqueue({ event: eventName || "message", data: data.join("\n") });
      eventName = "";
      data = [];
      return;
    }
    if (line.startsWith(":")) return; // comment, ví dụ heartbeat
    const colon = line.indexOf(":");
    const field = colon === -1 ? line : line.slice(0, colon);
    let value = colon === -1 ? "" : line.slice(colon + 1);
    if (value.startsWith(" ")) value = value.slice(1);
    if (field === "event") eventName = value;
    else if (field === "data") data.push(value);
  };

  return new TransformStream({
    transform(chunk, out) {
      if (pendingCR && chunk.startsWith("\n")) chunk = chunk.slice(1); // CRLF bị cắt đôi
      pendingCR = chunk.endsWith("\r");
      buffer += chunk;
      const lines = buffer.split(/\r\n|\r|\n/);
      buffer = lines.pop() ?? ""; // dòng chưa hoàn chỉnh, chờ chunk sau
      for (const line of lines) processLine(line, out);
    },
    // flush: event dở dang khi stream đóng bị bỏ, đúng theo spec
  });
}

export async function* readChatEvents(response: Response): AsyncGenerator<ChatEvent> {
  if (!response.body) throw new Error("Response has no body");
  const reader = response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(createSSEParser())
    .getReader();
  try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) return;
      yield JSON.parse(value.data) as ChatEvent;
    }
  } finally {
    reader.cancel().catch(() => {}); // người dùng break sớm → huỷ cả pipeline
  }
}

import type { ChatEvent } from "./chat-protocol";

export type Part =
  | { type: "text"; text: string }
  | { type: "tool-call"; id: string; name: string; input: unknown; state: "running" | "done" | "error"; output?: unknown }
  | { type: "error"; message: string; retryable: boolean };

export type Message = {
  id: string;
  role: "user" | "assistant";
  status: "submitted" | "thinking" | "streaming" | "done" | "stopped" | "error";
  parts: Part[];
};

export type Action =
  | { type: "submit"; user: Message; assistantId: string }
  | { type: "events"; id: string; events: ChatEvent[] } // cả một frame event một lúc
  | { type: "stopped"; id: string }
  | { type: "failed"; id: string; message: string; retryable: boolean }
  | { type: "retry"; id: string };

function applyEvent(msg: Message, e: ChatEvent): Message {
  switch (e.type) {
    case "status":
      return { ...msg, status: e.value === "thinking" ? "thinking" : "streaming" };
    case "text-delta": {
      const parts = msg.parts.slice();
      const last = parts.at(-1);
      // Nối vào text part cuối; sau một tool call thì mở text part mới
      if (last?.type === "text") parts[parts.length - 1] = { ...last, text: last.text + e.text };
      else parts.push({ type: "text", text: e.text });
      return { ...msg, status: "streaming", parts };
    }
    case "tool-call":
      return { ...msg, parts: [...msg.parts, { type: "tool-call", id: e.id, name: e.name, input: e.input, state: "running" }] };
    case "tool-result":
      return {
        ...msg,
        parts: msg.parts.map((p) =>
          p.type === "tool-call" && p.id === e.id ? { ...p, state: e.isError ? "error" : "done", output: e.output } : p,
        ),
      };
    case "error":
      return { ...msg, status: "error", parts: [...msg.parts, { type: "error", message: e.message, retryable: e.retryable }] };
    case "done":
      return { ...msg, status: "done" };
  }
}

export function chatReducer(state: Message[], action: Action): Message[] {
  switch (action.type) {
    case "submit":
      return [...state, action.user, { id: action.assistantId, role: "assistant", status: "submitted", parts: [] }];
    case "events":
      return state.map((m) => (m.id === action.id ? action.events.reduce(applyEvent, m) : m));
    case "stopped":
      return state.map((m) => (m.id === action.id && m.status !== "done" ? { ...m, status: "stopped" } : m));
    case "failed":
      return state.map((m) =>
        m.id === action.id
          ? { ...m, status: "error", parts: [...m.parts, { type: "error", message: action.message, retryable: action.retryable }] }
          : m,
      );
    case "retry":
      return state.map((m) => (m.id === action.id ? { ...m, status: "submitted", parts: [] } : m));
  }
}

/** Thứ duy nhất gửi lên server: role + text. */
export function toWire(messages: Message[]) {
  return messages
    .map((m) => ({ role: m.role, text: m.parts.flatMap((p) => (p.type === "text" ? [p.text] : [])).join("") }))
    .filter((m) => m.text.length > 0);
}

"use client";
import { useCallback, useEffect, useReducer, useRef } from "react";
import type { ChatEvent } from "@/lib/chat-protocol";
import { chatReducer, toWire, type Message } from "@/lib/chat-state";
import { readChatEvents } from "@/lib/sse";

class HttpError extends Error {
  constructor(readonly status: number) {
    super(`HTTP ${status}`);
  }
}
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const ACTIVE = new Set<Message["status"]>(["submitted", "thinking", "streaming"]);

export function useChatStream(endpoint = "/api/chat") {
  const [messages, dispatch] = useReducer(chatReducer, []);
  const messagesRef = useRef(messages);
  useEffect(() => {
    messagesRef.current = messages; // send/retry đọc qua ref → identity ổn định cho memo
  });
  const abortRef = useRef<AbortController | null>(null);
  const pending = useRef<{ id: string; events: ChatEvent[] } | null>(null);
  const frame = useRef<number | null>(null);

  const flush = useCallback(() => {
    if (frame.current !== null) cancelAnimationFrame(frame.current);
    frame.current = null;
    const batch = pending.current;
    pending.current = null;
    if (batch) dispatch({ type: "events", id: batch.id, events: batch.events });
  }, []);

  const enqueue = useCallback(
    (id: string, event: ChatEvent) => {
      (pending.current ??= { id, events: [] }).events.push(event);
      frame.current ??= requestAnimationFrame(flush); // tối đa một dispatch mỗi frame
    },
    [flush],
  );

  const run = useCallback(
    async (assistantId: string, history: Message[]) => {
      const controller = new AbortController();
      abortRef.current = controller;
      let received = 0;
      let finished = false;

      for (let attempt = 0; ; attempt++) {
        try {
          const res = await fetch(endpoint, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ messages: toWire(history) }),
            signal: controller.signal,
          });
          if (!res.ok) throw new HttpError(res.status);
          for await (const event of readChatEvents(res)) {
            received++;
            if (event.type === "done" || event.type === "error") finished = true;
            enqueue(assistantId, event);
          }
          // Stream đóng êm mà không có "done" = bị cắt giữa đường (proxy timeout, deploy...)
          if (!finished) throw new Error("Stream ended unexpectedly");
          break;
        } catch (err) {
          if (controller.signal.aborted) break;
          const retryable = !(err instanceof HttpError) || err.status === 429 || err.status >= 500;
          // Chỉ tự retry khi người dùng chưa thấy chữ nào — tránh chữ bị lặp
          if (received === 0 && retryable && attempt < 2) {
            await sleep(500 * 2 ** attempt);
            continue;
          }
          flush();
          dispatch({ type: "failed", id: assistantId, message: "Connection lost", retryable });
          break;
        }
      }
      flush(); // tab ẩn thì rAF không chạy — flush đồng bộ khi kết thúc
      if (controller.signal.aborted) dispatch({ type: "stopped", id: assistantId });
      if (abortRef.current === controller) abortRef.current = null;
    },
    [endpoint, enqueue, flush],
  );

  const send = useCallback(
    (text: string) => {
      abortRef.current?.abort(); // mỗi lúc chỉ một câu trả lời
      const user: Message = { id: crypto.randomUUID(), role: "user", status: "done", parts: [{ type: "text", text }] };
      const assistantId = crypto.randomUUID();
      dispatch({ type: "submit", user, assistantId });
      void run(assistantId, [...messagesRef.current, user]);
    },
    [run],
  );

  const retry = useCallback(
    (assistantId: string) => {
      const all = messagesRef.current;
      const index = all.findIndex((m) => m.id === assistantId);
      if (index === -1) return;
      dispatch({ type: "retry", id: assistantId });
      void run(assistantId, all.slice(0, index));
    },
    [run],
  );

  const stop = useCallback(() => abortRef.current?.abort(), []);
  const last = messages.at(-1);
  const isStreaming = last?.role === "assistant" && ACTIVE.has(last.status);

  return { messages, send, stop, retry, isStreaming };
}

import { marked } from "marked";

export type BlockState = { source: string; stable: string[]; stableLength: number; tail: string };
export const EMPTY_BLOCKS: BlockState = { source: "", stable: [], stableLength: 0, tail: "" };

export function splitBlocks(prev: BlockState, text: string): BlockState {
  // Văn bản không còn nối tiếp phần đã ổn định (retry, sửa message) → làm lại từ đầu
  const base = text.startsWith(prev.source.slice(0, prev.stableLength)) ? prev : EMPTY_BLOCKS;
  const tokens = marked.lexer(text.slice(base.stableLength));

  let last = tokens.length - 1; // token nội dung cuối cùng = phần đuôi đang chạy
  while (last >= 0 && tokens[last].type === "space") last--;

  const stable = base.stable.slice();
  let consumed = 0;
  for (let i = 0; i < last; i++) {
    if (tokens[i].type !== "space") stable.push(tokens[i].raw);
    consumed += tokens[i].raw.length;
  }
  const stableLength = base.stableLength + consumed;
  return { source: text, stable, stableLength, tail: text.slice(stableLength) };
}

const FENCE = /^ {0,3}(`{3,}|~{3,})/gm;

/** Làm phần đuôi "trông như đã xong" — chỉ để render, không sửa dữ liệu. */
export function prepareTail(tail: string): { markdown: string; inCode: boolean } {
  if ((tail.match(FENCE)?.length ?? 0) % 2 === 1) {
    return { markdown: tail + "\n" + "`".repeat(3), inCode: true }; // fence chưa đóng
  }
  let md = tail.replace(/!?\[([^\]]*)\]\([^)]*$/, "$1"); // link/ảnh viết dở → chỉ giữ text
  if ((md.match(/\*\*/g)?.length ?? 0) % 2 === 1) md += "**"; // in đậm chưa đóng
  return { markdown: md, inCode: false };
}

"use client";
import { memo, useState } from "react";
import Markdown, { type Components } from "react-markdown";
import remarkGfm from "remark-gfm";
import rehypeHighlight from "rehype-highlight";
import { EMPTY_BLOCKS, prepareTail, splitBlocks } from "@/lib/markdown-blocks";

const REMARK = [remarkGfm];
const REHYPE = [rehypeHighlight];

function externalHost(href: string): string | null {
  try {
    return new URL(href).host; // URL tương đối ném lỗi → coi là nội bộ
  } catch {
    return null;
  }
}

const components: Components = {
  a: ({ href, children }) => (
    <a href={href} target="_blank" rel="noopener noreferrer nofollow">
      {children}
      {href && externalHost(href) && <span className="link-host"> ({externalHost(href)})</span>}
    </a>
  ),
  // Không tự tải ảnh do model sinh ra: URL ảnh có thể mang dữ liệu hội thoại ra ngoài
  img: ({ alt }) => <span className="img-placeholder">[{alt || "image"}]</span>,
};

const Block = memo(function Block({ source, live }: { source: string; live: boolean }) {
  return (
    <Markdown remarkPlugins={REMARK} rehypePlugins={live ? [] : REHYPE} components={components} skipHtml>
      {source}
    </Markdown>
  );
});

export function StreamingMarkdown({ text, streaming }: { text: string; streaming: boolean }) {
  const [blocks, setBlocks] = useState(() => splitBlocks(EMPTY_BLOCKS, text));
  let current = blocks;
  if (blocks.source !== text) {
    current = splitBlocks(blocks, text);
    setBlocks(current); // pattern "lưu thông tin từ lần render trước" của React
  }
  const tail = streaming ? prepareTail(current.tail).markdown : current.tail;

  return (
    <div className="prose">
      {current.stable.map((source, i) => (
        <Block key={i} source={source} live={false} />
      ))}
      {tail && <Block source={tail} live={streaming} />}
    </div>
  );
}

"use client";
import { useCallback, useEffect, useRef, useState } from "react";

const THRESHOLD = 48; // px

export function useStickToBottom<T extends HTMLElement>() {
  const scrollRef = useRef<T>(null);
  const contentRef = useRef<HTMLDivElement>(null);
  const pinned = useRef(true);
  const [showJump, setShowJump] = useState(false);

  useEffect(() => {
    const scroller = scrollRef.current;
    const content = contentRef.current;
    if (!scroller || !content) return;
    let lastTop = scroller.scrollTop;

    const onScroll = () => {
      const distance = scroller.scrollHeight - scroller.scrollTop - scroller.clientHeight;
      if (scroller.scrollTop < lastTop && distance > THRESHOLD) pinned.current = false; // người dùng cuộn lên
      if (distance <= THRESHOLD) pinned.current = true;
      lastTop = scroller.scrollTop;
      setShowJump(!pinned.current);
    };
    // Bắt mọi thay đổi chiều cao: chữ mới, khối code, card tool, ảnh placeholder
    const observer = new ResizeObserver(() => {
      if (pinned.current) scroller.scrollTop = scroller.scrollHeight;
    });

    observer.observe(content);
    scroller.addEventListener("scroll", onScroll, { passive: true });
    return () => {
      observer.disconnect();
      scroller.removeEventListener("scroll", onScroll);
    };
  }, []);

  const jumpToBottom = useCallback(() => {
    pinned.current = true;
    scrollRef.current?.scrollTo({ top: scrollRef.current.scrollHeight, behavior: "smooth" });
  }, []);

  return { scrollRef, contentRef, showJump, jumpToBottom };
}

import type { Message } from "@/lib/chat-state";

const STATUS_TEXT: Partial<Record<Message["status"], string>> = {
  thinking: "Trợ lý đang suy nghĩ",
  streaming: "Trợ lý đang trả lời",
  stopped: "Đã dừng câu trả lời",
  error: "Câu trả lời gặp lỗi",
};

export function ChatAnnouncer({ message }: { message?: Message }) {
  let text = "";
  if (message?.role === "assistant") {
    text =
      message.status === "done"
        ? message.parts
            .flatMap((p) => (p.type === "text" ? [p.text] : []))
            .join("")
            .replace(/[*_`#>|]/g, "") // bỏ ký hiệu markdown để không bị đọc thành tiếng
        : (STATUS_TEXT[message.status] ?? "");
  }
  return (
    <div aria-live="polite" aria-atomic="true" className="sr-only">
      {text}
    </div>
  );
}

import type { Part } from "@/lib/chat-state";

type ToolCall = Extract<Part, { type: "tool-call" }>;
type Order = { id: string; status: "processing" | "shipping" | "delivered"; eta?: string };

const ORDER_STATUS: Record<Order["status"], string> = {
  processing: "Đang xử lý",
  shipping: "Đang giao",
  delivered: "Đã giao",
};

export function ToolPart({ part }: { part: ToolCall }) {
  if (part.state === "running") {
    const orderId = (part.input as { orderId?: string }).orderId;
    return (
      <div className="tool-card" aria-busy="true">
        {part.name === "lookup_order" ? `Đang tra cứu đơn ${orderId ?? ""}…` : "Đang xử lý…"}
      </div>
    );
  }
  if (part.state === "error") {
    return <div className="tool-card tool-card--error">Không tra cứu được — trợ lý sẽ giải thích bên dưới.</div>;
  }
  if (part.name !== "lookup_order") return null; // tool không có UI riêng: để model diễn giải bằng text

  const order = part.output as Order; // code thật: validate bằng zod trước khi tin
  return (
    <dl className="order-card">
      <dt>Đơn hàng</dt>
      <dd>{order.id}</dd>
      <dt>Trạng thái</dt>
      <dd>{ORDER_STATUS[order.status]}</dd>
      {order.eta && (
        <>
          <dt>Dự kiến giao</dt>
          <dd>{order.eta}</dd>
        </>
      )}
    </dl>
  );
}

// app/api/chat/route.ts — phiên bản dùng AI SDK 7
import { anthropic } from "@ai-sdk/anthropic";
import { convertToModelMessages, createUIMessageStreamResponse, streamText, toUIMessageStream, type UIMessage } from "ai";

export async function POST(req: Request) {
  const { messages }: { messages: UIMessage[] } = await req.json();
  const result = streamText({
    model: anthropic(process.env.CHAT_MODEL!),
    instructions: "You are the support assistant for our store.",
    messages: await convertToModelMessages(messages),
    abortSignal: req.signal, // vẫn phải tự nối abort lên upstream
  });
  return createUIMessageStreamResponse({ stream: toUIMessageStream({ stream: result.stream }) });
}

// components/SupportChat.tsx (client)
// const { messages, sendMessage, status, stop, regenerate } = useChat({
//   transport: new DefaultChatTransport({ api: "/api/chat" }),
//   throttle: 50,
// });

"use client";
import { memo, useState } from "react";
import { useChatStream } from "@/hooks/useChatStream";
import { useStickToBottom } from "@/hooks/useStickToBottom";
import type { Message } from "@/lib/chat-state";
import { ChatAnnouncer } from "./ChatAnnouncer";
import { StreamingMarkdown } from "./StreamingMarkdown";
import { ToolPart } from "./ToolPart";

const MessageItem = memo(function MessageItem({ message, onRetry }: { message: Message; onRetry: (id: string) => void }) {
  const active = message.status === "submitted" || message.status === "thinking" || message.status === "streaming";
  return (
    <article aria-label={message.role === "user" ? "Bạn" : "Trợ lý"} aria-busy={active} className={`msg msg--${message.role}`}>
      {message.status === "thinking" && <p className="msg-status">Đang suy nghĩ…</p>}
      {message.parts.map((part, i) => {
        if (part.type === "tool-call") return <ToolPart key={part.id} part={part} />;
        if (part.type === "error")
          return (
            <p key={i} className="msg-error">
              Trợ lý đang gặp sự cố.{" "}
              {part.retryable && <button onClick={() => onRetry(message.id)}>Thử lại</button>}
            </p>
          );
        return message.role === "user" ? (
          <p key={i}>{part.text}</p>
        ) : (
          <StreamingMarkdown key={i} text={part.text} streaming={active && i === message.parts.length - 1} />
        );
      })}
      {message.status === "stopped" && <p className="msg-status">Đã dừng.</p>}
    </article>
  );
});

export function SupportChat() {
  const { messages, send, stop, retry, isStreaming } = useChatStream();
  const { scrollRef, contentRef, showJump, jumpToBottom } = useStickToBottom<HTMLDivElement>();
  const [input, setInput] = useState("");

  return (
    <section className="chat" aria-label="Trợ lý hỗ trợ khách hàng">
      <div ref={scrollRef} className="chat-scroll">
        <div ref={contentRef}>
          {messages.map((m) => (
            <MessageItem key={m.id} message={m} onRetry={retry} />
          ))}
        </div>
      </div>
      {showJump && <button className="chat-jump" onClick={jumpToBottom}>Xuống tin mới nhất</button>}
      <ChatAnnouncer message={messages.at(-1)} />
      <form
        onSubmit={(e) => {
          e.preventDefault();
          const text = input.trim();
          if (!text || isStreaming) return;
          send(text);
          setInput("");
        }}
      >
        <label htmlFor="chat-input" className="sr-only">Câu hỏi của bạn</label>
        <textarea id="chat-input" value={input} onChange={(e) => setInput(e.target.value)} />
        <button type="submit" disabled={isStreaming}>Gửi</button>
        <button type="button" onClick={stop} disabled={!isStreaming}>Dừng</button>
      </form>
    </section>
  );
}

import { describe, expect, it } from "vitest";
import { encodeEvent, HEARTBEAT, type ChatEvent } from "./chat-protocol";
import { readChatEvents } from "./sse";

const EVENTS: ChatEvent[] = [
  { type: "status", value: "writing" },
  { type: "text-delta", text: "Đơn hàng của bạn đang được giao 🚚" },
  { type: "done", reason: "end_turn" },
];

/** Body giả: cắt dữ liệu tại các vị trí byte cho trước. */
function fakeResponse(bytes: Uint8Array, cuts: number[]): Response {
  const body = new ReadableStream<Uint8Array>({
    start(controller) {
      let prev = 0;
      for (const cut of [...cuts, bytes.length]) {
        controller.enqueue(bytes.slice(prev, cut));
        prev = cut;
      }
      controller.close();
    },
  });
  return new Response(body, { headers: { "Content-Type": "text/event-stream" } });
}

async function collect(res: Response): Promise<ChatEvent[]> {
  const out: ChatEvent[] = [];
  for await (const event of readChatEvents(res)) out.push(event);
  return out;
}

describe("readChatEvents", async () => {
  const parts = [encodeEvent(EVENTS[0]), HEARTBEAT, encodeEvent(EVENTS[1]), encodeEvent(EVENTS[2])];
  const bytes = new Uint8Array(await new Blob(parts).arrayBuffer());

  it("ráp đúng event dù chunk bị cắt ở bất kỳ byte nào", async () => {
    for (let cut = 1; cut < bytes.length; cut++) {
      expect(await collect(fakeResponse(bytes, [cut]))).toEqual(EVENTS);
    }
  });

  it("chịu được mỗi chunk chỉ có một byte", async () => {
    const cuts = Array.from({ length: bytes.length - 1 }, (_, i) => i + 1);
    expect(await collect(fakeResponse(bytes, cuts))).toEqual(EVENTS);
  });
});