Add exchange rate

pom.xml

@@ -177,7 +177,7 @@
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>damsel</artifactId>
-            <version>1.692-70b59b9</version>
+            <version>1.695-25e75a5</version>
         </dependency>
         <dependency>
             <groupId>dev.vality</groupId>

src/main/java/dev/vality/daway/dao/invoicing/iface/PaymentExchangeContextDao.java

@@ -0,0 +1,17 @@
+package dev.vality.daway.dao.invoicing.iface;
+
+import dev.vality.dao.GenericDao;
+import dev.vality.daway.domain.tables.pojos.PaymentExchangeContext;
+import dev.vality.daway.exception.DaoException;
+import dev.vality.daway.model.InvoicingKey;
+
+import java.util.List;
+import java.util.Set;
+
+public interface PaymentExchangeContextDao extends GenericDao {
+
+    void saveBatch(List<PaymentExchangeContext> paymentExchangeContexts) throws DaoException;
+
+    void switchCurrent(Set<InvoicingKey> invoicingKeys) throws DaoException;
+
+}

src/main/java/dev/vality/daway/dao/invoicing/impl/PaymentExchangeContextDaoImpl.java

@@ -0,0 +1,82 @@
+package dev.vality.daway.dao.invoicing.impl;
+
+import dev.vality.dao.impl.AbstractGenericDao;
+import dev.vality.daway.dao.invoicing.iface.PaymentExchangeContextDao;
+import dev.vality.daway.domain.tables.pojos.PaymentExchangeContext;
+import dev.vality.daway.domain.tables.records.PaymentExchangeContextRecord;
+import dev.vality.daway.exception.DaoException;
+import dev.vality.daway.model.InvoicingKey;
+import dev.vality.mapper.RecordRowMapper;
+import org.jooq.Query;
+import org.jooq.impl.DSL;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static dev.vality.daway.domain.Tables.PAYMENT_EXCHANGE_CONTEXT;
+
+@Component
+public class PaymentExchangeContextDaoImpl extends AbstractGenericDao implements PaymentExchangeContextDao {
+
+    private final RowMapper<PaymentExchangeContext> paymentExchangeContextRowMapper;
+
+    public PaymentExchangeContextDaoImpl(DataSource dataSource) {
+        super(dataSource);
+        paymentExchangeContextRowMapper = new RecordRowMapper<>(PAYMENT_EXCHANGE_CONTEXT, PaymentExchangeContext.class);
+    }
+
+    @Override
+    public void saveBatch(List<PaymentExchangeContext> paymentExchangeContexts) throws DaoException {
+        List<Query> queries = paymentExchangeContexts.stream()
+                .map(paymentExchangeContext -> getDslContext().newRecord(
+                        PAYMENT_EXCHANGE_CONTEXT, paymentExchangeContext))
+                .map(this::prepareInsertQuery)
+                .toList();
+        batchExecute(queries);
+    }
+
+    private Query prepareInsertQuery(PaymentExchangeContextRecord record) {
+        return getDslContext().insertInto(PAYMENT_EXCHANGE_CONTEXT)
+                .set(record)
+                .onConflict(
+                        PAYMENT_EXCHANGE_CONTEXT.INVOICE_ID,
+                        PAYMENT_EXCHANGE_CONTEXT.PAYMENT_ID,
+                        PAYMENT_EXCHANGE_CONTEXT.SEQUENCE_ID,
+                        PAYMENT_EXCHANGE_CONTEXT.CHANGE_ID
+                )
+                .doNothing();
+    }
+
+    @Override
+    public void switchCurrent(Set<InvoicingKey> invoicingKeys) throws DaoException {
+        invoicingKeys.forEach(key -> {
+            setOldPaymentExchangeContextNotCurrent(key);
+            setLatestPaymentExchangeContextCurrent(key);
+        });
+    }
+
+    private void setOldPaymentExchangeContextNotCurrent(InvoicingKey key) {
+        execute(getDslContext().update(PAYMENT_EXCHANGE_CONTEXT)
+                .set(PAYMENT_EXCHANGE_CONTEXT.CURRENT, false)
+                .where(PAYMENT_EXCHANGE_CONTEXT.INVOICE_ID.eq(key.getInvoiceId())
+                        .and(PAYMENT_EXCHANGE_CONTEXT.PAYMENT_ID.eq(key.getPaymentId()))
+                        .and(PAYMENT_EXCHANGE_CONTEXT.CURRENT))
+        );
+    }
+
+    private void setLatestPaymentExchangeContextCurrent(InvoicingKey key) {
+        execute(getDslContext().update(PAYMENT_EXCHANGE_CONTEXT)
+                .set(PAYMENT_EXCHANGE_CONTEXT.CURRENT, true)
+                .where(PAYMENT_EXCHANGE_CONTEXT.ID.eq(
+                        DSL.select(DSL.max(PAYMENT_EXCHANGE_CONTEXT.ID))
+                                .from(PAYMENT_EXCHANGE_CONTEXT)
+                                .where(PAYMENT_EXCHANGE_CONTEXT.INVOICE_ID.eq(key.getInvoiceId())
+                                        .and(PAYMENT_EXCHANGE_CONTEXT.PAYMENT_ID.eq(key.getPaymentId())))
+                ))
+        );
+    }
+}

src/main/java/dev/vality/daway/factory/cash/flow/CashFlowFactory.java

@@ -37,6 +37,12 @@ public static List<CashFlow> build(List<FinalCashFlowPosting> cashFlowPostings,
             pcf.setAmount(cf.getVolume().getAmount());
             pcf.setCurrencyCode(cf.getVolume().getCurrency().getSymbolicCode());
             pcf.setDetails(cf.getDetails());
+            if (cf.isSetExchangeContext()) {
+                pcf.setExchangeSourceCurrencyCode(cf.getExchangeContext().getSourceCurrency());
+                pcf.setExchangeDestinationCurrencyCode(cf.getExchangeContext().getDestinationCurrency());
+                pcf.setExchangeRateRationalP(cf.getExchangeContext().getExchangeRate().getP());
+                pcf.setExchangeRateRationalQ(cf.getExchangeContext().getExchangeRate().getQ());
+            }
             return pcf;
         }).collect(Collectors.toList());
     }

src/main/java/dev/vality/daway/handler/wrapper/payment/PaymentExchangeContextWrapperHandler.java

@@ -0,0 +1,39 @@
+package dev.vality.daway.handler.wrapper.payment;
+
+import dev.vality.daway.dao.invoicing.iface.PaymentExchangeContextDao;
+import dev.vality.daway.domain.tables.pojos.PaymentExchangeContext;
+import dev.vality.daway.handler.wrapper.WrapperHandler;
+import dev.vality.daway.model.PaymentWrapper;
+import dev.vality.daway.util.PaymentWrapperUtil;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+@RequiredArgsConstructor
+@Component
+public class PaymentExchangeContextWrapperHandler implements WrapperHandler<PaymentWrapper> {
+
+    private final PaymentExchangeContextDao paymentExchangeContextDao;
+
+    @Override
+    public boolean accept(List<PaymentWrapper> wrappers) {
+        return wrappers.stream()
+                .map(PaymentWrapper::getPaymentExchangeContext)
+                .anyMatch(Objects::nonNull);
+    }
+
+    @Override
+    public void saveBatch(List<PaymentWrapper> wrappers) {
+        List<PaymentWrapper> processableWrappers = wrappers.stream()
+                .filter(paymentWrapper -> Objects.nonNull(paymentWrapper.getPaymentExchangeContext()))
+                .collect(Collectors.toList());
+        List<PaymentExchangeContext> paymentExchangeContexts = processableWrappers.stream()
+                .map(PaymentWrapper::getPaymentExchangeContext)
+                .collect(Collectors.toList());
+        paymentExchangeContextDao.saveBatch(paymentExchangeContexts);
+        paymentExchangeContextDao.switchCurrent(PaymentWrapperUtil.getInvoicingKeys(processableWrappers));
+    }
+}

src/main/java/dev/vality/daway/mapper/payment/InvoicePaymentExchangeContextChangedMapper.java

@@ -0,0 +1,73 @@
+package dev.vality.daway.mapper.payment;
+
+import dev.vality.damsel.base.Rational;
+import dev.vality.damsel.domain.ExchangeContext;
+import dev.vality.damsel.payment_processing.InvoiceChange;
+import dev.vality.damsel.payment_processing.InvoicePaymentChange;
+import dev.vality.damsel.payment_processing.InvoicePaymentExchangeContextChanged;
+import dev.vality.daway.domain.tables.pojos.PaymentExchangeContext;
+import dev.vality.daway.mapper.Mapper;
+import dev.vality.daway.model.InvoicingKey;
+import dev.vality.daway.model.PaymentWrapper;
+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.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class InvoicePaymentExchangeContextChangedMapper implements Mapper<PaymentWrapper> {
+
+    private Filter filter = new PathConditionFilter(new PathConditionRule(
+            "invoice_payment_change.payload.invoice_payment_exchange_context_changed",
+            new IsNullCondition().not()));
+
+    @Override
+    public PaymentWrapper map(InvoiceChange change, MachineEvent event, Integer changeId) {
+        InvoicePaymentChange invoicePaymentChange = change.getInvoicePaymentChange();
+        String invoiceId = event.getSourceId();
+        String paymentId = invoicePaymentChange.getId();
+        long sequenceId = event.getEventId();
+        log.info("Start mapping payment exchange context change, sequenceId='{}', changeId='{}', invoiceId='{}', "
+                        + "paymentId='{}'",
+                sequenceId, changeId, invoiceId, paymentId);
+
+        InvoicePaymentExchangeContextChanged exchangeContextChanged =
+                invoicePaymentChange.getPayload().getInvoicePaymentExchangeContextChanged();
+        ExchangeContext exchangeContext = exchangeContextChanged.getExchangeContext();
+        Rational exchangeRate = exchangeContext.getExchangeRate();
+
+        PaymentExchangeContext paymentExchangeContext = new PaymentExchangeContext();
+        paymentExchangeContext.setWtime(null);
+        paymentExchangeContext.setId(null);
+        paymentExchangeContext.setChangeId(changeId);
+        paymentExchangeContext.setSequenceId(sequenceId);
+        paymentExchangeContext.setPaymentId(paymentId);
+        paymentExchangeContext.setInvoiceId(invoiceId);
+        paymentExchangeContext.setCurrent(true);
+        paymentExchangeContext.setEventCreatedAt(TypeUtil.stringToLocalDateTime(event.getCreatedAt()));
+        paymentExchangeContext.setSourceCurrencyCode(exchangeContext.getSourceCurrency());
+        paymentExchangeContext.setDestinationCurrencyCode(exchangeContext.getDestinationCurrency());
+        paymentExchangeContext.setExchangeRateRationalP(exchangeRate.getP());
+        paymentExchangeContext.setExchangeRateRationalQ(exchangeRate.getQ());
+
+        PaymentWrapper paymentWrapper = new PaymentWrapper();
+        paymentWrapper.setKey(InvoicingKey.buildKey(invoiceId, paymentId));
+        paymentWrapper.setPaymentExchangeContext(paymentExchangeContext);
+        log.info("Payment exchange context has been mapped, sequenceId='{}', changeId='{}', invoiceId='{}', "
+                        + "paymentId='{}'",
+                sequenceId, changeId, invoiceId, paymentId);
+        return paymentWrapper;
+    }
+
+    @Override
+    public Filter<InvoiceChange> getFilter() {
+        return filter;
+    }
+}

src/main/java/dev/vality/daway/model/PaymentWrapper.java

@@ -18,6 +18,7 @@ public class PaymentWrapper {
     private PaymentRiskData paymentRiskData;
     private PaymentFee paymentFee;
     private PaymentRoute paymentRoute;
+    private PaymentExchangeContext paymentExchangeContext;
     private CashFlowWrapper cashFlowWrapper;
     private PaymentCashChange paymentCashChange;
     private InvoicingKey key;

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

@@ -0,0 +1,24 @@
+CREATE TABLE IF NOT EXISTS dw.payment_exchange_context
+(
+    id                              bigserial                   NOT NULL,
+    event_created_at                timestamp without time zone NOT NULL,
+    invoice_id                      character varying           NOT NULL,
+    payment_id                      character varying           NOT NULL,
+    source_currency_code            character varying           NOT NULL,
+    destination_currency_code       character varying           NOT NULL,
+    exchange_rate_rational_p        bigint                      NOT NULL,
+    exchange_rate_rational_q        bigint                      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,
+    change_id                       integer,
+
+    CONSTRAINT payment_exchange_context_pkey PRIMARY KEY (id),
+    CONSTRAINT payment_exchange_context_uniq UNIQUE (invoice_id, payment_id, sequence_id, change_id)
+);
+
+ALTER TABLE dw.cash_flow
+    ADD COLUMN IF NOT EXISTS exchange_source_currency_code character varying,
+    ADD COLUMN IF NOT EXISTS exchange_destination_currency_code character varying,
+    ADD COLUMN IF NOT EXISTS exchange_rate_rational_p bigint,
+    ADD COLUMN IF NOT EXISTS exchange_rate_rational_q bigint;

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

@@ -1,5 +1,6 @@
 package dev.vality.daway;
 
+import dev.vality.damsel.base.Rational;
 import dev.vality.damsel.domain.*;
 import dev.vality.damsel.domain.CashFlowAccount;
 import dev.vality.damsel.domain.InvoicePaymentChargeback;
@@ -14,7 +15,10 @@
 import dev.vality.daway.domain.enums.*;
 import dev.vality.daway.domain.tables.pojos.Chargeback;
 import dev.vality.daway.domain.tables.pojos.FistfulCashFlow;
+import dev.vality.daway.domain.tables.pojos.PaymentExchangeContext;
 import dev.vality.daway.domain.tables.pojos.WithdrawalAdjustment;
+import dev.vality.daway.model.InvoicingKey;
+import dev.vality.daway.model.PaymentWrapper;
 import dev.vality.fistful.account.Account;
 import dev.vality.fistful.base.Realm;
 import dev.vality.fistful.cashflow.FinalCashFlow;
@@ -234,6 +238,70 @@ public static String randomString() {
         return UUID.randomUUID().toString();
     }
 
+    public static ExchangeContext createExchangeContext() {
+        return createExchangeContext("RUB", "USD", 60797502L, 1000000L);
+    }
+
+    public static ExchangeContext createExchangeContext(String sourceCurrencyCode,
+                                                        String destinationCurrencyCode,
+                                                        Long rationalP,
+                                                        Long rationalQ) {
+        return new ExchangeContext(sourceCurrencyCode, destinationCurrencyCode, new Rational(rationalP, rationalQ));
+    }
+
+    public static FinalCashFlowPosting createPaymentCashFlowPosting() {
+        return new FinalCashFlowPosting()
+                .setSource(new FinalCashFlowAccount()
+                        .setAccountId(1)
+                        .setAccountType(CashFlowAccount.merchant(MerchantCashFlowAccount.settlement)))
+                .setDestination(new FinalCashFlowAccount()
+                        .setAccountId(2)
+                        .setAccountType(CashFlowAccount.system(SystemCashFlowAccount.settlement)))
+                .setVolume(new Cash(1000L, new CurrencyRef("RUB")));
+    }
+
+    public static FinalCashFlowPosting createPaymentCashFlowPostingWithExchangeContext() {
+        return createPaymentCashFlowPosting()
+                .setExchangeContext(createExchangeContext());
+    }
+
+    public static MachineEvent createInvoiceEvent(String invoiceId, Long sequenceId, LocalDateTime createdAt) {
+        return new MachineEvent()
+                .setSourceId(invoiceId)
+                .setEventId(sequenceId)
+                .setCreatedAt(TypeUtil.temporalToString(createdAt));
+    }
+
+    public static InvoiceChange createInvoicePaymentExchangeContextChanged(String paymentId) {
+        return InvoiceChange.invoice_payment_change(new InvoicePaymentChange()
+                .setId(paymentId)
+                .setPayload(InvoicePaymentChangePayload.invoice_payment_exchange_context_changed(
+                        new InvoicePaymentExchangeContextChanged(createExchangeContext()))));
+    }
+
+    public static PaymentWrapper createPaymentExchangeContextWrapper(String invoiceId,
+                                                                     String paymentId,
+                                                                     Long sequenceId,
+                                                                     Integer changeId,
+                                                                     Long exchangeRateP) {
+        PaymentExchangeContext paymentExchangeContext = new PaymentExchangeContext();
+        paymentExchangeContext.setEventCreatedAt(LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS));
+        paymentExchangeContext.setInvoiceId(invoiceId);
+        paymentExchangeContext.setPaymentId(paymentId);
+        paymentExchangeContext.setSourceCurrencyCode("RUB");
+        paymentExchangeContext.setDestinationCurrencyCode("USD");
+        paymentExchangeContext.setExchangeRateRationalP(exchangeRateP);
+        paymentExchangeContext.setExchangeRateRationalQ(1000000L);
+        paymentExchangeContext.setSequenceId(sequenceId);
+        paymentExchangeContext.setChangeId(changeId);
+        paymentExchangeContext.setCurrent(true);
+
+        PaymentWrapper paymentWrapper = new PaymentWrapper();
+        paymentWrapper.setKey(InvoicingKey.buildKey(invoiceId, paymentId));
+        paymentWrapper.setPaymentExchangeContext(paymentExchangeContext);
+        return paymentWrapper;
+    }
+
     public static CountryObject buildCountryObject() {
         Country country = new Country();
         country.setName(randomString());