feat: 이벤트 발행 로직 작성

1분에 한 번씩 outbox 테이블 내 row 중 status가 PENDING인 데이터 최대 2개 조회해서 메시지로 발행

Author: Kimjipang

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

@@ -3,6 +3,7 @@
 public enum OutboxType {
     PRODUCT_VIEWED,
     PRODUCT_LIKED,
+    PRODUCT_SALES,
     ORDER_CREATED,
     PAYMENT_COMPLETED
 }

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

@@ -2,6 +2,7 @@
 
 import com.loopers.domain.outbox.OutboxEvent;
 import com.loopers.domain.outbox.OutboxRepository;
+import com.loopers.domain.outbox.OutboxType;
 import lombok.RequiredArgsConstructor;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -17,13 +18,21 @@ public class KafkaOutboxPublisher {
     private final OutboxRepository outboxRepository;
     private final KafkaTemplate<Object, Object> kafkaTemplate;
 
-    @Scheduled(fixedDelayString = "1000")
+    @Scheduled(fixedDelayString = "60000")
     @Transactional
-    public void publish() {
-        List<OutboxEvent> events = outboxRepository.findPending(5);
+    public void publishProductViewed() {
+        List<OutboxEvent> events = outboxRepository.findPending(2);
 
         for (OutboxEvent event : events) {
-            kafkaTemplate.send("product-viewed", String.valueOf(event.getAggregateId()), event);
+            OutboxType type = event.getEventType();
+
+            if (type.equals(OutboxType.PRODUCT_VIEWED)) {
+                kafkaTemplate.send("product-viewed", String.valueOf(event.getAggregateId()), event);
+            }
+
+            else if (type.equals(OutboxType.PRODUCT_LIKED)) {
+                kafkaTemplate.send("product-liked", String.valueOf(event.getAggregateId()), event);
+            }
 
             event.markAsProcessed();
         }

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

@@ -2,14 +2,8 @@
 
 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

@@ -2,15 +2,25 @@
 
 import com.loopers.domain.outbox.OutboxEvent;
 import com.loopers.domain.outbox.OutboxRepository;
-import lombok.RequiredArgsConstructor;
+import com.loopers.domain.outbox.OutboxStatus;
+import com.loopers.domain.outbox.QOutboxEvent;
+import com.querydsl.jpa.impl.JPAQueryFactory;
+import jakarta.persistence.EntityManager;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
 
+import static com.loopers.domain.outbox.QOutboxEvent.outboxEvent;
+
 @Component
-@RequiredArgsConstructor
 public class OutboxRepositoryImpl implements OutboxRepository {
     private final OutboxJpaRepository outBoxJpaRepository;
+    private final JPAQueryFactory queryFactory;
+
+    public OutboxRepositoryImpl(OutboxJpaRepository outBoxJpaRepository, EntityManager entityManager) {
+        this.outBoxJpaRepository = outBoxJpaRepository;
+        this.queryFactory = new JPAQueryFactory(entityManager);
+    }
 
     @Override
     public OutboxEvent save(OutboxEvent outBoxEvent) {
@@ -19,7 +29,12 @@ public OutboxEvent save(OutboxEvent outBoxEvent) {
 
     @Override
     public List<OutboxEvent> findPending(int limit) {
-        return outBoxJpaRepository.findPending(limit);
+        return queryFactory
+                .selectFrom(outboxEvent)
+                .where(outboxEvent.status.eq(OutboxStatus.PENDING))
+                .orderBy(outboxEvent.createdAt.asc())
+                .limit(limit)
+                .fetch();
     }
 
 }