update 1inch v6

internal/worker/worker.go

@@ -7,7 +7,6 @@ import (
 	"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,10 +53,8 @@ func (w *Worker) Run(ctx context.Context) error {
 }
 func (w *Worker) processMessages(m []evmlistenerclient.Message) error {
 	var (
-		insertOrders         []storage.TradeLog
-		deleteBlocks         []uint64
-		wp                   = workerpool.New(100)
-		insertOrdersResultCh = make(chan parseEventLogResult)
+		insertOrders []storage.TradeLog
+		deleteBlocks []uint64
 	)
 
 	for _, message := range m {
@@ -70,41 +67,18 @@ func (w *Worker) processMessages(m []evmlistenerclient.Message) error {
 				if ps == nil {
 					continue
 				}
-				// 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,
-					}
-				})
+				order, err := ps.Parse(convert.ToETHLog(log), block.Timestamp)
+				if err != nil {
+					continue
+				}
+				insertOrders = append(insertOrders, order)
 			}
 		}
 		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/parser/oneinch/parser.go

@@ -24,8 +24,8 @@ const (
 
 var (
 	ErrInvalidOneInchFilledTopic = errors.New("invalid oneinch order filled topic")
-
-	RFQOrderOutputArgument abi.Arguments
+	ErrNotFoundTraceCall         = errors.New("not found trace call")
+	RFQOrderOutputArgument       abi.Arguments
 )
 
 func init() {
@@ -149,22 +149,22 @@ func (p *Parser) recursiveDetectOneInchRFQTrades(tradeLog storage.TradeLog, trac
 	var (
 		err error
 	)
-	isOneInchRFQTrade := p.isOneInchRFQTrades(tradeLog.OrderHash, traceCall)
+	isOneInchRFQTrade := p.isOneInchRFQTrades(tradeLog.MakerTokenAmount, tradeLog.OrderHash, traceCall)
 
 	if isOneInchRFQTrade {
 		return p.ParseFromInternalCall(tradeLog, traceCall)
 	}
 
 	for _, subCall := range traceCall.Calls {
 		tradeLog, err = p.recursiveDetectOneInchRFQTrades(tradeLog, subCall)
-		if err != nil {
-			return tradeLog, err
+		if err == nil {
+			return tradeLog, nil
 		}
 	}
-	return tradeLog, nil
+	return tradeLog, ErrNotFoundTraceCall
 }
 
-func (p *Parser) isOneInchRFQTrades(orderHash string, traceCall types.CallFrame) bool {
+func (p *Parser) isOneInchRFQTrades(makingAmountOrder string, orderHash string, traceCall types.CallFrame) bool {
 	for _, eventLog := range traceCall.Logs {
 		if len(eventLog.Topics) == 0 {
 			continue
@@ -174,11 +174,11 @@ func (p *Parser) isOneInchRFQTrades(orderHash string, traceCall types.CallFrame)
 			continue
 		}
 
-		_, _, orderHashFromOutput, err := p.decodeOutput(traceCall.Output)
+		makingAmount, _, orderHashFromOutput, err := p.decodeOutput(traceCall.Output)
 		if err != nil {
 			return false
 		}
-		return orderHash == orderHashFromOutput
+		return orderHash == orderHashFromOutput && makingAmount == makingAmountOrder
 	}
 	return false
 }

pkg/parser/oneinchv6/gen.sh

@@ -0,0 +1 @@
+abigen --abi=abi.json --pkg=oneinchv6 --out=oneinchv6.go

pkg/parser/oneinchv6/name.go

@@ -0,0 +1,3 @@
+package oneinchv6
+
+const NameExchange = "1inchv6"

pkg/parser/oneinchv6/oneinchv6_test.go

@@ -0,0 +1,73 @@
+package oneinchv6
+
+import (
+	"context"
+	"encoding/json"
+	"math/big"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/KyberNetwork/tradelogs/pkg/rpcnode"
+	"github.com/KyberNetwork/tradelogs/pkg/tracecall"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/stretchr/testify/require"
+)
+
+const rpcURL = ""
+
+func TestFetchEvent(t *testing.T) {
+	t.Skip("Need to add the rpc url that enables the trace call JSON-RPC")
+	rpcClient, err := rpcnode.NewClient(http.DefaultClient, rpcURL)
+	if err != nil {
+		panic(err)
+	}
+	traceCalls := tracecall.NewCache(rpcClient)
+	p := MustNewParser(traceCalls)
+	require.Equal(t, p.abi.Events[FilledEvent].ID, common.HexToHash("0xfec331350fce78ba658e082a71da20ac9f8d798a99b3c79681c8440cbfe77e07"))
+	client, err := ethclient.Dial(rpcURL)
+	require.NoError(t, err)
+	logs, err := client.FilterLogs(context.Background(), ethereum.FilterQuery{
+		BlockHash: nil,
+		FromBlock: big.NewInt(19517464),
+		ToBlock:   big.NewInt(19517464),
+		Addresses: nil,
+		Topics: [][]common.Hash{
+			{
+				p.abi.Events[FilledEvent].ID,
+			},
+		},
+	})
+	require.NoError(t, err)
+	d, err := json.Marshal(logs)
+	require.NoError(t, err)
+	t.Log(string(d))
+}
+
+func TestParseEvent(t *testing.T) {
+	t.Skip("Need to add the rpc url that enables the trace call JSON-RPC")
+	eventRaw := `[{"address":"0x111111125421ca6dc452d289314280a0f8842a65","topics":["0xfec331350fce78ba658e082a71da20ac9f8d798a99b3c79681c8440cbfe77e07"],"data":"0x2c680fc902966ed935ddf9dc75550a32f1b2d1fba7dcaa7e4ab3b541734acddb00000000000000000000000000000000000000000000021246e3d52a827b39b7","blockNumber":"0x129d018","transactionHash":"0x4fa77df92aa4726ddbf54a41f72ca78a57035793c50e86643a25a058a27e35f4","transactionIndex":"0x6","blockHash":"0xa8eaa16d7ca8d934343bfe5fbb14bfeab272143f65fbd51f9caecf67754cfd29","logIndex":"0x53","removed":false},{"address":"0x111111125421ca6dc452d289314280a0f8842a65","topics":["0xfec331350fce78ba658e082a71da20ac9f8d798a99b3c79681c8440cbfe77e07"],"data":"0x2c680fc902966ed935ddf9dc75550a32f1b2d1fba7dcaa7e4ab3b541734acddb00000000000000000000000000000000000000000000020c219b06bffa614286","blockNumber":"0x129d018","transactionHash":"0x4fa77df92aa4726ddbf54a41f72ca78a57035793c50e86643a25a058a27e35f4","transactionIndex":"0x6","blockHash":"0xa8eaa16d7ca8d934343bfe5fbb14bfeab272143f65fbd51f9caecf67754cfd29","logIndex":"0x60","removed":false},{"address":"0x111111125421ca6dc452d289314280a0f8842a65","topics":["0xfec331350fce78ba658e082a71da20ac9f8d798a99b3c79681c8440cbfe77e07"],"data":"0xe0f3e2f9d0dbc1d72d355d2f80328b4e432d316c91a011d2fa26b56b7a39c371000000000000000000000000000000000000000000000000e1394e6d5fa7bd71","blockNumber":"0x129d018","transactionHash":"0x4fa77df92aa4726ddbf54a41f72ca78a57035793c50e86643a25a058a27e35f4","transactionIndex":"0x6","blockHash":"0xa8eaa16d7ca8d934343bfe5fbb14bfeab272143f65fbd51f9caecf67754cfd29","logIndex":"0x6a","removed":false}]`
+	events := []types.Log{}
+	err := json.Unmarshal([]byte(eventRaw), &events)
+	require.NoError(t, err)
+	rpcClient, err := rpcnode.NewClient(http.DefaultClient, rpcURL)
+	if err != nil {
+		panic(err)
+	}
+	traceCalls := tracecall.NewCache(rpcClient)
+	p := MustNewParser(traceCalls)
+	for _, event := range events {
+		log, err := p.Parse(event, uint64(time.Now().Unix()))
+		require.NoError(t, err)
+		require.Equal(t, log.EventHash, p.eventHash)
+		t.Log(log.Maker)
+		t.Log(log.MakerToken)
+		t.Log(log.MakerTokenAmount)
+		t.Log(log.Taker)
+		t.Log(log.TakerToken)
+		t.Log(log.TakerTokenAmount)
+	}
+}