feature: Outbox 기반 주문 및 판매 이벤트 처리, Kafka 연동 구현

- `OrderEventOutboxHandler` 추가로 주문 생성 이벤트 처리 및 Kafka 발행 로직 구현.
- `OrderSalesAggregateListener`를 통해 판매 집계 Kafka 이벤트 발행 및 비동기 처리 지원.
- `OrderCreatedEvent` 확장: 고유 `eventId` 추가 및 주문 품목 전달 정보 구조 변경.
- 신규 `SalesCountEvent` 추가로 Kafka 기반 판매 메트릭 관리 가능.
- `OrderFacade`, `PointPaymentEventListener`, `PgPaymentEventListener` 등 관련 클래스 수정 및 패키지 정리.

Author: yeonjiyeon

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

@@ -1,5 +1,6 @@
 package com.loopers.application.order;
 
+import com.loopers.application.order.event.OrderCreatedEvent;
 import com.loopers.domain.coupon.CouponService;
 import com.loopers.domain.order.Order;
 import com.loopers.domain.order.OrderCommand.Item;
@@ -62,9 +63,10 @@ public OrderInfo placeOrder(PlaceOrder command) {
 
     productService.deductStock(products, orderItems);
 
-    OrderCreatedEvent orderEvent = new OrderCreatedEvent(
+    OrderCreatedEvent orderEvent = OrderCreatedEvent.of(
         order.getId(),
         user,
+        orderItems,
         finalPaymentAmount,
         command.paymentType(),
         command.cardType(),

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

@@ -1,14 +1,52 @@
 package com.loopers.application.order.event;
 
+import com.loopers.domain.order.OrderItem;
 import com.loopers.domain.payment.PaymentType;
 import com.loopers.domain.user.User;
+import java.util.List;
+import java.util.UUID;
 
 public record OrderCreatedEvent(
+    String eventId,
     Long orderId,
-    User user,
+    Long userId,
+    List<OrderItemInfo> items,
     long finalAmount,
     PaymentType paymentType,
     String cardType,
     String cardNo,
     Long couponId
-) {}
+) {
+
+  public record OrderItemInfo(Long productId, int quantity) {
+
+  }
+
+  public static OrderCreatedEvent of(
+      Long orderId,
+      User user,
+      List<OrderItem> orderItems,
+      long finalAmount,
+      PaymentType paymentType,
+      String cardType,
+      String cardNo,
+      Long couponId
+  ) {
+
+    List<OrderItemInfo> itemInfos = orderItems.stream()
+        .map(item -> new OrderItemInfo(item.getProductId(), item.getQuantity()))
+        .toList();
+
+    return new OrderCreatedEvent(
+        UUID.randomUUID().toString(),
+        orderId,
+        user.getId(),
+        itemInfos,
+        finalAmount,
+        paymentType,
+        cardType,
+        cardNo,
+        couponId
+    );
+  }
+}

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

@@ -0,0 +1,25 @@
+package com.loopers.application.order.event;
+
+import com.loopers.domain.event.OutboxService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+@Component
+@RequiredArgsConstructor
+public class OrderEventOutboxHandler {
+  private final KafkaTemplate<Object, Object> kafkaTemplate;
+  private final OutboxService outboxService;
+
+  @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+  public void handle(OrderCreatedEvent event) {
+    kafkaTemplate.send("order-events", String.valueOf(event.orderId()), event)
+        .whenComplete((result, ex) -> {
+          if (ex == null) {
+            outboxService.markPublished(event.eventId());
+          }
+        });
+  }
+}

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

@@ -0,0 +1,34 @@
+package com.loopers.application.order.event;
+
+import com.loopers.domain.event.OutboxService;
+import com.loopers.event.SalesCountEvent;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+@Component
+@RequiredArgsConstructor
+public class OrderSalesAggregateListener {
+
+  private final OutboxService outboxService;
+  private final ApplicationEventPublisher eventPublisher;
+
+  @Async
+  @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+  public void handleOrderCreated(OrderCreatedEvent event) {
+
+    event.items().forEach(item -> {
+
+      SalesCountEvent kafkaEvent = SalesCountEvent.of(
+          item.productId(),
+          item.quantity()
+      );
+
+      outboxService.saveEvent("SALES_METRICS", String.valueOf(item.productId()), kafkaEvent);
+      eventPublisher.publishEvent(kafkaEvent);
+    });
+  }
+}

apps/commerce-api/src/main/java/com/loopers/application/payment/PgPaymentEventListener.java

@@ -1,6 +1,6 @@
 package com.loopers.application.payment;
 
-import com.loopers.application.order.OrderCreatedEvent;
+import com.loopers.application.order.event.OrderCreatedEvent;
 import com.loopers.application.payment.PaymentEvent.PaymentRequestFailedEvent;
 import com.loopers.application.payment.PaymentEvent.PaymentRequestedEvent;
 import com.loopers.domain.payment.Payment;
@@ -36,7 +36,7 @@ public void handleOrderCreatedEvent(OrderCreatedEvent event) {
       return;
     }
 
-    User user = userService.findById(event.user().getId())
+    User user = userService.findById(event.userId())
         .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "유저 정보를 찾을 수 없습니다."));
 
     PaymentProcessor processor = paymentProcessors.stream()

apps/commerce-api/src/main/java/com/loopers/application/point/PointPaymentEventListener.java

@@ -1,13 +1,15 @@
 package com.loopers.application.point;
 
-import com.loopers.application.order.OrderCreatedEvent;
+import com.loopers.application.order.event.OrderCreatedEvent;
 import com.loopers.application.payment.PaymentEvent.PaymentCompletedEvent;
 import com.loopers.application.payment.PaymentEvent.PaymentRequestFailedEvent;
 import com.loopers.domain.order.OrderService;
 import com.loopers.domain.order.OrderStatus;
 import com.loopers.domain.payment.Payment;
 import com.loopers.domain.payment.PaymentProcessor;
 import com.loopers.domain.payment.PaymentType;
+import com.loopers.domain.user.User;
+import com.loopers.domain.user.UserService;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import java.util.List;
@@ -23,8 +25,10 @@
 @Component
 @RequiredArgsConstructor
 public class PointPaymentEventListener {
+
   private final List<PaymentProcessor> paymentProcessors;
   private final OrderService orderService;
+  private final UserService userService;
   private final ApplicationEventPublisher eventPublisher;
 
   @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
@@ -40,10 +44,13 @@ public void handleOrderCreatedEvent(OrderCreatedEvent event) {
         .findFirst()
         .orElseThrow(() -> new CoreException(ErrorType.BAD_REQUEST, "포인트 결제 프로세서를 찾을 수 없습니다."));
 
+    User user = userService.findById(event.userId())
+        .orElseThrow(() -> new CoreException(ErrorType.NOT_FOUND, "유저를 찾을 수 없습니다."));
+
     try {
       Payment payment = processor.process(
           event.orderId(),
-          event.user(),
+          user,
           event.finalAmount(),
           Map.of()
       );

modules/kafka/src/main/java/com/loopers/event/SalesCountEvent.java

@@ -0,0 +1,20 @@
+package com.loopers.event;
+
+import java.util.UUID;
+
+public record SalesCountEvent(
+    String eventId,
+    Long productId,
+    int quantity,
+    long timestamp
+) {
+
+  public static SalesCountEvent of(Long productId, int quantity) {
+    return new SalesCountEvent(
+        UUID.randomUUID().toString(),
+        productId,
+        quantity,
+        System.currentTimeMillis()
+    );
+  }
+}