Implemented KAS repository to load token transfers data

datasync/chaindatafetcher/kas/model.go

@@ -1,6 +1,9 @@
 package kas
 
-const TxTableName = "klay_transfers"
+const (
+	TxTableName          = "klay_transfers"
+	KctTransferTableName = "kct_transfers"
+)
 
 type Tx struct {
 	TransactionId   int64  `gorm:"column:transactionId;type:BIGINT;INDEX:idIdx;NOT NULL;PRIMARY_KEY"`
@@ -21,3 +24,17 @@ type Tx struct {
 func (Tx) TableName() string {
 	return TxTableName
 }
+
+type KCTTransfer struct {
+	ContractAddress  []byte `gorm:"column:contractAddress;type:VARBINARY(20);INDEX:ttFromCompIdx,ttToCompIdx;NOT NULL"`
+	From             []byte `gorm:"column:fromAddr;type:VARBINARY(20);INDEX:ttFromCompIdx,ttFromIdx"`
+	To               []byte `gorm:"column:toAddr;type:VARBINARY(20);INDEX:ttToCompIdx,ttToIdx"`
+	TransactionLogId int64  `gorm:"column:transactionLogId;type:BIGINT;PRIMARY_KEY;INDEX:ttFromCompIdx,ttToCompIdx"`
+	Value            string `gorm:"column:value;type:VARCHAR(80)"`
+	TransactionHash  []byte `gorm:"column:transactionHash;type:VARBINARY(32);INDEX:ttHashIdx;NOT NULL"`
+	Timestamp        int64  `gorm:"column:timestamp;type:INT(11)"`
+}
+
+func (KCTTransfer) TableName() string {
+	return KctTransferTableName
+}

datasync/chaindatafetcher/kas/repository.go

@@ -14,7 +14,8 @@ const (
 
 	maxPlaceholders = 65535
 
-	placeholdersPerTxItem = 13
+	placeholdersPerTxItem          = 13
+	placeholdersPerKCTTransferItem = 7
 
 	maxDBRetryCount = 20
 	DBRetryInterval = 1 * time.Second

datasync/chaindatafetcher/kas/repository_token_transfer.go

@@ -0,0 +1,158 @@
+// Copyright 2020 The klaytn Authors
+// This file is part of the klaytn library.
+//
+// The klaytn library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The klaytn library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the klaytn library. If not, see <http://www.gnu.org/licenses/>.
+
+package kas
+
+import (
+	"fmt"
+	"github.com/klaytn/klaytn/blockchain"
+	"github.com/klaytn/klaytn/blockchain/types"
+	"github.com/klaytn/klaytn/common"
+	"math/big"
+	"strings"
+)
+
+var tokenTransferEventHash = common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
+
+// splitToWords divides log data to the words.
+func splitToWords(data []byte) ([]common.Hash, error) {
+	if len(data)%common.HashLength != 0 {
+		return nil, fmt.Errorf("data length is not valid. want: %v, actual: %v", common.HashLength, len(data))
+	}
+	var words []common.Hash
+	for i := 0; i < len(data); i += common.HashLength {
+		words = append(words, common.BytesToHash(data[i:i+common.HashLength]))
+	}
+	return words, nil
+}
+
+// wordToAddress trims input word to get address field only.
+func wordToAddress(word common.Hash) common.Address {
+	return common.BytesToAddress(word[common.HashLength-common.AddressLength:])
+}
+
+// transformLogsToTokenTransfers converts the given event into Klaytn Compatible Token transfers.
+func transformLogsToTokenTransfers(event blockchain.ChainEvent) ([]*KCTTransfer, error) {
+	timestamp := event.Block.Time().Int64()
+	var kctTransfers []*KCTTransfer
+	for _, log := range event.Logs {
+		if len(log.Topics) > 0 && log.Topics[0] == tokenTransferEventHash {
+			transfer, err := transformLogToTokenTransfer(log)
+			if err != nil {
+				return nil, err
+			}
+			transfer.Timestamp = timestamp
+			kctTransfers = append(kctTransfers, transfer)
+		}
+	}
+
+	return kctTransfers, nil
+}
+
+// transformLogToTokenTransfer converts the given log to Klaytn Compatible Token transfer.
+func transformLogToTokenTransfer(log *types.Log) (*KCTTransfer, error) {
+	// in case of token transfer,
+	// case 1:
+	//   log.LogTopics[0] = token transfer event hash
+	//   log.LogData = concat(fromAddress, toAddress, value)
+	// case 2:
+	//   log.LogTopics[0] = token transfer event hash
+	//   log.LogTopics[1] = fromAddress
+	//   log.LogTopics[2] = toAddresss
+	//   log.LogData = value
+	words, err := splitToWords(log.Data)
+	if err != nil {
+		return nil, err
+	}
+	data := append(log.Topics, words...)
+	from := wordToAddress(data[1])
+	to := wordToAddress(data[2])
+	value := new(big.Int).SetBytes(data[3].Bytes())
+
+	txLogId := int64(log.BlockNumber)*maxTxCountPerBlock*maxTxLogCountPerTx + int64(log.TxIndex)*maxTxLogCountPerTx + int64(log.Index)
+
+	return &KCTTransfer{
+		ContractAddress:  log.Address.Bytes(),
+		From:             from.Bytes(),
+		To:               to.Bytes(),
+		TransactionLogId: txLogId,
+		Value:            "0x" + value.Text(16),
+		TransactionHash:  log.TxHash.Bytes(),
+	}, nil
+}
+
+// InsertTokenTransfers inserts token transfers in the given chain event into KAS database.
+// The token transfers are divided into chunkUnit because of max number of place holders.
+func (r *repository) InsertTokenTransfers(event blockchain.ChainEvent) error {
+	tokenTransfers, err := transformLogsToTokenTransfers(event)
+	if err != nil {
+		return err
+	}
+
+	chunkUnit := maxPlaceholders / placeholdersPerKCTTransferItem
+	var chunks []*KCTTransfer
+
+	for tokenTransfers != nil {
+		if placeholdersPerKCTTransferItem*len(tokenTransfers) > maxPlaceholders {
+			chunks = tokenTransfers[:chunkUnit]
+			tokenTransfers = tokenTransfers[chunkUnit:]
+		} else {
+			chunks = tokenTransfers
+			tokenTransfers = nil
+		}
+
+		if err := r.bulkInsertTokenTransfers(chunks); err != nil {
+			logger.Error("failed to insertTokenTransfers", "err", err, "numTokenTransfers", len(chunks))
+			return err
+		}
+	}
+
+	return nil
+}
+
+// bulkInsertTokenTransfers inserts the given token transfers in multiple rows at once.
+func (r *repository) bulkInsertTokenTransfers(tokenTransfers []*KCTTransfer) error {
+	if len(tokenTransfers) == 0 {
+		logger.Debug("the token transfer list is empty")
+		return nil
+	}
+	var valueStrings []string
+	var valueArgs []interface{}
+
+	for _, transfer := range tokenTransfers {
+		valueStrings = append(valueStrings, "(?,?,?,?,?,?,?)")
+
+		valueArgs = append(valueArgs, transfer.TransactionLogId)
+		valueArgs = append(valueArgs, transfer.From)
+		valueArgs = append(valueArgs, transfer.To)
+		valueArgs = append(valueArgs, transfer.Value)
+		valueArgs = append(valueArgs, transfer.ContractAddress)
+		valueArgs = append(valueArgs, transfer.TransactionHash)
+		valueArgs = append(valueArgs, transfer.Timestamp)
+	}
+
+	rawQuery := `
+			INSERT INTO kct_transfers(transactionLogId, fromAddr, toAddr, value, contractAddress, transactionHash, timestamp)
+			VALUES %s
+			ON DUPLICATE KEY
+			UPDATE transactionLogId=transactionLogId`
+	query := fmt.Sprintf(rawQuery, strings.Join(valueStrings, ","))
+
+	if _, err := r.db.DB().Exec(query, valueArgs...); err != nil {
+		return err
+	}
+	return nil
+}

datasync/chaindatafetcher/kas/repository_token_transfers_test.go

@@ -0,0 +1,65 @@
+// Copyright 2020 The klaytn Authors
+// This file is part of the klaytn library.
+//
+// The klaytn library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The klaytn library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the klaytn library. If not, see <http://www.gnu.org/licenses/>.
+
+package kas
+
+import (
+	"github.com/klaytn/klaytn/common"
+	"github.com/klaytn/klaytn/common/hexutil"
+	"github.com/stretchr/testify/assert"
+	"strings"
+	"testing"
+)
+
+func TestSplitToWords_Success(t *testing.T) {
+	data := "0x0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000850f0263a87af6dd51acb8baab96219041e28fda00000000000000000000000000000000000000000000d3c21bcecceda1000000"
+	bytes, err := hexutil.Decode(data)
+	assert.NoError(t, err)
+
+	hashes, err := splitToWords(bytes)
+	assert.NoError(t, err)
+	assert.Equal(t, common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000"), hashes[0])
+	assert.Equal(t, common.HexToHash("0x000000000000000000000000850f0263a87af6dd51acb8baab96219041e28fda"), hashes[1])
+	assert.Equal(t, common.HexToHash("0x00000000000000000000000000000000000000000000d3c21bcecceda1000000"), hashes[2])
+}
+
+func TestSplitToWords_Fail_DataLengthError(t *testing.T) {
+	data := "0x0000000000000000000000000000000000850f0263a87af6dd51acb8baab96219041e28fda00000000000000000000000000000000000000000000d3c21bcecceda1000000"
+	bytes, err := hexutil.Decode(data)
+	assert.NoError(t, err)
+
+	_, err = splitToWords(bytes)
+	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), "data length is not valid"))
+}
+
+func TestWordsToAddress_Success(t *testing.T) {
+	hash1 := common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000")
+	hash2 := common.HexToHash("0x000000000000000000000000850f0263a87af6dd51acb8baab96219041e28fda")
+	hash3 := common.HexToHash("0x00000000000000000000000000000000000000000000d3c21bcecceda1000000")
+
+	expected1 := common.HexToAddress("0x0000000000000000000000000000000000000000")
+	expected2 := common.HexToAddress("0x850f0263a87af6dd51acb8baab96219041e28fda")
+	expected3 := common.HexToAddress("0x00000000000000000000d3c21bcecceda1000000")
+
+	addr1 := wordToAddress(hash1)
+	addr2 := wordToAddress(hash2)
+	addr3 := wordToAddress(hash3)
+
+	assert.Equal(t, expected1, addr1)
+	assert.Equal(t, expected2, addr2)
+	assert.Equal(t, expected3, addr3)
+}

datasync/chaindatafetcher/repository.go

@@ -4,4 +4,5 @@ import "github.com/klaytn/klaytn/blockchain"
 
 type repository interface {
 	InsertTransactions(event blockchain.ChainEvent) error
+	InsertTokenTransfers(event blockchain.ChainEvent) error
 }