feat: parse oneinch rfq trades

.gitignore

@@ -3,4 +3,7 @@ cmd/*/*
 !cmd/*/*.go
 !cmd/*/migrations
 */cmd/*/*
-!*/cmd/*/*.go
+!*/cmd/*/*.go
+.idea
+.env.example
+.env

cmd/tradelogs/main.go

@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"fmt"
+	"github.com/KyberNetwork/tradelogs/pkg/parser"
+	"github.com/KyberNetwork/tradelogs/pkg/parser/oneinch"
 	"log"
 	"os"
 
@@ -13,7 +15,6 @@ import (
 	tradelogs "github.com/KyberNetwork/tradelogs/internal/server/tradelogs"
 	"github.com/KyberNetwork/tradelogs/internal/worker"
 	"github.com/KyberNetwork/tradelogs/pkg/evmlistenerclient"
-	"github.com/KyberNetwork/tradelogs/pkg/parser"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/hashflow"
 	hashflowv3 "github.com/KyberNetwork/tradelogs/pkg/parser/hashflow_v3"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/kyberswap"
@@ -39,6 +40,7 @@ func main() {
 	app.Flags = append(app.Flags, libapp.EvmListenerFlags()...)
 	app.Flags = append(app.Flags, libapp.HTTPServerFlags()...)
 	app.Flags = append(app.Flags, libapp.BigqueryFlags()...)
+	app.Flags = append(app.Flags, libapp.RPCNodeFlags()...)
 
 	if err := app.Run(os.Args); err != nil {
 		log.Panic(err)
@@ -72,6 +74,7 @@ func run(c *cli.Context) error {
 		l.Errorw("Error while init listener service")
 		return err
 	}
+
 	w, err := worker.New(l, s, listener,
 		kyberswap.MustNewParser(),
 		zxotc.MustNewParser(),
@@ -98,7 +101,9 @@ func run(c *cli.Context) error {
 		"native":       native.MustNewParser(),
 		"kyberswaprfq": kyberswaprfq.MustNewParser(),
 		"hashflowv3":   hashflowv3.MustNewParser(),
+		"1inch":        oneinch.MustNewParser(c.String(libapp.RPCUrlFlagName)),
 	}
+
 	backfillWorker, err := bigquery.NewWorker(libapp.BigqueryProjectIDFFromCli(c), s, parserMap)
 	if err != nil {
 		l.Errorw("Error while init backfillWorker")

go.mod

@@ -37,6 +37,8 @@ require (
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
+	github.com/gammazero/deque v0.2.0 // indirect
+	github.com/gammazero/workerpool v1.1.3 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
 	github.com/golang/protobuf v1.5.3 // indirect

go.sum

@@ -420,6 +420,10 @@ github.com/gabriel-vasile/mimetype v1.3.1/go.mod h1:fA8fi6KUiG7MgQQ+mEWotXoEOvmx
 github.com/gabriel-vasile/mimetype v1.4.0/go.mod h1:fA8fi6KUiG7MgQQ+mEWotXoEOvmxRtOJlERCzSmRvr8=
 github.com/gabriel-vasile/mimetype v1.4.2 h1:w5qFW6JKBz9Y393Y4q372O9A7cUSequkh1Q7OhCmWKU=
 github.com/gabriel-vasile/mimetype v1.4.2/go.mod h1:zApsH/mKG4w07erKIaJPFiX0Tsq9BFQgN3qGY5GnNgA=
+github.com/gammazero/deque v0.2.0 h1:SkieyNB4bg2/uZZLxvya0Pq6diUlwx7m2TeT7GAIWaA=
+github.com/gammazero/deque v0.2.0/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
+github.com/gammazero/workerpool v1.1.3 h1:WixN4xzukFoN0XSeXF6puqEqFTl2mECI9S6W44HWy9Q=
+github.com/gammazero/workerpool v1.1.3/go.mod h1:wPjyBLDbyKnUn2XwwyD3EEwo9dHutia9/fwNmSHWACc=
 github.com/garyburd/redigo v0.0.0-20150301180006-535138d7bcd7/go.mod h1:NR3MbYisc3/PwhQ00EMzDiPmrwpPxAn5GI05/YaO1SY=
 github.com/gavv/httpexpect v2.0.0+incompatible/go.mod h1:x+9tiU1YnrOvnB725RkpoLv1M62hOWzwo5OXotisrKc=
 github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff h1:tY80oXqGNY4FhTFhk+o9oFHGINQ/+vhlm8HFzi6znCI=

internal/app/rpcnode.go

@@ -0,0 +1,21 @@
+package app
+
+import "github.com/urfave/cli"
+
+const (
+	RPCUrlFlagName = "rpc-url"
+)
+
+var (
+	RPCUrlFlag = &cli.StringFlag{
+		Name:   RPCUrlFlagName,
+		EnvVar: "RPC_URL",
+		Usage:  "RPC node url",
+	}
+)
+
+func RPCNodeFlags() []cli.Flag {
+	return []cli.Flag{
+		RPCUrlFlag,
+	}
+}

internal/types/contract_call.go

@@ -0,0 +1,12 @@
+package types
+
+type ContractCallParam struct {
+	Name  string      `json:"name"`
+	Value interface{} `json:"value"`
+	Type  string      `json:"type"`
+}
+type ContractCall struct {
+	ContractType string              `json:"contract_type,omitempty"`
+	Name         string              `json:"name"`
+	Params       []ContractCallParam `json:"params"`
+}

internal/types/trace_call.go

@@ -0,0 +1,39 @@
+package types
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	ethereumTypes "github.com/ethereum/go-ethereum/core/types"
+)
+
+type TraceCallResponse struct {
+	Jsonrpc string    `json:"jsonrpc"`
+	ID      int       `json:"id"`
+	Result  CallFrame `json:"result"`
+}
+
+type CallLog struct {
+	Address common.Address `json:"address"`
+	Topics  []common.Hash  `json:"topics"`
+	Data    string         `json:"data"`
+}
+
+func (l CallLog) ToEthereumLog() ethereumTypes.Log {
+	return ethereumTypes.Log{
+		Address: l.Address,
+		Topics:  l.Topics,
+		Data:    common.Hex2Bytes(l.Data),
+	}
+}
+
+type CallFrame struct {
+	From    string      `json:"from"`
+	Gas     string      `json:"gas"`
+	GasUsed string      `json:"gasUsed"`
+	To      string      `json:"to"`
+	Input   string      `json:"input"`
+	Output  string      `json:"output"`
+	Calls   []CallFrame `json:"calls"`
+	Value   string      `json:"value"`
+	Type    string      `json:"type"`
+	Logs    []CallLog   `json:"logs"`
+}

internal/worker/type.go

@@ -0,0 +1,8 @@
+package worker
+
+import "github.com/KyberNetwork/tradelogs/pkg/storage"
+
+type parseEventLogResult struct {
+	tradeLogOrder storage.TradeLog
+	err           error
+}

internal/worker/worker.go

@@ -4,10 +4,10 @@ import (
 	"context"
 
 	"github.com/KyberNetwork/tradelogs/pkg/convert"
-
-	"github.com/KyberNetwork/tradelogs/pkg/storage"
 	"github.com/KyberNetwork/tradelogs/pkg/evmlistenerclient"
 	"github.com/KyberNetwork/tradelogs/pkg/parser"
+	"github.com/KyberNetwork/tradelogs/pkg/storage"
+	"github.com/gammazero/workerpool"
 	"go.uber.org/zap"
 )
 
@@ -54,8 +54,10 @@ func (w *Worker) Run(ctx context.Context) error {
 }
 func (w *Worker) processMessages(m []evmlistenerclient.Message) error {
 	var (
-		insertOrders []storage.TradeLog
-		deleteBlocks []uint64
+		insertOrders         []storage.TradeLog
+		deleteBlocks         []uint64
+		wp                   = workerpool.New(100)
+		insertOrdersResultCh = make(chan parseEventLogResult)
 	)
 
 	for _, message := range m {
@@ -68,17 +70,41 @@ func (w *Worker) processMessages(m []evmlistenerclient.Message) error {
 				if ps == nil {
 					continue
 				}
-				order, err := ps.Parse(convert.ToETHLog(log), block.Timestamp)
-				if err != nil {
-					return err
-				}
-				insertOrders = append(insertOrders, order)
+				// Create new instance of req for the goroutine. Can remove these variables after go1.22
+				block := block
+				log := log
+				wp.Submit(func() {
+					order, err := ps.Parse(convert.ToETHLog(log), block.Timestamp)
+					if err != nil {
+						insertOrdersResultCh <- parseEventLogResult{
+							err: err,
+						}
+						return
+					}
+					insertOrdersResultCh <- parseEventLogResult{
+						tradeLogOrder: order,
+						err:           nil,
+					}
+				})
 			}
 		}
 		for _, block := range message.RevertedBlocks {
 			deleteBlocks = append(deleteBlocks, block.Number.Uint64())
 		}
 	}
+
+	go func() {
+		wp.StopWait()
+		close(insertOrdersResultCh)
+	}()
+
+	for result := range insertOrdersResultCh {
+		if result.err != nil {
+			return result.err
+		}
+		insertOrders = append(insertOrders, result.tradeLogOrder)
+	}
+
 	err := w.s.Delete(deleteBlocks)
 	if err != nil {
 		return err

pkg/abitypes/abitypes.go

@@ -0,0 +1,8 @@
+package abitypes
+
+import "github.com/ethereum/go-ethereum/accounts/abi"
+
+var (
+	Uint256, _ = abi.NewType("uint256", "", nil)
+	Bytes32, _ = abi.NewType("bytes32", "", nil)
+)

pkg/decoder/decoder.go

@@ -0,0 +1,48 @@
+package decoder
+
+import (
+	"encoding/hex"
+	"fmt"
+	"github.com/KyberNetwork/tradelogs/internal/types"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+)
+
+func Decode(ABI *abi.ABI, input string) (*types.ContractCall, error) {
+	if ABI == nil {
+		return nil, fmt.Errorf("missing abi")
+	}
+	inputBytes := common.FromHex(input)
+	method, err := ABI.MethodById(inputBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	bytes, err := hex.DecodeString(input[10:])
+	if err != nil {
+		return nil, err
+	}
+
+	inputs, err := method.Inputs.Unpack(bytes)
+	if err != nil {
+		return nil, err
+	}
+
+	nonIndexedArgs := method.Inputs.NonIndexed()
+
+	contractCall := &types.ContractCall{
+		Name: method.Name,
+	}
+
+	for i, input := range inputs {
+		arg := nonIndexedArgs[i]
+		param := types.ContractCallParam{
+			Name:  arg.Name,
+			Value: input,
+			Type:  arg.Type.String(),
+		}
+		contractCall.Params = append(contractCall.Params, param)
+	}
+
+	return contractCall, nil
+}