Added ExecutionDateFrom filter in payment order based on configuration

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [4.1.9](https://github.com/Backbase/stream-services/compare/4.1.9...4.1.9)
+### Changed.
+- Adding configuration to filter payments by ExecutionDateFrom based in offset period days
+
 ## [4.1.8](https://github.com/Backbase/stream-services/compare/4.1.7...4.1.8)
 ### Changed.
 - Adding ingestion mode support for batch product groups
@@ -12,11 +16,11 @@ All notable changes to this project will be documented in this file.
 ## [4.1.6](https://github.com/Backbase/stream-services/compare/4.1.4...4.1.6)
 ### Fix
 - Fix arrangements being retrieved per payment order instead of once per payment order ingestion request.
-- 
+
 ## [4.1.4](https://github.com/Backbase/stream-services/compare/4.1.3...4.1.4)
 ### Fix
 - Fixed deletion of non-repository custom data-group items on data-group update
-- 
+
 ## [4.1.3](https://github.com/Backbase/stream-services/compare/4.1.3...4.1.4)
 ### Changed
 - Query for existing payments by using arrangement IDs instead of user IDs. This will eliminate duplicate payments from being ingested when joint owners are added.

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

@@ -15,4 +15,6 @@ public class PaymentOrderTypeConfiguration {
 
     @NotNull
     private List<String> types;
+
+    private Long startOffsetInDays;
 }

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

@@ -23,6 +23,7 @@
 import com.backbase.dbs.paymentorder.api.service.v2.model.Status;
 import java.math.BigDecimal;
 import java.time.Duration;
+import java.time.LocalDate;
 import com.backbase.dbs.paymentorder.api.service.v2.model.AccessFilter;
 import com.backbase.stream.config.PaymentOrderTypeConfiguration;
 import java.util.*;
@@ -236,20 +237,21 @@ private record DBSPaymentOrderPageResult(int next, int total, List<GetPaymentOrd
     }
 
     private Mono<List<GetPaymentOrderResponse>> pullFromDBS(final @NotNull List<String> arrangementIds) {
-        return defer(() -> retrieveNextPage(0, arrangementIds)
+        LocalDate startDate = paymentOrderTypeConfiguration.getStartOffsetInDays() != null ? LocalDate.now().minusDays(paymentOrderTypeConfiguration.getStartOffsetInDays()) : null;
+        return defer(() -> retrieveNextPage(0, arrangementIds, startDate)
             .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, arrangementIds);
+                    return retrieveNextPage(page.next, arrangementIds, startDate);
                 }
             }))
             .collectList()
             .map(pages -> pages.stream().flatMap(page -> page.requests.stream()).toList());
     }
 
-    private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final @NotNull List<String> arrangementIds) {
+    private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final @NotNull List<String> arrangementIds, LocalDate executionDateFrom) {
         List<String> paymentTypes = paymentOrderTypeConfiguration.getTypes();
         var paymentOrderPostFilterRequest = new PaymentOrderPostFilterRequest();
         List<AccessFilter> accessFilters = paymentTypes.stream()
@@ -260,7 +262,7 @@ private Mono<DBSPaymentOrderPageResult> retrieveNextPage(int currentCount, final
         paymentOrderPostFilterRequest.setAccessFilters(accessFilters);
         paymentOrderPostFilterRequest.setStatuses(FILTER);
 
-        return paymentOrdersApi.postFilterPaymentOrders(null, null, null, null, null, null, null, null,
+        return paymentOrdersApi.postFilterPaymentOrders(null, null, null, executionDateFrom, null, null, null, null,
                 null, null, null, null, null, null, currentCount / PAGE_SIZE, PAGE_SIZE, null,
                 null, paymentOrderPostFilterRequest)
             .retryWhen(fixedDelay(3, Duration.of(2000, MILLIS)).filter(

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

@@ -1,11 +1,23 @@
 package com.backbase.stream.task;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.lenient;
-
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
 import com.backbase.dbs.arrangement.api.service.v2.ArrangementsApi;
 import com.backbase.dbs.arrangement.api.service.v2.model.AccountArrangementItem;
 import com.backbase.dbs.arrangement.api.service.v2.model.AccountArrangementItems;
@@ -24,23 +36,10 @@
 import com.backbase.stream.paymentorder.PaymentOrderUnitOfWorkExecutor;
 import com.backbase.stream.worker.model.UnitOfWork;
 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;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.junit.jupiter.MockitoExtension;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
-import java.util.List;
-
 @ExtendWith(MockitoExtension.class)
 public class PaymentOrderUnitOfWorkExecutorTest extends PaymentOrderBaseTest {
 
@@ -53,6 +52,9 @@ public class PaymentOrderUnitOfWorkExecutorTest extends PaymentOrderBaseTest {
     @Mock
     private UnitOfWorkRepository<PaymentOrderTask, String> repository;
 
+    @Captor
+    ArgumentCaptor<LocalDate> fromDateCaptor;
+
     private PaymentOrderTypeConfiguration paymentOrderTypeConfiguration = new PaymentOrderTypeConfiguration();
 
     private final PaymentOrderTaskExecutor streamTaskExecutor = new PaymentOrderTaskExecutor(paymentOrdersApi);
@@ -145,7 +147,85 @@ void test_prepareUnitOfWork_paymentOrderPostRequestFlux() {
                 })
                 .verifyComplete();
     }
+
+    @Test
+    void test_prepareUnitOfWork_paymentOrderWithStartDateOffset() {
+        paymentOrderTypeConfiguration.setStartOffsetInDays(30l);
+        Flux<PaymentOrderPostRequest> paymentOrderPostRequestFlux = Flux.fromIterable(paymentOrderPostRequest);
+
+        PaymentOrderPostResponse paymentOrderPostResponse = new PaymentOrderPostResponse()
+                .id("po_post_resp_id")
+                .putAdditionsItem("key", "val");
+
+        lenient().when(paymentOrdersApi.postPaymentOrder(any()))
+                .thenReturn(Mono.just(paymentOrderPostResponse));
+
+        GetPaymentOrderResponse getPaymentOrderResponse = new GetPaymentOrderResponse()
+                .id("arrangementId_1")
+                .bankReferenceId("bankReferenceId_1");
+        PaymentOrderPostFilterResponse paymentOrderPostFilterResponse = new PaymentOrderPostFilterResponse()
+                .addPaymentOrdersItem(getPaymentOrderResponse)
+                .totalElements(new BigDecimal(1));
+        doReturn(Mono.just(paymentOrderPostFilterResponse)).when(paymentOrdersApi).postFilterPaymentOrders(any(), any(), any(), fromDateCaptor.capture(), any(),any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any());
+
+        AccountArrangementItem accountArrangementItem = new AccountArrangementItem()
+                .id("arrangementId_1")
+                .externalArrangementId("externalArrangementId_1");
+        AccountArrangementItems accountArrangementItems = new AccountArrangementItems()
+                .addArrangementElementsItem(accountArrangementItem);
+
+        Mockito.when(arrangementsApi.postFilter(Mockito.any()))
+                .thenReturn(Mono.just(accountArrangementItems));
+
+        StepVerifier.create(paymentOrderUnitOfWorkExecutor.prepareUnitOfWork(paymentOrderPostRequestFlux))
+                .assertNext(unitOfWork -> {
+                    Assertions.assertTrue(unitOfWork.getUnitOfOWorkId().startsWith("payment-orders-mixed-"));
+                    Assertions.assertEquals(UnitOfWork.State.NEW, unitOfWork.getState());
+                    Assertions.assertEquals(1, unitOfWork.getStreamTasks().size());
+                    Assertions.assertEquals(paymentOrderPostRequest.size(), unitOfWork.getStreamTasks().get(0).getData().size());
+                    Assertions.assertEquals(LocalDate.now().minusDays(30), fromDateCaptor.getValue());
+                })
+                .verifyComplete();
+    }
+
+    @Test
+    void test_prepareUnitOfWork_paymentOrderWithoutStartDateOffset() {
+        Flux<PaymentOrderPostRequest> paymentOrderPostRequestFlux = Flux.fromIterable(paymentOrderPostRequest);
+
+        PaymentOrderPostResponse paymentOrderPostResponse = new PaymentOrderPostResponse()
+                .id("po_post_resp_id")
+                .putAdditionsItem("key", "val");
+
+        lenient().when(paymentOrdersApi.postPaymentOrder(any()))
+                .thenReturn(Mono.just(paymentOrderPostResponse));
+
+        GetPaymentOrderResponse getPaymentOrderResponse = new GetPaymentOrderResponse()
+                .id("arrangementId_1")
+                .bankReferenceId("bankReferenceId_1");
+        PaymentOrderPostFilterResponse paymentOrderPostFilterResponse = new PaymentOrderPostFilterResponse()
+                .addPaymentOrdersItem(getPaymentOrderResponse)
+                .totalElements(new BigDecimal(1));
+        doReturn(Mono.just(paymentOrderPostFilterResponse)).when(paymentOrdersApi).postFilterPaymentOrders(any(), any(), any(), fromDateCaptor.capture(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any());
+
+        AccountArrangementItem accountArrangementItem = new AccountArrangementItem()
+                .id("arrangementId_1")
+                .externalArrangementId("externalArrangementId_1");
+        AccountArrangementItems accountArrangementItems = new AccountArrangementItems()
+                .addArrangementElementsItem(accountArrangementItem);
 
+        Mockito.when(arrangementsApi.postFilter(Mockito.any()))
+                .thenReturn(Mono.just(accountArrangementItems));
+
+        StepVerifier.create(paymentOrderUnitOfWorkExecutor.prepareUnitOfWork(paymentOrderPostRequestFlux))
+                .assertNext(unitOfWork -> {
+                    Assertions.assertTrue(unitOfWork.getUnitOfOWorkId().startsWith("payment-orders-mixed-"));
+                    Assertions.assertEquals(UnitOfWork.State.NEW, unitOfWork.getState());
+                    Assertions.assertEquals(1, unitOfWork.getStreamTasks().size());
+                    Assertions.assertEquals(paymentOrderPostRequest.size(), unitOfWork.getStreamTasks().get(0).getData().size());
+                    Assertions.assertNull(fromDateCaptor.getValue());
+                })
+                .verifyComplete();
+    }
     @Test
     void test_prepareunitofwork_blankuserid() {