Traders chat

common/src/main/java/bisq/common/storage/FileManager.java

@@ -72,6 +72,10 @@ public FileManager(File dir, File storageFile, long delay, PersistenceProtoResol
             try {
                 Thread.currentThread().setName("Save-file-task-" + new Random().nextInt(10000));
                 // Runs in an auto save thread.
+                // TODO: this looks like it could cause corrupt data as the savePending is unset before the actual
+                // save. By moving to after the save there might be some persist operations that are not performed
+                // and data would be lost. Probably all persist operations should happen sequencially rather than
+                // skip one when there is already one scheduled
                 if (!savePending.getAndSet(false)) {
                     // Some other scheduled request already beat us to it.
                     return null;

common/src/main/proto/pb.proto

@@ -282,6 +282,10 @@ message PeerOpenedDisputeMessage {
 }
 
 message DisputeCommunicationMessage {
+    enum Type {
+        DISPUTE = 0;
+        TRADE = 1;
+    }
     int64 date = 1;
     string trade_id = 2;
     int32 trader_id = 3;
@@ -296,6 +300,8 @@ message DisputeCommunicationMessage {
     string send_message_error = 12;
     bool acknowledged = 13;
     string ack_error = 14;
+    Type type = 15;
+    bool was_displayed = 16;
 }
 
 message DisputeResultMessage {
@@ -1257,6 +1263,7 @@ message Trade {
     PubKeyRing arbitrator_pub_key_ring = 26;
     PubKeyRing mediator_pub_key_ring = 27;
     string counter_currency_tx_id = 28;
+    repeated DisputeCommunicationMessage communication_messages = 29;
 }
 
 message BuyerAsMakerTrade {

core/src/main/java/bisq/core/arbitration/DisputeChatSession.java

@@ -0,0 +1,177 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.core.arbitration;
+
+import bisq.core.arbitration.messages.DisputeCommunicationMessage;
+import bisq.core.arbitration.messages.DisputeMessage;
+import bisq.core.arbitration.messages.DisputeResultMessage;
+import bisq.core.arbitration.messages.OpenNewDisputeMessage;
+import bisq.core.arbitration.messages.PeerOpenedDisputeMessage;
+import bisq.core.arbitration.messages.PeerPublishedDisputePayoutTxMessage;
+import bisq.core.chat.ChatSession;
+
+import bisq.network.p2p.NodeAddress;
+
+import bisq.common.crypto.PubKeyRing;
+
+import javafx.collections.FXCollections;
+import javafx.collections.ObservableList;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+public class DisputeChatSession extends ChatSession {
+    private static final Logger log = LoggerFactory.getLogger(DisputeChatSession.class);
+
+    @Nullable
+    private Dispute dispute;
+    private DisputeManager disputeManager;
+
+    public DisputeChatSession(@Nullable Dispute dispute,
+                              DisputeManager disputeManager) {
+        super(DisputeCommunicationMessage.Type.DISPUTE);
+        this.dispute = dispute;
+        this.disputeManager = disputeManager;
+    }
+
+    ///////////////////////////////////////////////////////////////////////////////////////////
+    // Dependent on selected dispute
+    ///////////////////////////////////////////////////////////////////////////////////////////
+
+    @Override
+    public boolean isClient() {
+        return dispute != null && disputeManager.isTrader(dispute);
+    }
+
+    @Override
+    public String getTradeId() {
+        return dispute != null ? dispute.getTradeId() : "";
+    }
+
+    @Override
+    public PubKeyRing getClientPubKeyRing() {
+        // Get pubkeyring of trader. Arbitrator is considered server for the chat session
+        return dispute != null ? dispute.getTraderPubKeyRing() : null;
+    }
+
+    @Override
+    public void addDisputeCommunicationMessage(DisputeCommunicationMessage message) {
+        if (dispute != null && (isClient() || (!isClient() && !message.isSystemMessage())))
+            dispute.addDisputeCommunicationMessage(message);
+    }
+
+    @Override
+    public void persist() {
+        disputeManager.getDisputes().persist();
+    }
+
+    @Override
+    public ObservableList<DisputeCommunicationMessage> getDisputeCommunicationMessages() {
+        return dispute != null ? dispute.getDisputeCommunicationMessages() : FXCollections.observableArrayList();
+    }
+
+    @Override
+    public boolean chatIsOpen() {
+        return dispute != null && !dispute.isClosed();
+    }
+
+
+    ///////////////////////////////////////////////////////////////////////////////////////////
+    // Not dependent on selected dispute
+    ///////////////////////////////////////////////////////////////////////////////////////////
+
+    @Nullable
+    @Override
+    public NodeAddress getPeerNodeAddress(DisputeCommunicationMessage message) {
+        Optional<Dispute> disputeOptional = disputeManager.findDispute(message.getTradeId(), message.getTraderId());
+        if (!disputeOptional.isPresent()) {
+            log.warn("Could not find dispute for tradeId = {} traderId = {}",
+                    message.getTradeId(), message.getTraderId());
+            return null;
+        }
+        return disputeManager.getNodeAddressPubKeyRingTuple(disputeOptional.get()).first;
+    }
+
+    @Nullable
+    @Override
+    public PubKeyRing getPeerPubKeyRing(DisputeCommunicationMessage message) {
+        Optional<Dispute> disputeOptional = disputeManager.findDispute(message.getTradeId(), message.getTraderId());
+        if (!disputeOptional.isPresent()) {
+            log.warn("Could not find dispute for tradeId = {} traderId = {}",
+                    message.getTradeId(), message.getTraderId());
+            return null;
+        }
+
+        return disputeManager.getNodeAddressPubKeyRingTuple(disputeOptional.get()).second;
+    }
+
+    @Override
+    public void dispatchMessage(DisputeMessage message) {
+        log.info("Received {} with tradeId {} and uid {}",
+                message.getClass().getSimpleName(), message.getTradeId(), message.getUid());
+
+        if (message instanceof OpenNewDisputeMessage) {
+            disputeManager.onOpenNewDisputeMessage((OpenNewDisputeMessage) message);
+        } else if (message instanceof PeerOpenedDisputeMessage) {
+            disputeManager.onPeerOpenedDisputeMessage((PeerOpenedDisputeMessage) message);
+        } else if (message instanceof DisputeCommunicationMessage) {
+            if (((DisputeCommunicationMessage) message).getType() != DisputeCommunicationMessage.Type.DISPUTE) {
+                log.debug("Ignore non dispute type communication message");
+                return;
+            }
+            disputeManager.getChatManager().onDisputeDirectMessage((DisputeCommunicationMessage) message);
+        } else if (message instanceof DisputeResultMessage) {
+            disputeManager.onDisputeResultMessage((DisputeResultMessage) message);
+        } else if (message instanceof PeerPublishedDisputePayoutTxMessage) {
+            disputeManager.onDisputedPayoutTxMessage((PeerPublishedDisputePayoutTxMessage) message);
+        } else {
+            log.warn("Unsupported message at dispatchMessage.\nmessage=" + message);
+        }
+    }
+
+    @Override
+    public List<DisputeCommunicationMessage> getChatMessages() {
+        return disputeManager.getDisputes().getList().stream()
+                .flatMap(dispute -> dispute.getDisputeCommunicationMessages().stream())
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public boolean channelOpen(DisputeCommunicationMessage message) {
+        return disputeManager.findDispute(message.getTradeId(), message.getTraderId()).isPresent();
+    }
+
+    @Override
+    public void storeDisputeCommunicationMessage(DisputeCommunicationMessage message) {
+        Optional<Dispute> disputeOptional = disputeManager.findDispute(message.getTradeId(), message.getTraderId());
+        if (disputeOptional.isPresent()) {
+            if (disputeOptional.get().getDisputeCommunicationMessages().stream().noneMatch(m -> m.getUid().equals(message.getUid()))) {
+                disputeOptional.get().addDisputeCommunicationMessage(message);
+            } else {
+                log.warn("We got a disputeCommunicationMessage what we have already stored. UId = {} TradeId = {}",
+                        message.getUid(), message.getTradeId());
+            }
+        }
+    }
+}