public interface ChatService {
    /**
     * 创建 SSE 连接
     */
    SseEmitter createSseConnection(String userId);
    
    /**
     * 异步发送消息
     */
    void sendMessageAsync(String userId, String message, String mode);
    
    /**
     * 停止生成
     */
    void stopGeneration(String userId);
}

@Slf4j
@Service
public class ChatServiceImpl implements ChatService {
    
    // 注入 DeepSeek 对话模型
    @Autowired
    @Qualifier("deepSeekChatModel")
    private ChatModel chatModel;
    
    @Autowired
    private RagService ragService;
    
    private ChatClient chatClient;
    
    // 存储用户的流订阅，用于取消
    private final Map<String, Disposable> userSubscriptions = new ConcurrentHashMap<>();
}

@Autowired
public void initChatClient() {
    this.chatClient = ChatClient.builder(chatModel).build();
}

private void processChat(String userId, String message) {
    try {
        log.info("【普通对话】用户: {}, 消息: {}", userId, message);
        
        // 1. 创建提示词
        Prompt prompt = new Prompt(message);
        
        // 2. 获取流式响应
        Flux<String> stream = chatClient.prompt(prompt).stream().content();
        
        // 3. 订阅流并处理
        Disposable subscription = stream
                .doOnError(throwable -> {
                    log.error("【对话错误】用户: {}, 错误: {}", userId, throwable.getMessage());
                    SSEServer.sendMsg(userId, "抱歉，服务出现了一点问题", SSEMsgType.FINISH);
                    userSubscriptions.remove(userId);
                })
                .subscribe(
                        // onNext: 接收每个文本片段
                        content -> {
                            if (SSEServer.exists(userId)) {
                                SSEServer.sendMsg(userId, content, SSEMsgType.ADD);
                            }
                        },
                        // onError: 处理错误
                        error -> {
                            log.error("【对话流失败】用户: {}, 错误: {}", userId, error.getMessage());
                            userSubscriptions.remove(userId);
                        },
                        // onComplete: 完成回调
                        () -> {
                            log.info("【对话完成】用户: {}", userId);
                            SSEServer.sendMsg(userId, "done", SSEMsgType.FINISH);
                            SSEServer.close(userId);
                            userSubscriptions.remove(userId);
                        }
                );
        
        // 4. 保存订阅引用，用于后续取消
        userSubscriptions.put(userId, subscription);
        
    } catch (Exception e) {
        log.error("【对话异常】用户: {}, 错误: {}", userId, e.getMessage());
        SSEServer.sendMsg(userId, "系统错误，请稍后重试", SSEMsgType.ERROR);
    }
}

@Override
public void stopGeneration(String userId) {
    log.info("【停止生成】用户: {}", userId);
    
    // 取消当前的订阅
    Disposable subscription = userSubscriptions.get(userId);
    if (subscription != null && !subscription.isDisposed()) {
        subscription.dispose();
        userSubscriptions.remove(userId);
    }
    
    // 发送停止事件
    SSEServer.sendMsg(userId, "stopped", SSEMsgType.FINISH);
}

@Slf4j
@RestController
@RequestMapping("/api/chat")
public class ChatController {
    
    @Autowired
    private ChatService chatService;
    
    /**
     * 发送聊天消息
     */
    @PostMapping("/send")
    public ChatResponse sendMessage(@RequestBody ChatRequest request) {
        log.info("收到聊天请求，用户: {}, 消息: {}, 模式: {}", 
                request.getCurrentUserName(), request.getMessage(), request.getMode());
        
        try {
            // 异步处理聊天，通过 SSE 推送结果
            chatService.sendMessageAsync(
                    request.getCurrentUserName(),
                    request.getMessage(),
                    request.getMode()
            );
            return ChatResponse.success();
        } catch (Exception e) {
            log.error("处理聊天请求失败", e);
            return ChatResponse.error("处理请求失败: " + e.getMessage());
        }
    }
}

// 使用 PromptTemplate 构建复杂提示词
PromptTemplate promptTemplate = new PromptTemplate(
    "你是一个{role}，请回答以下问题：{question}"
);

Prompt prompt = promptTemplate.create(Map.of(
    "role", "股票分析师",
    "question", "如何分析一只股票？"
));

// 通过 ChatOptions 调整模型参数
ChatOptions options = DeepSeekChatOptions.builder()
        .temperature(0.7)        // 温度：控制随机性
        .maxTokens(2000)         // 最大 token 数
        .topP(0.9)               // Top-P 采样
        .build();

Flux<String> stream = chatClient
        .prompt(prompt)
        .options(options)
        .stream()
        .content();

// 构建多轮对话
List<Message> messages = List.of(
    new UserMessage("你好"),
    new AssistantMessage("你好！有什么可以帮助你的？"),
    new UserMessage("介绍一下你自己")
);

Prompt prompt = new Prompt(messages);

// 添加重试机制
stream
    .retry(3)  // 失败时重试 3 次
    .timeout(Duration.ofSeconds(30))  // 30 秒超时
    .doOnError(error -> {
        // 记录错误日志
        log.error("AI 调用失败", error);
    });

// 限制同时进行的对话数量
private final Semaphore semaphore = new Semaphore(10);

public void processChat(String userId, String message) {
    if (!semaphore.tryAcquire()) {
        throw new RuntimeException("系统繁忙，请稍后重试");
    }
    
    try {
        // 处理对话
    } finally {
        semaphore.release();
    }
}

@Cacheable(value = "ai-responses", key = "#message")
public String getCachedResponse(String message) {
    // 对于常见问题，使用缓存避免重复调用 API
}

@Async
@Override
public void sendMessageAsync(String userId, String message, String mode) {
    // 使用 @Async 注解，避免阻塞主线程
}