feat/116 :: Order 주문 서비스 구현

services/order/Dockerfile

@@ -1,6 +1,7 @@
 FROM mcr.microsoft.com/openjdk/jdk:21-ubuntu
-ARG JAR_FILE=build/libs/*.jar
-WORKDIR /app
+ARG JAR_FILE=services/order/build/libs/*.jar
+RUN groupadd -r appuser && useradd -r -g appuser appuserWORKDIR /app
 COPY ${JAR_FILE} app.jar
-EXPOSE 8080
-ENTRYPOINT ["java", "-jar", "app.jar"]
+RUN chown -R appuser:appuser /app
+USER appuserEXPOSE 8080
+ENTRYPOINT ["java", "-jar", "app.jar"]

services/order/build.gradle

@@ -1,5 +1,47 @@
+plugins {
+    id 'com.google.protobuf' version '0.9.6'
+}
+
+ext {
+    springGrpcVersion = "1.0.2"
+    springCloudVersion = "2025.1.0"
+}
 version = '0.0.1'
 description = 'order'
 
 dependencies {
-}
+    implementation 'org.springframework.boot:spring-boot-starter-web'
+    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
+    implementation 'org.springframework:spring-aspects'
+    implementation 'org.aspectj:aspectjweaver'
+    implementation 'org.springframework.boot:spring-boot-starter-kafka'
+    implementation 'org.springframework.boot:spring-boot-starter-security'
+    implementation 'org.springframework.grpc:spring-grpc-client-spring-boot-starter'
+    implementation 'org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j'
+    runtimeOnly 'com.mysql:mysql-connector-j'
+
+    testImplementation 'org.springframework.kafka:spring-kafka-test'
+}
+
+dependencyManagement {
+    imports {
+        mavenBom "org.springframework.grpc:spring-grpc-dependencies:$springGrpcVersion"
+        mavenBom "org.springframework.cloud:spring-cloud-dependencies:$springCloudVersion"
+    }
+}
+
+protobuf {
+    protoc {
+        artifact = "com.google.protobuf:protoc:4.34.0"
+    }
+    plugins {
+        grpc {
+            artifact = 'io.grpc:protoc-gen-grpc-java:1.79.0'
+        }
+    }
+    generateProtoTasks {
+        all()*.plugins {
+            grpc {}
+        }
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/DeliveryEventKafkaListener.java

@@ -0,0 +1,40 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.DeliveryCompletedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.DeliveryStartedEvent;
+import kr.magicbox.order.application.port.in.HandleDeliveryCompletedUseCase;
+import kr.magicbox.order.application.port.in.HandleDeliveryStartedUseCase;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class DeliveryEventKafkaListener {
+
+    private final HandleDeliveryStartedUseCase handleDeliveryStartedUseCase;
+    private final HandleDeliveryCompletedUseCase handleDeliveryCompletedUseCase;
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.delivery-started", groupId = "order-service")
+    public void handleDeliveryStarted(ConsumerRecord<String, DeliveryStartedEvent> consumerRecord) {
+        log.info("[Inbox] delivery.started 이벤트 수신. eventId={}", consumerRecord.key());
+        DeliveryStartedEvent event = consumerRecord.value();
+        handleDeliveryStartedUseCase.handleDeliveryStarted(event.orderId(), event.orderLineId());
+    }
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.delivery-completed", groupId = "order-service")
+    public void handleDeliveryCompleted(ConsumerRecord<String, DeliveryCompletedEvent> consumerRecord) {
+        log.info("[Inbox] delivery.completed 이벤트 수신. eventId={}", consumerRecord.key());
+        DeliveryCompletedEvent event = consumerRecord.value();
+        handleDeliveryCompletedUseCase.handleDeliveryCompleted(event.orderId(), event.orderLineId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/KafkaConfiguration.java

@@ -0,0 +1,22 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.properties.InboxProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafkaRetryTopic;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+@EnableKafkaRetryTopic
+@Configuration
+@EnableConfigurationProperties(InboxProperties.class)
+public class KafkaConfiguration {
+
+    @Bean
+    public ThreadPoolTaskScheduler taskScheduler() {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        scheduler.setPoolSize(1);
+        scheduler.setThreadNamePrefix("kafka-retry-");
+        return scheduler;
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/OrderStateKafkaListener.java

@@ -0,0 +1,27 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.OrderPrepareEventDto;
+import kr.magicbox.order.application.port.in.HandleOrderPrepareUseCase;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OrderStateKafkaListener {
+
+    private final HandleOrderPrepareUseCase handleOrderPrepareUseCase;
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.order-prepare", groupId = "order-service")
+    public void handleOrderPrepare(ConsumerRecord<String, OrderPrepareEventDto> consumerRecord) {
+        log.info("[Inbox] order.prepare 이벤트 수신. eventId={}", consumerRecord.key());
+        handleOrderPrepareUseCase.handleOrderPrepare(consumerRecord.value().orderId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/PaymentEventKafkaListener.java

@@ -0,0 +1,60 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentCancelFailedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentCancelSucceededEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentFailedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentSucceededEvent;
+import kr.magicbox.order.application.port.in.HandlePaymentCancelFailedUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentCancelSucceededUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentFailedUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentSucceededUseCase;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class PaymentEventKafkaListener {
+
+    private final HandlePaymentSucceededUseCase handlePaymentSucceededUseCase;
+    private final HandlePaymentFailedUseCase handlePaymentFailedUseCase;
+    private final HandlePaymentCancelSucceededUseCase handlePaymentCancelSucceededUseCase;
+    private final HandlePaymentCancelFailedUseCase handlePaymentCancelFailedUseCase;
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.payment-succeeded", groupId = "order-service")
+    public void handlePaymentSucceeded(ConsumerRecord<String, PaymentSucceededEvent> consumerRecord) {
+        log.info("[Inbox] payment.succeeded 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentSucceededUseCase.handlePaymentSucceeded(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.payment-failed", groupId = "order-service")
+    public void handlePaymentFailed(ConsumerRecord<String, PaymentFailedEvent> consumerRecord) {
+        log.info("[Inbox] payment.failed 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentFailedUseCase.handlePaymentFailed(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.payment-cancel-succeeded", groupId = "order-service")
+    public void handlePaymentCancelSucceeded(ConsumerRecord<String, PaymentCancelSucceededEvent> consumerRecord) {
+        log.info("[Inbox] payment.cancel.succeeded 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentCancelSucceededUseCase.handlePaymentCancelSucceeded(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.payment-cancel-failed", groupId = "order-service")
+    public void handlePaymentCancelFailed(ConsumerRecord<String, PaymentCancelFailedEvent> consumerRecord) {
+        log.info("[Inbox] payment.cancel.failed 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentCancelFailedUseCase.handlePaymentCancelFailed(consumerRecord.value().orderId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/StockEventKafkaListener.java

@@ -0,0 +1,38 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.StockReserveFailedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.StockReserveSucceededEvent;
+import kr.magicbox.order.application.port.in.HandleStockReserveFailedUseCase;
+import kr.magicbox.order.application.port.in.HandleStockReserveSucceededUseCase;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class StockEventKafkaListener {
+
+    private final HandleStockReserveSucceededUseCase handleStockReserveSucceededUseCase;
+    private final HandleStockReserveFailedUseCase handleStockReserveFailedUseCase;
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.stock-reserve-succeeded", groupId = "order-service")
+    public void handleStockReserveSucceeded(ConsumerRecord<String, StockReserveSucceededEvent> consumerRecord) {
+        log.info("[Inbox] stock.reserve.succeeded 이벤트 수신. eventId={}", consumerRecord.key());
+        handleStockReserveSucceededUseCase.handleStockReserveSucceeded(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic
+    @KafkaListener(topics = "outbox.event.stock-reserve-failed", groupId = "order-service")
+    public void handleStockReserveFailed(ConsumerRecord<String, StockReserveFailedEvent> consumerRecord) {
+        log.info("[Inbox] stock.reserve.failed 이벤트 수신. eventId={}", consumerRecord.key());
+        handleStockReserveFailedUseCase.handleStockReserveFailed(consumerRecord.value().orderId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/annotation/Idempotent.java

@@ -0,0 +1,11 @@
+package kr.magicbox.order.adapter.in.kafka.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Idempotent {
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/aop/IdempotentAspect.java

@@ -0,0 +1,97 @@
+package kr.magicbox.order.adapter.in.kafka.aop;
+
+import kr.magicbox.order.adapter.in.kafka.properties.InboxProperties;
+import kr.magicbox.order.adapter.out.persistence.entity.OrderInboxEntity;
+import kr.magicbox.order.adapter.out.persistence.entity.OrderInboxStatus;
+import kr.magicbox.order.adapter.out.persistence.repository.OrderInboxJpaRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.lang.reflect.RecordComponent;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+
+@Slf4j
+@Aspect
+@Component
+@RequiredArgsConstructor
+public class IdempotentAspect {
+
+    private final OrderInboxJpaRepository orderInboxJpaRepository;
+    private final TransactionTemplate transactionTemplate;
+    private final InboxProperties inboxProperties;
+
+    @Around("@annotation(kr.magicbox.order.adapter.in.kafka.annotation.Idempotent)")
+    public Object around(ProceedingJoinPoint pjp) {
+        ConsumerRecord<String, ?> consumerRecord = extractRecord(pjp);
+        Long eventId = Long.parseLong(consumerRecord.key());
+        Instant occurredAt = extractOccurredAt(consumerRecord.value());
+
+        if (isTooOld(occurredAt)) {
+            log.warn("[Inbox] 만료된 메시지 폐기. eventId={}, occurredAt={}", eventId, occurredAt);
+            return null;
+        }
+
+        return transactionTemplate.execute(status -> {
+            if (orderInboxJpaRepository.existsByEventId(eventId)) {
+                log.warn("[Inbox] 중복 메시지 폐기. eventId={}", eventId);
+                return null;
+            }
+            OrderInboxEntity inbox = orderInboxJpaRepository.save(OrderInboxEntity.builder()
+                    .eventId(eventId)
+                    .topic(consumerRecord.topic())
+                    .partition(consumerRecord.partition())
+                    .offset(consumerRecord.offset())
+                    .status(OrderInboxStatus.PENDING)
+                    .occurredAt(occurredAt)
+                    .build());
+            try {
+                pjp.proceed();
+            } catch (Throwable e) {
+                status.setRollbackOnly();
+                throw new RuntimeException(e);
+            }
+            inbox.markProcessed();
+            return null;
+        });
+    }
+
+    private boolean isTooOld(Instant occurredAt) {
+        return occurredAt.isBefore(Instant.now().minus(inboxProperties.getMaxEventAgeMinutes(), ChronoUnit.MINUTES));
+    }
+
+    private Instant extractOccurredAt(Object payload) {
+        if (payload == null) {
+            return Instant.now();
+        }
+        try {
+            for (RecordComponent component : payload.getClass().getRecordComponents()) {
+                if (component.getName().equals("occurredAt")) {
+                    Object value = component.getAccessor().invoke(payload);
+                    if (value instanceof Instant instant) {
+                        return instant;
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.warn("[Inbox] occurredAt 추출 실패, 현재 시각으로 대체. payload={}", payload.getClass().getSimpleName());
+        }
+        return Instant.now();
+    }
+
+    @SuppressWarnings("unchecked")
+    private ConsumerRecord<String, ?> extractRecord(ProceedingJoinPoint pjp) {
+        return Arrays.stream(pjp.getArgs())
+                .filter(ConsumerRecord.class::isInstance)
+                .map(arg -> (ConsumerRecord<String, ?>) arg)
+                .findFirst()
+                .orElseThrow(() -> new IllegalStateException("@Idempotent 메서드에 ConsumerRecord 파라미터가 없습니다."));
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/event/DeliveryCompletedEvent.java

@@ -0,0 +1,22 @@
+package kr.magicbox.order.adapter.in.kafka.event;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.Instant;
+import java.util.List;
+
+public record DeliveryCompletedEvent(
+        @JsonProperty("order_id") Long orderId,
+        @JsonProperty("order_line_id") Long orderLineId,
+        @JsonProperty("customer_id") Long customerId,
+        @JsonProperty("delivery_id") Long deliveryId,
+        @JsonProperty("tracking_number") String trackingNumber,
+        @JsonProperty("delivered_at") Instant deliveredAt,
+        @JsonProperty("items") List<ItemPayload> items,
+        @JsonProperty("occurred_at") Instant occurredAt
+) {
+    public record ItemPayload(
+            @JsonProperty("product_id") Long productId,
+            @JsonProperty("quantity") int quantity
+    ) {}
+}