refactor: Interaction/Order 컨슈머 멱등성 4-TX → 단일 TX로 전환

기존 REQUIRES_NEW tryMark 방식은 event_handled INSERT가 독립 커밋되어,
upsert 실패 시 재시도가 영구 스킵되는 데이터 유실 버그가 있었음.

existsCheck + eventHandled save + upsert를 단일 @transactional로 묶고,
DIV(PK 충돌)는 TX 바깥에서 catch하여 rollback-only 마킹 문제를 원천 차단.

Author: ghtjr410

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/InteractionMetricProcessor.java

@@ -1,41 +1,38 @@
 package com.loopers.application.metrics;
 
-import com.loopers.application.idempotent.IdempotencyChecker;
 import com.loopers.application.log.EventLogWriter;
-import com.loopers.domain.metrics.ProductLikeMetricRepository;
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Component;
 
 import java.time.Instant;
-import java.time.LocalDateTime;
 
 @Component
 @RequiredArgsConstructor
 public class InteractionMetricProcessor {
 
-    private final IdempotencyChecker idempotencyChecker;
-    private final ProductLikeMetricRepository likeMetricRepository;
+    private final InteractionMetricWriter interactionMetricWriter;
     private final EventLogWriter eventLogWriter;
     private final ConsumerMetrics consumerMetrics;
 
     public void process(String eventId, String eventType, String topic, String groupId,
                         Long productId, Instant occurredAt) {
         long start = System.currentTimeMillis();
 
-        if (!idempotencyChecker.tryMark(eventId, eventType)) {
+        try {
+            boolean processed = interactionMetricWriter.write(eventId, eventType, productId, occurredAt);
+
+            if (processed) {
+                long durationMs = System.currentTimeMillis() - start;
+                eventLogWriter.saveProcessed(eventId, eventType, topic, groupId, durationMs);
+                consumerMetrics.recordProcessed(topic, groupId, eventType, durationMs);
+            } else {
+                eventLogWriter.saveSkipped(eventId, eventType, topic, groupId);
+                consumerMetrics.recordSkipped(topic, groupId, eventType);
+            }
+        } catch (DataIntegrityViolationException e) {
             eventLogWriter.saveSkipped(eventId, eventType, topic, groupId);
             consumerMetrics.recordSkipped(topic, groupId, eventType);
-            return;
-        }
-
-        try {
-            LocalDateTime bucketTime = BucketTimeUtils.toLocalDateTime(BucketTimeUtils.truncate5min(occurredAt));
-            int delta = "product.liked".equals(eventType) ? 1 : -1;
-            likeMetricRepository.upsert(productId, bucketTime, delta);
-
-            long durationMs = System.currentTimeMillis() - start;
-            eventLogWriter.saveProcessed(eventId, eventType, topic, groupId, durationMs);
-            consumerMetrics.recordProcessed(topic, groupId, eventType, durationMs);
         } catch (Exception e) {
             long durationMs = System.currentTimeMillis() - start;
             eventLogWriter.saveFailed(eventId, eventType, topic, groupId, e.getMessage(), durationMs);

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/InteractionMetricWriter.java

@@ -0,0 +1,32 @@
+package com.loopers.application.metrics;
+
+import com.loopers.domain.idempotent.EventHandled;
+import com.loopers.domain.idempotent.EventHandledRepository;
+import com.loopers.domain.metrics.ProductLikeMetricRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+@Component
+@RequiredArgsConstructor
+public class InteractionMetricWriter {
+
+    private final EventHandledRepository eventHandledRepository;
+    private final ProductLikeMetricRepository likeMetricRepository;
+
+    @Transactional
+    public boolean write(String eventId, String eventType, Long productId, Instant occurredAt) {
+        if (eventHandledRepository.existsByEventId(eventId)) {
+            return false;
+        }
+        eventHandledRepository.save(EventHandled.create(eventId, eventType));
+
+        LocalDateTime bucketTime = BucketTimeUtils.toLocalDateTime(BucketTimeUtils.truncate5min(occurredAt));
+        int delta = "product.liked".equals(eventType) ? 1 : -1;
+        likeMetricRepository.upsert(productId, bucketTime, delta);
+        return true;
+    }
+}

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/OrderMetricProcessor.java

@@ -1,19 +1,17 @@
 package com.loopers.application.metrics;
 
-import com.loopers.application.idempotent.IdempotencyChecker;
 import com.loopers.application.log.EventLogWriter;
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Component;
 
 import java.time.Instant;
-import java.time.LocalDateTime;
 import java.util.List;
 
 @Component
 @RequiredArgsConstructor
 public class OrderMetricProcessor {
 
-    private final IdempotencyChecker idempotencyChecker;
     private final OrderMetricWriter orderMetricWriter;
     private final EventLogWriter eventLogWriter;
     private final ConsumerMetrics consumerMetrics;
@@ -22,19 +20,20 @@ public void process(String eventId, String eventType, String topic, String group
                         Instant occurredAt, List<OrderItemMetric> items) {
         long start = System.currentTimeMillis();
 
-        if (!idempotencyChecker.tryMark(eventId, eventType)) {
+        try {
+            boolean processed = orderMetricWriter.write(eventId, eventType, occurredAt, items);
+
+            if (processed) {
+                long durationMs = System.currentTimeMillis() - start;
+                eventLogWriter.saveProcessed(eventId, eventType, topic, groupId, durationMs);
+                consumerMetrics.recordProcessed(topic, groupId, eventType, durationMs);
+            } else {
+                eventLogWriter.saveSkipped(eventId, eventType, topic, groupId);
+                consumerMetrics.recordSkipped(topic, groupId, eventType);
+            }
+        } catch (DataIntegrityViolationException e) {
             eventLogWriter.saveSkipped(eventId, eventType, topic, groupId);
             consumerMetrics.recordSkipped(topic, groupId, eventType);
-            return;
-        }
-
-        try {
-            LocalDateTime bucketTime = BucketTimeUtils.toLocalDateTime(BucketTimeUtils.truncate5min(occurredAt));
-            orderMetricWriter.upsertAll(bucketTime, items);
-
-            long durationMs = System.currentTimeMillis() - start;
-            eventLogWriter.saveProcessed(eventId, eventType, topic, groupId, durationMs);
-            consumerMetrics.recordProcessed(topic, groupId, eventType, durationMs);
         } catch (Exception e) {
             long durationMs = System.currentTimeMillis() - start;
             eventLogWriter.saveFailed(eventId, eventType, topic, groupId, e.getMessage(), durationMs);

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/OrderMetricWriter.java

@@ -1,27 +1,38 @@
 package com.loopers.application.metrics;
 
 import com.loopers.application.metrics.OrderMetricProcessor.OrderItemMetric;
+import com.loopers.domain.idempotent.EventHandled;
+import com.loopers.domain.idempotent.EventHandledRepository;
 import com.loopers.domain.metrics.ProductOrderMetricRepository;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.time.Instant;
 import java.time.LocalDateTime;
 import java.util.List;
 
 @Component
 @RequiredArgsConstructor
 public class OrderMetricWriter {
 
+    private final EventHandledRepository eventHandledRepository;
     private final ProductOrderMetricRepository orderMetricRepository;
 
     @Transactional
-    public void upsertAll(LocalDateTime bucketTime, List<OrderItemMetric> items) {
+    public boolean write(String eventId, String eventType, Instant occurredAt, List<OrderItemMetric> items) {
+        if (eventHandledRepository.existsByEventId(eventId)) {
+            return false;
+        }
+        eventHandledRepository.save(EventHandled.create(eventId, eventType));
+
+        LocalDateTime bucketTime = BucketTimeUtils.toLocalDateTime(BucketTimeUtils.truncate5min(occurredAt));
         for (OrderItemMetric item : items) {
             orderMetricRepository.upsert(
                     item.productId(), bucketTime,
                     1, item.quantity(), item.salesAmount()
             );
         }
+        return true;
     }
 }