Prevent the wallet from creating duplicate UTXOs (#4290)

* Prevent the wallet from creating duplicate UTXOs

* respond to the PR comments

* cleanup

core/src/main/scala/org/bitcoins/core/api/wallet/CoinSelectionAlgo.scala

@@ -1,6 +1,6 @@
 package org.bitcoins.core.api.wallet
 
-import org.bitcoins.crypto.StringFactory
+import org.bitcoins.crypto.{DoubleSha256Digest, StringFactory}
 
 /** Represents the various ways the wallet can do coin selection */
 sealed abstract class CoinSelectionAlgo
@@ -30,6 +30,9 @@ object CoinSelectionAlgo extends StringFactory[CoinSelectionAlgo] {
   /** Tries all coin selection algos and uses the one with the least waste */
   case object LeastWaste extends CoinSelectionAlgo
 
+  case class SelectedUtxos(outPoints: Set[(DoubleSha256Digest, Int)])
+      extends CoinSelectionAlgo
+
   /** Coin selection algos that don't call other ones */
   val independentAlgos: Vector[CoinSelectionAlgo] =
     Vector(RandomSelection,

core/src/main/scala/org/bitcoins/core/api/wallet/CoinSelector.scala

@@ -146,6 +146,25 @@ object CoinSelector extends CoinSelector {
             throw new IllegalArgumentException(
               "longTermFeeRateOpt must be defined for LeastWaste")
         }
+      case SelectedUtxos(outPoints) =>
+        val result = walletUtxos.foldLeft(Vector.empty[SpendingInfoDb]) {
+          (acc, utxo) =>
+            val outPoint = (utxo.outPoint.txId, utxo.outPoint.vout.toInt)
+            if (outPoints(outPoint)) acc :+ utxo else acc
+        }
+        if (result.toSet.size < outPoints.size) {
+          outPoints.foreach { outPoint =>
+            if (
+              !result.exists(utxo =>
+                utxo.outPoint.txId == outPoint._1 && utxo.outPoint.vout.toInt == outPoint._2)
+            )
+              throw new IllegalArgumentException(
+                s"Unknown UTXO: ${outPoint._1}:${outPoint._2}")
+          }
+        } else if (result.size > outPoints.size) {
+          throw new IllegalArgumentException(s"Found too many UTXOs")
+        }
+        result
     }
 
   private case class CoinSelectionResults(

wallet-test/src/test/scala/org/bitcoins/wallet/UTXOLifeCycleTest.scala

@@ -1,6 +1,7 @@
 package org.bitcoins.wallet
 
 import grizzled.slf4j.Logging
+import org.bitcoins.core.api.wallet.CoinSelectionAlgo
 import org.bitcoins.core.api.wallet.db.SpendingInfoDb
 import org.bitcoins.core.currency.{Bitcoins, Satoshis}
 import org.bitcoins.core.number._
@@ -570,8 +571,12 @@ class UTXOLifeCycleTest
         _ <- wallet.processBlock(block)
         broadcastSpentUtxo <- wallet.listUtxos(
           TxoState.PendingConfirmationsSpent)
+        pendingConfirmationsReceivedUtxos <- wallet.listUtxos(
+          TxoState.PendingConfirmationsReceived)
         finalReservedUtxos <- wallet.listUtxos(TxoState.Reserved)
       } yield {
+        assert(newReservedUtxos == finalReservedUtxos)
+        assert(pendingConfirmationsReceivedUtxos.isEmpty)
         assert(broadcastSpentUtxo.length == 1)
         //make sure spendingTxId got set correctly
         assert(broadcastSpentUtxo.head.spendingTxIdOpt.get == txIdBE)
@@ -579,4 +584,123 @@ class UTXOLifeCycleTest
         assert(finalReservedUtxos.length == newReservedUtxos.length)
       }
   }
+
+  it should "track a utxo state change to broadcast spent and then to pending confirmations received (the spend transaction gets confirmed together with the receive transaction))" in {
+    param =>
+      val WalletWithBitcoindRpc(wallet, bitcoind) = param
+
+      val receiveValue = Bitcoins(8)
+      val sendValue = Bitcoins(4)
+
+      for {
+        // receive a UTXO from an external wallet
+        walletAddress <- wallet.getNewAddress()
+        txId <- bitcoind.sendToAddress(walletAddress, receiveValue)
+        receiveTx <- bitcoind.getRawTransactionRaw(txId)
+        _ <- wallet.processTransaction(receiveTx, None)
+        receiveOutPointPair = receiveTx.outputs.zipWithIndex
+          .find(_._1.value == receiveValue)
+          .map(out => (receiveTx.txId, out._2))
+          .get
+
+        receiveOutPoint = TransactionOutPoint(receiveOutPointPair._1,
+                                              UInt32(receiveOutPointPair._2))
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+        _ = assert(receivedUtxo.size == 1)
+        _ = assert(receivedUtxo.head.state == BroadcastReceived)
+
+        // spend and broadcast unconfirmed
+        feeRate <- wallet.feeRateApi.getFeeRate()
+        sendTx <- wallet.sendWithAlgo(
+          testAddr,
+          sendValue,
+          feeRate,
+          CoinSelectionAlgo.SelectedUtxos(Set(receiveOutPointPair)))
+        _ <- wallet.broadcastTransaction(sendTx)
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+        _ = assert(receivedUtxo.size == 1)
+        _ = assert(receivedUtxo.head.state == BroadcastSpent)
+
+        // confirm receive and spend
+        blockHashes <- bitcoind.getNewAddress.flatMap(
+          bitcoind.generateToAddress(1, _))
+        block <- bitcoind.getBlockRaw(blockHashes.head)
+        _ <- wallet.processBlock(block)
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+      } yield {
+        assert(receivedUtxo.size == 1)
+        assert(receivedUtxo.head.state == PendingConfirmationsSpent)
+      }
+  }
+
+  it should "track a utxo state change to broadcast spent and then to pending confirmations received (the spend transaction gets confirmed after the receive transaction))" in {
+    param =>
+      val WalletWithBitcoindRpc(wallet, bitcoind) = param
+
+      val receiveValue = Bitcoins(8)
+      val sendValue = Bitcoins(4)
+
+      for {
+        // receive a UTXO from an external wallet
+        walletAddress <- wallet.getNewAddress()
+        txId <- bitcoind.sendToAddress(walletAddress, receiveValue)
+        receiveTx <- bitcoind.getRawTransactionRaw(txId)
+        _ <- wallet.processTransaction(receiveTx, None)
+        receiveOutPointPair = receiveTx.outputs.zipWithIndex
+          .find(_._1.value == receiveValue)
+          .map(out => (receiveTx.txId, out._2))
+          .get
+
+        receiveOutPoint = TransactionOutPoint(receiveOutPointPair._1,
+                                              UInt32(receiveOutPointPair._2))
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+        _ = assert(receivedUtxo.size == 1)
+        _ = assert(receivedUtxo.head.state == BroadcastReceived)
+
+        // spend unconfirmed
+        feeRate <- wallet.feeRateApi.getFeeRate()
+        sendTx <- wallet.sendWithAlgo(
+          testAddr,
+          sendValue,
+          feeRate,
+          CoinSelectionAlgo.SelectedUtxos(Set(receiveOutPointPair)))
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+        _ = assert(receivedUtxo.size == 1)
+        _ = assert(receivedUtxo.head.state == BroadcastSpent)
+
+        // confirm receive
+        blockHashes <- bitcoind.getNewAddress.flatMap(
+          bitcoind.generateToAddress(1, _))
+        block <- bitcoind.getBlockRaw(blockHashes.head)
+        _ <- wallet.processBlock(block)
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+        _ = assert(receivedUtxo.size == 1)
+        _ = assert(receivedUtxo.head.state == BroadcastSpent)
+
+        // broadcast and confirm spend
+        _ <- wallet.broadcastTransaction(sendTx)
+        blockHashes <- bitcoind.getNewAddress.flatMap(
+          bitcoind.generateToAddress(1, _))
+        block <- bitcoind.getBlockRaw(blockHashes.head)
+        _ <- wallet.processBlock(block)
+
+        receivedUtxo <- wallet.spendingInfoDAO.findByOutPoints(
+          Vector(receiveOutPoint))
+      } yield {
+        assert(receivedUtxo.size == 1)
+        assert(receivedUtxo.head.state == PendingConfirmationsSpent)
+      }
+  }
 }

wallet/src/main/scala/org/bitcoins/wallet/internal/UtxoHandling.scala

@@ -280,7 +280,11 @@ private[wallet] trait UtxoHandling extends WalletLogger {
 
     for {
       utxo <- utxoF
-      written <- spendingInfoDAO.create(utxo)
+      written <- spendingInfoDAO.createUnless(utxo) {
+        (foundUtxo, utxoToCreate) =>
+          foundUtxo.state.isInstanceOf[SpentState] && utxoToCreate.state
+            .isInstanceOf[ReceivedState]
+      }
     } yield {
       val writtenOut = written.outPoint
       logger.info(

wallet/src/main/scala/org/bitcoins/wallet/models/SpendingInfoDAO.scala

@@ -47,10 +47,63 @@ case class SpendingInfoDAO()(implicit
   }
 
   def create(si: SpendingInfoDb): Future[SpendingInfoDb] = {
+    val actions = for {
+      utxo: UTXORecord <- insertAction(si)
+      spk <-
+        spkTable
+          .filter(_.id === utxo.scriptPubKeyId)
+          .result
+          .headOption
+    } yield (utxo, spk)
+
+    safeDatabase
+      .run(actions)
+      .map {
+        case (utxo, Some(spk)) => utxo.toSpendingInfoDb(spk.scriptPubKey)
+        case _ =>
+          throw new SQLException(
+            s"Unexpected result: Cannot create either a UTXO or a SPK record for $si")
+      }
+  }
+
+  def createUnless(si: SpendingInfoDb)(
+      condition: (UTXORecord, UTXORecord) => Boolean): Future[
+    SpendingInfoDb] = {
+    val actions = for {
+      foundOpt <- table.filter(_.outPoint === si.outPoint).result.headOption
+      cond <- foundOpt match {
+        case Some(foundUtxo) =>
+          val utxoToCreate =
+            UTXORecord.fromSpendingInfoDb(si, foundUtxo.scriptPubKeyId)
+          DBIO.successful(condition(foundUtxo, utxoToCreate))
+        case None => DBIO.successful(false)
+      }
+      utxo <-
+        if (cond) DBIO.successful(foundOpt.get) else insertAction(si)
+      spk <-
+        spkTable
+          .filter(_.id === utxo.scriptPubKeyId)
+          .result
+          .headOption
+    } yield (utxo, spk)
+
+    safeDatabase
+      .run(actions)
+      .map {
+        case (utxo, Some(spk)) => utxo.toSpendingInfoDb(spk.scriptPubKey)
+        case _ =>
+          throw new SQLException(
+            s"Unexpected result: Cannot create either a UTXO or a SPK record for $si")
+      }
+  }
+
+  private def insertAction(si: SpendingInfoDb): DBIOAction[
+    UTXORecord,
+    NoStream,
+    Effect.Read with Effect.Write] = {
     val query =
       table.returning(table.map(_.id)).into((t, id) => t.copyWithId(id = id))
-
+    for {
-    val actions = for {
       spkOpt <-
         spkTable
           .filter(_.scriptPubKey === si.output.scriptPubKey)
@@ -70,21 +123,7 @@ case class SpendingInfoDAO()(implicit
             query += utxo
           }).flatten
       }
-      spk <-
+    } yield utxo
-        spkTable
-          .filter(_.id === utxo.scriptPubKeyId)
-          .result
-          .headOption
-    } yield (utxo, spk)
-
-    safeDatabase
-      .run(actions)
-      .map {
-        case (utxo, Some(spk)) => utxo.toSpendingInfoDb(spk.scriptPubKey)
-        case _ =>
-          throw new SQLException(
-            s"Unexpected result: Cannot create either a UTXO or a SPK record for $si")
-      }
   }
 
   def upsertAllSpendingInfoDb(
@@ -455,6 +494,11 @@ case class SpendingInfoDAO()(implicit
     safeDatabase.runVec(query.result).map(_.toVector)
   }
 
+  def findByOutPoint(
+      outPoint: TransactionOutPoint): Future[Option[SpendingInfoDb]] = {
+    findByOutPoints(Vector(outPoint)).map(_.headOption)
+  }
+
   /** Enumerates all TX outpoints in the wallet */
   def findByOutPoints(outPoints: Vector[TransactionOutPoint]): Future[
     Vector[SpendingInfoDb]] = {