add mark fusion for oneinchv6 parser

v2/cmd/backfill/main.go

@@ -27,7 +27,7 @@ import (
 	promotion1inchv2 "github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser/oneinchv2"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/rpcnode"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/backfill"
-	promoteeTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
+	promotee_storage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/state"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs"
 	bebopStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/bebop"
@@ -99,7 +99,7 @@ func run(c *cli.Context) error {
 	manager := tradelogs.NewManager(l, storages)
 
 	//promotee storage
-	promoteeStorage := promoteeTypes.New(l, db)
+	promoteeStorage := promotee_storage.New(l, db)
 
 	// backfill storage
 	backfillStorage := backfill.New(l, db)
@@ -132,7 +132,7 @@ func run(c *cli.Context) error {
 		paraswap.MustNewParser(),
 		kyberswaprfq.MustNewParser(),
 		hashflowv3.MustNewParser(),
-		oneinchv6.MustNewParser(),
+		oneinchv6.MustNewParser(promoteeStorage),
 		uniswapx.MustNewParser(),
 		bebop.MustNewParser(),
 		zxrfqv3.MustNewParserWithDeployer(l, zxv3DeployStorage, ethClients[0], common.HexToAddress(constant.Deployer0xV3)),

v2/cmd/parse_log/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser"
 	promotion1inchv2 "github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser/oneinchv2"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/rpcnode"
-	promoteeTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
+	promotee_storage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/state"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs"
 	bebopStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/bebop"
@@ -98,7 +98,7 @@ func run(c *cli.Context) error {
 	})
 
 	//promotee storage
-	promoteeStorage := promoteeTypes.New(l, db)
+	promoteeStorage := promotee_storage.New(l, db)
 
 	// state storage
 	s := state.New(l, db)
@@ -128,7 +128,7 @@ func run(c *cli.Context) error {
 		paraswap.MustNewParser(),
 		kyberswaprfq.MustNewParser(),
 		hashflowv3.MustNewParser(),
-		oneinchv6.MustNewParser(),
+		oneinchv6.MustNewParser(promoteeStorage),
 		uniswapx.MustNewParser(),
 		bebop.MustNewParser(),
 		zxrfqv3.MustNewParserWithDeployer(l, zxv3DeployStorage, ethClients[0], common.HexToAddress(constant.Deployer0xV3)),

v2/cmd/price_filler/main.go

@@ -80,7 +80,7 @@ func run(c *cli.Context) error {
 		l.Errorw("Error while init price filler")
 		return err
 	}
-	priceFiller.Run()
+	priceFiller.Run(c.Int(libapp.FillPriceTimeIntervalFlag.Name))
 	return nil
 }
 

v2/pkg/app/price_filler.go

@@ -13,15 +13,22 @@ var BinanceSecretKeyFlag = cli.StringFlag{
 	Name:   "binance-secret-key",
 	EnvVar: "BINANCE_SECRET_KEY",
 }
+
 var MarkToMarketURLFlag = cli.StringFlag{
 	Name:   "mark-to-market-url",
 	EnvVar: "MARK_TO_MARKET_URL",
 }
 
+var FillPriceTimeIntervalFlag = cli.StringFlag{
+	Name:   "fill-price-time-interval",
+	EnvVar: "FILL_PRICE_TIME_INTERVAL",
+}
+
 func PriceFillerFlags() []cli.Flag {
 	return []cli.Flag{
 		BinanceAPIKeyFlag,
 		BinanceSecretKeyFlag,
 		MarkToMarketURLFlag,
+		FillPriceTimeIntervalFlag,
 	}
 }

v2/pkg/handler/trade_logs.go

@@ -73,12 +73,12 @@ func (h *TradeLogHandler) ProcessBlockWithExclusion(blockHash string, blockNumbe
 		return fmt.Errorf("delete blocks error: %w", err)
 	}
 
-	err = h.processForTradelog(calls, blockHash, blockNumber, timestamp, exclusions)
+	err = h.processForPromotion(calls, blockHash, blockNumber, timestamp)
 	if err != nil {
 		return fmt.Errorf("error when process block: %d", blockNumber)
 	}
 
-	err = h.processForPromotion(calls, blockHash, blockNumber, timestamp)
+	err = h.processForTradelog(calls, blockHash, blockNumber, timestamp, exclusions)
 	if err != nil {
 		return fmt.Errorf("error when process block: %d", blockNumber)
 	}

v2/pkg/parser/oneinchv6/parser.go

@@ -10,6 +10,7 @@ import (
 	"github.com/KyberNetwork/tradelogs/v2/pkg/constant"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/decoder"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/parser"
+	promotee_storage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
 	storageTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/types"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/types"
 	"github.com/ethereum/go-ethereum/accounts/abi"
@@ -36,12 +37,13 @@ func init() {
 }
 
 type Parser struct {
-	abi       *abi.ABI
-	ps        *Oneinchv6Filterer
-	eventHash string
+	abi             *abi.ABI
+	ps              *Oneinchv6Filterer
+	eventHash       string
+	promoteeStorage *promotee_storage.Storage
 }
 
-func MustNewParser() *Parser {
+func MustNewParser(promoteeStorage *promotee_storage.Storage) *Parser {
 	ps, err := NewOneinchv6Filterer(common.Address{}, nil)
 	if err != nil {
 		panic(err)
@@ -56,9 +58,10 @@ func MustNewParser() *Parser {
 	}
 
 	return &Parser{
-		ps:        ps,
-		abi:       ab,
-		eventHash: event.ID.String(),
+		ps:              ps,
+		abi:             ab,
+		eventHash:       event.ID.String(),
+		promoteeStorage: promoteeStorage,
 	}
 }
 
@@ -116,8 +119,19 @@ func (p *Parser) ParseFromInternalCall(order storageTypes.TradeLog, internalCall
 	if err != nil {
 		return order, fmt.Errorf("error when parse contract call to order %w", err)
 	}
-
 	order.Taker = internalCall.From
+
+	if order.Type == storageTypes.RFQType {
+		return order, nil
+	}
+
+	fusion, err := p.promoteeStorage.CheckPromoteeExist(strings.ToLower(order.Taker))
+	if err != nil {
+		return order, nil
+	}
+	if fusion {
+		order.Type = storageTypes.FusionType
+	}
 	return order, nil
 }
 

v2/pkg/price_filler/price_filler.go

@@ -18,17 +18,16 @@ import (
 )
 
 const (
-	NetworkETHChainID              = 1
-	NetworkETHChainIDString        = "1"
-	NetworkETH                     = "ETH"
-	updateAllCoinInfoInterval      = 12 * time.Hour
-	backfillTradeLogsPriceInterval = 10 * time.Minute
-	backfillTradeLogsLimit         = 60
-	addressETH1                    = "0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
-	addressETH2                    = "0x0000000000000000000000000000000000000000"
-	coinUSDT                       = "USDT"
-	USDTAddress                    = "0xdac17f958d2ee523a2206206994597c13d831ec7"
-	invalidSymbolErrString         = "<APIError> code=-1121, msg=Invalid symbol."
+	NetworkETHChainID         = 1
+	NetworkETHChainIDString   = "1"
+	NetworkETH                = "ETH"
+	updateAllCoinInfoInterval = 12 * time.Hour
+	backfillTradeLogsLimit    = 60
+	addressETH1               = "0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
+	addressETH2               = "0x0000000000000000000000000000000000000000"
+	coinUSDT                  = "USDT"
+	USDTAddress               = "0xdac17f958d2ee523a2206206994597c13d831ec7"
+	invalidSymbolErrString    = "<APIError> code=-1121, msg=Invalid symbol."
 )
 
 var (
@@ -86,9 +85,9 @@ func NewPriceFiller(l *zap.SugaredLogger,
 	return p, nil
 }
 
-func (p *PriceFiller) Run() {
+func (p *PriceFiller) Run(fillPriceInterval int) {
 	go p.runUpdateAllCoinInfoRoutine()
-	p.runBackFillTradelogPriceRoutine()
+	p.runBackFillTradelogPriceRoutine(fillPriceInterval)
 }
 
 func (p *PriceFiller) getPrice(token string, timestamp int64) (float64, error) {
@@ -141,8 +140,8 @@ func (p *PriceFiller) runUpdateAllCoinInfoRoutine() {
 	}
 }
 
-func (p *PriceFiller) runBackFillTradelogPriceRoutine() {
-	ticker := time.NewTicker(backfillTradeLogsPriceInterval)
+func (p *PriceFiller) runBackFillTradelogPriceRoutine(fillPriceInterval int) {
+	ticker := time.NewTicker(time.Duration(fillPriceInterval) * time.Minute)
 	defer ticker.Stop()
 
 	for ; ; <-ticker.C {

v2/pkg/storage/promotees/storage.go

@@ -59,9 +59,9 @@ func (s *Storage) Insert(promotees []Promotee) error {
 
 func (s *Storage) Get(query PromoteesQuery) ([]Promotee, error) {
 	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
-		Select(fmt.Sprintf("%s.promoter, promotee, chain_id, tx_hash, timestamp, name", promoteesTable)).
+		Select(fmt.Sprintf("%s.promoter, promotee, chain_id, tx_hash, timestamp, COALESCE(%s.name, '') AS name", promoteesTable, nameTable)).
 		From(promoteesTable).
-		Join(fmt.Sprintf("%s ON %s.promoter = %s.promoter", nameTable, promoteesTable, nameTable))
+		LeftJoin(fmt.Sprintf("%s ON %s.promoter = %s.promoter", nameTable, promoteesTable, nameTable))
 
 	v := reflect.ValueOf(query)
 	types := v.Type()
@@ -128,3 +128,25 @@ func (s *Storage) InsertPromoterName(promotees []Promotee) error {
 	}
 	return nil
 }
+
+func (s *Storage) CheckPromoteeExist(promotee string) (bool, error) {
+	builder := squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar).
+		Select("COUNT(*)").
+		From(promoteesTable).
+		Where(squirrel.Eq{"promotee": promotee}).
+		Where(squirrel.Eq{"chain_id": "1"})
+
+	q, p, err := builder.ToSql()
+	if err != nil {
+		return false, err
+	}
+
+	var count int
+	if err := s.db.Get(&count, q, p...); err != nil {
+		return false, err
+	}
+	if count >= 1 {
+		return true, nil
+	}
+	return false, nil
+}

v2/pkg/storage/promotees/storage_test.go

@@ -27,16 +27,13 @@ func TestInsert(t *testing.T) {
 		t.Log(promotee)
 	}
 
-	// byteValue2, err := os.Open("test2.json")
+	byteValue2, err := os.Open("test2.json")
 
-	// assert.NoError(t, err)
-	// assert.NoError(t, json.NewDecoder(byteValue2).Decode(&promotees), "failed to parse promotees")
-	// assert.NoError(t, s.Insert(promotees), "failed to insert promotees")
-	// t.Log(promotees)
-	var query PromoteesQuery
-	query.Promoter = ""
-	query.Promotee = ""
-	query.ChainId = ""
+	assert.NoError(t, err)
+	assert.NoError(t, json.NewDecoder(byteValue2).Decode(&promotees), "failed to parse promotees")
+	assert.NoError(t, s.Insert(promotees), "failed to insert promotees")
+	t.Log(promotees)
+	query := PromoteesQuery{}
 	pwithname, err := s.Get(query)
 
 	assert.NoError(t, err, "Failed to get promotees")
@@ -64,4 +61,5 @@ func TestInsert(t *testing.T) {
 		}
 
 	}
+	t.Log(s.CheckPromoteeExist("0x0b8a49d816cc709b6eadb09498030ae3416b66dc"))
 }

v2/pkg/storage/promotees/test.json

@@ -3,31 +3,31 @@
         "promoter": "0xd7f6f541d4210550ca56f7b4c4a549efd4cafb49",
         "promotee": "0x1e9d349cec77fea6481f009593101d0e20a69490",
         "timestamp": 498529484,
-        "event_hash": "0xf8cf20ddcd5eaf81ea4967c60402d1f3805804126fd4371bf9e39a7605ccad02",
+        "tx_hash": "0xf8cf20ddcd5eaf81ea4967c60402d1f3805804126fd4371bf9e39a7605ccad02",
         "chain_id": "1",
         "block_number": 1
       },
       {
         "promoter": "0xe023f53f735c196e4a028233c2ee425957812a41",
         "promotee": "0xd1742b3c4fbb096990c8950fa635aec75b30781a",
         "timestamp": 498539324,
-        "event_hash": "0x4dc9aa484e1afe6f9e01280925504841c26a81e0db09c6406ca7b1869bdc6ea8",
+        "tx_hash": "0x4dc9aa484e1afe6f9e01280925504841c26a81e0db09c6406ca7b1869bdc6ea8",
         "chain_id": "2",
         "block_number": 2
       },
       {
         "promoter": "0xa260f8b7c8f37c2f1bc11b04c19902829de6ac8a",
         "promotee": "0x0b8a49d816cc709b6eadb09498030ae3416b66dc",
         "timestamp": 498539524,
-        "event_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
+        "tx_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
         "chain_id": "1",
         "block_number": 3
       },
       {
         "promoter": "0xe023f53f735c196e4a028233c2ee425957812a41",
         "promotee": "0x01372be9616ff00469d3e509b77d6960af757180",
         "timestamp": 4985437524,
-        "event_hash": "0x46ab12e2cb6e5e7c2a1e0a1da7735785b713124aeb93ba3093e64ed69451e624",
+        "tx_hash": "0x46ab12e2cb6e5e7c2a1e0a1da7735785b713124aeb93ba3093e64ed69451e624",
         "chain_id": "2",
         "block_number": 4
       }

v2/pkg/storage/promotees/test2.json

@@ -3,39 +3,39 @@
         "promoter": "0xd7f6f541d4210550ca56f7b4c4a549efd4cafb49",
         "promotee": "0x1e9d349cec77fea6481f009593101d0e20a69490",
         "timestamp": 498529484,
-        "event_hash": "0xf8cf20ddcd5eaf81ea4967c60402d1f3805804126fd4371bf9e39a7605ccad02",
+        "tx_hash": "0xf8cf20ddcd5eaf81ea4967c60402d1f3805804126fd4371bf9e39a7605ccad02",
         "chain_id": "1",
         "block_number": 1
       },
       {
         "promoter": "0xe023f53f735c196e4a028233c2ee425957812a41",
         "promotee": "0xd1742b3c4fbb096990c8950fa635aec75b30781a",
         "timestamp": 498539324,
-        "event_hash": "0x4dc9aa484e1afe6f9e01280925504841c26a81e0db09c6406ca7b1869bdc6ea8",
+        "tx_hash": "0x4dc9aa484e1afe6f9e01280925504841c26a81e0db09c6406ca7b1869bdc6ea8",
         "chain_id": "2",
         "block_number": 2
       },
       {
         "promoter": "0xa260f8b7c8f37c2f1bc11b04c19902829de6ac8a",
         "promotee": "0x0b8a49d816cc709b6eadb09498030ae3416b66dc",
         "timestamp": 498539524,
-        "event_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
+        "tx_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
         "chain_id": "1",
         "block_number": 3
       },
       {
         "promoter": "0xa260f8b7c8f37c2f1bc11b04c19902829de6ac8a",
         "promotee": "0x0b8a49d816cc709b6eadb09498030ae3416b66dc",
         "timestamp": 498539524,
-        "event_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
+        "tx_hash": "0xeb0bd6036ef28377b2288e0d1fb38c064c14635085777c6e88c617f9b21ac85b",
         "chain_id": "2",
         "block_number": 3
       },
       {
         "promoter": "0xe023f53f735c196e4a028233c257812a41",
         "promotee": "0x01372be9616ff00469d3e509b77d6daf757180",
         "timestamp": 4985437524,
-        "event_hash": "0x46ab12e2cb6e5e7c2a1e0a1da7735785b713124aeb93ba3093e64ed69451e624",
+        "tx_hash": "0x46ab12e2cb6e5e7c2a1e0a1da7735785b713124aeb93ba3093e64ed69451e624",
         "chain_id": "2",
         "block_number": 4
       }

v2/pkg/storage/tradelogs/types/oneinch_v6.go

@@ -3,5 +3,6 @@ package types
 type OneInchType string
 
 const (
-	RFQType OneInchType = "RFQ"
+	RFQType    OneInchType = "RFQ"
+	FusionType OneInchType = "Fusion"
 )