Round8

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/event/listener/OrderStatusEventListener.java

@@ -50,6 +50,13 @@ public void handlePaymentCompleted(PaymentCompletedEvent event) {
                 order.getOrderNo(),
                 order.getMemberId(),
                 order.getTotalPrice(),
+                order.getItems().stream()
+                    .map(item -> new OrderCompletedEvent.OrderItemInfo(
+                        item.getProductId(),
+                        item.getQuantity(),
+                        item.getUnitPrice()
+                    ))
+                    .toList(),
                 java.time.LocalDateTime.now()
             ));
 

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

@@ -2,11 +2,18 @@
 
 import com.loopers.domain.common.vo.Money;
 import java.time.LocalDateTime;
+import java.util.List;
 
 public record OrderCompletedEvent(
     String orderNo,
     Long memberId,
     Money totalPrice,
+    List<OrderItemInfo> items,
     LocalDateTime completedAt
 ) {
+    public record OrderItemInfo(
+        Long productId,
+        int quantity,
+        Money price
+    ) {}
 }

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

@@ -0,0 +1,11 @@
+package com.loopers.application.event.product;
+
+import java.time.LocalDateTime;
+
+public record ProductViewedEvent(
+    Long memberId,      // nullable (비로그인 사용자도 추적)
+    Long productId,
+    Long brandId,
+    LocalDateTime viewedAt
+) {
+}

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

@@ -1,18 +1,23 @@
 package com.loopers.application.product;
 
+import com.loopers.application.event.product.ProductViewedEvent;
 import com.loopers.domain.like.service.LikeReadService;
+import com.loopers.domain.product.Product;
+import com.loopers.domain.product.repository.ProductRepository;
 import com.loopers.domain.product.service.ProductReadService;
 import com.loopers.domain.product.command.ProductSearchFilter;
 import com.loopers.domain.product.enums.ProductSortCondition;
 import com.loopers.infrastructure.cache.ProductDetailCache;
 import com.loopers.infrastructure.cache.ProductListCache;
 import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.time.LocalDateTime;
 import java.util.List;
 
 @RequiredArgsConstructor
@@ -24,6 +29,8 @@ public class ProductFacade {
     private final LikeReadService likeReadService;
     private final ProductDetailCache productDetailCache;
     private final ProductListCache productListCache;
+    private final ProductRepository productRepository;
+    private final ApplicationEventPublisher eventPublisher;
 
     @Transactional(readOnly = true)
     public Page<ProductSummaryInfo> getProducts(ProductSearchCommand command) {
@@ -93,16 +100,17 @@ public ProductDetailInfo getProductDetail(Long productId, Long memberIdOrNull) {
                     return result;
                 });
 
-        // 2. 로그인하지 않은 경우 바로 반환
-        if (memberIdOrNull == null) {
-            return cachedInfo;  // isLikedByMember=false 그대로
-        }
+        // 2. Product 엔티티 조회 (brandId 획득용)
+        Product product = productRepository.findById(productId)
+                .orElseThrow(() -> new com.loopers.support.error.CoreException(
+                        com.loopers.support.error.ErrorType.NOT_FOUND,
+                        "상품을 찾을 수 없습니다."));
 
-        // 3. isLikedByMember만 동적 계산
-        boolean isLiked = likeReadService.isLikedBy(memberIdOrNull, productId);
+        // 3. isLikedByMember 동적 계산
+        boolean isLiked = memberIdOrNull != null && likeReadService.isLikedBy(memberIdOrNull, productId);
 
         // 4. isLikedByMember 필드만 교체해서 반환
-        return ProductDetailInfo.builder()
+        ProductDetailInfo result = ProductDetailInfo.builder()
                 .id(cachedInfo.getId())
                 .name(cachedInfo.getName())
                 .description(cachedInfo.getDescription())
@@ -113,6 +121,16 @@ public ProductDetailInfo getProductDetail(Long productId, Long memberIdOrNull) {
                 .likeCount(cachedInfo.getLikeCount())
                 .isLikedByMember(isLiked)  // ⭐ 동적 계산
                 .build();
+
+        // 5. ProductViewedEvent 발행 (조회수 집계)
+        eventPublisher.publishEvent(new ProductViewedEvent(
+            memberIdOrNull,  // 비로그인 사용자는 null
+            productId,
+            product.getBrandId(),
+            LocalDateTime.now()
+        ));
+
+        return result;
     }
 
     @Transactional(readOnly = true)

apps/commerce-api/src/main/java/com/loopers/domain/order/service/OrderPlacementService.java

@@ -11,14 +11,17 @@
 import com.loopers.domain.order.repository.OrderRepository;
 import com.loopers.domain.product.Product;
 import com.loopers.domain.product.repository.ProductRepository;
+import com.loopers.infrastructure.cache.CacheInvalidationService;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
 import java.util.List;
 
+@Slf4j
 @RequiredArgsConstructor
 @Component
 public class OrderPlacementService {
@@ -27,6 +30,7 @@ public class OrderPlacementService {
     private final ProductRepository productRepository;
     private final MemberRepository memberRepository;
     private final MemberCouponRepository memberCouponRepository;
+    private final CacheInvalidationService cacheInvalidationService;
 
     public Order placeOrder(OrderPlacementCommand command) {
         validateMemberExists(command.getMemberId());
@@ -83,6 +87,13 @@ private List<OrderItem> processOrderLines(List<OrderLineCommand> orderLines) {
                 throw new CoreException(ErrorType.BAD_REQUEST, "재고가 부족합니다.");
             }
 
+            // 재고 소진 시 캐시 무효화
+            int remainingStock = productRepository.getStockQuantity(product.getId());
+            if (remainingStock == 0) {
+                log.info("[Order] Stock depleted for productId={}, invalidating cache", product.getId());
+                cacheInvalidationService.invalidateOnStockDepletion(product.getId());
+            }
+
             items.add(new OrderItem(product.getId(), line.getQuantity(), product.getPrice()));
         }
 

apps/commerce-api/src/main/java/com/loopers/domain/product/repository/ProductRepository.java

@@ -26,6 +26,8 @@ public interface ProductRepository {
 
     int increaseStock(Long productId, int quantity);
 
+    int getStockQuantity(Long productId);
+
     int incrementLikeCount(Long productId);
 
     int decrementLikeCount(Long productId);

apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java

@@ -35,4 +35,13 @@ public void invalidateOnProductUpdate(Long productId) {
         log.info("[CacheInvalidation] Invalidating cache for product update, productId={}", productId);
         productDetailCache.delete(productId);
     }
+
+    /**
+     * Invalidate cache when product stock is depleted
+     */
+    public void invalidateOnStockDepletion(Long productId) {
+        log.info("[CacheInvalidation] Invalidating cache for stock depletion, productId={}", productId);
+        productDetailCache.delete(productId);
+        // Note: Product list cache는 TTL(60초)에 의존하여 자동 무효화
+    }
 }

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/KafkaTopicRouter.java

@@ -0,0 +1,48 @@
+package com.loopers.infrastructure.kafka;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Kafka Topic Router
+ * - 이벤트 타입에 따라 적절한 Kafka 토픽을 반환
+ * - 토픽 네이밍 규칙: loopers.commerce.{event-name}-v1
+ */
+@Component
+public class KafkaTopicRouter {
+
+    private static final String TOPIC_PREFIX = "loopers.commerce.";
+    private static final String TOPIC_VERSION = "-v1";
+
+    /**
+     * 이벤트 타입에 맞는 토픽 이름 반환
+     *
+     * @param eventType 이벤트 타입 (ORDER_PLACED, PRODUCT_LIKED 등)
+     * @return Kafka 토픽 이름
+     */
+    public String getTopicName(String eventType) {
+        return switch (eventType) {
+            // Order Events
+            case "ORDER_PLACED" -> TOPIC_PREFIX + "order-placed" + TOPIC_VERSION;
+            case "ORDER_COMPLETED" -> TOPIC_PREFIX + "order-completed" + TOPIC_VERSION;
+
+            // Payment Events
+            case "PAYMENT_COMPLETED" -> TOPIC_PREFIX + "payment-completed" + TOPIC_VERSION;
+
+            // Product Events
+            case "PRODUCT_LIKED" -> TOPIC_PREFIX + "product-liked" + TOPIC_VERSION;
+            case "PRODUCT_UNLIKED" -> TOPIC_PREFIX + "product-unliked" + TOPIC_VERSION;
+            case "PRODUCT_VIEWED" -> TOPIC_PREFIX + "product-viewed" + TOPIC_VERSION;
+
+            default -> throw new IllegalArgumentException("Unknown event type: " + eventType);
+        };
+    }
+
+    /**
+     * 토픽 이름이 유효한지 검증
+     */
+    public boolean isValidTopic(String topicName) {
+        return topicName != null &&
+               topicName.startsWith(TOPIC_PREFIX) &&
+               topicName.endsWith(TOPIC_VERSION);
+    }
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java

@@ -0,0 +1,43 @@
+package com.loopers.infrastructure.kafka.event;
+
+import java.time.ZonedDateTime;
+
+/**
+ * Kafka 이벤트 Envelope
+ * - Application Event를 Kafka로 전송할 때 메타데이터를 포함하기 위한 래퍼
+ * - 멱등성 체크 및 순서 보장을 위한 정보 포함
+ *
+ * @param <T> Application Event 타입 (OrderPlacedEvent, ProductLikedEvent 등)
+ */
+public record KafkaEventEnvelope<T>(
+    /**
+     * 이벤트 ID (Outbox Event의 ID)
+     * - Consumer에서 멱등성 체크에 사용
+     */
+    String eventId,
+
+    /**
+     * 이벤트 타입
+     * - 예: ORDER_PLACED, PRODUCT_LIKED, PAYMENT_COMPLETED
+     */
+    String eventType,
+
+    /**
+     * Partition Key
+     * - Kafka 파티션 분배 기준
+     * - 같은 값은 항상 같은 파티션으로 전송되어 순서 보장
+     */
+    String partitionKey,
+
+    /**
+     * 실제 이벤트 Payload
+     * - Application Event 객체 (OrderPlacedEvent, ProductLikedEvent 등)
+     */
+    T payload,
+
+    /**
+     * 이벤트 발생 시각
+     */
+    ZonedDateTime occurredAt
+) {
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java

@@ -0,0 +1,113 @@
+package com.loopers.infrastructure.outbox;
+
+import com.loopers.application.event.like.ProductLikedEvent;
+import com.loopers.application.event.like.ProductUnlikedEvent;
+import com.loopers.application.event.order.OrderCompletedEvent;
+import com.loopers.application.event.order.OrderPlacedEvent;
+import com.loopers.application.event.payment.PaymentCompletedEvent;
+import com.loopers.application.event.product.ProductViewedEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+/**
+ * Kafka Outbox Event Listener
+ * - Application Event를 받아서 Outbox 테이블에 저장
+ * - BEFORE_COMMIT: 같은 트랜잭션 내에서 처리되어 원자성 보장
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class KafkaOutboxEventListener {
+
+    private final OutboxEventWriter outboxWriter;
+
+    /**
+     * 주문 생성 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void handleOrderPlaced(OrderPlacedEvent event) {
+        log.debug("[Outbox] OrderPlacedEvent 수신 - orderNo: {}", event.orderNo());
+
+        outboxWriter.write(
+            event.orderNo(),           // partition key
+            "ORDER_PLACED",            // event type
+            event                      // payload
+        );
+    }
+
+    /**
+     * 주문 완료 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void handleOrderCompleted(OrderCompletedEvent event) {
+        log.debug("[Outbox] OrderCompletedEvent 수신 - orderNo: {}", event.orderNo());
+
+        outboxWriter.write(
+            event.orderNo(),
+            "ORDER_COMPLETED",
+            event
+        );
+    }
+
+    /**
+     * 결제 완료 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void handlePaymentCompleted(PaymentCompletedEvent event) {
+        log.debug("[Outbox] PaymentCompletedEvent 수신 - orderNo: {}", event.orderNo());
+
+        outboxWriter.write(
+            event.orderNo(),
+            "PAYMENT_COMPLETED",
+            event
+        );
+    }
+
+    /**
+     * 상품 좋아요 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void handleProductLiked(ProductLikedEvent event) {
+        log.debug("[Outbox] ProductLikedEvent 수신 - productId: {}", event.productId());
+
+        outboxWriter.write(
+            String.valueOf(event.productId()),  // partition key
+            "PRODUCT_LIKED",
+            event
+        );
+    }
+
+    /**
+     * 상품 좋아요 취소 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    public void handleProductUnliked(ProductUnlikedEvent event) {
+        log.debug("[Outbox] ProductUnlikedEvent 수신 - productId: {}", event.productId());
+
+        outboxWriter.write(
+            String.valueOf(event.productId()),
+            "PRODUCT_UNLIKED",
+            event
+        );
+    }
+
+    /**
+     * 상품 조회 이벤트 → Outbox 저장
+     */
+    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public void handleProductViewed(ProductViewedEvent event) {
+        log.debug("[Outbox] ProductViewedEvent 수신 - productId: {}", event.productId());
+
+        outboxWriter.write(
+            String.valueOf(event.productId()),  // partition key
+            "PRODUCT_VIEWED",
+            event
+        );
+    }
+}