Chat 기능 bug fix

README.md

@@ -8,9 +8,9 @@
 - [서비스 소개](#-서비스-소개)
 - [기술 스택](#-기술-스택)
 - [프로젝트 설치 및 실행법](#-프로젝트-설치-및-실행법)
-- [프로젝트 구조](#-프로젝트-구조)
-- [주요기능](#-주요기능)
-- [Developer](#-developer)
+- [프로젝트 구조 ](#-프로젝트-구조 )
+- [주요 기능](#-주요기능)
+- [Developer](#-Developer)
 
 
 ## 💁‍♀️ 서비스 소개

src/main/java/com/sounganization/botanify/common/config/TransactionConfig.java

@@ -0,0 +1,19 @@
+package com.sounganization.botanify.common.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Configuration
+public class TransactionConfig {
+
+    @Bean
+    public TransactionTemplate transactionTemplate(PlatformTransactionManager transactionManager) {
+        TransactionTemplate template = new TransactionTemplate(transactionManager);
+        template.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
+        template.setTimeout(30);
+        return template;
+    }
+}

src/main/java/com/sounganization/botanify/common/config/async/AsyncConfig.java

@@ -0,0 +1,36 @@
+package com.sounganization.botanify.common.config.async;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+
+@Configuration
+@EnableAsync
+@Slf4j
+public class AsyncConfig implements AsyncConfigurer {
+
+    @Bean(name = "chatThreadPoolTaskExecutor")
+    public Executor getAsyncExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(5);
+        executor.setMaxPoolSize(10);
+        executor.setQueueCapacity(25);
+        executor.setThreadNamePrefix("ChatAsync-");
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        executor.initialize();
+        return executor;
+    }
+
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+        return (ex, method, params) -> {
+            log.error("비동기 메서드 {}에서 예외 발생: {}", method.getName(), ex.getMessage());
+        };
+    }
+}

src/main/java/com/sounganization/botanify/common/config/redis/RedisConfig.java

@@ -1,20 +1,24 @@
 package com.sounganization.botanify.common.config.redis;
 
 import com.sounganization.botanify.common.config.websocket.ChatMessageListener;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.listener.PatternTopic;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 @Configuration
 @ConditionalOnProperty(name = "spring.redis.enabled", havingValue = "true", matchIfMissing = true)
+@Slf4j
 public class RedisConfig {
 
     @Value("${spring.redis.master.host}")
@@ -46,13 +50,36 @@ public RedisTemplate<String, String> redisTemplate() {
         return redisTemplate;
     }
 
+    @Bean
+    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
+        StringRedisTemplate template = new StringRedisTemplate(connectionFactory);
+        template.setEnableTransactionSupport(true);
+        return template;
+    }
+
+    @Bean
+    public ThreadPoolTaskExecutor redisThreadPoolTaskExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(10);
+        executor.setMaxPoolSize(20);
+        executor.setQueueCapacity(100);
+        executor.setThreadNamePrefix("RedisExecutor-");
+        executor.initialize();
+        return executor;
+    }
+
     @Bean
     public RedisMessageListenerContainer redisMessageListenerContainer(
             RedisConnectionFactory connectionFactory,
             ChatMessageListener chatMessageListener) {
         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+
         container.setConnectionFactory(connectionFactory);
-        container.addMessageListener(chatMessageListener, new PatternTopic("chat_room_*"));
+        container.setTaskExecutor(redisThreadPoolTaskExecutor());
+        container.setRecoveryInterval(5000L);
+        container.addMessageListener(chatMessageListener, new PatternTopic("chat_room_broadcast_*"));
+
+        log.info("최적화된 설정으로 Redis message listener container 구성");
         return container;
     }
 }

src/main/java/com/sounganization/botanify/common/config/websocket/ChatMessageListener.java

@@ -1,29 +1,67 @@
 package com.sounganization.botanify.common.config.websocket;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.sounganization.botanify.common.config.websocket.service.WebSocketChatService;
 import com.sounganization.botanify.domain.chat.dto.req.ChatMessageReqDto;
+import com.sounganization.botanify.domain.chat.entity.ChatMessage;
+import com.sounganization.botanify.domain.chat.service.ChatMessageService;
+import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
 
-@Component
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+@Service
 @RequiredArgsConstructor
 @Slf4j
 public class ChatMessageListener implements MessageListener {
     private final ObjectMapper objectMapper;
-    private final WebSocketChatService webSocketChatService;
+    private final ChatMessageService chatMessageService;
+    private final Set<String> processedMessages = ConcurrentHashMap.newKeySet();
 
     @Override
-    public void onMessage(Message message, byte[] pattern) {
+    public void onMessage(@NonNull Message message, byte[] pattern) {
         try {
-            String messageBody = new String(message.getBody());
-            ChatMessageReqDto chatMessage = objectMapper.readValue(messageBody, ChatMessageReqDto.class);
-            webSocketChatService.handleChatMessage(chatMessage);
+            String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);
+            if (!processedMessages.add(messageContent)) {
+                log.debug("Listener를 통해 이미 처리된 메시지입니다. 건너뜁니다.");
+                return;
+            }
+
+            ChatMessageReqDto chatMessage = objectMapper.readValue(messageContent, ChatMessageReqDto.class);
+
+            if (chatMessage.source() != ChatMessageReqDto.MessageSource.WEBSOCKET) {
+                log.debug("WebSocket 메시지가 아니므로 건너뜁니다.");
+                return;
+            }
+
+            ChatMessage savedMessage = chatMessageService.findExistingMessage(
+                    chatMessage.roomId(),
+                    chatMessage.senderId(),
+                    chatMessage.content()
+            );
+
+            if (savedMessage != null) {
+                chatMessageService.markMessageAsDelivered(savedMessage.getId());
+                log.debug("브로드캐스트 후 메시지가 전달됨으로 표시되었습니다 - messageId: {}", savedMessage.getId());
+            }
+
+            removeProcessedMessageAfterDelay(messageContent);
+
         } catch (Exception e) {
-            log.error("채팅 메시지 처리 중 오류 발생: {}", e.getMessage());
+            log.error("Redis Pub/Sub에서 메시지 처리 실패", e);
         }
     }
-}
+
+    private void removeProcessedMessageAfterDelay(String messageContent) {
+        CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS).execute(() -> {
+            processedMessages.remove(messageContent);
+        });
+    }
+}

src/main/java/com/sounganization/botanify/common/config/websocket/WebSocketConfig.java

@@ -2,6 +2,7 @@
 
 import com.sounganization.botanify.common.config.websocket.handler.ChatWebSocketHandler;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.socket.config.annotation.EnableWebSocket;
 import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
@@ -10,8 +11,8 @@
 @Configuration
 @EnableWebSocket
 @RequiredArgsConstructor
+@Slf4j
 public class WebSocketConfig implements WebSocketConfigurer {
-
     private final ChatWebSocketHandler chatWebSocketHandler;
 
     @Override

src/main/java/com/sounganization/botanify/common/config/websocket/handler/ChatWebSocketHandler.java

@@ -17,40 +17,41 @@
 @Slf4j
 @RequiredArgsConstructor
 public class ChatWebSocketHandler extends TextWebSocketHandler {
-
     private final ObjectMapper objectMapper;
     private final WebSocketChatService webSocketChatService;
     private final ConnectionFailureHandler connectionFailureHandler;
 
-    @Override
-    public void afterConnectionEstablished(WebSocketSession session) {
-        log.info("새로운 WebSocket 연결이 열렸습니다. 세션 ID: {}", session.getId());
-    }
-
     @Override
     protected void handleTextMessage(@NonNull WebSocketSession session, @NonNull TextMessage message) throws Exception {
         String payload = message.getPayload();
+        log.info("WebSocket 메시지를 수신했습니다: {}", payload);
+
         ChatMessageReqDto chatMessage = objectMapper.readValue(payload, ChatMessageReqDto.class);
 
         try {
             switch (chatMessage.type()) {
                 case ENTER:
+                    log.info("사용자 {}이(가) 방 {}에 입장합니다.", chatMessage.senderId(), chatMessage.roomId());
                     webSocketChatService.handleEnterRoom(session, chatMessage.roomId(), chatMessage.senderId());
                     break;
                 case TALK:
                     if (!session.isOpen()) {
+                        log.warn("메시지를 보내는 중 세션이 종료되었습니다.");
                         connectionFailureHandler.handleConnectionFailure(chatMessage);
                         return;
                     }
+                    log.info("사용자 {}이(가) 방 {}에서 TALK 메시지를 처리 중입니다.",
+                            chatMessage.senderId(), chatMessage.roomId());
                     webSocketChatService.handleChatMessage(chatMessage);
                     break;
                 case LEAVE:
+                    log.info("사용자 {}이(가) 방 {}에서 나갑니다.", chatMessage.senderId(), chatMessage.roomId());
                     webSocketChatService.handleLeaveRoom(chatMessage.roomId(), chatMessage.senderId());
                     break;
             }
         } catch (Exception e) {
+            log.error("메시지 처리 중 오류 발생: ", e);
             handleError(session, e);
-            // 메시지 처리 중 오류 발생시 연결 실패로 처리
             if (chatMessage.type() == ChatMessageReqDto.MessageType.TALK) {
                 connectionFailureHandler.handleConnectionFailure(chatMessage);
             }

src/main/java/com/sounganization/botanify/common/config/websocket/handler/ConnectionFailureHandler.java

@@ -54,7 +54,8 @@ private void retryMessageDelivery(ChatMessage message) {
                 convertToDtoMessageType(message.getType()),
                 message.getChatRoom().getId(),
                 message.getSenderId(),
-                message.getContent()
+                message.getContent(),
+                ChatMessageReqDto.MessageSource.WEBSOCKET
         );
 
         try {

src/main/java/com/sounganization/botanify/common/config/websocket/service/MessageBroadcastService.java

@@ -6,32 +6,68 @@
 import com.sounganization.botanify.domain.chat.dto.req.ChatMessageReqDto;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.web.socket.WebSocketSession;
 
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 
 @Service
 @RequiredArgsConstructor
 @Slf4j
 public class MessageBroadcastService {
-
     private final ObjectMapper objectMapper;
+    private final StringRedisTemplate redisTemplate;
     private final Map<Long, Map<Long, WebSocketSession>> chatRoomSessions = new ConcurrentHashMap<>();
+    private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();
 
     public void broadcastMessage(Long roomId, ChatMessageReqDto message) {
+        String messageId = generateMessageId(roomId, message);
+
+        if (!processedMessageIds.add(messageId)) {
+            log.debug("이미 브로드캐스트된 메시지입니다. 건너뜁니다: {}", messageId);
+            return;
+        }
+
         try {
-            String messageJson = objectMapper.writeValueAsString(message);
-            WebSocketUtils.broadcastMessageToRoom(roomId, messageJson, chatRoomSessions);
+            log.debug("방 {}에 메시지를 브로드캐스트 중입니다.", roomId);
+            redisTemplate.convertAndSend(
+                    "chat_room_broadcast_" + roomId,
+                    objectMapper.writeValueAsString(message)
+            );
+
+            removeProcessedMessageIdAfterDelay(messageId);
         } catch (JsonProcessingException e) {
-            log.error("메시지 변환 중 오류 발생: {}", e.getMessage());
+            log.error("Redis에 메시지 게시 실패", e);
+            processedMessageIds.remove(messageId);
         }
     }
 
+    private String generateMessageId(Long roomId, ChatMessageReqDto message) {
+        return String.format("%d:%d:%d:%s",
+                roomId,
+                message.senderId(),
+                message.content().hashCode(),
+                LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS)
+        );
+    }
+
+    private void removeProcessedMessageIdAfterDelay(String messageId) {
+        CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS).execute(() -> {
+            processedMessageIds.remove(messageId);
+        });
+    }
+
     public void addSession(Long roomId, Long userId, WebSocketSession session) {
         chatRoomSessions.computeIfAbsent(roomId, k -> new ConcurrentHashMap<>())
                 .put(userId, session);
+        log.debug("roomId: {}, userId: {}에 대한 세션을 추가했습니다.", roomId, userId);
     }
 
     public void removeSession(Long roomId, Long userId) {
@@ -40,10 +76,12 @@ public void removeSession(Long roomId, Long userId) {
             WebSocketSession session = roomSessions.remove(userId);
             if (session != null) {
                 WebSocketUtils.closeSession(session);
+                log.debug("roomId: {}, userId: {}에 대한 세션을 제거했습니다.", roomId, userId);
             }
             if (roomSessions.isEmpty()) {
                 chatRoomSessions.remove(roomId);
+                log.debug("빈 방을 제거했습니다: {}", roomId);
             }
         }
     }
-}
+}