Persist and republish mailbox messages

core/src/main/java/bisq/core/alert/PrivateNotificationManager.java

@@ -50,6 +50,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import static org.bitcoinj.core.Utils.HEX;
 
 public class PrivateNotificationManager {
@@ -64,7 +66,8 @@ public class PrivateNotificationManager {
     private final String pubKeyAsHex;
 
     private ECKey privateNotificationSigningKey;
-    private DecryptedMessageWithPubKey decryptedMessageWithPubKey;
+    @Nullable
+    private PrivateNotificationMessage privateNotificationMessage;
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -91,10 +94,9 @@ public PrivateNotificationManager(P2PService p2PService,
     }
 
     private void handleMessage(DecryptedMessageWithPubKey decryptedMessageWithPubKey, NodeAddress senderNodeAddress) {
-        this.decryptedMessageWithPubKey = decryptedMessageWithPubKey;
         NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
         if (networkEnvelope instanceof PrivateNotificationMessage) {
-            PrivateNotificationMessage privateNotificationMessage = (PrivateNotificationMessage) networkEnvelope;
+            privateNotificationMessage = (PrivateNotificationMessage) networkEnvelope;
             log.info("Received PrivateNotificationMessage from {} with uid={}",
                     senderNodeAddress, privateNotificationMessage.getUid());
             if (privateNotificationMessage.getSenderNodeAddress().equals(senderNodeAddress)) {
@@ -116,8 +118,11 @@ public ReadOnlyObjectProperty<PrivateNotificationPayload> privateNotificationPro
         return privateNotificationMessageProperty;
     }
 
-    public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPayload privateNotification, PubKeyRing pubKeyRing, NodeAddress peersNodeAddress,
-                                                              String privKeyString, SendMailboxMessageListener sendMailboxMessageListener) {
+    public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPayload privateNotification,
+                                                              PubKeyRing pubKeyRing,
+                                                              NodeAddress peersNodeAddress,
+                                                              String privKeyString,
+                                                              SendMailboxMessageListener sendMailboxMessageListener) {
         boolean isKeyValid = isKeyValid(privKeyString);
         if (isKeyValid) {
             signAndAddSignatureToPrivateNotificationMessage(privateNotification);
@@ -137,7 +142,9 @@ public boolean sendPrivateNotificationMessageIfKeyIsValid(PrivateNotificationPay
     }
 
     public void removePrivateNotification() {
-        mailboxMessageService.removeMailboxMsg(decryptedMessageWithPubKey);
+        if (privateNotificationMessage != null) {
+            mailboxMessageService.removeMailboxMsg(privateNotificationMessage);
+        }
     }
 
     private boolean isKeyValid(String privKeyString) {

core/src/main/java/bisq/core/support/SupportManager.java

@@ -84,7 +84,7 @@ public SupportManager(P2PService p2PService, WalletsSetup walletsSetup) {
     // Abstract methods
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    protected abstract void dispatchMessage(SupportMessage networkEnvelope);
+    protected abstract void onSupportMessage(SupportMessage networkEnvelope);
 
     public abstract NodeAddress getPeerNodeAddress(ChatMessage message);
 
@@ -159,8 +159,7 @@ protected void onChatMessage(ChatMessage chatMessage) {
             sendAckMessage(chatMessage, receiverPubKeyRing, true, null);
     }
 
-    private void onAckMessage(AckMessage ackMessage,
-                              @Nullable DecryptedMessageWithPubKey decryptedMessageWithPubKey) {
+    private void onAckMessage(AckMessage ackMessage) {
         if (ackMessage.getSourceType() == getAckMessageSourceType()) {
             if (ackMessage.isSuccess()) {
                 log.info("Received AckMessage for {} with tradeId {} and uid {}",
@@ -179,10 +178,6 @@ private void onAckMessage(AckMessage ackMessage,
                             msg.setAckError(ackMessage.getErrorMessage());
                     });
             requestPersistence();
-
-            if (decryptedMessageWithPubKey != null) {
-                mailboxMessageService.removeMailboxMsg(decryptedMessageWithPubKey);
-            }
         }
     }
 
@@ -307,9 +302,9 @@ private void applyMessages() {
         decryptedDirectMessageWithPubKeys.forEach(decryptedMessageWithPubKey -> {
             NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
             if (networkEnvelope instanceof SupportMessage) {
-                dispatchMessage((SupportMessage) networkEnvelope);
+                onSupportMessage((SupportMessage) networkEnvelope);
             } else if (networkEnvelope instanceof AckMessage) {
-                onAckMessage((AckMessage) networkEnvelope, null);
+                onAckMessage((AckMessage) networkEnvelope);
             }
         });
         decryptedDirectMessageWithPubKeys.clear();
@@ -318,10 +313,13 @@ private void applyMessages() {
             NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
             log.debug("decryptedMessageWithPubKey.message " + networkEnvelope);
             if (networkEnvelope instanceof SupportMessage) {
-                dispatchMessage((SupportMessage) networkEnvelope);
-                mailboxMessageService.removeMailboxMsg(decryptedMessageWithPubKey);
+                SupportMessage supportMessage = (SupportMessage) networkEnvelope;
+                onSupportMessage(supportMessage);
+                mailboxMessageService.removeMailboxMsg(supportMessage);
             } else if (networkEnvelope instanceof AckMessage) {
-                onAckMessage((AckMessage) networkEnvelope, decryptedMessageWithPubKey);
+                AckMessage ackMessage = (AckMessage) networkEnvelope;
+                onAckMessage(ackMessage);
+                mailboxMessageService.removeMailboxMsg(ackMessage);
             }
         });
         decryptedMailboxMessageWithPubKeys.clear();

core/src/main/java/bisq/core/support/dispute/arbitration/ArbitrationManager.java

@@ -112,7 +112,7 @@ public SupportType getSupportType() {
     }
 
     @Override
-    public void dispatchMessage(SupportMessage message) {
+    public void onSupportMessage(SupportMessage message) {
         if (canProcessMessage(message)) {
             log.info("Received {} with tradeId {} and uid {}",
                     message.getClass().getSimpleName(), message.getTradeId(), message.getUid());

core/src/main/java/bisq/core/support/dispute/mediation/MediationManager.java

@@ -102,7 +102,7 @@ public SupportType getSupportType() {
     }
 
     @Override
-    public void dispatchMessage(SupportMessage message) {
+    public void onSupportMessage(SupportMessage message) {
         if (canProcessMessage(message)) {
             log.info("Received {} with tradeId {} and uid {}",
                     message.getClass().getSimpleName(), message.getTradeId(), message.getUid());

core/src/main/java/bisq/core/support/dispute/refund/RefundManager.java

@@ -96,7 +96,7 @@ public SupportType getSupportType() {
     }
 
     @Override
-    public void dispatchMessage(SupportMessage message) {
+    public void onSupportMessage(SupportMessage message) {
         if (canProcessMessage(message)) {
             log.info("Received {} with tradeId {} and uid {}",
                     message.getClass().getSimpleName(), message.getTradeId(), message.getUid());

core/src/main/java/bisq/core/support/traderchat/TraderChatManager.java

@@ -139,7 +139,7 @@ protected AckMessageSourceType getAckMessageSourceType() {
     // API
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    public void dispatchMessage(SupportMessage message) {
+    public void onSupportMessage(SupportMessage message) {
         if (canProcessMessage(message)) {
             log.info("Received {} with tradeId {} and uid {}",
                     message.getClass().getSimpleName(), message.getTradeId(), message.getUid());

core/src/main/java/bisq/core/trade/closed/CleanupMailboxMessages.java

@@ -25,6 +25,7 @@
 import bisq.network.p2p.BootstrapListener;
 import bisq.network.p2p.DecryptedMessageWithPubKey;
 import bisq.network.p2p.P2PService;
+import bisq.network.p2p.mailbox.MailboxMessage;
 import bisq.network.p2p.mailbox.MailboxMessageService;
 
 import bisq.common.crypto.PubKeyRing;
@@ -88,7 +89,8 @@ private void handleDecryptedMessageWithPubKey(DecryptedMessageWithPubKey decrypt
         trades.stream()
                 .filter(trade -> isMessageForTrade(decryptedMessageWithPubKey, trade))
                 .filter(trade -> isPubKeyValid(decryptedMessageWithPubKey, trade))
-                .forEach(trade -> removeEntryFromMailbox(decryptedMessageWithPubKey, trade));
+                .filter(trade -> decryptedMessageWithPubKey.getNetworkEnvelope() instanceof MailboxMessage)
+                .forEach(trade -> removeEntryFromMailbox((MailboxMessage) decryptedMessageWithPubKey.getNetworkEnvelope(), trade));
     }
 
     private boolean isMessageForTrade(DecryptedMessageWithPubKey decryptedMessageWithPubKey, Trade trade) {
@@ -102,10 +104,11 @@ private boolean isMessageForTrade(DecryptedMessageWithPubKey decryptedMessageWit
         return false;
     }
 
-    private void removeEntryFromMailbox(DecryptedMessageWithPubKey decryptedMessageWithPubKey, Trade trade) {
-        log.info("We found a pending mailbox message ({}) for trade {}. As the trade is closed we remove the mailbox message.",
-                decryptedMessageWithPubKey.getNetworkEnvelope().getClass().getSimpleName(), trade.getId());
-        mailboxMessageService.removeMailboxMsg(decryptedMessageWithPubKey);
+    private void removeEntryFromMailbox(MailboxMessage mailboxMessage, Trade trade) {
+        log.info("We found a pending mailbox message ({}) for trade {}. " +
+                        "As the trade is closed we remove the mailbox message.",
+                mailboxMessage.getClass().getSimpleName(), trade.getId());
+        mailboxMessageService.removeMailboxMsg(mailboxMessage);
     }
 
     private boolean isMyMessage(TradeMessage message, Trade trade) {
@@ -117,15 +120,15 @@ private boolean isMyMessage(AckMessage ackMessage, Trade trade) {
                 ackMessage.getSourceId().equals(trade.getId());
     }
 
-    private boolean isPubKeyValid(DecryptedMessageWithPubKey message, Trade trade) {
+    private boolean isPubKeyValid(DecryptedMessageWithPubKey decryptedMessageWithPubKey, Trade trade) {
         // We can only validate the peers pubKey if we have it already. If we are the taker we get it from the offer
         // Otherwise it depends on the state of the trade protocol if we have received the peers pubKeyRing already.
         PubKeyRing peersPubKeyRing = trade.getProcessModel().getTradingPeer().getPubKeyRing();
         boolean isValid = true;
         if (peersPubKeyRing != null &&
-                !message.getSignaturePubKey().equals(peersPubKeyRing.getSignaturePubKey())) {
+                !decryptedMessageWithPubKey.getSignaturePubKey().equals(peersPubKeyRing.getSignaturePubKey())) {
             isValid = false;
-            log.warn("SignaturePubKey in message does not match the SignaturePubKey we have set for our trading peer.");
+            log.warn("SignaturePubKey in decryptedMessageWithPubKey does not match the SignaturePubKey we have set for our trading peer.");
         }
         return isValid;
     }

core/src/main/java/bisq/core/trade/protocol/TradeProtocol.java

@@ -40,8 +40,6 @@
 import bisq.common.proto.network.NetworkEnvelope;
 import bisq.common.taskrunner.Task;
 
-import java.security.PublicKey;
-
 import java.util.concurrent.TimeUnit;
 
 import lombok.extern.slf4j.Slf4j;
@@ -140,42 +138,37 @@ private void handleDecryptedMessageWithPubKey(DecryptedMessageWithPubKey decrypt
     protected void handleDecryptedMessageWithPubKey(DecryptedMessageWithPubKey decryptedMessageWithPubKey,
                                                     NodeAddress peer) {
         NetworkEnvelope networkEnvelope = decryptedMessageWithPubKey.getNetworkEnvelope();
-        if (networkEnvelope instanceof TradeMessage &&
-                isMyMessage((TradeMessage) networkEnvelope) &&
-                isPubKeyValid(decryptedMessageWithPubKey)) {
+        if (networkEnvelope instanceof MailboxMessage && networkEnvelope instanceof TradeMessage) {
             TradeMessage tradeMessage = (TradeMessage) networkEnvelope;
-
-            // We only remove here if we have already completed the trade.
-            // Otherwise removal is done after successfully applied the task runner.
-            if (trade.isWithdrawn()) {
-                processModel.getP2PService().getMailboxMessageService().removeMailboxMsg(decryptedMessageWithPubKey);
-                log.info("Remove {} from the P2P network.", tradeMessage.getClass().getSimpleName());
-                return;
+            if (isMyMessage(tradeMessage) && isPubKeyValid(decryptedMessageWithPubKey)) {
+                // We only remove here if we have already completed the trade.
+                // Otherwise removal is done after successfully applied the task runner.
+                if (trade.isWithdrawn()) {
+                    MailboxMessage mailboxMessage = (MailboxMessage) tradeMessage;
+                    processModel.getP2PService().getMailboxMessageService().removeMailboxMsg(mailboxMessage);
+                    log.info("Remove {} from the P2P network as trade is already completed.",
+                            tradeMessage.getClass().getSimpleName());
+                    return;
+                }
+                onMailboxMessage(tradeMessage, peer);
             }
-
-            onMailboxMessage(tradeMessage, peer);
-        } else if (networkEnvelope instanceof AckMessage &&
-                isMyMessage((AckMessage) networkEnvelope) &&
-                isPubKeyValid(decryptedMessageWithPubKey)) {
-            if (!trade.isWithdrawn()) {
-                // We only apply the msg if we have not already completed the trade
-                onAckMessage((AckMessage) networkEnvelope, peer);
+        } else if (networkEnvelope instanceof AckMessage) {
+            AckMessage ackMessage = (AckMessage) networkEnvelope;
+            if (isMyMessage(ackMessage) && isPubKeyValid(decryptedMessageWithPubKey)) {
+                if (!trade.isWithdrawn()) {
+                    // We only apply the msg if we have not already completed the trade
+                    onAckMessage(ackMessage, peer);
+                }
+                // In any case we remove the msg
+                processModel.getP2PService().getMailboxMessageService().removeMailboxMsg(ackMessage);
+                log.info("Remove {} from the P2P network.", ackMessage.getClass().getSimpleName());
             }
-            // In any case we remove the msg
-            processModel.getP2PService().getMailboxMessageService().removeMailboxMsg(decryptedMessageWithPubKey);
-            log.info("Remove {} from the P2P network.", networkEnvelope.getClass().getSimpleName());
         }
     }
 
     public void removeMailboxMessageAfterProcessing(TradeMessage tradeMessage) {
-        if (tradeMessage instanceof MailboxMessage &&
-                processModel.getTradingPeer() != null &&
-                processModel.getTradingPeer().getPubKeyRing() != null &&
-                processModel.getTradingPeer().getPubKeyRing().getSignaturePubKey() != null) {
-            PublicKey sigPubKey = processModel.getTradingPeer().getPubKeyRing().getSignaturePubKey();
-            // We reconstruct the DecryptedMessageWithPubKey from the message and the peers signature pubKey
-            DecryptedMessageWithPubKey decryptedMessageWithPubKey = new DecryptedMessageWithPubKey(tradeMessage, sigPubKey);
-            processModel.getP2PService().getMailboxMessageService().removeMailboxMsg(decryptedMessageWithPubKey);
+        if (tradeMessage instanceof MailboxMessage) {
+            processModel.getP2PService().getMailboxMessageService().removeMailboxMsg((MailboxMessage) tradeMessage);
             log.info("Remove {} from the P2P network.", tradeMessage.getClass().getSimpleName());
         }
     }

p2p/src/main/java/bisq/network/p2p/mailbox/MailboxMessageService.java

@@ -261,16 +261,15 @@ public void onFailure(@NotNull Throwable throwable) {
     }
 
     /**
-     * The DecryptedMessageWithPubKey has been applied and we remove it from our local storage and from the network.
+     * The mailboxMessage has been applied and we remove it from our local storage and from the network.
      *
-     * @param decryptedMessageWithPubKey The DecryptedMessageWithPubKey to be removed
+     * @param mailboxMessage The MailboxMessage to be removed
      */
-    public void removeMailboxMsg(DecryptedMessageWithPubKey decryptedMessageWithPubKey) {
+    public void removeMailboxMsg(MailboxMessage mailboxMessage) {
         if (isBootstrapped) {
             // We need to delay a bit to not get a ConcurrentModificationException as we might iterate over
             // mailboxMessageList while getting called.
             UserThread.execute(() -> {
-                MailboxMessage mailboxMessage = (MailboxMessage) decryptedMessageWithPubKey.getNetworkEnvelope();
                 String uid = mailboxMessage.getUid();
 
                 // We called removeMailboxEntryFromNetwork at processMyMailboxItem,
@@ -287,7 +286,7 @@ public void removeMailboxMsg(DecryptedMessageWithPubKey decryptedMessageWithPubK
             });
         } else {
             // In case the network was not ready yet we try again later
-            UserThread.runAfter(() -> removeMailboxMsg(decryptedMessageWithPubKey), 30);
+            UserThread.runAfter(() -> removeMailboxMsg(mailboxMessage), 30);
         }
     }