Merge to main

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/in/web/presentation/ChatMessageQueryController.java

@@ -40,8 +40,6 @@ public GetMessagesResponseWrapperVo getMessagesHistory(
             @RequestHeader("X-Member-Uuid") String memberUuid,
             @ModelAttribute @Valid GetMessagesRequestVo getMessagesRequestVo
     ) {
-        log.info("lastMessageSentAt: {}, lastMessageUuid: {}", getMessagesRequestVo.getLastMessageSentAt(), getMessagesRequestVo.getLastMessageUuid());
-
 
         return chatMessageQueryVoMapper.toGetMessagesResponseWrapperVo(
                 chatMessageQueryUseCase.getMessages(

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/in/websocket/exception/WebSocketErrorMessage.java

@@ -5,9 +5,11 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
 
 @Getter
 @AllArgsConstructor
+@Slf4j
 public class WebSocketErrorMessage {
     private final int code;
     private final String message;
@@ -21,6 +23,7 @@ public String toJson() {
         try {
             return new ObjectMapper().writeValueAsString(this);
         } catch (JsonProcessingException e) {
+            log.error("에러 직렬화 실패: {}", e.getMessage());
             return "{\"code\":500,\"message\":\"에러 직렬화 실패\"}";
         }
     }

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/kafka/KafkaChatMessageConsumer.java → src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/kafka/consumer/KafkaChatMessageConsumer.java

@@ -1,8 +1,11 @@
-package com.chalnakchalnak.chatservice.chatmessage.adpater.out.kafka;
+package com.chalnakchalnak.chatservice.chatmessage.adpater.out.kafka.consumer;
 
 import com.chalnakchalnak.chatservice.chatmessage.adpater.out.redis.pub.RedisMessagePublisher;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.ChatMessageDto;
 import com.chalnakchalnak.chatservice.chatmessage.application.port.out.ChatMessageRepositoryPort;
+import com.chalnakchalnak.chatservice.common.exception.BaseException;
+import com.chalnakchalnak.chatservice.common.response.BaseResponseStatus;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.mongodb.DuplicateKeyException;
 import lombok.RequiredArgsConstructor;
@@ -12,8 +15,10 @@
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 @Component
 @RequiredArgsConstructor
@@ -24,7 +29,7 @@ public class KafkaChatMessageConsumer {
     private final ChatMessageRepositoryPort chatMessageRepositoryPort;
     private final ObjectMapper objectMapper;
 
-    @Transactional
+//    @Transactional
     @KafkaListener(topics = "chat.private.room")
     public void consume(List<String> payloads, Acknowledgment ack) {
 
@@ -36,10 +41,17 @@ public void consume(List<String> payloads, Acknowledgment ack) {
             }
 
             chatMessageRepositoryPort.bulkSaveMessages(messageList);
-            chatMessageRepositoryPort.bulkUpsertMessages(messageList);
+            chatMessageRepositoryPort.bulkUpsertSummary(messageList);
 
             for (ChatMessageDto message : messageList) {
-                redisMessagePublisher.publish(message.getChatRoomUuid(), objectMapper.writeValueAsString(message));
+                CompletableFuture.runAsync(() -> {
+                    try {
+                        redisMessagePublisher.publish(message.getChatRoomUuid(), objectMapper.writeValueAsString(message));
+                    } catch (Exception e) {
+                        log.error("Redis 메시지 발행 실패", e.getMessage());
+                        throw new BaseException(BaseResponseStatus.REDIS_PUBLISH_ERROR);
+                    }
+                });
             }
 
             ack.acknowledge();

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/kafka/KafkaChatMessageProducer.java → src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/kafka/producer/KafkaChatMessageProducer.java

@@ -1,4 +1,4 @@
-package com.chalnakchalnak.chatservice.chatmessage.adpater.out.kafka;
+package com.chalnakchalnak.chatservice.chatmessage.adpater.out.kafka.producer;
 
 import com.chalnakchalnak.chatservice.chatmessage.adpater.in.websocket.exception.WebSocketErrorMessage;
 import com.chalnakchalnak.chatservice.chatmessage.adpater.out.kafka.mapper.KafkaEventDtoMapper;
@@ -18,6 +18,7 @@
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
 
+import java.time.LocalDateTime;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -56,7 +57,6 @@ public void publishChatMessage(SendMessageRequestDto sendMessageRequestDto) {
                         sendErrorToUser(chatMessageDto.getSenderUuid());
                     }
                 });
-
     }
 
     public String toJson(ChatMessageDto chatMessageDto) {

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/entity/ChatMessageDocument.java → src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/document/ChatMessageDocument.java

@@ -1,17 +1,22 @@
-package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity;
+package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document;
 
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import org.bson.types.ObjectId;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.CompoundIndex;
+import org.springframework.data.mongodb.core.index.CompoundIndexes;
 import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.time.LocalDateTime;
 
 @Document(collection = "chat_message")
+@CompoundIndexes({
+        @CompoundIndex(name = "idx_message_sentAt", def = "{'messageUuid': 1, 'sentAt': -1}")
+})
 @Getter
 @NoArgsConstructor(access = AccessLevel.PROTECTED)
 public class ChatMessageDocument {

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/entity/ChatReadCheckPointDocument.java → src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/document/ChatReadCheckPointDocument.java

@@ -1,15 +1,24 @@
-package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity;
+package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document;
 
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.CompoundIndex;
+import org.springframework.data.mongodb.core.index.CompoundIndexes;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.time.LocalDateTime;
 
 @Document(collection = "chat_read_checkpoint")
+@CompoundIndexes({
+        @CompoundIndex(
+                name = "idx_message_read_checkpoint",
+                def = "{'chatRoomUuid': 1, 'memberUuid': 1}",
+                unique = true
+        )
+})
 @Getter
 @NoArgsConstructor(access = AccessLevel.PROTECTED)
 public class ChatReadCheckPointDocument {

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/entity/ReplyPreview.java → src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/document/ReplyPreview.java

@@ -1,4 +1,4 @@
-package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity;
+package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document;
 
 import com.chalnakchalnak.chatservice.chatmessage.domain.enums.MessageType;
 import lombok.Builder;

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/mapper/ChatMessageDocumentMapper.java

@@ -1,7 +1,7 @@
 package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.mapper;
 
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatMessageDocument;
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ReplyPreview;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatMessageDocument;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ReplyPreview;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.ChatMessageDto;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.ReplyPreviewDto;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.out.GetMessagesResponseDto;

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/repository/ChatMessageMongoRepository.java

@@ -1,7 +1,6 @@
 package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.repository;
 
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatMessageDocument;
-import org.bson.types.ObjectId;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatMessageDocument;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.mongodb.repository.MongoRepository;
 

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/repository/ChatMessageQueryRepository.java

@@ -1,7 +1,7 @@
 package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.repository;
 
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatMessageDocument;
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatReadCheckPointDocument;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatMessageDocument;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatReadCheckPointDocument;
 import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.mapper.ChatMessageDocumentMapper;
 import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.mapper.ChatReadCheckPointDocumentMapper;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.in.GetMessagesRequestDto;
@@ -11,7 +11,6 @@
 import com.chalnakchalnak.chatservice.chatmessage.application.port.out.ChatMessageQueryRepositoryPort;
 import com.chalnakchalnak.chatservice.chatroom.application.port.out.ChatRoomMemberExitRepositoryPort;
 import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.stereotype.Repository;
 
@@ -28,34 +27,47 @@ public class ChatMessageQueryRepository implements ChatMessageQueryRepositoryPor
     private final ChatRoomMemberExitRepositoryPort chatRoomMemberExitRepositoryPort;
     private final ChatReadCheckPointDocumentMapper chatReadCheckPointDocumentMapper;
 
+    private static final LocalDateTime DEFAULT_EXITED_AT = LocalDateTime.of(1970, 1, 1, 0, 0);
+
     @Override
-    public List<GetMessagesResponseDto> getMessages(GetMessagesRequestDto getMessagesRequestDto) {
+    public List<GetMessagesResponseDto> getMessages(GetMessagesRequestDto dto) {
+        if (dto.getLastMessageSentAt() == null) {
+            return getFirstPageMessages(dto);
+        }
+        return getPagedMessages(dto);
+    }
 
-        final PageRequest pageable = PageRequest.of(0, getMessagesRequestDto.getLimit());
-        List<ChatMessageDocument> messages;
+    private List<GetMessagesResponseDto> getFirstPageMessages(GetMessagesRequestDto dto) {
+        LocalDateTime exitedAt = chatRoomMemberExitRepositoryPort
+                .findByChatRoomUuidAndMemberUuid(dto.getChatRoomUuid(), dto.getMemberUuid())
+                .map(exit -> exit.getExitedAt())
+                .orElse(DEFAULT_EXITED_AT);
 
-        if (getMessagesRequestDto.getLastMessageSentAt() == null) {
-            final LocalDateTime exitedAt = chatRoomMemberExitRepositoryPort
-                    .findByChatRoomUuidAndMemberUuid(getMessagesRequestDto.getChatRoomUuid(), getMessagesRequestDto.getMemberUuid())
-                    .map(chatRoomMemberExitDto -> chatRoomMemberExitDto.getExitedAt())
-                    .orElse(LocalDateTime.of(1970, 1, 1, 0, 0));
+        List<ChatMessageDocument> messages = chatMessageMongoRepository
+                .findByChatRoomUuidAndSentAtGreaterThanOrderBySentAtDescMessageUuidDesc(
+                        dto.getChatRoomUuid(), exitedAt, PageRequest.of(0, dto.getLimit()));
 
-            messages = chatMessageMongoRepository.findByChatRoomUuidAndSentAtGreaterThanOrderBySentAtDescMessageUuidDesc(
-                    getMessagesRequestDto.getChatRoomUuid(), exitedAt, pageable);
-        } else {
-            LocalDateTime lastSentAt = getMessagesRequestDto.getLastMessageSentAt();
+        return toResponseDtoList(messages);
+    }
 
-            messages = chatMessageMongoRepository.findByChatRoomUuidAndSentAtLessThanOrderBySentAtDescMessageUuidDesc(
-                    getMessagesRequestDto.getChatRoomUuid(), lastSentAt, pageable);
+    private List<GetMessagesResponseDto> getPagedMessages(GetMessagesRequestDto dto) {
+        PageRequest pageable = PageRequest.of(0, dto.getLimit());
 
-            if (messages.isEmpty()) {
-                messages = chatMessageMongoRepository.findByChatRoomUuidAndSentAtAndMessageUuidLessThanOrderBySentAtDescMessageUuidDesc(
-                        getMessagesRequestDto.getChatRoomUuid(), lastSentAt, getMessagesRequestDto.getLastMessageUuid(), pageable);
-            }
+        List<ChatMessageDocument> messages = chatMessageMongoRepository
+                .findByChatRoomUuidAndSentAtLessThanOrderBySentAtDescMessageUuidDesc(
+                        dto.getChatRoomUuid(), dto.getLastMessageSentAt(), pageable);
+
+        if (messages.isEmpty()) {
+            messages = chatMessageMongoRepository
+                    .findByChatRoomUuidAndSentAtAndMessageUuidLessThanOrderBySentAtDescMessageUuidDesc(
+                            dto.getChatRoomUuid(), dto.getLastMessageSentAt(), dto.getLastMessageUuid(), pageable);
         }
 
-        return messages
-                .stream()
+        return toResponseDtoList(messages);
+    }
+
+    private List<GetMessagesResponseDto> toResponseDtoList(List<ChatMessageDocument> messages) {
+        return messages.stream()
                 .map(chatMessageDocumentMapper::toGetMessagesResponseDto)
                 .toList();
     }

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/repository/ChatMessageRepository.java

@@ -14,6 +14,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -35,7 +36,7 @@ public class ChatMessageRepository implements ChatMessageRepositoryPort {
     private final ChatMessageMapper chatMessageMapper;
 
     @Override
-    public void bulkUpsertMessages(List<ChatMessageDto> messageDtoList) {
+    public void bulkUpsertSummary(List<ChatMessageDto> messageDtoList) {
         try {
             List<String> chatRoomUuids = messageDtoList.stream()
                     .map(ChatMessageDto::getChatRoomUuid)
@@ -65,22 +66,14 @@ public void bulkUpsertMessages(List<ChatMessageDto> messageDtoList) {
                 );
             }
         } catch (Exception e) {
+            log.error("MongoDB Bulk Upsert 실패: {}", e.getMessage());
             throw new BaseException(BaseResponseStatus.FAILED_MESSAGE_PROCESSING);
         }
     }
 
     @Override
     public void bulkSaveMessages(List<ChatMessageDto> messages) {
-        try {
-            chatMessageBulkOps.saveMessages(messages);
-
-        } catch (MongoBulkWriteException e) {
-            log.error("MongoDB Bulk Insert 실패: {}", e.getMessage(), e);
-            throw new BaseException(BaseResponseStatus.FAILED_MESSAGE_PROCESSING);
-        } catch (Exception e) {
-            log.error("MongoDB 저장 중 예외 발생", e);
-            throw new BaseException(BaseResponseStatus.FAILED_MESSAGE_PROCESSING);
-        }
+        chatMessageBulkOps.saveMessages(messages);
     }
 
     @Override

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/repository/ChatReadCheckPointMongoRepository.java

@@ -1,6 +1,6 @@
 package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.repository;
 
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatReadCheckPointDocument;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatReadCheckPointDocument;
 import org.springframework.data.mongodb.repository.MongoRepository;
 
 import java.util.Optional;

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/mongo/repository/ChatReadCheckPointUpdater.java

@@ -1,6 +1,6 @@
 package com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.repository;
 
-import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.entity.ChatReadCheckPointDocument;
+import com.chalnakchalnak.chatservice.chatmessage.adpater.out.mongo.document.ChatReadCheckPointDocument;
 import com.chalnakchalnak.chatservice.chatmessage.application.dto.in.ReadMessageRequestDto;
 import com.chalnakchalnak.chatservice.chatmessage.application.port.out.ChatReadCheckPointUpdaterPort;
 import lombok.RequiredArgsConstructor;

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/s3/S3ImageKeyValidator.java

@@ -8,11 +8,13 @@
 import com.chalnakchalnak.chatservice.common.exception.BaseException;
 import com.chalnakchalnak.chatservice.common.response.BaseResponseStatus;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
 @Component
 @RequiredArgsConstructor
+@Slf4j
 public class S3ImageKeyValidator implements ImageKeyValidatorPort {
 
     private final AmazonS3 amazonS3;
@@ -59,8 +61,10 @@ private void validateS3Existence(String key) {
             amazonS3.getObjectMetadata(new GetObjectMetadataRequest(bucket, key));
         } catch (AmazonS3Exception e) {
             if (e.getStatusCode() == 404) {
+                log.warn("S3에서 이미지 파일을 찾을 수 없음: {}", key);
                 throw new BaseException(BaseResponseStatus.IMAGE_FILE_NOT_FOUND_IN_S3);
             }
+            log.error("S3 접근 중 오류 발생: {}", e.getMessage());
             throw new BaseException(BaseResponseStatus.FAILED_TO_ACCESS_S3);
         }
     }

src/main/java/com/chalnakchalnak/chatservice/chatmessage/adpater/out/websocket/message/SendMessageToClient.java

@@ -22,7 +22,7 @@ public void sendMessage(ReadMessageRequestDto readMessageRequestDto, String oppo
             String message = objectMapper.writeValueAsString(readMessageRequestDto);
             redisMessagePublisher.publishRead(opponentUuid, message);
         } catch (Exception e) {
-            log.error("클라이언트 메시지 수신 실패: {}", e.getMessage(), e);
+            log.error("클라이언트 메시지 수신 실패: {}", e.getMessage());
         }
     }
 }

src/main/java/com/chalnakchalnak/chatservice/chatmessage/application/port/out/ChatMessageRepositoryPort.java

@@ -8,7 +8,7 @@
 
 public interface ChatMessageRepositoryPort {
 
-    void bulkUpsertMessages(List<ChatMessageDto> messages);
+    void bulkUpsertSummary(List<ChatMessageDto> messages);
     void bulkSaveMessages(List<ChatMessageDto> messages);
     GetMessagesResponseDto findByMessageUuid(String messageUuid);
 

src/main/java/com/chalnakchalnak/chatservice/chatmessage/application/service/ChatMessageService.java

@@ -15,12 +15,14 @@
 import com.chalnakchalnak.chatservice.common.exception.BaseException;
 import com.chalnakchalnak.chatservice.common.response.BaseResponseStatus;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 
 @Service
 @RequiredArgsConstructor
+@Slf4j
 public class ChatMessageService implements ChatMessageUseCase {
 
     private final PublishChatMessagePort publishChatMessagePort;
@@ -62,6 +64,7 @@ public void updateReadCheckPoint(ReadMessageRequestDto readMessageRequestDto) {
                     readMessageMapper.toChatRoomSummaryUpdateEventByRead(readMessageRequestDto)
             );
         } catch (Exception e) {
+            log.error("읽음 체크포인트 업데이트 실패: {}", e.getMessage());
             throw new BaseException(BaseResponseStatus.FAILED_UPDATE_READ_CHECK_POINT, readMessageRequestDto.getMemberUuid());
         }
     }