Execute payout transactions

CHANGELOG.md

@@ -11,6 +11,7 @@
 - Expose stats endpoints [\#114](https://github.com/NodeFactoryIo/vedran/pull/114) ([MakMuftic](https://github.com/MakMuftic))
 - Calculating reward distribution [\#124](https://github.com/NodeFactoryIo/vedran/pull/124) ([MakMuftic](https://github.com/MakMuftic))
 - Add payout CLI command [\#126](https://github.com/NodeFactoryIo/vedran/pull/126) ([MakMuftic](https://github.com/MakMuftic))
+- Execute payout transactions [\#127](https://github.com/NodeFactoryIo/vedran/pull/127) ([MakMuftic](https://github.com/MakMuftic))
 
 ### Fix
 

cmd/payout.go

@@ -65,10 +65,15 @@ func init() {
 func payoutCommand(_ *cobra.Command, _ []string) {
 	DisplayBanner()
 	fmt.Println("Payout script running...")
-	err := script.ExecutePayout(secret, totalRewardAsFloat64, loadbalancerURL)
+	transactions, err := script.ExecutePayout(secret, totalRewardAsFloat64, loadbalancerURL)
+	if transactions != nil {
+		// display even if only part of transactions executed
+		DisplayTransactionsStatus(transactions)
+	}
 	if err != nil {
 		log.Errorf("Unable to execute payout, because of: %v", err)
 		return
+	} else {
+		log.Info("Payout execution finished")
 	}
-	log.Info("Payout execution finished")
 }

cmd/table.go

@@ -0,0 +1,18 @@
+package cmd
+
+import (
+	"fmt"
+	"github.com/NodeFactoryIo/vedran/internal/payout"
+	"github.com/gosuri/uitable"
+)
+
+func DisplayTransactionsStatus(transactions []*payout.TransactionDetails) {
+	table := uitable.New()
+	table.MaxColWidth = 80
+	table.Wrap = true
+	table.AddRow("ID (node)", "To", "Amount", "Status")
+	for _, tx := range transactions {
+		table.AddRow(tx.NodeId, tx.To, tx.Amount.String(), tx.Status)
+	}
+	fmt.Println(table)
+}

go.mod

@@ -3,14 +3,25 @@ module github.com/NodeFactoryIo/vedran
 go 1.15
 
 require (
+	github.com/NodeFactoryIo/go-substrate-rpc-client v1.1.1-0.20201116144250-e4075b704b1c
 	github.com/asdine/storm/v3 v3.2.1
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
+	github.com/fatih/color v1.3.0 // indirect
 	github.com/golang/gddo v0.0.0-20200831202555-721e228c7686
+	github.com/golang/protobuf v1.4.2 // indirect
+	github.com/golang/snappy v0.0.2-0.20200707131729-196ae77b8a26 // indirect
 	github.com/gorilla/mux v1.8.0
+	github.com/gosuri/uitable v0.0.4
 	github.com/inconshreveable/go-vhost v0.0.0-20160627193104-06d84117953b
+	github.com/mattn/go-colorable v0.1.0 // indirect
+	github.com/mattn/go-isatty v0.0.5-0.20180830101745-3fb116b82035 // indirect
+	github.com/mattn/go-runewidth v0.0.4 // indirect
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/cobra v1.0.0
 	github.com/stretchr/testify v1.6.1
-	golang.org/x/net v0.0.0-20200625001655-4c5254603344
+	golang.org/x/net v0.0.0-20200822124328-c89045814202
+	golang.org/x/sys v0.0.0-20200824131525-c12d262b63d8 // indirect
+	golang.org/x/text v0.3.3 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 )

internal/payout/payout.go

@@ -39,7 +39,7 @@ func GetStatsForPayout(
 			PaymentDetails: statistics,
 		})
 		if err != nil {
-			log.Errorf("Unable to save payout information to database, because of: %v", err)
+			log.Errorf("Unable to save payout information To database, because of: %v", err)
 			return nil, err
 		}
 	}

internal/payout/transaction.go

@@ -0,0 +1,202 @@
+package payout
+
+import (
+	gsrpc "github.com/NodeFactoryIo/go-substrate-rpc-client"
+	"github.com/NodeFactoryIo/go-substrate-rpc-client/signature"
+	"github.com/NodeFactoryIo/go-substrate-rpc-client/types"
+	log "github.com/sirupsen/logrus"
+	"math/big"
+	"sync"
+)
+
+type TransactionStatus string
+
+const (
+	Finalized = TransactionStatus("Finalized")
+	Dropped   = TransactionStatus("Dropped")
+	Invalid   = TransactionStatus("Invalid")
+)
+
+type TransactionDetails struct {
+	NodeId string
+	To     string
+	Amount big.Int
+	Status TransactionStatus
+}
+
+func ExecuteAllPayoutTransactions(
+	payoutDistribution map[string]big.Int,
+	payoutDetails map[string]NodePayoutDetails,
+	secret string,
+	substrateRPCurl string,
+) ([]*TransactionDetails, error) {
+	api, err := gsrpc.NewSubstrateAPI(substrateRPCurl)
+	if err != nil {
+		return nil, err
+	}
+
+	keyringPair, err := signature.KeyringPairFromSecret(secret, "")
+	if err != nil {
+		return nil, err
+	}
+
+	var wg sync.WaitGroup
+	var mux sync.Mutex
+
+	resultsChannel := make(chan *TransactionDetails, len(payoutDistribution))
+	fatalErrorsChannel := make(chan error)
+	wgDoneChannel := make(chan bool, 1)
+
+	wg.Add(len(payoutDistribution)) // define number of goroutines
+	for nodeId, amount := range payoutDistribution {
+		nId := nodeId
+		go func(to string, amount big.Int, wg *sync.WaitGroup, mux *sync.Mutex) {
+			defer wg.Done()
+			transactionDetails, err := executeTransaction(api, nId, to, amount, keyringPair, mux)
+			if err != nil {
+				fatalErrorsChannel <- err
+			} else {
+				resultsChannel <- transactionDetails
+			}
+		}(payoutDetails[nodeId].PayoutAddress, amount, &wg, &mux)
+	}
+
+	go func() {
+		// wait for group To finish
+		wg.Wait()
+		close(wgDoneChannel)
+		close(resultsChannel)
+	}()
+
+	var transactionDetails []*TransactionDetails
+	var fatalErr error
+	select {
+	case <-wgDoneChannel:
+		break
+	case fatalErr = <-fatalErrorsChannel:
+		break
+	}
+	// return even if just some of transaction have been executed
+	for result := range resultsChannel {
+		transactionDetails = append(transactionDetails, result)
+	}
+	return transactionDetails, fatalErr
+}
+
+func executeTransaction(
+	api *gsrpc.SubstrateAPI,
+	nodeId string,
+	to string,
+	amount big.Int,
+	keyringPair signature.KeyringPair,
+	mux *sync.Mutex,
+) (*TransactionDetails, error) {
+	// lock segment so goroutines don't access api at the same time
+	mux.Lock()
+
+	metadataLatest, err := api.RPC.State.GetMetadataLatest()
+	if err != nil {
+		return nil, err
+	}
+
+	toAddress, err := types.NewAddressFromHexAccountID(to)
+	if err != nil {
+		return nil, err
+	}
+
+	call, err := types.NewCall(
+		metadataLatest,
+		"Balances.transfer",
+		toAddress,
+		types.NewUCompact(&amount),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	extrinsic := types.NewExtrinsic(call)
+
+	genesisHash, err := api.RPC.Chain.GetBlockHash(0)
+	if err != nil {
+		return nil, err
+	}
+
+	runtimeVersionLatest, err := api.RPC.State.GetRuntimeVersionLatest()
+	if err != nil {
+		return nil, err
+	}
+
+	storageKey, err := types.CreateStorageKey(
+		metadataLatest,
+		"System",
+		"AccountNonce",
+		keyringPair.PublicKey,
+		nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	var nonce uint32
+	_, err = api.RPC.State.GetStorageLatest(storageKey, nonce)
+	if err != nil {
+		return nil, err
+	}
+
+	signatureOptions := types.SignatureOptions{
+		Era:         types.ExtrinsicEra{IsMortalEra: false},
+		Nonce:       types.NewUCompactFromUInt(uint64(nonce)),
+		Tip:         types.NewUCompactFromUInt(0),
+		SpecVersion: runtimeVersionLatest.SpecVersion,
+		GenesisHash: genesisHash,
+		BlockHash:   genesisHash,
+	}
+
+	err = extrinsic.Sign(keyringPair, signatureOptions)
+	if err != nil {
+		return nil, err
+	}
+
+	sub, err := api.RPC.Author.SubmitAndWatchExtrinsic(extrinsic)
+	if err != nil {
+		return nil, err
+	}
+
+	// unlock segment
+	mux.Unlock()
+
+	// listen for transaction Status
+	defer sub.Unsubscribe()
+	for {
+		status := <-sub.Chan()
+		if status.IsDropped {
+			tx := &TransactionDetails{
+				NodeId: nodeId,
+				To:     to,
+				Amount: amount,
+				Status: Dropped,
+			}
+			log.Warningf("Dropped transaction: %v", tx)
+			return tx, nil
+		}
+		if status.IsInvalid {
+			tx := &TransactionDetails{
+				NodeId: nodeId,
+				To:     to,
+				Amount: amount,
+				Status: Invalid,
+			}
+			log.Warningf("Invalid transaction: %v", tx)
+			return tx, nil
+		}
+		if status.IsFinalized {
+			log.Debugf("Transaction for node %s completed at block hash: %#x\n", nodeId, status.AsFinalized)
+			return &TransactionDetails{
+				NodeId: nodeId,
+				To:     to,
+				Amount: amount,
+				Status: Finalized,
+			}, nil
+		}
+	}
+}

internal/script/payout.go

@@ -6,17 +6,16 @@ import (
 	"github.com/NodeFactoryIo/vedran/internal/controllers"
 	"github.com/NodeFactoryIo/vedran/internal/models"
 	"github.com/NodeFactoryIo/vedran/internal/payout"
-	log "github.com/sirupsen/logrus"
 	"net/http"
 	"net/url"
 )
 
 var statsEndpoint, _ = url.Parse("/api/v1/stats")
 
-func ExecutePayout(secret string, totalReward float64, loadbalancerUrl *url.URL) error {
+func ExecutePayout(secret string, totalReward float64, loadbalancerUrl *url.URL) ([]*payout.TransactionDetails, error) {
 	stats, err := fetchStatsFromEndpoint(loadbalancerUrl.ResolveReference(statsEndpoint))
 	if err != nil {
-		return fmt.Errorf("unable to fetch stats from loadbalancer, %v", err)
+		return nil, fmt.Errorf("unable to fetch stats from loadbalancer, %v", err)
 	}
 
 	// calculate distribution
@@ -30,9 +29,12 @@ func ExecutePayout(secret string, totalReward float64, loadbalancerUrl *url.URL)
 		float64(stats.Fee),
 	)
 
-	// todo - call sending payout transactions
-	log.Info(distributionByNode)
-	return nil
+	return payout.ExecuteAllPayoutTransactions(
+		distributionByNode,
+		stats.Stats,
+		secret,
+		loadbalancerUrl.String(),
+	)
 }
 
 func fetchStatsFromEndpoint(endpoint *url.URL) (*controllers.StatsResponse, error) {