work

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

Author: GarakChoi

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

@@ -6,6 +6,10 @@
 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;
@@ -20,9 +24,12 @@
 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;
@@ -37,48 +44,78 @@ public class ChatBotService {
     private final ChatClient chatClient;
     private final QdrantService qdrantService;
     private final HistoryService historyService;
-    private final AsyncPostChatProcessingService asyncPostChatProcessingService;
-
+    private final MemberRepository memberRepository;
     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);
 
-        // 메시지 기억 관리
-        ChatMemory chatMemory = saveChatMemory(chatRequestDto, history);
+            // 메시지 기억 관리 (User 메시지 추가)
+            ChatMemory chatMemory = saveChatMemory(chatRequestDto, history);
 
-        // 프롬프트 생성
-        Prompt prompt = getPrompt(caseContext, lawContext, chatMemory, history);
+            // 프롬프트 생성
+            Prompt prompt = getPrompt(caseContext, lawContext, chatMemory, history);
 
-        return chatClient.prompt(prompt)
+            // 준비된 데이터를 담은 컨텍스트 객체 반환
+            return new PreparedChatContext(prompt, history, similarCaseDocuments, similarLawDocuments);
+        })
+        .subscribeOn(Schedulers.boundedElastic()) // 블로킹 작업을 별도 스레드에서 실행
+        .flatMapMany(context -> {
+            // LLM 스트리밍 호출 및 클라이언트에게 즉시 응답
+            return chatClient.prompt(context.prompt)
                 .stream()
                 .content()
                 .collectList()
                 .map(fullResponseList -> String.join("", fullResponseList))
-                .doOnNext(fullResponse -> asyncPostChatProcessingService.processHandlerTasks(history.getHistoryId(), chatRequestDto.getMessage(), fullResponse, similarCaseDocuments, similarLawDocuments))
-                .map(fullResponse -> createChatResponse(history, fullResponse, similarCaseDocuments, similarLawDocuments))
+                .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))
                 .flux()
                 .onErrorResume(throwable -> {
-                    log.error("스트리밍 처리 중 에러 발생 (historyId: {})", history.getHistoryId(), throwable);
-                    return Flux.just(handleError(history));
+                    log.error("스트리밍 처리 중 에러 발생 (historyId: {})", context.history.getHistoryId(), throwable);
+                    return Flux.just(handleError(context.history));
                 });
-
+        });
     }
 
     private ChatResponse createChatResponse(History history, String fullResponse, List<Document> cases, List<Document> laws) {
@@ -148,4 +185,23 @@ 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;
+        }
+    }
 }