[7주차] 이벤트 기반 아키텍처, 카프카 - 최호석

apps/commerce-api/build.gradle.kts

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

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

@@ -1,19 +1,27 @@
 package com.loopers.application.coupon;
 
+import com.loopers.application.event.CouponIssueRequestedEvent;
+import com.loopers.confg.kafka.KafkaTopics;
 import com.loopers.application.user.UserService;
+import com.loopers.infrastructure.outbox.OutboxEventService;
 import com.loopers.domain.coupon.Coupon;
+import com.loopers.domain.coupon.CouponIssueRequest;
+import com.loopers.domain.coupon.CouponIssueRequestRepository;
 import com.loopers.domain.coupon.IssuedCoupon;
 import com.loopers.domain.user.User;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -24,6 +32,9 @@ public class CouponFacade {
     private final CouponService couponService;
     private final IssuedCouponService issuedCouponService;
     private final UserService userService;
+    private final CouponIssueRequestRepository couponIssueRequestRepository;
+    private final ApplicationEventPublisher eventPublisher;
+    private final OutboxEventService outboxEventService;
 
     // Command
 
@@ -51,6 +62,31 @@ public IssuedCouponInfo issueCoupon(Long couponId, Long userId) {
         return IssuedCouponInfo.from(issuedCoupon);
     }
 
+    @Transactional
+    public CouponIssueRequestInfo issueAsync(Long couponId, Long userId) {
+        couponService.validateActiveCoupon(couponId);
+
+        Optional<CouponIssueRequest> existing = couponIssueRequestRepository.findByCouponIdAndUserId(couponId, userId);
+        if (existing.isPresent()) {
+            return CouponIssueRequestInfo.from(existing.get());
+        }
+
+        String eventId = UUID.randomUUID().toString();
+        CouponIssueRequest request = couponIssueRequestRepository.save(
+                CouponIssueRequest.create(eventId, couponId, userId));
+        outboxEventService.saveAndPublish("coupon.issue.requested", "Coupon",
+                String.valueOf(couponId), KafkaTopics.COUPON_ISSUE_REQUESTS,
+                new CouponIssueRequestedEvent(eventId, couponId, userId));
+        return CouponIssueRequestInfo.from(request);
+    }
+
+    @Transactional(readOnly = true)
+    public CouponIssueRequestInfo getIssueStatus(Long couponId, Long userId) {
+        CouponIssueRequest request = couponIssueRequestRepository.findByCouponIdAndUserId(couponId, userId)
+                .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "발급 요청이 존재하지 않습니다"));
+        return CouponIssueRequestInfo.from(request);
+    }
+
     @Transactional
     public CouponInfo updateInfo(Long couponId, CouponCommand.UpdateInfo command) {
         if (command.type() != null) {

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

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

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

@@ -0,0 +1,5 @@
+package com.loopers.application.event;
+
+
+public record CouponIssueRequestedEvent(String eventId, Long couponId, Long userId) {
+}

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

@@ -0,0 +1,61 @@
+package com.loopers.application.event;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.application.product.ProductService;
+import com.loopers.confg.kafka.KafkaTopics;
+import com.loopers.infrastructure.product.ProductCacheManager;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+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
+@RequiredArgsConstructor
+public class LikeCountEventHandler {
+
+    private final ProductService productService;
+    private final ProductCacheManager productCacheManager;
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+    private final ObjectMapper objectMapper;
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handleLiked(ProductLikedEvent event) {
+        try {
+            productService.incrementLikeCount(event.productId());
+            productCacheManager.evictDetail(event.productId());
+            publishToKafka("product.liked", event.productId(), event);
+        } catch (Exception e) {
+            log.error("좋아요 집계 실패: productId={}", event.productId(), e);
+        }
+    }
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handleUnliked(ProductUnlikedEvent event) {
+        try {
+            productService.decrementLikeCountIfPositive(event.productId());
+            productCacheManager.evictDetail(event.productId());
+            publishToKafka("product.unliked", event.productId(), event);
+        } catch (Exception e) {
+            log.error("좋아요 취소 집계 실패: productId={}", event.productId(), e);
+        }
+    }
+
+    /**
+     * 비핵심 지표(좋아요)는 Outbox 없이 직접 Kafka fire-and-forget.
+     * 유실돼도 비즈니스 불변식이 깨지지 않는다. product_metrics 근사치로 충분.
+     */
+    private void publishToKafka(String eventType, Long productId, Object event) {
+        try {
+            String payload = objectMapper.writeValueAsString(event);
+            kafkaTemplate.send(KafkaTopics.CATALOG_EVENTS, String.valueOf(productId), payload);
+        } catch (Exception e) {
+            log.warn("Kafka 직접 발행 실패 (fire-and-forget): eventType={}, productId={}", eventType, productId, e);
+        }
+    }
+}

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

@@ -0,0 +1,5 @@
+package com.loopers.application.event;
+
+
+public record PaymentCanceledEvent(Long paymentId, Long orderId, Long userId) {
+}

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

@@ -0,0 +1,7 @@
+package com.loopers.application.event;
+
+
+import java.math.BigDecimal;
+
+public record PaymentCompletedEvent(Long paymentId, Long orderId, Long userId, BigDecimal amount) {
+}

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

@@ -0,0 +1,5 @@
+package com.loopers.application.event;
+
+
+public record PaymentFailedEvent(Long paymentId, Long orderId, Long userId, String reason) {
+}

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

@@ -0,0 +1,5 @@
+package com.loopers.application.event;
+
+
+public record ProductLikedEvent(Long userId, Long productId) {
+}

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

@@ -0,0 +1,5 @@
+package com.loopers.application.event;
+
+
+public record ProductUnlikedEvent(Long userId, Long productId) {
+}

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

@@ -0,0 +1,4 @@
+package com.loopers.application.event;
+
+public record ProductViewedEvent(Long userId, Long productId) {
+}

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

@@ -0,0 +1,55 @@
+package com.loopers.application.event;
+
+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 UserActivityEventHandler {
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handleProductViewed(ProductViewedEvent event) {
+        try {
+            log.info("상품 조회: userId={}, productId={}", event.userId(), event.productId());
+        } catch (Exception e) {
+            log.error("상품 조회 로깅 실패: productId={}", event.productId(), e);
+        }
+    }
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handlePaymentCompleted(PaymentCompletedEvent event) {
+        try {
+            log.info("결제 완료: paymentId={}, orderId={}, userId={}, amount={}",
+                    event.paymentId(), event.orderId(), event.userId(), event.amount());
+        } catch (Exception e) {
+            log.error("결제 완료 로깅 실패: paymentId={}", event.paymentId(), e);
+        }
+    }
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handlePaymentFailed(PaymentFailedEvent event) {
+        try {
+            log.info("결제 실패: paymentId={}, orderId={}, userId={}, reason={}",
+                    event.paymentId(), event.orderId(), event.userId(), event.reason());
+        } catch (Exception e) {
+            log.error("결제 실패 로깅 실패: paymentId={}", event.paymentId(), e);
+        }
+    }
+
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Async
+    public void handlePaymentCanceled(PaymentCanceledEvent event) {
+        try {
+            log.info("결제 취소: paymentId={}, orderId={}, userId={}",
+                    event.paymentId(), event.orderId(), event.userId());
+        } catch (Exception e) {
+            log.error("결제 취소 로깅 실패: paymentId={}", event.paymentId(), e);
+        }
+    }
+}

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

@@ -1,6 +1,8 @@
 package com.loopers.application.like;
 
 import com.loopers.application.brand.BrandService;
+import com.loopers.application.event.ProductLikedEvent;
+import com.loopers.application.event.ProductUnlikedEvent;
 import com.loopers.application.product.ProductService;
 import com.loopers.domain.brand.Brand;
 import com.loopers.infrastructure.product.ProductCacheManager;
@@ -9,11 +11,11 @@
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
-import static com.loopers.support.transaction.TransactionHelper.afterCommit;
 
 import java.util.Map;
 import java.util.Set;
@@ -27,6 +29,7 @@ public class LikeFacade {
     private final ProductService productService;
     private final BrandService brandService;
     private final ProductCacheManager productCacheManager;
+    private final ApplicationEventPublisher eventPublisher;
 
     // Command
 
@@ -36,17 +39,15 @@ public void like(Long userId, Long productId) {
 
         boolean created = likeService.like(userId, productId);
         if (created) {
-            productService.incrementLikeCount(productId);
-            afterCommit(() -> productCacheManager.evictDetail(productId));
+            eventPublisher.publishEvent(new ProductLikedEvent(userId, productId));
         }
     }
 
     @Transactional
     public void unlike(Long userId, Long productId) {
         boolean deleted = likeService.unlike(userId, productId);
         if (deleted) {
-            productService.decrementLikeCountIfPositive(productId);
-            afterCommit(() -> productCacheManager.evictDetail(productId));
+            eventPublisher.publishEvent(new ProductUnlikedEvent(userId, productId));
         }
     }
 

apps/commerce-api/src/main/java/com/loopers/application/order/OrderFacade.java

@@ -63,6 +63,19 @@ public OrderInfo placeOrder(Long userId, OrderCommand.Place command) {
         return OrderInfo.from(order);
     }
 
+    @Transactional
+    public void expireOrder(Long orderId) {
+        boolean expired = orderService.expireIfCreated(orderId);
+        if (!expired) return;
+
+        Order order = orderService.getOrder(orderId);
+        stockService.releaseReserved(order.getProductQuantities());
+
+        if (order.hasCoupon()) {
+            issuedCouponService.restore(order.getIssuedCouponId());
+        }
+    }
+
     public void cancelOrder(Long userId, Long orderId) {
         Order order = orderService.getOrder(orderId);
         order.validateOwnership(userId);

apps/commerce-api/src/main/java/com/loopers/application/order/OrderService.java

@@ -59,6 +59,11 @@ public void cancelOrder(Long orderId) {
         order.cancel();
     }
 
+    @Transactional
+    public boolean expireIfCreated(Long orderId) {
+        return orderRepository.updateStatusIfCurrent(orderId, OrderStatus.CANCELED, OrderStatus.CREATED) > 0;
+    }
+
     // Query
 
     @Transactional(readOnly = true)
@@ -91,4 +96,9 @@ public Page<Order> findOrdersByProductId(Long productId, Pageable pageable) {
     public List<Order> findOrdersByStatusWithItems(OrderStatus status) {
         return orderRepository.findAllByStatusWithItems(status);
     }
+
+    @Transactional(readOnly = true)
+    public List<Order> findCreatedOlderThanWithItems(ZonedDateTime threshold) {
+        return orderRepository.findAllByStatusAndCreatedAtBeforeWithItems(OrderStatus.CREATED, threshold);
+    }
 }