Merge pull request #328 from dd-jiny/volume-8

[Volume 8] 대기열 시스템 구현

Author: dd-jiny

CLAUDE.md

@@ -6,7 +6,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 **loopers-java-spring-template** — Multi-module Java 21 / Spring Boot 3.4.4 commerce backend (Gradle Kotlin DSL).
 
-Group: `com.loopers` | 감성 이커머스 MVP: 좋아요 → 장바구니 → 주문(결제 대기) 흐름.
+Group: `com.loopers` | 감성 이커머스 MVP: 좋아요 → 장바구니 → 주문 → 결제 흐름. 이벤트 기반 아키텍처(Transactional Outbox + Kafka).
 
 ## Build & Test Commands
 
@@ -105,7 +105,7 @@ interfaces/ → application/ → domain/ ← infrastructure/
 | 구분 | 구조 | 해당 도메인 |
 |------|------|------------|
 | **단순 도메인** | Controller → **Service** 직접 호출 | User, Brand, Like, Stats |
-| **복잡한 도메인** | Controller → **Facade** → 여러 Service | Product, Cart, Order |
+| **복잡한 도메인** | Controller → **Facade** → 여러 Service | Product, Cart, Order, Coupon, Payment |
 
 - **Facade**: 여러 도메인 서비스를 조합(orchestration)하고, 복잡한 비즈니스 플로우(재고 hold/release, 보상 트랜잭션 등)를 담당하는 application 레이어 클래스.
 - **단순 도메인은 AppService 없이 Controller에서 Service를 직접 호출**한다.
@@ -156,6 +156,9 @@ interfaces/ → application/ → domain/ ← infrastructure/
 | **Like** | 상품 좋아요 등록/취소 (멱등), 복합 PK, ProductService 참조 | `BaseStringIdEntity` | Service 직접 호출 |
 | **Cart** | 장바구니 CRUD, 주문 연계 복원, 복합 PK | `BaseStringIdEntity` | Facade |
 | **Order** | 주문 생성(DIRECT/CART), 취소, 만료 | `BaseStringIdEntity` | Facade |
+| **Coupon** | 쿠폰 발급 (Rush 선착순), 3-상태 머신 (AVAILABLE→RESERVED→USED) | `BaseStringIdEntity` | Facade |
+| **Payment** | PG 결제 연동, 분산락, Resilience4j | `BaseStringIdEntity` | Facade |
+| **Outbox** | Transactional Outbox 이벤트 발행 | - | Service |
 | **Stats** | 운영 통계 (주문 현황, 인기 상품) | - | Service 직접 호출 |
 
 ## Entity Base Classes
@@ -226,6 +229,41 @@ Both provide: `createdAt`, `updatedAt`, `guard()` override for entity validation
 
 DIRECT 주문 취소/만료 시 `order_cart_restore` 테이블 `existsById` 확인으로 1회 복원 보장. CART 주문은 장바구니 유지.
 
+### Transactional Outbox + Kafka 이벤트 파이프라인
+
+비즈니스 트랜잭션과 이벤트 발행의 원자성을 보장하는 Outbox 패턴:
+
+1. **Outbox 저장**: `OutboxEventService`가 비즈니스 TX 내에서 `OutboxEventModel` 저장 (`Propagation.MANDATORY`)
+2. **Relay 발행**: `OutboxEventRelay`가 5초 주기 폴링 → Kafka 발행 (PENDING → PUBLISHED, 실패 시 지수 백오프 10s→60s)
+3. **Consumer 처리**: `commerce-streamer`가 토픽별 Processor로 이벤트 소비 (멱등성 보장: `EventHandledModel`)
+4. **Cleanup**: `OutboxCleanupScheduler`가 7일 경과 PUBLISHED 이벤트 삭제
+
+**Kafka 토픽**:
+- `catalog-events` (3 partitions, productId 파티셔닝): 좋아요/조회 이벤트
+- `order-events` (3 partitions, orderId 파티셔닝): 주문 생성/취소/만료 이벤트
+- `coupon-issue-requests` (1 partition, 순서 보장): 선착순 쿠폰 발급
+
+**ApplicationEvent → Outbox 흐름**: `@TransactionalEventListener(AFTER_COMMIT)`로 도메인 이벤트 핸들링 후 Outbox 저장.
+
+### 결제 분산락 + Resilience4j
+
+**분산락**: `PaymentLock` 인터페이스 → Redis SETNX + Lua owner-verify unlock. 30초 TTL. Redis 장애 시 락 없이 진행 (가용성 우선).
+
+**Resilience4j 데코레이터 체인** (`ResilientPgClient`):
+- Bulkhead (40 semaphore) → CircuitBreaker (80% failure-rate, 15s open) → Retry (3x, jitter 0.5-1.5s) → HTTP
+
+**TX 분리 패턴**: Outbox 이벤트 TX-1 저장 → PG 호출 (TX 밖) → 결과 TX-2 저장. 커넥션 풀 점유 방지.
+
+### 쿠폰 Rush 발급
+
+선착순 쿠폰의 4-레이어 멱등 방어:
+1. `CouponDeduplicationCache`: Redis SETNX 중복 요청 차단
+2. `CouponRemainingCache`: Redis 잔여 수량 사전 차감
+3. CAS `issued_count` 업데이트로 오버셀 방지
+4. `CouponPendingActionRelay`: 3초 폴링으로 RESERVED → USED 확정 또는 복원
+
+**DIP**: `CouponDeduplicationCache`, `CouponRemainingCache`, `CouponIssueMetrics`, `OutboxRelayMetrics` — 도메인 인터페이스로 추출, infrastructure에서 구현.
+
 ### 상품 변경 이력 (ProductRevision)
 
 복합 PK (`product_id` + `revision_seq`). 상품 수정/삭제/복구 시 `before_snapshot`/`after_snapshot` JSON 저장.
@@ -242,6 +280,29 @@ DIRECT 주문 취소/만료 시 `order_cart_restore` 테이블 `existsById` 확
 | `UnavailableReason` | `DELETED`, `HIDDEN`, `BRAND_DELETED`, `BRAND_HIDDEN`, `STOPPED`, `TEMP_SOLD_OUT`, `OUT_OF_STOCK`, `INVALID_QUANTITY` | 장바구니 항목 주문 불가 사유 |
 | `RestoreReason` | `USER_CANCELLED`, `EXPIRED`, `PAYMENT_FAILED`, `PG_CANCELLED` | 복원 사유 |
 | `RestoreTriggerSource` | `CANCEL_API`, `PG_WEBHOOK`, `EXPIRE_JOB`, `MANUAL` | 복원 트리거 출처 |
+| `CouponActionStatus` | `AVAILABLE`, `RESERVED`, `USED` | 쿠폰 발급 3-상태 머신 |
+| `OutboxEventStatus` | `PENDING`, `PUBLISHED`, `CLEANUP` | Outbox 이벤트 상태 |
+
+## Schedulers (commerce-api)
+
+| 스케줄러 | 주기 | 역할 |
+|----------|------|------|
+| `OutboxEventRelay` | 5초 | Outbox → Kafka 발행 (지수 백오프) |
+| `OutboxCleanupScheduler` | - | 7일 경과 PUBLISHED 이벤트 삭제 |
+| `OrderExpiryScheduler` | 60초 | PENDING_PAYMENT → EXPIRED (결제 요청 중인 주문 제외) |
+| `PaymentPollingScheduler` | - | PG 결제 상태 폴링 (CB-aware fail-fast) |
+| `CouponActionRelay` | 3초 | RESERVED → USED 확정 또는 복원 |
+| `CartRestoreRetryScheduler` | 매일 05:30 | 실패한 장바구니 복원 재시도 |
+
+별도 `ThreadPoolTaskScheduler` (poolSize=3)로 `@Scheduled` 태스크 실행.
+
+## commerce-streamer (Kafka Consumer)
+
+Kafka 이벤트를 소비하여 집계/처리하는 별도 애플리케이션 (Web 없음):
+- `CatalogEventProcessor`: 좋아요/조회 집계 → `ProductMetricsModel` upsert
+- `OrderEventProcessor`: 상품별 주문 수 추적 (멱등)
+- `CouponIssueProcessor`: 선착순 쿠폰 발급 4-레이어 멱등 처리
+- `EventHandledModel`: event_id PK로 중복 처리 방지, 30일 후 자동 삭제
 
 ## Key Conventions
 
@@ -280,6 +341,8 @@ DIRECT 주문 취소/만료 시 `order_cart_restore` 테이블 `existsById` 확
 - `04-erd.md` — ERD (복합 PK, 인덱스 전략)
 - `05-architecture.md` — 종합 아키텍처 설계서
 - `07-facade-analysis.md` — Facade 필요성 분석
+- `07-index-cache-strategy.md` — 인덱스 + Redis 캐시 전략 (TTL, cache-aside, write-invalidate)
+- `08-order-payment-flow.md` — 주문-결제 상태 머신 + TX 분리 패턴 + PG 콜백 흐름
 
 ## 설계 원칙
 

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/CommerceApiApplication.java

@@ -4,6 +4,7 @@
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import java.util.TimeZone;
 
@@ -12,6 +13,7 @@
  * REST API 서버를 기동하며, 타임존을 Asia/Seoul로 설정하고 스케줄링을 활성화한다.
  */
 @ConfigurationPropertiesScan
+@EnableAsync
 @EnableScheduling
 @SpringBootApplication
 public class CommerceApiApplication {

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

@@ -0,0 +1,115 @@
+package com.loopers.application.coupon;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.domain.coupon.CouponDeduplicationCache;
+import com.loopers.domain.coupon.CouponIssueMetrics;
+import com.loopers.domain.coupon.CouponIssueRequestMessage;
+import com.loopers.domain.coupon.CouponIssueResultModel;
+import com.loopers.domain.coupon.CouponIssueResultRepository;
+import com.loopers.domain.coupon.CouponRemainingCache;
+import com.loopers.support.error.CoreException;
+import com.loopers.support.error.ErrorType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 선착순 쿠폰 발급 Facade — Thin Producer 전략.
+ *
+ * <p>Producer에서 DB TX를 사용하지 않는다.
+ * Redis SETNX(중복 거절) + Redis DECR(수량 게이트키퍼) + Kafka send만 수행.
+ * 모든 DB 작업은 Consumer(commerce-streamer)에서 단일 TX로 원자적 처리.</p>
+ */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class CouponIssueFacade {
+
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+    private final ObjectMapper objectMapper;
+    private final CouponRemainingCache couponRemainingCache;
+    private final CouponDeduplicationCache couponDeduplicationCache;
+    private final CouponIssueResultRepository resultRepository;
+    private final CouponIssueMetrics couponIssueMetrics;
+
+    public String requestRushIssue(Long userId, Long couponId) {
+        String requestId = UUID.randomUUID().toString();
+
+        // Phase 1: Redis SETNX — 중복 요청 거절
+        if (!tryDeduplication(userId, couponId)) {
+            throw new CoreException(ErrorType.CONFLICT, "이미 발급 요청한 쿠폰입니다");
+        }
+
+        // Phase 2: Redis DECR — 수량 소진 즉시 거절
+        if (!tryDecrementRemaining(couponId)) {
+            removeDeduplication(userId, couponId);
+            throw new CoreException(ErrorType.BAD_REQUEST, "쿠폰이 모두 소진되었습니다.");
+        }
+
+        // Phase 3: Kafka send — Consumer에게 처리 위임
+        try {
+            String jsonMessage = objectMapper.writeValueAsString(
+                new CouponIssueRequestMessage(requestId, userId, couponId, LocalDateTime.now()));
+            kafkaTemplate.send("coupon-issue-requests",
+                couponId.toString(), jsonMessage).get(5, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            restoreRemaining(couponId);
+            removeDeduplication(userId, couponId);
+            throw new CoreException(ErrorType.INTERNAL_ERROR, "발급 요청 실패. 다시 시도해주세요.");
+        }
+
+        return requestId;
+    }
+
+    public CouponIssueResultModel getIssueResult(String requestId) {
+        return resultRepository.findById(requestId)
+            .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "발급 처리 중입니다"));
+    }
+
+    private boolean tryDeduplication(Long userId, Long couponId) {
+        try {
+            return couponDeduplicationCache.trySetIfAbsent(userId, couponId, Duration.ofHours(24));
+        } catch (Exception e) {
+            log.warn("[Redis장애] 중복 확인 실패, Consumer에서 방어", e);
+            return true;
+        }
+    }
+
+    private void removeDeduplication(Long userId, Long couponId) {
+        try {
+            couponDeduplicationCache.delete(userId, couponId);
+        } catch (Exception e) {
+            log.warn("[Redis장애] dedup key 삭제 실패, TTL(24h) 후 자동 만료", e);
+        }
+    }
+
+    private boolean tryDecrementRemaining(Long couponId) {
+        try {
+            Long remaining = couponRemainingCache.decrementAndGet(couponId);
+            if (remaining == null || remaining < 0) {
+                restoreRemaining(couponId);
+                return false;
+            }
+            return true;
+        } catch (Exception e) {
+            log.warn("[Redis장애] DECR 실패, DB CAS fallback", e);
+            couponIssueMetrics.incrementRedisFallback();
+            return true;
+        }
+    }
+
+    private void restoreRemaining(Long couponId) {
+        try {
+            couponRemainingCache.increment(couponId);
+        } catch (Exception e) {
+            log.warn("[Redis장애] INCR 복원 실패, 동기화 배치에서 보정", e);
+            couponIssueMetrics.incrementIncrRestoreFail();
+        }
+    }
+}

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

@@ -0,0 +1,93 @@
+package com.loopers.application.event;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.domain.like.event.ProductLikedEvent;
+import com.loopers.domain.like.event.ProductUnlikedEvent;
+import com.loopers.domain.product.ProductService;
+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.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+
+/**
+ * 좋아요 이벤트 핸들러.
+ *
+ * <p>트랜잭션 커밋 후(AFTER_COMMIT) 비동기로 실행되며,
+ * 상품의 좋아요 수를 증감시킨다. REQUIRES_NEW 트랜잭션으로 독립 실행된다.</p>
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class LikeEventHandler {
+
+    private final ProductService productService;
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+    private final ObjectMapper objectMapper;
+
+    /**
+     * 좋아요 등록 이벤트를 처리한다 (좋아요 수 증가).
+     */
+    @Async("eventTaskExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public void handleProductLiked(ProductLikedEvent event) {
+        try {
+            productService.incrementLikeCount(event.productId());
+            log.info("[ProductLiked] userId={}, productId={}", event.userId(), event.productId());
+
+            // Kafka direct send: 카탈로그 이벤트 (Outbox 불필요 — 비핵심 지표)
+            sendCatalogEvent("PRODUCT_LIKED", event.productId(), event.userId());
+        } catch (Exception e) {
+            log.error("[ProductLiked] Failed to increment like count. userId={}, productId={}",
+                    event.userId(), event.productId(), e);
+        }
+    }
+
+    /**
+     * 좋아요 취소 이벤트를 처리한다 (좋아요 수 감소).
+     */
+    @Async("eventTaskExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public void handleProductUnliked(ProductUnlikedEvent event) {
+        try {
+            productService.decrementLikeCount(event.productId());
+            log.info("[ProductUnliked] userId={}, productId={}", event.userId(), event.productId());
+
+            // Kafka direct send: 카탈로그 이벤트 (Outbox 불필요 — 비핵심 지표)
+            sendCatalogEvent("PRODUCT_UNLIKED", event.productId(), event.userId());
+        } catch (Exception e) {
+            log.error("[ProductUnliked] Failed to decrement like count. userId={}, productId={}",
+                    event.userId(), event.productId(), e);
+        }
+    }
+
+    private void sendCatalogEvent(String eventType, Long productId, Long userId) {
+        try {
+            Map<String, Object> message = Map.of(
+                "eventType", eventType,
+                "productId", productId,
+                "userId", userId,
+                "occurredAt", LocalDateTime.now().toString()
+            );
+            String jsonMessage = objectMapper.writeValueAsString(message);
+            kafkaTemplate.send("catalog-events", String.valueOf(productId), jsonMessage)
+                .whenComplete((result, ex) -> {
+                    if (ex != null) {
+                        log.error("[CatalogEvent] Kafka 발행 실패 — eventType={}, productId={}",
+                            eventType, productId, ex);
+                    }
+                });
+        } catch (Exception e) {
+            log.warn("[CatalogEvent] Kafka 전송 실패 — eventType={}, productId={}", eventType, productId, e);
+        }
+    }
+}

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

@@ -0,0 +1,53 @@
+package com.loopers.application.event;
+
+import com.loopers.domain.order.event.OrderCancelledEvent;
+import com.loopers.domain.order.event.OrderCreatedEvent;
+import com.loopers.domain.order.event.OrderExpiredEvent;
+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;
+
+/**
+ * 주문 이벤트 핸들러.
+ *
+ * <p>트랜잭션 커밋 후(AFTER_COMMIT) 비동기로 실행되며,
+ * 주문 생성/취소/만료 이벤트를 처리한다.</p>
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OrderEventHandler {
+
+    /**
+     * 주문 생성 이벤트를 처리한다 (로깅).
+     */
+    @Async("eventTaskExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleOrderCreated(OrderCreatedEvent event) {
+        log.info("[OrderCreated] orderId={}, userId={}, orderType={}, totalAmount={}",
+                event.orderId(), event.userId(), event.orderType(), event.totalAmount());
+    }
+
+    /**
+     * 주문 취소 이벤트를 처리한다 (로깅).
+     */
+    @Async("eventTaskExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleOrderCancelled(OrderCancelledEvent event) {
+        log.info("[OrderCancelled] orderId={}, userId={}",
+                event.orderId(), event.userId());
+    }
+
+    /**
+     * 주문 만료 이벤트를 처리한다 (로깅).
+     */
+    @Async("eventTaskExecutor")
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void handleOrderExpired(OrderExpiredEvent event) {
+        log.info("[OrderExpired] orderId={}, userId={}",
+                event.orderId(), event.userId());
+    }
+}