feature: 메트릭 이벤트 컨슈머 확장 및 Kafka 기반 지표 관리 강화

- `LikeEventConsumer`를 `MetricsEventConsumer`로 변경하고 상품 지표(좋아요, 조회수, 판매량) 처리 지원.
- `ProductMetricsService` 확장: 조회수(`ProductViewEvent`), 판매량(`SalesCountEvent`) 메트릭 처리 로직 추가.
- `ProductMetrics` 엔티티에 조회수와 판매량 속성 및 관련 업데이트 메서드 추가.

Author: yeonjiyeon

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

@@ -15,7 +15,9 @@ public class ProductMetrics {
   @Id
   private Long productId;
 
-  private int likeCount;
+  private int likeCount = 0;
+  private int viewCount = 0;
+  private int salesCount = 0;
 
   private LocalDateTime updatedAt;
 
@@ -30,4 +32,14 @@ public void updateLikeCount(int newCount, LocalDateTime eventTime) {
     this.likeCount = newCount;
     this.updatedAt = eventTime;
   }
+
+  public void incrementViewCount() {
+    this.viewCount += 1;
+    this.updatedAt = LocalDateTime.now();
+  }
+
+  public void addSalesCount(int quantity) {
+    this.salesCount += quantity;
+    this.updatedAt = LocalDateTime.now();
+  }
 }

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

@@ -1,7 +1,9 @@
 package com.loopers.domain.metrics;
 
 import com.loopers.domain.event.EventHandled;
-import com.loopers.event.LikeKafkaEvent;
+import com.loopers.event.LikeCountEvent;
+import com.loopers.event.ProductViewEvent;
+import com.loopers.event.SalesCountEvent;
 import com.loopers.infrastructure.EventHandledRepository;
 import com.loopers.infrastructure.ProductMetricsRepository;
 import java.time.LocalDateTime;
@@ -17,17 +19,49 @@ public class ProductMetricsService {
   private final EventHandledRepository eventHandledRepository;
 
   @Transactional
-  public void processLikeEvent(LikeKafkaEvent event) {
-    if (eventHandledRepository.existsById(event.eventId())) {
-      return;
-    }
+  public void processLikeCountEvent(LikeCountEvent event) {
+    if (isAlreadyHandled(event.eventId())) return;
 
-    ProductMetrics metrics = metricsRepository.findById(event.productId())
-        .orElseGet(() -> new ProductMetrics(event.productId()));
+    ProductMetrics metrics = getOrCreateMetrics(event.productId());
 
     metrics.updateLikeCount(event.currentLikeCount(), LocalDateTime.now());
-    metricsRepository.save(metrics);
 
-    eventHandledRepository.save(new EventHandled(event.eventId()));
+    completeProcess(event.eventId(), metrics);
+  }
+
+  @Transactional
+  public void processProductViewEvent(ProductViewEvent event) {
+    if (isAlreadyHandled(event.eventId())) return;
+
+    ProductMetrics metrics = getOrCreateMetrics(event.productId());
+
+    metrics.incrementViewCount();
+
+    completeProcess(event.eventId(), metrics);
+  }
+
+  @Transactional
+  public void processSalesCountEvent(SalesCountEvent event) {
+    if (isAlreadyHandled(event.eventId())) return;
+
+    ProductMetrics metrics = getOrCreateMetrics(event.productId());
+
+    metrics.addSalesCount(event.quantity());
+
+    completeProcess(event.eventId(), metrics);
+  }
+
+  private boolean isAlreadyHandled(String eventId) {
+    return eventHandledRepository.existsById(eventId);
+  }
+
+  private ProductMetrics getOrCreateMetrics(Long productId) {
+    return metricsRepository.findById(productId)
+        .orElseGet(() -> new ProductMetrics(productId));
+  }
+
+  private void completeProcess(String eventId, ProductMetrics metrics) {
+    metricsRepository.save(metrics);
+    eventHandledRepository.save(new EventHandled(eventId));
   }
 }

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

@@ -0,0 +1,59 @@
+package com.loopers.interfaces.consumer;
+
+import com.loopers.domain.metrics.ProductMetricsService;
+import com.loopers.event.LikeCountEvent;
+import com.loopers.event.ProductViewEvent;
+import com.loopers.event.SalesCountEvent;
+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;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class MetricsEventConsumer {
+
+  private final ProductMetricsService metricsService;
+
+  @KafkaListener(
+      topics = "catalog-events",
+      groupId = "metrics-group"
+  )
+  public void consumeLikeCount(ConsumerRecord<String, LikeCountEvent> record, Acknowledgment ack) {
+    try {
+      metricsService.processLikeCountEvent(record.value());
+      ack.acknowledge();
+    } catch (Exception e) {
+      log.error("좋아요 메트릭 처리 실패: {}", e.getMessage());
+    }
+  }
+
+  @KafkaListener(
+      topics = "catalog-events",
+      groupId = "metrics-group"
+  )
+  public void consumeProductView(ProductViewEvent event, Acknowledgment ack) {
+    try {
+      metricsService.processProductViewEvent(event);
+      ack.acknowledge();
+    } catch (Exception e) {
+      log.error("조회수 메트릭 처리 실패: {}", event.eventId(), e);
+    }
+  }
+
+  @KafkaListener(
+      topics = "catalog-events",
+      groupId = "metrics-group"
+  )
+  public void consumeSalesCount(SalesCountEvent event, Acknowledgment ack) {
+    try {
+      metricsService.processSalesCountEvent(event);
+      ack.acknowledge();
+    } catch (Exception e) {
+      log.error("판매량 메트릭 처리 실패: {}", event.eventId(), e);
+    }
+  }
+}