#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 cart, String customer, String product, String tax, String payment) {
 }

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

@@ -0,0 +1,157 @@
+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.JsonElement;
+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.model.enumeration.PaymentMethod;
+import com.yas.order.model.request.CheckoutPaymentRequest;
+import com.yas.order.repository.CheckoutRepository;
+import com.yas.order.service.PaymentService;
+import com.yas.order.utils.Constants;
+import java.io.IOException;
+import java.math.BigDecimal;
+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.retry.annotation.Backoff;
+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(
+        backoff = @Backoff(delay = 5000)
+    )
+    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.CHECKOUT_ID_COLUMN,
+            Constants.ErrorCode.ID_NOT_EXISTED);
+        String status = getJsonValueOrThrow(after, Constants.Column.CHECKOUT_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 {} hasn't status 'PAYMENT_PROCESSING' and progress 'STOCK_LOCKED'",
+                id);
+            return;
+        }
+
+        LOGGER.info("Checkout record with ID {} has status 'PAYMENT_PROCESSING' and process 'STOCK_LOCKED'",
+            id);
+
+        Checkout checkout = checkoutRepository
+            .findById(id)
+            .orElseThrow(() -> new NotFoundException(Constants.ErrorCode.CHECKOUT_NOT_FOUND, id));
+
+        processPaymentAndUpdateCheckout(after, checkout);
+    }
+
+    private boolean isPaymentProcessing(String status, String process) {
+        return CheckoutState.PAYMENT_PROCESSING.name().equalsIgnoreCase(status)
+            && CheckoutProgress.STOCK_LOCKED.name().equalsIgnoreCase(process);
+    }
+
+    private void processPaymentAndUpdateCheckout(JsonObject after, Checkout checkout) {
+
+        try {
+
+            String paymentMethodId = getJsonValueOrThrow(after, Constants.Column.CHECKOUT_PAYMENT_METHOD_ID_COLUMN,
+                Constants.ErrorCode.PAYMENT_METHOD_NOT_EXISTED, checkout.getId());
+
+            BigDecimal totalAmount = Optional.ofNullable(after.get(Constants.Column.CHECKOUT_TOTAL_AMOUNT_COLUMN))
+                .filter(jsonElement -> !jsonElement.isJsonNull())
+                .map(JsonElement::getAsBigDecimal)
+                .orElse(null);
+
+            Long paymentId = processPayment(checkout.getId(), PaymentMethod.fromValue(paymentMethodId), totalAmount);
+            checkout.setProgress(CheckoutProgress.PAYMENT_CREATED);
+
+            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(String id, PaymentMethod paymentMethod, BigDecimal totalAmount) {
+
+        CheckoutPaymentRequest requestDto = CheckoutPaymentRequest.builder()
+            .checkoutId(id)
+            .paymentMethod(paymentMethod)
+            .totalAmount(totalAmount)
+            .build();
+
+        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

@@ -15,10 +15,23 @@
 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 = "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 = "totalAmount", ignore = true)
+    @Mapping(target = "totalShipmentFee", ignore = true)
+    @Mapping(target = "totalShipmentTax", ignore = true)
+    @Mapping(target = "totalTax", ignore = true)
+    @Mapping(target = "totalDiscountAmount", ignore = true)
     Checkout toModel(CheckoutPostVm checkoutPostVm);
 
     CheckoutItemVm toVm(CheckoutItem checkoutItem);

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

@@ -1,5 +1,6 @@
 package com.yas.order.model;
 
+import com.yas.order.model.enumeration.CheckoutProgress;
 import com.yas.order.model.enumeration.CheckoutState;
 import jakarta.persistence.Column;
 import jakarta.persistence.Entity;
@@ -39,8 +40,8 @@ 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;
@@ -54,7 +55,6 @@ public class Checkout extends AbstractAuditEntity {
     @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,27 @@
 package com.yas.order.model.enumeration;
 
 public enum PaymentMethod {
-    COD, BANKING, PAYPAL
+
+    COD("COD"),
+    BANKING("BANKING"),
+    PAYPAL("PAYPAL");
+
+    private final String value;
+
+    PaymentMethod(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public static PaymentMethod fromValue(String value) {
+        for (PaymentMethod method : PaymentMethod.values()) {
+            if (method.getValue().equalsIgnoreCase(value)) {
+                return method;
+            }
+        }
+        throw new IllegalArgumentException("Unknown payment method: " + value);
+    }
 }