[VOLUME-10] Spring Batch 를 이용한 주간, 월간 랭킹 구현 - 조용민

.gitignore

@@ -38,3 +38,10 @@ out/
 
 ### Kotlin ###
 .kotlin
+
+### k6 load test outputs ###
+tests/k6/results/
+
+### Claude Code workspace ###
+.claude/
+infra.md

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

@@ -13,19 +13,36 @@
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 import java.time.Instant;
 import java.util.UUID;
 
+/**
+ * 실험용 플래그
+ * BYPASS_OUTBOX=true   → Outbox 없이 TX commit 후 직접 Kafka 발행
+ * CRASH_BEFORE_SEND=true → TX commit 완료 후, Kafka send 직전 크래시 시뮬레이션 → 이벤트 유실 재현
+ */
+@Slf4j
 @RequiredArgsConstructor
 @Component
 public class CouponService {
 
+    private static final boolean BYPASS_OUTBOX =
+        Boolean.parseBoolean(System.getenv().getOrDefault("BYPASS_OUTBOX", "false"));
+
+    private static final boolean CRASH_BEFORE_SEND =
+        Boolean.parseBoolean(System.getenv().getOrDefault("CRASH_BEFORE_SEND", "false"));
+
     private final CouponTemplateRepository couponTemplateRepository;
     private final IssuedCouponRepository issuedCouponRepository;
     private final OutboxEventRepository outboxEventRepository;
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
     private final ObjectMapper objectMapper;
 
     // 기존 동기 발급 (주문 플로우에서 사용)
@@ -53,6 +70,29 @@ public CouponInfo.IssuedCouponInfo issue(Long memberId, Long couponTemplateId) {
     public String requestIssue(Long memberId, Long couponTemplateId) {
         String eventId = UUID.randomUUID().toString();
         CouponIssueRequestEvent event = new CouponIssueRequestEvent(eventId, memberId, couponTemplateId, Instant.now().toEpochMilli());
+
+        if (BYPASS_OUTBOX) {
+            // [실험 3] Outbox 없이 직접 발행 — TX commit 후 afterCommit()에서 Kafka 발행 시도
+            // DB commit과 Kafka send가 원자적이지 않음 → 그 사이 크래시 시 이벤트 유실
+            String payload = objectMapper.writeValueAsString(event);
+            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                @Override
+                public void afterCommit() {
+                    if (CRASH_BEFORE_SEND) {
+                        log.error("[Bypass] CRASH_BEFORE_SEND — DB commit 완료, Kafka send 전 크래시! eventId={}", eventId);
+                        throw new RuntimeException("simulated crash before Kafka send");
+                    }
+                    try {
+                        kafkaTemplate.send(KafkaTopics.COUPON_ISSUE_REQUESTS, String.valueOf(couponTemplateId), objectMapper.readTree(payload));
+                        log.info("[Bypass] direct Kafka send eventId={}", eventId);
+                    } catch (Exception e) {
+                        log.error("[Bypass] Kafka send failed eventId={}", eventId, e);
+                    }
+                }
+            });
+            return eventId;
+        }
+
         outboxEventRepository.save(OutboxEvent.create(eventId, KafkaTopics.COUPON_ISSUE_REQUESTS, String.valueOf(couponTemplateId), objectMapper.writeValueAsString(event)));
         return eventId; // 클라이언트가 결과 polling 시 사용
     }

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

@@ -18,6 +18,7 @@
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.support.TransactionSynchronization;
@@ -28,6 +29,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+@Slf4j
 @RequiredArgsConstructor
 @Component
 public class OrderService {
@@ -138,16 +140,21 @@ public OrderResult placeOrder(Long memberId, List<OrderLineRequest> items, Long
             .collect(Collectors.toList());
 
         // Outbox에 ORDERED 이벤트 저장 (주문 라인별 1건씩 — 랭킹은 상품 단위)
-        for (Order.OrderLineEntity ol : order.getOrderLines()) {
-            String eventId = UUID.randomUUID().toString();
-            CatalogEvent event = CatalogEvent.ordered(
-                eventId, ol.getProductId(), memberId, Instant.now().toEpochMilli(),
-                ol.getUnitPrice(), ol.getQuantity()
-            );
-            outboxEventRepository.save(OutboxEvent.create(
-                eventId, KafkaTopics.CATALOG_EVENTS, String.valueOf(ol.getProductId()),
-                objectMapper.writeValueAsString(event)
-            ));
+        // 랭킹은 보조 지표이므로 실패해도 주문을 롤백하지 않는다 (fail-open)
+        try {
+            for (Order.OrderLineEntity ol : order.getOrderLines()) {
+                String eventId = UUID.randomUUID().toString();
+                CatalogEvent event = CatalogEvent.ordered(
+                    eventId, ol.getProductId(), memberId, Instant.now().toEpochMilli(),
+                    ol.getUnitPrice(), ol.getQuantity()
+                );
+                outboxEventRepository.save(OutboxEvent.create(
+                    eventId, KafkaTopics.CATALOG_EVENTS, String.valueOf(ol.getProductId()),
+                    objectMapper.writeValueAsString(event)
+                ));
+            }
+        } catch (Exception e) {
+            log.warn("[Order] ORDERED 이벤트 Outbox 저장 실패 orderId={}", order.getId(), e);
         }
 
         // 주문 완료 후 entered 키 삭제 — 슬롯 즉시 반환 (TTL 만료 대기 없이)

apps/commerce-api/src/main/java/com/loopers/application/ranking/RankingFacade.java

@@ -4,13 +4,18 @@
 import com.loopers.domain.product.BrandRepository;
 import com.loopers.domain.product.Product;
 import com.loopers.domain.product.ProductRepository;
+import com.loopers.domain.ranking.ProductRankEntry;
+import com.loopers.domain.ranking.RankingRepository;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import org.springframework.stereotype.Component;
 
+import java.time.DayOfWeek;
 import java.time.LocalDate;
+import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -21,24 +26,28 @@
 public class RankingFacade {
 
     private static final DateTimeFormatter KEY_DATE_FMT = DateTimeFormatter.BASIC_ISO_DATE;
+    private static final ZoneId ZONE = ZoneId.of("Asia/Seoul");
     private static final String DEFAULT_REDIS_TEMPLATE = "defaultRedisTemplate";
 
     private final RedisTemplate<String, String> redisTemplate;
     private final ProductRepository productRepository;
     private final BrandRepository brandRepository;
+    private final RankingRepository rankingRepository;
 
     public RankingFacade(
         @Qualifier(DEFAULT_REDIS_TEMPLATE) RedisTemplate<String, String> redisTemplate,
         ProductRepository productRepository,
-        BrandRepository brandRepository
+        BrandRepository brandRepository,
+        RankingRepository rankingRepository
     ) {
         this.redisTemplate = redisTemplate;
         this.productRepository = productRepository;
         this.brandRepository = brandRepository;
+        this.rankingRepository = rankingRepository;
     }
 
     /**
-     * Top-N 랭킹 조회 (ZREVRANGE + 상품 정보 Aggregation)
+     * 일간 Top-N 랭킹 조회 (Redis ZSET)
      */
     public List<RankingInfo> getTopRankings(String date, int page, int size) {
         String key = rankingKey(date);
@@ -54,18 +63,11 @@ public List<RankingInfo> getTopRankings(String date, int page, int size) {
             .map(t -> Long.parseLong(t.getValue()))
             .toList();
 
-        Map<Long, Product> productMap = productRepository.findAllByIds(productIds).stream()
-            .collect(Collectors.toMap(Product::getId, p -> p));
-
-        Map<Long, Brand> brandMap = productMap.values().stream()
-            .map(Product::getBrandId)
-            .filter(id -> id != null)
-            .distinct()
-            .flatMap(id -> brandRepository.findById(id).stream())
-            .collect(Collectors.toMap(Brand::getId, b -> b));
+        Map<Long, Product> productMap = fetchProductMap(productIds);
+        Map<Long, Brand> brandMap = fetchBrandMap(productMap);
 
         long rank = start + 1;
-        List<RankingInfo> result = new java.util.ArrayList<>();
+        List<RankingInfo> result = new ArrayList<>();
         for (TypedTuple<String> tuple : tuples) {
             Long productId = Long.parseLong(tuple.getValue());
             Product product = productMap.get(productId);
@@ -86,15 +88,77 @@ public List<RankingInfo> getTopRankings(String date, int page, int size) {
         return result;
     }
 
+    /**
+     * 주간 랭킹 조회 (MV 테이블)
+     */
+    public List<RankingInfo> getWeeklyRankings(LocalDate date, int page, int size) {
+        LocalDate weekStart = date.with(DayOfWeek.MONDAY);
+        List<ProductRankEntry> entries = rankingRepository.findWeeklyRankings(weekStart, page, size);
+        return enrichRankEntries(entries);
+    }
+
+    /**
+     * 월간 랭킹 조회 (MV 테이블)
+     */
+    public List<RankingInfo> getMonthlyRankings(LocalDate date, int page, int size) {
+        LocalDate monthStart = date.withDayOfMonth(1);
+        List<ProductRankEntry> entries = rankingRepository.findMonthlyRankings(monthStart, page, size);
+        return enrichRankEntries(entries);
+    }
+
     /**
      * 개별 상품 순위 조회 (ZREVRANK, 0-based → 1-based 변환)
      */
     public Long getRank(Long productId) {
-        String key = rankingKey(LocalDate.now().format(KEY_DATE_FMT));
+        String key = rankingKey(LocalDate.now(ZONE).format(KEY_DATE_FMT));
         Long rank = redisTemplate.opsForZSet().reverseRank(key, String.valueOf(productId));
         return rank != null ? rank + 1 : null;
     }
 
+    private List<RankingInfo> enrichRankEntries(List<ProductRankEntry> entries) {
+        if (entries.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<Long> productIds = entries.stream().map(ProductRankEntry::productId).toList();
+        Map<Long, Product> productMap = fetchProductMap(productIds);
+        Map<Long, Brand> brandMap = fetchBrandMap(productMap);
+
+        List<RankingInfo> result = new ArrayList<>();
+        for (ProductRankEntry entry : entries) {
+            Product product = productMap.get(entry.productId());
+            if (product == null) continue;
+            Brand brand = product.getBrandId() != null ? brandMap.get(product.getBrandId()) : null;
+            result.add(new RankingInfo(
+                entry.rankPosition(),
+                entry.productId(),
+                product.getName(),
+                product.getPrice(),
+                brand != null ? brand.getName() : null,
+                entry.totalScore()
+            ));
+        }
+        return result;
+    }
+
+    private Map<Long, Product> fetchProductMap(List<Long> productIds) {
+        return productRepository.findAllByIds(productIds).stream()
+            .collect(Collectors.toMap(Product::getId, p -> p));
+    }
+
+    private Map<Long, Brand> fetchBrandMap(Map<Long, Product> productMap) {
+        List<Long> brandIds = productMap.values().stream()
+            .map(Product::getBrandId)
+            .filter(id -> id != null)
+            .distinct()
+            .toList();
+        if (brandIds.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        return brandRepository.findAllByIds(brandIds).stream()
+            .collect(Collectors.toMap(Brand::getId, b -> b));
+    }
+
     private String rankingKey(String date) {
         return "ranking:all:" + date;
     }

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

@@ -1,10 +1,13 @@
 package com.loopers.domain.product;
 
+import java.util.List;
 import java.util.Optional;
 
 public interface BrandRepository {
 
     Brand save(Brand brand);
 
     Optional<Brand> findById(Long id);
+
+    List<Brand> findAllByIds(List<Long> ids);
 }

apps/commerce-api/src/main/java/com/loopers/domain/ranking/ProductRankEntry.java

@@ -0,0 +1,11 @@
+package com.loopers.domain.ranking;
+
+/**
+ * MV 테이블 조회 결과를 담는 도메인 VO.
+ * weekly/monthly 공통으로 사용한다.
+ */
+public record ProductRankEntry(
+    Long productId,
+    double totalScore,
+    int rankPosition
+) {}

apps/commerce-api/src/main/java/com/loopers/domain/ranking/RankingPeriod.java

@@ -0,0 +1,5 @@
+package com.loopers.domain.ranking;
+
+public enum RankingPeriod {
+    DAILY, WEEKLY, MONTHLY
+}

apps/commerce-api/src/main/java/com/loopers/domain/ranking/RankingRepository.java

@@ -0,0 +1,11 @@
+package com.loopers.domain.ranking;
+
+import java.time.LocalDate;
+import java.util.List;
+
+public interface RankingRepository {
+
+    List<ProductRankEntry> findWeeklyRankings(LocalDate weekStartDate, int page, int size);
+
+    List<ProductRankEntry> findMonthlyRankings(LocalDate monthStartDate, int page, int size);
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/product/BrandRepositoryImpl.java

@@ -5,6 +5,7 @@
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
 import java.util.Optional;
 
 @RequiredArgsConstructor
@@ -22,4 +23,9 @@ public Brand save(Brand brand) {
     public Optional<Brand> findById(Long id) {
         return brandJpaRepository.findById(id);
     }
+
+    @Override
+    public List<Brand> findAllByIds(List<Long> ids) {
+        return brandJpaRepository.findAllById(ids);
+    }
 }

apps/commerce-api/src/main/java/com/loopers/infrastructure/ranking/MvProductRankMonthlyEntity.java

@@ -0,0 +1,59 @@
+package com.loopers.infrastructure.ranking;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.IdClass;
+import jakarta.persistence.Table;
+import org.hibernate.annotations.Immutable;
+
+import java.io.Serializable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Objects;
+
+@Immutable
+@Entity
+@Table(name = "mv_product_rank_monthly")
+@IdClass(MvProductRankMonthlyEntity.PK.class)
+public class MvProductRankMonthlyEntity {
+
+    @Id
+    @Column(name = "product_id")
+    private Long productId;
+
+    @Id
+    @Column(name = "month_start_date")
+    private LocalDate monthStartDate;
+
+    @Column(name = "total_score")
+    private double totalScore;
+
+    @Column(name = "rank_position")
+    private int rankPosition;
+
+    @Column(name = "aggregated_at")
+    private LocalDateTime aggregatedAt;
+
+    public Long getProductId() { return productId; }
+    public LocalDate getMonthStartDate() { return monthStartDate; }
+    public double getTotalScore() { return totalScore; }
+    public int getRankPosition() { return rankPosition; }
+
+    public static class PK implements Serializable {
+        private Long productId;
+        private LocalDate monthStartDate;
+
+        protected PK() {}
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (!(o instanceof PK pk)) return false;
+            return Objects.equals(productId, pk.productId) && Objects.equals(monthStartDate, pk.monthStartDate);
+        }
+
+        @Override
+        public int hashCode() { return Objects.hash(productId, monthStartDate); }
+    }
+}