[Feat/OPS-344] message queue 구현

.github/workflows/test-server-ci.yml

@@ -23,6 +23,20 @@ jobs:
     env:
       SENTRY_AUTH_TOKEN: ${{ secrets.SENTRY_AUTH_TOKEN }}
 
+
+    # CI 작업이 실행되는 동안 RabbitMQ 서비스 컨테이너를 함께 실행
+    services:
+      rabbitmq:
+        image: rabbitmq:3-management
+        ports:
+          - 5672:5672
+        # RabbitMQ가 완전히 준비될 때까지 기다리는 상태 확인 옵션
+        options: >-
+          --health-cmd "rabbitmq-diagnostics check_running"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     steps:
       # 1. 소스 코드 체크아웃
       - name: Checkout source code
@@ -74,6 +88,12 @@ jobs:
 
       # 7. Gradle 테스트 실행
       - name: Test with Gradle
+        # 테스트 단계에서 RabbitMQ 연결을 위한 환경 변수 설정
+        env:
+          SPRING_RABBITMQ_HOST: localhost
+          SPRING_RABBITMQ_PORT: 5672
+          SPRING_RABBITMQ_USERNAME: guest
+          SPRING_RABBITMQ_PASSWORD: guest
         run: ./gradlew test
 
       # 8. 테스트 결과 요약 출력

build.gradle

@@ -116,8 +116,12 @@ dependencies {
     // Redis (Spring starter)
     implementation 'org.springframework.boot:spring-boot-starter-data-redis'
 
-    // RabbitMQ
+    // RabbitMQ (Spring starter)
     implementation 'org.springframework.boot:spring-boot-starter-amqp'
+    testImplementation 'org.springframework.amqp:spring-rabbit-test'
+
+    // Awaitility (비동기 테스트 지원)
+    testImplementation 'org.awaitility:awaitility:4.2.0'
 }
 
 dependencyManagement {

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/controller/ApiV1DashboardController.java

@@ -33,18 +33,18 @@ public class ApiV1DashboardController {
      */
     @PutMapping("/{dashboardId}/graph")
     @Operation(summary = "React-flow 데이터 저장(갱신)")
-    public ResponseEntity<RsData<Void>> updateGraph(
+    public ResponseEntity<RsData<Void>> queueGraphUpdate(
             @PathVariable Integer dashboardId,
             @RequestBody String requestBody,
             @RequestHeader("Liveblocks-Signature") String signature
     ) {
-        dashboardService.verifyAndUpdateGraph(dashboardId, requestBody, signature);
+        dashboardService.queueGraphUpdate(dashboardId, requestBody, signature);
 
         return ResponseEntity
-                .status(HttpStatus.OK)
+                .status(HttpStatus.ACCEPTED)
                 .body(new RsData<>(
-                        "200",
-                        "React-flow 데이터를 저장했습니다.",
+                        "202",
+                        "데이터 업데이트 요청이 성공적으로 접수되었습니다.",
                         null
                 ));
     }

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/dto/GraphUpdateMessage.java

@@ -0,0 +1,7 @@
+package org.tuna.zoopzoop.backend.domain.dashboard.dto;
+
+public record GraphUpdateMessage(
+        Integer dashboardId,
+        String requestBody
+){
+}

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/entity/Graph.java

@@ -3,6 +3,7 @@
 import jakarta.persistence.CascadeType;
 import jakarta.persistence.Entity;
 import jakarta.persistence.OneToMany;
+import jakarta.persistence.Version;
 import lombok.Getter;
 import lombok.Setter;
 import org.tuna.zoopzoop.backend.global.jpa.entity.BaseEntity;
@@ -14,6 +15,10 @@
 @Setter
 @Entity
 public class Graph extends BaseEntity {
+
+    @Version
+    private Long version;
+
     @OneToMany(mappedBy = "graph", cascade = CascadeType.ALL, orphanRemoval = true)
     private List<Node> nodes = new ArrayList<>();
 

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/extraComponent/GraphUpdateConsumer.java

@@ -0,0 +1,39 @@
+package org.tuna.zoopzoop.backend.domain.dashboard.extraComponent;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.orm.ObjectOptimisticLockingFailureException;
+import org.springframework.stereotype.Component;
+import org.tuna.zoopzoop.backend.domain.dashboard.dto.BodyForReactFlow;
+import org.tuna.zoopzoop.backend.domain.dashboard.dto.GraphUpdateMessage;
+import org.tuna.zoopzoop.backend.domain.dashboard.service.DashboardService;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class GraphUpdateConsumer {
+    private final DashboardService dashboardService;
+    private final ObjectMapper objectMapper;
+
+    @RabbitListener(queues = "graph.update.queue")
+    public void handleGraphUpdate(GraphUpdateMessage message) {
+        log.info("Received graph update message for dashboardId: {}", message.dashboardId());
+        try {
+            BodyForReactFlow dto = objectMapper.readValue(message.requestBody(), BodyForReactFlow.class);
+            dashboardService.updateGraph(message.dashboardId(), dto);
+            log.info("Successfully updated graph for dashboardId: {}", message.dashboardId());
+        } catch (ObjectOptimisticLockingFailureException e) {
+            // Optimistic Lock 충돌 발생!
+            // 내가 처리하려던 메시지는 이미 구버전 데이터에 대한 요청이었음.
+            // 따라서 이 메시지는 무시하고 정상 처리된 것으로 간주.
+            log.warn("Stale update attempt for dashboardId: {}. A newer version already exists. Discarding message.", message.dashboardId());
+            // 예외를 다시 던지지 않으므로, 메시지는 큐에서 정상적으로 제거(ACK)됩니다.
+        } catch (Exception e) {
+            // 실제 운영에서는 메시지를 재시도하거나, 실패 큐(Dead Letter Queue)로 보내는 등의
+            // 정교한 에러 처리 로직이 필요합니다.
+            log.error("Failed to process graph update for dashboardId: {}", message.dashboardId(), e);
+        }
+    }
+}

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/service/DashboardService.java

@@ -3,9 +3,13 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.persistence.NoResultException;
 import lombok.RequiredArgsConstructor;
+import org.apache.commons.codec.binary.Hex;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.tuna.zoopzoop.backend.domain.dashboard.dto.BodyForReactFlow;
+import org.tuna.zoopzoop.backend.domain.dashboard.dto.GraphUpdateMessage;
 import org.tuna.zoopzoop.backend.domain.dashboard.entity.Dashboard;
 import org.tuna.zoopzoop.backend.domain.dashboard.entity.Edge;
 import org.tuna.zoopzoop.backend.domain.dashboard.entity.Graph;
@@ -25,6 +29,7 @@ public class DashboardService {
     private final MembershipService membershipService;
     private final ObjectMapper objectMapper;
     private final SignatureService signatureService;
+    private final RabbitTemplate rabbitTemplate;
 
 
 
@@ -103,6 +108,33 @@ public void verifyAccessPermission(Member member, Integer dashboardId) throws Ac
         }
     }
 
+    // =========================== message 관리 메서드  ===========================
+
+    /**
+     * Graph 업데이트 요청을 RabbitMQ 큐에 비동기적으로 발행하는 메서드
+     * @param dashboardId 대시보드 ID
+     * @param requestBody 요청 바디
+     * @param signatureHeader 서명 헤더
+     */
+    public void queueGraphUpdate(Integer dashboardId, String requestBody, String signatureHeader){
+        // 서명 검증은 동기적으로 즉시 처리
+        if (!signatureService.isValidSignature(requestBody, signatureHeader)) {
+            throw new SecurityException("Invalid webhook signature.");
+        }
+
+        // 대시보드 존재 여부 확인
+        if (!dashboardRepository.existsById(dashboardId)) {
+            throw new NoResultException(dashboardId + " ID를 가진 대시보드를 찾을 수 없습니다.");
+        }
+
+        // 큐에 보낼 메시지 생성
+        GraphUpdateMessage message = new GraphUpdateMessage(dashboardId, requestBody);
+
+        // RabbitMQ에 메시지 발행
+        rabbitTemplate.convertAndSend("zoopzoop.exchange", "graph.update.rk", message);
+    }
+
+
 
 
 

src/main/java/org/tuna/zoopzoop/backend/domain/dashboard/service/SignatureService.java

@@ -26,6 +26,11 @@ public class SignatureService {
      * @return 서명이 유효하면 true, 그렇지 않으면 false
      */
     public boolean isValidSignature(String requestBody, String signatureHeader) {
+        // [임시 코드] 로컬 테스트를 위해 무조건 true 반환
+//        if ("true".equals(System.getProperty("local.test.skip.signature"))) {
+//            return true;
+//        }
+
         try {
             // 1. 헤더 파싱
             String[] parts = signatureHeader.split(",");

src/main/java/org/tuna/zoopzoop/backend/global/config/mq/RabbitMQConfig.java

@@ -0,0 +1,69 @@
+package org.tuna.zoopzoop.backend.global.config.mq;
+
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMQConfig {
+    private static final String EXCHANGE_NAME = "zoopzoop.exchange";
+    private static final String QUEUE_NAME = "graph.update.queue";
+    private static final String ROUTING_KEY = "graph.update.#";
+
+    private static final String DLQ_EXCHANGE_NAME = EXCHANGE_NAME + ".dlx";
+    private static final String DLQ_QUEUE_NAME = QUEUE_NAME + ".dlq";
+    private static final String DLQ_ROUTING_KEY = "graph.update.dlq";
+
+    @Bean
+    public TopicExchange exchange() {
+        return new TopicExchange(EXCHANGE_NAME);
+    }
+
+    @Bean
+    public Queue queue() {
+        return QueueBuilder.durable(QUEUE_NAME)
+                .withArgument("x-dead-letter-exchange", DLQ_EXCHANGE_NAME) // 실패 시 메시지를 보낼 Exchange
+                .withArgument("x-dead-letter-routing-key", DLQ_ROUTING_KEY) // 실패 시 사용할 라우팅 키
+                .build();
+    }
+
+    @Bean
+    public Binding binding(Queue queue, TopicExchange exchange) {
+        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
+    }
+
+    // ================= DLQ 인프라 구성 추가 ================= //
+
+    @Bean
+    public TopicExchange dlqExchange() {
+        return new TopicExchange(DLQ_EXCHANGE_NAME);
+    }
+
+    @Bean
+    public Queue dlqQueue() {
+        return new Queue(DLQ_QUEUE_NAME);
+    }
+
+    @Bean
+    public Binding dlqBinding(Queue dlqQueue, TopicExchange dlqExchange) {
+        return BindingBuilder.bind(dlqQueue).to(dlqExchange).with(DLQ_ROUTING_KEY);
+    }
+
+    // ================= DLQ 인프라 구성 추가 ================= //
+    @Bean
+    public MessageConverter messageConverter() {
+        // 메시지를 JSON 형식으로 직렬화/역직렬화하는 컨버터
+        return new Jackson2JsonMessageConverter();
+    }
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
+        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+        rabbitTemplate.setMessageConverter(messageConverter);
+        return rabbitTemplate;
+    }
+}

src/main/resources/application.yml

@@ -40,6 +40,23 @@ spring:
     port: 5672
     username: ${SPRING_RABBITMQ_USERNAME:guest}
     password: ${SPRING_RABBITMQ_PASSWORD:guest}
+    listener:
+      simple:
+        retry:
+          enabled: true
+          initial-interval: 2000
+          max-attempts: 3
+  data: #RedisTemplate 등을 사용하기 위한 직접 연결용
+    redis:
+      host: localhost
+      port: 6379
+      timeout: 6000
+  cache: #Spring Cache를 사용하기 위한 Redis
+    type: redis
+    redis:
+      time-to-live: 300000
+      cache-null-values: false
+      key-prefix:
 
 springdoc:
   default-produces-media-type: application/json;charset=UTF-8
@@ -49,6 +66,7 @@ logging:
     org.hibernate.orm.jdbc.extract: TRACE
     org.springframework.transaction.interceptor: TRACE
     com.back: DEBUG
+    org.springframework.retry: DEBUG
 
 server:
   port: 8080