Merge pull request #196 from rnqhstmd/round8

[volume - 8] Decoupling with Kafka

Author: rnqhstmd

apps/commerce-api/build.gradle.kts

@@ -2,6 +2,7 @@ dependencies {
     // add-ons
     implementation(project(":modules:jpa"))
     implementation(project(":modules:redis"))
+    implementation(project(":modules:kafka"))
     implementation(project(":supports:jackson"))
     implementation(project(":supports:logging"))
     implementation(project(":supports:monitoring"))
@@ -22,6 +23,10 @@ dependencies {
 
     testImplementation("net.datafaker:datafaker:2.0.2")
 
+    // Kafka
+    testImplementation(testFixtures(project(":modules:kafka")))
+    testImplementation("org.springframework.kafka:spring-kafka-test")
+
     // Resilience4j
     implementation("io.github.resilience4j:resilience4j-spring-boot3")
     implementation("org.springframework.boot:spring-boot-starter-aop")

apps/commerce-api/src/main/java/com/loopers/application/order/OrderFacade.java

@@ -90,6 +90,11 @@ public OrderInfo createOrder(OrderPlaceCommand command) {
     public void useCoupon(Long couponId) {
         log.info("쿠폰 사용 처리: couponId={}", couponId);
         Coupon coupon = couponService.getCouponWithOptimisticLock(couponId);
+        if (!coupon.canUse()) {
+            log.info("이미 사용된 쿠폰입니다 (멱등성 처리): couponId={}", couponId);
+            return;
+        }
+
         coupon.use();
         couponService.save(coupon);
     }

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

@@ -0,0 +1,96 @@
+package com.loopers.domain.outbox;
+
+import jakarta.persistence.*;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Entity
+@Table(name = "outbox_event")
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class OutboxEvent {
+
+    @Id
+    private String id;
+
+    @Column(name = "aggregate_type", nullable = false)
+    private String aggregateType;
+
+    @Column(name = "aggregate_id", nullable = false)
+    private String aggregateId;
+
+    @Column(name = "event_type", nullable = false)
+    private String eventType;
+
+    @Column(name = "topic", nullable = false)
+    private String topic;
+
+    @Column(name = "partition_key")
+    private String partitionKey;
+
+    @Column(name = "payload", columnDefinition = "TEXT", nullable = false)
+    private String payload;
+
+    @Column(name = "status", nullable = false)
+    @Enumerated(EnumType.STRING)
+    private OutboxStatus status;
+
+    @Column(name = "created_at", nullable = false)
+    private LocalDateTime createdAt;
+
+    @Column(name = "processed_at")
+    private LocalDateTime processedAt;
+
+    @Column(name = "retry_count")
+    private int retryCount;
+
+    @Column(name = "last_error")
+    private String lastError;
+
+    public static OutboxEvent create(
+            String aggregateType,
+            String aggregateId,
+            String eventType,
+            String topic,
+            String partitionKey,
+            String payload
+    ) {
+        OutboxEvent event = new OutboxEvent();
+        event.id = UUID.randomUUID().toString();
+        event.aggregateType = aggregateType;
+        event.aggregateId = aggregateId;
+        event.eventType = eventType;
+        event.topic = topic;
+        event.partitionKey = partitionKey;
+        event.payload = payload;
+        event.status = OutboxStatus.PENDING;
+        event.createdAt = LocalDateTime.now();
+        event.retryCount = 0;
+        return event;
+    }
+
+    public void markAsProcessed() {
+        this.status = OutboxStatus.PROCESSED;
+        this.processedAt = LocalDateTime.now();
+    }
+
+    public void markAsFailed(String error) {
+        this.status = OutboxStatus.FAILED;
+        this.lastError = error;
+        this.retryCount++;
+    }
+
+    public void markForRetry() {
+        this.status = OutboxStatus.PENDING;
+    }
+
+    public enum OutboxStatus {
+        PENDING,
+        PROCESSED,
+        FAILED
+    }
+}

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

@@ -0,0 +1,9 @@
+package com.loopers.domain.outbox;
+
+import java.util.List;
+
+public interface OutboxEventRepository {
+    OutboxEvent save(OutboxEvent event);
+    List<OutboxEvent> findPendingEvents(int limit);
+    List<OutboxEvent> findFailedEventsForRetry(int maxRetryCount, int limit);
+}

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

@@ -0,0 +1,46 @@
+package com.loopers.domain.outbox;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class OutboxService {
+
+    private final OutboxEventRepository outboxEventRepository;
+    private final ObjectMapper objectMapper;
+
+    @Transactional
+    public void saveEvent(
+            String aggregateType,
+            String aggregateId,
+            String eventType,
+            String topic,
+            String partitionKey,
+            Object payload
+    ) {
+        try {
+            String payloadJson = objectMapper.writeValueAsString(payload);
+            OutboxEvent event = OutboxEvent.create(
+                    aggregateType,
+                    aggregateId,
+                    eventType,
+                    topic,
+                    partitionKey,
+                    payloadJson
+            );
+            outboxEventRepository.save(event);
+            log.debug("Outbox 이벤트 저장: aggregateType={}, aggregateId={}, eventType={}",
+                    aggregateType, aggregateId, eventType);
+        } catch (JsonProcessingException e) {
+            log.error("Outbox 이벤트 직렬화 실패: aggregateType={}, aggregateId={}",
+                    aggregateType, aggregateId, e);
+            throw new RuntimeException("이벤트 직렬화 실패", e);
+        }
+    }
+}

apps/commerce-api/src/main/java/com/loopers/domain/payment/event/PaymentSucceededEvent.java

@@ -12,6 +12,7 @@ public record PaymentSucceededEvent(
         Long couponId,
         Long amount,
         PaymentMethod paymentMethod,
+        String transactionId,
         ZonedDateTime paidAt
 ) {
     public static PaymentSucceededEvent of(Payment payment, Long couponId, ZonedDateTime paidAt) {
@@ -22,6 +23,7 @@ public static PaymentSucceededEvent of(Payment payment, Long couponId, ZonedDate
                 couponId,
                 payment.getAmountValue(),
                 payment.getPaymentMethod(),
+                payment.getTransactionId(),
                 paidAt != null ? paidAt : ZonedDateTime.now()
         );
     }

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/dto/LikeChangedDto.java

@@ -0,0 +1,17 @@
+package com.loopers.infrastructure.kafka.dto;
+
+import java.util.UUID;
+
+public record LikeChangedDto(
+        String eventId,
+        Long productId,
+        String likeType
+) {
+    public static LikeChangedDto of(Long productId, String likeType) {
+        return new LikeChangedDto(
+                UUID.randomUUID().toString(),
+                productId,
+                likeType
+        );
+    }
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/dto/OrderEventDto.java

@@ -0,0 +1,67 @@
+package com.loopers.infrastructure.kafka.dto;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+public record OrderEventDto(
+        String eventId,
+        Long orderId,
+        Long userId,
+        String orderStatus,
+        Long totalAmount,
+        Long discountAmount,
+        List<OrderItemDto> items,
+        LocalDateTime occurredAt
+) {
+    public static OrderEventDto created(
+            Long orderId,
+            Long userId,
+            Long totalAmount,
+            Long discountAmount,
+            List<OrderItemDto> items
+    ) {
+        return new OrderEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "CREATED",
+                totalAmount,
+                discountAmount,
+                items,
+                LocalDateTime.now()
+        );
+    }
+
+    public static OrderEventDto completed(Long orderId, Long userId) {
+        return new OrderEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "COMPLETED",
+                null,
+                null,
+                null,
+                LocalDateTime.now()
+        );
+    }
+
+    public static OrderEventDto failed(Long orderId, Long userId) {
+        return new OrderEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "FAILED",
+                null,
+                null,
+                null,
+                LocalDateTime.now()
+        );
+    }
+
+    public record OrderItemDto(
+            Long productId,
+            int quantity,
+            Long unitPrice
+    ) {}
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/dto/PaymentEventDto.java

@@ -0,0 +1,54 @@
+package com.loopers.infrastructure.kafka.dto;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+public record PaymentEventDto(
+        String eventId,
+        Long orderId,
+        Long userId,
+        String paymentStatus,
+        String transactionId,
+        Long amount,
+        String failureReason,
+        LocalDateTime occurredAt
+) {
+    public static PaymentEventDto success(Long orderId, Long userId, String transactionId, Long amount) {
+        return new PaymentEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "SUCCESS",
+                transactionId,
+                amount,
+                null,
+                LocalDateTime.now()
+        );
+    }
+
+    public static PaymentEventDto failed(Long orderId, Long userId, String failureReason) {
+        return new PaymentEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "FAILED",
+                null,
+                null,
+                failureReason,
+                LocalDateTime.now()
+        );
+    }
+
+    public static PaymentEventDto pending(Long orderId, Long userId, String transactionId) {
+        return new PaymentEventDto(
+                UUID.randomUUID().toString(),
+                orderId,
+                userId,
+                "PENDING",
+                transactionId,
+                null,
+                null,
+                LocalDateTime.now()
+        );
+    }
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/dto/ProductViewedDto.java

@@ -0,0 +1,15 @@
+package com.loopers.infrastructure.kafka.dto;
+
+import java.util.UUID;
+
+public record ProductViewedDto(
+        String eventId,
+        Long productId
+) {
+    public static ProductViewedDto of(Long productId) {
+        return new ProductViewedDto(
+                UUID.randomUUID().toString(),
+                productId
+        );
+    }
+}