diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/OrchestratorCommandKafkaListener.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/OrchestratorCommandKafkaListener.java new file mode 100644 index 00000000..0c75123a --- /dev/null +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/OrchestratorCommandKafkaListener.java @@ -0,0 +1,29 @@ +package kr.magicbox.generalgoods.adapter.in.kafka; + +import kr.magicbox.generalgoods.adapter.in.kafka.annotation.Idempotent; +import kr.magicbox.generalgoods.adapter.in.kafka.event.StockReserveCommandEvent; +import kr.magicbox.generalgoods.application.port.in.HandleStockReserveUseCase; +import kr.magicbox.generalgoods.global.exception.BusinessException; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.annotation.RetryableTopic; +import org.springframework.kafka.retrytopic.DltStrategy; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class OrchestratorCommandKafkaListener { + + private final HandleStockReserveUseCase handleStockReserveUseCase; + + @Idempotent + @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {BusinessException.class}) + @KafkaListener(topics = "outbox.event.stock-reserve.general-good", groupId = "general-goods-service") + public void handleStockReserve(ConsumerRecord consumerRecord) { + log.info("[Inbox] stock-reserve.general-good 커맨드 수신. key={}", consumerRecord.key()); + handleStockReserveUseCase.handleStockReserve(consumerRecord.value()); + } +} diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/StockReserveCommand.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/event/StockReserveCommandEvent.java similarity index 61% rename from services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/StockReserveCommand.java rename to services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/event/StockReserveCommandEvent.java index e8577587..4a638c2d 100644 --- a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/StockReserveCommand.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/event/StockReserveCommandEvent.java @@ -1,4 +1,4 @@ -package kr.magicbox.orchestrator.domain.event; +package kr.magicbox.generalgoods.adapter.in.kafka.event; import com.fasterxml.jackson.annotation.JsonProperty; import lombok.Builder; @@ -7,29 +7,19 @@ import java.util.List; @Builder -public record StockReserveCommand( - @JsonProperty("event_id") Long eventId, +public record StockReserveCommandEvent( @JsonProperty("order_id") Long orderId, @JsonProperty("customer_id") Long customerId, @JsonProperty("total_amount") Long totalAmount, @JsonProperty("items") List items, @JsonProperty("occurred_at") Instant occurredAt -) implements OrchestratorCommandEvent { +) implements InboxEvent { @Builder public record ItemPayload( + @JsonProperty("order_line_id") Long orderLineId, @JsonProperty("product_id") Long productId, @JsonProperty("quantity") int quantity, @JsonProperty("unit_price") long unitPrice ) {} - - @Override - public String key() { - return orderId.toString(); - } - - @Override - public OrchestratorCommandEventType eventType() { - return OrchestratorCommandEventType.STOCK_RESERVE; - } } diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/GeneralGoodsJpaAdapter.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/GeneralGoodsJpaAdapter.java index d7884bdd..54a42afb 100644 --- a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/GeneralGoodsJpaAdapter.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/GeneralGoodsJpaAdapter.java @@ -74,4 +74,9 @@ public List findAllByCursor(Long cursorId, int size) { .map(generalGoodsMapper::toDomain) .toList(); } + + @Override + public boolean decreaseStock(Long productId, long quantity) { + return generalGoodsJpaRepository.decreaseStock(productId, quantity) > 0; + } } diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/entity/GeneralGoodsEntity.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/entity/GeneralGoodsEntity.java index 97275658..43380bee 100644 --- a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/entity/GeneralGoodsEntity.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/entity/GeneralGoodsEntity.java @@ -39,6 +39,7 @@ public class GeneralGoodsEntity extends BaseEntity { @ElementCollection @CollectionTable(name = "general_goods_category", joinColumns = @JoinColumn(name = "general_goods_id")) + @Enumerated(EnumType.STRING) @Column(name = "category") private Set categories; @@ -48,9 +49,6 @@ public class GeneralGoodsEntity extends BaseEntity { @Column(name = "is_deleted", nullable = false) private boolean isDeleted = false; - @Version - private Integer version; - @Builder public GeneralGoodsEntity(Long creatorId, String name, Long price, Long stock, String description, GeneralGoodsLevel level, Set categories) { this.creatorId = creatorId; diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/repository/GeneralGoodsJpaRepository.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/repository/GeneralGoodsJpaRepository.java index 673b44b7..84541f0f 100644 --- a/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/repository/GeneralGoodsJpaRepository.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/persistence/repository/GeneralGoodsJpaRepository.java @@ -26,4 +26,8 @@ public interface GeneralGoodsJpaRepository extends JpaRepository= :quantity AND g.isDeleted = false") + int decreaseStock(@Param("id") Long id, @Param("quantity") long quantity); } \ No newline at end of file diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/in/HandleStockReserveUseCase.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/in/HandleStockReserveUseCase.java new file mode 100644 index 00000000..69e1ac45 --- /dev/null +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/in/HandleStockReserveUseCase.java @@ -0,0 +1,7 @@ +package kr.magicbox.generalgoods.application.port.in; + +import kr.magicbox.generalgoods.adapter.in.kafka.event.StockReserveCommandEvent; + +public interface HandleStockReserveUseCase { + void handleStockReserve(StockReserveCommandEvent event); +} diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/out/GeneralGoodsRepositoryPort.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/out/GeneralGoodsRepositoryPort.java index 5296e142..fc8e381d 100644 --- a/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/out/GeneralGoodsRepositoryPort.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/port/out/GeneralGoodsRepositoryPort.java @@ -16,4 +16,6 @@ public interface GeneralGoodsRepositoryPort { GeneralGoods findById(GeneralGoodsId id); List findAllByCursor(Long cursorId, int size); + + boolean decreaseStock(Long productId, long quantity); } \ No newline at end of file diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/service/HandleStockReserveService.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/service/HandleStockReserveService.java new file mode 100644 index 00000000..75b978cb --- /dev/null +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/application/service/HandleStockReserveService.java @@ -0,0 +1,60 @@ +package kr.magicbox.generalgoods.application.service; + +import kr.magicbox.generalgoods.adapter.in.kafka.event.StockReserveCommandEvent; +import kr.magicbox.generalgoods.application.port.in.HandleStockReserveUseCase; +import kr.magicbox.generalgoods.application.port.out.GeneralGoodsOutboxPort; +import kr.magicbox.generalgoods.application.port.out.GeneralGoodsRepositoryPort; +import kr.magicbox.generalgoods.domain.event.StockReserveFailedEvent; +import kr.magicbox.generalgoods.domain.event.StockReserveSucceededEvent; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Instant; + +@Slf4j +@Service +@RequiredArgsConstructor +public class HandleStockReserveService implements HandleStockReserveUseCase { + + private final GeneralGoodsRepositoryPort generalGoodsRepositoryPort; + private final GeneralGoodsOutboxPort generalGoodsOutboxPort; + + @Transactional + @Override + public void handleStockReserve(StockReserveCommandEvent event) { + if (event.items() == null || event.items().isEmpty()) { + log.warn("[StockReserve] items 없음. orderId={}", event.orderId()); + generalGoodsOutboxPort.save(StockReserveFailedEvent.builder() + .orderId(event.orderId()) + .customerId(event.customerId()) + .reason("items 없음") + .occurredAt(Instant.now()) + .build()); + return; + } + + for (StockReserveCommandEvent.ItemPayload item : event.items()) { + boolean decreased = generalGoodsRepositoryPort.decreaseStock(item.productId(), item.quantity()); + if (!decreased) { + log.warn("[StockReserve] 재고 부족. orderId={}, productId={}", event.orderId(), item.productId()); + generalGoodsOutboxPort.save(StockReserveFailedEvent.builder() + .orderId(event.orderId()) + .customerId(event.customerId()) + .reason("재고 부족. productId=" + item.productId()) + .occurredAt(Instant.now()) + .build()); + return; + } + } + + log.info("[StockReserve] 재고 예약 성공. orderId={}", event.orderId()); + generalGoodsOutboxPort.save(StockReserveSucceededEvent.builder() + .orderId(event.orderId()) + .customerId(event.customerId()) + .totalAmount(event.totalAmount()) + .occurredAt(Instant.now()) + .build()); + } +} diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/GeneralGoodsDomainEventType.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/GeneralGoodsDomainEventType.java index bd19ba20..c3969698 100644 --- a/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/GeneralGoodsDomainEventType.java +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/GeneralGoodsDomainEventType.java @@ -9,7 +9,9 @@ public enum GeneralGoodsDomainEventType { GENERAL_GOODS_CREATED("general-goods-created"), GENERAL_GOODS_UPDATED("general-goods-updated"), - GENERAL_GOODS_DELETED("general-goods-deleted"); + GENERAL_GOODS_DELETED("general-goods-deleted"), + STOCK_RESERVE_SUCCEEDED("stock-reserve-general-goods-succeeded"), + STOCK_RESERVE_FAILED("stock-reserve-general-goods-failed"); private final String value; } diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveFailedEvent.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveFailedEvent.java new file mode 100644 index 00000000..5ff2fa3e --- /dev/null +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveFailedEvent.java @@ -0,0 +1,25 @@ +package kr.magicbox.generalgoods.domain.event; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.Builder; + +import java.time.Instant; + +@Builder +public record StockReserveFailedEvent( + @JsonProperty("order_id") Long orderId, + @JsonProperty("customer_id") Long customerId, + @JsonProperty("reason") String reason, + @JsonProperty("occurred_at") Instant occurredAt +) implements GeneralGoodsDomainEvent { + + @Override + public String key() { + return orderId.toString(); + } + + @Override + public GeneralGoodsDomainEventType eventType() { + return GeneralGoodsDomainEventType.STOCK_RESERVE_FAILED; + } +} diff --git a/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveSucceededEvent.java b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveSucceededEvent.java new file mode 100644 index 00000000..907eeaf7 --- /dev/null +++ b/services/general-goods/src/main/java/kr/magicbox/generalgoods/domain/event/StockReserveSucceededEvent.java @@ -0,0 +1,25 @@ +package kr.magicbox.generalgoods.domain.event; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.Builder; + +import java.time.Instant; + +@Builder +public record StockReserveSucceededEvent( + @JsonProperty("order_id") Long orderId, + @JsonProperty("customer_id") Long customerId, + @JsonProperty("total_amount") Long totalAmount, + @JsonProperty("occurred_at") Instant occurredAt +) implements GeneralGoodsDomainEvent { + + @Override + public String key() { + return orderId.toString(); + } + + @Override + public GeneralGoodsDomainEventType eventType() { + return GeneralGoodsDomainEventType.STOCK_RESERVE_SUCCEEDED; + } +} diff --git a/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/controller/NotificationCommandController.java b/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/controller/NotificationCommandController.java index c7809672..41384b97 100644 --- a/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/controller/NotificationCommandController.java +++ b/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/controller/NotificationCommandController.java @@ -1,12 +1,16 @@ package kr.magicbox.notification.adapter.in.web.controller; +import kr.magicbox.notification.adapter.in.web.dto.request.ReadNotificationsRequest; import kr.magicbox.notification.adapter.in.web.dto.request.RegisterFcmTokenRequest; +import kr.magicbox.notification.application.dto.command.ReadNotificationCommand; import kr.magicbox.notification.application.dto.command.RegisterFcmTokenCommand; +import kr.magicbox.notification.application.port.in.ReadNotificationUseCase; import kr.magicbox.notification.application.port.in.RegisterFcmTokenUseCase; import kr.magicbox.notification.domain.vo.UserId; import lombok.RequiredArgsConstructor; import org.springframework.http.ResponseEntity; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.bind.annotation.PatchMapping; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -18,6 +22,7 @@ public class NotificationCommandController { private final RegisterFcmTokenUseCase registerFcmTokenUseCase; + private final ReadNotificationUseCase readNotificationUseCase; @PostMapping("/fcm-token") public ResponseEntity registerFcmToken( @@ -26,4 +31,12 @@ public ResponseEntity registerFcmToken( registerFcmTokenUseCase.register(RegisterFcmTokenCommand.of(userId.value(), request.token())); return ResponseEntity.ok().build(); } + + @PatchMapping("/read") + public ResponseEntity readNotifications( + @AuthenticationPrincipal UserId userId, + @RequestBody ReadNotificationsRequest request) { + readNotificationUseCase.readAll(ReadNotificationCommand.of(request.notificationIds(), userId.value())); + return ResponseEntity.noContent().build(); + } } diff --git a/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/dto/request/ReadNotificationsRequest.java b/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/dto/request/ReadNotificationsRequest.java new file mode 100644 index 00000000..067fb9a2 --- /dev/null +++ b/services/notification/src/main/java/kr/magicbox/notification/adapter/in/web/dto/request/ReadNotificationsRequest.java @@ -0,0 +1,6 @@ +package kr.magicbox.notification.adapter.in.web.dto.request; + +import java.util.List; + +public record ReadNotificationsRequest(List notificationIds) { +} diff --git a/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/NotificationJpaAdapter.java b/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/NotificationJpaAdapter.java index f1382ff1..bb5cb0d0 100644 --- a/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/NotificationJpaAdapter.java +++ b/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/NotificationJpaAdapter.java @@ -4,10 +4,12 @@ import kr.magicbox.notification.adapter.out.persistence.repository.NotificationJpaRepository; import kr.magicbox.notification.application.port.out.NotificationRepositoryPort; import kr.magicbox.notification.domain.aggregate.Notification; +import kr.magicbox.notification.domain.enums.NotificationStatus; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Repository; import java.util.List; +import java.util.Optional; @Repository @RequiredArgsConstructor @@ -20,6 +22,24 @@ public void save(Notification notification) { notificationJpaRepository.save(NotificationEntity.from(notification)); } + @Override + public void update(Notification notification) { + notificationJpaRepository.findByIdAndUserId( + notification.getId().value(), notification.getUserId().value()) + .ifPresent(NotificationEntity::markRead); + } + + @Override + public void updateAllByIdsAndUserId(List notificationIds, Long userId) { + notificationJpaRepository.updateStatusByIdsAndUserId(notificationIds, userId, NotificationStatus.READ); + } + + @Override + public Optional findByIdAndUserId(Long notificationId, Long userId) { + return notificationJpaRepository.findByIdAndUserId(notificationId, userId) + .map(NotificationEntity::toDomain); + } + @Override public List findAllByUserId(Long userId) { return notificationJpaRepository.findAllByUserIdOrderByCreatedAtDesc(userId) diff --git a/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/repository/NotificationJpaRepository.java b/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/repository/NotificationJpaRepository.java index 2a55e0bf..b0a91df5 100644 --- a/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/repository/NotificationJpaRepository.java +++ b/services/notification/src/main/java/kr/magicbox/notification/adapter/out/persistence/repository/NotificationJpaRepository.java @@ -1,10 +1,20 @@ package kr.magicbox.notification.adapter.out.persistence.repository; import kr.magicbox.notification.adapter.out.persistence.entity.NotificationEntity; +import kr.magicbox.notification.domain.enums.NotificationStatus; 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; +import java.util.Optional; public interface NotificationJpaRepository extends JpaRepository { List findAllByUserIdOrderByCreatedAtDesc(Long userId); + Optional findByIdAndUserId(Long id, Long userId); + + @Modifying + @Query("UPDATE NotificationEntity n SET n.status = :status WHERE n.id IN :ids AND n.userId = :userId") + void updateStatusByIdsAndUserId(@Param("ids") List ids, @Param("userId") Long userId, @Param("status") NotificationStatus status); } diff --git a/services/notification/src/main/java/kr/magicbox/notification/application/dto/command/ReadNotificationCommand.java b/services/notification/src/main/java/kr/magicbox/notification/application/dto/command/ReadNotificationCommand.java new file mode 100644 index 00000000..05feccf8 --- /dev/null +++ b/services/notification/src/main/java/kr/magicbox/notification/application/dto/command/ReadNotificationCommand.java @@ -0,0 +1,12 @@ +package kr.magicbox.notification.application.dto.command; + +import java.util.List; + +public record ReadNotificationCommand( + List notificationIds, + Long userId +) { + public static ReadNotificationCommand of(List notificationIds, Long userId) { + return new ReadNotificationCommand(notificationIds, userId); + } +} diff --git a/services/notification/src/main/java/kr/magicbox/notification/application/port/in/ReadNotificationUseCase.java b/services/notification/src/main/java/kr/magicbox/notification/application/port/in/ReadNotificationUseCase.java new file mode 100644 index 00000000..6afd0fce --- /dev/null +++ b/services/notification/src/main/java/kr/magicbox/notification/application/port/in/ReadNotificationUseCase.java @@ -0,0 +1,7 @@ +package kr.magicbox.notification.application.port.in; + +import kr.magicbox.notification.application.dto.command.ReadNotificationCommand; + +public interface ReadNotificationUseCase { + void readAll(ReadNotificationCommand command); +} diff --git a/services/notification/src/main/java/kr/magicbox/notification/application/port/out/NotificationRepositoryPort.java b/services/notification/src/main/java/kr/magicbox/notification/application/port/out/NotificationRepositoryPort.java index 122eb316..e99c7af1 100644 --- a/services/notification/src/main/java/kr/magicbox/notification/application/port/out/NotificationRepositoryPort.java +++ b/services/notification/src/main/java/kr/magicbox/notification/application/port/out/NotificationRepositoryPort.java @@ -3,8 +3,12 @@ import kr.magicbox.notification.domain.aggregate.Notification; import java.util.List; +import java.util.Optional; public interface NotificationRepositoryPort { void save(Notification notification); + void update(Notification notification); + void updateAllByIdsAndUserId(List notificationIds, Long userId); + Optional findByIdAndUserId(Long notificationId, Long userId); List findAllByUserId(Long userId); } diff --git a/services/notification/src/main/java/kr/magicbox/notification/application/service/ReadNotificationService.java b/services/notification/src/main/java/kr/magicbox/notification/application/service/ReadNotificationService.java new file mode 100644 index 00000000..01bd57d5 --- /dev/null +++ b/services/notification/src/main/java/kr/magicbox/notification/application/service/ReadNotificationService.java @@ -0,0 +1,21 @@ +package kr.magicbox.notification.application.service; + +import kr.magicbox.notification.application.dto.command.ReadNotificationCommand; +import kr.magicbox.notification.application.port.in.ReadNotificationUseCase; +import kr.magicbox.notification.application.port.out.NotificationRepositoryPort; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +@Service +@RequiredArgsConstructor +public class ReadNotificationService implements ReadNotificationUseCase { + + private final NotificationRepositoryPort notificationRepositoryPort; + + @Transactional + @Override + public void readAll(ReadNotificationCommand command) { + notificationRepositoryPort.updateAllByIdsAndUserId(command.notificationIds(), command.userId()); + } +} diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/StockEventKafkaListener.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/StockEventKafkaListener.java index 619a901b..955a3fb4 100644 --- a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/StockEventKafkaListener.java +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/StockEventKafkaListener.java @@ -1,6 +1,8 @@ package kr.magicbox.orchestrator.adapter.in.kafka; import kr.magicbox.orchestrator.adapter.in.kafka.annotation.Idempotent; +import kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveFailedEvent; +import kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveSucceededEvent; import kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveFailedEvent; import kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveSucceededEvent; import kr.magicbox.orchestrator.application.port.in.HandleStockReserveFailedUseCase; @@ -37,4 +39,22 @@ public void handleStockReserveFailed(ConsumerRecord consumerRecord) { + log.info("[Inbox] stock-reserve-general-goods-succeeded 이벤트 수신. key={}", consumerRecord.key()); + GeneralGoodStockReserveSucceededEvent event = consumerRecord.value(); + handleStockReserveSucceededUseCase.handleStockReserveSucceeded( + event.orderId(), event.customerId(), event.totalAmount()); + } + + @Idempotent + @RetryableTopic + @KafkaListener(topics = "outbox.event.stock-reserve-general-goods-failed", groupId = "orchestrator-service") + public void handleGeneralGoodStockReserveFailed(ConsumerRecord consumerRecord) { + log.info("[Inbox] stock-reserve-general-goods-failed 이벤트 수신. key={}", consumerRecord.key()); + handleStockReserveFailedUseCase.handleStockReserveFailed(consumerRecord.value().orderId()); + } } diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveFailedEvent.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveFailedEvent.java new file mode 100644 index 00000000..5fb4594b --- /dev/null +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveFailedEvent.java @@ -0,0 +1,12 @@ +package kr.magicbox.orchestrator.adapter.in.kafka.event; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.Builder; + +import java.time.Instant; + +@Builder +public record GeneralGoodStockReserveFailedEvent( + @JsonProperty("order_id") Long orderId, + @JsonProperty("occurred_at") Instant occurredAt +) implements InboxEvent {} diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveSucceededEvent.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveSucceededEvent.java new file mode 100644 index 00000000..ac14e3b6 --- /dev/null +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/GeneralGoodStockReserveSucceededEvent.java @@ -0,0 +1,14 @@ +package kr.magicbox.orchestrator.adapter.in.kafka.event; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.Builder; + +import java.time.Instant; + +@Builder +public record GeneralGoodStockReserveSucceededEvent( + @JsonProperty("order_id") Long orderId, + @JsonProperty("customer_id") Long customerId, + @JsonProperty("total_amount") Long totalAmount, + @JsonProperty("occurred_at") Instant occurredAt +) implements InboxEvent {} diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/OrderPrepareEvent.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/OrderPrepareEvent.java index ec1c396b..2206aacd 100644 --- a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/OrderPrepareEvent.java +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/adapter/in/kafka/event/OrderPrepareEvent.java @@ -20,6 +20,7 @@ public record OrderPrepareEvent( public record ItemPayload( @JsonProperty("product_id") Long productId, @JsonProperty("quantity") int quantity, - @JsonProperty("unit_price") long unitPrice + @JsonProperty("unit_price") long unitPrice, + @JsonProperty("product_type") String productType ) {} } diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/application/service/HandleOrderPrepareService.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/application/service/HandleOrderPrepareService.java index 2e38eb35..a9fa1e1c 100644 --- a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/application/service/HandleOrderPrepareService.java +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/application/service/HandleOrderPrepareService.java @@ -3,7 +3,8 @@ import kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPrepareEvent; import kr.magicbox.orchestrator.application.port.in.HandleOrderPrepareUseCase; import kr.magicbox.orchestrator.application.port.out.OrchestratorOutboxPort; -import kr.magicbox.orchestrator.domain.event.StockReserveCommand; +import kr.magicbox.orchestrator.domain.event.StockReserveGeneralGoodCommand; +import kr.magicbox.orchestrator.domain.event.StockReserveReleaseCommand; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -14,7 +15,7 @@ /** * order.prepare 이벤트 처리 - * 플로우: order.prepare 수신 → stock-reserve 커맨드 발행 + * 플로우: order.prepare 수신 → productType에 따라 stock-reserve-release 또는 stock-reserve.general-good 커맨드 발행 */ @Slf4j @Service @@ -27,19 +28,39 @@ public class HandleOrderPrepareService implements HandleOrderPrepareUseCase { @Transactional public void handleOrderPrepare(Long orderId, Long customerId, Long sellerId, Long totalAmount, List items) { log.info("[Orchestrator] order.prepare 처리. orderId={}", orderId); - List commandItems = items == null ? List.of() : items.stream() - .map(i -> StockReserveCommand.ItemPayload.builder() - .productId(i.productId()) - .quantity(i.quantity()) - .unitPrice(i.unitPrice()) - .build()) - .toList(); - orchestratorOutboxPort.save(StockReserveCommand.builder() - .orderId(orderId) - .customerId(customerId) - .totalAmount(totalAmount) - .items(commandItems) - .occurredAt(Instant.now()) - .build()); + List safeItems = items == null ? List.of() : items; + String productType = safeItems.isEmpty() ? null : safeItems.get(0).productType(); + + if ("GENERAL_GOODS".equals(productType)) { + List commandItems = safeItems.stream() + .map(i -> StockReserveGeneralGoodCommand.ItemPayload.builder() + .productId(i.productId()) + .quantity(i.quantity()) + .unitPrice(i.unitPrice()) + .build()) + .toList(); + orchestratorOutboxPort.save(StockReserveGeneralGoodCommand.builder() + .orderId(orderId) + .customerId(customerId) + .totalAmount(totalAmount) + .items(commandItems) + .occurredAt(Instant.now()) + .build()); + } else { + List commandItems = safeItems.stream() + .map(i -> StockReserveReleaseCommand.ItemPayload.builder() + .productId(i.productId()) + .quantity(i.quantity()) + .unitPrice(i.unitPrice()) + .build()) + .toList(); + orchestratorOutboxPort.save(StockReserveReleaseCommand.builder() + .orderId(orderId) + .customerId(customerId) + .totalAmount(totalAmount) + .items(commandItems) + .occurredAt(Instant.now()) + .build()); + } } } diff --git a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/OrchestratorCommandEventType.java b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/OrchestratorCommandEventType.java index db9bca90..121d40ec 100644 --- a/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/OrchestratorCommandEventType.java +++ b/services/orchestrator/src/main/java/kr/magicbox/orchestrator/domain/event/OrchestratorCommandEventType.java @@ -6,7 +6,8 @@ @Getter @RequiredArgsConstructor public enum OrchestratorCommandEventType { - STOCK_RESERVE("stock-reserve"), + STOCK_RESERVE_RELEASE("stock-reserve-release"), + STOCK_RESERVE_GENERAL_GOOD("stock-reserve.general-good"), STOCK_RELEASE("stock-release"), PAYMENT_APPROVE("payment-approve"), ORDER_PREPARE_CONFIRMED("order-prepare-confirmed"), diff --git a/services/orchestrator/src/main/resources/application-dev.yml b/services/orchestrator/src/main/resources/application-dev.yml index 4f1ac1d6..e764f43c 100644 --- a/services/orchestrator/src/main/resources/application-dev.yml +++ b/services/orchestrator/src/main/resources/application-dev.yml @@ -16,7 +16,25 @@ spring: properties: spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer spring.json.trusted.packages: "kr.magicbox.orchestrator.adapter.in.kafka.event" - spring.json.type.mapping: order-prepare:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPrepareEvent,order-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderConfirmedEvent,order-cancel:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderCancelEvent,order-purchase-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPurchaseConfirmedEvent,order-auto-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderAutoConfirmedEvent,delivery-completed:kr.magicbox.orchestrator.adapter.in.kafka.event.DeliveryCompletedEvent,payment-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentSucceededEvent,payment-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentFailedEvent,payment-cancel-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelSucceededEvent,payment-cancel-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelFailedEvent,stock-reserve-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveSucceededEvent,stock-reserve-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveFailedEvent + spring.json.type.mapping: >- + order-prepare:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPrepareEvent, + order-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderConfirmedEvent, + order-cancel:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderCancelEvent, + order-purchase-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPurchaseConfirmedEvent, + order-auto-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderAutoConfirmedEvent, + payment-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentSucceededEvent, + payment-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentFailedEvent, + payment-cancel-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelSucceededEvent, + payment-cancel-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelFailedEvent, + delivery-completed:kr.magicbox.orchestrator.adapter.in.kafka.event.DeliveryCompletedEvent, + stock-reserve-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveSucceededEvent, + stock-reserve-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveFailedEvent, + stock-reserve-general-goods-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveSucceededEvent, + stock-reserve-general-goods-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveFailedEvent + spring.json.use.type.headers: false + producer: + key-serializer: org.apache.kafka.common.serialization.StringSerializer + value-serializer: org.springframework.kafka.support.serializer.JsonSerializer retry: topic: attempts: 3 diff --git a/services/orchestrator/src/main/resources/application-prod.yml b/services/orchestrator/src/main/resources/application-prod.yml index f130cca4..1dd73090 100644 --- a/services/orchestrator/src/main/resources/application-prod.yml +++ b/services/orchestrator/src/main/resources/application-prod.yml @@ -16,7 +16,25 @@ spring: properties: spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JacksonJsonDeserializer spring.json.trusted.packages: "kr.magicbox.orchestrator.adapter.in.kafka.event" - spring.json.type.mapping: order-prepare:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPrepareEvent,order-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderConfirmedEvent,order-cancel:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderCancelEvent,order-purchase-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPurchaseConfirmedEvent,order-auto-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderAutoConfirmedEvent,delivery-completed:kr.magicbox.orchestrator.adapter.in.kafka.event.DeliveryCompletedEvent,payment-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentSucceededEvent,payment-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentFailedEvent,payment-cancel-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelSucceededEvent,stock-reserve-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveSucceededEvent,stock-reserve-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveFailedEvent + spring.json.type.mapping: >- + order-prepare:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPrepareEvent, + order-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderConfirmedEvent, + order-cancel:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderCancelEvent, + order-purchase-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderPurchaseConfirmedEvent, + order-auto-confirmed:kr.magicbox.orchestrator.adapter.in.kafka.event.OrderAutoConfirmedEvent, + payment-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentSucceededEvent, + payment-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentFailedEvent, + payment-cancel-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelSucceededEvent, + payment-cancel-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.PaymentCancelFailedEvent, + delivery-completed:kr.magicbox.orchestrator.adapter.in.kafka.event.DeliveryCompletedEvent, + stock-reserve-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveSucceededEvent, + stock-reserve-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.StockReserveFailedEvent, + stock-reserve-general-goods-succeeded:kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveSucceededEvent, + stock-reserve-general-goods-failed:kr.magicbox.orchestrator.adapter.in.kafka.event.GeneralGoodStockReserveFailedEvent + spring.json.use.type.headers: false + producer: + key-serializer: org.apache.kafka.common.serialization.StringSerializer + value-serializer: org.springframework.kafka.support.serializer.JsonSerializer retry: topic: attempts: 5 diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateOrderRequest.java b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateOrderRequest.java new file mode 100644 index 00000000..0c3c330b --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateOrderRequest.java @@ -0,0 +1,38 @@ +package kr.magicbox.order.adapter.in.web.dto.request; + +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Positive; +import kr.magicbox.order.application.dto.command.CreateOrderCommand; + +import java.util.List; + +public record CreateOrderRequest( + @NotNull(message = "판매자 ID는 필수입니다.") @Positive(message = "판매자 ID는 양수여야 합니다.") Long sellerId, + @NotNull(message = "총 금액은 필수입니다.") @Positive(message = "총 금액은 양수여야 합니다.") Long totalAmount, + @Valid @NotNull(message = "배송지 정보는 필수입니다.") ShippingAddressRequest shippingAddress, + @NotEmpty(message = "주문 항목은 하나 이상이어야 합니다.") List<@Valid OrderLineRequest> orderLines +) { + public CreateOrderCommand toCommand(Long customerId) { + List lineCommands = orderLines.stream() + .map(line -> CreateOrderCommand.OrderLineCommand.builder() + .productId(line.productId()) + .sellerId(line.sellerId()) + .productName(line.productName()) + .quantity(line.quantity()) + .unitPrice(line.unitPrice()) + .productType(line.productType()) + .thumbnailUrl(line.thumbnailUrl()) + .build()) + .toList(); + + return CreateOrderCommand.builder() + .customerId(customerId) + .sellerId(sellerId) + .totalAmount(totalAmount) + .shippingAddress(shippingAddress.toDomain()) + .orderLines(lineCommands) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateReleaseOrderRequest.java b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateReleaseOrderRequest.java new file mode 100644 index 00000000..2b5f3c01 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/CreateReleaseOrderRequest.java @@ -0,0 +1,30 @@ +package kr.magicbox.order.adapter.in.web.dto.request; + +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Positive; +import kr.magicbox.order.application.dto.command.CreateReleaseOrderCommand; + +public record CreateReleaseOrderRequest( + @NotNull(message = "판매자 ID는 필수입니다.") @Positive(message = "판매자 ID는 양수여야 합니다.") Long sellerId, + @NotNull(message = "릴리즈 ID는 필수입니다.") @Positive(message = "릴리즈 ID는 양수여야 합니다.") Long releaseId, + @NotBlank(message = "구매 토큰은 필수입니다.") String purchaseToken, + @NotBlank(message = "상품명은 필수입니다.") String productName, + @NotNull(message = "단가는 필수입니다.") @Positive(message = "단가는 양수여야 합니다.") Long unitPrice, + String thumbnailUrl, + @Valid @NotNull(message = "배송지 정보는 필수입니다.") ShippingAddressRequest shippingAddress +) { + public CreateReleaseOrderCommand toCommand(Long customerId) { + return CreateReleaseOrderCommand.builder() + .customerId(customerId) + .sellerId(sellerId) + .releaseId(releaseId) + .purchaseToken(purchaseToken) + .productName(productName) + .unitPrice(unitPrice) + .thumbnailUrl(thumbnailUrl) + .shippingAddress(shippingAddress.toDomain()) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/OrderLineRequest.java b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/OrderLineRequest.java new file mode 100644 index 00000000..9aad8123 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/request/OrderLineRequest.java @@ -0,0 +1,17 @@ +package kr.magicbox.order.adapter.in.web.dto.request; + +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Positive; +import kr.magicbox.order.domain.enums.ProductType; + +public record OrderLineRequest( + @NotNull(message = "상품 ID는 필수입니다.") @Positive(message = "상품 ID는 양수여야 합니다.") Long productId, + @NotNull(message = "판매자 ID는 필수입니다.") @Positive(message = "판매자 ID는 양수여야 합니다.") Long sellerId, + @NotBlank(message = "상품명은 필수입니다.") String productName, + @NotNull(message = "수량은 필수입니다.") @Min(value = 1, message = "수량은 1 이상이어야 합니다.") Integer quantity, + @NotNull(message = "단가는 필수입니다.") @Min(value = 0, message = "단가는 0 이상이어야 합니다.") Long unitPrice, + @NotNull(message = "상품 타입은 필수입니다.") ProductType productType, + String thumbnailUrl +) {} diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/response/OrderLineResponse.java b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/response/OrderLineResponse.java new file mode 100644 index 00000000..0230deb9 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/in/web/dto/response/OrderLineResponse.java @@ -0,0 +1,25 @@ +package kr.magicbox.order.adapter.in.web.dto.response; + +import kr.magicbox.order.application.dto.result.OrderLineResult; +import lombok.Builder; + +@Builder +public record OrderLineResponse( + Long orderLineId, + Long productId, + String productName, + int quantity, + long unitPrice, + String thumbnailUrl +) { + public static OrderLineResponse from(OrderLineResult result) { + return OrderLineResponse.builder() + .orderLineId(result.orderLineId()) + .productId(result.productId()) + .productName(result.productName()) + .quantity(result.quantity()) + .unitPrice(result.unitPrice()) + .thumbnailUrl(result.thumbnailUrl()) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/entity/OrderLineEntity.java b/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/entity/OrderLineEntity.java new file mode 100644 index 00000000..3e6824c9 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/entity/OrderLineEntity.java @@ -0,0 +1,66 @@ +package kr.magicbox.order.adapter.out.persistence.entity; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.Table; +import kr.magicbox.order.domain.enums.OrderLineDeliveryStatus; +import kr.magicbox.order.domain.enums.ProductType; +import lombok.AccessLevel; +import lombok.Builder; +import lombok.Getter; +import lombok.NoArgsConstructor; + +@Getter +@NoArgsConstructor(access = AccessLevel.PROTECTED) +@Entity +@Table(name = "order_line") +public class OrderLineEntity extends BaseEntity { + + @Column(name = "order_id", nullable = false) + private Long orderId; + + @Column(name = "product_id", nullable = false) + private Long productId; + + @Column(name = "seller_id", nullable = false) + private Long sellerId; + + @Column(name = "product_name", nullable = false) + private String productName; + + @Column(name = "quantity", nullable = false) + private Integer quantity; + + @Column(name = "unit_price", nullable = false) + private Long unitPrice; + + @Enumerated(EnumType.STRING) + @Column(name = "delivery_status", nullable = false) + private OrderLineDeliveryStatus deliveryStatus; + + @Enumerated(EnumType.STRING) + @Column(name = "product_type", nullable = false) + private ProductType productType; + + @Column(name = "thumbnail_url") + private String thumbnailUrl; + + @Builder + public OrderLineEntity(Long orderId, Long productId, Long sellerId, String productName, Integer quantity, Long unitPrice, OrderLineDeliveryStatus deliveryStatus, ProductType productType, String thumbnailUrl) { + this.orderId = orderId; + this.productId = productId; + this.sellerId = sellerId; + this.productName = productName; + this.quantity = quantity; + this.unitPrice = unitPrice; + this.deliveryStatus = deliveryStatus != null ? deliveryStatus : OrderLineDeliveryStatus.PENDING; + this.productType = productType; + this.thumbnailUrl = thumbnailUrl; + } + + public void updateDeliveryStatus(OrderLineDeliveryStatus deliveryStatus) { + this.deliveryStatus = deliveryStatus; + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/mapper/OrderMapper.java b/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/mapper/OrderMapper.java new file mode 100644 index 00000000..9644a79e --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/adapter/out/persistence/mapper/OrderMapper.java @@ -0,0 +1,83 @@ +package kr.magicbox.order.adapter.out.persistence.mapper; + +import kr.magicbox.order.adapter.out.persistence.entity.OrderEntity; +import kr.magicbox.order.adapter.out.persistence.entity.OrderLineEntity; +import kr.magicbox.order.domain.aggregate.Order; +import kr.magicbox.order.domain.aggregate.OrderLine; +import kr.magicbox.order.domain.vo.OrderId; +import kr.magicbox.order.domain.vo.OrderLineId; +import kr.magicbox.order.domain.vo.ShippingAddress; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +public class OrderMapper { + + public OrderEntity toEntity(Order domain) { + ShippingAddress addr = domain.getShippingAddress(); + return OrderEntity.builder() + .customerId(domain.getCustomerId()) + .sellerId(domain.getSellerId()) + .status(domain.getStatus()) + .totalAmount(domain.getTotalAmount()) + .recipient(addr.recipient()) + .phone(addr.phone()) + .zipCode(addr.zipCode()) + .address1(addr.address1()) + .address2(addr.address2()) + .build(); + } + + public OrderLineEntity toLineEntity(Long orderId, OrderLine domain) { + return OrderLineEntity.builder() + .orderId(orderId) + .productId(domain.getProductId()) + .sellerId(domain.getSellerId()) + .productName(domain.getProductName()) + .quantity(domain.getQuantity()) + .unitPrice(domain.getUnitPrice()) + .deliveryStatus(domain.getDeliveryStatus()) + .productType(domain.getProductType()) + .thumbnailUrl(domain.getThumbnailUrl()) + .build(); + } + + public Order toDomain(OrderEntity entity, List lineEntities) { + List orderLines = lineEntities.stream() + .map(this::toLineDomain) + .toList(); + + return Order.reconstructBuilder() + .id(OrderId.of(entity.getId())) + .customerId(entity.getCustomerId()) + .sellerId(entity.getSellerId()) + .status(entity.getStatus()) + .totalAmount(entity.getTotalAmount()) + .shippingAddress(ShippingAddress.of( + entity.getRecipient(), + entity.getPhone(), + entity.getZipCode(), + entity.getAddress1(), + entity.getAddress2() + )) + .orderLines(orderLines) + .createdAt(entity.getCreatedAt()) + .updatedAt(entity.getUpdatedAt()) + .build(); + } + + private OrderLine toLineDomain(OrderLineEntity entity) { + return OrderLine.reconstructBuilder() + .id(OrderLineId.of(entity.getId())) + .productId(entity.getProductId()) + .sellerId(entity.getSellerId()) + .productName(entity.getProductName()) + .quantity(entity.getQuantity()) + .unitPrice(entity.getUnitPrice()) + .deliveryStatus(entity.getDeliveryStatus()) + .productType(entity.getProductType()) + .thumbnailUrl(entity.getThumbnailUrl()) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateOrderCommand.java b/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateOrderCommand.java new file mode 100644 index 00000000..aaba97bd --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateOrderCommand.java @@ -0,0 +1,27 @@ +package kr.magicbox.order.application.dto.command; + +import kr.magicbox.order.domain.enums.ProductType; +import kr.magicbox.order.domain.vo.ShippingAddress; +import lombok.Builder; + +import java.util.List; + +@Builder +public record CreateOrderCommand( + Long customerId, + Long sellerId, + Long totalAmount, + ShippingAddress shippingAddress, + List orderLines +) { + @Builder + public record OrderLineCommand( + Long productId, + Long sellerId, + String productName, + Integer quantity, + Long unitPrice, + ProductType productType, + String thumbnailUrl + ) {} +} diff --git a/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateReleaseOrderCommand.java b/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateReleaseOrderCommand.java new file mode 100644 index 00000000..2a99d2a9 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/dto/command/CreateReleaseOrderCommand.java @@ -0,0 +1,16 @@ +package kr.magicbox.order.application.dto.command; + +import kr.magicbox.order.domain.vo.ShippingAddress; +import lombok.Builder; + +@Builder +public record CreateReleaseOrderCommand( + Long customerId, + Long sellerId, + Long releaseId, + String purchaseToken, + String productName, + Long unitPrice, + String thumbnailUrl, + ShippingAddress shippingAddress +) {} diff --git a/services/order/src/main/java/kr/magicbox/order/application/dto/result/OrderLineResult.java b/services/order/src/main/java/kr/magicbox/order/application/dto/result/OrderLineResult.java new file mode 100644 index 00000000..e53f27f0 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/dto/result/OrderLineResult.java @@ -0,0 +1,13 @@ +package kr.magicbox.order.application.dto.result; + +import lombok.Builder; + +@Builder +public record OrderLineResult( + Long orderLineId, + Long productId, + String productName, + int quantity, + long unitPrice, + String thumbnailUrl +) {} diff --git a/services/order/src/main/java/kr/magicbox/order/application/service/CreateOrderService.java b/services/order/src/main/java/kr/magicbox/order/application/service/CreateOrderService.java new file mode 100644 index 00000000..207ef954 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/service/CreateOrderService.java @@ -0,0 +1,70 @@ +package kr.magicbox.order.application.service; + +import kr.magicbox.order.application.dto.command.CreateOrderCommand; +import kr.magicbox.order.application.dto.result.CreateOrderResult; +import kr.magicbox.order.application.dto.result.OrderLineResult; +import kr.magicbox.order.application.port.in.CreateOrderUseCase; +import kr.magicbox.order.application.port.out.OrderOutboxPort; +import kr.magicbox.order.application.port.out.OrderRepositoryPort; +import kr.magicbox.order.domain.aggregate.Order; +import kr.magicbox.order.domain.aggregate.OrderLine; +import kr.magicbox.order.domain.event.OrderPrepareEvent; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.List; + +@Service +@RequiredArgsConstructor +public class CreateOrderService implements CreateOrderUseCase { + + private final OrderRepositoryPort orderRepositoryPort; + private final OrderOutboxPort orderOutboxPort; + + @Transactional + @Override + public CreateOrderResult createOrder(CreateOrderCommand command) { + List orderLines = command.orderLines().stream() + .map(line -> OrderLine.createBuilder() + .productId(line.productId()) + .sellerId(line.sellerId()) + .productName(line.productName()) + .quantity(line.quantity()) + .unitPrice(line.unitPrice()) + .productType(line.productType()) + .thumbnailUrl(line.thumbnailUrl()) + .build()) + .toList(); + + Order order = Order.createBuilder() + .customerId(command.customerId()) + .sellerId(command.sellerId()) + .totalAmount(command.totalAmount()) + .shippingAddress(command.shippingAddress()) + .orderLines(orderLines) + .build(); + + Order savedOrder = orderRepositoryPort.save(order); + Long savedOrderId = savedOrder.getId().value(); + orderOutboxPort.save(OrderPrepareEvent.from(savedOrder)); + + List orderLineResults = savedOrder.getOrderLines().stream() + .map(line -> OrderLineResult.builder() + .orderLineId(line.getId().value()) + .productId(line.getProductId()) + .productName(line.getProductName()) + .quantity(line.getQuantity()) + .unitPrice(line.getUnitPrice()) + .thumbnailUrl(line.getThumbnailUrl()) + .build()) + .toList(); + + return CreateOrderResult.builder() + .orderId(savedOrderId) + .sellerId(command.sellerId()) + .totalAmount(command.totalAmount()) + .orderLines(orderLineResults) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/application/service/CreateReleaseOrderService.java b/services/order/src/main/java/kr/magicbox/order/application/service/CreateReleaseOrderService.java new file mode 100644 index 00000000..698708d6 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/service/CreateReleaseOrderService.java @@ -0,0 +1,85 @@ +package kr.magicbox.order.application.service; + +import kr.magicbox.order.application.dto.command.CreateReleaseOrderCommand; +import kr.magicbox.order.application.dto.result.CreateOrderResult; +import kr.magicbox.order.application.dto.result.OrderLineResult; +import kr.magicbox.order.application.port.in.CreateReleaseOrderUseCase; +import kr.magicbox.order.application.port.out.OrderOutboxPort; +import kr.magicbox.order.application.port.out.OrderRepositoryPort; +import kr.magicbox.order.application.port.out.PurchaseTokenValidationPort; +import kr.magicbox.order.domain.aggregate.Order; +import kr.magicbox.order.domain.aggregate.OrderLine; +import kr.magicbox.order.domain.enums.ProductType; +import kr.magicbox.order.domain.event.OrderPrepareEvent; +import kr.magicbox.order.domain.event.ReleaseSoldQuantityIncreaseEvent; +import kr.magicbox.order.domain.exception.InvalidPurchaseTokenException; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.List; + +@Service +@RequiredArgsConstructor +public class CreateReleaseOrderService implements CreateReleaseOrderUseCase { + + private final PurchaseTokenValidationPort purchaseTokenValidationPort; + private final OrderRepositoryPort orderRepositoryPort; + private final OrderOutboxPort orderOutboxPort; + + @Override + public CreateOrderResult createReleaseOrder(CreateReleaseOrderCommand command) { + boolean valid = purchaseTokenValidationPort.validate( + command.releaseId(), command.customerId(), command.purchaseToken()); + if (!valid) { + throw new InvalidPurchaseTokenException(); + } + + return saveOrderWithOutbox(command); + } + + @Transactional + protected CreateOrderResult saveOrderWithOutbox(CreateReleaseOrderCommand command) { + OrderLine orderLine = OrderLine.createBuilder() + .productId(command.releaseId()) + .sellerId(command.sellerId()) + .productName(command.productName()) + .quantity(1) + .unitPrice(command.unitPrice()) + .productType(ProductType.RELEASE) + .thumbnailUrl(command.thumbnailUrl()) + .build(); + + Order order = Order.createBuilder() + .customerId(command.customerId()) + .sellerId(command.sellerId()) + .totalAmount(command.unitPrice()) + .shippingAddress(command.shippingAddress()) + .orderLines(List.of(orderLine)) + .build(); + + Order savedOrder = orderRepositoryPort.save(order); + Long savedOrderId = savedOrder.getId().value(); + orderOutboxPort.save(OrderPrepareEvent.from(savedOrder)); + orderOutboxPort.save(ReleaseSoldQuantityIncreaseEvent.of(command.releaseId())); + + OrderLineResult orderLineResult = savedOrder.getOrderLines().stream() + .findFirst() + .map(line -> OrderLineResult.builder() + .orderLineId(line.getId().value()) + .productId(line.getProductId()) + .productName(line.getProductName()) + .quantity(line.getQuantity()) + .unitPrice(line.getUnitPrice()) + .thumbnailUrl(line.getThumbnailUrl()) + .build()) + .orElseThrow(); + + return CreateOrderResult.builder() + .orderId(savedOrderId) + .sellerId(command.sellerId()) + .totalAmount(command.unitPrice()) + .orderLines(List.of(orderLineResult)) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/application/service/OrderResultMapper.java b/services/order/src/main/java/kr/magicbox/order/application/service/OrderResultMapper.java new file mode 100644 index 00000000..1ac2712b --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/application/service/OrderResultMapper.java @@ -0,0 +1,37 @@ +package kr.magicbox.order.application.service; + +import kr.magicbox.order.application.dto.result.OrderLineResult; +import kr.magicbox.order.application.dto.result.OrderResult; +import kr.magicbox.order.application.dto.result.ShippingAddressResult; +import kr.magicbox.order.domain.aggregate.Order; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +public class OrderResultMapper { + + public OrderResult toResult(Order order) { + List lineResults = order.getOrderLines().stream() + .map(line -> OrderLineResult.builder() + .orderLineId(line.getId() != null ? line.getId().value() : null) + .productId(line.getProductId()) + .productName(line.getProductName()) + .quantity(line.getQuantity()) + .unitPrice(line.getUnitPrice()) + .thumbnailUrl(line.getThumbnailUrl()) + .build()) + .toList(); + + return OrderResult.builder() + .orderId(order.getId().value()) + .customerId(order.getCustomerId()) + .sellerId(order.getSellerId()) + .status(order.getStatus()) + .totalAmount(order.getTotalAmount()) + .shippingAddress(ShippingAddressResult.from(order.getShippingAddress())) + .orderLines(lineResults) + .createdAt(order.getCreatedAt()) + .build(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/domain/aggregate/OrderLine.java b/services/order/src/main/java/kr/magicbox/order/domain/aggregate/OrderLine.java new file mode 100644 index 00000000..d1083cd2 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/domain/aggregate/OrderLine.java @@ -0,0 +1,134 @@ +package kr.magicbox.order.domain.aggregate; + +import kr.magicbox.order.domain.enums.OrderLineDeliveryStatus; +import kr.magicbox.order.domain.enums.ProductType; +import kr.magicbox.order.domain.exception.InvalidFieldException; +import kr.magicbox.order.domain.exception.OrderLineComplainNotAllowedException; +import kr.magicbox.order.domain.exception.OrderStatusConflictException; +import kr.magicbox.order.domain.vo.OrderLineId; +import lombok.Builder; +import lombok.Getter; + +@Getter +public class OrderLine { + + private final OrderLineId id; + private final Long productId; + private final Long sellerId; + private final String productName; + private final Integer quantity; + private final Long unitPrice; + private final ProductType productType; + private final String thumbnailUrl; + private OrderLineDeliveryStatus deliveryStatus; + + @Builder(builderMethodName = "createBuilder", builderClassName = "CreateBuilder") + public OrderLine(Long productId, Long sellerId, String productName, Integer quantity, Long unitPrice, ProductType productType, String thumbnailUrl) { + validateCreate(productId, quantity, unitPrice); + this.id = null; + this.productId = productId; + this.sellerId = sellerId; + this.productName = productName; + this.quantity = quantity; + this.unitPrice = unitPrice; + this.productType = productType; + this.thumbnailUrl = thumbnailUrl; + this.deliveryStatus = OrderLineDeliveryStatus.PENDING; + } + + @Builder(builderMethodName = "reconstructBuilder", builderClassName = "ReconstructBuilder") + public OrderLine(OrderLineId id, Long productId, Long sellerId, String productName, + Integer quantity, Long unitPrice, ProductType productType, String thumbnailUrl, OrderLineDeliveryStatus deliveryStatus) { + validateReconstruct(id, productId, sellerId, productName, quantity, unitPrice, deliveryStatus); + this.id = id; + this.productId = productId; + this.sellerId = sellerId; + this.productName = productName; + this.quantity = quantity; + this.unitPrice = unitPrice; + this.productType = productType; + this.thumbnailUrl = thumbnailUrl; + this.deliveryStatus = deliveryStatus; + } + + private void validateCreate(Long productId, Integer quantity, Long unitPrice) { + if (productId == null || productId <= 0) throw new InvalidFieldException("상품 ID는 양수여야 합니다."); + if (quantity == null || quantity <= 0) throw new InvalidFieldException("주문 수량은 1 이상이어야 합니다."); + if (unitPrice == null || unitPrice < 0) throw new InvalidFieldException("단가는 0 이상이어야 합니다."); + } + + private void validateReconstruct(OrderLineId id, Long productId, Long sellerId, String productName, + Integer quantity, Long unitPrice, OrderLineDeliveryStatus deliveryStatus) { + if (id == null) throw new InvalidFieldException("주문 라인 ID는 필수입니다."); + if (productId == null || productId <= 0) throw new InvalidFieldException("상품 ID는 양수여야 합니다."); + if (sellerId == null || sellerId <= 0) throw new InvalidFieldException("판매자 ID는 양수여야 합니다."); + if (productName == null || productName.isBlank()) throw new InvalidFieldException("상품명은 필수입니다."); + if (quantity == null || quantity <= 0) throw new InvalidFieldException("주문 수량은 1 이상이어야 합니다."); + if (unitPrice == null || unitPrice < 0) throw new InvalidFieldException("단가는 0 이상이어야 합니다."); + if (deliveryStatus == null) throw new InvalidFieldException("배송 상태는 필수입니다."); + } + + public long lineTotal() { + return unitPrice * quantity; + } + + public void prepare() { + if (this.deliveryStatus != OrderLineDeliveryStatus.PENDING) { + throw new OrderStatusConflictException("준비 처리할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.PREPARING; + } + + public void confirm() { + if (this.deliveryStatus != OrderLineDeliveryStatus.PREPARING) { + throw new OrderStatusConflictException("확정 처리할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.CONFIRMED; + } + + public void startDelivery() { + if (this.deliveryStatus != OrderLineDeliveryStatus.CONFIRMED) { + throw new OrderStatusConflictException("배송 시작할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.SHIPPING; + } + + public void completeDelivery() { + if (this.deliveryStatus != OrderLineDeliveryStatus.SHIPPING) { + throw new OrderStatusConflictException("배송 완료 처리할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.SHIPPED; + } + + public void complain() { + if (this.deliveryStatus != OrderLineDeliveryStatus.SHIPPING) { + throw new OrderLineComplainNotAllowedException(); + } + this.deliveryStatus = OrderLineDeliveryStatus.COMPLAINING; + } + + public void requestCancel() { + if (this.deliveryStatus == OrderLineDeliveryStatus.SHIPPED + || this.deliveryStatus == OrderLineDeliveryStatus.COMPLAINING + || this.deliveryStatus == OrderLineDeliveryStatus.CANCEL_REQUESTED + || this.deliveryStatus == OrderLineDeliveryStatus.CANCELLED) { + throw new OrderStatusConflictException("취소 요청할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.CANCEL_REQUESTED; + } + + public void completeCancellation() { + if (this.deliveryStatus != OrderLineDeliveryStatus.CANCEL_REQUESTED) { + throw new OrderStatusConflictException("취소 완료 처리할 수 없는 상태입니다: " + this.deliveryStatus); + } + this.deliveryStatus = OrderLineDeliveryStatus.CANCELLED; + } + + public boolean isCancelRequested() { + return this.deliveryStatus == OrderLineDeliveryStatus.CANCEL_REQUESTED; + } + + public boolean isAtLeast(OrderLineDeliveryStatus target) { + return this.deliveryStatus.ordinal() >= target.ordinal(); + } +} diff --git a/services/order/src/main/java/kr/magicbox/order/domain/enums/ProductType.java b/services/order/src/main/java/kr/magicbox/order/domain/enums/ProductType.java new file mode 100644 index 00000000..6c2b4a93 --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/domain/enums/ProductType.java @@ -0,0 +1,6 @@ +package kr.magicbox.order.domain.enums; + +public enum ProductType { + RELEASE, + GENERAL_GOODS +} diff --git a/services/order/src/main/java/kr/magicbox/order/domain/event/OrderPrepareEvent.java b/services/order/src/main/java/kr/magicbox/order/domain/event/OrderPrepareEvent.java new file mode 100644 index 00000000..a269fe3f --- /dev/null +++ b/services/order/src/main/java/kr/magicbox/order/domain/event/OrderPrepareEvent.java @@ -0,0 +1,86 @@ +package kr.magicbox.order.domain.event; + +import com.fasterxml.jackson.annotation.JsonProperty; +import kr.magicbox.order.domain.aggregate.Order; +import kr.magicbox.order.domain.enums.ProductType; +import kr.magicbox.order.domain.vo.ShippingAddress; +import lombok.Builder; + +import java.time.Instant; +import java.util.List; + +@Builder +public record OrderPrepareEvent( + @JsonProperty("event_id") Long eventId, + @JsonProperty("order_id") Long orderId, + @JsonProperty("customer_id") Long customerId, + @JsonProperty("seller_id") Long sellerId, + @JsonProperty("items") List items, + @JsonProperty("total_amount") long totalAmount, + @JsonProperty("shipping_address") ShippingAddressPayload shippingAddress, + @JsonProperty("occurred_at") Instant occurredAt +) implements OrderDomainEvent { + + public static OrderPrepareEvent from(Order order) { + Long orderId = order.getId().value(); + List items = order.getOrderLines().stream() + .map(line -> OrderItemPayload.builder() + .orderLineId(line.getId().value()) + .productId(line.getProductId()) + .quantity(line.getQuantity()) + .unitPrice(line.getUnitPrice()) + .productName(line.getProductName()) + .productType(line.getProductType()) + .build()) + .toList(); + + ShippingAddress addr = order.getShippingAddress(); + ShippingAddressPayload shippingAddress = ShippingAddressPayload.builder() + .recipient(addr.recipient()) + .phone(addr.phone()) + .zipCode(addr.zipCode()) + .address1(addr.address1()) + .address2(addr.address2()) + .build(); + + return OrderPrepareEvent.builder() + .eventId(orderId) + .orderId(orderId) + .customerId(order.getCustomerId()) + .sellerId(order.getSellerId()) + .items(items) + .totalAmount(order.getTotalAmount()) + .shippingAddress(shippingAddress) + .occurredAt(Instant.now()) + .build(); + } + + @Override + public String key() { + return orderId.toString(); + } + + @Override + public OrderDomainEventType eventType() { + return OrderDomainEventType.ORDER_PREPARE; + } + + @Builder + public record OrderItemPayload( + @JsonProperty("order_line_id") Long orderLineId, + @JsonProperty("product_id") Long productId, + @JsonProperty("quantity") int quantity, + @JsonProperty("unit_price") long unitPrice, + @JsonProperty("product_name") String productName, + @JsonProperty("product_type") ProductType productType + ) {} + + @Builder + public record ShippingAddressPayload( + @JsonProperty("recipient") String recipient, + @JsonProperty("phone") String phone, + @JsonProperty("zip_code") String zipCode, + @JsonProperty("address1") String address1, + @JsonProperty("address2") String address2 + ) {} +} diff --git a/services/release/src/main/java/kr/magicbox/release/adapter/in/kafka/OrchestratorCommandKafkaListener.java b/services/release/src/main/java/kr/magicbox/release/adapter/in/kafka/OrchestratorCommandKafkaListener.java index 43406da1..f3359bb8 100644 --- a/services/release/src/main/java/kr/magicbox/release/adapter/in/kafka/OrchestratorCommandKafkaListener.java +++ b/services/release/src/main/java/kr/magicbox/release/adapter/in/kafka/OrchestratorCommandKafkaListener.java @@ -3,11 +3,13 @@ import kr.magicbox.release.adapter.in.kafka.annotation.Idempotent; import kr.magicbox.release.adapter.in.kafka.event.StockReserveCommandEvent; import kr.magicbox.release.application.port.in.HandleStockReserveUseCase; +import kr.magicbox.release.global.exception.BusinessException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.RetryableTopic; +import org.springframework.kafka.retrytopic.DltStrategy; import org.springframework.stereotype.Component; @Slf4j @@ -18,10 +20,10 @@ public class OrchestratorCommandKafkaListener { private final HandleStockReserveUseCase handleStockReserveUseCase; @Idempotent - @RetryableTopic - @KafkaListener(topics = "outbox.event.stock-reserve", groupId = "release-service") + @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {BusinessException.class}) + @KafkaListener(topics = "outbox.event.stock-reserve-release", groupId = "release-service") public void handleStockReserve(ConsumerRecord consumerRecord) { - log.info("[Inbox] stock-reserve 커맨드 수신. key={}", consumerRecord.key()); + log.info("[Inbox] stock-reserve-release 커맨드 수신. key={}", consumerRecord.key()); handleStockReserveUseCase.handleStockReserve(consumerRecord.value()); } } diff --git a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/ReleaseJpaAdapter.java b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/ReleaseJpaAdapter.java index ad1ccdc0..9503cbdc 100644 --- a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/ReleaseJpaAdapter.java +++ b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/ReleaseJpaAdapter.java @@ -80,6 +80,17 @@ public List findAllByCursor(Long cursorId, int size) { .toList(); } + @Override + public Long findCreatorIdById(Long id) { + return releaseJpaRepository.findCreatorIdById(id) + .orElseThrow(ReleaseNotFoundException::new); + } + + @Override + public int increaseSoldQuantity(Long id, int quantity) { + return releaseJpaRepository.increaseSoldQuantity(id, quantity); + } + private void syncMediaList(ReleaseEntity entity, List newMediaList) { Set incomingKeys = newMediaList.stream() .map(m -> m.getMediaUrl() + "|" + m.getSortOrder()) diff --git a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/entity/ReleaseEntity.java b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/entity/ReleaseEntity.java index 59443d4d..85d04527 100644 --- a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/entity/ReleaseEntity.java +++ b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/entity/ReleaseEntity.java @@ -18,9 +18,6 @@ @Table(name = "releases") public class ReleaseEntity extends BaseEntity { - @Version - private Long version; - @Column(name = "creator_id", nullable = false) private Long creatorId; diff --git a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/repository/ReleaseJpaRepository.java b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/repository/ReleaseJpaRepository.java index bf318636..b7505f34 100644 --- a/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/repository/ReleaseJpaRepository.java +++ b/services/release/src/main/java/kr/magicbox/release/adapter/out/persistence/repository/ReleaseJpaRepository.java @@ -4,6 +4,7 @@ import kr.magicbox.release.domain.enums.ReleaseStatus; import org.springframework.data.domain.Pageable; 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; @@ -28,4 +29,13 @@ public interface ReleaseJpaRepository extends JpaRepository @Query("SELECT DISTINCT r FROM ReleaseEntity r LEFT JOIN FETCH r.releaseMediaList ORDER BY r.id DESC") List findAllByOrderByIdDesc(Pageable pageable); + + @Query("SELECT r.creatorId FROM ReleaseEntity r WHERE r.id = :id") + Optional findCreatorIdById(@Param("id") Long id); + + @Modifying + @Query("UPDATE ReleaseEntity r SET r.soldQuantity = r.soldQuantity + :quantity, " + + "r.status = CASE WHEN r.soldQuantity + :quantity >= r.limitedQuantity THEN 'SOLD_OUT' ELSE r.status END " + + "WHERE r.id = :id AND r.status = 'ON_SALE' AND r.soldQuantity + :quantity <= r.limitedQuantity") + int increaseSoldQuantity(@Param("id") Long id, @Param("quantity") int quantity); } diff --git a/services/release/src/main/java/kr/magicbox/release/application/port/out/ReleaseRepositoryPort.java b/services/release/src/main/java/kr/magicbox/release/application/port/out/ReleaseRepositoryPort.java index 9c466103..4ab7bed7 100644 --- a/services/release/src/main/java/kr/magicbox/release/application/port/out/ReleaseRepositoryPort.java +++ b/services/release/src/main/java/kr/magicbox/release/application/port/out/ReleaseRepositoryPort.java @@ -22,4 +22,8 @@ public interface ReleaseRepositoryPort { List findScheduledBefore(Instant scheduledAt, int limit); List findAllByCursor(Long cursorId, int size); + + Long findCreatorIdById(Long id); + + int increaseSoldQuantity(Long id, int quantity); } diff --git a/services/release/src/main/java/kr/magicbox/release/application/service/HandleStockReserveService.java b/services/release/src/main/java/kr/magicbox/release/application/service/HandleStockReserveService.java index e98ef742..4aa80060 100644 --- a/services/release/src/main/java/kr/magicbox/release/application/service/HandleStockReserveService.java +++ b/services/release/src/main/java/kr/magicbox/release/application/service/HandleStockReserveService.java @@ -4,12 +4,9 @@ import kr.magicbox.release.application.port.in.HandleStockReserveUseCase; import kr.magicbox.release.application.port.out.ReleaseOutboxPort; import kr.magicbox.release.application.port.out.ReleaseRepositoryPort; -import kr.magicbox.release.domain.aggregate.Release; import kr.magicbox.release.domain.event.StockReserveFailedEvent; import kr.magicbox.release.domain.event.StockReserveSucceededEvent; import kr.magicbox.release.domain.exception.ReleaseNotFoundException; -import kr.magicbox.release.domain.exception.ReleaseStatusConflictException; -import kr.magicbox.release.domain.vo.ReleaseId; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -44,26 +41,23 @@ public void handleStockReserve(StockReserveCommandEvent event) { List succeededItems = new ArrayList<>(); for (StockReserveCommandEvent.ItemPayload item : event.items()) { Long releaseId = item.productId(); - try { - Release release = releaseRepositoryPort.findById(ReleaseId.of(releaseId)); - release.increaseSoldQuantity(); - releaseRepositoryPort.update(release); - succeededItems.add(StockReserveSucceededEvent.ItemPayload.builder() - .orderLineId(item.orderLineId()) - .sellerId(release.getCreatorId().value()) - .amount((long) item.quantity() * item.unitPrice()) - .build()); - } catch (ReleaseNotFoundException | ReleaseStatusConflictException e) { - log.warn("[StockReserve] 재고 예약 실패. orderId={}, releaseId={}, reason={}", - event.orderId(), releaseId, e.getMessage()); + int updated = releaseRepositoryPort.increaseSoldQuantity(releaseId, item.quantity()); + if (updated == 0) { + log.warn("[StockReserve] 재고 예약 실패. orderId={}, releaseId={}", event.orderId(), releaseId); releaseOutboxPort.save(StockReserveFailedEvent.builder() .orderId(event.orderId()) .customerId(event.customerId()) - .reason(e.getMessage()) + .reason("재고 부족 또는 판매 중 상태 아님") .occurredAt(Instant.now()) .build()); return; } + Long sellerId = releaseRepositoryPort.findCreatorIdById(releaseId); + succeededItems.add(StockReserveSucceededEvent.ItemPayload.builder() + .orderLineId(item.orderLineId()) + .sellerId(sellerId) + .amount((long) item.quantity() * item.unitPrice()) + .build()); } log.info("[StockReserve] 재고 예약 성공. orderId={}", event.orderId()); diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/LogoutKafkaListener.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/LogoutKafkaListener.java index 6341fe7f..00f6fe02 100644 --- a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/LogoutKafkaListener.java +++ b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/LogoutKafkaListener.java @@ -2,7 +2,7 @@ import tools.jackson.databind.ObjectMapper; import kr.magicbox.ssegateway.adapter.in.kafka.event.UserLoggedOutKafkaEvent; -import kr.magicbox.ssegateway.adapter.out.redis.RedisPubSubAdapter; +import kr.magicbox.ssegateway.adapter.out.redis.RedisStreamAdapter; import kr.magicbox.ssegateway.domain.vo.UserId; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -15,7 +15,7 @@ @RequiredArgsConstructor public class LogoutKafkaListener { - private final RedisPubSubAdapter redisPubSubAdapter; + private final RedisStreamAdapter redisStreamAdapter; private final ObjectMapper objectMapper; @KafkaListener(topics = "outbox.event.user-logged-out", groupId = "sse-gateway-service") @@ -28,7 +28,7 @@ public void handleLogout(ConsumerRecord record) { return; } - log.debug("로그아웃 이벤트 수신, Redis logout 발행 userId={}", event.userId()); - redisPubSubAdapter.publishLogout(UserId.of(event.userId())).subscribe(); + log.debug("로그아웃 이벤트 수신, Redis Stream logout 적재 userId={}", event.userId()); + redisStreamAdapter.appendLogout(UserId.of(event.userId())).subscribe(); } } diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/PurchaseTokenKafkaListener.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/PurchaseTokenKafkaListener.java index 73025eb6..9be8aead 100644 --- a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/PurchaseTokenKafkaListener.java +++ b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/in/kafka/PurchaseTokenKafkaListener.java @@ -3,7 +3,7 @@ import tools.jackson.databind.ObjectMapper; import kr.magicbox.ssegateway.adapter.in.kafka.event.PurchaseTokenIssuedKafkaEvent; import kr.magicbox.ssegateway.adapter.in.web.dto.response.SseNotificationResponse; -import kr.magicbox.ssegateway.adapter.out.redis.RedisPubSubAdapter; +import kr.magicbox.ssegateway.adapter.out.redis.RedisStreamAdapter; import kr.magicbox.ssegateway.domain.vo.UserId; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -16,7 +16,7 @@ @RequiredArgsConstructor public class PurchaseTokenKafkaListener { - private final RedisPubSubAdapter redisPubSubAdapter; + private final RedisStreamAdapter redisStreamAdapter; private final ObjectMapper objectMapper; @KafkaListener(topics = "sse.purchase-token-issued", groupId = "sse-gateway-service") @@ -35,6 +35,6 @@ public void handlePurchaseTokenIssued(ConsumerRecord record) { .purchaseToken(event.purchaseToken()) .build(); - redisPubSubAdapter.publishNotification(UserId.of(event.userId()), payload).subscribe(); + redisStreamAdapter.appendNotification(UserId.of(event.userId()), payload).subscribe(); } } diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisConfiguration.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisConfiguration.java index e78f01d1..47de8ebd 100644 --- a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisConfiguration.java +++ b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisConfiguration.java @@ -3,23 +3,22 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; -import org.springframework.data.redis.core.ReactiveStringRedisTemplate; -import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.core.ReactiveRedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializationContext; +import org.springframework.data.redis.serializer.StringRedisSerializer; @Configuration public class RedisConfiguration { @Bean - public ReactiveRedisMessageListenerContainer redisMessageListenerContainer( + public ReactiveRedisTemplate reactiveRedisTemplate( ReactiveRedisConnectionFactory factory ) { - return new ReactiveRedisMessageListenerContainer(factory); - } - - @Bean - public ReactiveStringRedisTemplate reactiveStringRedisTemplate( - ReactiveRedisConnectionFactory factory - ) { - return new ReactiveStringRedisTemplate(factory); + RedisSerializationContext context = RedisSerializationContext + .newSerializationContext(new StringRedisSerializer()) + .value(new StringRedisSerializer()) + .build(); + return new ReactiveRedisTemplate<>(factory, context); } } \ No newline at end of file diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisPubSubAdapter.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisPubSubAdapter.java deleted file mode 100644 index f63153f3..00000000 --- a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisPubSubAdapter.java +++ /dev/null @@ -1,126 +0,0 @@ -package kr.magicbox.ssegateway.adapter.out.redis; - -import tools.jackson.databind.ObjectMapper; -import kr.magicbox.ssegateway.adapter.in.web.dto.response.SseNotificationResponse; -import kr.magicbox.ssegateway.adapter.out.cache.SseSinkRegistry; -import kr.magicbox.ssegateway.domain.vo.UserId; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.connection.ReactiveSubscription; -import org.springframework.data.redis.core.ReactiveStringRedisTemplate; -import org.springframework.data.redis.listener.ChannelTopic; -import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer; -import org.springframework.stereotype.Component; -import reactor.core.Disposable; -import reactor.core.publisher.Mono; - -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * Redis Pub/Sub 기반 SSE 알림 어댑터. - * - * 채널 명명 규칙: - * - 구매 토큰 알림: sse:notification:{userId} - * - 로그아웃: sse:logout:{userId} - * - * 클라이언트가 SSE에 연결되면 해당 userId의 채널을 subscribe하고, - * 연결이 끊기면 구독을 해제한다. - * 모든 인스턴스가 동일한 채널을 구독하지만, 로컬 Sink가 있는 인스턴스만 emit한다. - */ -@Slf4j -@Component -@RequiredArgsConstructor -public class RedisPubSubAdapter { - - private static final String NOTIFICATION_CHANNEL_PREFIX = "sse:notification:"; - private static final String LOGOUT_CHANNEL_PREFIX = "sse:logout:"; - - private final ReactiveRedisMessageListenerContainer listenerContainer; - private final ReactiveStringRedisTemplate redisTemplate; - private final SseSinkRegistry sinkRegistry; - private final ObjectMapper objectMapper; - - private final Map subscriptions = new ConcurrentHashMap<>(); - - public void subscribe(UserId userId) { - String notificationChannel = NOTIFICATION_CHANNEL_PREFIX + userId.value(); - String logoutChannel = LOGOUT_CHANNEL_PREFIX + userId.value(); - - Disposable subscription = listenerContainer - .receive(ChannelTopic.of(notificationChannel), ChannelTopic.of(logoutChannel)) - .flatMap(message -> handleMessage(userId, message)) - .subscribe( - unused -> {}, - error -> log.error("Redis 메시지 처리 오류 userId={}", userId.value(), error) - ); - - subscriptions.put(userId.value(), subscription); - log.debug("Redis 채널 구독 등록 userId={}", userId.value()); - } - - public void unsubscribe(UserId userId) { - Disposable subscription = subscriptions.remove(userId.value()); - if (subscription != null && !subscription.isDisposed()) { - subscription.dispose(); - log.debug("Redis 채널 구독 해제 userId={}", userId.value()); - } - } - - public Mono publishNotification(UserId userId, SseNotificationResponse payload) { - String channel = NOTIFICATION_CHANNEL_PREFIX + userId.value(); - String message = serialize(payload); - return redisTemplate.convertAndSend(channel, message) - .doOnNext(count -> log.debug("Redis 알림 발행 userId={} channel={} receivers={}", userId.value(), channel, count)); - } - - public Mono publishLogout(UserId userId) { - String channel = LOGOUT_CHANNEL_PREFIX + userId.value(); - return redisTemplate.convertAndSend(channel, "logout") - .doOnNext(count -> log.debug("Redis 로그아웃 발행 userId={} channel={} receivers={}", userId.value(), channel, count)); - } - - private Mono handleMessage(UserId userId, ReactiveSubscription.Message message) { - String channel = message.getChannel(); - - if (channel.startsWith(LOGOUT_CHANNEL_PREFIX)) { - if (sinkRegistry.contains(userId)) { - log.debug("로그아웃 메시지 수신, Sink 종료 userId={}", userId.value()); - sinkRegistry.remove(userId); - } - return Mono.empty(); - } - - if (channel.startsWith(NOTIFICATION_CHANNEL_PREFIX)) { - if (!sinkRegistry.contains(userId)) { - return Mono.empty(); - } - SseNotificationResponse payload = deserialize(message.getMessage()); - if (payload != null) { - log.debug("알림 메시지 수신, Sink emit userId={}", userId.value()); - sinkRegistry.emit(userId, payload); - } - return Mono.empty(); - } - - return Mono.empty(); - } - - private String serialize(SseNotificationResponse payload) { - try { - return objectMapper.writeValueAsString(payload); - } catch (Exception e) { - log.error("SseNotificationResponse 직렬화 실패", e); - return "{}"; - } - } - - private SseNotificationResponse deserialize(String message) { - try { - return objectMapper.readValue(message, SseNotificationResponse.class); - } catch (Exception e) { - log.error("SseNotificationResponse 역직렬화 실패: {}", message, e); - return null; - } - } -} diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisStreamAdapter.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisStreamAdapter.java new file mode 100644 index 00000000..2c376857 --- /dev/null +++ b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/adapter/out/redis/RedisStreamAdapter.java @@ -0,0 +1,149 @@ +package kr.magicbox.ssegateway.adapter.out.redis; + +import tools.jackson.databind.ObjectMapper; +import kr.magicbox.ssegateway.adapter.in.web.dto.response.SseNotificationResponse; +import kr.magicbox.ssegateway.adapter.out.cache.SseSinkRegistry; +import kr.magicbox.ssegateway.domain.vo.UserId; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.ReactiveRedisTemplate; +import org.springframework.stereotype.Component; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Redis Stream 기반 SSE 알림 어댑터. + * + * 스트림 키 명명 규칙: + * - 구매 토큰 알림: sse:stream:notification:{userId} + * - 로그아웃: sse:stream:logout:{userId} + * + * 클라이언트가 SSE에 연결되면 해당 userId의 스트림을 XREAD BLOCK으로 구독하고, + * 연결이 끊기면 구독을 해제한다. + * 메시지는 영속화되어 연결 전 발행된 메시지도 lastSeenId 이후 재조회 가능. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class RedisStreamAdapter { + + private static final String NOTIFICATION_STREAM_PREFIX = "sse:stream:notification:"; + private static final String LOGOUT_STREAM_PREFIX = "sse:stream:logout:"; + private static final String PAYLOAD_FIELD = "payload"; + + private final ReactiveRedisTemplate redisTemplate; + private final SseSinkRegistry sinkRegistry; + private final ObjectMapper objectMapper; + + private final Map subscriptions = new ConcurrentHashMap<>(); + + public void subscribe(UserId userId) { + String notificationStream = NOTIFICATION_STREAM_PREFIX + userId.value(); + String logoutStream = LOGOUT_STREAM_PREFIX + userId.value(); + + Disposable subscription = pollStream(userId, notificationStream, logoutStream); + subscriptions.put(userId.value(), subscription); + log.debug("Redis Stream 구독 등록 userId={}", userId.value()); + } + + public void unsubscribe(UserId userId) { + Disposable subscription = subscriptions.remove(userId.value()); + if (subscription != null && !subscription.isDisposed()) { + subscription.dispose(); + log.debug("Redis Stream 구독 해제 userId={}", userId.value()); + } + } + + public Mono appendNotification(UserId userId, SseNotificationResponse payload) { + String stream = NOTIFICATION_STREAM_PREFIX + userId.value(); + String json = serialize(payload); + return redisTemplate.opsForStream() + .add(MapRecord.create(stream, Map.of(PAYLOAD_FIELD, json))) + .map(recordId -> { + log.debug("Redis Stream 알림 적재 userId={} id={}", userId.value(), recordId); + return recordId.getValue(); + }); + } + + public Mono appendLogout(UserId userId) { + String stream = LOGOUT_STREAM_PREFIX + userId.value(); + return redisTemplate.opsForStream() + .add(MapRecord.create(stream, Map.of(PAYLOAD_FIELD, "logout"))) + .map(recordId -> { + log.debug("Redis Stream 로그아웃 적재 userId={} id={}", userId.value(), recordId); + return recordId.getValue(); + }); + } + + private Disposable pollStream(UserId userId, String notificationStream, String logoutStream) { + return readLatest(notificationStream) + .mergeWith(readLatest(logoutStream)) + .repeat() + .flatMap(record -> handleRecord(userId, record)) + .subscribe( + unused -> {}, + error -> log.error("Redis Stream 처리 오류 userId={}", userId.value(), error) + ); + } + + private Flux> readLatest(String stream) { + return redisTemplate.opsForStream() + .read(StreamOffset.create(stream, ReadOffset.lastConsumed())) + .onErrorResume(e -> { + log.warn("Redis Stream 읽기 오류 stream={}: {}", stream, e.getMessage()); + return Flux.empty(); + }); + } + + private Mono handleRecord(UserId userId, MapRecord record) { + String stream = record.getStream(); + String payload = String.valueOf(record.getValue().get(PAYLOAD_FIELD)); + + if (stream.startsWith(LOGOUT_STREAM_PREFIX)) { + if (sinkRegistry.contains(userId)) { + log.debug("로그아웃 스트림 수신, Sink 종료 userId={}", userId.value()); + sinkRegistry.remove(userId); + } + return Mono.empty(); + } + + if (stream.startsWith(NOTIFICATION_STREAM_PREFIX)) { + if (!sinkRegistry.contains(userId)) { + return Mono.empty(); + } + SseNotificationResponse response = deserialize(payload); + if (response != null) { + log.debug("알림 스트림 수신, Sink emit userId={}", userId.value()); + sinkRegistry.emit(userId, response); + } + return Mono.empty(); + } + + return Mono.empty(); + } + + private String serialize(SseNotificationResponse payload) { + try { + return objectMapper.writeValueAsString(payload); + } catch (Exception e) { + log.error("SseNotificationResponse 직렬화 실패", e); + return "{}"; + } + } + + private SseNotificationResponse deserialize(String message) { + try { + return objectMapper.readValue(message, SseNotificationResponse.class); + } catch (Exception e) { + log.error("SseNotificationResponse 역직렬화 실패: {}", message, e); + return null; + } + } +} diff --git a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/application/service/SseSubscribeService.java b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/application/service/SseSubscribeService.java index 5e91b58b..3998dfaa 100644 --- a/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/application/service/SseSubscribeService.java +++ b/services/sse-gateway/src/main/java/kr/magicbox/ssegateway/application/service/SseSubscribeService.java @@ -3,7 +3,7 @@ import kr.magicbox.ssegateway.adapter.in.web.dto.response.SseNotificationResponse; import kr.magicbox.ssegateway.adapter.out.cache.SseSinkRegistry; import kr.magicbox.ssegateway.adapter.out.kafka.SseEventKafkaAdapter; -import kr.magicbox.ssegateway.adapter.out.redis.RedisPubSubAdapter; +import kr.magicbox.ssegateway.adapter.out.redis.RedisStreamAdapter; import kr.magicbox.ssegateway.application.port.in.SubscribeSseUseCase; import kr.magicbox.ssegateway.domain.vo.UserId; import lombok.RequiredArgsConstructor; @@ -24,12 +24,12 @@ public class SseSubscribeService implements SubscribeSseUseCase { private final SseSinkRegistry sinkRegistry; private final SseEventKafkaAdapter sseEventKafkaAdapter; - private final RedisPubSubAdapter redisPubSubAdapter; + private final RedisStreamAdapter redisStreamAdapter; @Override public Flux> subscribe(UserId userId) { Sinks.One sink = sinkRegistry.register(userId); - redisPubSubAdapter.subscribe(userId); + redisStreamAdapter.subscribe(userId); sseEventKafkaAdapter.publishConnected(userId); Flux> notificationStream = sink.asMono() @@ -39,7 +39,7 @@ public Flux> subscribe(UserId userId) { .build()) .flux() .doFinally(signal -> { - redisPubSubAdapter.unsubscribe(userId); + redisStreamAdapter.unsubscribe(userId); sinkRegistry.remove(userId); sseEventKafkaAdapter.publishDisconnected(userId); });