this will let joint owners own the same payments and avoid duplicates

CHANGELOG.md

@@ -1,6 +1,9 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [5.11.0](https://github.com/Backbase/stream-services/compare/5.10.0...5.11.0)
+### Changed
+- feature - improved payment ingestion to allow for joint owner accounts to be shared.
 ## [5.10.0](https://github.com/Backbase/stream-services/compare/5.9.1...5.10.0)
 ### Added
 - Introducing Grand Central customer canonical model: Processing the `PartyUpsertEvent` using the Legal Entity Saga via the [legal-entity-composition-service](stream-compositions/services/legal-entity-composition-service)

stream-compositions/services/payment-order-composition-service/src/main/resources/application-local.yml

@@ -24,7 +24,7 @@ backbase:
           direct-uri: http://localhost:8050
       payment:
         order:
-          direct-uri: http://localhost:8051
+          direct-uri: http://localhost:8090
       transaction:
         manager:
           direct-uri: http://localhost:8083
@@ -39,6 +39,8 @@ backbase:
     enabled: true
   stream:
     paymentorder:
+      types:
+        - INT_TRANS_CLOSED
       worker:
         deletePaymentOrder: false
 

stream-compositions/services/payment-order-composition-service/src/main/resources/application.yml

@@ -29,5 +29,7 @@ backbase:
         required-scope: api:service
   stream:
     paymentorder:
+      types:
+        - INT_TRANS_CLOSED
       worker:
         deletePaymentOrder: false

stream-compositions/services/payment-order-composition-service/src/test/java/com/backbase/stream/compositions/paymentorders/http/PaymentOrderControllerIT.java

@@ -46,7 +46,9 @@
 import static org.mockserver.model.HttpResponse.response;
 
 @DirtiesContext
-@SpringBootTest
+@SpringBootTest(properties = {
+        "backbase.stream.paymentorder.types=type1,type2,type3"
+})
 @AutoConfigureWebTestClient
 @ExtendWith({SpringExtension.class})
 @Slf4j

stream-compositions/services/payment-order-composition-service/src/test/resources/application.yml

@@ -42,6 +42,8 @@ backbase:
           direct-uri: http://localhost:8083/transaction-manager
       stream:
         payment-order:
+          types:
+            - INT_TRANS_CLOSED
           integration:
             direct-uri: http://localhost:18000
     http:

stream-payment-order/payment-order-core/src/main/java/com/backbase/stream/config/PaymentOrderServiceConfiguration.java

@@ -17,13 +17,15 @@
 import org.springframework.context.annotation.Configuration;
 
 @EnableConfigurationProperties({
-    PaymentOrderWorkerConfigurationProperties.class
+    PaymentOrderWorkerConfigurationProperties.class,
+    PaymentOrderTypeConfiguration.class
 })
 @AllArgsConstructor
 @Configuration
 public class PaymentOrderServiceConfiguration {
 
     private final PaymentOrderTypeMapper paymentOrderTypeMapper;
+    private final PaymentOrderTypeConfiguration paymentOrderTypeConfiguration;
 
     @Bean
     public PaymentOrderTaskExecutor paymentOrderTaskExecutor(PaymentOrdersApi paymentOrdersApi) {
@@ -39,7 +41,8 @@ public PaymentOrderUnitOfWorkExecutor paymentOrderUnitOfWorkExecutor(
             ArrangementsApi arrangementsApi) {
 
         return new PaymentOrderUnitOfWorkExecutor(paymentOrderUnitOfWorkRepository, paymentOrderTaskExecutor,
-                paymentOrderWorkerConfigurationProperties, paymentOrdersApi, arrangementsApi, paymentOrderTypeMapper);
+                paymentOrderWorkerConfigurationProperties, paymentOrdersApi, arrangementsApi, paymentOrderTypeMapper,
+                paymentOrderTypeConfiguration);
     }
 
     @Bean

stream-payment-order/payment-order-core/src/main/java/com/backbase/stream/config/PaymentOrderTypeConfiguration.java

@@ -0,0 +1,18 @@
+package com.backbase.stream.config;
+
+import java.util.List;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+@ConfigurationProperties(prefix = "backbase.stream.paymentorder")
+@Validated
+@Slf4j
+@Data
+public class PaymentOrderTypeConfiguration {
+
+    @NotNull
+    private List<String> types;
+}

stream-payment-order/payment-order-core/src/main/java/com/backbase/stream/paymentorder/PaymentOrderUnitOfWorkExecutor.java

@@ -16,11 +16,15 @@
 import com.backbase.dbs.arrangement.api.service.v2.model.AccountArrangementItem;
 import com.backbase.dbs.arrangement.api.service.v2.model.AccountArrangementItems;
 import com.backbase.dbs.arrangement.api.service.v2.model.AccountArrangementsFilter;
+import com.backbase.dbs.paymentorder.api.service.v3.model.AccessFilter;
+import com.backbase.dbs.paymentorder.api.service.v3.model.Status;
+import com.backbase.stream.config.PaymentOrderTypeConfiguration;
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import com.backbase.dbs.paymentorder.api.service.v3.PaymentOrdersApi;
@@ -51,24 +55,27 @@
 @Slf4j
 public class PaymentOrderUnitOfWorkExecutor extends UnitOfWorkExecutor<PaymentOrderTask> {
 
-    private static final PaymentOrderPostFilterRequest FILTER = new PaymentOrderPostFilterRequest().statuses(List.of(READY, ACCEPTED, PROCESSED, CANCELLED, REJECTED, CANCELLATION_PENDING));
+    private static final List<Status> FILTER = List.of(READY, ACCEPTED, PROCESSED, CANCELLED, REJECTED, CANCELLATION_PENDING);
 
     private static final int PAGE_SIZE = 1000;
 
     private final PaymentOrdersApi paymentOrdersApi;
     private final ArrangementsApi arrangementsApi;
     private final PaymentOrderTypeMapper paymentOrderTypeMapper;
+    private final PaymentOrderTypeConfiguration paymentOrderTypeConfiguration;
 
     public PaymentOrderUnitOfWorkExecutor(UnitOfWorkRepository<PaymentOrderTask, String> repository,
             StreamTaskExecutor<PaymentOrderTask> streamTaskExecutor,
             StreamWorkerConfiguration streamWorkerConfiguration,
             PaymentOrdersApi paymentOrdersApi,
             ArrangementsApi arrangementsApi,
-            PaymentOrderTypeMapper paymentOrderTypeMapper) {
+            PaymentOrderTypeMapper paymentOrderTypeMapper,
+            PaymentOrderTypeConfiguration paymentOrderTypeConfiguration) {
         super(repository, streamTaskExecutor, streamWorkerConfiguration);
         this.paymentOrdersApi = paymentOrdersApi;
         this.arrangementsApi = arrangementsApi;
         this.paymentOrderTypeMapper = paymentOrderTypeMapper;
+        this.paymentOrderTypeConfiguration = paymentOrderTypeConfiguration;
     }
 
     public Flux<UnitOfWork<PaymentOrderTask>> prepareUnitOfWork(List<PaymentOrderIngestRequest> items) {
@@ -101,19 +108,19 @@ private PaymentOrderIngestContext createPaymentOrderIngestContext(List<PaymentOr
      * Gets the list of payments that are persisted in DBS for a specific user.
      * The transfers have been divided by destination product type.
      *
-     * @param paymentOrderIngestContext2 Holds all the Ingestion details.
+     * @param paymentOrderIngestContext Holds all the Ingestion details.
      * @return A Mono of List of GetPaymentOrderResponse.
      */
-    private @NotNull @Valid Mono<PaymentOrderIngestContext> getPersistedScheduledTransfers(PaymentOrderIngestContext paymentOrderIngestContext2) {
+    private @NotNull @Valid Mono<PaymentOrderIngestContext> getPersistedScheduledTransfers(PaymentOrderIngestContext paymentOrderIngestContext) {
 
         List<GetPaymentOrderResponse> listOfPayments = new ArrayList<>();
 
-        return getPayments(paymentOrderIngestContext2.internalUserId())
+        return getPayments(paymentOrderIngestContext.arrangementIds())
                 .map(response -> {
                     listOfPayments.addAll(response.getPaymentOrders());
                     return listOfPayments;
                 })
-                .map(getPaymentOrderResponses -> paymentOrderIngestContext2.existingPaymentOrder(getPaymentOrderResponses))
+                .map(getPaymentOrderResponses -> paymentOrderIngestContext.existingPaymentOrder(getPaymentOrderResponses))
                 .doOnSuccess(result ->
                         log.debug("Successfully fetched dbs scheduled payment orders"));
     }
@@ -136,15 +143,21 @@ private Mono<AccountArrangementItems> getArrangement(PaymentOrderPostRequest pay
     /**
      * Calls the payment order service to retrieve existing payments.
      *
-     * @param internalUserId   The user's internal id that came with the Payments.
+     * @param arrangementIds   Check for all the arrangements that belong to this PMT.
      * @return A Mono with the response from the service api.
      */
-    private Mono<PaymentOrderPostFilterResponse> getPayments(String internalUserId) {
+    private Mono<PaymentOrderPostFilterResponse> getPayments(List<AccountArrangementItems> arrangementIds) {
 
-        if (isEmptyUserId(internalUserId)) {
+        if (isEmptyArrangmetIds(arrangementIds)) {
             return Mono.just(new PaymentOrderPostFilterResponse().paymentOrders(emptyList()).totalElements(new BigDecimal(0)));
         }
-        return pullFromDBS(internalUserId).map(result -> {
+        List<String> allArrangementIds = arrangementIds
+                .stream().flatMap(accountArrangementItems ->
+                        accountArrangementItems.getArrangementElements().stream())
+                .map(AccountArrangementItem::getId)
+                .collect(Collectors.toList());
+
+        return pullFromDBS(allArrangementIds).map(result -> {
             final var response = new PaymentOrderPostFilterResponse();
             response.setPaymentOrders(result);
             response.setTotalElements(new BigDecimal(result.size()));
@@ -224,24 +237,34 @@ private record DBSPaymentOrderPageResult(int next, int total, List<GetPaymentOrd
 
     }
 
-    private Mono<List<GetPaymentOrderResponse>> pullFromDBS(final @NotNull String userid) {
-        return defer(() -> retrieveNextPage(0, userid)
+    private Mono<List<GetPaymentOrderResponse>> pullFromDBS(final @NotNull List<String> arrangementIds) {
+        return defer(() -> retrieveNextPage(0, arrangementIds)
                 .expand(page -> {
                     // If there are no more pages, return an empty flux.
                     if (page.next >= page.total || page.requests.isEmpty()) {
                         return empty();
                     } else {
-                        return retrieveNextPage(page.next, userid);
+                        return retrieveNextPage(page.next, arrangementIds);
                     }
                 }))
                 .collectList()
                 .map(pages -> pages.stream().flatMap(page -> page.requests.stream()).toList());
     }
 
-    private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final @NotNull String userId) {
+    private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final @NotNull List<String> arrangementIds) {
+        List<String> paymentTypes = paymentOrderTypeConfiguration.getTypes();
+        var paymentOrderPostFilterRequest = new PaymentOrderPostFilterRequest();
+        List<AccessFilter> accessFilters = paymentTypes.stream()
+                .map(paymentType -> new AccessFilter()
+                        .paymentType(paymentType)
+                        .arrangementIds(arrangementIds))
+                .collect(Collectors.toList());
+        paymentOrderPostFilterRequest.setAccessFilters(accessFilters);
+        paymentOrderPostFilterRequest.setStatuses(FILTER);
+
         return paymentOrdersApi.postFilterPaymentOrders(null, null, null, null, null, null, null, null, null,
-                        null, null, null, userId, null, null, currentCount / PAGE_SIZE, PAGE_SIZE, null,
-                        null, FILTER)
+                        null, null, null, null, null, null, currentCount / PAGE_SIZE, PAGE_SIZE, null,
+                        null, paymentOrderPostFilterRequest)
                 .retryWhen(fixedDelay(3, Duration.of(2000, MILLIS)).filter(
                         t -> t instanceof WebClientRequestException
                                 || t instanceof WebClientResponseException.ServiceUnavailable))
@@ -252,6 +275,10 @@ private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final
                 });
     }
 
+    private boolean isEmptyArrangmetIds(List<AccountArrangementItems> arrangementItems) {
+        return arrangementItems == null || arrangementItems.isEmpty();
+    }
+
     private boolean isEmptyUserId(String userId) {
         return userId == null || userId.isBlank();
     }

stream-payment-order/payment-order-core/src/test/java/com/backbase/stream/config/PaymentOrderServiceConfigurationTest.java

@@ -24,6 +24,10 @@ void configurationTest() {
             .withBean(DbsApiClientsAutoConfiguration.class)
             .withBean(InterServiceWebClientConfiguration.class)
             .withUserConfiguration(PaymentOrderServiceConfiguration.class)
+            .withUserConfiguration(PaymentOrderTypeConfiguration.class)
+            .withPropertyValues(
+                    "backbase.stream.paymentorder.types=type1,type2,type3"
+            )
             .run(context -> {
                 assertThat(context).hasSingleBean(PaymentOrderTaskExecutor.class);
             });

stream-payment-order/payment-order-core/src/test/java/com/backbase/stream/task/PaymentOrderUnitOfWorkExecutorTest.java

@@ -15,6 +15,7 @@
 import com.backbase.dbs.paymentorder.api.service.v3.model.PaymentOrderPostRequest;
 import com.backbase.dbs.paymentorder.api.service.v3.model.PaymentOrderPostResponse;
 import com.backbase.stream.common.PaymentOrderBaseTest;
+import com.backbase.stream.config.PaymentOrderTypeConfiguration;
 import com.backbase.stream.config.PaymentOrderWorkerConfigurationProperties;
 import com.backbase.stream.model.request.NewPaymentOrderIngestRequest;
 import com.backbase.stream.model.request.PaymentOrderIngestRequest;
@@ -25,6 +26,7 @@
 import com.backbase.stream.worker.repository.UnitOfWorkRepository;
 import java.math.BigDecimal;
 
+import java.util.ArrayList;
 import org.apache.commons.lang3.StringUtils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -52,6 +54,8 @@ public class PaymentOrderUnitOfWorkExecutorTest extends PaymentOrderBaseTest {
     @Mock
     private UnitOfWorkRepository<PaymentOrderTask, String> repository;
 
+    private PaymentOrderTypeConfiguration paymentOrderTypeConfiguration = new PaymentOrderTypeConfiguration();
+
     private final PaymentOrderTaskExecutor streamTaskExecutor = new PaymentOrderTaskExecutor(paymentOrdersApi);
 
     private final PaymentOrderWorkerConfigurationProperties streamWorkerConfiguration = new PaymentOrderWorkerConfigurationProperties();
@@ -61,9 +65,13 @@ public class PaymentOrderUnitOfWorkExecutorTest extends PaymentOrderBaseTest {
 
     @BeforeEach
     void setup() {
+        List<String> pmtTypes = new ArrayList<>();
+        pmtTypes.add("INTRA_PMT");
+        paymentOrderTypeConfiguration.setTypes(pmtTypes);
         paymentOrderUnitOfWorkExecutor = new PaymentOrderUnitOfWorkExecutor(
                 repository, streamTaskExecutor, streamWorkerConfiguration,
-                paymentOrdersApi, arrangementsApi, paymentOrderTypeMapper);
+                paymentOrdersApi, arrangementsApi, paymentOrderTypeMapper,
+                paymentOrderTypeConfiguration);
     }
 
     @Test
@@ -142,7 +150,7 @@ void test_prepareunitofwork_blankuserid() {
 
         paymentOrderUnitOfWorkExecutor = new PaymentOrderUnitOfWorkExecutor(
                 repository, streamTaskExecutor, streamWorkerConfiguration,
-                paymentOrdersApi, arrangementsApi, null);
+                paymentOrdersApi, arrangementsApi, null, paymentOrderTypeConfiguration);
 
         paymentOrderPostRequest.get(0).setInternalUserId(StringUtils.EMPTY);
         paymentOrderPostRequest.get(1).setInternalUserId(null);