(#1919) Migrate to AddWatchedAddresses (#1920)

* Migrate to AddWatchedAddresses
* Use variadic parameters
* Vendored fixes
* Re-watch addresses on socket restart
* Convert Mutex to RWMutex
* Protect with listenLock
* Update to use bulk call
* Eliminate redundant conversion
* Move statements into the socketclient section
* Remove duplicate lock
* Refactored to be closer to it’s usage
* DB Transaction for Putting PubKeys
* Fix linter errors
* Migrate to PutAll for watched scripts
* Fix missing initialization of err
* Interface check for implemented wallets
* Handle errors from transaction; Report Rollback failures
* Append to BlockBookClient.listenAddrs instead of replace
* Fix race bug and remove redundant replay handling in c/b/client.go

Co-authored-by: Mike Greenberg <mikeg@ob1.io>

Author: drwasho

core/images.go

@@ -168,8 +168,7 @@ func (n *OpenBazaarNode) FetchImage(peerID string, imageType string, size string
 func (n *OpenBazaarNode) GetBase64Image(url string) (base64ImageData, filename string, err error) {
 	var client *http.Client
 	if n.TorDialer != nil {
-		dial := n.TorDialer.Dial
-		tbTransport := &http.Transport{Dial: dial}
+		tbTransport := &http.Transport{Dial: n.TorDialer.Dial}
 		client = &http.Client{Transport: tbTransport, Timeout: time.Second * 30}
 	} else {
 		client = &http.Client{Timeout: time.Second * 30}

core/order.go

@@ -278,7 +278,7 @@ func prepareModeratedOrderContract(data *repo.PurchaseData, n *OpenBazaarNode, c
 	fee := wal.GetFeePerByte(wallet.NORMAL)
 	contract.BuyerOrder.BigRefundFee = fee.String()
 
-	err = wal.AddWatchedAddress(addr)
+	err = wal.AddWatchedAddresses(addr)
 	if err != nil {
 		return nil, err
 	}
@@ -315,7 +315,7 @@ func processOnlineDirectOrder(resp *pb.Message, n *OpenBazaarNode, wal wallet.Wa
 	if err != nil {
 		return "", "", *big.NewInt(0), false, err
 	}
-	err = wal.AddWatchedAddress(addr)
+	err = wal.AddWatchedAddresses(addr)
 	if err != nil {
 		return "", "", *big.NewInt(0), false, err
 	}
@@ -374,7 +374,7 @@ func processOfflineDirectOrder(n *OpenBazaarNode, wal wallet.Wallet, contract *p
 	payment.RedeemScript = hex.EncodeToString(redeemScript)
 	payment.Chaincode = hex.EncodeToString(chaincode)
 
-	err = wal.AddWatchedAddress(addr)
+	err = wal.AddWatchedAddresses(addr)
 	if err != nil {
 		return "", "", *big.NewInt(0), err
 	}

net/retriever/retriever.go

@@ -69,8 +69,7 @@ type offlineMessage struct {
 func NewMessageRetriever(cfg MRConfig) *MessageRetriever {
 	var client *http.Client
 	if cfg.Dialer != nil {
-		dial := cfg.Dialer.Dial
-		tbTransport := &http.Transport{Dial: dial}
+		tbTransport := &http.Transport{Dial: cfg.Dialer.Dial}
 		client = &http.Client{Transport: tbTransport, Timeout: time.Second * 30}
 	} else {
 		client = &http.Client{Timeout: time.Second * 30}

net/service/handlers.go

@@ -414,17 +414,9 @@ func (service *OpenBazaarService) handleOrder(peer peer.ID, pmes *pb.Message, op
 		if err != nil {
 			return errorResponse(err.Error()), err
 		}
-		err = wal.AddWatchedAddress(addr)
-		if err != nil {
-			log.Error(err)
-		}
-		log.Debugf("added address to wallet to watch: %s", addr)
-		log.Debugf("storing sales order %s in database", orderId)
-		err = service.node.Datastore.Sales().Put(orderId, *contract, pb.OrderState_AWAITING_PAYMENT, false)
-		if err != nil {
-			log.Error(err)
-		}
-		log.Debugf("successfully processed direct ORDER message from %s", peer.Pretty())
+		wal.AddWatchedAddresses(addr)
+		service.node.Datastore.Sales().Put(orderId, *contract, pb.OrderState_AWAITING_PAYMENT, false)
+		log.Debugf("Received direct ORDER message from %s", peer.Pretty())
 		return nil, nil
 	} else if order.Payment.Method == pb.Order_Payment_MODERATED && !offline {
 		log.Debugf("processing moderated online order from %s", peer.Pretty())
@@ -451,7 +443,8 @@ func (service *OpenBazaarService) handleOrder(peer peer.ID, pmes *pb.Message, op
 		if err != nil {
 			return errorResponse(err.Error()), err
 		}
-		err = wal.AddWatchedAddress(addr)
+		wal.AddWatchedAddresses(addr)
+		contract, err = service.node.NewOrderConfirmation(contract, false, false)
 		if err != nil {
 			log.Error(err)
 		}
@@ -492,16 +485,9 @@ func (service *OpenBazaarService) handleOrder(peer peer.ID, pmes *pb.Message, op
 			log.Error(err)
 			return errorResponse(err.Error()), err
 		}
-		err = wal.AddWatchedAddress(addr)
-		if err != nil {
-			log.Error(err)
-		}
-		log.Debugf("storing sales order %s in database", orderId)
-		err = service.node.Datastore.Sales().Put(orderId, *contract, pb.OrderState_AWAITING_PAYMENT, false)
-		if err != nil {
-			log.Error(err)
-		}
-		log.Debugf("successfully processed offline moderated ORDER message from %s", peer.Pretty())
+		wal.AddWatchedAddresses(addr)
+		log.Debugf("Received offline moderated ORDER message from %s", peer.Pretty())
+		service.node.Datastore.Sales().Put(orderId, *contract, pb.OrderState_AWAITING_PAYMENT, false)
 		return nil, nil
 	}
 	log.Errorf("Unrecognized payment type on order (%s)", contract.VendorOrderConfirmation.OrderID)

repo/db/watched_scripts.go

@@ -19,6 +19,37 @@ func NewWatchedScriptStore(db *sql.DB, lock *sync.Mutex, coinType wallet.CoinTyp
 	return &WatchedScriptsDB{modelStore{db, lock}, coinType}
 }
 
+func (w *WatchedScriptsDB) PutAll(scriptPubKeys [][]byte) error {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+
+	tx, err := w.db.Begin()
+	if err != nil {
+		return err
+	}
+
+	stmt, err := tx.Prepare("insert or replace into watchedscripts(coin, scriptPubKey) values(?,?)")
+	if err != nil {
+		if rErr := tx.Rollback(); rErr != nil {
+			return fmt.Errorf("put AND rollback failed: %s (rollback error: %s)", err.Error(), rErr.Error())
+		}
+		return err
+	}
+	defer stmt.Close()
+
+	for _, scriptPubKey := range scriptPubKeys {
+		_, err = stmt.Exec(w.coinType.CurrencyCode(), hex.EncodeToString(scriptPubKey))
+		if err != nil {
+			if rErr := tx.Rollback(); rErr != nil {
+				return fmt.Errorf("put AND rollback failed: %s (rollback error: %s)", err.Error(), rErr.Error())
+			}
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
 func (w *WatchedScriptsDB) Put(scriptPubKey []byte) error {
 	w.lock.Lock()
 	defer w.lock.Unlock()