feat: consumer 로직 작성

Author: Kimjipang

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

@@ -34,8 +34,31 @@ public ProductMetric(Long productId, Long viewCount, Long likeCount, Long salesV
         this.eventType = eventType;
     }
 
-    public static ProductMetric of(Long productId, Long viewCount, Long likeCount, Long salesVolume, ProductEventType eventType) {
-        return new ProductMetric(productId, viewCount, likeCount, salesVolume, eventType);
+    /*
+    - [ ] 리팩토링 예정
+     */
+    public static ProductMetric of(Long productId, ProductEventType eventType) {
+        if (eventType == null) {
+            throw new IllegalArgumentException("정의되지 않은 event type입니다.");
+        }
+
+        return switch (eventType) {
+            case PRODUCT_VIEWED -> ofProductViewed(productId);
+            case PRODUCT_LIKED -> ofProductLiked(productId);
+            case PRODUCT_SALES -> ofProductSales(productId);
+        };
+    }
+
+    public static ProductMetric ofProductViewed(Long productId) {
+        return new ProductMetric(productId, 1L, 0L, 0L, ProductEventType.PRODUCT_VIEWED);
+    }
+
+    public static ProductMetric ofProductLiked(Long productId) {
+        return new ProductMetric(productId, 0L, 1L, 0L, ProductEventType.PRODUCT_LIKED);
+    }
+
+    public static ProductMetric ofProductSales(Long productId) {
+        return new ProductMetric(productId, 0L, 0L, 1L, ProductEventType.PRODUCT_SALES);
     }
 
     public void increaseProductMetric(ProductEventType eventType) {

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

@@ -23,11 +23,11 @@ public class KafkaOutboxConsumer {
     private final ObjectMapper objectMapper;
 
     @KafkaListener(
-            topics = {"product-viewed"},
+            topics = {"product-viewed", "product-liked"},
             containerFactory = KafkaConfig.BATCH_LISTENER
     )
     @Transactional
-    public void demoListener(
+    public void productViewedListener(
             List<ConsumerRecord<String, String>> messages,
             Acknowledgment acknowledgment
     ) throws JsonProcessingException {
@@ -39,15 +39,9 @@ public void demoListener(
             ProductMetric productMetric = productMetricRepository.findByProductId(productId);
 
             if (productMetric == null) {
-                productMetricRepository.save(
-                        ProductMetric.of(
-                                productId,
-                                0L,
-                                0L,
-                                0L,
-                                ProductEventType.valueOf(eventType)
-                        )
-                );
+                ProductMetric newProductMetric = ProductMetric.of(productId, ProductEventType.valueOf(eventType));
+
+                productMetricRepository.save(newProductMetric);
             }
             else {
                 productMetric.increaseProductMetric(ProductEventType.valueOf(eventType));