Fix cashflow adjustment handling

src/main/java/dev/vality/daway/handler/event/stock/impl/withdrawal/WithdrawalAdjustmentCreatedHandler.java

@@ -59,6 +59,8 @@ public void handle(TimestampedChange timestampedChange, MachineEvent event) {
             DataRevisionChangePlan newDomainRevision = changesPlan.getNewDomainRevision();
             withdrawalAdjustment.setType(WithdrawalAdjustmentType.domain_revision);
             withdrawalAdjustment.setDomainRevision(newDomainRevision.getNewDomainRevision());
+        } else if (changesPlan.isSetNewCashFlow()) {
+            withdrawalAdjustment.setType(WithdrawalAdjustmentType.cash_flow);
         }
         if (changesPlan.isSetNewCashFlow()) {
             CashFlowChangePlan cashFlow = changesPlan.getNewCashFlow();

src/main/resources/db/migration/V55__add_withdrawal_adjustment_cash_flow_type.sql

@@ -0,0 +1 @@
+ALTER TYPE dw.withdrawal_adjustment_type ADD VALUE IF NOT EXISTS 'cash_flow';

src/test/java/dev/vality/daway/TestData.java

@@ -418,6 +418,31 @@ public static TimestampedChange createWithdrawalAdjustmentCreatedDomainRevisionC
         return timestampedChange;
     }
 
+    public static TimestampedChange createWithdrawalAdjustmentCreatedCashFlowChange(String id) {
+        Adjustment adjustment = new Adjustment();
+        adjustment.setId(id);
+        adjustment.setOperationTimestamp(OCCURED_AT);
+        adjustment.setCreatedAt(OCCURED_AT);
+        adjustment.setStatus(Status.pending(new Pending()));
+        adjustment.setChangesPlan(
+                new ChangesPlan()
+                        .setNewCashFlow(new CashFlowChangePlan()
+                                .setOldCashFlowInverted(new FinalCashFlow().setPostings(getFinalCashFlowPostings()))
+                                .setNewCashFlow(new FinalCashFlow().setPostings(getFinalCashFlowPostings())))
+        );
+        var payload = new dev.vality.fistful.withdrawal.adjustment.Change();
+        payload.setCreated(new CreatedChange().setAdjustment(adjustment));
+        AdjustmentChange adjustmentChange = new AdjustmentChange();
+        adjustmentChange.setId("id");
+        adjustmentChange.setPayload(payload);
+        Change change = new Change();
+        change.setAdjustment(adjustmentChange);
+        TimestampedChange timestampedChange = new TimestampedChange();
+        timestampedChange.setOccuredAt(OCCURED_AT);
+        timestampedChange.setChange(change);
+        return timestampedChange;
+    }
+
     public static TimestampedChange createWithdrawalAdjustmentStatusChange(String id) {
         var payload = new dev.vality.fistful.withdrawal.adjustment.Change();
         payload.setStatusChanged(new StatusChange(Status.succeeded(new Succeeded())));

src/test/java/dev/vality/daway/dao/DaoTests.java

@@ -142,7 +142,7 @@ void cashFlowDaoTest() {
         List<CashFlow> cashFlowList = RandomBeans.randomListOf(100, CashFlow.class);
         cashFlowList.forEach(cf -> {
             cf.setObjId(cashFlowLink);
-            cf.setAmount((long) new Random().nextInt(100));
+            cf.setAmount(42L);
             cf.setObjType(PaymentChangeType.payment);
             cf.setAdjFlowType(null);
             cf.setSourceAccountTypeValue("settlement");
@@ -212,17 +212,16 @@ void invoiceCartDaoTest() {
 
     @Test
     void paymentRecurrentInfoDaoTest() {
-        Random random = new Random();
         jdbcTemplate.execute("truncate table dw.payment_recurrent_info cascade");
-        List<PaymentRecurrentInfo> list = RandomBeans.randomListOf(random.nextLong(), 2, PaymentRecurrentInfo.class);
+        List<PaymentRecurrentInfo> list = RandomBeans.randomListOf(1L, 2, PaymentRecurrentInfo.class);
         list.forEach(statusInfo -> statusInfo.setCurrent(true));
         paymentRecurrentInfoDao.saveBatch(list);
         PaymentRecurrentInfo first = list.get(0);
         assertEquals(first, paymentRecurrentInfoDao.get(first.getInvoiceId(), first.getPaymentId()));
         PaymentRecurrentInfo second = list.get(1);
         assertEquals(second, paymentRecurrentInfoDao.get(second.getInvoiceId(), second.getPaymentId()));
 
-        PaymentRecurrentInfo third = RandomBeans.random(random.nextLong(), PaymentRecurrentInfo.class);
+        PaymentRecurrentInfo third = RandomBeans.random(2L, PaymentRecurrentInfo.class);
         third.setId(first.getId() + 1);
         third.setCurrent(false);
         third.setInvoiceId(first.getInvoiceId());

src/test/java/dev/vality/daway/dao/partition/DaoTests.java

@@ -17,7 +17,6 @@
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Random;
 import java.util.Set;
 
 import static dev.vality.daway.domain.tables.Invoice.INVOICE;
@@ -111,11 +110,10 @@ void invoiceStatusInfoDaoTest() {
 
     @Test
     void paymentDaoTest() {
-        Random random = new Random();
-        Payment first = RandomBeans.random(random.nextLong(100), Payment.class);
+        Payment first = RandomBeans.random(1L, Payment.class);
         first.setId(1L);
         first.setEventCreatedAt(LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS));
-        Payment second = RandomBeans.random(random.nextLong(100),Payment.class);
+        Payment second = RandomBeans.random(2L, Payment.class);
         second.setId(2L);
         second.setEventCreatedAt(LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS));
         paymentDao.saveBatch(Arrays.asList(first, second));
@@ -147,10 +145,9 @@ void paymentStatusInfoDaoTest() {
 
     @Test
     void paymentPayerInfoDaoTest() {
-        Random random = new Random();
-        PaymentPayerInfo first = RandomBeans.random(random.nextLong(100), PaymentPayerInfo.class);
+        PaymentPayerInfo first = RandomBeans.random(1L, PaymentPayerInfo.class);
         first.setId(1L);
-        PaymentPayerInfo second = RandomBeans.random(random.nextLong(100), PaymentPayerInfo.class);
+        PaymentPayerInfo second = RandomBeans.random(2L, PaymentPayerInfo.class);
         second.setId(2L);
         paymentPayerInfoDao.saveBatch(Arrays.asList(first, second));
         assertEquals(first, paymentPayerInfoDao.get(first.getInvoiceId(), first.getPaymentId()));

src/test/java/dev/vality/daway/handler/event/stock/impl/withdrawal/WithdrawalAdjustmentCreatedHandlerTest.java

@@ -3,6 +3,7 @@
 import dev.vality.daway.TestData;
 import dev.vality.daway.config.PostgresqlJooqSpringBootITest;
 import dev.vality.daway.dao.withdrawal.impl.WithdrawalAdjustmentDaoImpl;
+import dev.vality.daway.domain.enums.WithdrawalAdjustmentType;
 import dev.vality.daway.domain.tables.records.WithdrawalAdjustmentRecord;
 import dev.vality.daway.factory.machine.event.WithdrawalAdjustmentMachineEventCopyFactoryImpl;
 import dev.vality.fistful.withdrawal.TimestampedChange;
@@ -13,6 +14,7 @@
 import org.springframework.test.context.ContextConfiguration;
 
 import static dev.vality.daway.domain.tables.WithdrawalAdjustment.WITHDRAWAL_ADJUSTMENT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
@@ -55,4 +57,21 @@ void handleDomainRevisionChange() {
         assertNull(record.getWithdrawalStatus());
         assertNotNull(record.getDomainRevision());
     }
+
+    @Test
+    void handleCashFlowChange() {
+        TimestampedChange timestampedChange =
+                TestData.createWithdrawalAdjustmentCreatedCashFlowChange("adjustmentId");
+
+        handler.handle(timestampedChange, TestData.createMachineEvent(timestampedChange));
+
+        WithdrawalAdjustmentRecord record = dslContext.fetchAny(WITHDRAWAL_ADJUSTMENT);
+        assertNotNull(record);
+        assertEquals(WithdrawalAdjustmentType.cash_flow, record.getType());
+        assertNotNull(record.getAmount());
+        assertNotNull(record.getFee());
+        assertNotNull(record.getProviderFee());
+        assertNull(record.getWithdrawalStatus());
+        assertNull(record.getDomainRevision());
+    }
 }

src/test/java/dev/vality/daway/kafka/WithdrawalKafkaListenerAdjustmentTest.java

@@ -77,7 +77,7 @@ void listenWithdrawalAdjustmentStatusChange() {
 
         Mockito.verify(withdrawalAdjustmentDao, Mockito.timeout(TimeUnit.MINUTES.toMillis(3)).times(1))
                 .getByIds(anyString(), anyString());
-        Mockito.verify(withdrawalAdjustmentDao, Mockito.times(1))
+        Mockito.verify(withdrawalAdjustmentDao, Mockito.timeout(TimeUnit.MINUTES.toMillis(3)).times(1))
                 .save(any());
     }
 }

src/test/java/dev/vality/daway/service/InvoiceWrapperServiceTest.java

@@ -15,7 +15,6 @@
 import org.springframework.test.context.jdbc.Sql;
 
 import java.util.List;
-import java.util.Random;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -60,12 +59,11 @@ void duplicationTest() {
     }
 
     private List<InvoiceWrapper> prepareInvoiceWrappers() {
-        Random random = new Random();
         List<InvoiceWrapper> invoiceWrappers = IntStream.range(1, 5)
                 .mapToObj(x -> new InvoiceWrapper(
-                        RandomBeans.random(random.nextLong(), Invoice.class, "id"),
-                        RandomBeans.random(random.nextLong(), InvoiceStatusInfo.class, "id", "invoiceId"),
-                        RandomBeans.randomListOf(random.nextLong(), 3, InvoiceCart.class, "id", "invoiceId")))
+                        RandomBeans.random((long) x * 3, Invoice.class, "id"),
+                        RandomBeans.random((long) x * 3 + 1, InvoiceStatusInfo.class, "id", "invoiceId"),
+                        RandomBeans.randomListOf((long) x * 3 + 2, 3, InvoiceCart.class, "id", "invoiceId")))
                 .collect(Collectors.toList());
 
         invoiceWrappers.forEach(iw -> {

src/test/java/dev/vality/daway/service/PaymentWrapperServiceTest.java

@@ -22,7 +22,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static dev.vality.daway.utils.JdbcUtil.countEntities;
 import static dev.vality.daway.utils.JdbcUtil.countPaymentEntity;
@@ -89,7 +89,8 @@ void duplicationTest() {
     }
 
     private List<PaymentWrapper> preparePaymentWrappers() {
-        Random random = new Random();
+        AtomicLong cashFlowSeed = new AtomicLong();
+        AtomicLong cashFlowId = new AtomicLong();
         List<PaymentWrapper> paymentWrappers = RandomBeans.randomListOf(2, PaymentWrapper.class);
         paymentWrappers.stream()
                 .map(PaymentWrapper::getPayment)
@@ -98,9 +99,12 @@ private List<PaymentWrapper> preparePaymentWrappers() {
             pw.setPaymentExchangeContext(null);
             pw.setCashFlowWrapper(new CashFlowWrapper(
                     RandomBeans.random(CashFlowLink.class),
-                    RandomBeans.randomListOf(random.nextLong(100), 3, CashFlow.class)
+                    RandomBeans.randomListOf(cashFlowSeed.getAndIncrement(), 3, CashFlow.class)
             ));
-            pw.getCashFlowWrapper().getCashFlows().forEach(cf -> cf.setObjType(PaymentChangeType.payment));
+            pw.getCashFlowWrapper().getCashFlows().forEach(cf -> {
+                cf.setId(cashFlowId.incrementAndGet());
+                cf.setObjType(PaymentChangeType.payment);
+            });
             PaymentWrapperTestUtil.setCurrent(pw, true);
         });
         PaymentWrapperTestUtil.setInvoiceIdAndPaymentId(paymentWrappers.get(0), invoiceIdFirst, paymentIdFirst);