feat: 상품 집계 정보 Redis ZSET에 적재

Author: Kimjipang

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

@@ -9,45 +9,81 @@
 import com.loopers.domain.ProductMetricRepository;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ZSetOperations;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 @Component
 @RequiredArgsConstructor
 public class KafkaOutboxConsumer {
     private final ProductMetricRepository productMetricRepository;
     private final ObjectMapper objectMapper;
+    private final StringRedisTemplate redisTemplate;
+
+    private static final ZoneId KST = ZoneId.of("Asia/Seoul");
+    private static final DateTimeFormatter YYYYMMDD = DateTimeFormatter.BASIC_ISO_DATE;
 
     @KafkaListener(
-            topics = {"product-viewed", "product-liked"},
+            topics = {"product-viewed", "product-liked", "product-sales"},
             containerFactory = KafkaConfig.BATCH_LISTENER
     )
     @Transactional
     public void productViewedListener(
             List<ConsumerRecord<String, String>> messages,
             Acknowledgment acknowledgment
     ) throws JsonProcessingException {
+
+        // 1) 배치 내 productId별 점수 누적
+        Map<Long, Double> scoreDelta = new HashMap<>();
+
         for (var record : messages) {
             OutboxEvent value = objectMapper.readValue(record.value(), OutboxEvent.class);
             Long productId = value.aggregateId();
-            String eventType = value.eventType();
+            ProductEventType eventType = ProductEventType.valueOf(value.eventType());
+
+            scoreDelta.merge(productId, weight(eventType), Double::sum);
 
             ProductMetric productMetric = productMetricRepository.findByProductId(productId);
 
             if (productMetric == null) {
-                ProductMetric newProductMetric = ProductMetric.of(productId, ProductEventType.valueOf(eventType));
+                ProductMetric newProductMetric = ProductMetric.of(productId, eventType);
 
                 productMetricRepository.save(newProductMetric);
             }
             else {
-                productMetric.increaseProductMetric(ProductEventType.valueOf(eventType));
+                productMetric.increaseProductMetric(eventType);
             }
+        }
 
+        String key = "ranking:all:" + LocalDate.now(KST).format(YYYYMMDD);
+        ZSetOperations<String, String> zset = redisTemplate.opsForZSet();
+
+        for (var e : scoreDelta.entrySet()) {
+            zset.incrementScore(key, String.valueOf(e.getKey()), e.getValue()); // ZINCRBY
         }
+
+        // 3) 일간 키는 TTL 걸어두는 게 운영에 유리 (예: 8일 보관)
+        redisTemplate.expire(key, Duration.ofDays(8));
+
         acknowledgment.acknowledge();
     }
+
+    double weight(ProductEventType eventType) {
+        return switch (eventType) {
+            case PRODUCT_VIEWED -> 0.1;
+            case PRODUCT_LIKED -> 0.3;
+            case PRODUCT_SALES -> 0.6;
+        };
+    }
 }