[Feat/#223] Implement Chatting Feature (#227)

* [Feat/#84] Implement Chatting Room Feature

* [Feat/#223] Implement Chatting Feature

* [Feat/#223] Configure MongoDB client and implement message saving logic

* [Feat/#223] Implement sorting chat rooms by unread message count

* [Feat/#223] Implement logic to retrieve chat history

* [Feat/#223] Update SendTime Logic

* [Feat/#223] Enhance STOMP Authentication Handling

* [Feat/#223] Update File Path

* [Feat/#223] Add @SuperBuilder to ChatRoom and ChatRoomMember

build.gradle

@@ -72,15 +72,28 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-thymeleaf'
     implementation 'nz.net.ultraq.thymeleaf:thymeleaf-layout-dialect'
 
-    // kafka
+    // view
+    implementation 'org.springframework.boot:spring-boot-starter-freemarker'
+    developmentOnly 'org.springframework.boot:spring-boot-devtools'
+    implementation 'org.webjars.bower:bootstrap:4.3.1'
+    implementation 'org.webjars.bower:vue:2.5.16'
+    implementation 'org.webjars.bower:axios:0.17.1'
+    implementation 'com.google.code.gson:gson:2.8.0'
+
+
+    // WebSocket, Kafka, Stomp, MongoDB
+    implementation 'org.springframework.boot:spring-boot-starter-websocket'
+    implementation 'org.webjars:sockjs-client:1.1.2'
+    implementation 'org.webjars:stomp-websocket:2.3.3-1'
     implementation 'org.springframework.kafka:spring-kafka'
+    implementation 'org.springframework.boot:spring-boot-starter-data-mongodb'
+    annotationProcessor "org.springframework.boot:spring-boot-configuration-processor"
 
     // QueryDsl
     implementation 'com.querydsl:querydsl-jpa:5.0.0:jakarta'
     annotationProcessor "com.querydsl:querydsl-apt:5.0.0:jakarta"
     annotationProcessor "jakarta.annotation:jakarta.annotation-api"
     annotationProcessor "jakarta.persistence:jakarta.persistence-api"
-
 }
 
 tasks.named('test') {

src/main/java/com/example/waggle/domain/chat/config/StompHandler.java

@@ -0,0 +1,57 @@
+package com.example.waggle.domain.chat.config;
+
+import static org.springframework.messaging.simp.stomp.StompCommand.CONNECT;
+import static org.springframework.messaging.simp.stomp.StompCommand.SEND;
+import static org.springframework.messaging.simp.stomp.StompCommand.SUBSCRIBE;
+
+import com.example.waggle.domain.member.entity.Member;
+import com.example.waggle.domain.member.service.MemberQueryService;
+import com.example.waggle.global.exception.handler.MemberHandler;
+import com.example.waggle.global.payload.code.ErrorStatus;
+import com.example.waggle.global.security.service.TokenService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.MessageHeaderAccessor;
+import org.springframework.stereotype.Component;
+
+@Order(Ordered.HIGHEST_PRECEDENCE + 99) // 우선 순위를 높게 설정해서 SecurityFilter들 보다 앞서 실행되게 해준다.
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class StompHandler implements ChannelInterceptor {
+
+    private final TokenService tokenService;
+    private final MemberQueryService memberQueryService;
+
+    @Override
+    public Message<?> preSend(Message<?> message, MessageChannel channel) {
+        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
+        if (accessor.getCommand() == CONNECT || accessor.getCommand() == SEND || accessor.getCommand() == SUBSCRIBE) {
+            Member member = getMemberByAccessToken(getAccessToken(accessor));
+            accessor.setUser(() -> member.getUsername());
+        }
+        return message;
+    }
+
+    private String getAccessToken(StompHeaderAccessor accessor) {
+        String token = accessor.getFirstNativeHeader("Authorization");
+        if (token != null && token.startsWith("Bearer ")) {
+            return token.substring(7).trim();
+        }
+        throw new MemberHandler(ErrorStatus.AUTH_IS_NULL);
+    }
+
+    private Member getMemberByAccessToken(String accessToken) {
+        if (!tokenService.validateToken(accessToken)) {
+            throw new MemberHandler(ErrorStatus.AUTH_INVALID_TOKEN);
+        }
+        String username = tokenService.getAuthentication(accessToken).getName();
+        return memberQueryService.getMemberByUsername(username);
+    }
+}

src/main/java/com/example/waggle/domain/chat/config/WebSocketConfig.java

@@ -0,0 +1,43 @@
+package com.example.waggle.domain.chat.config;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.ChannelRegistration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
+
+@RequiredArgsConstructor
+@EnableWebSocketMessageBroker
+@Configuration
+public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
+
+    private final StompHandler stompHandler;
+
+    @Override
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        registry.addEndpoint("/ws/chat")
+                .setAllowedOriginPatterns("*")
+                .withSockJS();
+    }
+
+    @Override
+    public void configureMessageBroker(MessageBrokerRegistry registry) {
+        registry.enableSimpleBroker("/subscribe"); // /subscribe/{chatRoomId}로 주제 구독 가능
+        registry.setApplicationDestinationPrefixes("/publish"); // /publish/message로 메시지 전송 컨트롤러 라우팅 가능
+    }
+
+    @Override
+    public void configureClientInboundChannel(ChannelRegistration registration) {
+        registration.interceptors(stompHandler);
+    }
+
+    @Override
+    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
+        registry.setMessageSizeLimit(160 * 64 * 1024);
+        registry.setSendTimeLimit(100 * 10000);
+        registry.setSendBufferSizeLimit(3 * 512 * 1024);
+    }
+}

src/main/java/com/example/waggle/domain/chat/config/kafka/KafkaConsumerConfig.java

@@ -0,0 +1,53 @@
+//package com.example.waggle.domain.chat.config.kafka;
+//
+//import com.example.waggle.domain.chat.entity.MessageDto;
+//import com.google.common.collect.ImmutableMap;
+//import java.util.Map;
+//import lombok.RequiredArgsConstructor;
+//import org.apache.kafka.clients.consumer.ConsumerConfig;
+//import org.apache.kafka.common.serialization.StringDeserializer;
+//import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Configuration;
+//import org.springframework.kafka.annotation.EnableKafka;
+//import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+//import org.springframework.kafka.core.ConsumerFactory;
+//import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+//import org.springframework.kafka.support.serializer.JsonDeserializer;
+//
+//@RequiredArgsConstructor
+//@EnableKafka
+//@Configuration
+//public class KafkaConsumerConfig {
+//
+//    private final KafkaProperties kafkaProperties;
+//
+//
+//    // KafkaListener 컨테이너 팩토리를 생성하는 Bean 메서드
+//    @Bean
+//    ConcurrentKafkaListenerContainerFactory<String, MessageDto> kafkaListenerContainerFactory() {
+//        ConcurrentKafkaListenerContainerFactory<String, MessageDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
+//        factory.setConsumerFactory(consumerFactory());
+//        return factory;
+//    }
+//
+//    // Kafka ConsumerFactory를 생성하는 Bean 메서드
+//    @Bean
+//    public ConsumerFactory<String, MessageDto> consumerFactory() {
+//        JsonDeserializer<MessageDto> deserializer = new JsonDeserializer<>();
+//        // 패키지 신뢰 오류로 인해 모든 패키지를 신뢰하도록 작성
+//        deserializer.addTrustedPackages("*");
+//
+//        // Kafka Consumer 구성을 위한 설정값들을 설정 -> 변하지 않는 값이므로 ImmutableMap을 이용하여 설정
+//        Map<String, Object> consumerConfigurations =
+//                ImmutableMap.<String, Object>builder()
+//                        .put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBroker())
+//                        .put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getGroupId())
+//                        .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
+//                        .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer)
+//                        .put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
+//                        .build();
+//
+//        return new DefaultKafkaConsumerFactory<>(consumerConfigurations, new StringDeserializer(), deserializer);
+//    }
+//
+//}

src/main/java/com/example/waggle/domain/chat/config/kafka/KafkaProducerConfig.java

@@ -0,0 +1,46 @@
+//package com.example.waggle.domain.chat.config.kafka;
+//
+//import com.example.waggle.domain.chat.entity.MessageDto;
+//import com.google.common.collect.ImmutableMap;
+//import java.util.Map;
+//import lombok.RequiredArgsConstructor;
+//import org.apache.kafka.clients.producer.ProducerConfig;
+//import org.apache.kafka.common.serialization.StringSerializer;
+//import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Configuration;
+//import org.springframework.kafka.annotation.EnableKafka;
+//import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+//import org.springframework.kafka.core.KafkaTemplate;
+//import org.springframework.kafka.core.ProducerFactory;
+//import org.springframework.kafka.support.serializer.JsonSerializer;
+//
+//@RequiredArgsConstructor
+//@EnableKafka
+//@Configuration
+//public class KafkaProducerConfig {
+//
+//    private final KafkaProperties kafkaProperties;
+//
+//    // Kafka ProducerFactory를 생성하는 Bean 메서드
+//    @Bean
+//    public ProducerFactory<String, MessageDto> producerFactory() {
+//        return new DefaultKafkaProducerFactory<>(producerConfigurations());
+//    }
+//
+//    // Kafka Producer 구성을 위한 설정값들을 포함한 맵을 반환하는 메서드
+//    @Bean
+//    public Map<String, Object> producerConfigurations() {
+//        return ImmutableMap.<String, Object>builder()
+//                .put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBroker())
+//                .put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class)
+//                .put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class)
+//                .build();
+//    }
+//
+//    // KafkaTemplate을 생성하는 Bean 메서드
+//    @Bean
+//    public KafkaTemplate<String, MessageDto> kafkaTemplate() {
+//        return new KafkaTemplate<>(producerFactory());
+//    }
+//
+//}

src/main/java/com/example/waggle/domain/chat/config/kafka/KafkaProperties.java

@@ -0,0 +1,14 @@
+package com.example.waggle.domain.chat.config.kafka;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+@ConfigurationProperties(prefix = "kafka.waggle")
+public class KafkaProperties {
+    private String topic;
+    private String groupId;
+    private String broker;
+}

src/main/java/com/example/waggle/domain/chat/entity/ChatMessage.java

@@ -0,0 +1,39 @@
+package com.example.waggle.domain.chat.entity;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+@ToString
+@Document(collection = "chatMessage")
+@Getter
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class ChatMessage {
+
+    @Id
+    private String id;
+
+    @NotNull
+    private Long chatRoomId;
+
+    @NotNull
+    private ChatMessageType chatMessageType;
+
+    @NotBlank
+    private String content;
+
+    @NotBlank
+    private String senderUserUrl;
+
+    @NotNull
+    private Long sendTime;
+
+}

src/main/java/com/example/waggle/domain/chat/entity/ChatMessageType.java

@@ -0,0 +1,5 @@
+package com.example.waggle.domain.chat.entity;
+
+public enum ChatMessageType {
+    ENTER, EXIT, TALK
+}

src/main/java/com/example/waggle/domain/chat/entity/ChatRoom.java

@@ -1,6 +1,7 @@
 package com.example.waggle.domain.chat.entity;
 
 import com.example.waggle.domain.member.entity.Member;
+import com.example.waggle.global.component.auditing.BaseEntity;
 import jakarta.persistence.CascadeType;
 import jakarta.persistence.Entity;
 import jakarta.persistence.FetchType;
@@ -16,11 +17,13 @@
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 @Getter
+@SuperBuilder
 @NoArgsConstructor(access = AccessLevel.PROTECTED)
 @Entity
-public class ChatRoom {
+public class ChatRoom extends BaseEntity {
 
     @Id
     @GeneratedValue(strategy = GenerationType.IDENTITY)

src/main/java/com/example/waggle/domain/chat/entity/ChatRoomMember.java

@@ -1,22 +1,26 @@
 package com.example.waggle.domain.chat.entity;
 
 import com.example.waggle.domain.member.entity.Member;
+import com.example.waggle.global.component.auditing.BaseEntity;
 import jakarta.persistence.Entity;
 import jakarta.persistence.FetchType;
 import jakarta.persistence.GeneratedValue;
 import jakarta.persistence.GenerationType;
 import jakarta.persistence.Id;
 import jakarta.persistence.JoinColumn;
 import jakarta.persistence.ManyToOne;
+import java.time.LocalDateTime;
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 @Getter
+@SuperBuilder
 @NoArgsConstructor(access = AccessLevel.PROTECTED)
 @Entity
-public class ChatRoomMember {
+public class ChatRoomMember extends BaseEntity {
 
     @Id
     @GeneratedValue(strategy = GenerationType.IDENTITY)
@@ -30,9 +34,16 @@ public class ChatRoomMember {
     @JoinColumn(name = "chat_room_id")
     private ChatRoom chatRoom;
 
+    private LocalDateTime lastAccessTime;
+
     @Builder
-    public ChatRoomMember(Member member, ChatRoom chatRoom) {
+    public ChatRoomMember(Member member, ChatRoom chatRoom, LocalDateTime lastAccessTime) {
         this.member = member;
         this.chatRoom = chatRoom;
+        this.lastAccessTime = lastAccessTime;
+    }
+
+    public void updateLastAccessTime(LocalDateTime time) {
+        this.lastAccessTime = time;
     }
 }

src/main/java/com/example/waggle/domain/chat/repository/ChatMessageRepository.java

@@ -0,0 +1,17 @@
+package com.example.waggle.domain.chat.repository;
+
+import com.example.waggle.domain.chat.entity.ChatMessage;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.data.mongodb.repository.Query;
+
+public interface ChatMessageRepository extends MongoRepository<ChatMessage, String> {
+
+    @Query(value = "{'chatRoomId': ?0, 'sendTime': {$gt: ?1}}", count = true)
+    long countByChatRoomIdAndSendTimeAfter(Long chatRoomId, long lastAccessTime);
+
+    @Query(value = "{'chatRoomId': ?0}", sort = "{'sendTime': -1}")
+    Page<ChatMessage> findByChatRoomIdSortedBySendTimeDesc(Long chatRoomId, Pageable pageable);
+
+}