package consumer

import (
	"context"
	"strings"

	"gitea.com/red-future/common/db/mongo"
	"gitea.com/red-future/common/rabbitmq"
	"gitea.com/red-future/common/ragflow"
	"gitea.com/red-future/common/redis"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gfile"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
)

// processorConfig 处理器配置（避免重复读取配置）
type processorConfig struct {
	chatId             string // RAGFlow Chat ID（默认）
	responseExchange   string // 响应队列 Exchange
	responseRoutingKey string // 响应队列 RoutingKey
	followUpExchange   string // 追问队列 Exchange
	followUpRoutingKey string // 追问队列 RoutingKey
	archiveExchange    string // 归档队列 Exchange
	archiveRoutingKey  string // 归档队列 RoutingKey
	// Mock模式已删除，不再用于压测
}

var cfg *processorConfig

// responsePublisher 响应消息发布器（单例）
var responsePublisher *rabbitmq.Publisher

// followUpPublisher 追问消息发布器（单例）
var followUpPublisher *rabbitmq.Publisher

// archivePublisher 归档消息发布器（单例）
var archivePublisher *rabbitmq.Publisher

// Processor 消息处理器
type Processor struct {
	processor *ragflow.QueueProcessor
}

// NewProcessor 创建消息处理器
func NewProcessor(ctx context.Context) (*Processor, error) {
	// 从配置读取参数
	streamKey := g.Cfg().MustGet(ctx, "stream.streamKey").String()
	groupName := g.Cfg().MustGet(ctx, "stream.groupName").String()
	consumerName := g.Cfg().MustGet(ctx, "stream.consumerName").String()
	batchSize := g.Cfg().MustGet(ctx, "stream.batchSize", 200).Int64()
	blockTimeout := g.Cfg().MustGet(ctx, "stream.blockTimeout", 2000).Int64()

	// 初始化处理器配置（单例）
	cfg = &processorConfig{
		chatId:             g.Cfg().MustGet(ctx, "ragflow.chat_id").String(),
		responseExchange:   g.Cfg().MustGet(ctx, "rabbitmq.responseExchange", "ragflow.response").String(),
		responseRoutingKey: g.Cfg().MustGet(ctx, "rabbitmq.responseRoutingKey", "response").String(),
		followUpExchange:   g.Cfg().MustGet(ctx, "followUp.exchange", "followup.delayed").String(),
		followUpRoutingKey: g.Cfg().MustGet(ctx, "followUp.routingKey", "followup").String(),
		archiveExchange:    g.Cfg().MustGet(ctx, "archive.exchange", "archive.delayed").String(),
		archiveRoutingKey:  g.Cfg().MustGet(ctx, "archive.routingKey", "archive").String(),
	}

	// 咨询方向配置已从Consul自动加载（common/config包init时自动执行）

	// 初始化响应发布器
	responsePublisher = rabbitmq.NewPublisher(cfg.responseExchange, cfg.responseRoutingKey)
	glog.Infof(ctx, "响应发布器已初始化 - Exchange: %s, RoutingKey: %s", cfg.responseExchange, cfg.responseRoutingKey)

	// 初始化追问发布器
	followUpPublisher = rabbitmq.NewPublisher(cfg.followUpExchange, cfg.followUpRoutingKey)
	glog.Info(ctx, "追问发布器已初始化")

	// 初始化归档发布器
	archivePublisher = rabbitmq.NewPublisher(cfg.archiveExchange, cfg.archiveRoutingKey)
	glog.Info(ctx, "归档发布器已初始化")

	// 创建消息处理器（批量读取 + 并发发送，削峰填谷）
	return &Processor{
		processor: ragflow.NewQueueProcessor(
			streamKey,
			groupName,
			consumerName,
			blockTimeout,
			batchSize,
			handleMessage,
		),
	}, nil
}

// Start 启动消息处理
func (p *Processor) Start(ctx context.Context) error {
	glog.Info(ctx, "开始消费消息...")
	return p.processor.Start(ctx)
}

// Stop 停止消息处理
func (p *Processor) Stop() {
	p.processor.Stop()
}

// getChatIdByAccountName 根据客服账号名称从ragflow_config表查询chat_id
func getChatIdByAccountName(ctx context.Context, tenantId, accountName string) string {
	if accountName == "" {
		return ""
	}

	db := mongo.GetDB()
	if db == nil {
		glog.Error(ctx, "MongoDB未初始化")
		return ""
	}

	collection := db.Collection("ragflow_config")

	// 从MongoDB查询ragflow_config（先尝试字符串tenantId）
	filter := map[string]interface{}{
		"accountName": accountName,
		"isDeleted":   false,
	}

	if tenantId != "" {
		filter["tenantId"] = tenantId
	}

	var config struct {
		ChatId string `json:"chatId" bson:"chatId"`
	}

	err := collection.FindOne(ctx, filter).Decode(&config)

	// 如果未找到且tenantId可以转为数字，尝试用数字查询（兼容MongoDB中存储为int的情况）
	if err != nil && tenantId != "" {
		tenantIdInt := gconv.Int(tenantId)
		if tenantIdInt > 0 {
			filter["tenantId"] = tenantIdInt
			err = collection.FindOne(ctx, filter).Decode(&config)
		}
	}

	if err != nil {
		glog.Warningf(ctx, "未找到客服账号对应的RAGFlow配置 - 账号: %s, tenantId: %s, err: %v", accountName, tenantId, err)
		return ""
	}

	glog.Infof(ctx, "使用客服账号对应的chat_id - 账号: %s, chat_id: %s", accountName, config.ChatId)
	return config.ChatId
}

// getChatIdByDirection 根据用户选择的咨询方向获取对应的chat_id（从Consul读取）
func getChatIdByDirection(ctx context.Context, userId, platform string) string {
	// 从Redis获取用户状态
	userState, err := redis.GetUserState(ctx, userId, platform)
	if err != nil || userState.Direction == "" {
		// 无方向或获取失败，返回默认chat_id
		if cfg != nil {
			return cfg.chatId
		}
		return ""
	}

	// 直接使用accountName查询
	chatId := ""
	if chatId != "" {
		glog.Infof(ctx, "使用咨询方向对应的chat_id - 用户: %s, 方向: %s, chat_id: %s", userId, userState.Direction, chatId)
		return chatId
	}

	// 未找到匹配方向，返回默认chat_id
	glog.Warningf(ctx, "未找到方向对应的chat_id，使用默认 - 用户: %s, 方向: %s", userId, userState.Direction)
	if cfg != nil {
		return cfg.chatId
	}
	return ""
}

// HandleMessageHTTP 处理HTTP请求的消息（导出供controller调用）
func HandleMessageHTTP(ctx context.Context, message map[string]interface{}) error {
	return handleMessage(ctx, message)
}

// handleMessage 处理单条消息
func handleMessage(ctx context.Context, message map[string]interface{}) (err error) {
	// gconv.Map转换结构体时使用驼峰字段名，而非json标签
	userId := gconv.String(message["UserId"])
	content := gconv.String(message["Content"])
	messageId := gconv.String(message["MessageId"])
	platform := gconv.String(message["Platform"])
	tenantId := gconv.String(message["TenantId"])
	accountId := gconv.String(message["AccountId"])     // 客服账号ID
	accountName := gconv.String(message["AccountName"]) // 客服账号名称（如cs_xhs_qixue）
	messageChatId := gconv.String(message["ChatId"])    // 消息中携带的chat_id（从ragflow_config查询）
	if platform == "" {
		platform = "xiaohongshu" // 默认平台
	}

	// 解析历史对话（由 customerservice 从 MongoDB 读取后携带）
	var history []redis.HistoryMessage
	if historyData := message["History"]; historyData != nil {
		_ = gjson.New(historyData).Scan(&history)
	}

	glog.Infof(ctx, "处理消息 - 用户: %s, 客服账号: %s, ChatId: %s, 内容: %s, 历史轮数: %d",
		userId, accountName, messageChatId, content, len(history))

	var answer string
	var sessionId string
	startTime := gtime.Now()

	// 调用RAGFlow处理消息
	{
		// 1. 获取chat_id（优先级：消息携带 > 客服账号查询 > 用户方向查询 > 默认）
		var chatId string
		if messageChatId != "" {
			chatId = messageChatId
			glog.Infof(ctx, "使用消息携带的chat_id: %s", chatId)
		} else if accountName != "" {
			chatId = getChatIdByAccountName(ctx, tenantId, accountName)
		}
		if chatId == "" {
			chatId = getChatIdByDirection(ctx, userId, platform)
		}

		// 2. 检测chatId是否变更不再需要，因为缓存key已包含chatId
		// 不同chatId自动使用不同的session缓存	}

		// 3. 获取或创建 Session（使用正确的chat_id，包含租户隔离）
		var isNewSession bool
		sessionId, isNewSession, err = getOrCreateSession(ctx, tenantId, userId, chatId)
		if err != nil {
			// 检测Chat权限错误（assistant/chat不属于当前API Key）
			if strings.Contains(err.Error(), "do not own the assistant") || strings.Contains(err.Error(), "don't own the chat") || strings.Contains(err.Error(), "doesn't exist") || strings.Contains(err.Error(), "not found") {
				glog.Warningf(ctx, "创建Session时检测到Chat权限错误，尝试自动重建 - 用户: %s, accountName: %s, chat_id: %s", userId, accountName, chatId)

				// 调用重建逻辑
				if accountName != "" {
					if recreateErr := recreateChatIfNeeded(ctx, tenantId, accountName, platform); recreateErr != nil {
						glog.Errorf(ctx, "自动重建Chat失败: %v", recreateErr)
						return
					}

					// 重新查询chat_id
					newChatId := getChatIdByAccountName(ctx, tenantId, accountName)
					if newChatId == "" {
						glog.Errorf(ctx, "重建Chat后仍无法获取chat_id")
						return
					}
					chatId = newChatId
					glog.Infof(ctx, "Chat重建成功，新chat_id: %s", chatId)

					// 清理旧session缓存
					redis.DelSessionCache(ctx, tenantId, userId)

					// 使用新chat_id重新创建session
					sessionId, isNewSession, err = getOrCreateSession(ctx, tenantId, userId, chatId)
					if err != nil {
						glog.Errorf(ctx, "使用新Chat创建Session失败: %v", err)
						return
					}
				} else {
					glog.Errorf(ctx, "Chat权限错误但缺少accountName，无法自动重建")
					return
				}
			} else {
				glog.Errorf(ctx, "获取 Session 失败: %v", err)
				return
			}
		}

		// 3. 调用 RAGFlow API
		client := ragflow.GetGlobalClient()
		if client == nil {
			glog.Error(ctx, "RAGFlow 客户端未初始化")
			return
		}

		// 如果是新 Session 且有历史对话，把历史拼接到问题中调用 RAGFlow
		if isNewSession && len(history) > 0 {
			var newSessionId string
			newSessionId, answer, err = callWithHistory(ctx, client, userId, content, history, chatId)
			if err != nil {
				glog.Errorf(ctx, "带历史调用 RAGFlow 失败: %v", err)
				return
			}
			// 更新 Session 缓存（包含租户隔离）
			if newSessionId != "" {
				redis.SetSessionCache(ctx, tenantId, userId, newSessionId)
				sessionId = newSessionId
			}
		} else {
			// 已有 Session 或无历史，使用普通 API（RAGFlow 内部维护上下文）
			var res *ragflow.ChatCompletionRes
			res, err = client.ChatCompletion(ctx, chatId, &ragflow.ChatCompletionReq{
				Question:  content,
				SessionId: sessionId,
				UserId:    userId,
				Stream:    false,
			})
			if err != nil {
				// 检测Chat不存在或权限错误（RAGFlow中Chat被删除或不属于当前API Key）
				if strings.Contains(err.Error(), "doesn't exist") || strings.Contains(err.Error(), "not found") || strings.Contains(err.Error(), "404") || strings.Contains(err.Error(), "don't own the chat") {
					glog.Warningf(ctx, "检测到Chat不存在，尝试自动重建 - 用户: %s, accountName: %s, chat_id: %s", userId, accountName, chatId)

					// 调用重建逻辑（需要accountName）
					if accountName != "" {
						if recreateErr := recreateChatIfNeeded(ctx, tenantId, accountName, platform); recreateErr != nil {
							glog.Errorf(ctx, "自动重建Chat失败: %v", recreateErr)
							return
						}

						// 重新查询chat_id
						newChatId := getChatIdByAccountName(ctx, tenantId, accountName)
						if newChatId == "" {
							glog.Errorf(ctx, "重建Chat后仍无法获取chat_id")
							return
						}
						chatId = newChatId
						glog.Infof(ctx, "Chat重建成功，新chat_id: %s", chatId)

						// 清理旧session缓存
						redis.DelSessionCache(ctx, tenantId, userId)

						// 重新创建session并调用
						sessionId, _, err = getOrCreateSession(ctx, tenantId, userId, chatId)
						if err != nil {
							glog.Errorf(ctx, "创建新Session失败: %v", err)
							return
						}

						// 重试调用
						res, err = client.ChatCompletion(ctx, chatId, &ragflow.ChatCompletionReq{
							Question:  content,
							SessionId: sessionId,
							UserId:    userId,
							Stream:    false,
						})
						if err != nil {
							glog.Errorf(ctx, "使用新Chat重试失败: %v", err)
							return
						}
					} else {
						glog.Errorf(ctx, "Chat不存在但缺少accountName，无法自动重建")
						return
					}
					// 检测session ownership错误（session不属于当前chat_id）
				} else if strings.Contains(err.Error(), "don't own the session") {
					glog.Warningf(ctx, "Session不属于当前chat_id，清理缓存并重新创建 - 用户: %s, 旧session: %s, chat_id: %s", userId, sessionId, chatId)
					// 清理缓存（包含租户隔离）
					redis.DelSessionCache(ctx, tenantId, userId)
					// 重新创建session并调用
					sessionId, _, err = getOrCreateSession(ctx, tenantId, userId, chatId)
					if err != nil {
						glog.Errorf(ctx, "重新创建Session失败: %v", err)
						return
					}
					// 重试调用
					res, err = client.ChatCompletion(ctx, chatId, &ragflow.ChatCompletionReq{
						Question:  content,
						SessionId: sessionId,
						UserId:    userId,
						Stream:    false,
					})
					if err != nil {
						glog.Errorf(ctx, "重试调用RAGFlow失败: %v", err)
						return
					}
				} else {
					glog.Errorf(ctx, "调用 RAGFlow 失败: %v", err)
					return
				}
			}
			answer = res.Data.Answer

			// 更新 Session 缓存（如果 RAGFlow 返回了新的 session_id，包含租户隔离）
			if res.Data.SessionId != "" && res.Data.SessionId != sessionId {
				redis.SetSessionCache(ctx, tenantId, userId, res.Data.SessionId)
				sessionId = res.Data.SessionId
			}
		}
	}

	// 计算耗时并写入文件
	endTime := gtime.Now()
	elapsed := endTime.Sub(startTime)
	// 截取回复前20字
	answerPreview := answer
	if len([]rune(answerPreview)) > 20 {
		answerPreview = string([]rune(answerPreview)[:20]) + "..."
	}
	glog.Infof(ctx, "回复 - 用户: %s, 耗时: %s, 回复: %s", userId, elapsed, answerPreview)
	// 日志格式：发送时间 | 响应时间 | 耗时 | 用户 | 问题 | 回复前20字
	// 写入时间日志，失败时记录警告但不影响主流程
	logContent := startTime.Format("Y-m-d H:i:s") + "\t" + endTime.Format("Y-m-d H:i:s") + "\t" + elapsed.String() + "\t" + userId + "\t" + content + "\t" + answerPreview + "\n"
	if err := gfile.PutContentsAppend("timelog/ragflow_time.log", logContent); err != nil {
		glog.Warningf(ctx, "写入时间日志失败: %v", err)
	}

	// 3. 更新用户最后活跃时间
	redis.SetSessionLastActive(ctx, userId)

	// 6. 原样写入 RabbitMQ 结果队列（透传 TenantId、AccountId、AccountName）
	responseMsg := &redis.ResponseStreamMessage{
		UserId:      userId,
		Platform:    platform,
		TenantId:    tenantId,
		AccountId:   accountId,
		AccountName: accountName,
		Question:    content,
		Content:     answer,
		SessionId:   sessionId,
		Timestamp:   gtime.Now().Timestamp(),
		MessageId:   messageId,
	}

	// 读取请求中的ReplyQueue字段，支持多实例独立队列
	replyQueue := gconv.String(message["reply_queue"])
	if replyQueue != "" && replyQueue != cfg.responseRoutingKey {
		// 使用自定义响应队列（多实例场景）
		// routing key使用队列名，实现精确路由（避免广播到所有实例）
		glog.Infof(ctx, "使用自定义响应队列: %s - 用户: %s", replyQueue, userId)
		customPublisher := rabbitmq.NewPublisher(cfg.responseExchange, replyQueue)
		if err = customPublisher.PublishWithRoutingKey(ctx, replyQueue, responseMsg); err != nil {
			glog.Errorf(ctx, "写入自定义响应队列失败: %v", err)
			return
		}
		glog.Infof(ctx, "响应已写入 RabbitMQ - 用户: %s, routingKey: %s (队列名)", userId, replyQueue)
	} else {
		// 使用默认响应队列（单实例或旧版本兼容）
		// routing key使用：tenantId.userId
		routingKey := tenantId + "." + userId
		if err = responsePublisher.PublishWithRoutingKey(ctx, routingKey, responseMsg); err != nil {
			glog.Errorf(ctx, "写入 RabbitMQ 结果队列失败: %v", err)
			return
		}
		glog.Infof(ctx, "响应已写入 RabbitMQ - 用户: %s, routingKey: %s (租户.用户)", userId, routingKey)
	}

	// 7. 发送追问消息到延时队列
	sendFollowUpMessages(ctx, userId, platform)

	// 8. 发送归档消息到延时队列（60分钟后）
	sendArchiveMessage(ctx, userId, platform, sessionId, tenantId)

	glog.Infof(ctx, "消息处理完成 - 用户: %s", userId)
	return
}

// sendArchiveMessage 发送归档消息到延时队列
func sendArchiveMessage(ctx context.Context, userId, platform, sessionId, tenantId string) {
	msg := &redis.ArchiveMessage{
		UserId:    userId,
		Platform:  platform,
		SessionId: sessionId,
		TenantId:  tenantId,
		Timestamp: gtime.Now().Timestamp(),
	}

	if err := archivePublisher.PublishDelayed(ctx, msg, redis.GetArchiveDelay()); err != nil {
		glog.Errorf(ctx, "发送归档消息失败: %v", err)
	}
}

// callWithHistory 带历史上下文调用 RAGFlow（用于新 Session）
// history 由 customerservice 从 MongoDB 读取后通过消息携带
// 使用 RAGFlow 原生 API，把历史对话拼接到问题中，保留提示词和知识库功能
// chatId 为用户选择方向对应的chat_id，而非默认chat_id
func callWithHistory(ctx context.Context, client *ragflow.Client, userId, content string, history []redis.HistoryMessage, chatId string) (sessionId, answer string, err error) {
	// 构建带历史上下文的问题
	var question string
	if len(history) > 0 {
		// 限制历史对话长度：最多3轮或总字符数不超过8000（避免超过RAGFlow输入限制）
		const maxHistoryRounds = 3
		const maxHistoryChars = 8000

		var builder strings.Builder
		builder.WriteString("[以下是之前的对话历史，请参考]\n")

		// 限制历史轮数
		historyToUse := history
		if len(history) > maxHistoryRounds {
			historyToUse = history[len(history)-maxHistoryRounds:] // 只取最近3轮
			glog.Infof(ctx, "历史对话超过%d轮，截取最近%d轮 - 用户: %s", len(history), maxHistoryRounds, userId)
		}

		// 拼接历史对话，同时检查字符数限制
		for _, h := range historyToUse {
			builder.WriteString("用户: ")
			builder.WriteString(h.Question)
			builder.WriteString("\nAI: ")
			builder.WriteString(h.Answer)
			builder.WriteString("\n")

			// 检查是否超过字符数限制
			if builder.Len() > maxHistoryChars {
				glog.Warningf(ctx, "历史对话超过%d字符，停止追加 - 用户: %s", maxHistoryChars, userId)
				break
			}
		}

		builder.WriteString("\n[当前问题]\n")
		builder.WriteString(content)
		question = builder.String()
	} else {
		question = content
	}

	glog.Infof(ctx, "注入 %d 轮历史对话上下文 - chat_id: %s", len(history), chatId)

	// 先创建新 session（使用方向对应的chat_id）
	session, err := client.CreateSession(ctx, chatId, &ragflow.CreateSessionReq{
		Name:   "session_" + userId,
		UserId: userId,
	})
	if err != nil {
		glog.Errorf(ctx, "创建 Session 失败: %v", err)
		return
	}
	sessionId = session.Id

	// 使用 RAGFlow 原生 API（保留提示词和知识库，使用方向对应的chat_id）
	res, err := client.ChatCompletion(ctx, chatId, &ragflow.ChatCompletionReq{
		Question:  question,
		SessionId: sessionId,
		UserId:    userId,
		Stream:    false,
	})
	if err != nil {
		return
	}

	answer = res.Data.Answer
	return
}

// sendFollowUpMessages 发送追问消息到延时队列
func sendFollowUpMessages(ctx context.Context, userId, platform string) {
	now := gtime.Now().Timestamp()

	for followUpType := redis.FollowUpType1; followUpType <= redis.FollowUpType3; followUpType++ {
		msg := &redis.FollowUpMessage{
			UserId:       userId,
			Platform:     platform,
			Content:      redis.GetFollowUpContent(followUpType),
			FollowUpType: followUpType,
			Timestamp:    now,
		}

		if err := followUpPublisher.PublishDelayed(ctx, msg, redis.GetFollowUpDelay(followUpType)); err != nil {
			glog.Errorf(ctx, "发送追问消息失败 - 类型: %d, 错误: %v", followUpType, err)
		}
	}
}

// getOrCreateSession 获取或创建 RAGFlow Session（支持租户隔离）
// 返回 isNew=true 表示是新创建的 session，需要注入历史上下文
func getOrCreateSession(ctx context.Context, tenantId, userId, chatId string) (sessionId string, isNew bool, err error) {
	// 先从缓存获取（包含租户隔离）
	if sessionId, err = redis.GetSessionCache(ctx, tenantId, userId); err != nil {
		return
	}
	if sessionId != "" {
		glog.Infof(ctx, "使用缓存的session - 租户: %s, 用户: %s, chat_id: %s, session: %s", tenantId, userId, chatId, sessionId)
		return // 已有 session，不是新的
	}

	// 缓存不存在，创建新 Session（使用传入的chat_id）
	client := ragflow.GetGlobalClient()
	if client == nil {
		return
	}

	glog.Infof(ctx, "创建新session - 租户: %s, 用户: %s, chat_id: %s", tenantId, userId, chatId)
	session, err := client.CreateSession(ctx, chatId, &ragflow.CreateSessionReq{
		Name:   "session_" + tenantId + "_" + userId,
		UserId: userId,
	})
	if err != nil {
		return
	}

	sessionId = session.Id
	isNew = true // 标记为新创建

	// 缓存 Session ID（包含租户隔离）
	redis.SetSessionCache(ctx, tenantId, userId, sessionId)
	glog.Infof(ctx, "新session已创建并缓存 - 租户: %s, 用户: %s, session: %s", tenantId, userId, sessionId)
	return
}