Fix[poll]:중복투표오류해결

Author: GarakChoi

backend/src/main/java/com/ai/lawyer/domain/chatbot/service/ChatBotService.java

@@ -6,10 +6,6 @@
 import com.ai.lawyer.domain.chatbot.dto.ChatDto.ChatResponse;
 import com.ai.lawyer.domain.chatbot.entity.History;
 import com.ai.lawyer.domain.chatbot.repository.HistoryRepository;
-import com.ai.lawyer.infrastructure.kafka.dto.ChatPostProcessEvent;
-import com.ai.lawyer.infrastructure.kafka.dto.DocumentDto;
-import com.ai.lawyer.domain.member.entity.Member;
-import com.ai.lawyer.domain.member.repositories.MemberRepository;
 import com.ai.lawyer.global.qdrant.service.QdrantService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,12 +20,9 @@
 import org.springframework.ai.chat.prompt.PromptTemplate;
 import org.springframework.ai.document.Document;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 import java.util.HashMap;
 import java.util.List;
@@ -44,78 +37,48 @@ public class ChatBotService {
     private final ChatClient chatClient;
     private final QdrantService qdrantService;
     private final HistoryService historyService;
-    private final MemberRepository memberRepository;
+    private final AsyncPostChatProcessingService asyncPostChatProcessingService;
+
     private final HistoryRepository historyRepository;
     private final ChatMemoryRepository chatMemoryRepository;
 
-    // KafkaTemplate 주입
-    private final KafkaTemplate<String, ChatPostProcessEvent> kafkaTemplate;
-
     @Value("${custom.ai.system-message}")
     private String systemMessageTemplate;
 
-    // Kafka 토픽 이름 -> 추후 application.yml로 이동 고려
-    private static final String POST_PROCESSING_TOPIC = "chat-post-processing";
-
     // 핵심 로직
     // 멤버 조회 -> 벡터 검색 -> 프롬프트 생성 -> LLM 호출 (스트림) -> Kafka 이벤트 발행 -> 응답 반환
     @Transactional
     public Flux<ChatResponse> sendMessage(Long memberId, ChatRequest chatRequestDto, Long roomId) {
-        return Mono.fromCallable(() -> {
 
-            // 벡터 검색 (판례, 법령) (블로킹)
-            List<Document> similarCaseDocuments = qdrantService.searchDocument(chatRequestDto.getMessage(), "type", "판례");
-            List<Document> similarLawDocuments = qdrantService.searchDocument(chatRequestDto.getMessage(), "type", "법령");
+        // 벡터 검색 (판례, 법령)
+        List<Document> similarCaseDocuments = qdrantService.searchDocument(chatRequestDto.getMessage(), "type", "판례");
+        List<Document> similarLawDocuments = qdrantService.searchDocument(chatRequestDto.getMessage(), "type", "법령");
 
-            String caseContext = formatting(similarCaseDocuments);
-            String lawContext = formatting(similarLawDocuments);
+        String caseContext = formatting(similarCaseDocuments);
+        String lawContext = formatting(similarLawDocuments);
 
-            // 채팅방 조회 또는 생성 (블로킹)
-            History history = getOrCreateRoom(memberId, roomId);
+        // 채팅방 조회 또는 생성
+        History history = getOrCreateRoom(memberId, roomId);
 
-            // 메시지 기억 관리 (User 메시지 추가)
-            ChatMemory chatMemory = saveChatMemory(chatRequestDto, history);
+        // 메시지 기억 관리
+        ChatMemory chatMemory = saveChatMemory(chatRequestDto, history);
 
-            // 프롬프트 생성
-            Prompt prompt = getPrompt(caseContext, lawContext, chatMemory, history);
+        // 프롬프트 생성
+        Prompt prompt = getPrompt(caseContext, lawContext, chatMemory, history);
 
-            // 준비된 데이터를 담은 컨텍스트 객체 반환
-            return new PreparedChatContext(prompt, history, similarCaseDocuments, similarLawDocuments);
-        })
-        .subscribeOn(Schedulers.boundedElastic()) // 블로킹 작업을 별도 스레드에서 실행
-        .flatMapMany(context -> {
-            // LLM 스트리밍 호출 및 클라이언트에게 즉시 응답
-            return chatClient.prompt(context.prompt)
+        return chatClient.prompt(prompt)
                 .stream()
                 .content()
                 .collectList()
                 .map(fullResponseList -> String.join("", fullResponseList))
-                .doOnNext(fullResponse -> {
-
-                    // Document를 DTO로 변환
-                    List<DocumentDto> caseDtos = context.similarCaseDocuments.stream().map(DocumentDto::from).collect(Collectors.toList());
-                    List<DocumentDto> lawDtos = context.similarLawDocuments.stream().map(DocumentDto::from).collect(Collectors.toList());
-
-                    // Kafka로 보낼 이벤트 객체
-                    ChatPostProcessEvent event = new ChatPostProcessEvent(
-                            context.history.getHistoryId(),
-                            chatRequestDto.getMessage(),
-                            fullResponse,
-                            caseDtos,
-                            lawDtos
-                    );
-
-                    // Kafka 이벤트 발행
-                    kafkaTemplate.send(POST_PROCESSING_TOPIC, event);
-
-                })
-                .map(fullResponse -> createChatResponse(context.history, fullResponse, context.similarCaseDocuments, context.similarLawDocuments))
+                .doOnNext(fullResponse -> asyncPostChatProcessingService.processHandlerTasks(history.getHistoryId(), chatRequestDto.getMessage(), fullResponse, similarCaseDocuments, similarLawDocuments))
+                .map(fullResponse -> createChatResponse(history, fullResponse, similarCaseDocuments, similarLawDocuments))
                 .flux()
                 .onErrorResume(throwable -> {
-                    log.error("스트리밍 처리 중 에러 발생 (historyId: {})", context.history.getHistoryId(), throwable);
-                    return Flux.just(handleError(context.history));
+                    log.error("스트리밍 처리 중 에러 발생 (historyId: {})", history.getHistoryId(), throwable);
+                    return Flux.just(handleError(history));
                 });
-        });
+
     }
 
     private ChatResponse createChatResponse(History history, String fullResponse, List<Document> cases, List<Document> laws) {
@@ -185,23 +148,4 @@ private ChatResponse handleError(History history) {
                 .build();
     }
 
-    /**
-     * 블로킹 작업에서 준비된 데이터를 담는 컨텍스트 클래스
-     * 리액티브 체인에서 데이터를 전달하기 위한 내부 클래스
-     */
-    private static class PreparedChatContext {
-        final Prompt prompt;
-        final History history;
-        final List<Document> similarCaseDocuments;
-        final List<Document> similarLawDocuments;
-
-        PreparedChatContext(Prompt prompt, History history,
-                          List<Document> similarCaseDocuments,
-                          List<Document> similarLawDocuments) {
-            this.prompt = prompt;
-            this.history = history;
-            this.similarCaseDocuments = similarCaseDocuments;
-            this.similarLawDocuments = similarLawDocuments;
-        }
-    }
 }