add withdrawal cash changed support

pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>dev.vality</groupId>
         <artifactId>service-parent-pom</artifactId>
-        <version>3.1.9</version>
+        <version>3.1.11</version>
     </parent>
 
     <artifactId>daway</artifactId>
@@ -177,12 +177,12 @@
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>damsel</artifactId>
-            <version>1.696-b07c077</version>
+            <version>1.698-99c91c9</version>
         </dependency>
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>fistful-proto</artifactId>
-            <version>1.188-f7ce08e</version>
+            <version>1.191-94b75ce</version>
         </dependency>
         <dependency>
             <groupId>dev.vality</groupId>

src/main/java/dev/vality/daway/dao/withdrawal/iface/WithdrawalCashChangeDao.java

@@ -0,0 +1,13 @@
+package dev.vality.daway.dao.withdrawal.iface;
+
+import dev.vality.dao.GenericDao;
+import dev.vality.daway.domain.tables.pojos.WithdrawalCashChange;
+import dev.vality.daway.exception.DaoException;
+
+public interface WithdrawalCashChangeDao extends GenericDao {
+
+    void save(WithdrawalCashChange withdrawalCashChange) throws DaoException;
+
+    void switchCurrent(String withdrawalId) throws DaoException;
+
+}

src/main/java/dev/vality/daway/dao/withdrawal/impl/WithdrawalCashChangeDaoImpl.java

@@ -0,0 +1,63 @@
+package dev.vality.daway.dao.withdrawal.impl;
+
+import dev.vality.dao.impl.AbstractGenericDao;
+import dev.vality.daway.dao.withdrawal.iface.WithdrawalCashChangeDao;
+import dev.vality.daway.domain.tables.pojos.WithdrawalCashChange;
+import dev.vality.daway.domain.tables.records.WithdrawalCashChangeRecord;
+import dev.vality.daway.exception.DaoException;
+import org.jooq.Query;
+import org.jooq.impl.DSL;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+
+import static dev.vality.daway.domain.Tables.WITHDRAWAL_CASH_CHANGE;
+
+@Component
+public class WithdrawalCashChangeDaoImpl extends AbstractGenericDao implements WithdrawalCashChangeDao {
+
+    public WithdrawalCashChangeDaoImpl(DataSource dataSource) {
+        super(dataSource);
+    }
+
+    @Override
+    public void save(WithdrawalCashChange withdrawalCashChange) throws DaoException {
+        WithdrawalCashChangeRecord record = getDslContext().newRecord(WITHDRAWAL_CASH_CHANGE, withdrawalCashChange);
+        execute(prepareInsertQuery(record));
+    }
+
+    private Query prepareInsertQuery(WithdrawalCashChangeRecord record) {
+        return getDslContext().insertInto(WITHDRAWAL_CASH_CHANGE)
+                .set(record)
+                .onConflict(
+                        WITHDRAWAL_CASH_CHANGE.WITHDRAWAL_ID,
+                        WITHDRAWAL_CASH_CHANGE.SEQUENCE_ID
+                )
+                .doNothing();
+    }
+
+    @Override
+    public void switchCurrent(String withdrawalId) throws DaoException {
+        setOldWithdrawalCashChangeNotCurrent(withdrawalId);
+        setLatestWithdrawalCashChangeCurrent(withdrawalId);
+    }
+
+    private void setOldWithdrawalCashChangeNotCurrent(String withdrawalId) {
+        execute(getDslContext().update(WITHDRAWAL_CASH_CHANGE)
+                .set(WITHDRAWAL_CASH_CHANGE.CURRENT, false)
+                .where(WITHDRAWAL_CASH_CHANGE.WITHDRAWAL_ID.eq(withdrawalId)
+                        .and(WITHDRAWAL_CASH_CHANGE.CURRENT))
+        );
+    }
+
+    private void setLatestWithdrawalCashChangeCurrent(String withdrawalId) {
+        execute(getDslContext().update(WITHDRAWAL_CASH_CHANGE)
+                .set(WITHDRAWAL_CASH_CHANGE.CURRENT, true)
+                .where(WITHDRAWAL_CASH_CHANGE.ID.eq(
+                        DSL.select(DSL.max(WITHDRAWAL_CASH_CHANGE.ID))
+                                .from(WITHDRAWAL_CASH_CHANGE)
+                                .where(WITHDRAWAL_CASH_CHANGE.WITHDRAWAL_ID.eq(withdrawalId))
+                ))
+        );
+    }
+}

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

@@ -0,0 +1,62 @@
+package dev.vality.daway.handler.event.stock.impl.withdrawal;
+
+import dev.vality.daway.dao.withdrawal.iface.WithdrawalCashChangeDao;
+import dev.vality.daway.domain.tables.pojos.WithdrawalCashChange;
+import dev.vality.fistful.base.Cash;
+import dev.vality.fistful.withdrawal.BodyChange;
+import dev.vality.fistful.withdrawal.Change;
+import dev.vality.fistful.withdrawal.TimestampedChange;
+import dev.vality.geck.common.util.TypeUtil;
+import dev.vality.geck.filter.Filter;
+import dev.vality.geck.filter.PathConditionFilter;
+import dev.vality.geck.filter.condition.IsNullCondition;
+import dev.vality.geck.filter.rule.PathConditionRule;
+import dev.vality.machinegun.eventsink.MachineEvent;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class WithdrawalBodyChangedHandler implements WithdrawalHandler {
+
+    private final WithdrawalCashChangeDao withdrawalCashChangeDao;
+
+    @Getter
+    private final Filter filter = new PathConditionFilter(
+            new PathConditionRule("change.body_changed", new IsNullCondition().not()));
+
+    @Override
+    @Transactional(propagation = Propagation.REQUIRED)
+    public void handle(TimestampedChange timestampedChange, MachineEvent event) {
+        Change change = timestampedChange.getChange();
+        long sequenceId = event.getEventId();
+        String withdrawalId = event.getSourceId();
+        log.info("Start withdrawal body changed handling, sequenceId={}, withdrawalId={}", sequenceId, withdrawalId);
+
+        BodyChange bodyChange = change.getBodyChanged();
+        WithdrawalCashChange cashChange = new WithdrawalCashChange();
+        cashChange.setWtime(null);
+        cashChange.setId(null);
+        cashChange.setSequenceId(sequenceId);
+        cashChange.setWithdrawalId(withdrawalId);
+        cashChange.setCurrent(true);
+        cashChange.setEventCreatedAt(TypeUtil.stringToLocalDateTime(event.getCreatedAt()));
+
+        Cash newCash = bodyChange.getNewBody();
+        cashChange.setNewAmount(newCash.getAmount());
+        cashChange.setNewCurrencyCode(newCash.getCurrency().getSymbolicCode());
+        Cash oldCash = bodyChange.getOldBody();
+        cashChange.setOldAmount(oldCash.getAmount());
+        cashChange.setOldCurrencyCode(oldCash.getCurrency().getSymbolicCode());
+
+        withdrawalCashChangeDao.save(cashChange);
+        withdrawalCashChangeDao.switchCurrent(withdrawalId);
+        log.info("Withdrawal body change has been saved, sequenceId={}, withdrawalId={}", sequenceId, withdrawalId);
+    }
+
+}

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

@@ -0,0 +1,18 @@
+CREATE TABLE dw.withdrawal_cash_change
+(
+    id                bigserial                   NOT NULL,
+    event_created_at  timestamp without time zone NOT NULL,
+    withdrawal_id     character varying           NOT NULL,
+
+    new_amount        bigint                      NOT NULL,
+    new_currency_code character varying           NOT NULL,
+    old_amount        bigint                      NOT NULL,
+    old_currency_code character varying           NOT NULL,
+
+    current           BOOLEAN                     NOT NULL DEFAULT false,
+    wtime             timestamp without time zone NOT NULL DEFAULT (now() AT TIME ZONE 'utc'::text),
+    sequence_id       bigint,
+
+    CONSTRAINT withdrawal_cash_change_pkey PRIMARY KEY (id),
+    CONSTRAINT withdrawal_cash_change_uniq UNIQUE (withdrawal_id, sequence_id)
+);

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

@@ -625,6 +625,24 @@ public static TimestampedChange createWithdrawalCreatedChange(String id) {
         return timestampedChange;
     }
 
+    public static TimestampedChange createWithdrawalBodyChangedChange() {
+        BodyChange bodyChange = new BodyChange()
+                .setOldBody(new dev.vality.fistful.base.Cash()
+                        .setAmount(100L)
+                        .setCurrency(new dev.vality.fistful.base.CurrencyRef()
+                                .setSymbolicCode("RUB")))
+                .setNewBody(new dev.vality.fistful.base.Cash()
+                        .setAmount(200L)
+                        .setCurrency(new dev.vality.fistful.base.CurrencyRef()
+                                .setSymbolicCode("USD")));
+        Change change = new Change();
+        change.setBodyChanged(bodyChange);
+        TimestampedChange timestampedChange = new TimestampedChange();
+        timestampedChange.setOccuredAt(OCCURED_AT);
+        timestampedChange.setChange(change);
+        return timestampedChange;
+    }
+
     public static MachineEvent createInvoice(InvoicePaymentChangePayload invoicePaymentChangePayload) {
         PaymentEventPayloadSerializer paymentEventPayloadSerializer = new PaymentEventPayloadSerializer();
         MachineEvent message = new MachineEvent();

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

@@ -0,0 +1,52 @@
+package dev.vality.daway.handler.event.stock.impl.withdrawal;
+
+import dev.vality.daway.TestData;
+import dev.vality.daway.config.PostgresqlJooqSpringBootITest;
+import dev.vality.daway.dao.withdrawal.impl.WithdrawalCashChangeDaoImpl;
+import dev.vality.daway.domain.tables.records.WithdrawalCashChangeRecord;
+import dev.vality.fistful.withdrawal.TimestampedChange;
+import dev.vality.machinegun.eventsink.MachineEvent;
+import org.jooq.DSLContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+
+import static dev.vality.daway.domain.tables.WithdrawalCashChange.WITHDRAWAL_CASH_CHANGE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@PostgresqlJooqSpringBootITest
+@ContextConfiguration(classes = {WithdrawalCashChangeDaoImpl.class, WithdrawalBodyChangedHandler.class})
+class WithdrawalBodyChangedHandlerTest {
+
+    @Autowired
+    private WithdrawalBodyChangedHandler handler;
+
+    @Autowired
+    private DSLContext dslContext;
+
+    @BeforeEach
+    void setUp() {
+        dslContext.deleteFrom(WITHDRAWAL_CASH_CHANGE).execute();
+    }
+
+    @Test
+    void handle() {
+        TimestampedChange timestampedChange = TestData.createWithdrawalBodyChangedChange();
+        MachineEvent event = TestData.createMachineEvent(timestampedChange);
+
+        handler.handle(timestampedChange, event);
+
+        WithdrawalCashChangeRecord record = dslContext.fetchAny(WITHDRAWAL_CASH_CHANGE);
+        assertNotNull(record);
+        assertEquals(event.getSourceId(), record.getWithdrawalId());
+        assertEquals(event.getEventId(), record.getSequenceId());
+        assertEquals(100L, record.getOldAmount());
+        assertEquals("RUB", record.getOldCurrencyCode());
+        assertEquals(200L, record.getNewAmount());
+        assertEquals("USD", record.getNewCurrencyCode());
+        assertTrue(record.getCurrent());
+    }
+}