feat: kafka consumer 정의

Author: Kimjipang

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

@@ -0,0 +1,17 @@
+package com.loopers.domain;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+import java.math.BigDecimal;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public record OutboxEvent(
+        Long id,
+        String aggregateType,
+        Long aggregateId,
+        String eventType,
+        String status,
+        BigDecimal createdAt,
+        BigDecimal updatedAt
+) {
+}

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

@@ -0,0 +1,7 @@
+package com.loopers.domain;
+
+public enum ProductEventType {
+    PRODUCT_VIEWED,
+    PRODUCT_LIKED,
+    PRODUCT_SALES
+}

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

@@ -0,0 +1,61 @@
+package com.loopers.domain;
+
+import jakarta.persistence.*;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Table(name = "product_metrics")
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@Getter
+public class ProductMetric extends BaseEntity {
+    @Column(name = "ref_product_id", nullable = false)
+    private Long productId;
+
+    @Column(name = "view_count", nullable = false)
+    private Long viewCount;
+
+    @Column(name = "like_count", nullable = false)
+    private Long likeCount;
+
+    @Column(name = "sales_volume", nullable = false)
+    private Long salesVolume;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "event_type", nullable = false)
+    private ProductEventType eventType;
+
+    public ProductMetric(Long productId, Long viewCount, Long likeCount, Long salesVolume, ProductEventType eventType) {
+        this.productId = productId;
+        this.viewCount = viewCount;
+        this.likeCount = likeCount;
+        this.salesVolume = salesVolume;
+        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 void increaseProductMetric(ProductEventType eventType) {
+        switch (eventType) {
+            case PRODUCT_VIEWED -> increaseViewCount();
+            case PRODUCT_LIKED -> increaseLikeCount();
+            case PRODUCT_SALES -> increaseSalesVolume();
+        }
+    }
+
+    public void increaseViewCount() {
+        this.viewCount++;
+    }
+
+    public void increaseLikeCount() {
+        this.likeCount++;
+    }
+
+    public void increaseSalesVolume() {
+        this.salesVolume++;
+    }
+
+}

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

@@ -0,0 +1,8 @@
+package com.loopers.domain;
+
+
+public interface ProductMetricRepository {
+    ProductMetric findByProductId(Long productId);
+
+    ProductMetric save(ProductMetric of);
+}

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/ProductMetricJpaRepository.java

@@ -0,0 +1,9 @@
+package com.loopers.infrastructure;
+
+import com.loopers.domain.ProductMetric;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+
+public interface ProductMetricJpaRepository extends JpaRepository<ProductMetric, Long> {
+    ProductMetric findByProductId(Long productId);
+}

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/ProductMetricRepositoryImpl.java

@@ -0,0 +1,23 @@
+package com.loopers.infrastructure;
+
+import com.loopers.domain.ProductMetric;
+import com.loopers.domain.ProductMetricRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+
+@Component
+@RequiredArgsConstructor
+public class ProductMetricRepositoryImpl implements ProductMetricRepository {
+    private final ProductMetricJpaRepository productMetricJpaRepository;
+
+    @Override
+    public ProductMetric findByProductId(Long productId) {
+        return productMetricJpaRepository.findByProductId(productId);
+    }
+
+    @Override
+    public ProductMetric save(ProductMetric productMetric) {
+        return productMetricJpaRepository.save(productMetric);
+    }
+}

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

@@ -0,0 +1,57 @@
+package com.loopers.interfaces.consumer;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.confg.kafka.KafkaConfig;
+import com.loopers.domain.OutboxEvent;
+import com.loopers.domain.ProductEventType;
+import com.loopers.domain.ProductMetric;
+import com.loopers.domain.ProductMetricRepository;
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class KafkaOutboxConsumer {
+    private final ProductMetricRepository productMetricRepository;
+    private final ObjectMapper objectMapper;
+
+    @KafkaListener(
+            topics = {"product-viewed"},
+            containerFactory = KafkaConfig.BATCH_LISTENER
+    )
+    public void demoListener(
+            List<ConsumerRecord<String, String>> messages,
+            Acknowledgment acknowledgment
+    ) throws JsonProcessingException {
+        for (var record : messages) {
+            OutboxEvent value = objectMapper.readValue(record.value(), OutboxEvent.class);
+            Long productId = value.aggregateId();
+            String eventType = value.eventType();
+
+            ProductMetric productMetric = productMetricRepository.findByProductId(productId);
+
+            if (productMetric == null) {
+                productMetricRepository.save(
+                        ProductMetric.of(
+                                productId,
+                                0L,
+                                0L,
+                                0L,
+                                ProductEventType.valueOf(eventType)
+                        )
+                );
+            }
+            else {
+                productMetric.increaseProductMetric(ProductEventType.valueOf(eventType));
+            }
+
+        }
+        acknowledgment.acknowledge();
+    }
+}