[volume - 8] Decoupling with Kafka

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"))
@@ -34,4 +35,5 @@ dependencies {
     // test-fixtures
     testImplementation(testFixtures(project(":modules:jpa")))
     testImplementation(testFixtures(project(":modules:redis")))
+    testImplementation(testFixtures(project(":modules:kafka")))
 }

apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java

@@ -6,10 +6,13 @@
 import com.loopers.domain.brand.Brand;
 import com.loopers.domain.product.Product;
 import com.loopers.domain.product.ProductDetail;
+import com.loopers.domain.product.ProductEvent;
+import com.loopers.domain.product.ProductEventPublisher;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 import java.util.Map;
@@ -30,6 +33,7 @@ public class CatalogFacade {
     private final BrandService brandService;
     private final ProductService productService;
     private final ProductCacheService productCacheService;
+    private final ProductEventPublisher productEventPublisher;
 
     /**
      * 상품 목록을 조회합니다.
@@ -103,16 +107,20 @@ public ProductInfoList getProducts(Long brandId, String sort, int page, int size
      * 상품 정보를 조회합니다.
      * <p>
      * Redis 캐시를 먼저 확인하고, 캐시에 없으면 DB에서 조회한 후 캐시에 저장합니다.
+     * 상품 조회 시 ProductViewed 이벤트를 발행하여 메트릭 집계에 사용합니다.
      * </p>
      *
      * @param productId 상품 ID
      * @return 상품 정보와 좋아요 수
      * @throws CoreException 상품을 찾을 수 없는 경우
      */
+    @Transactional(readOnly = true)
     public ProductInfo getProduct(Long productId) {
         // 캐시에서 조회 시도
         ProductInfo cachedResult = productCacheService.getCachedProduct(productId);
         if (cachedResult != null) {
+            // 캐시 히트 시에도 조회 수 집계를 위해 이벤트 발행
+            productEventPublisher.publish(ProductEvent.ProductViewed.from(productId));
             return cachedResult;
         }
 
@@ -133,6 +141,9 @@ public ProductInfo getProduct(Long productId) {
         // 캐시에 저장
         productCacheService.cacheProduct(productId, result);
 
+        // ✅ 상품 조회 이벤트 발행 (메트릭 집계용)
+        productEventPublisher.publish(ProductEvent.ProductViewed.from(productId));
+
         // 로컬 캐시의 좋아요 수 델타 적용 (DB 조회 결과에도 델타 반영)
         return productCacheService.applyLikeCountDelta(result);
     }

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

@@ -0,0 +1,141 @@
+package com.loopers.application.outbox;
+
+import com.loopers.domain.like.LikeEvent;
+import com.loopers.domain.order.OrderEvent;
+import com.loopers.domain.product.ProductEvent;
+import lombok.RequiredArgsConstructor;
+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;
+
+/**
+ * Outbox Bridge Event Listener.
+ * <p>
+ * ApplicationEvent를 구독하여 외부 시스템(Kafka)으로 전송해야 하는 이벤트를
+ * Transactional Outbox Pattern을 통해 Outbox에 저장합니다.
+ * </p>
+ * <p>
+ * <b>표준 패턴:</b>
+ * <ul>
+ *   <li>EventPublisher는 ApplicationEvent만 발행 (단일 책임)</li>
+ *   <li>이 컴포넌트가 ApplicationEvent를 구독하여 Outbox에 저장 (관심사 분리)</li>
+ *   <li>트랜잭션 커밋 후(AFTER_COMMIT) 처리하여 에러 격리</li>
+ * </ul>
+ * </p>
+ * <p>
+ * <b>처리 이벤트:</b>
+ * <ul>
+ *   <li><b>LikeEvent:</b> LikeAdded, LikeRemoved → like-events</li>
+ *   <li><b>OrderEvent:</b> OrderCreated → order-events</li>
+ *   <li><b>ProductEvent:</b> ProductViewed → product-events</li>
+ * </ul>
+ * </p>
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OutboxBridgeEventListener {
+
+    private final OutboxEventService outboxEventService;
+
+    /**
+     * LikeAdded 이벤트를 Outbox에 저장합니다.
+     *
+     * @param event LikeAdded 이벤트
+     */
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleLikeAdded(LikeEvent.LikeAdded event) {
+        try {
+            outboxEventService.saveEvent(
+                "LikeAdded",
+                event.productId().toString(),
+                "Product",
+                event,
+                "like-events",
+                event.productId().toString()
+            );
+            log.debug("LikeAdded 이벤트를 Outbox에 저장: productId={}", event.productId());
+        } catch (Exception e) {
+            log.error("LikeAdded 이벤트 Outbox 저장 실패: productId={}", event.productId(), e);
+            // 외부 시스템 전송 실패는 내부 처리에 영향 없음
+        }
+    }
+
+    /**
+     * LikeRemoved 이벤트를 Outbox에 저장합니다.
+     *
+     * @param event LikeRemoved 이벤트
+     */
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleLikeRemoved(LikeEvent.LikeRemoved event) {
+        try {
+            outboxEventService.saveEvent(
+                "LikeRemoved",
+                event.productId().toString(),
+                "Product",
+                event,
+                "like-events",
+                event.productId().toString()
+            );
+            log.debug("LikeRemoved 이벤트를 Outbox에 저장: productId={}", event.productId());
+        } catch (Exception e) {
+            log.error("LikeRemoved 이벤트 Outbox 저장 실패: productId={}", event.productId(), e);
+            // 외부 시스템 전송 실패는 내부 처리에 영향 없음
+        }
+    }
+
+    /**
+     * OrderCreated 이벤트를 Outbox에 저장합니다.
+     *
+     * @param event OrderCreated 이벤트
+     */
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleOrderCreated(OrderEvent.OrderCreated event) {
+        try {
+            outboxEventService.saveEvent(
+                "OrderCreated",
+                event.orderId().toString(),
+                "Order",
+                event,
+                "order-events",
+                event.orderId().toString()
+            );
+            log.debug("OrderCreated 이벤트를 Outbox에 저장: orderId={}", event.orderId());
+        } catch (Exception e) {
+            log.error("OrderCreated 이벤트 Outbox 저장 실패: orderId={}", event.orderId(), e);
+            // 외부 시스템 전송 실패는 내부 처리에 영향 없음
+        }
+    }
+
+    /**
+     * ProductViewed 이벤트를 Outbox에 저장합니다.
+     *
+     * @param event ProductViewed 이벤트
+     */
+    @Async
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleProductViewed(ProductEvent.ProductViewed event) {
+        try {
+            outboxEventService.saveEvent(
+                "ProductViewed",
+                event.productId().toString(),
+                "Product",
+                event,
+                "product-events",
+                event.productId().toString()
+            );
+            log.debug("ProductViewed 이벤트를 Outbox에 저장: productId={}", event.productId());
+        } catch (Exception e) {
+            log.error("ProductViewed 이벤트 Outbox 저장 실패: productId={}", event.productId(), e);
+            // 외부 시스템 전송 실패는 내부 처리에 영향 없음
+        }
+    }
+}

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

@@ -0,0 +1,100 @@
+package com.loopers.application.outbox;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.domain.outbox.OutboxEventRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.UUID;
+
+/**
+ * Outbox 이벤트 저장 서비스.
+ * <p>
+ * 도메인 트랜잭션과 같은 트랜잭션에서 Outbox에 이벤트를 저장합니다.
+ * Application 레이어에 위치하여 비즈니스 로직(이벤트 저장 결정)을 처리합니다.
+ * </p>
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class OutboxEventService {
+
+    private final OutboxEventRepository outboxEventRepository;
+    private final ObjectMapper objectMapper;
+
+    /**
+     * Kafka로 전송할 이벤트를 Outbox에 저장합니다.
+     * <p>
+     * 도메인 트랜잭션과 같은 트랜잭션에서 실행되어야 합니다.
+     * 집계 ID별로 순차적인 버전을 자동으로 부여합니다.
+     * </p>
+     * <p>
+     * 버전 충돌 시 최대 3회까지 재시도합니다.
+     * 유니크 제약 조건을 통해 경쟁 조건을 감지하고 재시도합니다.
+     * </p>
+     *
+     * @param eventType 이벤트 타입 (예: "OrderCreated", "LikeAdded")
+     * @param aggregateId 집계 ID (예: orderId, productId)
+     * @param aggregateType 집계 타입 (예: "Order", "Product")
+     * @param event 이벤트 객체
+     * @param topic Kafka 토픽 이름
+     * @param partitionKey 파티션 키
+     */
+    @Transactional
+    public void saveEvent(
+        String eventType,
+        String aggregateId,
+        String aggregateType,
+        Object event,
+        String topic,
+        String partitionKey
+    ) {
+        int maxRetries = 3;
+        for (int i = 0; i < maxRetries; i++) {
+            try {
+                String eventId = UUID.randomUUID().toString();
+                String payload = objectMapper.writeValueAsString(event);
+
+                // 집계 ID별 최신 버전 조회 후 +1
+                Long latestVersion = outboxEventRepository.findLatestVersionByAggregateId(aggregateId, aggregateType);
+                Long nextVersion = latestVersion + 1L;
+
+                OutboxEvent outboxEvent = OutboxEvent.builder()
+                    .eventId(eventId)
+                    .eventType(eventType)
+                    .aggregateId(aggregateId)
+                    .aggregateType(aggregateType)
+                    .payload(payload)
+                    .topic(topic)
+                    .partitionKey(partitionKey)
+                    .version(nextVersion)
+                    .build();
+
+                outboxEventRepository.save(outboxEvent);
+                log.debug("Outbox 이벤트 저장: eventType={}, aggregateId={}, topic={}, version={}", 
+                    eventType, aggregateId, topic, nextVersion);
+                return; // 성공
+            } catch (DataIntegrityViolationException e) {
+                // 유니크 제약 조건 위반 (버전 충돌)
+                if (i == maxRetries - 1) {
+                    log.error("Outbox 이벤트 저장 실패 (최대 재시도 횟수 초과): eventType={}, aggregateId={}, retryCount={}", 
+                        eventType, aggregateId, i + 1, e);
+                    throw new RuntimeException("Outbox 이벤트 저장 실패: 버전 충돌", e);
+                }
+                log.warn("Outbox 이벤트 저장 재시도: eventType={}, aggregateId={}, retryCount={}", 
+                    eventType, aggregateId, i + 1);
+            } catch (Exception e) {
+                log.error("Outbox 이벤트 저장 실패: eventType={}, aggregateId={}", 
+                    eventType, aggregateId, e);
+                throw new RuntimeException("Outbox 이벤트 저장 실패", e);
+            }
+        }
+    }
+}