Adjusted moving funds handling for the new wallet coordination

pkg/chain/ethereum/tbtc.go

@@ -1,7 +1,6 @@
 package ethereum
 
 import (
-	"context"
 	"crypto/ecdsa"
 	"crypto/elliptic"
 	"encoding/binary"
@@ -1913,32 +1912,6 @@ func (tc *TbtcChain) OnRedemptionProposalSubmitted(
 		OnEvent(onEvent)
 }
 
-func (tc *TbtcChain) OnMovingFundsCommitmentInitiated(handler func(uint64)) subscription.EventSubscription {
-	// TODO: Replace this function with a real trigger of moving funds commitment.
-	ctx, cancelCtx := context.WithCancel(context.Background())
-	blocksChan := tc.blockCounter.WatchBlocks(ctx)
-
-	go func() {
-		for {
-			select {
-			case blockNumber := <-blocksChan:
-				// Generate an event every 500th block.
-				logger.Infof("Received block %d", blockNumber)
-
-				if blockNumber%500 == 0 {
-					go handler(blockNumber)
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
-
-	return subscription.NewEventSubscription(func() {
-		cancelCtx()
-	})
-}
-
 func (tc *TbtcChain) SubmitMovingFundsCommitment(
 	walletPublicKeyHash [20]byte,
 	walletMainUTXO bitcoin.UnspentTransactionOutput,

pkg/tbtc/chain.go

@@ -367,12 +367,6 @@ type WalletCoordinatorChain interface {
 		func(event *RedemptionProposalSubmittedEvent),
 	) subscription.EventSubscription
 
-	// OnMovingFundsCommitmentInitiated registers a callback that is invoked
-	// when moving funds commitment is initiated.
-	OnMovingFundsCommitmentInitiated(
-		func(blockNumber uint64),
-	) subscription.EventSubscription
-
 	// Submits the moving funds target wallets commitment.
 	SubmitMovingFundsCommitment(
 		walletPublicKeyHash [20]byte,

pkg/tbtc/chain_test.go

@@ -760,10 +760,6 @@ func (lc *localChain) OnRedemptionProposalSubmitted(
 	panic("unsupported")
 }
 
-func (lc *localChain) OnMovingFundsCommitmentInitiated(handler func(uint64)) subscription.EventSubscription {
-	panic("unsupported")
-}
-
 func (lc *localChain) SubmitMovingFundsCommitment(
 	walletPublicKeyHash [20]byte,
 	walletMainUTXO bitcoin.UnspentTransactionOutput,

pkg/tbtc/moving_funds.go

@@ -0,0 +1,5 @@
+package tbtc
+
+type MovingFundsProposal struct {
+	WalletPublicKeyHash [20]byte
+}

pkg/tbtc/node.go

@@ -6,11 +6,11 @@ import (
 	"encoding/hex"
 	"fmt"
 	"math/big"
+	"sort"
 	"sync"
 	"time"
 
 	"github.com/keep-network/keep-core/pkg/bitcoin"
-
 	"github.com/keep-network/keep-core/pkg/chain"
 
 	"go.uber.org/zap"
@@ -643,6 +643,170 @@ func (n *node) handleRedemptionProposal(
 	walletActionLogger.Infof("wallet action dispatched successfully")
 }
 
+func (n *node) handleMovingFundsProposal(proposal *MovingFundsProposal) {
+	go func() {
+		sourceWalletPublicKeyHash := proposal.WalletPublicKeyHash
+		logger.Info(
+			"moving funds proposal initiated for wallet with PKH [0x%x]",
+			sourceWalletPublicKeyHash,
+		)
+
+		// Make sure the wallet meets the criteria for moving funds proposal.
+		_, found := n.walletRegistry.getWalletByPublicKeyHash(
+			sourceWalletPublicKeyHash,
+		)
+		if !found {
+			logger.Errorf(
+				"skipping moving funds proposal for wallet with PKH"+
+					"[0x%x] as the node does not control it",
+				sourceWalletPublicKeyHash,
+			)
+			return
+		}
+
+		sourceWalletChainData, err := n.chain.GetWallet(sourceWalletPublicKeyHash)
+		if err != nil {
+			logger.Errorf(
+				"failed to get wallet data for source wallet with PKH "+
+					"[0x%x]: [%v]",
+				sourceWalletPublicKeyHash,
+				err,
+			)
+			return
+		}
+
+		walletMainUtxo, err := DetermineWalletMainUtxo(
+			sourceWalletPublicKeyHash,
+			n.chain,
+			n.btcChain,
+		)
+		if err != nil {
+			logger.Errorf(
+				"skipping moving funds proposal for wallet with PKH "+
+					"[0x%x] due to error determining wallet main UTXO: [%v]",
+				sourceWalletPublicKeyHash,
+				err,
+			)
+			return
+		}
+
+		walletBalance := walletMainUtxo.Value
+
+		// Check if the wallet meets the condition for moving funds
+		// commitment.
+		if sourceWalletChainData.State != StateMovingFunds ||
+			sourceWalletChainData.PendingRedemptionsValue > 0 ||
+			sourceWalletChainData.PendingMovedFundsSweepRequestsCount > 0 ||
+			sourceWalletChainData.MovingFundsTargetWalletsCommitmentHash != [32]byte{} ||
+			walletBalance <= 0 {
+			logger.Errorf(
+				"skipping moving funds proposal for wallet with PKH "+
+					"[0x%x] due to unmet conditions",
+				sourceWalletPublicKeyHash,
+			)
+			return
+		}
+
+		logger.Infof(
+			"proceeding with moving funds commitment for wallet with "+
+				"PKH [0x%x]",
+			sourceWalletPublicKeyHash,
+		)
+
+		// Retrieve all the live wallets.
+		liveWalletsCount, err := n.chain.GetLiveWalletsCount()
+		if err != nil {
+			logger.Errorf("failed to get live wallets count: [%v]", err)
+			return
+		}
+
+		if liveWalletsCount == 0 {
+			logger.Infof(
+				"skipping moving funds proposal due to lack of live wallets",
+			)
+			return
+		}
+
+		events, err := n.chain.PastNewWalletRegisteredEvents(nil)
+		if err != nil {
+			logger.Errorf(
+				"failed to get past new wallet registered events: [%v]",
+				err,
+			)
+			return
+		}
+
+		liveWallets := make([][20]byte, 0)
+
+		for _, event := range events {
+			walletPubKeyHash := event.WalletPublicKeyHash
+			wallet, err := n.chain.GetWallet(walletPubKeyHash)
+			if err != nil {
+				logger.Errorf(
+					"failed to get wallet data for wallet with PKH [0x%x]: [%v]",
+					walletPubKeyHash,
+					err,
+				)
+				continue
+			}
+			if wallet.State == StateLive {
+				liveWallets = append(liveWallets, walletPubKeyHash)
+			}
+		}
+
+		// Make sure all the live wallets data has been retrieved by
+		// comparing the number of live wallets to the on-chain counter.
+		if len(liveWallets) != int(liveWalletsCount) {
+			logger.Errorf(
+				"mismatch between the number of retrieved live wallets "+
+					"and the on-chain live wallet count [%v:%v]",
+				len(liveWallets),
+				int(liveWalletsCount),
+			)
+			return
+		}
+
+		// Sort the live wallets according to their numerical representation
+		// as the on-chain contract expects.
+		sort.Slice(liveWallets, func(i, j int) bool {
+			bigIntI := new(big.Int).SetBytes(liveWallets[i][:])
+			bigIntJ := new(big.Int).SetBytes(liveWallets[j][:])
+			return bigIntI.Cmp(bigIntJ) < 0
+		})
+
+		logger.Infof("found [%v] live wallets", len(liveWallets))
+
+		_, _, _, _, _, walletMaxBtcTransfer, _, _ := n.chain.GetWalletParameters()
+
+		ceilingDivide := func(x, y uint64) uint64 {
+			return (x + y - 1) / y
+		}
+		min := func(x, y uint64) uint64 {
+			if x < y {
+				return x
+			}
+			return y
+		}
+
+		targetWalletsCount := min(
+			uint64(liveWalletsCount),
+			ceilingDivide(uint64(walletBalance), walletMaxBtcTransfer),
+		)
+
+		targetWallets := liveWallets[0:targetWalletsCount]
+		logger.Infof("Target wallets length [%v]", len(targetWallets))
+
+		// TODO: Continue with moving funds commitment
+
+		logger.Infof(
+			"Finished moving funds commitment for wallet with PKH [0x%x]",
+			sourceWalletPublicKeyHash,
+		)
+
+		// TODO: Add construction of the move funds Bitcoin transaction.
+	}()
+}
+
 // coordinationLayerSettings represents settings for the coordination layer.
 type coordinationLayerSettings struct {
 	// executeCoordinationProcedureFn is a function executing the coordination