go-svc-code/internal/service/event/dispatcher.go

package event

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"git.x2erp.com/qdy/go-base/logger"
)

// MessageType 消息类型枚举
type MessageType string

const (
	MessageTypeThinking MessageType = "thinking" // 思考过程
	MessageTypeTool     MessageType = "tool"     // 工具调用
	MessageTypeReply    MessageType = "reply"    // 最终回复
	MessageTypeUnknown  MessageType = "unknown"  // 未知类型
)

// CompletionHook 完成钩子接口，用于消息完成时的处理（如保存到数据库）
type CompletionHook interface {
	OnMessageComplete(sessionID string, messageID string, completeText string, eventType string, metadata map[string]interface{})
}

// MessageState 消息状态，用于跟踪单个消息的增量内容
type MessageState struct {
	SessionID  string
	MessageID  string
	StartTime  time.Time
	LastUpdate time.Time
	Metadata   map[string]interface{}

	// 分离的缓冲区，用于不同类型的内容
	ReasoningBuffer strings.Builder // 思考内容
	ReplyBuffer     strings.Builder // 回答内容
	ToolBuffer      strings.Builder // 工具调用

	// 类型完成状态跟踪
	CompletedTypes map[string]bool // 已完成的类型: "reasoning", "text", "tool"
	HookTriggered  map[string]bool // 钩子触发状态，按类型记录

	// 当前活跃类型（用于跟踪正在处理的内容）
	CurrentType string // 当前正在处理的类型
}

// MessageAggregator 消息聚合器，负责合并增量内容并检测完成状态
type MessageAggregator struct {
	mu                    sync.RWMutex
	messages              map[string]*MessageState // key: sessionID_messageID
	hooks                 []CompletionHook
	OnMessageCompleteFunc func(sessionID string, messageID string, completeText string, messageType MessageType, metadata map[string]interface{}) // 消息完成回调函数
	OnEventProcessedFunc  func(sessionID string, eventType string, eventData string, eventMap map[string]interface{})                             // 事件处理回调函数
}

// NewMessageAggregator 创建新的消息聚合器
func NewMessageAggregator() *MessageAggregator {
	return &MessageAggregator{
		messages:              make(map[string]*MessageState),
		hooks:                 make([]CompletionHook, 0),
		OnMessageCompleteFunc: nil,
		OnEventProcessedFunc:  nil,
	}
}

// RegisterHook 注册完成钩子
func (ma *MessageAggregator) RegisterHook(hook CompletionHook) {
	ma.mu.Lock()
	defer ma.mu.Unlock()
	ma.hooks = append(ma.hooks, hook)
}

// ProcessEvent 处理事件，合并增量内容并检测完成状态
func (ma *MessageAggregator) ProcessEvent(eventData string, sessionID string) {
	// 解析事件数据
	var eventMap map[string]interface{}
	if err := json.Unmarshal([]byte(eventData), &eventMap); err != nil {
		logger.Debug(fmt.Sprintf("无法解析事件JSON error=%s dataPreview=%s",
			err.Error(), safeSubstring(eventData, 0, 200)))
		return
	}

	eventType := getEventType(eventMap)

	// 诊断日志：记录处理的事件类型
	logger.Debug(fmt.Sprintf("🔍 ProcessEvent: sessionID=%s eventType=%s dataPreview=%s",
		sessionID, eventType, safeSubstring(eventData, 0, 100)))

	// 调用事件处理回调函数（如果有设置）
	if ma.OnEventProcessedFunc != nil {
		ma.OnEventProcessedFunc(sessionID, eventType, eventData, eventMap)
	}

	// 只处理 message.part.updated, message.updated, session.status 事件
	if eventType == "message.part.updated" {
		ma.handleMessagePartUpdated(eventMap, sessionID, eventData)
	} else if eventType == "message.updated" {
		ma.handleMessageUpdated(eventMap, sessionID)
	} else if eventType == "session.status" {
		ma.handleSessionStatus(eventMap, sessionID)
	} else if eventType == "session.idle" {
		ma.handleSessionIdle(sessionID)
	}
	// 其他事件类型忽略
}

// handleMessagePartUpdated 处理 message.part.updated 事件
func (ma *MessageAggregator) handleMessagePartUpdated(eventMap map[string]interface{}, sessionID string, eventData string) {
	// 提取消息部分信息
	payload, _ := eventMap["payload"].(map[string]interface{})
	props, _ := payload["properties"].(map[string]interface{})
	part, _ := props["part"].(map[string]interface{})

	messageID, _ := part["messageID"].(string)
	partType, _ := part["type"].(string)

	if messageID == "" || (partType != "text" && partType != "reasoning" && partType != "step-finish" && partType != "tool") {
		return
	}

	ma.mu.Lock()
	defer ma.mu.Unlock()

	key := sessionID + "_" + messageID
	state, exists := ma.messages[key]

	if !exists {
		// step-finish 事件不应该创建新状态，它应该总是跟随在text/reasoning事件之后
		if partType == "step-finish" {
			logger.Debug(fmt.Sprintf("忽略step-finish事件，无对应消息状态 sessionID=%s messageID=%s",
				sessionID, messageID))
			return
		}

		// 新消息，初始化状态
		state = &MessageState{
			SessionID:      sessionID,
			MessageID:      messageID,
			StartTime:      time.Now(),
			LastUpdate:     time.Now(),
			Metadata:       make(map[string]interface{}),
			CompletedTypes: make(map[string]bool),
			HookTriggered:  make(map[string]bool),
			CurrentType:    partType,
		}
		ma.messages[key] = state
		logger.Debug(fmt.Sprintf("开始跟踪新消息 sessionID=%s messageID=%s type=%s",
			sessionID, messageID, partType))
	}

	// 更新增量内容 - 根据类型写入不同的缓冲区
	state.CurrentType = partType

	if text, ok := part["text"].(string); ok && text != "" {
		// 根据类型选择缓冲区
		switch partType {
		case "reasoning":
			state.ReasoningBuffer.WriteString(text)
		case "text":
			state.ReplyBuffer.WriteString(text)
		case "tool":
			// 工具调用可能有text字段，也可能通过其他方式处理
			state.ToolBuffer.WriteString(text)
		}
		state.LastUpdate = time.Now()

		// 记录增量合并日志（仅调试）- 已禁用以减少日志量
		// logger.Debug(fmt.Sprintf("合并增量内容 sessionID=%s messageID=%s type=%s deltaLength=%d",
		// 	sessionID, messageID, partType, len(text)))
	} else if partType == "tool" {
		// 处理工具调用事件，提取工具信息
		if name, ok := part["name"].(string); ok && name != "" {
			toolText := fmt.Sprintf("[工具调用: %s", name)
			if args, ok := part["arguments"].(string); ok && args != "" {
				// 参数可能是JSON字符串，可以尝试美化
				toolText += fmt.Sprintf(" 参数: %s", args)
			}
			toolText += "]"
			state.ToolBuffer.WriteString(toolText)
			state.LastUpdate = time.Now()
			logger.Debug(fmt.Sprintf("合并工具调用内容 sessionID=%s messageID=%s toolName=%s",
				sessionID, messageID, name))
		}
	}

	// 检查是否为 step-finish
	if partType == "step-finish" {
		// 标记当前类型为完成
		if state.CurrentType != "" {
			state.CompletedTypes[state.CurrentType] = true
			logger.Info(fmt.Sprintf("步骤完成 sessionID=%s messageID=%s type=%s",
				sessionID, messageID, state.CurrentType))

			// 触发该类型的完成钩子
			ma.triggerTypeCompletionHooks(state, state.CurrentType)
		} else {
			logger.Warn(fmt.Sprintf("step-finish事件无当前类型 sessionID=%s messageID=%s",
				sessionID, messageID))
		}
	}
}

// handleMessageUpdated 处理 message.updated 事件
func (ma *MessageAggregator) handleMessageUpdated(eventMap map[string]interface{}, sessionID string) {
	payload, _ := eventMap["payload"].(map[string]interface{})
	props, _ := payload["properties"].(map[string]interface{})
	info, _ := props["info"].(map[string]interface{})

	messageID, _ := info["id"].(string)
	finishReason, _ := info["finish"].(string)

	if messageID == "" || finishReason == "" {
		return
	}

	key := sessionID + "_" + messageID
	ma.mu.Lock()
	defer ma.mu.Unlock()

	if state, exists := ma.messages[key]; exists {
		logger.Info(fmt.Sprintf("消息完成 sessionID=%s messageID=%s finishReason=%s",
			sessionID, messageID, finishReason))

		// 触发完成钩子（处理所有类型的内容）
		ma.triggerCompletionHooks(state)

		// 清理完成的消息状态
		delete(ma.messages, key)
	}
}

// handleSessionStatus 处理 session.status 事件
func (ma *MessageAggregator) handleSessionStatus(eventMap map[string]interface{}, sessionID string) {
	payload, _ := eventMap["payload"].(map[string]interface{})
	props, _ := payload["properties"].(map[string]interface{})
	status, _ := props["status"].(map[string]interface{})

	statusType, _ := status["type"].(string)
	if statusType == "idle" {
		logger.Info(fmt.Sprintf("会话进入空闲状态 sessionID=%s", sessionID))
		// 可以清理该会话的所有消息状态
		ma.cleanupSession(sessionID)
	}
}

// handleSessionIdle 处理 session.idle 事件
func (ma *MessageAggregator) handleSessionIdle(sessionID string) {
	logger.Info(fmt.Sprintf("会话空闲事件 sessionID=%s", sessionID))
	ma.cleanupSession(sessionID)
}

// cleanupSession 清理指定会话的所有消息状态
func (ma *MessageAggregator) cleanupSession(sessionID string) {
	ma.mu.Lock()
	defer ma.mu.Unlock()

	keysToDelete := make([]string, 0)
	for key, state := range ma.messages {
		if state.SessionID == sessionID {
			keysToDelete = append(keysToDelete, key)

			// 检查是否有未触发的类型内容，强制触发完成钩子
			hasUnfinishedContent := false

			// 检查每种类型是否有内容但钩子未触发
			if state.ReasoningBuffer.Len() > 0 {
				if triggered, exists := state.HookTriggered["reasoning"]; !exists || !triggered {
					hasUnfinishedContent = true
				}
			}
			if state.ReplyBuffer.Len() > 0 {
				if triggered, exists := state.HookTriggered["text"]; !exists || !triggered {
					hasUnfinishedContent = true
				}
			}
			if state.ToolBuffer.Len() > 0 {
				if triggered, exists := state.HookTriggered["tool"]; !exists || !triggered {
					hasUnfinishedContent = true
				}
			}

			if hasUnfinishedContent {
				logger.Warn(fmt.Sprintf("强制完成未完成消息 sessionID=%s messageID=%s",
					sessionID, state.MessageID))
				ma.triggerCompletionHooks(state)
			}
		}
	}

	for _, key := range keysToDelete {
		delete(ma.messages, key)
	}

	if len(keysToDelete) > 0 {
		logger.Info(fmt.Sprintf("清理会话消息状态 sessionID=%s cleanedCount=%d",
			sessionID, len(keysToDelete)))
	}
}

// triggerTypeCompletionHooks 触发特定类型的完成钩子
func (ma *MessageAggregator) triggerTypeCompletionHooks(state *MessageState, partType string) {
	// 避免重复触发
	if triggered, exists := state.HookTriggered[partType]; exists && triggered {
		logger.Debug(fmt.Sprintf("钩子已触发过，跳过 sessionID=%s messageID=%s type=%s",
			state.SessionID, state.MessageID, partType))
		return
	}

	// 获取该类型的缓冲区内容
	var completeText string
	var textLength int

	switch partType {
	case "reasoning":
		completeText = state.ReasoningBuffer.String()
		textLength = len(completeText)
	case "text":
		completeText = state.ReplyBuffer.String()
		textLength = len(completeText)
	case "tool":
		completeText = state.ToolBuffer.String()
		textLength = len(completeText)
	default:
		logger.Warn(fmt.Sprintf("未知类型，跳过钩子触发 sessionID=%s messageID=%s type=%s",
			state.SessionID, state.MessageID, partType))
		return
	}

	duration := time.Since(state.StartTime)

	// 记录完成事件
	logger.Info(fmt.Sprintf("🔔 类型完成总结 sessionID=%s messageID=%s type=%s textLength=%d duration=%v",
		state.SessionID, state.MessageID, partType, textLength, duration))

	if textLength > 0 {
		// 记录文本预览（前200字符）
		preview := completeText
		if len(preview) > 200 {
			preview = preview[:200] + "..."
		}
		logger.Info(fmt.Sprintf("📝 类型文本预览: %s", preview))
	} else {
		logger.Info("📭 空文本完成事件")
	}

	// 转换事件类型为消息类型枚举
	messageType := convertToMessageType(partType)

	// 调用消息完成回调函数（如果设置）- 使用新的类型化接口
	// 注意：这里调用现有的OnMessageCompleteFunc，传入特定类型的内容
	if ma.OnMessageCompleteFunc != nil {
		ma.OnMessageCompleteFunc(state.SessionID, state.MessageID, completeText, messageType, state.Metadata)
	}

	// 调用所有注册的钩子（保持向后兼容）
	for _, hook := range ma.hooks {
		hook.OnMessageComplete(state.SessionID, state.MessageID, completeText, partType, state.Metadata)
	}

	// 标记该类型的钩子已触发
	state.HookTriggered[partType] = true
	logger.Debug(fmt.Sprintf("✅ 类型钩子标记为已触发 sessionID=%s messageID=%s type=%s",
		state.SessionID, state.MessageID, partType))
}

// triggerCompletionHooks 触发完成钩子（向后兼容，触发所有类型）
func (ma *MessageAggregator) triggerCompletionHooks(state *MessageState) {
	// 遍历所有支持的类型，触发各自的完成钩子
	types := []string{"reasoning", "text", "tool"}
	hasAnyContent := false

	for _, partType := range types {
		// 检查该类型是否有内容
		var hasContent bool
		switch partType {
		case "reasoning":
			hasContent = state.ReasoningBuffer.Len() > 0
		case "text":
			hasContent = state.ReplyBuffer.Len() > 0
		case "tool":
			hasContent = state.ToolBuffer.Len() > 0
		}

		if hasContent {
			hasAnyContent = true
			// 触发该类型的完成钩子（函数内部会检查是否已触发）
			ma.triggerTypeCompletionHooks(state, partType)
		}
	}

	// 如果没有内容，至少记录一个事件（保持向后兼容）
	if !hasAnyContent {
		logger.Info(fmt.Sprintf("📭 空消息完成事件 sessionID=%s messageID=%s",
			state.SessionID, state.MessageID))
	}
}

// EventDispatcher 事件分发器 - 单例模式
type EventDispatcher struct {
	mu                sync.RWMutex
	baseURL           string
	port              int
	subscriptions     map[string]map[chan string]struct{} // sessionID -> set of channels
	sessionUserCache  *SessionUserCache                   // sessionID -> userID 映射缓存（用于用户验证）
	client            *http.Client
	cancelFunc        context.CancelFunc
	running           bool
	messageAggregator *MessageAggregator // 消息聚合器
}

// EventData opencode事件数据结构 - 匹配实际事件格式
type EventData struct {
	Directory string                 `json:"directory,omitempty"`
	Payload   map[string]interface{} `json:"payload"`
}

// PayloadData payload内部结构（辅助类型）
type PayloadData struct {
	Type       string                 `json:"type"`
	Properties map[string]interface{} `json:"properties,omitempty"`
}

// NewEventDispatcher 创建新的事件分发器
func NewEventDispatcher(baseURL string, port int) *EventDispatcher {
	return &EventDispatcher{
		baseURL:           baseURL,
		port:              port,
		subscriptions:     make(map[string]map[chan string]struct{}),
		sessionUserCache:  NewSessionUserCache(20 * time.Minute),
		messageAggregator: NewMessageAggregator(),
		client: &http.Client{
			Timeout: 0, // 无超时限制，用于长连接
		},
		running: false,
	}
}

// Start 启动事件分发器，连接到opencode全局事件流
func (ed *EventDispatcher) Start(ctx context.Context) error {
	ed.mu.Lock()
	if ed.running {
		ed.mu.Unlock()
		return fmt.Errorf("event dispatcher already running")
	}

	// 创建子上下文用于控制SSE连接
	sseCtx, cancel := context.WithCancel(ctx)
	ed.cancelFunc = cancel
	ed.running = true
	ed.mu.Unlock()

	// 启动SSE连接协程
	go ed.runSSEConnection(sseCtx)

	logger.Info(fmt.Sprintf("事件分发器已启动 baseURL=%s port=%d",
		ed.baseURL, ed.port))
	return nil
}

// Stop 停止事件分发器
func (ed *EventDispatcher) Stop() {
	ed.mu.Lock()
	if !ed.running {
		ed.mu.Unlock()
		return
	}

	if ed.cancelFunc != nil {
		ed.cancelFunc()
	}

	// 清理所有订阅通道
	for sessionID, channels := range ed.subscriptions {
		for ch := range channels {
			close(ch)
		}
		delete(ed.subscriptions, sessionID)
	}

	ed.running = false
	ed.mu.Unlock()

	logger.Info("事件分发器已停止")
}

// Subscribe 订阅指定会话的事件
func (ed *EventDispatcher) Subscribe(sessionID, userID string) (<-chan string, error) {
	ed.mu.Lock()
	defer ed.mu.Unlock()

	// 缓存会话-用户映射（用于未来需要用户验证时）
	ed.sessionUserCache.Set(sessionID, userID)

	// 创建缓冲通道
	ch := make(chan string, 100)

	// 添加到订阅列表
	if _, exists := ed.subscriptions[sessionID]; !exists {
		ed.subscriptions[sessionID] = make(map[chan string]struct{})
	}
	ed.subscriptions[sessionID][ch] = struct{}{}

	logger.Debug(fmt.Sprintf("新订阅添加 sessionID=%s userID=%s totalSubscriptions=%d",
		sessionID, userID, len(ed.subscriptions[sessionID])))

	return ch, nil
}

// Unsubscribe 取消订阅指定会话的事件
func (ed *EventDispatcher) Unsubscribe(sessionID string, ch <-chan string) {
	ed.mu.Lock()
	defer ed.mu.Unlock()

	if channels, exists := ed.subscriptions[sessionID]; exists {
		// 遍历查找对应的通道（因为ch是只读通道，无法直接作为key）
		var foundChan chan string
		for candidate := range channels {
			// 比较通道值
			if candidate == ch {
				foundChan = candidate
				break
			}
		}

		if foundChan != nil {
			close(foundChan)
			delete(channels, foundChan)
			logger.Debug(fmt.Sprintf("订阅已移除 sessionID=%s remainingSubscriptions=%d",
				sessionID, len(channels)))
		}

		// 如果没有订阅者了，清理该会话的映射
		if len(channels) == 0 {
			delete(ed.subscriptions, sessionID)
			ed.sessionUserCache.Delete(sessionID)
		}
	}
}

// RegisterSession 注册会话（前端调用SendPromptStream时调用）
func (ed *EventDispatcher) RegisterSession(sessionID, userID string) {
	ed.sessionUserCache.Set(sessionID, userID)
	logger.Debug(fmt.Sprintf("会话已注册 sessionID=%s userID=%s",
		sessionID, userID))
}

// buildSSEURL 构建SSE URL，避免端口重复
func (ed *EventDispatcher) buildSSEURL() string {
	// 检查baseURL是否已包含端口
	base := ed.baseURL
	// 简单检查：如果baseURL已经包含端口号模式（冒号后跟数字），就不再加端口
	// 查找最后一个冒号的位置
	lastColon := -1
	for i := len(base) - 1; i >= 0; i-- {
		if base[i] == ':' {
			lastColon = i
			break
		}
	}

	if lastColon != -1 {
		// 检查冒号后是否都是数字（端口号）
		hasPort := true
		for i := lastColon + 1; i < len(base); i++ {
			if base[i] < '0' || base[i] > '9' {
				hasPort = false
				break
			}
		}
		if hasPort {
			// baseURL已有端口，直接拼接路径
			if strings.HasSuffix(base, "/") {
				return base + "global/event"
			}
			return base + "/global/event"
		}
	}

	// baseURL没有端口或端口格式不正确，添加端口
	if strings.HasSuffix(base, "/") {
		return fmt.Sprintf("%s:%d/global/event", strings.TrimSuffix(base, "/"), ed.port)
	}
	return fmt.Sprintf("%s:%d/global/event", base, ed.port)
}

// runSSEConnection 运行SSE连接，读取全局事件并分发
func (ed *EventDispatcher) runSSEConnection(ctx context.Context) {
	// 构建SSE URL，避免重复端口
	url := ed.buildSSEURL()

	for {
		select {
		case <-ctx.Done():
			logger.Info("SSE连接停止（上下文取消）")
			return
		default:
			// 建立SSE连接
			logger.Info(fmt.Sprintf("正在连接SSE流 url=%s",
				url))
			if err := ed.connectAndProcessSSE(ctx, url); err != nil {
				logger.Error(fmt.Sprintf("SSE连接失败，5秒后重试 error=%s url=%s",
					err.Error(), url))

				select {
				case <-ctx.Done():
					return
				case <-time.After(5 * time.Second):
					continue
				}
			}
		}
	}
}

// connectAndProcessSSE 连接并处理SSE流
func (ed *EventDispatcher) connectAndProcessSSE(ctx context.Context, url string) error {
	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return fmt.Errorf("创建请求失败: %w", err)
	}
	req.Header.Set("Accept", "text/event-stream")

	resp, err := ed.client.Do(req)
	if err != nil {
		return fmt.Errorf("发送请求失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("SSE请求失败，状态码: %d, 响应: %s", resp.StatusCode, string(body))
	}

	logger.Info(fmt.Sprintf("SSE连接已建立 url=%s",
		url))

	reader := bufio.NewReader(resp.Body)
	eventCount := 0

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			line, err := reader.ReadString('\n')
			if err != nil {
				if err == io.EOF {
					logger.Info(fmt.Sprintf("SSE流正常结束 totalEvents=%d",
						eventCount))
				} else if ctx.Err() != nil {
					logger.Info("SSE流上下文取消")
				} else {
					logger.Error(fmt.Sprintf("读取SSE流错误 error=%s",
						err.Error()))
				}
				return err
			}

			line = strings.TrimSpace(line)
			if line == "" {
				continue
			}

			if strings.HasPrefix(line, "data: ") {
				data := strings.TrimPrefix(line, "data: ")
				eventCount++

				// 分发事件
				ed.dispatchEvent(data)

				if eventCount%100 == 0 {
					//logger.Debug(fmt.Sprintf("事件处理统计 totalEvents=%d activeSessions=%d",
					//	eventCount, len(ed.subscriptions)))
				}
			}
		}
	}
}

// dispatchEvent 分发事件到相关订阅者
func (ed *EventDispatcher) dispatchEvent(data string) {
	// 解析事件数据获取sessionID
	sessionID := extractSessionIDFromEvent(data)

	// 处理事件聚合（无论是否有sessionID都处理）
	if ed.messageAggregator != nil && sessionID != "" {
		ed.messageAggregator.ProcessEvent(data, sessionID)
	}

	if sessionID == "" {
		// 没有sessionID的事件（如全局心跳）分发给所有订阅者
		//logger.Debug(fmt.Sprintf("广播全局事件 dataPreview=%s", safeSubstring(data, 0, 100)))
		ed.broadcastToAll(data)
		return
	}

	// 只记录非增量文本事件的路由日志，减少日志量
	shouldLog := true
	var eventMap map[string]interface{}
	if err := json.Unmarshal([]byte(data), &eventMap); err == nil {
		eventType := getEventType(eventMap)
		// message.part.updated 是最频繁的事件，跳过其路由日志
		if eventType == "message.part.updated" {
			shouldLog = false
		}
	}

	if shouldLog {
		logger.Debug(fmt.Sprintf("路由事件到会话 sessionID=%s dataPreview=%s",
			sessionID, safeSubstring(data, 0, 100)))
	}

	// 只分发给订阅该会话的通道
	ed.mu.RLock()
	channels, exists := ed.subscriptions[sessionID]
	ed.mu.RUnlock()

	if !exists {
		// 没有该会话的订阅者，忽略事件
		//logger.Debug(fmt.Sprintf("忽略事件，无订阅者 sessionID=%s", sessionID))
		return
	}

	// 发送事件到所有订阅该会话的通道
	ed.mu.RLock()
	for ch := range channels {
		select {
		case ch <- data:
			// 成功发送
		default:
			// 通道已满，丢弃事件并记录警告
			logger.Warn(fmt.Sprintf("事件通道已满，丢弃事件 sessionID=%s",
				sessionID))
		}
	}
	ed.mu.RUnlock()
}

// broadcastToAll 广播事件给所有订阅者（用于全局事件如心跳）
func (ed *EventDispatcher) broadcastToAll(data string) {
	//logger.Debug(fmt.Sprintf("广播事件给所有订阅者 dataPreview=%s", safeSubstring(data, 0, 100)))
	ed.mu.RLock()
	defer ed.mu.RUnlock()

	for sessionID, channels := range ed.subscriptions {
		for ch := range channels {
			select {
			case ch <- data:
				// 成功发送
			default:
				// 通道已满，丢弃事件
				logger.Warn(fmt.Sprintf("事件通道已满，丢弃全局事件 sessionID=%s",
					sessionID))
			}
		}
	}
}

// extractSessionIDFromEvent 从事件数据中提取sessionID
func extractSessionIDFromEvent(data string) string {
	// 尝试解析为JSON
	var eventMap map[string]interface{}
	if err := json.Unmarshal([]byte(data), &eventMap); err != nil {
		logger.ErrorC(fmt.Sprintf("无法解析事件JSON error=%s dataPreview=%s",
			err.Error(), safeSubstring(data, 0, 200)))
		return ""
	}

	// 添加调试日志，显示完整事件结构（仅调试时启用）
	debugMode := true
	if debugMode {
		eventJSON, _ := json.MarshalIndent(eventMap, "", "  ")
		logger.Debug(fmt.Sprintf("事件数据结构 eventStructure=%s",
			string(eventJSON)))
	}

	// 递归查找sessionID字段
	sessionID := findSessionIDRecursive(eventMap)

	if sessionID == "" {
		//logger.Debug(fmt.Sprintf("未找到sessionID字段 eventType=%s dataPreview=%s",
		//	getEventType(eventMap), safeSubstring(data, 0, 100)))
	}

	return sessionID
}

// findSessionIDRecursive 递归查找sessionID字段
func findSessionIDRecursive(data interface{}) string {
	switch v := data.(type) {
	case map[string]interface{}:
		// 检查当前层级的sessionID字段（支持多种命名变体）
		for _, key := range []string{"sessionID", "session_id", "sessionId"} {
			if val, ok := v[key]; ok {
				if str, ok := val.(string); ok && str != "" {
					return str
				}
			}
		}

		// 检查常见嵌套路径
		// 1. payload.properties.sessionID (session.status事件)
		if payload, ok := v["payload"].(map[string]interface{}); ok {
			if props, ok := payload["properties"].(map[string]interface{}); ok {
				if sessionID, ok := props["sessionID"].(string); ok && sessionID != "" {
					return sessionID
				}
			}
		}

		// 2. payload.properties.part.sessionID (message.part.updated事件)
		if payload, ok := v["payload"].(map[string]interface{}); ok {
			if props, ok := payload["properties"].(map[string]interface{}); ok {
				if part, ok := props["part"].(map[string]interface{}); ok {
					if sessionID, ok := part["sessionID"].(string); ok && sessionID != "" {
						return sessionID
					}
				}
			}
		}

		// 3. payload.properties.info.sessionID (message.updated事件)
		if payload, ok := v["payload"].(map[string]interface{}); ok {
			if props, ok := payload["properties"].(map[string]interface{}); ok {
				if info, ok := props["info"].(map[string]interface{}); ok {
					if sessionID, ok := info["sessionID"].(string); ok && sessionID != "" {
						return sessionID
					}
				}
			}
		}

		// 递归遍历所有值
		for _, value := range v {
			if result := findSessionIDRecursive(value); result != "" {
				return result
			}
		}

	case []interface{}:
		// 遍历数组
		for _, item := range v {
			if result := findSessionIDRecursive(item); result != "" {
				return result
			}
		}
	}

	return ""
}

// getEventType 获取事件类型
func getEventType(eventMap map[string]interface{}) string {
	if payload, ok := eventMap["payload"].(map[string]interface{}); ok {
		if eventType, ok := payload["type"].(string); ok {
			return eventType
		}
	}
	return "unknown"
}

// convertToMessageType 将事件类型转换为消息类型枚举
func convertToMessageType(eventType string) MessageType {
	switch eventType {
	case "reasoning":
		return MessageTypeThinking
	case "text":
		return MessageTypeReply
	case "tool":
		return MessageTypeTool
	default:
		// 尝试识别其他类型
		if strings.Contains(eventType, "reasoning") || strings.Contains(eventType, "thinking") {
			return MessageTypeThinking
		}
		if strings.Contains(eventType, "tool") || strings.Contains(eventType, "function") {
			return MessageTypeTool
		}
		return MessageTypeUnknown
	}
}

// safeSubstring 安全的子字符串函数
func safeSubstring(s string, start, length int) string {
	if start < 0 {
		start = 0
	}
	if start >= len(s) {
		return ""
	}
	end := start + length
	if end > len(s) {
		end = len(s)
	}
	return s[start:end]
}

// getHookTriggerSource 获取钩子触发源（用于调试）
func getHookTriggerSource(state *MessageState) string {
	// 检查是否有任何类型的钩子已触发
	for _, triggered := range state.HookTriggered {
		if triggered {
			return "completed"
		}
	}
	return "unknown"
}

// RegisterHook 注册完成钩子
func (ed *EventDispatcher) RegisterHook(hook CompletionHook) {
	if ed.messageAggregator != nil {
		ed.messageAggregator.RegisterHook(hook)
	}
}

// SetOnMessageCompleteFunc 设置消息完成回调函数
func (ed *EventDispatcher) SetOnMessageCompleteFunc(f func(sessionID string, messageID string, completeText string, messageType MessageType, metadata map[string]interface{})) {
	if ed.messageAggregator != nil {
		ed.messageAggregator.OnMessageCompleteFunc = f
		logger.Info("✅ 消息完成回调函数已设置")
	}
}

// SetOnEventProcessedFunc 设置事件处理回调函数
func (ed *EventDispatcher) SetOnEventProcessedFunc(f func(sessionID string, eventType string, eventData string, eventMap map[string]interface{})) {
	if ed.messageAggregator != nil {
		ed.messageAggregator.OnEventProcessedFunc = f
		logger.Info("✅ 事件处理回调函数已设置")
	}
}

// GetInstance 获取单例实例（线程安全）
var (
	instance     *EventDispatcher
	instanceOnce sync.Once
)

// GetEventDispatcher 获取事件分发器单例
func GetEventDispatcher(baseURL string, port int) *EventDispatcher {
	instanceOnce.Do(func() {
		instance = NewEventDispatcher(baseURL, port)
	})
	return instance
}

// DiagnosticHook 诊断钩子实现（用于调试和测试）
type DiagnosticHook struct{}

func (h *DiagnosticHook) OnMessageComplete(sessionID string, messageID string, completeText string, eventType string, metadata map[string]interface{}) {
	logger.Info(fmt.Sprintf("🔍 诊断钩子触发: session=%s message=%s type=%s textLength=%d",
		sessionID, messageID, eventType, len(completeText)))

	if len(completeText) > 0 {
		preview := completeText
		if len(preview) > 150 {
			preview = preview[:150] + "..."
		}
		logger.Info(fmt.Sprintf("📋 诊断钩子文本预览: %s", preview))
	} else {
		logger.Info("📭 诊断钩子: 空文本")
	}

	// 记录元数据（如果有）
	if metadata != nil && len(metadata) > 0 {
		logger.Info(fmt.Sprintf("📊 诊断钩子元数据: %+v", metadata))
	}
}

// 使用示例:
// dispatcher := GetEventDispatcher("http://localhost", 3000)
// dispatcher.RegisterHook(&DiagnosticHook{})