/**
 * StreamBufferManager — per-session 流式事件节流缓冲
 *
 * WS 事件到达时写入 buffer（纯 JS 对象，不触发 React），
 * 每 FLUSH_INTERVAL ms 批量 flush 到 Zustand store。
 *
 * 设计为 singleton，不依赖 React 组件生命周期。
 * app-ws-shim 直接调用 streamBufferManager.handle(msg)。
 */

import type { ChatListItem, ChatMessage, ContentBlock } from '../stores/chat-types';
import { useStore } from '../stores';
import { renderMarkdown } from '../utils/markdown';
import { cleanMoodText } from '../utils/message-parser';
import { findOpenToolIndex, toolCallFromStartEvent, toolCallIdFromEvent } from '../utils/tool-call-identity';
import {
  registerStreamBufferInvalidator,
  registerStreamBufferSnapshot,
  type StreamBufferSnapshot,
} from '../stores/stream-invalidator';
import { bumpMessageLiveVersion } from '../stores/message-live-version';

/* eslint-disable @typescript-eslint/no-explicit-any -- 流式消息 handle(msg) 接收动态 JSON */

const STREAM_FLUSH_FPS = 30;
const FLUSH_INTERVAL = Math.round(1000 / STREAM_FLUSH_FPS);
let streamMessageSeq = 0;
type InterludeContentBlock = Extract<ContentBlock, { type: 'interlude' }>;

function nextStreamMessageId(): string {
  streamMessageSeq = (streamMessageSeq + 1) % Number.MAX_SAFE_INTEGER;
  return `stream-${Date.now()}-${streamMessageSeq}`;
}

interface Buffer {
  sessionPath: string;
  textAcc: string;
  thinkingAcc: string;
  moodAcc: string;
  moodYuan: string;
  inThinking: boolean;
  inMood: boolean;
  inCard: boolean;
  cardAttrs: { type: string; plugin: string; route: string; title?: string } | null;
  cardDescAcc: string;
  lastFlushTime: number;
  flushTimer: ReturnType<typeof setTimeout> | null;
  /** 当前 turn 绑定的 assistant message id */
  messageId: string | null;
  pendingInterludesByTaskId: Map<string, InterludeContentBlock[]>;
}

function createBuffer(sessionPath: string): Buffer {
  return {
    sessionPath,
    textAcc: '',
    thinkingAcc: '',
    moodAcc: '',
    moodYuan: 'hanako',
    inThinking: false,
    inMood: false,
    inCard: false,
    cardAttrs: null,
    cardDescAcc: '',
    lastFlushTime: 0,
    flushTimer: null,
    messageId: null,
    pendingInterludesByTaskId: new Map(),
  };
}

function resolveSessionYuan(sessionPath: string): string {
  const state = useStore.getState();
  const sessionAgentId = state.sessions.find((session: any) => session.path === sessionPath)?.agentId ?? null;
  if (!sessionAgentId) return 'hanako';
  return state.agents.find((agent: any) => agent.id === sessionAgentId)?.yuan || 'hanako';
}

class StreamBufferManager {
  private buffers = new Map<string, Buffer>();

  /** 获取或创建 session buffer */
  private getBuffer(sessionPath: string): Buffer {
    let buf = this.buffers.get(sessionPath);
    if (!buf) {
      buf = createBuffer(sessionPath);
      this.buffers.set(sessionPath, buf);
    }
    return buf;
  }

  private hasTurnState(buf: Buffer): boolean {
    return !!(
      buf.messageId ||
      buf.textAcc ||
      buf.thinkingAcc ||
      buf.moodAcc ||
      buf.inThinking ||
      buf.inMood ||
      buf.inCard ||
      buf.cardAttrs ||
      buf.cardDescAcc
    );
  }

  private resetTurnState(buf: Buffer): void {
    if (buf.flushTimer) {
      clearTimeout(buf.flushTimer);
      buf.flushTimer = null;
    }
    buf.textAcc = '';
    buf.thinkingAcc = '';
    buf.moodAcc = '';
    buf.inThinking = false;
    buf.inMood = false;
    buf.inCard = false;
    buf.cardAttrs = null;
    buf.cardDescAcc = '';
    buf.messageId = null;
  }

  private finishBufferTurn(buf: Buffer): void {
    if (this.hasTurnState(buf)) {
      this.flush(buf);
    } else if (buf.flushTimer) {
      clearTimeout(buf.flushTimer);
      buf.flushTimer = null;
    }
    this.resetTurnState(buf);
  }

  /** 确保 store 中已存在当前 turn 绑定的 assistant message */
  private ensureMessage(buf: Buffer): void {
    const store = useStore.getState();
    const session = store.chatSessions[buf.sessionPath];
    if (!session) return; // session 未初始化（loadMessages 尚未完成）

    const targetId = buf.messageId || trailingDeferredTextRebindTargetId(session.items);
    const existing = targetId
      ? session.items.find((item) =>
        item.type === 'message' &&
        item.data.id === targetId &&
        item.data.role === 'assistant',
      )
      : null;
    if (existing) {
      buf.messageId = targetId;
      return;
    }

    const id = targetId || nextStreamMessageId();
    const msg: ChatMessage = { id, role: 'assistant', blocks: [], timestamp: Date.now() };
    store.appendItem(buf.sessionPath, { type: 'message', data: msg });
    bumpMessageLiveVersion(buf.sessionPath);
    buf.messageId = id;
  }

  private updateTargetMessage(buf: Buffer, updater: (msg: ChatMessage) => ChatMessage): void {
    this.ensureMessage(buf);
    if (!buf.messageId) return;
    const updated = useStore.getState().updateMessageById(buf.sessionPath, buf.messageId, updater);
    if (!updated) {
      console.warn('[stream] target assistant message missing after ensureMessage:', buf.sessionPath, buf.messageId);
      return;
    }
    bumpMessageLiveVersion(buf.sessionPath);
  }

  private tryInsertInterlude(buf: Buffer, block: InterludeContentBlock): boolean {
    const consumed = useStore.getState().insertInterludeItemNearTaskResult(buf.sessionPath, block.taskId || null, block);
    if (consumed) bumpMessageLiveVersion(buf.sessionPath);
    return consumed;
  }

  private queuePendingInterlude(buf: Buffer, block: InterludeContentBlock): void {
    const taskId = block.taskId;
    if (!taskId) return;
    const existing = buf.pendingInterludesByTaskId.get(taskId) || [];
    const alreadyQueued = existing.some((item) => (
      (block.id && item.id === block.id) ||
      (!!block.taskId && item.taskId === block.taskId && item.status === block.status)
    ));
    if (alreadyQueued) return;
    buf.pendingInterludesByTaskId.set(taskId, [...existing, block]);
  }

  private drainPendingInterludesForTask(buf: Buffer, taskId: string | null): void {
    if (!taskId) return;
    const pending = buf.pendingInterludesByTaskId.get(taskId);
    if (!pending?.length) return;

    const remaining: InterludeContentBlock[] = [];
    for (const block of pending) {
      if (!this.tryInsertInterlude(buf, block)) remaining.push(block);
    }
    if (remaining.length > 0) {
      buf.pendingInterludesByTaskId.set(taskId, remaining);
    } else {
      buf.pendingInterludesByTaskId.delete(taskId);
    }
  }

  private drainPendingInterludesForBlock(buf: Buffer, block: ContentBlock): void {
    this.drainPendingInterludesForTask(buf, interludeAnchorTaskId(block));
  }

  /** 调度节流 flush */
  private scheduleFlush(buf: Buffer): void {
    const now = Date.now();
    if (now - buf.lastFlushTime >= FLUSH_INTERVAL) {
      this.flush(buf);
    } else if (!buf.flushTimer) {
      buf.flushTimer = setTimeout(() => {
        buf.flushTimer = null;
        this.flush(buf);
      }, FLUSH_INTERVAL - (now - buf.lastFlushTime));
    }
  }

  /** 把 buffer 中累积的内容一次性 flush 到 Zustand */
  private flush(buf: Buffer): void {
    buf.lastFlushTime = Date.now();
    if (buf.flushTimer) {
      clearTimeout(buf.flushTimer);
      buf.flushTimer = null;
    }

    this.updateTargetMessage(buf, (msg) => {
      const blocks = [...(msg.blocks || [])];

      // ── Thinking ──
      if (buf.thinkingAcc || buf.inThinking) {
        const idx = blocks.findIndex(b => b.type === 'thinking');
        const thinkingBlock: ContentBlock = {
          type: 'thinking',
          content: buf.thinkingAcc,
          sealed: !buf.inThinking,
        };
        if (idx >= 0) blocks[idx] = thinkingBlock;
        else blocks.unshift(thinkingBlock); // thinking 在最前面
      }

      // ── Mood ──
      if (buf.moodAcc || buf.inMood) {
        const idx = blocks.findIndex(b => b.type === 'mood');
        const moodBlock: ContentBlock = {
          type: 'mood',
          yuan: buf.moodYuan,
          text: buf.inMood ? buf.moodAcc : cleanMoodText(buf.moodAcc),
        };
        if (idx >= 0) blocks[idx] = moodBlock;
        else {
          // mood 在 thinking 后面
          const insertAt = blocks.findIndex(b => b.type !== 'thinking');
          blocks.splice(insertAt >= 0 ? insertAt : blocks.length, 0, moodBlock);
        }
      }

      // ── Text ──
      if (buf.textAcc) {
        const displayText = buf.textAcc.replace(/<tool_code>[\s\S]*?<\/tool_code>\s*/g, '');
        const html = renderMarkdown(displayText);
        const idx = blocks.findIndex(b => b.type === 'text');
        if (idx >= 0) {
          blocks[idx] = { type: 'text', html, source: displayText };
        } else {
          blocks.push({ type: 'text', html, source: displayText });
        }
      }

      return { ...msg, blocks };
    });
  }

  // ── 公开事件处理器 ──

  handle(msg: any): void {
    const sessionPath = msg.sessionPath;
    if (!sessionPath) {
      console.warn('[ws] stream event missing sessionPath:', msg.type);
      return;
    }
    const buf = this.getBuffer(sessionPath);

    switch (msg.type) {
      case 'text_delta':
        this.ensureMessage(buf);
        buf.textAcc += msg.delta || '';
        this.scheduleFlush(buf);
        break;

      case 'thinking_start':
        this.ensureMessage(buf);
        buf.inThinking = true;
        buf.thinkingAcc = '';
        this.flush(buf);
        break;

      case 'thinking_delta':
        buf.thinkingAcc += msg.delta || '';
        // 与 text/mood 共用时间节流，避免思考流只能在结束后显示。
        this.scheduleFlush(buf);
        break;

      case 'thinking_end':
        buf.inThinking = false;
        this.flush(buf);
        break;

      case 'mood_start':
        this.ensureMessage(buf);
        buf.inMood = true;
        buf.moodAcc = '';
        buf.moodYuan = resolveSessionYuan(sessionPath);
        this.flush(buf);
        break;

      case 'mood_text':
        buf.moodAcc += msg.delta || '';
        this.scheduleFlush(buf);
        break;

      case 'mood_end':
        buf.inMood = false;
        this.flush(buf);
        break;

      case 'card_start':
        this.ensureMessage(buf);
        buf.inCard = true;
        buf.cardAttrs = msg.attrs || null;
        buf.cardDescAcc = '';
        break;

      case 'card_text':
        buf.cardDescAcc += msg.delta || '';
        break;

      case 'card_end': {
        buf.inCard = false;
        if (buf.cardAttrs) {
          this.flush(buf); // flush pending text first
          const card = {
            type: buf.cardAttrs.type || 'iframe',
            pluginId: buf.cardAttrs.plugin || '',
            route: buf.cardAttrs.route || '',
            title: buf.cardAttrs.title,
            description: buf.cardDescAcc,
          };
          this.updateTargetMessage(buf, (m) => ({
            ...m,
            blocks: [...(m.blocks || []), { type: 'plugin_card' as const, card }],
          }));
        }
        buf.cardAttrs = null;
        buf.cardDescAcc = '';
        break;
      }

      case 'tool_start':
        this.ensureMessage(buf);
        // 工具事件频率低，直接写 store
        this.flush(buf); // 先 flush 文本
        this.updateTargetMessage(buf, (m) => {
          const blocks = [...(m.blocks || [])];
          // 找最后一个 tool_group 或创建新的
          let lastTg = blocks.length - 1;
          while (lastTg >= 0 && blocks[lastTg].type !== 'tool_group') lastTg--;
          if (lastTg >= 0 && blocks[lastTg].type === 'tool_group') {
            const tg = blocks[lastTg] as Extract<ContentBlock, { type: 'tool_group' }>;
            // 如果上一个 group 里还有未完成的工具，追加到同一个 group
            if (tg.tools.some(t => !t.done)) {
              blocks[lastTg] = {
                ...tg,
                tools: [...tg.tools, toolCallFromStartEvent(msg)],
              };
              return { ...m, blocks };
            }
          }
          // 新建 tool_group
          blocks.push({
            type: 'tool_group',
            tools: [toolCallFromStartEvent(msg)],
            collapsed: false,
          });
          return { ...m, blocks };
        });
        break;

      case 'tool_end':
        this.updateTargetMessage(buf, (m) => {
          const blocks = [...(m.blocks || [])];
          // 从后往前找含该 tool 名且未 done 的
          for (let i = blocks.length - 1; i >= 0; i--) {
            if (blocks[i].type !== 'tool_group') continue;
            const tg = blocks[i] as Extract<ContentBlock, { type: 'tool_group' }>;
            const toolIdx = findOpenToolIndex(tg.tools, msg);
            if (toolIdx >= 0) {
              const tools = [...tg.tools];
              const id = toolCallIdFromEvent(msg);
              tools[toolIdx] = {
                ...tools[toolIdx],
                ...(id ? { id } : {}),
                done: true,
                success: !!msg.success,
                details: msg.details,
              };
              const allDone = tools.every(t => t.done);
              blocks[i] = { ...tg, tools, collapsed: allDone && tools.length > 1 };
              return { ...m, blocks };
            }
          }
          return m;
        });
        break;

      case 'content_block': {
        let block = msg.block;
        // Apply cached patches (block_update 可能先于 content_block 到达)
        if (block.taskId) {
          const pending = (useStore.getState() as any)._pendingBlockPatches;
          const cached = pending?.[block.taskId];
          if (cached) {
            block = { ...block, ...cached };
            delete pending[block.taskId];
          }
        }

        if (isInterludeBlock(block)) {
          if (this.hasTurnState(buf)) this.flush(buf);
          if (this.tryInsertInterlude(buf, block)) break;
          this.queuePendingInterlude(buf, block);
          break;
        }

        const taskId = replacementTaskId(block);
        if (taskId) {
          if (this.hasTurnState(buf)) this.flush(buf);
          const consumed = useStore.getState().resolveBlockByTaskId(buf.sessionPath, taskId, block);
          if (consumed) {
            this.drainPendingInterludesForBlock(buf, block);
            bumpMessageLiveVersion(buf.sessionPath);
            break;
          }
        }

        this.ensureMessage(buf);
        this.flush(buf);
        this.updateTargetMessage(buf, (m) => ({
          ...m,
          blocks: mergeContentBlock([...(m.blocks || [])], block),
        }));
        this.drainPendingInterludesForBlock(buf, block);
        break;
      }

      case 'compaction_start':
        break;

      case 'compaction_end':
        break;

      case 'turn_end':
        this.finishBufferTurn(buf);
        break;

    }
  }

  /** 服务端确认新 turn 开始：释放任何遗留的本地 turn 绑定。 */
  beginTurn(sessionPath: string): void {
    const buf = this.getBuffer(sessionPath);
    this.finishBufferTurn(buf);
  }

  /** 服务端确认当前 turn 结束或被中止：flush 可见内容，然后释放 turn-local 绑定。 */
  finishTurn(sessionPath: string): void {
    const buf = this.buffers.get(sessionPath);
    if (!buf) return;
    this.finishBufferTurn(buf);
  }

  /** 清理指定 session 的 buffer */
  clear(sessionPath: string): void {
    const buf = this.buffers.get(sessionPath);
    if (buf?.flushTimer) clearTimeout(buf.flushTimer);
    this.buffers.delete(sessionPath);
  }

  /** 清理所有 */
  clearAll(): void {
    for (const [, buf] of this.buffers) {
      if (buf.flushTimer) clearTimeout(buf.flushTimer);
    }
    this.buffers.clear();
  }

  /**
   * 取当前 buffer 的快照。供 loadMessages 在 session 重建后合并 in-flight
   * 内容：jsonl 只在 turn_end 落盘，在 stream 进行中重建 session 时，
   * 这份快照是避免 UI 上"正在流的消息凭空消失"的唯一来源。
   */
  snapshot(sessionPath: string): StreamBufferSnapshot | null {
    const buf = this.buffers.get(sessionPath);
    if (!buf) return null;
    const hasContent = !!(buf.textAcc || buf.thinkingAcc || buf.moodAcc);
    if (!hasContent) return null;
    return {
      hasContent: true,
      messageId: buf.messageId,
      text: buf.textAcc,
      thinking: buf.thinkingAcc,
      mood: buf.inMood ? buf.moodAcc : cleanMoodText(buf.moodAcc),
      moodYuan: buf.moodYuan,
      inThinking: buf.inThinking,
      inMood: buf.inMood,
    };
  }
}

/** 全局 singleton */
export const streamBufferManager = new StreamBufferManager();

function mergeContentBlock(blocks: ContentBlock[], block: ContentBlock): ContentBlock[] {
  if (isInterludeBlock(block)) return blocks;
  if (block.type === 'media_generation' && block.status === 'pending') {
    const resolved = blocks.some((existing) => isResolvedTaskBlock(existing, block.taskId));
    if (resolved) return blocks;
  }
  const taskId = replacementTaskId(block);
  if (!taskId) return [...blocks, block];
  const idx = blocks.findIndex((existing) => (
    existing.type === 'media_generation' &&
    existing.taskId === taskId
  ));
  if (idx < 0) return [...blocks, block];
  const next = [...blocks];
  next[idx] = block;
  return next;
}

function replacementTaskId(block: ContentBlock): string | null {
  if (block.type === 'file') return block.replacesTaskId || null;
  if (block.type === 'media_generation' && block.status !== 'pending') return block.taskId;
  return null;
}

function interludeAnchorTaskId(block: ContentBlock): string | null {
  if (block.type === 'file') return block.replacesTaskId || null;
  if (block.type === 'media_generation') return block.taskId || null;
  if (block.type === 'subagent' || block.type === 'workflow') return block.taskId || null;
  return null;
}

function isResolvedTaskBlock(block: ContentBlock, taskId: string): boolean {
  if (block.type === 'file') return block.replacesTaskId === taskId;
  return block.type === 'media_generation' &&
    block.taskId === taskId &&
    block.status !== 'pending';
}

function isInterludeBlock(block: ContentBlock): block is Extract<ContentBlock, { type: 'interlude' }> {
  return block.type === 'interlude';
}

function trailingDeferredTextRebindTargetId(items: ChatListItem[]): string | null {
  const taskIds = new Set<string>();
  let idx = items.length - 1;

  while (idx >= 0 && items[idx]?.type === 'interlude') {
    const item = items[idx];
    if (item.type === 'interlude' && item.data.taskId) {
      taskIds.add(item.data.taskId);
    }
    idx -= 1;
  }

  if (taskIds.size === 0) return null;

  const candidate = items[idx];
  if (!candidate || candidate.type !== 'message' || candidate.data.role !== 'assistant') {
    return null;
  }

  const hasDeferredTextAnchor = (candidate.data.blocks || []).some((block) => (
    (block.type === 'workflow' || block.type === 'subagent') &&
    !!block.taskId &&
    taskIds.has(block.taskId)
  ));

  return hasDeferredTextAnchor ? candidate.data.id : null;
}


// 让 chat-slice / session-actions 通过桥接模块触达 manager，打破循环依赖。
registerStreamBufferInvalidator((sessionPath) => {
  if (sessionPath == null) streamBufferManager.clearAll();
  else streamBufferManager.clear(sessionPath);
});
registerStreamBufferSnapshot((sessionPath) => streamBufferManager.snapshot(sessionPath));
