Merge pull request #17 from XDagger/develop

rewards batch transfer

README.md

@@ -91,8 +91,8 @@ Configuration is self-describing, just copy *config.example.json* to *config.jso
 
 	"payout": {
 		"poolRation": 5.0,
-		"rewardRation": 5.0,
-		"directRation": 5.0,
+		"rewardRation": 0.0,
+		"directRation": 0.0,
     // threshhold to pay miner
 		"threshold": 3,
 		"paymentInterval": "10m",

chunk_test.go

@@ -0,0 +1,125 @@
+package main
+
+import (
+	"fmt"
+	"path"
+	"testing"
+
+	"github.com/XDagger/xdagpool/kvstore"
+	"github.com/XDagger/xdagpool/payouts"
+	"github.com/XDagger/xdagpool/pool"
+	"github.com/XDagger/xdagpool/util"
+	"github.com/XDagger/xdagpool/xdago/common"
+	bip "github.com/XDagger/xdagpool/xdago/wallet"
+)
+
+func TestFieldTypes(t *testing.T) {
+	s := payouts.FieldChunkTypes(false, true, true, 10)
+	fmt.Println(s)
+
+	s = payouts.FieldChunkTypes(false, false, true, 11)
+	fmt.Println(s)
+
+	s = payouts.FieldChunkTypes(false, true, false, 6)
+	fmt.Println(s)
+
+	s = payouts.FieldChunkTypes(true, false, false, 5)
+	fmt.Println(s)
+}
+
+func TestChunkSign(t *testing.T) {
+	// init log file
+	// _ = os.Mkdir("../logs", os.ModePerm)
+	iLogFile := "logs/info.log"
+	eLogFile := "logs/error.log"
+	sLogFile := "logs/share.log"
+	bLogFile := "logs/block.log"
+	util.InitLog(iLogFile, eLogFile, sLogFile, bLogFile, 10)
+
+	wallet := bip.NewWallet(path.Join(".", common.BIP32_WALLET_FOLDER, common.BIP32_WALLET_FILE_NAME))
+	res := wallet.UnlockWallet("111")
+	if res && wallet.IsHdWalletInitialized() {
+		key := wallet.GetDefKey()
+
+		r, s, n := payouts.ChunkTranxSign("0000000000000000000000000000000000000000000000000000000000000000", key)
+		fmt.Println(r, s, n)
+	}
+}
+
+func TestChunkBlock(t *testing.T) {
+	to := []string{"4duPWMbYUgAifVYkKDCWxLvRRkSByf5gb", "Fve2AF8NrEPjNcAj5BABTBeqn7LW7WfeT", "Fii9BuhR1KogfNzWbtSH1YJgQQDwFMomK"}
+	value := []int64{2 * 1e9, 7 * 1e9, 15 * 1e9}
+
+	iLogFile := "logs/info.log"
+	eLogFile := "logs/error.log"
+	sLogFile := "logs/share.log"
+	bLogFile := "logs/block.log"
+	util.InitLog(iLogFile, eLogFile, sLogFile, bLogFile, 10)
+
+	wallet := bip.NewWallet(path.Join(".", common.BIP32_WALLET_FOLDER, common.BIP32_WALLET_FILE_NAME))
+	res := wallet.UnlockWallet("111")
+	if res && wallet.IsHdWalletInitialized() {
+		key := wallet.GetDefKey()
+
+		block, total := payouts.TransactionChunkBlock("Dd2KRkRceHtx7ep3qWHVAHEdjYoyPpAYx", to, "hello", value, key)
+		fmt.Println(total)
+		fmt.Println(block)
+	}
+
+}
+
+func TestChunkRpc(t *testing.T) {
+	to := []string{"4duPWMbYUgAifVYkKDCWxLvRRkSByf5gb", "Fve2AF8NrEPjNcAj5BABTBeqn7LW7WfeT", "2UqAZBFqA3CbWa6oTR3PYi6BW2wXGK9tY"}
+	value := []int64{2 * 1e9, 3 * 1e9, 4 * 1e9}
+	cfg.NodeRpc = "http://testnet-rpc.xdagj.org:10001"
+	payouts.Cfg = &cfg
+	iLogFile := "logs/info.log"
+	eLogFile := "logs/error.log"
+	sLogFile := "logs/share.log"
+	bLogFile := "logs/block.log"
+	util.InitLog(iLogFile, eLogFile, sLogFile, bLogFile, 10)
+
+	wallet := bip.NewWallet(path.Join(".", common.BIP32_WALLET_FOLDER, common.BIP32_WALLET_FILE_NAME))
+	res := wallet.UnlockWallet("111")
+	if res && wallet.IsHdWalletInitialized() {
+		// fmt.Println(wallet.GetMnemonic())
+		key := wallet.GetDefKey()
+
+		hash, err := payouts.TransferChunkRpc(value, "Fii9BuhR1KogfNzWbtSH1YJgQQDwFMomK", to, "hello", key)
+		fmt.Println(hash)
+		fmt.Println(err)
+	}
+
+}
+
+func TestPayChunk(t *testing.T) {
+	to := []string{"4duPWMbYUgAifVYkKDCWxLvRRkSByf5gb", "Fve2AF8NrEPjNcAj5BABTBeqn7LW7WfeT", "Fii9BuhR1KogfNzWbtSH1YJgQQDwFMomK"}
+	value := []int64{2 * 1e9, 3 * 1e9, 4 * 1e9}
+	cfg.NodeRpc = "http://testnet-rpc.xdagj.org:10001"
+	cfg.Coin = "xdag"
+	cfg.KvRocks = pool.StorageConfig{
+		Endpoint:          "127.0.0.1:6379",
+		PasswordEncrypted: "MbRmWtAs7GA2E1B6ioBSoQ==",
+		Password:          "123456",
+		Database:          0,
+		PoolSize:          10,
+	}
+	cfg.Address = "Dd2KRkRceHtx7ep3qWHVAHEdjYoyPpAYx"
+	payouts.Cfg = &cfg
+	iLogFile := "logs/info.log"
+	eLogFile := "logs/error.log"
+	sLogFile := "logs/share.log"
+	bLogFile := "logs/block.log"
+	util.InitLog(iLogFile, eLogFile, sLogFile, bLogFile, 10)
+
+	wallet := bip.NewWallet(path.Join(".", common.BIP32_WALLET_FOLDER, common.BIP32_WALLET_FILE_NAME))
+	res := wallet.UnlockWallet("111")
+	if res && wallet.IsHdWalletInitialized() {
+		payouts.BipKey = wallet.GetDefKey()
+		backend = kvstore.NewKvClient(&cfg.KvRocks, cfg.Coin)
+
+		payouts.PayChunk(backend, &to, &value, "test pay")
+
+	}
+
+}

kvstore/kvrocks.go

@@ -2,6 +2,7 @@ package kvstore
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/big"
 	"strconv"
@@ -176,6 +177,31 @@ func (r *KvClient) SetPayment(login, txHash, remark string, payment float64, ms,
 	return err
 }
 
+func (r *KvClient) SetChunkPayment(logins []string, txHash, remark string, payments []int64, ms, ts int64) error {
+	if len(logins) != len(payments) {
+		return errors.New("set payment chunck size not match")
+	}
+	var total int64
+	for _, v := range payments {
+		total += v
+	}
+	tx := r.client.TxPipeline()
+	tx.HIncrBy(ctx, r.formatKey("pool", "account"), "payment", total)
+	tx.HIncrBy(ctx, r.formatKey("pool", "account"), "unpaid", -1*total)
+
+	for i := 0; i < len(logins); i++ {
+		tx.HIncrBy(ctx, r.formatKey("account", logins[i]), "payment", payments[i])
+		tx.HIncrBy(ctx, r.formatKey("account", logins[i]), "unpaid", -1*payments[i])
+
+		tx.ZAdd(ctx, r.formatKey("payment", logins[i]), redis.Z{Score: float64(ts),
+			Member: join(float64(payments[i])/float64(1e9), ms, txHash, remark)})
+		tx.ZAdd(ctx, r.formatKey("balance", logins[i]), redis.Z{Score: float64(ts),
+			Member: join("payment", float64(payments[i])/float64(1e9), ms, txHash, remark)})
+	}
+	_, err := tx.Exec(ctx)
+	return err
+}
+
 // func (r *KvClient) SetFund(fund, txHash, jobHash, remark string, payment float64, ms, ts int64) error {
 // 	_, err := r.client.ZAdd(ctx, r.formatKey("donate", fund), redis.Z{Score: float64(ts),
 // 		Member: join(payment, ms, txHash, jobHash, remark)}).Result()
@@ -331,6 +357,9 @@ func (r *KvClient) GetMinerName(jobHash string) []string {
 
 // set lowest hash finder reward of a job
 func (r *KvClient) SetFinderReward(login string, reward pool.XdagjReward, fee float64, ms, ts int64) {
+	if fee*float64(1e9) < 1.0 {
+		return
+	}
 	// minimum hash finder
 	raw, err := r.client.ZRange(ctx, r.formatKey("mini", reward.PreHash), 0, 0).Result()
 	if err != nil {

payouts/chunk_test.go

@@ -0,0 +1,47 @@
+package payouts
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+)
+
+func TestChunk(t *testing.T) {
+
+	miners := make(map[string]int64)
+	for i := 1; i <= 11; i++ {
+		miners[strconv.Itoa(i)] = int64(i)
+	}
+
+	remark := "hello"
+	var chunkSize int
+	if len(remark) > 0 {
+		chunkSize = 10
+	} else {
+		chunkSize = 11
+	}
+	batchAddress := make([]string, 0, chunkSize)
+	batchAmount := make([]int64, 0, chunkSize)
+	for address, amount := range miners {
+
+		if amount > 0 {
+			batchAddress = append(batchAddress, address)
+			batchAmount = append(batchAmount, amount)
+		}
+		if len(batchAddress) == chunkSize {
+			chunkReset(&batchAddress, &batchAmount)
+		}
+	}
+
+	if len(batchAddress) > 0 {
+		chunkReset(&batchAddress, &batchAmount)
+	}
+}
+
+func chunkReset(batchAddress *[]string, batchAmount *[]int64) {
+	fmt.Println("=========================")
+	fmt.Println(batchAmount)
+	fmt.Println(batchAddress)
+	*batchAddress = (*batchAddress)[:0]
+	*batchAmount = (*batchAmount)[:0]
+}

payouts/pay_chunk.go

@@ -0,0 +1,78 @@
+package payouts
+
+import (
+	"errors"
+
+	"github.com/XDagger/xdagpool/kvstore"
+	"github.com/XDagger/xdagpool/pool"
+	"github.com/XDagger/xdagpool/util"
+)
+
+func batchPayMiners(cfg *pool.Config, backend *kvstore.KvClient) {
+	var threshold int64
+	var remark string
+	cfg.RLock()
+	threshold = cfg.PayOut.Threshold
+	remark = cfg.PayOut.PaymentRemark
+	cfg.RUnlock()
+
+	// find miners balance more than payment threshold
+	miners := backend.GetMinersToPay(threshold)
+	if len(miners) == 0 {
+		return
+	}
+	var chunkSize int
+	if len(remark) > 0 {
+		chunkSize = 10
+	} else {
+		chunkSize = 11
+	}
+	batchAddress := make([]string, 0, chunkSize)
+	batchAmount := make([]int64, 0, chunkSize)
+	for address, amount := range miners {
+
+		if amount > 0 {
+			batchAddress = append(batchAddress, address)
+			batchAmount = append(batchAmount, amount)
+		}
+		if len(batchAddress) == chunkSize {
+			PayChunk(backend, &batchAddress, &batchAmount, remark)
+		}
+	}
+
+	if len(batchAddress) > 0 {
+		PayChunk(backend, &batchAddress, &batchAmount, remark)
+	}
+
+}
+
+func PayChunk(backend *kvstore.KvClient, batchAddress *[]string, batchAmount *[]int64, remark string) {
+	ms := util.MakeTimestamp()
+	ts := ms / 1000
+	txHash, err := transfer2chunk(*batchAddress, remark, *batchAmount)
+	if err != nil {
+		util.Error.Println("transfer chunk reward to miners error", err)
+		return
+	}
+	err = backend.SetChunkPayment(*batchAddress, txHash, remark, *batchAmount, ms, ts)
+	if err != nil {
+		util.Error.Println("kv store set chunk payment error", txHash, err)
+	}
+	*batchAddress = (*batchAddress)[:0]
+	*batchAmount = (*batchAmount)[:0]
+}
+
+func transfer2chunk(miners []string, remark string, amounts []int64) (txHash string, err error) {
+	if len(remark) > 0 && !ValidateRemark(remark) {
+		return "", errors.New("remark error")
+	}
+
+	if len(miners) != len(amounts) || len(miners) > 11 || (len(miners) == 11 && len(remark) > 0) {
+		return "", errors.New("transfer chunck size error")
+	}
+
+	txHash, err = TransferChunkRpc(amounts, Cfg.Address, miners, remark, BipKey)
+	return
+	// fmt.Println(amount, Cfg.Address, miner, remark)
+	// return getUuid(), nil
+}

payouts/payment.go

@@ -23,7 +23,8 @@ func PaymentTask(ctx context.Context, cfg *pool.Config, backend *kvstore.KvClien
 			util.Info.Println("exit payment task")
 			return
 		case <-ticker.C:
-			payMiners(cfg, backend)
+			// payMiners(cfg, backend)
+			batchPayMiners(cfg, backend)
 		}
 	}
 }