#1180 & #1237 create payment and order in paypal

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/java/com/yas/order/service/CustomerServiceIT.java

@@ -6,19 +6,22 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.verify;
 
+import com.yas.order.config.KafkaIntegrationTestConfiguration;
 import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
 import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Import;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 @SpringBootTest
 @Testcontainers
+@Import({KafkaIntegrationTestConfiguration.class})
 class CustomerServiceIT {
     @Container
     @ServiceConnection

order/src/it/java/com/yas/order/service/OrderServiceIT.java

@@ -9,6 +9,7 @@
 import com.yas.commonlibrary.IntegrationTestConfiguration;
 import com.yas.commonlibrary.exception.NotFoundException;
 import com.yas.order.OrderApplication;
+import com.yas.order.config.KafkaIntegrationTestConfiguration;
 import com.yas.order.model.Order;
 import com.yas.order.model.enumeration.OrderStatus;
 import com.yas.order.model.enumeration.PaymentStatus;
@@ -37,7 +38,7 @@
 import org.springframework.data.util.Pair;
 
 @SpringBootTest(classes = OrderApplication.class)
-@Import(IntegrationTestConfiguration.class)
+@Import({IntegrationTestConfiguration.class, KafkaIntegrationTestConfiguration.class})
 @AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
 class OrderServiceIT {
 

order/src/it/java/com/yas/order/service/ProductServiceIT.java

@@ -6,22 +6,24 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.verify;
 
+import com.yas.order.config.KafkaIntegrationTestConfiguration;
 import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
 import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
-import java.util.List;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Import;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 @SpringBootTest
 @Testcontainers
+@Import({KafkaIntegrationTestConfiguration.class})
 class ProductServiceIT {
     @Container
     @ServiceConnection

order/src/it/java/com/yas/order/service/TaxServiceIT.java

@@ -6,19 +6,22 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.verify;
 
+import com.yas.order.config.KafkaIntegrationTestConfiguration;
 import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
 import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Import;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 @SpringBootTest
 @Testcontainers
+@Import({KafkaIntegrationTestConfiguration.class})
 class TaxServiceIT {
     @Container
     @ServiceConnection

order/src/it/resources/application.properties

@@ -12,4 +12,23 @@ 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.fulfillment.group-id=checkout-fulfillment
+cdc.event.checkout.confirmed.status.group-id=checkout-confirmed-status
+
+cdc.event.payment.topic-name=dbpayment.public.payment
+cdc.event.payment.order.id.update.group-id=payment-order-id-update
+cdc.event.payment.status.update.group-id=payment-status-update
+
+kafka.version=7.0.9
+
+spring.kafka.consumer.bootstrap-servers=kafka:9092
+spring.aop.proxy-target-class=true
+
+spring.kafka.producer.bootstrap-servers=kafka:9092
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
+
+spring.kafka.bootstrap-servers=localhost:9092

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/controller/CheckoutController.java

@@ -32,16 +32,27 @@ public ResponseEntity<CheckoutVm> createCheckout(@Valid @RequestBody CheckoutPos
         return ResponseEntity.ok(checkoutService.createCheckout(checkoutPostVm));
     }
 
+    @PostMapping("/storefront/checkouts/{id}/process-payment")
+    public ResponseEntity<Void> processPayment(@PathVariable String id) {
+        checkoutService.processPayment(id);
+        return ResponseEntity.ok().build();
+    }
+
     @PutMapping("/storefront/checkouts/status")
     public ResponseEntity<Long> updateCheckoutStatus(@Valid @RequestBody CheckoutStatusPutVm checkoutStatusPutVm) {
         return ResponseEntity.ok(checkoutService.updateCheckoutStatus(checkoutStatusPutVm));
     }
 
-    @GetMapping("/storefront/checkouts/{id}")
-    public ResponseEntity<CheckoutVm> getOrderWithItemsById(@PathVariable String id) {
+    @GetMapping("/storefront/checkouts/pending/{id}")
+    public ResponseEntity<CheckoutVm> getPendingCheckoutDetailsById(@PathVariable String id) {
         return ResponseEntity.ok(checkoutService.getCheckoutPendingStateWithItemsById(id));
     }
 
+    @GetMapping("/storefront/checkouts/{id}")
+    public ResponseEntity<CheckoutVm> getCheckoutById(@PathVariable String id) {
+        return ResponseEntity.ok(checkoutService.findCheckoutWithItemsById(id));
+    }
+
     @PutMapping("/storefront/checkouts/{id}/payment-method")
     @ApiResponses(value = {
         @ApiResponse(responseCode = ApiConstant.CODE_200, description = ApiConstant.OK,

order/src/main/java/com/yas/order/kafka/AppKafkaListenerConfigurer.java

@@ -0,0 +1,24 @@
+package com.yas.order.kafka;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaListenerConfigurer;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
+import org.springframework.validation.beanvalidation.LocalValidatorFactoryBean;
+
+@EnableKafka
+@Configuration
+public class AppKafkaListenerConfigurer implements KafkaListenerConfigurer {
+
+    private LocalValidatorFactoryBean validator;
+
+    public AppKafkaListenerConfigurer(LocalValidatorFactoryBean validator) {
+        this.validator = validator;
+    }
+
+    @Override
+    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
+        // Enable message validation
+        registrar.setValidator(this.validator);
+    }
+}