Apply saga framework for yas (#723)

* Apply saga framework for yas

* Add tag custom version for postgres debezium

.github/workflows/cart-ci.yaml

@@ -25,7 +25,7 @@ jobs:
           fetch-depth: 0  # Shallow clones should be disabled for a better relevancy of analysis
       - uses: ./.github/workflows/actions
       - name: Run Maven Build Command
-        run: mvn clean install -DskipTests -f cart
+        run: mvn clean install -DskipTests
       - name: Run Maven Test
         run: mvn test -f cart
       - name: Unit Test Results

.github/workflows/order-ci.yaml

@@ -25,7 +25,7 @@ jobs:
           fetch-depth: 0 # Shallow clones should be disabled for a better relevancy of analysis
       - uses: ./.github/workflows/actions
       - name: Run Maven Build Command
-        run: mvn clean install -DskipTests -f order
+        run: mvn clean install -DskipTests
       - name: Run Maven Test
         run: mvn test -f order
       - name: Unit Test Results

cart/pom.xml

@@ -109,6 +109,12 @@
 			<groupId>org.liquibase</groupId>
 			<artifactId>liquibase-core</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.yas</groupId>
+			<artifactId>saga</artifactId>
+			<version>0.0.1-SNAPSHOT</version>
+			<scope>compile</scope>
+		</dependency>
 	</dependencies>
 	<dependencyManagement>
 		<dependencies>

cart/src/main/java/com/yas/cart/controller/CartController.java

@@ -11,6 +11,8 @@
 import jakarta.validation.constraints.NotEmpty;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
@@ -40,7 +42,7 @@ public ResponseEntity<List<CartGetDetailVm>> listCartDetailByCustomerId(@PathVar
     public ResponseEntity<CartGetDetailVm> getLastCart(Principal principal) {
         if (principal == null)
             return ResponseEntity.ok(null);
-        return ResponseEntity.ok(cartService.getLastCart());
+        return ResponseEntity.ok(cartService.getLastCart(principal.getName()));
     }
 
     @PostMapping(path = "/storefront/carts")
@@ -56,18 +58,21 @@ public ResponseEntity<CartGetDetailVm> createCart(@Valid @RequestBody @NotEmpty
 
     @PutMapping("cart-item")
     public ResponseEntity<CartItemPutVm> updateCart(@Valid @RequestBody CartItemVm cartItemVm) {
-        return new ResponseEntity<>(cartService.updateCartItems(cartItemVm), HttpStatus.OK);
+        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+        return new ResponseEntity<>(cartService.updateCartItems(cartItemVm, auth.getName()), HttpStatus.OK);
     }
 
     @DeleteMapping("/storefront/cart-item")
     public ResponseEntity<Void> removeCartItemByProductId(@RequestParam Long productId) {
-        cartService.removeCartItemByProductId(productId);
+        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+        cartService.removeCartItemByProductId(productId, auth.getName());
         return ResponseEntity.noContent().build();
     }
 
     @DeleteMapping("/storefront/cart-item/multi-delete")
     public ResponseEntity<Void> removeCartItemListByProductIdList(@RequestParam List<Long> productIds) {
-        cartService.removeCartItemListByProductIdList(productIds);
+        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+        cartService.removeCartItemListByProductIdList(productIds, auth.getName());
         return ResponseEntity.noContent().build();
     }
 

cart/src/main/java/com/yas/cart/repository/CartItemRepository.java

@@ -5,6 +5,7 @@
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 import java.util.Optional;
@@ -16,8 +17,10 @@ public interface CartItemRepository extends JpaRepository<CartItem, Long> {
 
     Optional<CartItem> findByCartIdAndProductId(Long cartId, Long productId);
 
+    @Transactional
     void deleteByCartIdAndProductId(Long cartId, Long productId);
 
+    @Transactional
     void deleteByCartIdAndProductIdIn(Long cartId, List<Long> productIds);
 
     @Query("select sum(ci.quantity) from CartItem ci where ci.cart.id = ?1")

cart/src/main/java/com/yas/cart/saga/HandlerDispatcherRegister.java

@@ -0,0 +1,17 @@
+package com.yas.cart.saga;
+
+import com.yas.cart.saga.handler.CartItemCommandHandler;
+import io.eventuate.tram.commands.consumer.CommandDispatcher;
+import io.eventuate.tram.sagas.participant.SagaCommandDispatcherFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class HandlerDispatcherRegister {
+    @Bean
+    public CommandDispatcher consumerCommandDispatcher(CartItemCommandHandler target,
+                                                       SagaCommandDispatcherFactory sagaCommandDispatcherFactory) {
+
+        return sagaCommandDispatcherFactory.make("cartItemCommandDispatcher", target.commandHandlerDefinitions());
+    }
+}

cart/src/main/java/com/yas/cart/saga/handler/CartItemCommandHandler.java

@@ -0,0 +1,43 @@
+package com.yas.cart.saga.handler;
+
+import com.yas.cart.exception.BadRequestException;
+import com.yas.cart.exception.NotFoundException;
+import com.yas.cart.service.CartService;
+import com.yas.saga.cart.command.DeleteCartItemCommand;
+import com.yas.saga.cart.reply.DeleteCartItemFailure;
+import com.yas.saga.cart.reply.DeleteCartItemSuccess;
+import io.eventuate.tram.commands.consumer.CommandHandlers;
+import io.eventuate.tram.commands.consumer.CommandMessage;
+import io.eventuate.tram.messaging.common.Message;
+import io.eventuate.tram.sagas.participant.SagaCommandHandlersBuilder;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import static io.eventuate.tram.commands.consumer.CommandHandlerReplyBuilder.withFailure;
+import static io.eventuate.tram.commands.consumer.CommandHandlerReplyBuilder.withSuccess;
+
+@Component
+@RequiredArgsConstructor
+public class CartItemCommandHandler {
+
+    private final CartService cartService;
+
+    public CommandHandlers commandHandlerDefinitions() {
+        return SagaCommandHandlersBuilder
+                .fromChannel("cartService")
+                .onMessage(DeleteCartItemCommand.class, this::deleteCartItem)
+                .build();
+    }
+
+    private Message deleteCartItem(CommandMessage<DeleteCartItemCommand> cm) {
+        DeleteCartItemCommand cmd = cm.getCommand();
+        try {
+            this.cartService.removeCartItemListByProductIdList(cmd.getProductIds(), cmd.getCustomerId());
+            String productIdsStr = StringUtils.join(cmd.getProductIds(), ",");
+            return withSuccess(new DeleteCartItemSuccess("Delete all card items of product ids [" + productIdsStr + "] success"));
+        } catch (BadRequestException | NotFoundException e) {
+            return withFailure(new DeleteCartItemFailure(e.getMessage()));
+        }
+    }
+}

cart/src/main/java/com/yas/cart/service/CartService.java

@@ -8,7 +8,6 @@
 import com.yas.cart.repository.CartRepository;
 import com.yas.cart.utils.Constants;
 import com.yas.cart.viewmodel.*;
-import org.springframework.http.HttpStatus;
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Service;
@@ -88,9 +87,8 @@ public CartGetDetailVm addToCart(List<CartItemVm> cartItemVms) {
         return CartGetDetailVm.fromModel(cart);
     }
 
-    public CartGetDetailVm getLastCart() {
-        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
-        return cartRepository.findByCustomerIdAndOrderIdIsNull(auth.getName())
+    public CartGetDetailVm getLastCart(String customerId) {
+        return cartRepository.findByCustomerIdAndOrderIdIsNull(customerId)
                 .stream().reduce((first, second) -> second)
                 .map(CartGetDetailVm::fromModel).orElse(CartGetDetailVm.fromModel(new Cart()));
     }
@@ -103,8 +101,8 @@ private CartItem getCartItemByProductId(Set<CartItem> cartItems, Long productId)
         return new CartItem();
     }
 
-    public CartItemPutVm updateCartItems(CartItemVm cartItemVm) {
-        CartGetDetailVm currentCart = getLastCart();
+    public CartItemPutVm updateCartItems(CartItemVm cartItemVm, String customerId) {
+        CartGetDetailVm currentCart = getLastCart(customerId);
 
         validateCart(currentCart, cartItemVm.productId());
 
@@ -124,11 +122,9 @@ public CartItemPutVm updateCartItems(CartItemVm cartItemVm) {
         }
     }
 
-    @Transactional
-    public void removeCartItemListByProductIdList(List<Long> productIdList) {
-        CartGetDetailVm currentCart = getLastCart();
-        productIdList.stream().forEach(id -> validateCart(currentCart, id));
-
+    public void removeCartItemListByProductIdList(List<Long> productIdList, String customerId) {
+        CartGetDetailVm currentCart = getLastCart(customerId);
+        productIdList.forEach(id -> validateCart(currentCart, id));
         cartItemRepository.deleteByCartIdAndProductIdIn(currentCart.id(), productIdList);
     }
 
@@ -145,8 +141,8 @@ private void validateCart(CartGetDetailVm cart, Long productId) {
     }
 
     @Transactional
-    public void removeCartItemByProductId(Long productId) {
-        CartGetDetailVm currentCart = getLastCart();
+    public void removeCartItemByProductId(Long productId, String customerId) {
+        CartGetDetailVm currentCart = getLastCart(customerId);
 
         validateCart(currentCart, productId);
 

cart/src/main/resources/application.properties

@@ -8,6 +8,11 @@ management.endpoints.web.exposure.include=prometheus
 management.metrics.distribution.percentiles-histogram.http.server.requests=true
 management.metrics.tags.application=${spring.application.name}
 
+eventuatelocal.kafka.bootstrap.servers=kafka:9092
+eventuatelocal.zookeeper.connection.string=zookeeper:2181
+eventuate.database.schema=eventuate
+spring.liquibase.parameters.eventualSlotName=cart
+
 logging.pattern.level=%5p [${spring.application.name:},%X{traceId:-},%X{spanId:-}]
 
 spring.security.oauth2.resourceserver.jwt.issuer-uri=http://identity/realms/Yas
@@ -26,6 +31,9 @@ spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.PostgreSQLDialec
 # Hibernate ddl auto (none, create, create-drop, validate, update)
 spring.jpa.hibernate.ddl-auto = none
 
+# Disable open in view transaction
+spring.jpa.open-in-view=false
+
 #Enable liquibase
 spring.liquibase.enabled=true
 

cart/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -2,4 +2,6 @@ databaseChangeLog:
   - includeAll:
       path: db/changelog/ddl/
   - includeAll:
-      path: db/changelog/data/
+      path: db/changelog/data/
+  - includeAll:
+      path: db/changelog/eventuate-dll/

docker-compose.yml

@@ -220,10 +220,12 @@ services:
     networks:
       - yas-network
   postgres:
-    image: debezium/postgres:15-alpine
+    image: debezium/postgres:15-alpine-custom
+    build: ./docker/postgres
     ports:
       - "5432:5432"
     volumes:
+      - ./docker/postgres/postgresql.conf.sample:/usr/share/postgresql/postgresql.conf.sample
       - ./postgres_init.sql:/docker-entrypoint-initdb.d/postgres_init.sql
       - postgres:/var/lib/postgresql/data
     command: postgres -c 'max_connections=500'
@@ -261,7 +263,7 @@ services:
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://kafka:29092
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
@@ -302,6 +304,52 @@ services:
     networks:
       - yas-network
 
+  eventuate-cdc:
+    image: eventuateio/eventuate-cdc-service:0.15.0.RELEASE
+    ports:
+      - "8099:8080"
+    depends_on:
+      - postgres
+      - kafka
+      - zookeeper
+    networks:
+      - yas-network
+    environment:
+      CDC_OPTS: "--debug"
+
+      EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS: kafka:29092
+      EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING: zookeeper:2181
+
+      EVENTUATE_CDC_READER_READER1_TYPE: postgres-wal
+      EVENTUATE_CDC_READER_READER1_DATASOURCEURL: jdbc:postgresql://postgres/order
+      EVENTUATE_CDC_READER_READER1_MONITORINGSCHEMA: eventuate
+      EVENTUATE_CDC_READER_READER1_DATASOURCEUSERNAME: admin
+      EVENTUATE_CDC_READER_READER1_DATASOURCEPASSWORD: admin
+      EVENTUATE_CDC_READER_READER1_DATASOURCEDRIVERCLASSNAME: org.postgresql.Driver
+      EVENTUATE_CDC_READER_READER1_LEADERSHIPLOCKPATH: /eventuate/cdc/leader/order
+      EVENTUATE_CDC_READER_READER1_OFFSETSTORAGETOPICNAME: db.history.common
+      EVENTUATE_CDC_READER_READER1_OUTBOXID: 1
+      EVENTUATE_CDC_READER_READER1_POSTGRESREPLICATIONSLOTNAME: eventuate_slot_order
+
+      EVENTUATE_CDC_READER_READER2_TYPE: postgres-wal
+      EVENTUATE_CDC_READER_READER2_DATASOURCEURL: jdbc:postgresql://postgres/cart
+      EVENTUATE_CDC_READER_READER2_MONITORINGSCHEMA: eventuate
+      EVENTUATE_CDC_READER_READER2_DATASOURCEUSERNAME: admin
+      EVENTUATE_CDC_READER_READER2_DATASOURCEPASSWORD: admin
+      EVENTUATE_CDC_READER_READER2_DATASOURCEDRIVERCLASSNAME: org.postgresql.Driver
+      EVENTUATE_CDC_READER_READER2_LEADERSHIPLOCKPATH: /eventuate/cdc/leader/cart
+      EVENTUATE_CDC_READER_READER2_OFFSETSTORAGETOPICNAME: db.history.common
+      EVENTUATE_CDC_READER_READER2_OUTBOXID: 2
+      EVENTUATE_CDC_READER_READER2_POSTGRESREPLICATIONSLOTNAME: eventuate_slot_cart
+
+      EVENTUATE_CDC_PIPELINE_PIPELINE1_TYPE: eventuate-tram
+      EVENTUATE_CDC_PIPELINE_PIPELINE1_READER: reader1
+      EVENTUATE_CDC_PIPELINE_PIPELINE1_EVENTUATEDATABASESCHEMA: eventuate
+
+      EVENTUATE_CDC_PIPELINE_PIPELINE2_TYPE: eventuate-tram
+      EVENTUATE_CDC_PIPELINE_PIPELINE2_READER: reader2
+      EVENTUATE_CDC_PIPELINE_PIPELINE2_EVENTUATEDATABASESCHEMA: eventuate
+
 networks:
   yas-network:
     driver: bridge

docker/postgres/Dockerfile

@@ -0,0 +1,7 @@
+FROM debezium/postgres:15-alpine
+ENV  WAL2JSON_TAG=wal2json_2_5
+RUN    apk add --no-cache --virtual .debezium-build-deps gcc clang15 llvm15 git make musl-dev pkgconf \
+         && git clone https://github.com/eulerto/wal2json -b master --single-branch \
+         && (cd /wal2json && git checkout tags/$WAL2JSON_TAG -b $WAL2JSON_TAG && make && make install) \
+         && rm -rf wal2json \
+         && apk del .debezium-build-deps

docker/postgres/postgresql.conf.sample

@@ -0,0 +1,16 @@
+# LOGGING
+# log_min_error_statement = fatal
+# log_min_messages = DEBUG1
+
+# CONNECTION
+listen_addresses = '*'
+
+# MODULES
+shared_preload_libraries = 'decoderbufs,wal2json'
+
+# REPLICATION
+wal_level = logical             # minimal, archive, hot_standby, or logical (change requires restart)
+max_wal_senders = 20             # max number of walsender processes (change requires restart)
+#wal_keep_segments = 4          # in logfile segments, 16MB each; 0 disables
+#wal_sender_timeout = 60s       # in milliseconds; 0 disables
+max_replication_slots = 20       # max number of replication slots (change requires restart)

k8s/charts/cart/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.0
+version: 0.2.0
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

k8s/charts/cart/values.yaml

@@ -10,3 +10,12 @@ backend:
     enabled: true
     host: api.yas.local.com
     path: /cart
+  extraVolumes:
+    - name: cart-application-config
+      configMap:
+        name: cart-application-configmap
+  extraVolumeMounts:
+    - name: cart-application-config
+      mountPath: /opt/yas/cart
+  extraApplicationConfigPaths:
+    - /opt/yas/cart/cart-application.yaml

k8s/charts/eventuate-cdc/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/