#1180-return-payment-id

kafka/connects/debezium-checkout-status.json

@@ -0,0 +1,16 @@
+{
+  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+  "topic.prefix": "dbcheckout-status",
+  "database.user": "admin",
+  "database.dbname": "order",
+  "database.hostname": "postgres",
+  "database.password": "admin",
+  "database.port": "5432",
+  "key.converter.schemas.enable": "false",
+  "value.converter.schemas.enable": "false",
+  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
+  "schema.include.list": "public",
+  "table.include.list": "public.checkout",
+  "slot.name": "checkout_status_slot"
+}

kafka/connects/debezium-payment.json

@@ -0,0 +1,16 @@
+{
+  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+  "topic.prefix": "dbpayment",
+  "database.user": "admin",
+  "database.dbname": "payment",
+  "database.hostname": "postgres",
+  "database.password": "admin",
+  "database.port": "5432",
+  "key.converter.schemas.enable": "false",
+  "value.converter.schemas.enable": "false",
+  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
+  "schema.include.list": "public",
+  "table.include.list": "public.payment",
+  "slot.name": "payment_slot"
+}

order/pom.xml

@@ -29,6 +29,14 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-validation</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-jpa</artifactId>
@@ -51,6 +59,11 @@
             <groupId>org.liquibase</groupId>
             <artifactId>liquibase-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.yas</groupId>
             <artifactId>common-library</artifactId>

order/src/it/resources/application.properties

@@ -12,4 +12,7 @@ spring.security.oauth2.resourceserver.jwt.issuer-uri=test
 springdoc.oauthflow.authorization-url=test
 springdoc.oauthflow.token-url=test
 spring.jpa.open-in-view=true
-cors.allowed-origins=*
+cors.allowed-origins=*
+
+cdc.event.checkout.status.topic-name=dbcheckout-status.public.checkout
+cdc.event.checkout.status.group-id=checkout-status

order/src/main/java/com/yas/order/config/JsonConfig.java

@@ -0,0 +1,22 @@
+package com.yas.order.config;
+
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.Gson;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+
+@Configuration
+public class JsonConfig {
+
+    @Bean
+    public ObjectMapper objectMapper() {
+        return new ObjectMapper();
+    }
+
+    @Bean
+    public Gson gson() {
+        return new Gson();
+    }
+}

order/src/main/java/com/yas/order/config/ServiceUrlConfig.java

@@ -4,5 +4,5 @@
 
 @ConfigurationProperties(prefix = "yas.services")
 public record ServiceUrlConfig(
-        String cart, String customer, String product, String tax, String promotion) {
+        String cart, String customer, String product, String tax, String promotion, String payment) {
 }

order/src/main/java/com/yas/order/consumer/OrderStatusConsumer.java

@@ -0,0 +1,145 @@
+package com.yas.order.consumer;
+
+import static com.yas.order.utils.JsonUtils.convertObjectToString;
+import static com.yas.order.utils.JsonUtils.createJsonErrorObject;
+import static com.yas.order.utils.JsonUtils.getAttributesNode;
+import static com.yas.order.utils.JsonUtils.getJsonValueOrThrow;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import com.yas.commonlibrary.exception.BadRequestException;
+import com.yas.commonlibrary.exception.NotFoundException;
+import com.yas.order.model.Checkout;
+import com.yas.order.model.enumeration.CheckoutProgress;
+import com.yas.order.model.enumeration.CheckoutState;
+import com.yas.order.repository.CheckoutRepository;
+import com.yas.order.service.PaymentService;
+import com.yas.order.utils.Constants;
+import com.yas.order.viewmodel.payment.CheckoutPaymentVm;
+import java.io.IOException;
+import java.util.Objects;
+import java.util.Optional;
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class OrderStatusConsumer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(OrderStatusConsumer.class);
+    private final PaymentService paymentService;
+    private final CheckoutRepository checkoutRepository;
+    private final ObjectMapper objectMapper;
+    private final Gson gson;
+
+    @KafkaListener(
+        topics = "${cdc.event.checkout.status.topic-name}",
+        groupId = "${cdc.event.checkout.status.group-id}"
+    )
+    @RetryableTopic(
+        attempts = "1"
+    )
+    public void listen(ConsumerRecord<?, ?> consumerRecord) {
+
+        if (Objects.isNull(consumerRecord)) {
+            LOGGER.info("ConsumerRecord is null");
+            return;
+        }
+        JsonObject valueObject = gson.fromJson((String) consumerRecord.value(), JsonObject.class);
+        processCheckoutEvent(valueObject);
+
+    }
+
+    private void processCheckoutEvent(JsonObject valueObject) {
+        Optional.ofNullable(valueObject)
+            .filter(value -> value.has("after"))
+            .map(value -> value.getAsJsonObject("after"))
+            .ifPresent(this::handleAfterJson);
+    }
+
+    private void handleAfterJson(JsonObject after) {
+
+        String id = getJsonValueOrThrow(after, Constants.Column.ID_COLUMN,
+            Constants.ErrorCode.ID_NOT_EXISTED);
+        String status = getJsonValueOrThrow(after, Constants.Column.STATUS_COLUMN,
+            Constants.ErrorCode.STATUS_NOT_EXISTED, id);
+        String progress = getJsonValueOrThrow(after, Constants.Column.CHECKOUT_PROGRESS_COLUMN,
+            Constants.ErrorCode.PROGRESS_NOT_EXISTED, id);
+
+        if (!isPaymentProcessing(status, progress)) {
+            LOGGER.info("Checkout record with ID {} lacks the status 'PAYMENT_PROCESSING' and progress 'STOCK_LOCKED'",
+                id);
+            return;
+        }
+
+        LOGGER.info("Checkout record with ID {} has the status 'PAYMENT_PROCESSING' and the process 'STOCK_LOCKED'",
+            id);
+
+        Checkout checkout = checkoutRepository
+            .findById(id)
+            .orElseThrow(() -> new NotFoundException(Constants.ErrorCode.CHECKOUT_NOT_FOUND, id));
+
+        processPaymentAndUpdateCheckout(checkout);
+    }
+
+    private boolean isPaymentProcessing(String status, String process) {
+        return CheckoutState.PAYMENT_PROCESSING.name().equalsIgnoreCase(status)
+            && CheckoutProgress.STOCK_LOCKED.name().equalsIgnoreCase(process);
+    }
+
+    private void processPaymentAndUpdateCheckout(Checkout checkout) {
+
+        try {
+            Long paymentId = processPayment(checkout);
+            checkout.setProgress(CheckoutProgress.PAYMENT_CREATED);
+            checkout.setLastError(null);
+
+            ObjectNode updatedAttributes = updateAttributesWithPayment(checkout.getAttributes(), paymentId);
+            checkout.setAttributes(convertObjectToString(objectMapper, updatedAttributes));
+
+        } catch (Exception e) {
+
+            checkout.setProgress(CheckoutProgress.PAYMENT_CREATED_FAILED);
+
+            ObjectNode error = createJsonErrorObject(objectMapper, CheckoutProgress.PAYMENT_CREATED_FAILED.name(),
+                e.getMessage());
+            checkout.setLastError(convertObjectToString(objectMapper, error));
+
+            LOGGER.error(e.getMessage());
+            throw new BadRequestException(Constants.ErrorCode.PROCESS_CHECKOUT_FAILED, checkout.getId());
+
+        } finally {
+            checkoutRepository.save(checkout);
+        }
+    }
+
+    private Long processPayment(Checkout checkout) {
+
+        CheckoutPaymentVm requestDto = new CheckoutPaymentVm(
+            checkout.getId(),
+            checkout.getPaymentMethodId(),
+            checkout.getTotalAmount()
+        );
+
+        Long paymentId = paymentService.createPaymentFromEvent(requestDto);
+        LOGGER.info("Payment created successfully with ID: {}", paymentId);
+
+        return paymentId;
+    }
+
+    private ObjectNode updateAttributesWithPayment(String attributes, Long paymentId) throws IOException {
+
+        ObjectNode attributesNode = getAttributesNode(objectMapper, attributes);
+        attributesNode.put(Constants.Column.CHECKOUT_ATTRIBUTES_PAYMENT_ID_FIELD, paymentId);
+
+        return attributesNode;
+    }
+
+}

order/src/main/java/com/yas/order/mapper/CheckoutMapper.java

@@ -16,11 +16,22 @@
 public interface CheckoutMapper {
     @Mapping(target = "id", ignore = true)
     @Mapping(target = "checkout", ignore = true)
+    @Mapping(target = "checkoutId", ignore = true)
     CheckoutItem toModel(CheckoutItemPostVm checkoutItemPostVm);
 
     @Mapping(target = "id", ignore = true)
     @Mapping(target = "checkoutState", ignore = true)
-    @Mapping(target = "totalAmount", source = "totalAmount") // Ánh xạ tường minh cho totalAmount
+    @Mapping(target = "progress", ignore = true)
+    @Mapping(target = "customerId", ignore = true)
+    @Mapping(target = "shipmentMethodId", ignore = true)
+    @Mapping(target = "paymentMethodId", ignore = true)
+    @Mapping(target = "shippingAddressId", ignore = true)
+    @Mapping(target = "lastError", ignore = true)
+    @Mapping(target = "attributes", ignore = true)
+    @Mapping(target = "totalShipmentFee", ignore = true)
+    @Mapping(target = "totalShipmentTax", ignore = true)
+    @Mapping(target = "totalTax", ignore = true)
+    @Mapping(target = "totalAmount", source = "totalAmount")
     @Mapping(target = "totalDiscountAmount", source = "totalDiscountAmount")
     Checkout toModel(CheckoutPostVm checkoutPostVm);
 

order/src/main/java/com/yas/order/model/Checkout.java

@@ -1,6 +1,8 @@
 package com.yas.order.model;
 
+import com.yas.order.model.enumeration.CheckoutProgress;
 import com.yas.order.model.enumeration.CheckoutState;
+import com.yas.order.model.enumeration.PaymentMethod;
 import jakarta.persistence.Column;
 import jakarta.persistence.Entity;
 import jakarta.persistence.EnumType;
@@ -39,22 +41,21 @@ public class Checkout extends AbstractAuditEntity {
     @Enumerated(EnumType.STRING)
     private CheckoutState checkoutState;
 
-    @SuppressWarnings("unused")
-    private String progress;
+    @Enumerated(EnumType.STRING)
+    private CheckoutProgress progress;
 
     @SuppressWarnings("unused")
     private Long customerId;
 
     @SuppressWarnings("unused")
     private String shipmentMethodId;
 
-    @Column(name = "payment_method_id")
-    private String paymentMethodId;
+    @Enumerated(EnumType.STRING)
+    private PaymentMethod paymentMethodId;
 
     @SuppressWarnings("unused")
     private Long shippingAddressId;
 
-    @SuppressWarnings("unused")
     @JdbcTypeCode(SqlTypes.JSON)
     @Column(name = "last_error", columnDefinition = "jsonb")
     private String lastError;

order/src/main/java/com/yas/order/model/enumeration/CheckoutProgress.java

@@ -0,0 +1,20 @@
+package com.yas.order.model.enumeration;
+
+public enum CheckoutProgress {
+    INIT("Init"),
+    PROMOTION_CODE_APPLIED("Promotion code applied"),
+    PROMOTION_CODE_APPLIED_FAILED("Promotion Code applied failed"),
+    STOCK_LOCKED("Stock locked"),
+    STOCK_LOCKED_FAILED("Stock locked failed"),
+    PAYMENT_CREATED("Payment created"),
+    PAYMENT_CREATED_FAILED("Payment created failed");
+    private final String name;
+
+    CheckoutProgress(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+}

order/src/main/java/com/yas/order/model/enumeration/CheckoutState.java

@@ -1,7 +1,15 @@
 package com.yas.order.model.enumeration;
 
 public enum CheckoutState {
-    COMPLETED("Completed"), PENDING("Pending"), LOCK("LOCK");
+    COMPLETED("Completed"),
+    PENDING("Pending"),
+    LOCK("LOCK"),
+    CHECKED_OUT("Checked Out"),
+    PAYMENT_PROCESSING("Payment Processing"),
+    PAYMENT_FAILED("Payment Failed"),
+    PAYMENT_CONFIRMED("Payment Confirmed"),
+    FULFILLED("Fulfilled")
+    ;
     private final String name;
 
     CheckoutState(String name) {

order/src/main/java/com/yas/order/model/enumeration/PaymentMethod.java

@@ -1,5 +1,14 @@
 package com.yas.order.model.enumeration;
 
 public enum PaymentMethod {
-    COD, BANKING, PAYPAL
+    COD, BANKING, PAYPAL;
+
+    public static PaymentMethod fromValue(String value) {
+        try {
+            return PaymentMethod.valueOf(value.toUpperCase());
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("Invalid payment method: " + value);
+        }
+    }
 }
+

order/src/main/java/com/yas/order/service/CheckoutService.java

@@ -9,6 +9,7 @@
 import com.yas.order.model.CheckoutItem;
 import com.yas.order.model.Order;
 import com.yas.order.model.enumeration.CheckoutState;
+import com.yas.order.model.enumeration.PaymentMethod;
 import com.yas.order.repository.CheckoutItemRepository;
 import com.yas.order.repository.CheckoutRepository;
 import com.yas.order.utils.AuthenticationUtils;
@@ -19,6 +20,7 @@
 import com.yas.order.viewmodel.checkout.CheckoutStatusPutVm;
 import com.yas.order.viewmodel.checkout.CheckoutVm;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
@@ -104,9 +106,14 @@ public Long updateCheckoutStatus(CheckoutStatusPutVm checkoutStatusPutVm) {
     }
 
     public void updateCheckoutPaymentMethod(String id, CheckoutPaymentMethodPutVm checkoutPaymentMethodPutVm) {
+
+        if (Objects.isNull(checkoutPaymentMethodPutVm.paymentMethodId())) {
+            return;
+        }
+
         Checkout checkout = checkoutRepository.findById(id)
             .orElseThrow(() -> new NotFoundException(CHECKOUT_NOT_FOUND, id));
-        checkout.setPaymentMethodId(checkoutPaymentMethodPutVm.paymentMethodId());
+        checkout.setPaymentMethodId(PaymentMethod.fromValue(checkoutPaymentMethodPutVm.paymentMethodId()));
         checkoutRepository.save(checkout);
     }
 }