/*
 * This file is part of Haveno.
 *
 * Haveno 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.
 *
 * Haveno 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 Haveno. If not, see <http://www.gnu.org/licenses/>.
 */

package haveno.core.trade;

import com.google.common.collect.ImmutableList;

import common.utils.GenUtils;
import haveno.common.ClockWatcher;
import haveno.common.UserThread;
import haveno.common.crypto.KeyRing;
import haveno.common.handlers.ErrorMessageHandler;
import haveno.common.handlers.FaultHandler;
import haveno.common.handlers.ResultHandler;
import haveno.common.persistence.PersistenceManager;
import haveno.common.proto.network.NetworkEnvelope;
import haveno.common.proto.persistable.PersistedDataHost;
import haveno.core.api.AccountServiceListener;
import haveno.core.api.CoreAccountService;
import haveno.core.api.CoreNotificationService;
import haveno.core.locale.Res;
import haveno.core.offer.Offer;
import haveno.core.offer.OfferBookService;
import haveno.core.offer.OfferDirection;
import haveno.core.offer.OfferUtil;
import haveno.core.offer.OpenOffer;
import haveno.core.offer.OpenOfferManager;
import haveno.core.offer.SignedOffer;
import haveno.core.offer.availability.OfferAvailabilityModel;
import haveno.core.provider.price.PriceFeedService;
import haveno.core.support.dispute.arbitration.arbitrator.Arbitrator;
import haveno.core.support.dispute.arbitration.arbitrator.ArbitratorManager;
import haveno.core.support.dispute.mediation.mediator.MediatorManager;
import haveno.core.trade.Trade.DisputeState;
import haveno.core.trade.Trade.Phase;
import haveno.core.trade.failed.FailedTradesManager;
import haveno.core.trade.handlers.TradeResultHandler;
import haveno.core.trade.messages.DepositRequest;
import haveno.core.trade.messages.DepositResponse;
import haveno.core.trade.messages.InitMultisigRequest;
import haveno.core.trade.messages.InitTradeRequest;
import haveno.core.trade.messages.SignContractRequest;
import haveno.core.trade.messages.SignContractResponse;
import haveno.core.trade.protocol.ArbitratorProtocol;
import haveno.core.trade.protocol.MakerProtocol;
import haveno.core.trade.protocol.ProcessModel;
import haveno.core.trade.protocol.ProcessModelServiceProvider;
import haveno.core.trade.protocol.TakerProtocol;
import haveno.core.trade.protocol.TradeProtocol;
import haveno.core.trade.protocol.TradeProtocolFactory;
import haveno.core.trade.protocol.TraderProtocol;
import haveno.core.trade.statistics.ReferralIdService;
import haveno.core.trade.statistics.TradeStatisticsManager;
import haveno.core.user.User;
import haveno.core.util.Validator;
import haveno.core.xmr.model.XmrAddressEntry;
import haveno.core.xmr.wallet.XmrWalletService;
import haveno.network.p2p.BootstrapListener;
import haveno.network.p2p.DecryptedDirectMessageListener;
import haveno.network.p2p.DecryptedMessageWithPubKey;
import haveno.network.p2p.NodeAddress;
import haveno.network.p2p.P2PService;
import haveno.network.p2p.network.TorNetworkNode;
import javafx.beans.property.BooleanProperty;
import javafx.beans.property.LongProperty;
import javafx.beans.property.SimpleBooleanProperty;
import javafx.beans.property.SimpleLongProperty;
import javafx.collections.ListChangeListener;
import javafx.collections.ObservableList;
import lombok.Getter;
import lombok.Setter;
import monero.daemon.model.MoneroTx;
import monero.wallet.model.MoneroOutputQuery;
import org.bitcoinj.core.Coin;
import org.bouncycastle.crypto.params.KeyParameter;
import org.fxmisc.easybind.EasyBind;
import org.fxmisc.easybind.monadic.MonadicBinding;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import javax.inject.Inject;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;


public class TradeManager implements PersistedDataHost, DecryptedDirectMessageListener {
    private static final Logger log = LoggerFactory.getLogger(TradeManager.class);

    private boolean isShutDown;
    private final User user;
    @Getter
    private final KeyRing keyRing;
    private final CoreAccountService accountService;
    private final XmrWalletService xmrWalletService;
    @Getter
    private final CoreNotificationService notificationService;
    private final OfferBookService offerBookService;
    private final OpenOfferManager openOfferManager;
    private final ClosedTradableManager closedTradableManager;
    private final FailedTradesManager failedTradesManager;
    private final P2PService p2PService;
    private final PriceFeedService priceFeedService;
    private final TradeStatisticsManager tradeStatisticsManager;
    private final OfferUtil offerUtil;
    private final TradeUtil tradeUtil;
    @Getter
    private final ArbitratorManager arbitratorManager;
    private final MediatorManager mediatorManager;
    private final ProcessModelServiceProvider processModelServiceProvider;
    private final ClockWatcher clockWatcher;

    private final Map<String, TradeProtocol> tradeProtocolByTradeId = new HashMap<>();
    private final PersistenceManager<TradableList<Trade>> persistenceManager;
    private final TradableList<Trade> tradableList = new TradableList<>();
    @Getter
    private final BooleanProperty persistedTradesInitialized = new SimpleBooleanProperty();
    @Setter
    @Nullable
    private ErrorMessageHandler takeOfferRequestErrorMessageHandler;
    @Getter
    private final LongProperty numPendingTrades = new SimpleLongProperty();
    private final ReferralIdService referralIdService;


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Constructor
    ///////////////////////////////////////////////////////////////////////////////////////////

    @Inject
    public TradeManager(User user,
                        KeyRing keyRing,
                        CoreAccountService accountService,
                        XmrWalletService xmrWalletService,
                        CoreNotificationService notificationService,
                        OfferBookService offerBookService,
                        OpenOfferManager openOfferManager,
                        ClosedTradableManager closedTradableManager,
                        FailedTradesManager failedTradesManager,
                        P2PService p2PService,
                        PriceFeedService priceFeedService,
                        TradeStatisticsManager tradeStatisticsManager,
                        OfferUtil offerUtil,
                        TradeUtil tradeUtil,
                        ArbitratorManager arbitratorManager,
                        MediatorManager mediatorManager,
                        ProcessModelServiceProvider processModelServiceProvider,
                        ClockWatcher clockWatcher,
                        PersistenceManager<TradableList<Trade>> persistenceManager,
                        ReferralIdService referralIdService) {
        this.user = user;
        this.keyRing = keyRing;
        this.accountService = accountService;
        this.xmrWalletService = xmrWalletService;
        this.notificationService = notificationService;
        this.offerBookService = offerBookService;
        this.openOfferManager = openOfferManager;
        this.closedTradableManager = closedTradableManager;
        this.failedTradesManager = failedTradesManager;
        this.p2PService = p2PService;
        this.priceFeedService = priceFeedService;
        this.tradeStatisticsManager = tradeStatisticsManager;
        this.offerUtil = offerUtil;
        this.tradeUtil = tradeUtil;
        this.arbitratorManager = arbitratorManager;
        this.mediatorManager = mediatorManager;
        this.processModelServiceProvider = processModelServiceProvider;
        this.clockWatcher = clockWatcher;
        this.referralIdService = referralIdService;
        this.persistenceManager = persistenceManager;

        this.persistenceManager.initialize(tradableList, "PendingTrades", PersistenceManager.Source.PRIVATE);

        p2PService.addDecryptedDirectMessageListener(this);

        failedTradesManager.setUnFailTradeCallback(this::unFailTrade);

        // initialize trades when connected to p2p network
        p2PService.addP2PServiceListener(new BootstrapListener() {
            public void onUpdatedDataReceived() {
                initPersistedTrades();
            }
        });
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // PersistedDataHost
    ///////////////////////////////////////////////////////////////////////////////////////////

    @Override
    public void readPersisted(Runnable completeHandler) {
        persistenceManager.readPersisted(persisted -> {
                    tradableList.setAll(persisted.getList());
                    tradableList.stream()
                            .filter(trade -> trade.getOffer() != null)
                            .forEach(trade -> trade.getOffer().setPriceFeedService(priceFeedService));
                    completeHandler.run();
                },
                completeHandler);
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // DecryptedDirectMessageListener
    ///////////////////////////////////////////////////////////////////////////////////////////

    @Override
    public void onDirectMessage(DecryptedMessageWithPubKey message, NodeAddress peer) {
        NetworkEnvelope networkEnvelope = message.getNetworkEnvelope();
        new Thread(() -> {
            if (networkEnvelope instanceof InitTradeRequest) {
                handleInitTradeRequest((InitTradeRequest) networkEnvelope, peer);
            } else if (networkEnvelope instanceof InitMultisigRequest) {
                handleInitMultisigRequest((InitMultisigRequest) networkEnvelope, peer);
            } else if (networkEnvelope instanceof SignContractRequest) {
                handleSignContractRequest((SignContractRequest) networkEnvelope, peer);
            } else if (networkEnvelope instanceof SignContractResponse) {
                handleSignContractResponse((SignContractResponse) networkEnvelope, peer);
            } else if (networkEnvelope instanceof DepositRequest) {
                handleDepositRequest((DepositRequest) networkEnvelope, peer);
            } else if (networkEnvelope instanceof DepositResponse) {
                handleDepositResponse((DepositResponse) networkEnvelope, peer);
            }
        }).start();
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Lifecycle
    ///////////////////////////////////////////////////////////////////////////////////////////

    public void onAllServicesInitialized() {

        // listen for account updates
        accountService.addListener(new AccountServiceListener() {

            @Override
            public void onAccountCreated() {
                log.info(TradeManager.class + ".accountService.onAccountCreated()");
                initPersistedTrades();
            }

            @Override
            public void onAccountOpened() {
                log.info(TradeManager.class + ".accountService.onAccountOpened()");
                initPersistedTrades();
            }

            @Override
            public void onAccountClosed() {
                log.info(TradeManager.class + ".accountService.onAccountClosed()");
                closeAllTrades();
            }

            @Override
            public void onPasswordChanged(String oldPassword, String newPassword) {
                // handled in XmrWalletService
            }
        });
    }

    public void onShutDownStarted() {
        log.info("{}.onShutDownStarted()", getClass().getSimpleName());

        // collect trades to prepare
        Set<Trade> trades = new HashSet<Trade>();
        trades.addAll(tradableList.getList());
        trades.addAll(closedTradableManager.getClosedTrades());
        trades.addAll(failedTradesManager.getObservableList());

        // prepare to shut down trades in parallel
        Set<Runnable> tasks = new HashSet<Runnable>();
        for (Trade trade : trades) tasks.add(() -> {
            try {
                trade.onShutDownStarted();
            } catch (Exception e) {
                if (e.getMessage() != null && e.getMessage().contains("Connection reset")) return; // expected if shut down with ctrl+c
                log.warn("Error notifying {} {} that shut down started {}", getClass().getSimpleName(), trade.getId());
                e.printStackTrace();
            }
        });
        try {
            HavenoUtils.executeTasks(tasks);
        } catch (Exception e) {
            log.warn("Error notifying trades that shut down started: {}", e.getMessage());
            e.printStackTrace();
        }
    }

    public void shutDown() {
        log.info("Shutting down {}", getClass().getSimpleName());
        isShutDown = true;
        closeAllTrades();
    }

    private void closeAllTrades() {

        // collect trades to shutdown
        Set<Trade> trades = new HashSet<Trade>();
        trades.addAll(tradableList.getList());
        trades.addAll(closedTradableManager.getClosedTrades());
        trades.addAll(failedTradesManager.getObservableList());

        // shut down trades in parallel
        Set<Runnable> tasks = new HashSet<Runnable>();
        for (Trade trade : trades) tasks.add(() -> {
            try {
                trade.shutDown();
            } catch (Exception e) {
                if (e.getMessage() != null && (e.getMessage().contains("Connection reset") || e.getMessage().contains("Connection refused"))) return; // expected if shut down with ctrl+c
                log.warn("Error closing {} {}", trade.getClass().getSimpleName(), trade.getId());
                e.printStackTrace();
            }
        });
        try {
            HavenoUtils.executeTasks(tasks);
        } catch (Exception e) {
            log.warn("Error shutting down trades: {}", e.getMessage());
            e.printStackTrace();
        }
    }

    private void thawUnreservedOutputs() {
        if (xmrWalletService.getWallet() == null) return;

        // collect reserved outputs
        Set<String> reservedKeyImages = new HashSet<String>();
        for (Trade trade : getObservableList()) {
            if (trade.getSelf().getReserveTxKeyImages() == null) continue;
            reservedKeyImages.addAll(trade.getSelf().getReserveTxKeyImages());
        }
        for (OpenOffer openOffer : openOfferManager.getObservableList()) {
            reservedKeyImages.addAll(openOffer.getOffer().getOfferPayload().getReserveTxKeyImages());
        }

        // thaw unreserved outputs
        Set<String> unreservedFrozenKeyImages = xmrWalletService.getWallet().getOutputs(new MoneroOutputQuery()
                .setIsFrozen(true)
                .setIsSpent(false))
                .stream()
                .map(output -> output.getKeyImage().getHex())
                .collect(Collectors.toSet());
        unreservedFrozenKeyImages.removeAll(reservedKeyImages);
        if (!unreservedFrozenKeyImages.isEmpty()) {
            log.warn("Thawing outputs which are not reserved for offer or trade: " + unreservedFrozenKeyImages);
            xmrWalletService.thawOutputs(unreservedFrozenKeyImages);
            xmrWalletService.saveMainWallet();
        }
    }

    public TradeProtocol getTradeProtocol(Trade trade) {
        synchronized (tradeProtocolByTradeId) {
            return tradeProtocolByTradeId.get(trade.getUid());
        }
    }

    public TradeProtocol createTradeProtocol(Trade trade) {
        synchronized (tradeProtocolByTradeId) {
            TradeProtocol tradeProtocol = TradeProtocolFactory.getNewTradeProtocol(trade);
            TradeProtocol prev = tradeProtocolByTradeId.put(trade.getUid(), tradeProtocol);
            if (prev != null) log.error("We had already an entry with uid {}", trade.getUid());
            return tradeProtocol;
        }
    }

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Init pending trade
    ///////////////////////////////////////////////////////////////////////////////////////////

    private void initPersistedTrades() {
        log.info("Initializing persisted trades");

        // initialize off main thread
        new Thread(() -> {

            // get all trades
            List<Trade> trades = getAllTrades();

            // initialize trades in parallel
            int threadPoolSize = 10;
            Set<Runnable> tasks = new HashSet<Runnable>();
            for (Trade trade : trades) {
                tasks.add(() -> {
                    try {
                        initPersistedTrade(trade);

                        // remove trade if protocol didn't initialize
                        if (getOpenTradeByUid(trade.getId()).isPresent() && !trade.isDepositRequested()) {
                            log.warn("Removing persisted {} {} with uid={} because it did not finish initializing (state={})", trade.getClass().getSimpleName(), trade.getId(), trade.getUid(), trade.getState());
                            maybeRemoveTradeOnError(trade);
                        }
                    } catch (Exception e) {
                        log.warn("Error initializing {} {}: {}", trade.getClass().getSimpleName(), trade.getId(), e.getMessage());
                        e.printStackTrace();
                        trade.prependErrorMessage(e.getMessage());
                    }
                });
            };
            HavenoUtils.executeTasks(tasks, threadPoolSize);
            log.info("Done initializing persisted trades");
            if (isShutDown) return;

            // sync idle trades once in background after active trades
            for (Trade trade : trades) {
                if (trade.isIdling())  {
                    HavenoUtils.submitTask(() -> trade.syncWallet());
                }
            }
    
            getObservableList().addListener((ListChangeListener<Trade>) change -> onTradesChanged());
            onTradesChanged();
    
            xmrWalletService.setTradeManager(this);

            // process after all wallets initialized
            if (HavenoUtils.havenoSetup == null) throw new IllegalStateException("HavenoSetup is null; is this an improperly registered seed node?");
            MonadicBinding<Boolean> walletsInitialized = EasyBind.combine(HavenoUtils.havenoSetup.getWalletInitialized(), persistedTradesInitialized, (a, b) -> a && b);
            walletsInitialized.subscribe((observable, oldValue, newValue) -> {
                if (!newValue) return;

                // thaw unreserved outputs
                thawUnreservedOutputs();

                // reset any available funded address entries
                xmrWalletService.getAddressEntriesForAvailableBalanceStream()
                        .filter(addressEntry -> addressEntry.getOfferId() != null)
                        .forEach(addressEntry -> {
                            log.warn("Swapping pending {} entries at startup. offerId={}", addressEntry.getContext(), addressEntry.getOfferId());
                            xmrWalletService.swapTradeEntryToAvailableEntry(addressEntry.getOfferId(), addressEntry.getContext());
                        });
            });

            // notify that persisted trades initialized
            persistedTradesInitialized.set(true);
    
            // We do not include failed trades as they should not be counted anyway in the trade statistics
            // TODO: remove stats?
            Set<Trade> nonFailedTrades = new HashSet<>(closedTradableManager.getClosedTrades());
            nonFailedTrades.addAll(tradableList.getList());
            String referralId = referralIdService.getOptionalReferralId().orElse(null);
            boolean isTorNetworkNode = p2PService.getNetworkNode() instanceof TorNetworkNode;
            tradeStatisticsManager.maybeRepublishTradeStatistics(nonFailedTrades, referralId, isTorNetworkNode);
        }).start();

        // allow execution to start
        GenUtils.waitFor(100);
    }

    private void initPersistedTrade(Trade trade) {
        if (isShutDown) return;
        if (getTradeProtocol(trade) != null) return;
        initTradeAndProtocol(trade, createTradeProtocol(trade));
        requestPersistence();
        scheduleDeletionIfUnfunded(trade);
    }

    private void initTradeAndProtocol(Trade trade, TradeProtocol tradeProtocol) {
        tradeProtocol.initialize(processModelServiceProvider, this);
        requestPersistence(); // TODO requesting persistence twice with initPersistedTrade()
    }

    public void requestPersistence() {
        persistenceManager.requestPersistence();
    }

    private void handleInitTradeRequest(InitTradeRequest request, NodeAddress sender) {
      log.info("Received InitTradeRequest from {} with tradeId {} and uid {}", sender, request.getTradeId(), request.getUid());

      try {
          Validator.nonEmptyStringOf(request.getTradeId());
      } catch (Throwable t) {
          log.warn("Invalid InitTradeRequest message " + request.toString());
          return;
      }

      // handle request as arbitrator
      boolean isArbitrator = request.getArbitratorNodeAddress().equals(p2PService.getNetworkNode().getNodeAddress());
      if (isArbitrator) {

          // verify this node is registered arbitrator
          Arbitrator thisArbitrator = user.getRegisteredArbitrator();
          NodeAddress thisAddress = p2PService.getNetworkNode().getNodeAddress();
          if (thisArbitrator == null || !thisArbitrator.getNodeAddress().equals(thisAddress)) {
              log.warn("Ignoring InitTradeRequest from {} with tradeId {} because we are not an arbitrator", sender, request.getTradeId());
              return;
          }

          // get offer associated with trade
          Offer offer = null;
          for (Offer anOffer : offerBookService.getOffers()) {
              if (anOffer.getId().equals(request.getTradeId())) {
                  offer = anOffer;
              }
          }
          if (offer == null) {
              log.warn("Ignoring InitTradeRequest from {} with tradeId {} because offer is not on the books", sender, request.getTradeId());
              return;
          }

          // verify arbitrator is payload signer unless they are offline
          // TODO (woodser): handle if payload signer differs from current arbitrator (verify signer is offline)

          // verify maker is offer owner
          // TODO (woodser): maker address might change if they disconnect and reconnect, should allow maker address to differ if pubKeyRing is same ?
          if (!offer.getOwnerNodeAddress().equals(request.getMakerNodeAddress())) {
              log.warn("Ignoring InitTradeRequest from {} with tradeId {} because maker is not offer owner", sender, request.getTradeId());
              return;
          }

          Trade trade;
          Optional<Trade> tradeOptional = getOpenTrade(offer.getId());
          if (tradeOptional.isPresent()) {
              trade = tradeOptional.get();

              // verify request is from maker
              if (!sender.equals(request.getMakerNodeAddress())) {
                  log.warn("Trade is already taken"); // TODO (woodser): need to respond with bad ack
                  return;
              }
          } else {

            // verify request is from taker
            if (!sender.equals(request.getTakerNodeAddress())) {
                log.warn("Ignoring InitTradeRequest from {} with tradeId {} because request must be from taker when trade is not initialized", sender, request.getTradeId());
                return;
            }

            // get expected taker fee
            BigInteger takerFee = HavenoUtils.getTakerFee(BigInteger.valueOf(offer.getOfferPayload().getAmount()));

            // create arbitrator trade
            trade = new ArbitratorTrade(offer,
                    BigInteger.valueOf(offer.getOfferPayload().getAmount()),
                    takerFee,
                    offer.getOfferPayload().getPrice(),
                    xmrWalletService,
                    getNewProcessModel(offer),
                    UUID.randomUUID().toString(),
                    request.getMakerNodeAddress(),
                    request.getTakerNodeAddress(),
                    request.getArbitratorNodeAddress());

            // set reserve tx hash if available
            Optional<SignedOffer> signedOfferOptional = openOfferManager.getSignedOfferById(request.getTradeId());
            if (signedOfferOptional.isPresent()) {
                SignedOffer signedOffer = signedOfferOptional.get();
                trade.getMaker().setReserveTxHash(signedOffer.getReserveTxHash());
            }

            // initialize trade protocol
            initTradeAndProtocol(trade, createTradeProtocol(trade));
            synchronized (tradableList) {
                tradableList.add(trade);
            }
          }

          ((ArbitratorProtocol) getTradeProtocol(trade)).handleInitTradeRequest(request, sender, errorMessage -> {
              log.warn("Arbitrator error during trade initialization for trade {}: {}", trade.getId(), errorMessage);
              if (trade.getMaker().getReserveTxHash() != null || trade.getTaker().getReserveTxHash() != null) {
                onMoveInvalidTradeToFailedTrades(trade); // arbitrator retains failed trades for analysis and penalty
              } else {
                maybeRemoveTradeOnError(trade);
              }
              if (takeOfferRequestErrorMessageHandler != null) takeOfferRequestErrorMessageHandler.handleErrorMessage(errorMessage);
          });

          requestPersistence();
      }

      // handle request as maker
      else {

          Optional<OpenOffer> openOfferOptional = openOfferManager.getOpenOfferById(request.getTradeId());
          if (!openOfferOptional.isPresent()) {
              return;
          }

          OpenOffer openOffer = openOfferOptional.get();
          if (openOffer.getState() != OpenOffer.State.AVAILABLE) {
              return;
          }

          Offer offer = openOffer.getOffer();

          // verify request is from arbitrator
          Arbitrator arbitrator = user.getAcceptedArbitratorByAddress(sender);
          if (arbitrator == null) {
              log.warn("Ignoring InitTradeRequest from {} with tradeId {} because request is not from accepted arbitrator", sender, request.getTradeId());
              return;
          }

          Optional<Trade> tradeOptional = getOpenTrade(request.getTradeId());
          if (tradeOptional.isPresent()) {
              log.warn("Maker trade already exists with id " + request.getTradeId() + ". This should never happen.");
              return;
          }

          // reserve open offer
          openOfferManager.reserveOpenOffer(openOffer); // TODO (woodser): reserve offer if arbitrator? probably. or, arbitrator does not have open offer?

          // get expected taker fee
          BigInteger takerFee = HavenoUtils.getTakerFee(BigInteger.valueOf(offer.getOfferPayload().getAmount()));

          Trade trade;
          if (offer.isBuyOffer())
              trade = new BuyerAsMakerTrade(offer,
                      BigInteger.valueOf(offer.getOfferPayload().getAmount()),
                      takerFee,
                      offer.getOfferPayload().getPrice(),
                      xmrWalletService,
                      getNewProcessModel(offer),
                      UUID.randomUUID().toString(),
                      request.getMakerNodeAddress(),
                      request.getTakerNodeAddress(),
                      request.getArbitratorNodeAddress());
          else
              trade = new SellerAsMakerTrade(offer,
                      BigInteger.valueOf(offer.getOfferPayload().getAmount()),
                      takerFee,
                      offer.getOfferPayload().getPrice(),
                      xmrWalletService,
                      getNewProcessModel(offer),
                      UUID.randomUUID().toString(),
                      request.getMakerNodeAddress(),
                      request.getTakerNodeAddress(),
                      request.getArbitratorNodeAddress());

          trade.getArbitrator().setPubKeyRing(arbitrator.getPubKeyRing());
          trade.getMaker().setPubKeyRing(trade.getOffer().getPubKeyRing());
          initTradeAndProtocol(trade, createTradeProtocol(trade));
          trade.getSelf().setPaymentAccountId(offer.getOfferPayload().getMakerPaymentAccountId());
          trade.getSelf().setReserveTxHash(openOffer.getReserveTxHash()); // TODO (woodser): initialize in initTradeAndProtocol?
          trade.getSelf().setReserveTxHex(openOffer.getReserveTxHex());
          trade.getSelf().setReserveTxKey(openOffer.getReserveTxKey());
          trade.getSelf().setReserveTxKeyImages(offer.getOfferPayload().getReserveTxKeyImages());
          synchronized (tradableList) {
              tradableList.add(trade);
          }

          // notify on phase changes
          // TODO (woodser): save subscription, bind on startup
          EasyBind.subscribe(trade.statePhaseProperty(), phase -> {
              if (phase == Phase.DEPOSITS_PUBLISHED) {
                  notificationService.sendTradeNotification(trade, "Offer Taken", "Your offer " + offer.getId() + " has been accepted"); // TODO (woodser): use language translation
              }
          });

          ((MakerProtocol) getTradeProtocol(trade)).handleInitTradeRequest(request, sender, errorMessage -> {
              log.warn("Maker error during trade initialization: " + errorMessage);
              maybeRemoveTradeOnError(trade);
              openOfferManager.unreserveOpenOffer(openOffer); // offer remains available // TODO: only unreserve if funds not deposited to multisig
              if (takeOfferRequestErrorMessageHandler != null) takeOfferRequestErrorMessageHandler.handleErrorMessage(errorMessage);
          });

          requestPersistence();
      }
    }

    private void handleInitMultisigRequest(InitMultisigRequest request, NodeAddress peer) {
      log.info("Received InitMultisigRequest from {} with tradeId {} and uid {}", peer, request.getTradeId(), request.getUid());

      try {
          Validator.nonEmptyStringOf(request.getTradeId());
      } catch (Throwable t) {
          log.warn("Invalid InitMultisigRequest " + request.toString());
          return;
      }

      Optional<Trade> tradeOptional = getOpenTrade(request.getTradeId());
      if (!tradeOptional.isPresent()) {
          log.warn("No trade with id " + request.getTradeId() + " at node " + P2PService.getMyNodeAddress());
          return;
      }
      Trade trade = tradeOptional.get();
      getTradeProtocol(trade).handleInitMultisigRequest(request, peer);
    }

    private void handleSignContractRequest(SignContractRequest request, NodeAddress peer) {
        log.info("Received SignContractRequest from {} with tradeId {} and uid {}", peer, request.getTradeId(), request.getUid());

        try {
            Validator.nonEmptyStringOf(request.getTradeId());
        } catch (Throwable t) {
            log.warn("Invalid SignContractRequest message " + request.toString());
            return;
        }

        Optional<Trade> tradeOptional = getOpenTrade(request.getTradeId());
        if (!tradeOptional.isPresent()) {
            log.warn("No trade with id " + request.getTradeId());
            return;
        }
        Trade trade = tradeOptional.get();
        getTradeProtocol(trade).handleSignContractRequest(request, peer);
    }

    private void handleSignContractResponse(SignContractResponse request, NodeAddress peer) {
        log.info("Received SignContractResponse from {} with tradeId {} and uid {}", peer, request.getTradeId(), request.getUid());

        try {
            Validator.nonEmptyStringOf(request.getTradeId());
        } catch (Throwable t) {
            log.warn("Invalid SignContractResponse message " + request.toString());
            return;
        }

        Optional<Trade> tradeOptional = getOpenTrade(request.getTradeId());
        if (!tradeOptional.isPresent()) {
            log.warn("No trade with id " + request.getTradeId());
            return;
        }
        Trade trade = tradeOptional.get();
        ((TraderProtocol) getTradeProtocol(trade)).handleSignContractResponse(request, peer);
    }

    private void handleDepositRequest(DepositRequest request, NodeAddress peer) {
        log.info("Received DepositRequest from {} with tradeId {} and uid {}", peer, request.getTradeId(), request.getUid());

        try {
            Validator.nonEmptyStringOf(request.getTradeId());
        } catch (Throwable t) {
            log.warn("Invalid DepositRequest message " + request.toString());
            return;
        }

        Optional<Trade> tradeOptional = getOpenTrade(request.getTradeId());
        if (!tradeOptional.isPresent()) {
            log.warn("No trade with id " + request.getTradeId());
            return;
        }
        Trade trade = tradeOptional.get();
        ((ArbitratorProtocol) getTradeProtocol(trade)).handleDepositRequest(request, peer);
    }

    private void handleDepositResponse(DepositResponse response, NodeAddress peer) {
        log.info("Received DepositResponse from {} with tradeId {} and uid {}", peer, response.getTradeId(), response.getUid());

        try {
            Validator.nonEmptyStringOf(response.getTradeId());
        } catch (Throwable t) {
            log.warn("Invalid DepositResponse message " + response.toString());
            return;
        }

        Optional<Trade> tradeOptional = getOpenTrade(response.getTradeId());
        if (!tradeOptional.isPresent()) {
            log.warn("No trade with id " + response.getTradeId());
            return;
        }
        Trade trade = tradeOptional.get();
        ((TraderProtocol) getTradeProtocol(trade)).handleDepositResponse(response, peer);
    }

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Take offer
    ///////////////////////////////////////////////////////////////////////////////////////////

    public void checkOfferAvailability(Offer offer,
                                       boolean isTakerApiUser,
                                       String paymentAccountId,
                                       ResultHandler resultHandler,
                                       ErrorMessageHandler errorMessageHandler) {
        offer.checkOfferAvailability(getOfferAvailabilityModel(offer, isTakerApiUser, paymentAccountId), resultHandler, errorMessageHandler);
    }

    // First we check if offer is still available then we create the trade with the protocol
    public void onTakeOffer(BigInteger amount,
                            BigInteger takerFee,
                            BigInteger fundsNeededForTrade,
                            Offer offer,
                            String paymentAccountId,
                            boolean useSavingsWallet,
                            boolean isTakerApiUser,
                            TradeResultHandler tradeResultHandler,
                            ErrorMessageHandler errorMessageHandler) {

        checkArgument(!wasOfferAlreadyUsedInTrade(offer.getId()));

        OfferAvailabilityModel model = getOfferAvailabilityModel(offer, isTakerApiUser, paymentAccountId);
        offer.checkOfferAvailability(model,
                () -> {
                    if (offer.getState() == Offer.State.AVAILABLE) {
                        Trade trade;
                        if (offer.isBuyOffer()) {
                            trade = new SellerAsTakerTrade(offer,
                                    amount,
                                    takerFee,
                                    model.getTradeRequest().getTradePrice(),
                                    xmrWalletService,
                                    getNewProcessModel(offer),
                                    UUID.randomUUID().toString(),
                                    model.getPeerNodeAddress(),
                                    P2PService.getMyNodeAddress(),
                                    offer.getOfferPayload().getArbitratorSigner());
                        } else {
                            trade = new BuyerAsTakerTrade(offer,
                                    amount,
                                    takerFee,
                                    model.getTradeRequest().getTradePrice(),
                                    xmrWalletService,
                                    getNewProcessModel(offer),
                                    UUID.randomUUID().toString(),
                                    model.getPeerNodeAddress(),
                                    P2PService.getMyNodeAddress(),
                                    offer.getOfferPayload().getArbitratorSigner());
                        }

                        trade.getProcessModel().setTradeMessage(model.getTradeRequest());
                        trade.getProcessModel().setMakerSignature(model.getMakerSignature());
                        trade.getProcessModel().setUseSavingsWallet(useSavingsWallet);
                        trade.getProcessModel().setFundsNeededForTrade(fundsNeededForTrade.longValueExact());
                        trade.getMaker().setPubKeyRing(trade.getOffer().getPubKeyRing());
                        trade.getSelf().setPubKeyRing(model.getPubKeyRing());
                        trade.getSelf().setPaymentAccountId(paymentAccountId);
                        trade.addInitProgressStep();

                        // ensure trade is not already open
                        Optional<Trade> tradeOptional = getOpenTrade(offer.getId());
                        if (tradeOptional.isPresent()) throw new RuntimeException("Cannot create trade protocol because trade with ID " + trade.getId() + " is already open");

                        // initialize trade protocol
                        TradeProtocol tradeProtocol = createTradeProtocol(trade);
                        synchronized (tradableList) {
                            tradableList.add(trade);
                        }

                        initTradeAndProtocol(trade, tradeProtocol);

                        // take offer and persist trade on success
                        ((TakerProtocol) tradeProtocol).onTakeOffer(result -> {
                            tradeResultHandler.handleResult(trade);
                            requestPersistence();
                        }, errorMessage -> {
                            log.warn("Taker error during trade initialization: " + errorMessage);
                            maybeRemoveTradeOnError(trade);
                            errorMessageHandler.handleErrorMessage(errorMessage);
                            if (takeOfferRequestErrorMessageHandler != null) takeOfferRequestErrorMessageHandler.handleErrorMessage(errorMessage);
                        });
                        requestPersistence();
                    } else {
                        log.warn("Cannot take offer {} because it's not available, state={}", offer.getId(), offer.getState());
                    }
                },
                errorMessage -> {
                    log.warn("Taker error during check offer availability: " + errorMessage);
                    errorMessageHandler.handleErrorMessage(errorMessage);
                    if (takeOfferRequestErrorMessageHandler != null) takeOfferRequestErrorMessageHandler.handleErrorMessage(errorMessage);
                });

        requestPersistence();
    }

    private ProcessModel getNewProcessModel(Offer offer) {
        return new ProcessModel(checkNotNull(offer).getId(),
                processModelServiceProvider.getUser().getAccountId(),
                processModelServiceProvider.getKeyRing().getPubKeyRing());
    }

    private OfferAvailabilityModel getOfferAvailabilityModel(Offer offer, boolean isTakerApiUser, String paymentAccountId) {
        return new OfferAvailabilityModel(
                offer,
                keyRing.getPubKeyRing(),
                xmrWalletService,
                p2PService,
                user,
                mediatorManager,
                tradeStatisticsManager,
                isTakerApiUser,
                paymentAccountId,
                offerUtil);
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Complete trade
    ///////////////////////////////////////////////////////////////////////////////////////////

    // TODO (woodser): remove this function
    public void onWithdrawRequest(String toAddress,
          Coin amount,
          Coin fee,
          KeyParameter aesKey,
          Trade trade,
          @Nullable String memo,
          ResultHandler resultHandler,
          FaultHandler faultHandler) {
        throw new RuntimeException("Withdraw trade funds after payout to Haveno wallet not supported");
    }

    // If trade was completed (closed without fault but might be closed by a dispute) we move it to the closed trades
    public void onTradeCompleted(Trade trade) {
        if (trade.isCompleted()) throw new RuntimeException("Trade " + trade.getId() + " was already completed");
        closedTradableManager.add(trade);
        trade.setState(Trade.State.TRADE_COMPLETED);
        removeTrade(trade);

        // TODO The address entry should have been removed already. Check and if its the case remove that.
        xmrWalletService.resetAddressEntriesForPendingTrade(trade.getId());
        requestPersistence();
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Dispute
    ///////////////////////////////////////////////////////////////////////////////////////////

    public void closeDisputedTrade(String tradeId, DisputeState disputeState) {
        Optional<Trade> tradeOptional = getOpenTrade(tradeId);
        if (tradeOptional.isPresent()) {
            Trade trade = tradeOptional.get();
            trade.setDisputeState(disputeState);
            onTradeCompleted(trade);
            xmrWalletService.swapTradeEntryToAvailableEntry(trade.getId(), XmrAddressEntry.Context.TRADE_PAYOUT);
            requestPersistence();
        }
    }

    ///////////////////////////////////////////////////////////////////////////////////////////
    // Trade period state
    ///////////////////////////////////////////////////////////////////////////////////////////

    public void applyTradePeriodState() {
        updateTradePeriodState();
        clockWatcher.addListener(new ClockWatcher.Listener() {
            @Override
            public void onSecondTick() {
            }

            @Override
            public void onMinuteTick() {
                updateTradePeriodState();
            }
        });
    }

    private void updateTradePeriodState() {
        for (Trade trade : new ArrayList<Trade>(tradableList.getList())) {
            if (!trade.isPayoutPublished()) {
                Date maxTradePeriodDate = trade.getMaxTradePeriodDate();
                Date halfTradePeriodDate = trade.getHalfTradePeriodDate();
                if (maxTradePeriodDate != null && halfTradePeriodDate != null) {
                    Date now = new Date();
                    if (now.after(maxTradePeriodDate)) {
                        trade.setPeriodState(Trade.TradePeriodState.TRADE_PERIOD_OVER);
                        requestPersistence();
                    } else if (now.after(halfTradePeriodDate)) {
                        trade.setPeriodState(Trade.TradePeriodState.SECOND_HALF);
                        requestPersistence();
                    }
                }
            }
        }
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Failed trade handling
    ///////////////////////////////////////////////////////////////////////////////////////////

    // If trade is in already in critical state (if taker role: taker fee; both roles: after deposit published)
    // we move the trade to failedTradesManager
    public void onMoveInvalidTradeToFailedTrades(Trade trade) {
        maybeRemoveTradeOnError(trade);
        failedTradesManager.add(trade);
    }

    public void addFailedTradeToPendingTrades(Trade trade) {
        if (!trade.isInitialized()) {
            initPersistedTrade(trade);
        }
        addTrade(trade);
    }

    public Stream<Trade> getTradesStreamWithFundsLockedIn() {
        synchronized (tradableList) {
            return getObservableList().stream().filter(Trade::isFundsLockedIn);
        }
    }

    public Set<String> getSetOfFailedOrClosedTradeIdsFromLockedInFunds() throws TradeTxException {
        AtomicReference<TradeTxException> tradeTxException = new AtomicReference<>();
        synchronized (tradableList) {
            Set<String> tradesIdSet = getTradesStreamWithFundsLockedIn()
                    .filter(Trade::hasFailed)
                    .map(Trade::getId)
                    .collect(Collectors.toSet());
            tradesIdSet.addAll(failedTradesManager.getTradesStreamWithFundsLockedIn()
                    .filter(trade -> trade.getMakerDepositTx() != null || trade.getTakerDepositTx() != null)
                    .map(trade -> {
                        log.warn("We found a failed trade with locked up funds. " +
                                "That should never happen. trade ID=" + trade.getId());
                        return trade.getId();
                    })
                    .collect(Collectors.toSet()));
            tradesIdSet.addAll(closedTradableManager.getTradesStreamWithFundsLockedIn()
                    .map(trade -> {
                        MoneroTx makerDepositTx = trade.getMakerDepositTx();
                        if (makerDepositTx != null) {
                            if (!makerDepositTx.isConfirmed()) {
                                tradeTxException.set(new TradeTxException(Res.get("error.closedTradeWithUnconfirmedDepositTx", trade.getShortId()))); // TODO (woodser): rename to closedTradeWithLockedDepositTx
                            } else {
                                log.warn("We found a closed trade with locked up funds. " +
                                        "That should never happen. {} ID={}, state={}, payoutState={}, disputeState={}", trade.getClass().getSimpleName(), trade.getId(), trade.getState(), trade.getPayoutState(), trade.getDisputeState());
                            }
                        } else {
                            log.warn("Closed trade with locked up funds missing maker deposit tx. {} ID={}, state={}, payoutState={}, disputeState={}", trade.getClass().getSimpleName(), trade.getId(), trade.getState(), trade.getPayoutState(), trade.getDisputeState());
                            tradeTxException.set(new TradeTxException(Res.get("error.closedTradeWithNoDepositTx", trade.getShortId())));
                        }

                        MoneroTx takerDepositTx = trade.getTakerDepositTx();
                        if (takerDepositTx != null) {
                            if (!takerDepositTx.isConfirmed()) {
                                tradeTxException.set(new TradeTxException(Res.get("error.closedTradeWithUnconfirmedDepositTx", trade.getShortId())));
                            } else {
                                log.warn("We found a closed trade with locked up funds. " +
                                        "That should never happen. trade ID={} ID={}, state={}, payoutState={}, disputeState={}", trade.getClass().getSimpleName(), trade.getId(), trade.getState(), trade.getPayoutState(), trade.getDisputeState());
                            }
                        } else {
                            log.warn("Closed trade with locked up funds missing taker deposit tx. {} ID={}, state={}, payoutState={}, disputeState={}", trade.getClass().getSimpleName(), trade.getId(), trade.getState(), trade.getPayoutState(), trade.getDisputeState());
                            tradeTxException.set(new TradeTxException(Res.get("error.closedTradeWithNoDepositTx", trade.getShortId())));
                        }
                        return trade.getId();
                    })
                    .collect(Collectors.toSet()));

            if (tradeTxException.get() != null)
                throw tradeTxException.get();

            return tradesIdSet;
        }
    }

    // If trade still has funds locked up it might come back from failed trades
    // Aborts unfailing if the address entries needed are not available
    private boolean unFailTrade(Trade trade) {
        if (!recoverAddresses(trade)) {
            log.warn("Failed to recover address during unFail trade");
            return false;
        }

        initPersistedTrade(trade);

        synchronized (tradableList) {
            if (!tradableList.contains(trade)) {
                tradableList.add(trade);
            }
        }

        return true;
    }

    // The trade is added to pending trades if the associated address entries are AVAILABLE and
    // the relevant entries are changed, otherwise it's not added and no address entries are changed
    private boolean recoverAddresses(Trade trade) {
        // Find addresses associated with this trade.
        var entries = tradeUtil.getAvailableAddresses(trade);
        if (entries == null)
            return false;

        xmrWalletService.recoverAddressEntry(trade.getId(), entries.first,
                XmrAddressEntry.Context.MULTI_SIG);
        xmrWalletService.recoverAddressEntry(trade.getId(), entries.second,
                XmrAddressEntry.Context.TRADE_PAYOUT);
        return true;
    }


    ///////////////////////////////////////////////////////////////////////////////////////////
    // Getters, Utils
    ///////////////////////////////////////////////////////////////////////////////////////////

    public ObservableList<Trade> getObservableList() {
        synchronized (tradableList) {
            return tradableList.getObservableList();
        }
    }

    public BooleanProperty persistedTradesInitializedProperty() {
        return persistedTradesInitialized;
    }

    public boolean isMyOffer(Offer offer) {
        return offer.isMyOffer(keyRing);
    }

    public boolean wasOfferAlreadyUsedInTrade(String offerId) {
        return getOpenTrade(offerId).isPresent() ||
                failedTradesManager.getTradeById(offerId).isPresent() ||
                closedTradableManager.getTradableById(offerId).isPresent();
    }

    public boolean isBuyer(Offer offer) {
        // If I am the maker, we use the OfferDirection, otherwise the mirrored direction
        if (isMyOffer(offer))
            return offer.isBuyOffer();
        else
            return offer.getDirection() == OfferDirection.SELL;
    }

    // TODO (woodser): make Optional<Trade> versus Trade return types consistent
    public Trade getTrade(String tradeId) {
        return getOpenTrade(tradeId).orElseGet(() -> getClosedTrade(tradeId).orElseGet(() -> null));
    }

    public Optional<Trade> getOpenTrade(String tradeId) {
        synchronized (tradableList) {
            return tradableList.stream().filter(e -> e.getId().equals(tradeId)).findFirst();
        }
    }

    public Optional<Trade> getOpenTradeByUid(String tradeUid) {
        synchronized (tradableList) {
            return tradableList.stream().filter(e -> e.getUid().equals(tradeUid)).findFirst();
        }
    }

    public List<Trade> getAllTrades() {
        List<Trade> trades = new ArrayList<Trade>();
        trades.addAll(tradableList.getList());
        trades.addAll(closedTradableManager.getClosedTrades());
        trades.addAll(failedTradesManager.getObservableList());
        return trades;
    }

    public List<Trade> getOpenTrades() {
        synchronized (tradableList) {
            return ImmutableList.copyOf(getObservableList().stream()
                    .filter(e -> e instanceof Trade)
                    .map(e -> e)
                    .collect(Collectors.toList()));
        }
    }

    public Optional<Trade> getClosedTrade(String tradeId) {
        return closedTradableManager.getClosedTrades().stream().filter(e -> e.getId().equals(tradeId)).findFirst();
    }

    private void addTrade(Trade trade) {
        synchronized (tradableList) {
            if (tradableList.add(trade)) {
                requestPersistence();
            }
        }
    }

    private void removeTrade(Trade trade) {
        log.info("TradeManager.removeTrade() " + trade.getId());
        synchronized (tradableList) {
            if (!tradableList.contains(trade)) return;

            // remove trade
            tradableList.remove(trade);

            // unregister and persist
            p2PService.removeDecryptedDirectMessageListener(getTradeProtocol(trade));
            requestPersistence();
        }
    }

    private void maybeRemoveTradeOnError(Trade trade) {
        log.info("TradeManager.maybeRemoveTradeOnError() " + trade.getId());
        synchronized (tradableList) {
            if (!tradableList.contains(trade)) return;

            // unreserve taker key images
            if (trade instanceof TakerTrade && trade.getSelf().getReserveTxKeyImages() != null) {
                xmrWalletService.thawOutputs(trade.getSelf().getReserveTxKeyImages());
                xmrWalletService.saveMainWallet();
                trade.getSelf().setReserveTxKeyImages(null);
            }

            // remove trade if wallet deleted
            if (!trade.walletExists()) {
                removeTrade(trade);
                return;
            }

            // remove trade and wallet unless deposit requested without nack
            if (!trade.isDepositRequested() || trade.isDepositFailed()) {
                removeTrade(trade);
                if (trade.walletExists()) trade.deleteWallet();
            } else {
                scheduleDeletionIfUnfunded(trade);
            }
        }
    }

    private void scheduleDeletionIfUnfunded(Trade trade) {
        if (getOpenTrade(trade.getId()).isPresent() && trade.isDepositRequested() && !trade.isDepositsPublished()) {
            log.warn("Scheduling to delete open trade if unfunded for {} {}", trade.getClass().getSimpleName(), trade.getId());
            UserThread.runAfter(() -> {
                if (isShutDown) return;

                // get trade's deposit txs from daemon
                MoneroTx makerDepositTx = trade.getMaker().getDepositTxHash() == null ? null : xmrWalletService.getDaemon().getTx(trade.getMaker().getDepositTxHash());
                MoneroTx takerDepositTx = trade.getTaker().getDepositTxHash() == null ? null : xmrWalletService.getDaemon().getTx(trade.getTaker().getDepositTxHash());

                // delete multisig trade wallet if neither deposit tx published
                if (makerDepositTx == null && takerDepositTx == null) {
                    log.warn("Deleting {} {} after protocol error", trade.getClass().getSimpleName(), trade.getId());
                    removeTrade(trade);
                    failedTradesManager.removeTrade(trade);
                    if (trade.walletExists()) trade.deleteWallet();
                } else {
                    log.warn("Refusing to delete {} {} after protocol timeout because its wallet might be funded", trade.getClass().getSimpleName(), trade.getId());
                }
            }, 60);
        }
    }

    // TODO Remove once tradableList is refactored to a final field
    //  (part of the persistence refactor PR)
    private void onTradesChanged() {
        this.numPendingTrades.set(getObservableList().size());
    }
}
