Add Message table, related entities and saving

Author: RaevskiySV

server/src/main/java/com/exactpro/blockchain/api/client/ClientHandler.java

@@ -7,6 +7,7 @@
 import com.exactpro.blockchain.repository.*;
 import com.exactpro.iso20022.CustomerCreditTransfer;
 import com.exactpro.iso20022.XmlCodec;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import jakarta.xml.bind.JAXBException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -36,6 +37,7 @@ public class ClientHandler {
     private final ConversionRateRepository conversionRateRepository;
     private final CurrencyRepository currencyRepository;
     private final TransferRepository transferRepository;
+    private final MessageRepository messageRepository;
     private final XmlCodec xmlCodec;
     private final CustomerCreditTransferConverter converter;
     private final KafkaPublisher kafkaPublisher;
@@ -47,6 +49,7 @@ public ClientHandler(
         CurrencyRepository currencyRepository,
         ConversionRateRepository conversionRateRepository,
         TransferRepository transferRepository,
+        MessageRepository messageRepository,
         XmlCodec xmlCodec,
         CustomerCreditTransferConverter converter,
         KafkaPublisher kafkaPublisher,
@@ -56,6 +59,7 @@ public ClientHandler(
         this.clientRepository = clientRepository;
         this.conversionRateRepository = conversionRateRepository;
         this.transferRepository = transferRepository;
+        this.messageRepository = messageRepository;
         this.currencyRepository = currencyRepository;
         this.xmlCodec = xmlCodec;
         this.converter = converter;
@@ -76,7 +80,7 @@ public Mono<ServerResponse> transfer(ServerRequest request) {
             saveTransferRequest(client, transferRequest)
             .flatMap(transfer ->
                 debitAccount(account, transferRequest)
-                .then(Mono.defer(() -> publishPacs008(transfer)))
+                .then(Mono.defer(() -> saveMessageAndPublishPacs008(transfer)))
                 .then(Mono.defer(() -> transferRepository.save(transfer.withStatus(TransferStatus.COMPLETED))))
                 .then(Mono.defer(() -> {
                     if (currency.getAddress() == null) {
@@ -181,14 +185,34 @@ Mono<Void> handle(
         return transferRepository.save(converter.newTransfer(client, transferRequest));
     }
 
-    private @NonNull Mono<Void> publishPacs008(@NonNull Transfer transfer) {
+    private @NonNull Mono<Void> saveMessageAndPublishPacs008(@NonNull Transfer transfer) {
         CustomerCreditTransfer customerCreditTransfer = converter.toCustomerCreditTransfer(transfer);
-        String pacs008XmlString;
+
+        String transferJson;
         try {
-            pacs008XmlString = xmlCodec.encode(customerCreditTransfer);
-        } catch (JAXBException | TransformerException ex) {
-            return Mono.error(new Exception("Failed to encode transfer to pacs.008 XML", ex));
+            transferJson = transfer.toJson();
+        } catch (JsonProcessingException e) {
+            return Mono.error(new Exception("Failed to convert Transfer to JSON", e));
         }
-        return kafkaPublisher.publishMessage(transfer.getCreditorBic(), pacs008XmlString);
+
+        Message message = new Message(
+            customerCreditTransfer.getGroupHeader().getMessageId(),
+            transfer.getTransferId(),
+            transferJson);
+
+        return messageRepository.save(message)
+            .flatMap(savedMessage -> {
+                logger.info("Message saved with ID: {}", savedMessage.getMessageId());
+
+                String pacs008XmlString;
+                try {
+                    pacs008XmlString = xmlCodec.encode(customerCreditTransfer);
+                } catch (JAXBException | TransformerException ex) {
+                    return Mono.error(new Exception("Failed to encode transfer to pacs.008 XML", ex));
+                }
+
+                return kafkaPublisher.publishMessage(transfer.getCreditorBic(), pacs008XmlString);
+            })
+            .then();
     }
 }

server/src/main/java/com/exactpro/blockchain/entity/Message.java

@@ -0,0 +1,50 @@
+package com.exactpro.blockchain.entity;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
+
+@Table("Message")
+public class Message {
+    @Id
+    @Column("messageId")
+    private String messageId;
+
+    @Column("transferId")
+    private Integer transferId;
+
+    private String content;
+
+    public Message() {
+    }
+
+    public Message(String messageId, Integer transferId, String content) {
+        this.messageId = messageId;
+        this.transferId = transferId;
+        this.content = content;
+    }
+
+    public String getMessageId() {
+        return messageId;
+    }
+
+    public void setMessageId(String messageId) {
+        this.messageId = messageId;
+    }
+
+    public Integer getTransferId() {
+        return transferId;
+    }
+
+    public void setTransferId(Integer transferId) {
+        this.transferId = transferId;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public void setContent(String content) {
+        this.content = content;
+    }
+}

server/src/main/java/com/exactpro/blockchain/entity/Transfer.java

@@ -1,5 +1,7 @@
 package com.exactpro.blockchain.entity;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.Table;
@@ -12,6 +14,8 @@
 
 @Table("Transfer")
 public class Transfer {
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
     @Id
     @Column("transferId")
     private Integer transferId;
@@ -250,5 +254,9 @@ public Builder transferTimestamp(Instant transferTimestamp) {
             return this;
         }
     }
+
+    public String toJson() throws JsonProcessingException {
+        return objectMapper.writeValueAsString(this);
+    }
 }
 

server/src/main/java/com/exactpro/blockchain/repository/MessageRepository.java

@@ -0,0 +1,7 @@
+package com.exactpro.blockchain.repository;
+
+import com.exactpro.blockchain.entity.Message;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+public interface MessageRepository extends ReactiveCrudRepository<Message, String> {
+}

server/src/main/resources/db/migration/V1__Initial_schema.sql

@@ -54,3 +54,10 @@ CREATE TABLE BankETHAddress (
     bic VARCHAR(255) NOT NULL PRIMARY KEY,
     ethAddress VARCHAR(255) NOT NULL UNIQUE
 );
+
+CREATE TABLE Message (
+    messageId VARCHAR(255) PRIMARY KEY,
+    transferId INTEGER,
+    content TEXT,
+    FOREIGN KEY (transferId) REFERENCES Transfer(transferId)
+);