package org.dromara.talk.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.satoken.utils.LoginHelper;
import org.dromara.talk.domain.bo.PhoneUserBo;
import org.dromara.talk.domain.bo.TalkAgentBo;
import org.dromara.talk.domain.bo.TalkSessionBo;
import org.dromara.talk.domain.dto.MessageStreamRequest;
import org.dromara.talk.domain.vo.TalkAgentVo;
import org.dromara.talk.domain.vo.TalkSessionVo;
import org.dromara.talk.service.IChatService;
import org.dromara.talk.service.IDifyService;
import org.dromara.talk.service.IPhoneUserService;
import org.dromara.talk.service.ITalkAgentService;
import org.dromara.talk.service.ITalkSessionService;
import org.dromara.talk.service.ITtsService;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.ByteArrayOutputStream;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
@RequiredArgsConstructor
@Service
public class ChatServiceImpl implements IChatService {

    private final ITtsService ttsService;
    private final ITalkAgentService talkAgentService;
    private final IDifyService difyService;
    private final ITalkSessionService talkSessionService;
    private final IPhoneUserService phoneUserService;
    @Qualifier("chatStreamExecutor")
    private final Executor chatStreamExecutor;
    @Qualifier("chatTtsExecutor")
    private final Executor chatTtsExecutor;

    // 存储每个用户的最新请求ID
    private final Map<Long, Integer> latestRequestIdMap = new ConcurrentHashMap<>();

    // 存储每个会话的对话内容
    private final Map<String, List<Map<String, String>>> conversationMap = new ConcurrentHashMap<>();

    @Override
    public void processMessageStream(MessageStreamRequest request, SseEmitter emitter) {
        long requestStartNs = System.nanoTime();
        String traceId = buildTraceId(request);

        // 在主线程中获取用户ID，避免在异步线程中访问ThreadLocal
        Long userId = LoginHelper.getUserId();
        if (userId == null) {
            userId = 0L;
            log.warn("获取登录用户ID失败，使用默认值");
        }

        log.info("traceId={} currentTime={} 聊天流开始 requestId={}, userId={}, conversationId={}, agentId={}, isGreeting={}, messageLength={}",
            traceId, currentTime(),
            request.getRequestId(), userId, request.getConversationId(), request.getAgentId(),
            request.getIsGreeting(), request.getMessage() != null ? request.getMessage().length() : 0);

        // 更新最新请求ID
        if (request.getRequestId() != null) {
            latestRequestIdMap.put(userId, request.getRequestId());
            log.info("traceId={} currentTime={} 更新用户 {} 的最新请求ID为: {}，elapsedMs={}",
                traceId, currentTime(), userId, request.getRequestId(), elapsedMs(requestStartNs));
        }

        Long finalUserId = userId;
        CompletableFuture.runAsync(() -> {
            long asyncStartNs = System.nanoTime();
            try {
                log.info("traceId={} currentTime={} 异步处理开始 queueWaitMs={}", traceId, currentTime(), elapsedMs(requestStartNs));
                // 获取客服配置
                TalkAgentVo agentConfig = null;
                if (request.getAgentId() != null) {
                    long agentQueryStartNs = System.nanoTime();
                    agentConfig = talkAgentService.queryById(request.getAgentId());
                    log.info("traceId={} currentTime={} 查询客服配置完成 elapsedMs={}, agentId={}, found={}",
                        traceId, currentTime(), elapsedMs(agentQueryStartNs), request.getAgentId(), agentConfig != null);
                }

                // 如果是欢迎语，合成语音但不发送text事件（避免前端重复显示）
                if (Boolean.TRUE.equals(request.getIsGreeting())) {
                    AtomicReference<CompletableFuture<Void>> greetingAudioChain =
                        new AtomicReference<>(CompletableFuture.completedFuture(null));
                    enqueueAudioSynthesis(greetingAudioChain, request.getMessage(), agentConfig, emitter, finalUserId,
                        request.getRequestId(), traceId, "greeting-tts", requestStartNs);
                    completeAfterAudio(greetingAudioChain, emitter, traceId, requestStartNs);
                    return;
                }

                TalkAgentVo finalAgentConfig = agentConfig;
                Integer finalRequestId = request.getRequestId();
                long[] firstTextChunkNs = {0L};
                int[] textChunkCount = {0};
                int[] sentenceCount = {0};
                AtomicReference<CompletableFuture<Void>> audioChain =
                    new AtomicReference<>(CompletableFuture.completedFuture(null));
                StringBuilder audioBatchBuilder = new StringBuilder();
                final int audioBatchMaxLength = 120;

                // 初始化对话记录列表
                String finalConversationId = request.getConversationId();
                if (finalConversationId != null) {
                    List<Map<String, String>> messages = getOrCreateConversation(finalConversationId);

                    // 添加用户消息（客户）
                    Map<String, String> userMsg = new HashMap<>();
                    userMsg.put("role", "user");
                    userMsg.put("content", request.getMessage());
                    userMsg.put("timestamp", String.valueOf(System.currentTimeMillis()));
                    messages.add(userMsg);
                }

                // 用于累积AI回复的完整文本
                StringBuilder aiReplyBuilder = new StringBuilder();

                // 检查 conversationId 是否是临时 sessionId
                // 如果是临时 sessionId（数据库中还未被 Dify 更新），则传递 null 给 Dify
                String difyConversationId = request.getConversationId();
                if (request.getConversationId() != null) {
                    long querySessionStartNs = System.nanoTime();
                    TalkSessionVo session = talkSessionService.queryBySessionId(request.getConversationId());
                    log.info("traceId={} currentTime={} 查询会话完成 elapsedMs={}, requestedConversationId={}, sessionFound={}",
                        traceId, currentTime(), elapsedMs(querySessionStartNs), request.getConversationId(), session != null);
                    if (session != null && request.getConversationId().equals(session.getSessionId())) {
                        // 检查这个 sessionId 是否是刚创建的（没有对话内容）
                        // 如果没有对话内容，说明这是第一次对话，不应该传递给 Dify
                        if (session.getConversationJson() == null || session.getConversationJson().isEmpty()) {
                            difyConversationId = null;
                            log.info("traceId={} currentTime={} 检测到临时 sessionId，第一次对话不传递 conversationId 给 Dify", traceId, currentTime());
                        }
                    }
                }

                // 处理source字段，去除前后空格
                String source = request.getSource() != null ? request.getSource().trim() : null;

                log.info("traceId={} currentTime={} 准备调用Dify elapsedMs={}, difyConversationId={}, source={}",
                    traceId, currentTime(), elapsedMs(requestStartNs), difyConversationId, source);

                difyService.callWorkflowStream(traceId,
                    request.getMessage(),
                    request.getAgentGender(),
                    request.getTtsVcnList(),
                    agentConfig,
                    finalUserId,
                    difyConversationId,
                    request.getCustomerPhone(),
                    source,
                    (textChunk) -> {
                        try {
                            textChunkCount[0]++;
                            if (firstTextChunkNs[0] == 0L) {
                                firstTextChunkNs[0] = System.nanoTime();
                                log.info("traceId={} currentTime={} 首个文本分片发送给前端 elapsedMs={}, chunkLength={}",
                                    traceId, currentTime(), elapsedMs(requestStartNs), textChunk.length());
                            }

                            // 累积AI回复文本
                            aiReplyBuilder.append(textChunk);

                            // 构建 JSON 格式的事件数据
                            Map<String, String> eventData = new HashMap<>();
                            eventData.put("name", "text");
                            eventData.put("data", textChunk);
                            emitter.send(SseEmitter.event()
                                .data(eventData));
                        } catch (Exception e) {
                            log.error("发送文本失败", e);
                        }
                    },
                    (sentence, newConversationId, isComplete) -> {
                        try {
                            sentenceCount[0]++;
                            log.info("traceId={} currentTime={} 句子回调 elapsedMs={}, sentenceIndex={}, sentenceLength={}, isComplete={}, conversationId={}",
                                traceId, currentTime(), elapsedMs(requestStartNs), sentenceCount[0], sentence != null ? sentence.length() : 0,
                                isComplete, newConversationId);

                            // 发送conversationId
                            if (newConversationId != null) {
                                List<Map<String, String>> messages = getOrCreateConversation(newConversationId);

                                // 更新conversationId
                                if (finalConversationId == null) {
                                    Map<String, String> userMsg = new HashMap<>();
                                    userMsg.put("role", "user");
                                    userMsg.put("content", request.getMessage());
                                    userMsg.put("timestamp", String.valueOf(System.currentTimeMillis()));
                                    messages.add(userMsg);
                                    log.info("traceId={} currentTime={} 首次收到Dify conversationId={}, 已补写用户消息", traceId, currentTime(), newConversationId);
                                } else if (!finalConversationId.equals(newConversationId)) {
                                    // 如果传入的conversationId与Dify返回的不同,说明传入的是临时sessionId
                                    // 需要更新数据库中的sessionId
                                    long updateSessionIdStartNs = System.nanoTime();
                                    talkSessionService.updateSessionId(finalConversationId, newConversationId);
                                    moveConversation(finalConversationId, newConversationId);
                                    log.info("traceId={} currentTime={} 更新临时sessionId为Dify conversationId old={}, new={}, elapsedMs={}",
                                        traceId, currentTime(), finalConversationId, newConversationId, elapsedMs(updateSessionIdStartNs));
                                }

                                Map<String, String> conversationEvent = new HashMap<>();
                                conversationEvent.put("name", "conversationId");
                                conversationEvent.put("data", newConversationId);
                                emitter.send(SseEmitter.event()
                                    .data(conversationEvent));
                            }

                            // 检查是否需要合成音频（只为最新请求合成音频）
                            boolean needAudio = true;
                            if (finalRequestId != null) {
                                Integer latestRequestId = latestRequestIdMap.get(finalUserId);
                                if (latestRequestId != null && !latestRequestId.equals(finalRequestId)) {
                                    needAudio = false;
                                    log.info("流式请求ID {} 不是最新请求（最新为 {}），跳过音频合成", finalRequestId, latestRequestId);
                                }
                            }

                            if (needAudio && sentence != null && !sentence.trim().isEmpty()) {
                                audioBatchBuilder.append(sentence.trim());
                                if (shouldFlushAudioBatch(audioBatchBuilder, isComplete, audioBatchMaxLength)) {
                                    String audioText = audioBatchBuilder.toString();
                                    audioBatchBuilder.setLength(0);
                                    log.info("traceId={} currentTime={} 音频批次入队 elapsedMs={}, textLength={}",
                                        traceId, currentTime(), elapsedMs(requestStartNs), audioText.length());
                                    enqueueAudioSynthesis(audioChain, audioText, finalAgentConfig, emitter, finalUserId,
                                        finalRequestId, traceId, "stream-tts", requestStartNs);
                                }
                            }

                            // 如果是最后一个句子，发送完成事件
                            if (isComplete) {
                                log.info("traceId={} currentTime={} 收到完成标志 elapsedMs={}, textChunkCount={}, sentenceCount={}, aiReplyLength={}",
                                    traceId, currentTime(), elapsedMs(requestStartNs), textChunkCount[0], sentenceCount[0], aiReplyBuilder.length());

                                // 保存对话内容到数据库
                                if (newConversationId != null && finalAgentConfig != null) {
                                    try {
                                        long saveConversationStartNs = System.nanoTime();
                                        // 获取当前会话的对话内容
                                        List<Map<String, String>> messages = conversationMap.getOrDefault(newConversationId, new ArrayList<>());

                                        // 添加AI回复消息（客服）
                                        if (aiReplyBuilder.length() > 0) {
                                            Map<String, String> assistantMsg = new HashMap<>();
                                            assistantMsg.put("role", "assistant");
                                            assistantMsg.put("content", aiReplyBuilder.toString());
                                            assistantMsg.put("timestamp", String.valueOf(System.currentTimeMillis()));
                                            messages.add(assistantMsg);
                                        }

                                        // 将对话内容转换为JSON字符串
                                        String conversationJson = new ObjectMapper().writeValueAsString(messages);

                                        // 保存到数据库
                                        talkSessionService.saveOrUpdateConversation(newConversationId, finalAgentConfig.getId(), conversationJson, null, finalUserId);

                                        log.info("traceId={} currentTime={} 对话内容已保存到数据库 elapsedMs={}, conversationId={}, messageCount={}",
                                            traceId, currentTime(), elapsedMs(saveConversationStartNs), newConversationId, messages.size());

                                        // 清理内存：对话完成后从 conversationMap 中移除
                                        conversationMap.remove(newConversationId);
                                        log.debug("traceId={} 已清理会话 {} 的对话内容缓存", traceId, newConversationId);
                                    } catch (Exception e) {
                                        log.error("traceId={} 保存对话内容失败", traceId, e);
                                    }
                                }

                                if (audioBatchBuilder.length() > 0) {
                                    String audioText = audioBatchBuilder.toString();
                                    audioBatchBuilder.setLength(0);
                                    log.info("traceId={} currentTime={} 尾部音频批次入队 elapsedMs={}, textLength={}",
                                        traceId, currentTime(), elapsedMs(requestStartNs), audioText.length());
                                    enqueueAudioSynthesis(audioChain, audioText, finalAgentConfig, emitter, finalUserId,
                                        finalRequestId, traceId, "tail-tts", requestStartNs);
                                }
                                completeAfterAudio(audioChain, emitter, traceId, requestStartNs);
                                log.info("traceId={} currentTime={} 已安排在音频队列完成后发送done事件 totalElapsedMs={}",
                                    traceId, currentTime(), elapsedMs(requestStartNs));
                            }
                        } catch (Exception e) {
                            log.error("traceId={} 处理句子回调失败 elapsedMs={}", traceId, elapsedMs(requestStartNs), e);
                            emitter.completeWithError(e);
                        }
                    });

                log.info("traceId={} currentTime={} Dify调用返回到ChatService totalElapsedMs={}, asyncElapsedMs={}",
                    traceId, currentTime(), elapsedMs(requestStartNs), elapsedMs(asyncStartNs));

            } catch (Exception e) {
                log.error("traceId={} 流式处理失败 totalElapsedMs={}", traceId, elapsedMs(requestStartNs), e);
                try {
                    emitter.completeWithError(e);
                } catch (Exception ex) {
                    log.error("traceId={} 发送错误失败", traceId, ex);
                }
            }
        }, chatStreamExecutor);
    }

    private List<Map<String, String>> getOrCreateConversation(String conversationId) {
        return conversationMap.computeIfAbsent(conversationId, key -> Collections.synchronizedList(new ArrayList<>()));
    }

    private void moveConversation(String oldConversationId, String newConversationId) {
        List<Map<String, String>> oldMessages = conversationMap.remove(oldConversationId);
        if (oldMessages == null) {
            return;
        }
        List<Map<String, String>> newMessages = getOrCreateConversation(newConversationId);
        synchronized (oldMessages) {
            newMessages.addAll(oldMessages);
        }
    }

    private boolean shouldFlushAudioBatch(StringBuilder audioBatchBuilder, boolean isComplete, int audioBatchMaxLength) {
        return isComplete || audioBatchBuilder.length() >= audioBatchMaxLength;
    }

    private void enqueueAudioSynthesis(AtomicReference<CompletableFuture<Void>> audioChain,
                                       String text,
                                       TalkAgentVo agentConfig,
                                       SseEmitter emitter,
                                       Long userId,
                                       Integer requestId,
                                       String traceId,
                                       String stageName,
                                       long requestStartNs) {
        if (text == null || text.trim().isEmpty()) {
            return;
        }
        log.info("traceId={} currentTime={} {} 排队等待TTS elapsedMs={}, textLength={}",
            traceId, currentTime(), stageName, elapsedMs(requestStartNs), text.length());
        audioChain.getAndUpdate(previous -> previous.thenRunAsync(() ->
            synthesizeAndSendAudio(text, agentConfig, emitter, userId, requestId, traceId, stageName, requestStartNs), chatTtsExecutor));
    }

    private void synthesizeAndSendAudio(String text,
                                        TalkAgentVo agentConfig,
                                        SseEmitter emitter,
                                        Long userId,
                                        Integer requestId,
                                        String traceId,
                                        String stageName,
                                        long requestStartNs) {
        if (!isLatestRequest(userId, requestId)) {
            log.info("traceId={} currentTime={} {} 请求ID {} 已过期，跳过音频合成 elapsedMs={}",
                traceId, currentTime(), stageName, requestId, elapsedMs(requestStartNs));
            return;
        }

        long ttsStartNs = System.nanoTime();
        long[] firstAudioChunkNs = {0L};
        int[] audioChunkCount = {0};
        ByteArrayOutputStream mergedAudioBytes = new ByteArrayOutputStream();
        AtomicReference<RuntimeException> audioFailure = new AtomicReference<>();
        log.info("traceId={} currentTime={} {} 开始TTS elapsedMs={}, textLength={}",
            traceId, currentTime(), stageName, elapsedMs(requestStartNs), text.length());
        ttsService.synthesize(text, agentConfig, new ITtsService.AudioCallback() {
            @Override
            public void onAudio(String base64Audio, int status) {
                try {
                    audioChunkCount[0]++;
                    if (firstAudioChunkNs[0] == 0L) {
                        firstAudioChunkNs[0] = System.nanoTime();
                        log.info("traceId={} currentTime={} {} 首个TTS音频分片到达 ttsElapsedMs={}, requestElapsedMs={}",
                            traceId, currentTime(), stageName, elapsedMs(ttsStartNs), elapsedMs(requestStartNs));
                    }
                    byte[] audioBytes = Base64.getDecoder().decode(base64Audio);
                    mergedAudioBytes.write(audioBytes);
                    if (status == 2 && isLatestRequest(userId, requestId)) {
                        Map<String, String> audioEvent = new HashMap<>();
                        audioEvent.put("name", "audio");
                        audioEvent.put("data", Base64.getEncoder().encodeToString(mergedAudioBytes.toByteArray()));
                        emitter.send(SseEmitter.event().data(audioEvent));
                        log.info("traceId={} currentTime={} {} TTS完成并发送音频 ttsElapsedMs={}, requestElapsedMs={}, audioChunkCount={}, mergedBytes={}",
                            traceId, currentTime(), stageName, elapsedMs(ttsStartNs), elapsedMs(requestStartNs),
                            audioChunkCount[0], mergedAudioBytes.size());
                    }
                } catch (Exception e) {
                    audioFailure.compareAndSet(null, new RuntimeException("发送音频失败", e));
                }
            }

            @Override
            public void onError(int code, String message) {
                audioFailure.compareAndSet(null, new RuntimeException("TTS合成失败: " + code + " - " + message));
            }
        });
        if (audioFailure.get() != null) {
            throw audioFailure.get();
        }
    }

    private boolean isLatestRequest(Long userId, Integer requestId) {
        if (requestId == null) {
            return true;
        }
        Integer latestRequestId = latestRequestIdMap.get(userId);
        return latestRequestId == null || latestRequestId.equals(requestId);
    }

    private void completeAfterAudio(AtomicReference<CompletableFuture<Void>> audioChain, SseEmitter emitter,
                                    String traceId, long requestStartNs) {
        audioChain.get()
            .whenComplete((unused, throwable) -> {
                try {
                    if (throwable != null) {
                        log.error("traceId={} 音频队列执行失败 totalElapsedMs={}", traceId, elapsedMs(requestStartNs), throwable);
                        emitter.completeWithError(throwable);
                        return;
                    }
                    Map<String, String> doneEvent = new HashMap<>();
                    doneEvent.put("name", "done");
                    doneEvent.put("data", "");
                    emitter.send(SseEmitter.event().data(doneEvent));
                    log.info("traceId={} currentTime={} done事件已发送，关闭SSE连接 totalElapsedMs={}",
                        traceId, currentTime(), elapsedMs(requestStartNs));
                    emitter.complete();
                } catch (Exception e) {
                    log.error("traceId={} 发送done事件失败 totalElapsedMs={}", traceId, elapsedMs(requestStartNs), e);
                    emitter.completeWithError(e);
                }
            });
    }

    private String buildTraceId(MessageStreamRequest request) {
        String requestIdPart = request.getRequestId() != null ? String.valueOf(request.getRequestId()) : "noReq";
        return "chat-" + requestIdPart + "-" + UUID.randomUUID().toString().substring(0, 8);
    }

    private long elapsedMs(long startNs) {
        return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
    }

    private String currentTime() {
        return java.time.LocalDateTime.now().toString();
    }

    @Override
    public Map<String, Object> startChatSession(Long agentId, Long userId, String customerPhone) {
        boolean success = talkAgentService.changeAgentStatus(agentId, "2");
        if (!success) {
            log.warn("客服 {} 开始对话失败，可能已被占用", agentId);
            return Map.of("success", false, "code", 400, "msg", "该客服已被其他用户占用，请选择其他客服");
        }

        String tempSessionId = UUID.randomUUID().toString();

        // 如果有客户电话，检查并保存到 phone_user 表
        if (customerPhone != null && !customerPhone.isEmpty()) {
            try {
                PhoneUserBo phoneUserBo = new PhoneUserBo();
                phoneUserBo.setCustomerPhone(customerPhone);
                phoneUserService.insertByBo(phoneUserBo);
                log.info("新客户电话已保存: {}", customerPhone);
            } catch (Exception e) {
                log.debug("客户电话可能已存在: {}", customerPhone);
            }
        }

        TalkSessionBo sessionBo = new TalkSessionBo();
        sessionBo.setSessionId(tempSessionId);
        sessionBo.setAgentId(agentId);
        sessionBo.setStartTime(new Date());
        sessionBo.setUserId(userId);
        if (customerPhone != null) {
            sessionBo.setCustomerPhone(customerPhone);
        }
        talkSessionService.insertByBo(sessionBo);

        log.info("客服 {} 开始对话，sessionId: {}, customerPhone: {}, 创建人: {}", agentId, tempSessionId, customerPhone, userId);
        return Map.of("success", true, "code", 200, "sessionId", tempSessionId);
    }

    @Override
    public Boolean hangupCall(Long agentId, String sessionId, List<Map<String, Object>> chatHistory) {
        // 更新会话结束时间并保存对话内容
        if (sessionId != null) {
            talkSessionService.updateEndTime(sessionId);

            // 保存对话内容
            if (chatHistory != null && !chatHistory.isEmpty()) {
                try {
                    // 转换前端格式到后端格式
                    List<Map<String, String>> messages = new ArrayList<>();
                    for (Map<String, Object> msg : chatHistory) {
                        Map<String, String> message = new HashMap<>();
                        String type = (String) msg.get("type");
                        String content = (String) msg.get("content");

                        // 转换 type: user/agent 到 role: user/assistant
                        message.put("role", "user".equals(type) ? "user" : "assistant");
                        message.put("content", content);
                        message.put("timestamp", String.valueOf(System.currentTimeMillis()));
                        messages.add(message);
                    }

                    String conversationJson = new ObjectMapper().writeValueAsString(messages);
                    talkSessionService.saveOrUpdateConversation(sessionId, agentId, conversationJson, null, null);
                    log.info("挂断时保存对话内容，会话ID: {}, 消息数量: {}", sessionId, messages.size());
                } catch (Exception e) {
                    log.error("保存对话内容失败", e);
                }
            }
        }

        // 将客服状态改为空闲
        TalkAgentBo bo = new TalkAgentBo();
        bo.setId(agentId);
        bo.setStatus("0");
        boolean success = talkAgentService.updateByBo(bo);
        log.info("客服 {} 挂断电话，状态已改为空闲中，会话ID: {}", agentId, sessionId);
        return success;
    }
}
