diff --git a/infrastructure/outbox/pom.xml b/infrastructure/outbox/pom.xml index da2aaff..b7c309e 100644 --- a/infrastructure/outbox/pom.xml +++ b/infrastructure/outbox/pom.xml @@ -12,5 +12,11 @@ outbox + + + org.springframework.boot + spring-boot-starter + + \ No newline at end of file diff --git a/infrastructure/outbox/src/main/java/com/food/order/system/outbox/OutboxScheduler.java b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/OutboxScheduler.java new file mode 100644 index 0000000..f37e0f5 --- /dev/null +++ b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/OutboxScheduler.java @@ -0,0 +1,5 @@ +package com.food.order.system.outbox; + +public interface OutboxScheduler { + void processOutboxMessage(); +} diff --git a/infrastructure/outbox/src/main/java/com/food/order/system/outbox/config/SchedulerConfig.java b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/config/SchedulerConfig.java new file mode 100644 index 0000000..43adacd --- /dev/null +++ b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/config/SchedulerConfig.java @@ -0,0 +1,12 @@ +package com.food.order.system.outbox.config; + +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableScheduling; + +@Configuration +@EnableScheduling +public class SchedulerConfig { + + + +} diff --git a/infrastructure/outbox/src/main/java/com/food/order/system/outbox/order/SagaConst.java b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/order/SagaConst.java new file mode 100644 index 0000000..3ec7b83 --- /dev/null +++ b/infrastructure/outbox/src/main/java/com/food/order/system/outbox/order/SagaConst.java @@ -0,0 +1,12 @@ +package com.food.order.system.outbox.order; + +public class SagaConst { + + private SagaConst() { + } + + public static final String ORDER_PROCESSING_SAGA = "OrderProcessingSaga"; + + + +} diff --git a/order-service/order-domain/order-application-service/pom.xml b/order-service/order-domain/order-application-service/pom.xml index 69a718c..8a4fc93 100644 --- a/order-service/order-domain/order-application-service/pom.xml +++ b/order-service/order-domain/order-application-service/pom.xml @@ -31,6 +31,16 @@ spring-boot-starter-validation + + com.food.order + outbox + + + + org.springframework.boot + spring-boot-starter-json + + org.springframework spring-tx diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/OrderCreateCommandHandler.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/OrderCreateCommandHandler.java index 0e0a51c..5bca96a 100644 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/OrderCreateCommandHandler.java +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/OrderCreateCommandHandler.java @@ -1,13 +1,18 @@ package com.food.order.sysyem; +import com.food.order.system.outbox.OutboxStatus; import com.food.order.sysyem.dto.create.CreateOrderCommand; import com.food.order.sysyem.dto.create.CreateOrderResponse; import com.food.order.sysyem.helper.OrderCreateHelper; +import com.food.order.sysyem.helper.OrderSagaHelper; import com.food.order.sysyem.mapper.OrderDataMapper; -import com.food.order.sysyem.ports.output.message.publisher.payment.OrderCreatedPaymentRequestMessagePublisher; +import com.food.order.sysyem.outbox.scheduler.payment.PaymentOutboxHelper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.util.UUID; @Component @Slf4j @@ -17,14 +22,28 @@ public class OrderCreateCommandHandler { private final OrderCreateHelper orderCreateHelper; private final OrderDataMapper orderDataMapper; - private final OrderCreatedPaymentRequestMessagePublisher orderCreatedPaymentRequestMessagePublisher; + private final OrderSagaHelper orderSagaHelper; + private final PaymentOutboxHelper paymentOutboxHelper; + + @Transactional public CreateOrderResponse createOrder(CreateOrderCommand createOrderCommand) { var persistOrder = orderCreateHelper.persistOrder(createOrderCommand); log.info("createOrder with id: {}", persistOrder.getOrder().getId().getValue()); - orderCreatedPaymentRequestMessagePublisher.publish(persistOrder); - return orderDataMapper.orderToCreateOrderResponse(persistOrder.getOrder(),"Order created successfully"); + var response = orderDataMapper.orderToCreateOrderResponse(persistOrder.getOrder(),"Order created successfully"); + + paymentOutboxHelper.savePaymentOutboxMessage( + orderDataMapper.orderCreatedEventToOrderPaymentEventPayload(persistOrder), + persistOrder.getOrder().getStatus(), + orderSagaHelper.orderStatusToSagaStatus(persistOrder.getOrder().getStatus()), + OutboxStatus.STARTED, + UUID.randomUUID() + ); + + log.info("Returning CreateOrderResponse with order id : {}", persistOrder.getOrder().getId()); + + return response; } diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderCreateHelper.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderCreateHelper.java index fd1163e..7afb14a 100644 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderCreateHelper.java +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderCreateHelper.java @@ -28,7 +28,6 @@ public class OrderCreateHelper { private final CustomerRepository customerRepository; private final RestaurantRepository restaurantRepository; private final OrderDataMapper orderDataMapper; - private final DomainEventPublisher publisher; @Transactional public OrderCreatedEvent persistOrder(CreateOrderCommand createOrderCommand) { @@ -36,7 +35,8 @@ public class OrderCreateHelper { checkCustomer(createOrderCommand.customerId()); Restaurant restaurant = checkRestaurant(createOrderCommand); var order = orderDataMapper.createOrderCommandToOrder(createOrderCommand); - var createdEventOrder = orderDomainService.validateAndInitiateOrder(order, restaurant,publisher); + var createdEventOrder = orderDomainService.validateAndInitiateOrder + (order, restaurant); saveOrder(order); log.info("Created Order Event : {}", createdEventOrder); return createdEventOrder; diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderSagaHelper.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderSagaHelper.java index 55eb54c..f0bc541 100644 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderSagaHelper.java +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/helper/OrderSagaHelper.java @@ -2,7 +2,9 @@ package com.food.order.sysyem.helper; import com.food.order.system.domain.entity.Order; import com.food.order.system.domain.exception.OrderNotFoundException; +import com.food.order.system.saga.SagaStatus; import com.food.order.sysyem.ports.output.repository.OrderRepository; +import com.food.order.sysyem.valueobject.OrderStatus; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -23,5 +25,15 @@ public class OrderSagaHelper { orderRepository.save(order); } + public SagaStatus orderStatusToSagaStatus(OrderStatus orderStatus) { + return switch (orderStatus) { + case PAID -> SagaStatus.PROCESSING; + case APPROVED -> SagaStatus.SUCCEEDED; + case CANCELLING -> SagaStatus.COMPENSATING; + case CANCELLED -> SagaStatus.COMPENSATED; + default -> SagaStatus.STARTED; + }; + } + } diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/mapper/OrderDataMapper.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/mapper/OrderDataMapper.java index 10e023c..be99271 100644 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/mapper/OrderDataMapper.java +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/mapper/OrderDataMapper.java @@ -4,15 +4,14 @@ import com.food.order.system.domain.entity.Order; import com.food.order.system.domain.entity.OrderItem; import com.food.order.system.domain.entity.Product; import com.food.order.system.domain.entity.Restaurant; +import com.food.order.system.domain.event.OrderCreatedEvent; import com.food.order.system.domain.valueobject.StreetAddress; import com.food.order.sysyem.dto.create.CreateOrderCommand; import com.food.order.sysyem.dto.create.CreateOrderResponse; import com.food.order.sysyem.dto.create.OrderAddress; import com.food.order.sysyem.dto.track.TrackOrderResponse; -import com.food.order.sysyem.valueobject.CustomerId; -import com.food.order.sysyem.valueobject.Money; -import com.food.order.sysyem.valueobject.ProductId; -import com.food.order.sysyem.valueobject.RestaurantId; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentEventPayload; +import com.food.order.sysyem.valueobject.*; import org.springframework.stereotype.Component; import java.util.List; @@ -21,6 +20,16 @@ import java.util.UUID; @Component public class OrderDataMapper { + public OrderPaymentEventPayload orderCreatedEventToOrderPaymentEventPayload(OrderCreatedEvent order) { + return OrderPaymentEventPayload.builder() + .orderId(order.getOrder().getId().getValue().toString()) + .customerId(order.getOrder().getCustomerId().getValue().toString()) + .price(order.getOrder().getPrice().getAmount()) + .createdAt(order.getCreatedAt()) + .paymentOrderStatus(PaymentOrderStatus.PENDING.name()) + .build(); + } + public TrackOrderResponse orderToTrackOrderResponse(Order order) { return TrackOrderResponse.builder() diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalEventPayload.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalEventPayload.java new file mode 100644 index 0000000..832ca19 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalEventPayload.java @@ -0,0 +1,36 @@ +package com.food.order.sysyem.outbox.model.approval; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; + +import java.math.BigDecimal; +import java.time.ZonedDateTime; +import java.util.List; + +@Getter +@Builder +@AllArgsConstructor +public class OrderApprovalEventPayload { + + @JsonProperty + private String orderId; + + @JsonProperty + private String restaurantId; + + @JsonProperty + private BigDecimal price; + + @JsonProperty + private ZonedDateTime createdAt; + + @JsonProperty + private String restaurantOrderStatus; + + @JsonProperty + private List products; + + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalOutboxMessage.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalOutboxMessage.java new file mode 100644 index 0000000..7d58527 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalOutboxMessage.java @@ -0,0 +1,42 @@ +package com.food.order.sysyem.outbox.model.approval; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.valueobject.OrderStatus; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; +import lombok.Setter; + +import java.time.ZonedDateTime; +import java.util.UUID; + +@Getter +@Builder +@AllArgsConstructor +public class OrderApprovalOutboxMessage { + + private UUID id; + + private UUID sagaId; + + private ZonedDateTime createdAt; + + @Setter + private ZonedDateTime processedAt; + + private String type; + + private String payload; + + @Setter + private SagaStatus sagaStatus; + + @Setter + private OrderStatus orderStatus; + + @Setter + private OutboxStatus outboxStatus; + + private int version; +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalProduct.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalProduct.java new file mode 100644 index 0000000..9c1524e --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/approval/OrderApprovalProduct.java @@ -0,0 +1,19 @@ +package com.food.order.sysyem.outbox.model.approval; + + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; + +@Getter +@Builder +@AllArgsConstructor +public class OrderApprovalProduct { + + @JsonProperty + private String id; + @JsonProperty + private Integer quantity; + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentEventPayload.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentEventPayload.java new file mode 100644 index 0000000..d175728 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentEventPayload.java @@ -0,0 +1,27 @@ +package com.food.order.sysyem.outbox.model.payment; + +import com.fasterxml.jackson.annotation.JsonProperty; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; + +import java.math.BigDecimal; +import java.time.ZonedDateTime; + +@Getter +@Builder +@AllArgsConstructor +public class OrderPaymentEventPayload { + + @JsonProperty + private String orderId; + @JsonProperty + private String customerId; + @JsonProperty + private BigDecimal price; + @JsonProperty + private ZonedDateTime createdAt; + @JsonProperty + private String paymentOrderStatus; + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentOutboxMessage.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentOutboxMessage.java new file mode 100644 index 0000000..4e3facd --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/model/payment/OrderPaymentOutboxMessage.java @@ -0,0 +1,39 @@ +package com.food.order.sysyem.outbox.model.payment; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.valueobject.OrderStatus; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; +import lombok.Setter; + +import java.time.ZonedDateTime; +import java.util.UUID; + +@Getter +@Builder +@AllArgsConstructor +public class OrderPaymentOutboxMessage { + private UUID id; + + private UUID sagaId; + + private ZonedDateTime createdAt; + + @Setter + private ZonedDateTime processedAt; + + private String type; + + private String payload; + @Setter + private SagaStatus sagaStatus; + @Setter + private OrderStatus orderStatus; + @Setter + private OutboxStatus outboxStatus; + private int version; + + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/ApprovalOutboxHelper.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/ApprovalOutboxHelper.java new file mode 100644 index 0000000..fbe42e8 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/ApprovalOutboxHelper.java @@ -0,0 +1,66 @@ +package com.food.order.sysyem.outbox.scheduler.approval; + +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.approval.OrderApprovalOutboxMessage; +import com.food.order.sysyem.ports.output.repository.ApprovalOutboxRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +import static com.food.order.system.outbox.order.SagaConst.ORDER_PROCESSING_SAGA; + +@Slf4j +@RequiredArgsConstructor +@Component +public class ApprovalOutboxHelper { + + private final ApprovalOutboxRepository approvalOutboxRepository; + + @Transactional(readOnly = true) + public Optional> getApprovalOutboxMessageByOutboxStatusAndSagaStatus + (OutboxStatus outboxStatus, SagaStatus... sagaStatus) { + return approvalOutboxRepository.findByTypeAndOutboxStatusAndSagaStatus( + ORDER_PROCESSING_SAGA + ,outboxStatus, + sagaStatus); + } + + + @Transactional(readOnly = true) + public Optional getApprovalOutboxMessageBySagaIdAndSagaStatus( + UUID sagaId, SagaStatus... sagaStatus) { + return approvalOutboxRepository.findByTypeAndSagaIdAndSagaStatus( + ORDER_PROCESSING_SAGA + ,sagaId, + sagaStatus); + } + + @Transactional + public void save(OrderApprovalOutboxMessage approvalOutboxMessage) { + var response = approvalOutboxRepository.save(approvalOutboxMessage); + if (Objects.isNull(response)) { + throw new OrderDomainException("Failed to save outbox message id : " + + approvalOutboxMessage.getId()); + } + log.info("Outbox message id : {} saved successfully", response.getId()); + } + + @Transactional + public void deleteApprovalOutboxMessageByOutboxStatusAndSagaStatus(OutboxStatus outboxStatus, + SagaStatus... sagaStatus) { + + approvalOutboxRepository.deleteByTypeAndOutboxStatusAndSagaStatus( + ORDER_PROCESSING_SAGA, + outboxStatus, + sagaStatus); + + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxCleaner.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxCleaner.java new file mode 100644 index 0000000..c6c6273 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxCleaner.java @@ -0,0 +1,51 @@ +package com.food.order.sysyem.outbox.scheduler.approval; + +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxScheduler; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.approval.OrderApprovalOutboxMessage; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.util.Objects; +import java.util.stream.Collectors; + +@Slf4j +@Component +@RequiredArgsConstructor +public class RestaurantApprovalOutboxCleaner implements OutboxScheduler { + + private final ApprovalOutboxHelper approvalOutboxHelper; + + @Override + @Scheduled(cron = "@midnight") + public void processOutboxMessage() { + + var outboxMessageResponse = + approvalOutboxHelper.getApprovalOutboxMessageByOutboxStatusAndSagaStatus( + OutboxStatus.COMPLETED, + SagaStatus.SUCCEEDED, + SagaStatus.FAILED, + SagaStatus.COMPENSATING). + orElseThrow(() -> new OrderDomainException("No outbox message found for processing")); + + if (Objects.nonNull(outboxMessageResponse)) { + + log.info("Received {} OrderPaymentOutboxMessage for clean-up. The Payloads :{}", + outboxMessageResponse.size(), + outboxMessageResponse.stream().map(OrderApprovalOutboxMessage::getPayload) + .collect(Collectors.joining(","))); + approvalOutboxHelper.deleteApprovalOutboxMessageByOutboxStatusAndSagaStatus( + OutboxStatus.COMPLETED, + SagaStatus.SUCCEEDED, + SagaStatus.FAILED, + SagaStatus.COMPENSATING); + log.info("Clean-up completed ! DELETED LOG SIZE : {}", outboxMessageResponse.size()); + + } + + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxMessage.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxMessage.java new file mode 100644 index 0000000..8d9d9f9 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/approval/RestaurantApprovalOutboxMessage.java @@ -0,0 +1,64 @@ +package com.food.order.sysyem.outbox.scheduler.approval; + +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxScheduler; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.approval.OrderApprovalOutboxMessage; +import com.food.order.sysyem.ports.output.message.publisher.restaurantapproval.RestaurantApprovalRequestMessagePublisher; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Objects; +import java.util.stream.Collectors; + +@Slf4j +@Component +@RequiredArgsConstructor +public class RestaurantApprovalOutboxMessage implements OutboxScheduler { + + private final ApprovalOutboxHelper approvalOutboxHelper; + private final RestaurantApprovalRequestMessagePublisher restaurantApprovalRequestMessagePublisher; + + @Override + @Transactional + @Scheduled(fixedDelayString = "${order-service.outbox-scheduler-fixed-rate}", + initialDelayString = "${order-service.outbox-scheduler-initial-delay}") + public void processOutboxMessage() { + + log.info("Processing outbox message STARTED !"); + + var outboxMessageResponse = + approvalOutboxHelper.getApprovalOutboxMessageByOutboxStatusAndSagaStatus( + OutboxStatus.STARTED, + SagaStatus.STARTED, + SagaStatus.COMPENSATING) + .orElseThrow( + () -> new OrderDomainException("No outbox message found for processing")); + + if (Objects.nonNull(outboxMessageResponse) && outboxMessageResponse.size() > 0) { + + log.info("Received {} OrderPaymentOutboxMessage with ids : {} , sending message bus !" , + outboxMessageResponse.size(), + outboxMessageResponse.stream().map(orderPaymentOutboxMessage -> orderPaymentOutboxMessage.getId().toString()) + .collect(Collectors.joining(","))); + outboxMessageResponse.forEach(orderPaymentOutboxMessage -> { + restaurantApprovalRequestMessagePublisher.publish + (orderPaymentOutboxMessage,this::updateOutboxStatus); + }); + log.info("Processing outbox message completed ! "); + } + + } + + + private void updateOutboxStatus(OrderApprovalOutboxMessage orderPaymentOutboxMessage, + OutboxStatus outboxStatus) { + orderPaymentOutboxMessage.setOutboxStatus(outboxStatus); + approvalOutboxHelper.save(orderPaymentOutboxMessage); + log.info("Outbox message id : {} updated successfully", orderPaymentOutboxMessage.getId()); + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxCleaner.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxCleaner.java new file mode 100644 index 0000000..2d7ce1e --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxCleaner.java @@ -0,0 +1,51 @@ +package com.food.order.sysyem.outbox.scheduler.payment; + +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxScheduler; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentOutboxMessage; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.util.Objects; +import java.util.stream.Collectors; + +@Slf4j +@Component +@RequiredArgsConstructor +public class PaymentOutboxCleaner implements OutboxScheduler { + + private final PaymentOutboxHelper paymentOutboxHelper; + + @Override + @Scheduled(cron = "@midnight") + public void processOutboxMessage() { + + var outboxMessageResponse = + paymentOutboxHelper.getPaymentOutboxMessageByOutboxMessageStatusAndSagaStatus( + OutboxStatus.COMPLETED, + SagaStatus.SUCCEEDED, + SagaStatus.FAILED, + SagaStatus.COMPENSATING). + orElseThrow(() -> new OrderDomainException("No outbox message found for processing")); + + if (Objects.nonNull(outboxMessageResponse)) { + + log.info("Received {} OrderPaymentOutboxMessage for clean-up. The Payloads :{}", + outboxMessageResponse.size(), + outboxMessageResponse.stream().map(OrderPaymentOutboxMessage::getPayload) + .collect(Collectors.joining(","))); + paymentOutboxHelper.deletePaymentOutboxMessageByOutboxStatusAndSagaStatus( + OutboxStatus.COMPLETED, + SagaStatus.SUCCEEDED, + SagaStatus.FAILED, + SagaStatus.COMPENSATING); + log.info("Clean-up completed ! DELETED LOG SIZE : {}", outboxMessageResponse.size()); + + } + + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxHelper.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxHelper.java new file mode 100644 index 0000000..1ef0ca2 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxHelper.java @@ -0,0 +1,100 @@ +package com.food.order.sysyem.outbox.scheduler.payment; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentEventPayload; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentOutboxMessage; +import com.food.order.sysyem.ports.output.repository.PaymentOutboxRepository; +import com.food.order.sysyem.valueobject.OrderStatus; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +import static com.food.order.system.outbox.order.SagaConst.ORDER_PROCESSING_SAGA; + +@Slf4j +@Component +@RequiredArgsConstructor +public class PaymentOutboxHelper { + + private final PaymentOutboxRepository paymentOutboxRepository; + private final ObjectMapper objectMapper; + + @Transactional(readOnly = true) + public Optional> getPaymentOutboxMessageByOutboxMessageStatusAndSagaStatus( + OutboxStatus outboxStatus, SagaStatus... sagaStatus) { + return paymentOutboxRepository.findByTypeAndOutboxStatusAndSagaStatus( + ORDER_PROCESSING_SAGA, + outboxStatus, + sagaStatus); + + } + + @Transactional(readOnly = true) + public Optional getPaymentOutboxMessageBySagaIdAndSagaStatus( + UUID sagaId, SagaStatus... sagaStatus) { + return paymentOutboxRepository.findByTypeAndSagaIdAndSagaStatus( + ORDER_PROCESSING_SAGA + ,sagaId, + sagaStatus); + } + + @Transactional + public void save(OrderPaymentOutboxMessage orderPaymentOutboxMessage) { + var response = paymentOutboxRepository.save(orderPaymentOutboxMessage); + if (Objects.isNull(response)) { + throw new OrderDomainException("Failed to save outbox message id : " + + orderPaymentOutboxMessage.getId()); + } + log.info("Outbox message id : {} saved successfully", response.getId()); + } + + @Transactional + public void savePaymentOutboxMessage(OrderPaymentEventPayload payload, + OrderStatus orderStatus, + SagaStatus sagaStatus, + OutboxStatus outboxStatus, + UUID sagaId) { + + save(OrderPaymentOutboxMessage.builder() + .id(UUID.randomUUID()) + .sagaId(sagaId) + .createdAt(payload.getCreatedAt()) + .type(ORDER_PROCESSING_SAGA) + .payload(createPayload(payload)) + .outboxStatus(outboxStatus) + .orderStatus(orderStatus) + .sagaStatus(sagaStatus) + .build()); + + } + + private String createPayload(OrderPaymentEventPayload payload) { + try { + return objectMapper.writeValueAsString(payload); + } catch (JsonProcessingException e) { + log.error("Failed to create payload for outbox message", e); + throw new OrderDomainException("Failed to create payload for outbox message"); + } + } + + @Transactional + public void deletePaymentOutboxMessageByOutboxStatusAndSagaStatus(OutboxStatus outboxStatus, + SagaStatus... sagaStatus) { + + paymentOutboxRepository.deleteByTypeAndOutboxStatusAndSagaStatus( + ORDER_PROCESSING_SAGA, + outboxStatus, + sagaStatus); + + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxScheduler.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxScheduler.java new file mode 100644 index 0000000..1afbfff --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/outbox/scheduler/payment/PaymentOutboxScheduler.java @@ -0,0 +1,63 @@ +package com.food.order.sysyem.outbox.scheduler.payment; + +import com.food.order.system.domain.exception.OrderDomainException; +import com.food.order.system.outbox.OutboxScheduler; +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentOutboxMessage; +import com.food.order.sysyem.ports.output.message.publisher.payment.PaymentRequestMessagePublisher; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Objects; +import java.util.stream.Collectors; + +@Slf4j +@Component +@RequiredArgsConstructor +public class PaymentOutboxScheduler implements OutboxScheduler { + + private final PaymentOutboxHelper paymentOutboxHelper; + private final PaymentRequestMessagePublisher paymentRequestMessagePublisher; + + @Override + @Transactional + @Scheduled(fixedDelayString = "${order-service.outbox-scheduler-fixed-rate}", + initialDelayString = "${order-service.outbox-scheduler-initial-delay}") + public void processOutboxMessage() { + + log.info("Processing outbox message STARTED !"); + + var outboxMessageResponse = + paymentOutboxHelper.getPaymentOutboxMessageByOutboxMessageStatusAndSagaStatus( + OutboxStatus.STARTED, + SagaStatus.STARTED, + SagaStatus.COMPENSATING) + .orElseThrow( + () -> new OrderDomainException("No outbox message found for processing")); + + if (Objects.nonNull(outboxMessageResponse) && outboxMessageResponse.size() > 0) { + + log.info("Received {} OrderPaymentOutboxMessage with ids : {} , sending message bus !" , + outboxMessageResponse.size(), + outboxMessageResponse.stream().map(orderPaymentOutboxMessage -> orderPaymentOutboxMessage.getId().toString()) + .collect(Collectors.joining(","))); + outboxMessageResponse.forEach(orderPaymentOutboxMessage -> { + paymentRequestMessagePublisher.publish(orderPaymentOutboxMessage,this::updateOutboxStatus); + }); + log.info("Processing outbox message completed ! "); + } + + } + + + private void updateOutboxStatus(OrderPaymentOutboxMessage orderPaymentOutboxMessage, + OutboxStatus outboxStatus) { + orderPaymentOutboxMessage.setOutboxStatus(outboxStatus); + paymentOutboxHelper.save(orderPaymentOutboxMessage); + log.info("Outbox message id : {} updated successfully", orderPaymentOutboxMessage.getId()); + } +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCancelledPaymentRequestMessagePublisher.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCancelledPaymentRequestMessagePublisher.java deleted file mode 100644 index 6a84a49..0000000 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCancelledPaymentRequestMessagePublisher.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.food.order.sysyem.ports.output.message.publisher.payment; - -import com.food.order.sysyem.event.publisher.DomainEventPublisher; -import com.food.order.system.domain.event.OrderCancelledEvent; - -public interface OrderCancelledPaymentRequestMessagePublisher extends DomainEventPublisher { - -} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCreatedPaymentRequestMessagePublisher.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCreatedPaymentRequestMessagePublisher.java deleted file mode 100644 index fb83ed0..0000000 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/OrderCreatedPaymentRequestMessagePublisher.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.food.order.sysyem.ports.output.message.publisher.payment; - -import com.food.order.sysyem.event.publisher.DomainEventPublisher; -import com.food.order.system.domain.event.OrderCreatedEvent; - -public interface OrderCreatedPaymentRequestMessagePublisher extends DomainEventPublisher { - -} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/PaymentRequestMessagePublisher.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/PaymentRequestMessagePublisher.java new file mode 100644 index 0000000..05750f7 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/payment/PaymentRequestMessagePublisher.java @@ -0,0 +1,13 @@ +package com.food.order.sysyem.ports.output.message.publisher.payment; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentOutboxMessage; + +import java.util.function.BiConsumer; + +public interface PaymentRequestMessagePublisher { + + void publish(OrderPaymentOutboxMessage message, + BiConsumer outboxCallback); + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/OrderPaidRestaurantRequestMessagePublisher.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/OrderPaidRestaurantRequestMessagePublisher.java deleted file mode 100644 index 1d03f3f..0000000 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/OrderPaidRestaurantRequestMessagePublisher.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.food.order.sysyem.ports.output.message.publisher.restaurantapproval; - -import com.food.order.sysyem.event.publisher.DomainEventPublisher; -import com.food.order.system.domain.event.OrderPaidEvent; - -public interface OrderPaidRestaurantRequestMessagePublisher extends DomainEventPublisher { - -} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/RestaurantApprovalRequestMessagePublisher.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/RestaurantApprovalRequestMessagePublisher.java new file mode 100644 index 0000000..8f14f8d --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/message/publisher/restaurantapproval/RestaurantApprovalRequestMessagePublisher.java @@ -0,0 +1,13 @@ +package com.food.order.sysyem.ports.output.message.publisher.restaurantapproval; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.sysyem.outbox.model.approval.OrderApprovalOutboxMessage; + +import java.util.function.BiConsumer; + +public interface RestaurantApprovalRequestMessagePublisher { + + void publish(OrderApprovalOutboxMessage message, + BiConsumer outboxCallback); + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/ApprovalOutboxRepository.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/ApprovalOutboxRepository.java new file mode 100644 index 0000000..4cc7355 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/ApprovalOutboxRepository.java @@ -0,0 +1,25 @@ +package com.food.order.sysyem.ports.output.repository; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.approval.OrderApprovalOutboxMessage; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +public interface ApprovalOutboxRepository { + OrderApprovalOutboxMessage save(OrderApprovalOutboxMessage message); + + Optional> findByTypeAndOutboxStatusAndSagaStatus(String type, + OutboxStatus outboxStatus, + SagaStatus... sagaStatus); + + Optional findByTypeAndSagaIdAndSagaStatus(String type, + UUID sagaId, + SagaStatus... sagaStatus); + + void deleteByTypeAndOutboxStatusAndSagaStatus(String type, + OutboxStatus outboxStatus, + SagaStatus... sagaStatus); +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/PaymentOutboxRepository.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/PaymentOutboxRepository.java new file mode 100644 index 0000000..a3016a7 --- /dev/null +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/ports/output/repository/PaymentOutboxRepository.java @@ -0,0 +1,28 @@ +package com.food.order.sysyem.ports.output.repository; + +import com.food.order.system.outbox.OutboxStatus; +import com.food.order.system.saga.SagaStatus; +import com.food.order.sysyem.outbox.model.payment.OrderPaymentOutboxMessage; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +public interface PaymentOutboxRepository { + + OrderPaymentOutboxMessage save(OrderPaymentOutboxMessage message); + + Optional> findByTypeAndOutboxStatusAndSagaStatus( String type, + OutboxStatus outboxStatus, + SagaStatus... sagaStatus); + + Optional findByTypeAndSagaIdAndSagaStatus(String type, + UUID sagaId, + SagaStatus... sagaStatus); + + void deleteByTypeAndOutboxStatusAndSagaStatus(String type, + OutboxStatus outboxStatus, + SagaStatus... sagaStatus); + + +} diff --git a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/saga/OrderApprovalSaga.java b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/saga/OrderApprovalSaga.java index d309c10..314aa46 100644 --- a/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/saga/OrderApprovalSaga.java +++ b/order-service/order-domain/order-application-service/src/main/java/com/food/order/sysyem/saga/OrderApprovalSaga.java @@ -6,7 +6,6 @@ import com.food.order.system.saga.SagaStep; import com.food.order.sysyem.dto.message.RestaurantApprovalResponse; import com.food.order.sysyem.event.EmptyEvent; import com.food.order.sysyem.helper.OrderSagaHelper; -import com.food.order.sysyem.ports.output.message.publisher.payment.OrderCancelledPaymentRequestMessagePublisher; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -18,9 +17,10 @@ import org.springframework.transaction.annotation.Transactional; public class OrderApprovalSaga implements SagaStep { private final OrderDomainService orderDomainService; - private final OrderCancelledPaymentRequestMessagePublisher messagePublisher; private final OrderSagaHelper orderSagaHelper; + + @Override @Transactional public EmptyEvent process(RestaurantApprovalResponse data) { @@ -38,7 +38,9 @@ public class OrderApprovalSaga implements SagaStep { private final OrderMessagingDataMapper orderMessagingDataMapper; private final OrderServiceConfigData configData; diff --git a/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/CreateOrderKafkaMessagePublisher.java b/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/CreateOrderKafkaMessagePublisher.java index 5ee687c..cce3831 100644 --- a/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/CreateOrderKafkaMessagePublisher.java +++ b/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/CreateOrderKafkaMessagePublisher.java @@ -7,7 +7,7 @@ import com.food.order.system.kafka.producer.KafkaMessageHelper; import com.food.order.system.kafka.producer.service.KafkaProducer; import com.food.order.system.order.messaging.mapper.OrderMessagingDataMapper; import com.food.order.sysyem.config.OrderServiceConfigData; -import com.food.order.sysyem.ports.output.message.publisher.payment.OrderCreatedPaymentRequestMessagePublisher; +import com.food.order.sysyem.event.publisher.DomainEventPublisher; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -15,7 +15,7 @@ import org.springframework.stereotype.Component; @Component @Slf4j @RequiredArgsConstructor -public class CreateOrderKafkaMessagePublisher implements OrderCreatedPaymentRequestMessagePublisher { +public class CreateOrderKafkaMessagePublisher implements DomainEventPublisher { private final OrderMessagingDataMapper orderMessagingDataMapper; private final OrderServiceConfigData configData; diff --git a/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/PayOrderKafkaMessagePublisher.java b/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/PayOrderKafkaMessagePublisher.java index bf5c71b..24551fb 100644 --- a/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/PayOrderKafkaMessagePublisher.java +++ b/order-service/order-messaging/src/main/java/com/food/order/system/order/messaging/publisher/kafka/PayOrderKafkaMessagePublisher.java @@ -6,7 +6,7 @@ import com.food.order.system.kafka.producer.KafkaMessageHelper; import com.food.order.system.kafka.producer.service.KafkaProducer; import com.food.order.system.order.messaging.mapper.OrderMessagingDataMapper; import com.food.order.sysyem.config.OrderServiceConfigData; -import com.food.order.sysyem.ports.output.message.publisher.restaurantapproval.OrderPaidRestaurantRequestMessagePublisher; +import com.food.order.sysyem.event.publisher.DomainEventPublisher; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -14,7 +14,7 @@ import org.springframework.stereotype.Component; @Component @Slf4j @RequiredArgsConstructor -public class PayOrderKafkaMessagePublisher implements OrderPaidRestaurantRequestMessagePublisher { +public class PayOrderKafkaMessagePublisher implements DomainEventPublisher { private final OrderMessagingDataMapper orderMessagingDataMapper; private final OrderServiceConfigData configData;