feature: 지표 메트릭 개선 및 RankingService 추가

- 이벤트 타임스탬프 필드를 `LocalDateTime`으로 변경하고 이벤트 생성 로직 업데이트
- `RankingService` 추가로 상품 점수 관리 로직 구현 (조회수, 좋아요, 판매량 가중치 적용)
- MetricsEventConsumer 확장: 랭킹 점수 계산 및 Redis 업데이트 지원
- Kafka 리스너 및 배치 처리 로직 개선, 이벤트별 점수 업데이트 처리 추가
- OutboxHandler 수정: 이벤트 타입 변경 및 카탈로그 이벤트 발행 정리

Author: yeonjiyeon

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

@@ -1,6 +1,7 @@
 package com.loopers.application.order.event;
 
 import com.loopers.domain.event.OutboxService;
+import com.loopers.event.ProductStockEvent;
 import lombok.RequiredArgsConstructor;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
@@ -14,8 +15,8 @@ public class OrderEventOutboxHandler {
   private final OutboxService outboxService;
 
   @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
-  public void handle(OrderCreatedEvent event) {
-    kafkaTemplate.send("order-events", String.valueOf(event.orderId()), event)
+  public void handle(ProductStockEvent event) {
+    kafkaTemplate.send("catalog-events", String.valueOf(event.productId()), event)
         .whenComplete((result, ex) -> {
           if (ex == null) {
             outboxService.markPublished(event.eventId());

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

@@ -2,8 +2,8 @@
 
 import com.loopers.application.event.FailedEventStore;
 import com.loopers.application.like.event.LikeCreatedEvent;
-import com.loopers.event.LikeCountEvent;
 import com.loopers.domain.product.ProductService;
+import com.loopers.event.LikeCountEvent;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.orm.ObjectOptimisticLockingFailureException;
@@ -30,7 +30,7 @@ public void handleLikeCreatedEvent(LikeCreatedEvent event) {
     try {
       int updatedLikeCount = performAggregation(event);
 
-      eventPublisher.publishEvent(new LikeCountEvent(
+      eventPublisher.publishEvent(LikeCountEvent.of(
           event.eventId(),
           event.productId(),
           updatedLikeCount

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

@@ -0,0 +1,38 @@
+package com.loopers.domain.rank;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import lombok.RequiredArgsConstructor;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class RankingService {
+
+  private final RedisTemplate<String, String> redisTemplate;
+
+  private static final String KEY_PREFIX = "ranking:all:";
+  private static final double VIEW_WEIGHT = 0.1;
+  private static final double LIKE_WEIGHT = 0.2;
+  private static final double ORDER_WEIGHT = 0.6;
+
+  public void addScore(Long productId, double baseScore, double weight, LocalDateTime dateTime) {
+    String dateKey = KEY_PREFIX + dateTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+    double finalScore = baseScore * weight;
+
+    redisTemplate.opsForZSet().incrementScore(dateKey, productId.toString(), finalScore);
+    redisTemplate.expire(dateKey, 2, TimeUnit.DAYS);
+  }
+
+  public void addOrderScoresBatch(Map<String, Map<Long, Double>> updates) {
+    updates.forEach((dateKey, productScores) -> {
+      productScores.forEach((productId, score) -> {
+        redisTemplate.opsForZSet().incrementScore(dateKey, productId.toString(), score);
+      });
+      redisTemplate.expire(dateKey, 2, TimeUnit.DAYS);
+    });
+  }
+}

apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/MetricsEventConsumer.java

@@ -5,9 +5,12 @@
 import com.loopers.event.LikeCountEvent;
 import com.loopers.event.ProductStockEvent;
 import com.loopers.event.ProductViewEvent;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Component;
@@ -24,9 +27,10 @@ public class MetricsEventConsumer {
       topics = "catalog-events",
       groupId = "metrics-group"
   )
-  public void consumeLikeCount(ConsumerRecord<String, LikeCountEvent> record, Acknowledgment ack) {
+  public void consumeLikeCount(LikeCountEvent event, Acknowledgment ack) {
     try {
-      metricsService.processLikeCountEvent(record.value());
+      metricsService.processLikeCountEvent(event);
+      rankingService.addScore(event.productId(), 1.0, 0.2, event.createdAt());
       ack.acknowledge();
     } catch (Exception e) {
       log.error("좋아요 메트릭 처리 실패: {}", e.getMessage());
@@ -40,6 +44,7 @@ public void consumeLikeCount(ConsumerRecord<String, LikeCountEvent> record, Ackn
   public void consumeProductView(ProductViewEvent event, Acknowledgment ack) {
     try {
       metricsService.processProductViewEvent(event);
+      rankingService.addScore(event.productId(), 1.0, 0.1, event.createdAt());
       ack.acknowledge();
     } catch (Exception e) {
       log.error("조회수 메트릭 처리 실패: {}", event.eventId(), e);
@@ -48,14 +53,27 @@ public void consumeProductView(ProductViewEvent event, Acknowledgment ack) {
 
   @KafkaListener(
       topics = "catalog-events",
-      groupId = "metrics-group"
+      groupId = "metrics-group",
+      containerFactory = "batchFactory"
   )
-  public void consumeSalesCount(ProductStockEvent event, Acknowledgment ack) {
+  public void consumeSalesCount(List<ProductStockEvent> events, Acknowledgment ack) {
     try {
-      metricsService.processSalesCountEvent(event);
+      Map<String, Map<Long, Double>> rankingUpdates = new HashMap<>();
+
+      for (ProductStockEvent event : events) {
+        metricsService.processSalesCountEvent(event);
+
+        String dateKey = "ranking:all:" + event.createdAt().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
+        double orderScore = (event.price() * event.sellQuantity()) * 0.6;
+
+        rankingUpdates.computeIfAbsent(dateKey, k -> new HashMap<>())
+            .merge(event.productId(), orderScore, Double::sum);
+      }
+
+      rankingService.addOrderScoresBatch(rankingUpdates);
       ack.acknowledge();
     } catch (Exception e) {
-      log.error("판매량 메트릭 처리 실패: {}", event.eventId(), e);
+      log.error("판매량 메트릭 처리 실패", e);
     }
   }
 }

modules/kafka/src/main/java/com/loopers/event/LikeCountEvent.java

@@ -1,9 +1,20 @@
 package com.loopers.event;
 
+import java.time.LocalDateTime;
+
 public record LikeCountEvent(
-   String eventId,
+    String eventId,
     Long productId,
-    int currentLikeCount
+    int currentLikeCount,
+    LocalDateTime createdAt
 ) {
 
+  public static LikeCountEvent of(String eventId, Long productId, int currentLikeCount) {
+    return new LikeCountEvent(
+        eventId,
+        productId,
+        currentLikeCount,
+        LocalDateTime.now()
+    );
+  }
 }

modules/kafka/src/main/java/com/loopers/event/ProductStockEvent.java

@@ -1,5 +1,6 @@
 package com.loopers.event;
 
+import java.time.LocalDateTime;
 import java.util.UUID;
 
 public record ProductStockEvent(
@@ -8,7 +9,7 @@ public record ProductStockEvent(
     int sellQuantity,
     int currentStock,
     long price,
-    long timestamp
+    LocalDateTime createdAt
 ) {
   public static ProductStockEvent of(Long productId, int sellQuantity, int currentStock, long price) {
     return new ProductStockEvent(
@@ -17,7 +18,7 @@ public static ProductStockEvent of(Long productId, int sellQuantity, int current
         sellQuantity,
         currentStock,
         price,
-        System.currentTimeMillis()
+        LocalDateTime.now()
     );
   }
 }

modules/kafka/src/main/java/com/loopers/event/ProductViewEvent.java

@@ -1,18 +1,19 @@
 package com.loopers.event;
 
+import java.time.LocalDateTime;
 import java.util.UUID;
 
 public record ProductViewEvent(
     String eventId,
     Long productId,
-    long timestamp
+    LocalDateTime createdAt
 ) {
 
   public static ProductViewEvent from(Long productId) {
     return new ProductViewEvent(
         UUID.randomUUID().toString(),
         productId,
-        System.currentTimeMillis()
+        LocalDateTime.now()
     );
   }
 }