feat: Transactional Outbox 패턴 적용

Author: Kimjipang

apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java

@@ -1,30 +1,31 @@
 package com.loopers.application.product;
 
-import com.loopers.domain.actionlog.ActionType;
 import com.loopers.domain.brand.BrandRepository;
+import com.loopers.domain.outbox.AggregateType;
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.domain.outbox.OutboxType;
+import com.loopers.domain.outbox.OutboxRepository;
 import com.loopers.domain.product.Product;
 import com.loopers.domain.product.ProductRepository;
-import com.loopers.domain.product.ProductViewedMessage;
 import com.loopers.infrastructure.product.ProductViewKafkaProducer;
 import com.loopers.interfaces.api.product.ProductV1Dto;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import lombok.RequiredArgsConstructor;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Cacheable;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.math.BigDecimal;
 import java.util.List;
-import java.util.UUID;
 
 @Component
 @RequiredArgsConstructor
 public class ProductFacade {
     private final ProductRepository productRepository;
     private final BrandRepository brandRepository;
+    private final OutboxRepository outBoxRepository;
     private final ProductViewKafkaProducer kafkaProducer;
 
     @Transactional
@@ -51,21 +52,20 @@ public List<ProductInfo> findAllProducts() {
                 .toList();
     }
 
-    @Transactional(readOnly = true)
+    @Transactional
     @Cacheable(value = "product", key = "#id")
     public ProductInfo findProductById(Long id) {
         Product product = productRepository.findById(id).orElseThrow(
                 () -> new CoreException(ErrorType.NOT_FOUND, "찾고자 하는 상품이 존재하지 않습니다.")
         );
 
-        // 유저 ID는 임시로 하드 코딩했습니다. 추후 인증/인가 기능이 추가되면 수정할 예정입니다.
-        ProductViewedMessage message = new ProductViewedMessage(
-                UUID.randomUUID().toString(),
-                1L,
-                product.getId()
+        OutboxEvent outBoxEvent = OutboxEvent.of(
+                AggregateType.PRODUCT,
+                product.getId(),
+                OutboxType.PRODUCT_VIEWED
         );
 
-        kafkaProducer.sendProductViewedMessage(message);
+        outBoxRepository.save(outBoxEvent);
 
         return ProductInfo.from(product);
     }

apps/commerce-api/src/main/java/com/loopers/domain/outbox/AggregateType.java

@@ -0,0 +1,7 @@
+package com.loopers.domain.outbox;
+
+public enum AggregateType {
+    PRODUCT,
+    ORDER,
+    USER
+}

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

@@ -0,0 +1,46 @@
+package com.loopers.domain.outbox;
+
+import com.loopers.domain.BaseEntity;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.Table;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@Entity
+@Table(name = "outbox")
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class OutboxEvent extends BaseEntity {
+    @Enumerated(EnumType.STRING)
+    @Column(name = "aggregate_type", nullable = false)
+    private AggregateType aggregateType;
+
+    @Column(name = "aggregate_id", nullable = false)
+    private Long aggregateId;
+
+//    @Enumerated(EnumType.STRING)
+    @Column(name = "event_type", nullable = false)
+    private OutboxType eventType;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false)
+    private OutboxStatus status;
+
+    public OutboxEvent(AggregateType aggregateType, Long aggregateId, OutboxType eventType, OutboxStatus status) {
+        this.aggregateType = aggregateType;
+        this.aggregateId = aggregateId;
+        this.eventType = eventType;
+        this.status = status;
+    }
+
+    public static OutboxEvent of(AggregateType aggregateType, Long aggregateId, OutboxType eventType) {
+        return new OutboxEvent(aggregateType, aggregateId, eventType, OutboxStatus.PENDING);
+    }
+
+    public void markAsProcessed() {
+        this.status = OutboxStatus.PROCESSED;
+    }
+
+}

apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxRepository.java

@@ -0,0 +1,9 @@
+package com.loopers.domain.outbox;
+
+import java.util.List;
+
+public interface OutboxRepository {
+    OutboxEvent save(OutboxEvent outBoxEvent);
+
+    List<OutboxEvent> findPending(int limit);
+}

apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxStatus.java

@@ -0,0 +1,7 @@
+package com.loopers.domain.outbox;
+
+public enum OutboxStatus {
+    PENDING,
+    PROCESSED,
+    FAILED
+}

apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxType.java

@@ -0,0 +1,7 @@
+package com.loopers.domain.outbox;
+
+public enum OutboxType {
+    PRODUCT_VIEWED,
+    ORDER_CREATED,
+    PAYMENT_COMPLETED
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxPublisher.java

@@ -0,0 +1,36 @@
+package com.loopers.infrastructure.outbox;
+
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.domain.outbox.OutboxRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class KafkaOutboxPublisher {
+
+    private final OutboxRepository outboxRepository;
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+
+    @Scheduled(fixedDelayString = "1000")
+    @Transactional
+    public void publish() {
+//        System.out.println("hello");
+        List<OutboxEvent> events = outboxRepository.findPending(5);
+        for (OutboxEvent event : events) {
+            System.out.println("event.toString() = " + event.toString());
+        }
+
+        for (OutboxEvent event : events) {
+            kafkaTemplate.send("product-viewed", event);
+
+            event.markAsProcessed();
+        }
+
+    }
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxJpaRepository.java

@@ -0,0 +1,15 @@
+package com.loopers.infrastructure.outbox;
+
+import com.loopers.domain.outbox.OutboxEvent;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import java.util.List;
+
+public interface OutboxJpaRepository extends JpaRepository<OutboxEvent, Long> {
+    @Query("SELECT o FROM OutboxEvent o WHERE o.status = 'PENDING' ORDER BY o.createdAt ASC")
+    List<OutboxEvent> findPending(int limit);
+
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxRepositoryImpl.java

@@ -0,0 +1,25 @@
+package com.loopers.infrastructure.outbox;
+
+import com.loopers.domain.outbox.OutboxEvent;
+import com.loopers.domain.outbox.OutboxRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class OutboxRepositoryImpl implements OutboxRepository {
+    private final OutboxJpaRepository outBoxJpaRepository;
+
+    @Override
+    public OutboxEvent save(OutboxEvent outBoxEvent) {
+        return outBoxJpaRepository.save(outBoxEvent);
+    }
+
+    @Override
+    public List<OutboxEvent> findPending(int limit) {
+        return outBoxJpaRepository.findPending(limit);
+    }
+
+}