[volume-9] Product Ranking with Redis

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

@@ -0,0 +1,17 @@
+package com.loopers.application.ranking;
+
+import com.loopers.application.product.ProductLikeSummary;
+import com.loopers.ranking.RankingEntry;
+
+import java.time.LocalDate;
+import java.util.List;
+
+public record RankingQueryResponse(
+        LocalDate date,
+        List<RankingEntry> rankingEntries,
+        List<ProductLikeSummary> productLikeSummary
+) {
+    public static RankingQueryResponse of(LocalDate date, List<RankingEntry> rankingEntries, List<ProductLikeSummary> productLikeSummary) {
+        return new RankingQueryResponse(date, rankingEntries, productLikeSummary);
+    }
+}

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

@@ -0,0 +1,82 @@
+package com.loopers.application.ranking;
+
+import com.loopers.application.like.event.ProductLikeEvent;
+import com.loopers.application.product.ProductLikeSummary;
+import com.loopers.application.product.ProductQueryService;
+import com.loopers.domain.product.ProductSortType;
+import com.loopers.ranking.DailyRankingResponse;
+import com.loopers.ranking.RankingEntry;
+import com.loopers.ranking.RankingZSetRepository;
+import com.loopers.support.error.CoreException;
+import com.loopers.support.error.ErrorType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.OptionalDouble;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class RankingQueryService {
+    private final RankingZSetRepository rankingZSetRepository;
+    private final ProductQueryService productQueryService;
+
+    @Transactional(readOnly = true)
+    public RankingQueryResponse getDailyPopularProducts(String date, int size) {
+        LocalDate target = initLocalDate(date);
+
+        int limit = (size <= 0) ? 20 : Math.min(size, 100);
+
+        List<RankingEntry> rankingEntries = rankingZSetRepository.findTopDailyAllByLimit(target, limit);
+        List<ProductLikeSummary> productLikeSummaries = findProductSummaryFrom(rankingEntries);
+
+        return new RankingQueryResponse(
+                target,
+                rankingEntries,
+                productLikeSummaries
+        );
+    }
+
+    @Transactional(readOnly = true)
+    public OptionalDouble getDailyRankingScore(Long productId) {
+        LocalDate now = LocalDate.now(ZoneId.systemDefault());
+        return rankingZSetRepository.findDailyRanking(now, productId);
+    }
+
+    private boolean hasValidDate(String date) {
+        return date == null || date.isBlank();
+    }
+
+    private LocalDate initLocalDate(String date) {
+         return (hasValidDate(date))
+                ? LocalDate.now(ZoneId.systemDefault())
+                : LocalDate.parse(date, DateTimeFormatter.BASIC_ISO_DATE);
+
+    }
+
+    private List<ProductLikeSummary> findProductSummaryFrom(List<RankingEntry> rankingEntries) {
+        List<ProductLikeSummary> result = new ArrayList<>();
+
+        for(RankingEntry rankingEntry : rankingEntries) {
+            ProductLikeSummary summary;
+            try {
+                summary = productQueryService.getProductLikeSummary(rankingEntry.productId());
+            } catch (CoreException e) {
+                if(e.getErrorType() == ErrorType.NOT_FOUND) {
+                    log.error("Could not find product like summary for {}", rankingEntry.productId());
+                }
+                summary =  null;
+            }
+            result.add(summary);
+        }
+        return result;
+    }
+}

apps/commerce-api/src/main/java/com/loopers/interfaces/api/product/ProductV1Controller.java

@@ -2,7 +2,9 @@
 
 import com.loopers.application.product.ProductLikeSummary;
 import com.loopers.application.product.ProductQueryService;
+import com.loopers.application.ranking.RankingQueryService;
 import com.loopers.domain.product.ProductSortType;
+import com.loopers.ranking.RankingEntry;
 import com.loopers.support.tracking.general.UserActionType;
 import com.loopers.interfaces.api.ApiResponse;
 import com.loopers.support.tracking.annotation.TrackUserAction;
@@ -12,11 +14,14 @@
 import org.springframework.data.web.PageableDefault;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.OptionalDouble;
+
 @RequiredArgsConstructor
 @RestController
 @RequestMapping("/api/v1/products")
 public class ProductV1Controller implements ProductV1ApiSpec{
     private final ProductQueryService productQueryService;
+    private final RankingQueryService rankingQueryService;
 
     @Override
     @GetMapping
@@ -41,7 +46,7 @@ public ApiResponse<ProductV1Dto.ProductListResponse<ProductLikeSummary>> getProd
     )
     public ApiResponse<ProductV1Dto.ProductDetailResponse<ProductLikeSummary>> getProductDetail(@PathVariable("productId") Long productId) {
         ProductLikeSummary productLikeSummary = productQueryService.getProductLikeSummary(productId);
-
-        return ApiResponse.success(ProductV1Dto.ProductDetailResponse.of(productLikeSummary));
+        OptionalDouble rankingEntry = rankingQueryService.getDailyRankingScore(productId);
+        return ApiResponse.success(ProductV1Dto.ProductDetailResponse.of(productLikeSummary, rankingEntry));
     }
 }

apps/commerce-api/src/main/java/com/loopers/interfaces/api/product/ProductV1Dto.java

@@ -1,8 +1,10 @@
 package com.loopers.interfaces.api.product;
 
+import com.loopers.ranking.RankingEntry;
 import org.springframework.data.domain.Page;
 
 import java.util.List;
+import java.util.OptionalDouble;
 
 public class ProductV1Dto {
 
@@ -28,10 +30,15 @@ public static <T> ProductListResponse<T> of(Page<T> page, List<T> content) {
     }
 
     public record ProductDetailResponse<T>(
-            T content
+            T content,
+            OptionalDouble rankingScore
     ){
         static <T> ProductDetailResponse<T> of(T content) {
-            return new ProductDetailResponse<>(content);
+            return new ProductDetailResponse<>(content, null);
+        }
+
+        static <T> ProductDetailResponse<T> of(T content, OptionalDouble rankingScore) {
+            return new ProductDetailResponse<>(content, rankingScore);
         }
     }
 }

apps/commerce-api/src/main/java/com/loopers/interfaces/api/ranking/RankingV1Controller.java

@@ -0,0 +1,26 @@
+package com.loopers.interfaces.api.ranking;
+
+import com.loopers.application.ranking.RankingQueryResponse;
+import com.loopers.application.ranking.RankingQueryService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RequiredArgsConstructor
+@RestController
+@RequestMapping("/api/v1/ranking")
+public class RankingV1Controller {
+
+    private final RankingQueryService rankingQueryService;
+
+    @GetMapping
+    public RankingQueryResponse getDailyRanking(
+            @RequestParam(required = false, name = "date") String date,
+            @RequestParam(defaultValue = "20", name = "size") int size
+    ) {
+        return rankingQueryService.getDailyPopularProducts(date, size);
+    }
+
+}

apps/commerce-streamer/src/main/java/com/loopers/application/ranking/RankingAggregationService.java

@@ -0,0 +1,77 @@
+package com.loopers.application.ranking;
+
+import com.loopers.ranking.RankingKey;
+import lombok.RequiredArgsConstructor;
+
+import org.springframework.data.redis.connection.RedisZSetCommands;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.time.*;
+
+@Service
+@RequiredArgsConstructor
+public class RankingAggregationService {
+    private static final double LIKE_WEIGHT = 0.2d;
+    private static final Duration TTL = Duration.ofDays(2);
+    private static final double CARRY_OVER_WEIGHT = 0.1d;
+
+    private final StringRedisTemplate redisTemplate;
+
+    public void applyLike(long productId, Instant occurredAt) {
+        LocalDate day = occurredAt.atZone(ZoneId.systemDefault()).toLocalDate();
+        String key = RankingKey.dailyAll(day);
+        String member = String.valueOf(productId);
+
+        // Score = Weight * 1
+        redisTemplate.opsForZSet().incrementScore(key, member, LIKE_WEIGHT);
+
+        setTTLOnlyOnce(key);
+    }
+
+    /**
+     * 일별 랭킹에 대한 전일자 Carry Over 스케줄러
+     * - 매일 23:30분 수행합니다.
+     */
+    @Scheduled(cron = "0 30 23 * * *")
+    public void carryOverDailyRanking() {
+        LocalDate today = LocalDate.now(ZoneId.systemDefault());
+        LocalDate tomorrow = today.plusDays(1);
+
+        String sourceKey = RankingKey.dailyAll(today);
+        String targetKey = RankingKey.dailyAll(tomorrow);
+
+        carryOver(sourceKey, targetKey, CARRY_OVER_WEIGHT);
+    }
+
+    private void carryOver(String sourceKey, String targetKey, double weight) {
+        redisTemplate.execute((RedisCallback<Long>) connection -> {
+            byte[] target = redisTemplate.getStringSerializer().serialize(targetKey);
+            byte[] source = redisTemplate.getStringSerializer().serialize(sourceKey);
+            // ZUNIONSTORE dest 1 src WEIGHTS 0.1 AGGREGATE SUM
+            Long result = (Long) connection.execute(
+                    "ZUNIONSTORE",
+                    target,
+                    "1".getBytes(),
+                    source,
+                    "WEIGHTS".getBytes(),
+                    String.valueOf(weight).getBytes(),
+                    "AGGREGATE".getBytes(),
+                    "SUM".getBytes()
+            );
+            // 순서 주의할 것
+            // ZUNIONSTORE target ... → 키 재생성 되므로 이전 TTL 소멸됨
+            connection.expire(target, TTL.getSeconds());
+            return result;
+        });
+    }
+
+    private void setTTLOnlyOnce(String key) {
+        Boolean exists = redisTemplate.hasKey(key);
+        if (Boolean.FALSE.equals(exists)) {
+            redisTemplate.expire(key, TTL);
+        }
+    }
+}

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

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.loopers.application.idempotency.EventHandledService;
 import com.loopers.application.metrics.MetricsAggregationService;
+import com.loopers.application.ranking.RankingAggregationService;
 import com.loopers.contract.event.ProductLikeEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -22,6 +23,7 @@ public class ProductLikeEventConsumer {
     private final ObjectMapper objectMapper;
     private final EventHandledService eventHandledService;
     private final MetricsAggregationService metricsAggregationService;
+    private final RankingAggregationService rankingAggregationService;
 
     @KafkaListener(topics = "product-like-events", groupId = "${spring.kafka.consumer.group-id}")
     public void consume( @Header(KafkaHeaders.RECEIVED_KEY) String key,
@@ -42,6 +44,7 @@ public void consume( @Header(KafkaHeaders.RECEIVED_KEY) String key,
         }
 
         metricsAggregationService.handleProductLiked(event.productId());
+        rankingAggregationService.applyLike(event.productId(), event.occurredAt());
 
         ack.acknowledge();
     }

apps/commerce-streamer/src/test/java/com/loopers/application/idempotency/EventHandledServiceTest.java

@@ -3,28 +3,52 @@
 import com.loopers.domain.ProductLikeMetricsModel;
 import com.loopers.infrastructure.EventHandleRepository;
 import com.loopers.infrastructure.ProductLikeMetricsRepository;
+import com.loopers.testcontainers.KafkaTestContainersConfig;
+import com.loopers.utils.KafkaCleanUp;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.testcontainers.junit.jupiter.Testcontainers;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.util.Map;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
-@Testcontainers
+@ActiveProfiles("local")
+@Import(KafkaTestContainersConfig.class)
 @SpringBootTest
 class EventHandledServiceTest {
+
+    @Autowired
+    private KafkaCleanUp kafkaCleanUp;
+
     @Autowired
     KafkaTemplate<Object,Object> kafkaTemplate;
     @Autowired
     ProductLikeMetricsRepository metricsRepo;
     @Autowired
     EventHandleRepository handledRepo;
 
+    @Autowired
+    private KafkaProperties kafkaProperties;
+
+    @BeforeEach
+    void setUp() {
+        // @BeforeEach는 이미 SpringBootTest 컨텍스트 + KafkaListener 컨테이너가 다 올라간 뒤에 실행
+        // kafkaCleanUp.resetAllTestTopics();
+        kafkaCleanUp.resetAllConsumerGroups();
+    }
+
     @Test
     @DisplayName("멱등성 테스트")
     void duplicate_message_should_be_applied_once() throws Exception {
+
         long productId = 1L;
         ProductLikeMetricsModel metrics = metricsRepo.findById(productId)
                 .orElseGet(() -> metricsRepo.save(ProductLikeMetricsModel.of(productId)));
@@ -42,6 +66,9 @@ void duplicate_message_should_be_applied_once() throws Exception {
         // TODO - 카프카 브로커/컨슈머 테스트환경에서 설정필요
         Thread.sleep(1500);
 
+        Map<String, Object> consumerProps = kafkaProperties.buildConsumerProperties();
+        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "${spring.kafka.consumer.group-id}");
+
 
         // then
         // like 카운트는 1번만 증가