[Volume7] 이벤트 기반 아키텍처 및 Kafka 파이프라인 구현 - 박시온

.gitignore

@@ -38,3 +38,6 @@ out/
 
 ### Kotlin ###
 .kotlin
+
+### QueryDSL ###
+**/generated/

apps/commerce-api/build.gradle.kts

@@ -2,6 +2,8 @@ dependencies {
     // add-ons
     implementation(project(":modules:jpa"))
     implementation(project(":modules:redis"))
+    implementation(project(":modules:kafka"))
+    implementation(project(":modules:event-contract"))
     implementation(project(":supports:jackson"))
     implementation(project(":supports:logging"))
     implementation(project(":supports:monitoring"))

apps/commerce-api/src/main/java/com/loopers/application/coupon/CouponIssueCountManager.java

@@ -0,0 +1,6 @@
+package com.loopers.application.coupon;
+
+public interface CouponIssueCountManager {
+    long increment(Long couponId);
+    void decrement(Long couponId);
+}

apps/commerce-api/src/main/java/com/loopers/application/coupon/CouponIssueRequestFacade.java

@@ -0,0 +1,62 @@
+package com.loopers.application.coupon;
+
+import com.loopers.application.outbox.OutboxEventPublisher;
+import com.loopers.domain.coupon.CouponIssueRequest;
+import com.loopers.domain.coupon.CouponIssueRequestRepository;
+import com.loopers.domain.coupon.CouponPromotion;
+import com.loopers.domain.coupon.CouponPromotionRepository;
+import com.loopers.event.EventType;
+import com.loopers.event.payload.CouponIssueRequestedEventPayload;
+import com.loopers.support.error.CoreException;
+import com.loopers.support.error.ErrorType;
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@RequiredArgsConstructor
+@Service
+public class CouponIssueRequestFacade {
+
+    private final CouponPromotionRepository couponPromotionRepository;
+    private final CouponIssueRequestRepository couponIssueRequestRepository;
+    private final CouponIssueCountManager couponIssueCountManager;
+    private final OutboxEventPublisher outboxEventPublisher;
+
+    @Transactional
+    public CouponIssueRequestInfo issueRequest(Long userId, Long couponId) {
+        CouponPromotion promotion = couponPromotionRepository.findByCouponId(couponId)
+                .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "선착순 프로모션 대상 쿠폰이 아닙니다."));
+
+        promotion.validateIssuable();
+
+        long issuedCount = couponIssueCountManager.increment(couponId);
+        if (issuedCount > promotion.getMaxQuantity()) {
+            couponIssueCountManager.decrement(couponId);
+            throw new CoreException(ErrorType.BAD_REQUEST, "선착순 쿠폰이 모두 소진되었습니다.");
+        }
+
+        CouponIssueRequest request;
+        try {
+            request = couponIssueRequestRepository.save(CouponIssueRequest.create(couponId, userId));
+        } catch (DataIntegrityViolationException e) {
+            couponIssueCountManager.decrement(couponId);
+            throw new CoreException(ErrorType.CONFLICT, "이미 발급 요청한 쿠폰입니다.");
+        }
+
+        outboxEventPublisher.publish(
+                EventType.COUPON_ISSUE_REQUESTED,
+                CouponIssueRequestedEventPayload.of(request.getId(), couponId, userId),
+                couponId
+        );
+
+        return CouponIssueRequestInfo.from(request);
+    }
+
+    @Transactional(readOnly = true)
+    public CouponIssueRequestInfo getIssueRequest(Long userId, Long requestId) {
+        CouponIssueRequest request = couponIssueRequestRepository.findByIdAndUserId(requestId, userId)
+                .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "발급 요청을 찾을 수 없습니다."));
+        return CouponIssueRequestInfo.from(request);
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/coupon/CouponIssueRequestInfo.java

@@ -0,0 +1,21 @@
+package com.loopers.application.coupon;
+
+import com.loopers.domain.coupon.CouponIssueRequest;
+
+public record CouponIssueRequestInfo(
+    Long requestId,
+    Long couponId,
+    Long userId,
+    String status,
+    String reason
+) {
+    public static CouponIssueRequestInfo from(CouponIssueRequest request) {
+        return new CouponIssueRequestInfo(
+            request.getId(),
+            request.getCouponId(),
+            request.getUserId(),
+            request.getStatus().name(),
+            request.getReason()
+        );
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/event/UserBehaviorLoggingListener.java

@@ -0,0 +1,22 @@
+package com.loopers.application.event;
+
+import com.loopers.domain.outbox.Outbox;
+import com.loopers.domain.outbox.OutboxEvent;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+@Slf4j
+@Component
+public class UserBehaviorLoggingListener {
+
+    @Async("outboxPublishExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleOutboxEvent(OutboxEvent event) {
+        Outbox outbox = event.getOutbox();
+        log.info("[UserBehavior] type={}, partitionKey={}, payload={}",
+                outbox.getEventType(), outbox.getPartitionKey(), outbox.getPayload());
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/like/LikeFacade.java

@@ -1,7 +1,11 @@
 package com.loopers.application.like;
 
+import com.loopers.application.outbox.OutboxEventPublisher;
 import com.loopers.application.product.ProductInfo;
 import com.loopers.application.product.ProductService;
+import com.loopers.event.EventType;
+import com.loopers.event.payload.ProductLikedEventPayload;
+import com.loopers.event.payload.ProductUnlikedEventPayload;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -15,18 +19,29 @@
 public class LikeFacade {
     private final LikeService likeService;
     private final ProductService productService;
+    private final OutboxEventPublisher outboxEventPublisher;
 
     @Transactional
     public LikeInfo register(Long userId, Long productId) {
+        productService.getActiveProduct(productId);
         LikeInfo like = likeService.register(userId, productId);
-        productService.increaseLikeCount(productId);
+        outboxEventPublisher.publish(
+                EventType.PRODUCT_LIKED,
+                ProductLikedEventPayload.of(productId, userId),
+                productId
+        );
+
         return like;
     }
 
     @Transactional
     public void cancel(Long userId, Long productId) {
         if (likeService.cancel(userId, productId)) {
-            productService.decreaseLikeCount(productId);
+            outboxEventPublisher.publish(
+                    EventType.PRODUCT_UNLIKED,
+                    ProductUnlikedEventPayload.of(productId, userId),
+                    productId
+            );
         }
     }
 

apps/commerce-api/src/main/java/com/loopers/application/outbox/MessageRelay.java

@@ -0,0 +1,64 @@
+package com.loopers.application.outbox;
+
+import com.loopers.domain.outbox.Outbox;
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.infrastructure.outbox.OutboxRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.domain.Pageable;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class MessageRelay {
+
+    private final OutboxRepository outboxRepository;
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void createOutbox(OutboxEvent event) {
+        outboxRepository.save(event.getOutbox());
+    }
+
+    @Async("outboxPublishExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void publishEvent(OutboxEvent event) {
+        publishToKafka(event.getOutbox());
+    }
+
+    private void publishToKafka(Outbox outbox) {
+        try {
+            kafkaTemplate.send(
+                    outbox.getEventType().getTopic(),
+                    String.valueOf(outbox.getPartitionKey()),
+                    outbox.getPayload()
+            ).get(1, TimeUnit.SECONDS);
+            outboxRepository.delete(outbox);
+        } catch (Exception e) {
+            log.error("[MessageRelay] Kafka 발행 실패, outboxId={}", outbox.getOutboxId(), e);
+        }
+    }
+
+    @Scheduled(fixedDelay = 10, timeUnit = TimeUnit.SECONDS)
+    public void publishPendingEvents() {
+        List<Outbox> pending = outboxRepository
+                                .findAllByCreatedAtLessThanEqualOrderByCreatedAtAsc(
+                                        LocalDateTime.now().minusSeconds(10),
+                                        Pageable.ofSize(100)
+                                );
+
+        for (Outbox outbox : pending) {
+            publishToKafka(outbox);
+        }
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventPublisher.java

@@ -0,0 +1,27 @@
+package com.loopers.application.outbox;
+
+import com.loopers.domain.outbox.Outbox;
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.event.*;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class OutboxEventPublisher {
+
+    private final Snowflake outboxIdSnowflake = new Snowflake();
+    private final Snowflake eventIdSnowflake = new Snowflake();
+    private final ApplicationEventPublisher applicationEventPublisher;
+
+    public void publish(EventType type, EventPayload payload, Long partitionKey) {
+        Outbox outbox = Outbox.create(
+                outboxIdSnowflake.nextId(),
+                type,
+                Event.of(eventIdSnowflake.nextId(), type, payload).toJson(),
+                partitionKey
+        );
+        applicationEventPublisher.publishEvent(OutboxEvent.of(outbox));
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/payment/PaymentFacade.java

@@ -2,18 +2,24 @@
 
 import com.loopers.application.order.OrderCompensationService;
 import com.loopers.application.order.OrderInfo;
+import com.loopers.application.order.OrderItemInfo;
 import com.loopers.application.order.OrderService;
+import com.loopers.application.outbox.OutboxEventPublisher;
 import com.loopers.domain.order.Order;
 import com.loopers.domain.payment.Payment;
 import com.loopers.domain.payment.PaymentRepository;
 import com.loopers.domain.payment.PaymentStatus;
+import com.loopers.event.EventType;
+import com.loopers.event.payload.PaymentCompletedEventPayload;
 import com.loopers.infrastructure.client.PgDeclinedException;
 import com.loopers.infrastructure.client.PgPaymentDto;
 import com.loopers.infrastructure.client.PgPaymentException;
 import com.loopers.infrastructure.client.PgPaymentGateway;
 import com.loopers.infrastructure.client.PgTransactionStatus;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
+
+import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -27,19 +33,22 @@ public class PaymentFacade {
     private final PaymentRepository paymentRepository;
     private final OrderService orderService;
     private final OrderCompensationService orderCompensationService;
+    private final OutboxEventPublisher outboxEventPublisher;
     private final PgPaymentGateway pgPaymentGateway;
     private final String callbackUrl;
 
     public PaymentFacade(
             PaymentRepository paymentRepository,
             OrderService orderService,
             OrderCompensationService orderCompensationService,
+            OutboxEventPublisher outboxEventPublisher,
             PgPaymentGateway pgPaymentGateway,
             @Value("${payment.callback-url}") String callbackUrl
     ) {
         this.paymentRepository = paymentRepository;
         this.orderService = orderService;
         this.orderCompensationService = orderCompensationService;
+        this.outboxEventPublisher = outboxEventPublisher;
         this.pgPaymentGateway = pgPaymentGateway;
         this.callbackUrl = callbackUrl;
     }
@@ -122,8 +131,18 @@ private void applyPgResult(Payment payment, PgTransactionStatus status, String r
                 int affected = paymentRepository.completeIfPending(payment.getId());
                 if (affected > 0) {
                     orderService.markOrderPaid(payment.getOrderId());
+                    List<Long> productIds = orderService.getOrderItems(payment.getOrderId()).stream()
+                                                        .map(OrderItemInfo::productId)
+                                                        .toList();
+
+                    outboxEventPublisher.publish(
+                            EventType.PAYMENT_COMPLETED,
+                            PaymentCompletedEventPayload.of(payment.getId(), payment.getOrderId(), null, productIds),
+                            payment.getOrderId()
+                    );
                 }
             }
+
             case FAILED -> {
                 int affected = paymentRepository.failIfPending(payment.getId(), reason);
                 if (affected > 0) {

apps/commerce-api/src/main/java/com/loopers/application/product/ProductInfo.java

@@ -27,7 +27,7 @@ public static ProductInfo from(Product product) {
                 product.getDescription(),
                 product.getPrice(),
                 product.getStockQuantity(),
-                product.getLikeCount(),
+                0,
                 product.getVisibility(),
                 product.getCreatedAt(),
                 product.getUpdatedAt(),

apps/commerce-api/src/main/java/com/loopers/application/product/ProductService.java

@@ -98,26 +98,6 @@ public void deleteAllByBrandId(Long brandId) {
         products.forEach(Product::delete);
     }
 
-    @Transactional
-    public void increaseLikeCount(Long productId) {
-        Product product = findById(productId);
-        if (!product.isActive()) {
-            throw new CoreException(ErrorType.NOT_FOUND, "[productId = " + productId + "] 를 찾을 수 없습니다.");
-        }
-
-        productRepository.increaseLikeCount(productId);
-    }
-
-    @Transactional
-    public void decreaseLikeCount(Long productId) {
-        Product product = findById(productId);
-        if (!product.isActive()) {
-            throw new CoreException(ErrorType.NOT_FOUND, "[productId = " + productId + "] 를 찾을 수 없습니다.");
-        }
-
-        productRepository.decreaseLikeCount(productId);
-    }
-
     @Transactional
     public void decreaseStock(List<OrderItemCommand> items) {
         List<OrderItemCommand> sorted = items.stream()