[Round-10] Spring Batch를 활용한 주간/월간 랭킹 집계 구현

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

@@ -4,7 +4,7 @@
 import com.loopers.domain.like.LikeService;
 import com.loopers.domain.product.Product;
 import com.loopers.domain.product.ProductService;
-import com.loopers.infrastructure.message.ProductProducer;
+import com.loopers.infrastructure.message.ProductLikeProducer;
 import com.loopers.domain.product.CatalogMessage;
 import com.loopers.shared.event.Envelope;
 import lombok.RequiredArgsConstructor;
@@ -18,7 +18,7 @@
 public class LikeEventListener {
     private final LikeService likeService;
     private final ProductService productService;
-    private final ProductProducer productProducer;
+    private final ProductLikeProducer productLikeProducer;
 
     @Async("applicationEventTaskExecutor")
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
@@ -28,15 +28,14 @@ public void onAdded(Envelope<LikeEvent.Added> e){
 
         productService.updateLikeCnt(product, likeCnt);
 
-        Envelope<CatalogMessage.LikeChanged> record = Envelope.of(
+        Envelope<CatalogMessage.LikeAdded> record = Envelope.of(
                 e.actorId(),
-                new CatalogMessage.LikeChanged(
-                        product.getId(),
-                        e.payload().targetType().name(),
-                        likeCnt
+                new CatalogMessage.LikeAdded(
+                        e.actorId(),
+                        product.getId()
                 )
         );
-        productProducer.send(String.valueOf(product.getId()),record);
+        productLikeProducer.send(String.valueOf(product.getId()),record);
     }
 
     @Async("applicationEventTaskExecutor")
@@ -46,14 +45,13 @@ public void onRemoved(Envelope<LikeEvent.Removed> e) {
         Product product = productService.getProduct(e.payload().targetId());
         productService.updateLikeCnt(product, likeCnt);
 
-        Envelope<CatalogMessage.LikeChanged> record = Envelope.of(
+        Envelope<CatalogMessage.LikeRemoved> record = Envelope.of(
                 e.actorId(),
-                new CatalogMessage.LikeChanged(
-                        product.getId(),
-                        e.payload().targetType().name(),
-                        likeCnt
+                new CatalogMessage.LikeRemoved(
+                        e.actorId(),
+                        product.getId()
                 )
         );
-        productProducer.send(String.valueOf(product.getId()),record);
+        productLikeProducer.send(String.valueOf(product.getId()),record);
     }
 }

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

@@ -52,7 +52,7 @@ public ProductInfo.Detail getProductDetail(String loginId, Long productId) {
         String brandName = brandService.get(product.getBrandId()).getName();
 
         // 오늘 기준 랭킹 조회 (없으면 null)
-        Rank rank = rankingService.getRankInfo(LocalDate.now(ZoneId.of("Asia/Seoul")), productId);
+        Rank rank = rankingService.getRankInfo(LocalDate.now(ZoneId.of("Asia/Seoul")), productId,"daily");
 
         // 사용자 활동로그(상품 조회)
         productActivityPublisher.productDetail(new ProductActivityPayload.ProductDetailViewed(loginId, productId));
@@ -78,7 +78,7 @@ public ProductInfo.Detail getProductDetailWithCacheable(String loginId, Long pro
         String brandName = brandService.get(product.getBrandId()).getName();
 
         // 오늘 기준 랭킹 조회 (없으면 null)
-        Rank rank = rankingService.getRankInfo(LocalDate.now(ZoneId.of("Asia/Seoul")), productId);
+        Rank rank = rankingService.getRankInfo(LocalDate.now(ZoneId.of("Asia/Seoul")), productId,"daily");
 
         // 사용자 활동로그(상품 조회)
         productActivityPublisher.productDetail(new ProductActivityPayload.ProductDetailViewed(loginId, productId));

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

@@ -3,7 +3,7 @@
 import com.loopers.domain.order.OrderEvent;
 import com.loopers.domain.payment.PaymentEvent;
 import com.loopers.domain.product.*;
-import com.loopers.infrastructure.message.ProductProducer;
+import com.loopers.infrastructure.message.ProductViewProducer;
 import com.loopers.shared.event.Envelope;
 import com.loopers.infrastructure.message.ProductSkuProducer;
 import lombok.RequiredArgsConstructor;
@@ -20,7 +20,7 @@ public class ProductEventListener {
 
     private final ProductSkuService productSkuService;
     private final ProductService productService;
-    private final ProductProducer productProducer;
+    private final ProductViewProducer productViewProducer;
     private final ProductSkuProducer productSkuProducer;
     private final CacheManager cacheManager;
 
@@ -68,6 +68,6 @@ public void onOrderSuccess(OrderEvent.ReCalStock e) {
     @Async("applicationEventTaskExecutor")
     @EventListener
     public void onDetailViewed(Envelope<ProductActivityPayload.ProductDetailViewed> event) {
-        productProducer.send(String.valueOf(event.payload().productId()),event);
+        productViewProducer.send(String.valueOf(event.payload().productId()),event);
     }
 }

apps/commerce-api/src/main/java/com/loopers/application/rank/RankApplicationService.java

@@ -11,6 +11,7 @@
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -26,10 +27,11 @@ public class RankApplicationService implements RankUsecase {
     @Override
     @Transactional(readOnly = true)
     public List<RankInfo.ProductRank> getProductRanking(RankCommand.ProductRanking command) {
-        LocalDate date = LocalDate.parse(command.date());
+        LocalDate date = LocalDate.parse(command.date(), DateTimeFormatter.BASIC_ISO_DATE);
         int size = command.size();
+        String period = command.period() == null ? "daily" : command.period().toLowerCase();
 
-        List<Rank> rankList = rankingService.getProductRank(date, size);
+        List<Rank> rankList = rankingService.getProductRank(date, size, period);
 
         if (rankList.isEmpty())
             return List.of();

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

@@ -1,9 +1,14 @@
 package com.loopers.domain.product;
 
 public class CatalogMessage {
-    public record LikeChanged(
-            Long productId,
-            String targetType,
-            long likeCount
-    ) {}
+
+    public record LikeAdded(
+            String loginId,
+            Long targetId
+    )  {}
+
+    public record LikeRemoved(
+            String loginId,
+            Long targetId
+    )  {}
 }

apps/commerce-api/src/main/java/com/loopers/domain/rank/Rank.java

@@ -2,9 +2,9 @@
 
 import lombok.*;
 
-import java.nio.charset.StandardCharsets;
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
+import java.time.temporal.WeekFields;
 
 @Getter
 @ToString
@@ -16,7 +16,31 @@ public class Rank {
     private final Long position;
     private final double score;
 
+    private static final DateTimeFormatter D_YYYYMMDD = DateTimeFormatter.BASIC_ISO_DATE;
+
     public static Rank create(Long productId, Long position, double score) {
         return new Rank(productId, position, score);
     }
+
+    public static String buildKey(String periodRaw, LocalDate date,
+                                  String dailyPrefix, String weeklyPrefix, String monthlyPrefix) {
+        final String period = (periodRaw == null ? "daily" : periodRaw.toLowerCase());
+
+        return switch (period) {
+            case "weekly" -> weeklyPrefix + ":" + toYearWeek(date);
+            case "monthly" -> monthlyPrefix + ":" + toYearMonth(date);
+            default -> dailyPrefix  + ":" + date.format(D_YYYYMMDD);
+        };
+    }
+
+    private static String toYearWeek(LocalDate date) {
+        var wf = WeekFields.ISO;
+        int w = date.get(wf.weekOfWeekBasedYear());
+        int y = date.get(wf.weekBasedYear());
+        return "%04dW%02d".formatted(y, w);
+    }
+
+    private static String toYearMonth(LocalDate date) {
+        return DateTimeFormatter.ofPattern("yyyyMM").format(date);
+    }
 }

apps/commerce-api/src/main/java/com/loopers/domain/rank/RankCommand.java

@@ -5,10 +5,11 @@
 public class RankCommand {
     public record ProductRanking(
             String date,
-            int size
+            int size,
+            String period
     ){
-        public static ProductRanking create(LocalDate date, int size) {
-            return new ProductRanking(date.toString(), size);
+        public static ProductRanking create(String date, int size,String period) {
+            return new ProductRanking(date.toString(), size,period);
         }
     }
 }

apps/commerce-api/src/main/java/com/loopers/domain/rank/RankingProperties.java

@@ -4,5 +4,7 @@
 
 @ConfigurationProperties(value = "ranking")
 public record RankingProperties(
-        String keyPrefix
+        String keyPrefix,
+        String weeklyPrefix,
+        String monthlyPrefix
 ){}

apps/commerce-api/src/main/java/com/loopers/domain/rank/RankingService.java

@@ -24,48 +24,49 @@ public class RankingService {
 
     private final RankingProperties props;
 
-    public List<Rank> getProductRank(LocalDate date, int size) {
+    public List<Rank> getProductRank(LocalDate date, int size, String period) {
         if (size <= 0) return List.of();
 
-        final String key = props.keyPrefix() + ":" + date.format(dateTimeFormatter);
+        String key = Rank.buildKey(period, date,
+                props.keyPrefix(),
+                props.weeklyPrefix(),
+                props.monthlyPrefix()
+        );
 
         Set<ZSetOperations.TypedTuple<String>> tuples =
                 stringRedisTemplate.opsForZSet().reverseRangeWithScores(key, 0, Math.max(0, size - 1));
 
-        if (tuples == null || tuples.isEmpty())
-            return List.of();
+        if (tuples == null || tuples.isEmpty()) return List.of();
 
         List<Rank> result = new ArrayList<>(tuples.size());
         int position = 1;
         for (ZSetOperations.TypedTuple<String> t : tuples) {
-
             String productIdStr = t.getValue();
             Double score = t.getScore();
-            if (productIdStr == null)
-                continue;
+            if (productIdStr == null) continue;
 
             long productId = Long.parseLong(productIdStr);
-
-            result.add(new Rank(productId, (long) position++, score == null ? 0d : score));
+            result.add(Rank.create(productId, (long) position++, score == null ? 0d : score));
         }
         return result;
     }
 
-    public Rank getRankInfo(LocalDate date, Long productId) {
+    public Rank getRankInfo(LocalDate date, Long productId, String period) {
         if (productId == null) return null;
 
-        final String key = props.keyPrefix() + ":" + date.format(dateTimeFormatter);
+        String key = Rank.buildKey(period, date,
+                props.keyPrefix(),
+                props.weeklyPrefix(),
+                props.monthlyPrefix()
+        );
         final String member = productId.toString();
 
         Long zeroBased = stringRedisTemplate.opsForZSet().reverseRank(key, member);
-
-        if (zeroBased == null)
-            return null;
+        if (zeroBased == null) return null;
 
         Double score = stringRedisTemplate.opsForZSet().score(key, member);
-
         int position = Math.toIntExact(zeroBased) + 1;
 
-        return new Rank(productId, (long) position, score == null ? 0d : score);
+        return Rank.create(productId, (long) position, score == null ? 0d : score);
     }
 }

apps/commerce-api/src/main/java/com/loopers/infrastructure/message/ProductProducer.java → apps/commerce-api/src/main/java/com/loopers/infrastructure/message/ProductLikeProducer.java

@@ -13,12 +13,12 @@
 
 @Component
 @RequiredArgsConstructor
-public class ProductProducer {
+public class ProductLikeProducer {
 
     private final KafkaTemplate<Object, Object> kafkaTemplate;
     private final RetryTemplate retryTemplate;
 
-    @Value("${kafka.topic.catalog-events}")
+    @Value("${kafka.topic.catalog-like-events}")
     private String topic;
 
     @Value("${kafka.topic.producer-dlq}")

apps/commerce-api/src/main/java/com/loopers/infrastructure/message/ProductSkuProducer.java

@@ -20,7 +20,7 @@ public class ProductSkuProducer {
     private final KafkaTemplate<Object, Object> kafkaTemplate;
     private final RetryTemplate retryTemplate;
 
-    @Value("${kafka.topic.sku-events}")
+    @Value("${kafka.topic.catalog-product-events}")
     private String topic;
 
     @Value("${kafka.topic.producer-dlq}")

apps/commerce-api/src/main/java/com/loopers/infrastructure/message/ProductViewProducer.java

@@ -0,0 +1,54 @@
+package com.loopers.infrastructure.message;
+
+import com.loopers.shared.event.Envelope;
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+
+@Component
+@RequiredArgsConstructor
+public class ProductViewProducer {
+
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+    private final RetryTemplate retryTemplate;
+
+    @Value("${kafka.topic.catalog-view-events}")
+    private String topic;
+
+    @Value("${kafka.topic.producer-dlq}")
+    private String dlq;
+
+    public <T> void send(String key, Envelope<T> envelope)  {
+        String eventType = envelope.payload().getClass().getSimpleName();
+
+        ProducerRecord<Object, Object> record = new ProducerRecord<>(topic, key, envelope);
+        addHeaders(record, envelope.id(), eventType, envelope.at().toString(), envelope.actorId());
+
+        try {
+            retryTemplate.execute(ctx -> {
+                kafkaTemplate.send(record).get();
+                return null;
+            });
+        } catch (Exception e) {
+            ProducerRecord<Object, Object> dead = new ProducerRecord<>(dlq, key, envelope);
+            dead.headers().add(new RecordHeader("source_topic", topic.getBytes(StandardCharsets.UTF_8)));
+            dead.headers().add(new RecordHeader("event_type",  eventType.getBytes(StandardCharsets.UTF_8)));
+            kafkaTemplate.send(dead);
+        }
+    }
+
+    private void addHeaders(ProducerRecord<Object, Object> record, String id, String type, String at, String actor) {
+        record.headers()
+                .add(new RecordHeader("event_id", id.getBytes(StandardCharsets.UTF_8)))
+                .add(new RecordHeader("event_type", type.getBytes(StandardCharsets.UTF_8)))
+                .add(new RecordHeader("event_version", "1".getBytes(StandardCharsets.UTF_8)))
+                .add(new RecordHeader("occurred_at", at.getBytes(StandardCharsets.UTF_8)))
+                .add(new RecordHeader("actor_id", actor.getBytes(StandardCharsets.UTF_8)));
+    }
+}

apps/commerce-api/src/main/java/com/loopers/interfaces/api/controller/rank/RankController.java

@@ -5,7 +5,6 @@
 import com.loopers.domain.rank.RankInfo;
 import com.loopers.interfaces.api.ApiResponse;
 import lombok.RequiredArgsConstructor;
-import org.springframework.format.annotation.DateTimeFormat;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -24,10 +23,11 @@ public class RankController implements RankV1ApiSpec{
     @Override
     @GetMapping
     public ApiResponse<RankV1Response.ProductRankList> rank(
-            @RequestParam @DateTimeFormat(pattern = "yyyyMMdd") LocalDate date,
-            @RequestParam(defaultValue = "20") int size
+            @RequestParam String date,
+            @RequestParam(defaultValue = "20") int size,
+            @RequestParam(defaultValue = "daily") String period
     ) {
-        List<RankInfo.ProductRank> infos = rankUsecase.getProductRanking(RankCommand.ProductRanking.create(date,size));
+        List<RankInfo.ProductRank> infos = rankUsecase.getProductRanking(RankCommand.ProductRanking.create(date,size,period));
         return ApiResponse.success(RankV1Response.ProductRankList.from(infos));
     }
 }