TRD-573 using tradelog as struct, add dockerfile & readme

Dockerfile

@@ -7,6 +7,8 @@ COPY . .
 
 RUN go build -o app ./cmd/tradelogs
 
+RUN go build -o parse_log ./v2/cmd/parse_log
+
 
 ## DEPLOY
 FROM debian:bullseye
@@ -15,10 +17,16 @@ RUN apt-get update && \
     apt install -y ca-certificates && \
     rm -rf /var/lib/apt/lists/*
 
+### tradelogs v1
 WORKDIR /cmd
 
 COPY --from=builder /src/app /cmd/app
 
 COPY cmd/tradelogs/migrations migrations
 
+### tradelogs v2
+WORKDIR /parse_log
+COPY --from=builder /src/parse_log /parse_log/parse_log
+COPY v2/cmd/parse_log/migrations migrations
+
 ENTRYPOINT /cmd/app

README.md

@@ -1 +1,22 @@
 ## This is service to indexing tradelogs from event
+
+### Re-generate mock file
+
+First, you need to install `mockery`
+
+- https://vektra.github.io/mockery/latest/installation/
+- https://github.com/vektra/mockery
+
+Then you use the `mockery` to generate mock files from interface
+
+```
+mockery --dir=<interface directory> --name=<interface name> --output=<output dir> --structname=<name of output struct> --filename=<output filename>
+```
+
+Example: generate `Storage` interface to a struct name `MockStorage`
+
+```
+mockery --dir=v2/pkg/storage/state --name=Storage --output=v2/mocks/ --structname=MockState --filename=State.go
+```
+
+For more information `mockery --help`

v2/cmd/parse_log/main.go

@@ -58,7 +58,7 @@ func run(c *cli.Context) error {
 
 	db, err := initDB(c)
 	if err != nil {
-		l.Panicw("cannot init DB", "err", err)
+		return fmt.Errorf("cannot init DB: %w", err)
 	}
 
 	// trade log manager
@@ -78,7 +78,7 @@ func run(c *cli.Context) error {
 	for i, url := range rpcURL {
 		client, err := ethclient.Dial(url)
 		if err != nil {
-			panic(fmt.Errorf("cannot dial eth client: %w", err))
+			return fmt.Errorf("cannot dial eth client: %w", err)
 		}
 		ethClients[i] = client
 	}
@@ -101,13 +101,13 @@ func run(c *cli.Context) error {
 	broadcastTopic := c.String(libapp.KafkaBroadcastTopic.Name)
 	err = kafka.ValidateTopicName(broadcastTopic)
 	if err != nil {
-		panic(fmt.Errorf("invalid kafka topic: %w", err))
+		return fmt.Errorf("invalid kafka topic: %w", err)
 	}
 
 	// kafka publisher for broadcasting trade logs
 	kafkaPublisher, err := kafka.NewPublisher(libapp.KafkaConfigFromFlags(c))
 	if err != nil {
-		panic(fmt.Errorf("cannot create kafka publisher: %w", err))
+		return fmt.Errorf("cannot create kafka publisher: %w", err)
 	}
 
 	// trade log handler
@@ -118,7 +118,7 @@ func run(c *cli.Context) error {
 
 	mostRecentBlock, err := getMostRecentBlock(l, s, rpcNode)
 	if err != nil {
-		panic(fmt.Errorf("cannot get most recent block: %w", err))
+		return fmt.Errorf("cannot get most recent block: %w", err)
 	}
 
 	// subscribe evm listener with worker as a consumer
@@ -129,6 +129,7 @@ func run(c *cli.Context) error {
 		c.Int(libapp.EVMListenerMaxTrackingBlock.Name),
 		w,
 		mostRecentBlock,
+		c.Int(libapp.EVMListenerClientTimeoutSecondFlag.Name),
 	)
 }
 

v2/internal/app/evmlistener.go

@@ -5,10 +5,10 @@ import (
 )
 
 const (
-	EVMHTTPRPCUrlFlagName = "http-rpc-url"
-	EVMwsRPCUrlFlagName   = "ws-rpc-url"
-	EVMMaxTrackingBlock   = "max-tracking-block"
-	EVMBlockExpiration    = "block-expiration"
+	EVMHTTPRPCUrlFlagName  = "http-rpc-url"
+	EVMwsRPCUrlFlagName    = "ws-rpc-url"
+	EVMMaxTrackingBlock    = "max-tracking-block"
+	EVMClientTimeoutSecond = "timeout-second"
 )
 
 //nolint:gochecknoglobals
@@ -17,26 +17,30 @@ var (
 		Name:   EVMHTTPRPCUrlFlagName,
 		EnvVar: "EVM_HTTP_RPC_URL",
 		Usage:  "HTTP RPC node url for EVM Listener",
-		Value:  "https://ethereum.kyberengineering.io/trading-tokyo",
 	}
 	EVMListenerWsRPCUrlFlag = &cli.StringFlag{
 		Name:   EVMwsRPCUrlFlagName,
 		EnvVar: "EVM_WS_RPC_URL",
 		Usage:  "Web socket RPC node url for EVM Listener",
-		Value:  "wss://ethereum.kyberengineering.io/trading-tokyo",
 	}
 	EVMListenerMaxTrackingBlock = &cli.IntFlag{
 		Name:   EVMMaxTrackingBlock,
 		EnvVar: "EVM_MAX_TRACKING_BLOCK",
 		Usage:  "Max tracking block number for block keeper",
 		Value:  32,
 	}
+	EVMListenerClientTimeoutSecondFlag = &cli.IntFlag{
+		Name:   EVMClientTimeoutSecond,
+		EnvVar: "EVM_CLIENT_TIMEOUT_SECOND",
+		Usage:  "Client timeout second",
+		Value:  15,
+	}
 )
 
 // NewEvmListenerFlags returns flags for evmlistener.
 func EvmListenerFlags() []cli.Flag {
 	return []cli.Flag{
 		EVMListenerHTTPRPCUrlFlag, EVMListenerWsRPCUrlFlag,
-		EVMListenerMaxTrackingBlock,
+		EVMListenerMaxTrackingBlock, EVMListenerClientTimeoutSecondFlag,
 	}
 }

v2/pkg/constant/exchange.go

@@ -0,0 +1,33 @@
+package constant
+
+const (
+	ExZeroX         = "zerox"
+	EXZeroXV3       = "zeroxV3"
+	ExParaswap      = "paraswap"
+	ExTokenlon      = "tokenlon"
+	ExZeroXRFQ      = "zerox_rfq"
+	ExKs            = "kyberswap"
+	ExHashflow      = "hashflow"
+	ExKsRFQ         = "kyberswapRFQ"
+	Ex1Inch         = "1inch"
+	Ex1InchV6       = "1inchV6"
+	ExHashflowV3    = "hashflowV3"
+	ExParaswapTaker = "paraswap_taker"
+	ExUniswapX      = "uniswapx"
+	ExNative        = "native"
+	ExBebop         = "bebop"
+	ExUniswapXV1    = "uniswapxV1"
+
+	Addr1InchV6      = "0x111111125421cA6dc452d289314280a0f8842A65"
+	AddrBebop        = "0xbbbbbBB520d69a9775E85b458C58c648259FAD5F"
+	AddrHashflowV3   = "0x24b9d98FABF4DA1F69eE10775F240AE3dA6856fd"
+	AddrKyberswap    = "0x6131B5fae19EA4f9D964eAc0408E4408b66337b5"
+	AddrKyberswapRFQ = "0x7A819Fa46734a49D0112796f9377E024c350FB26"
+	AddrParaswap     = "0xe92b586627ccA7a83dC919cc7127196d70f55a06"
+	AddrUniswapX     = "0x00000011F84B9aa48e5f8aA8B9897600006289Be"
+	AddrUniswapXV1   = "0x6000da47483062A0D734Ba3dc7576Ce6A0B645C4"
+	Addr0x           = "0xDef1C0ded9bec7F1a1670819833240f027b25EfF"
+	Deployer0xV3     = "0x00000000000004533Fe15556B1E086BB1A72cEae"
+
+	TableZeroX = "tradelogs_zerox"
+)

v2/pkg/evmlistenerclient/client.go

@@ -14,9 +14,9 @@ import (
 )
 
 func SubscribeEvent(l *zap.SugaredLogger, wsRPC, httpRPC string, maxTrackingBlock int,
-	publisher pubsub.Publisher, processedBlock uint64) error {
+	publisher pubsub.Publisher, processedBlock uint64, timeoutSecond int) error {
 	httpClient := &http.Client{
-		Timeout:   time.Second * 15,
+		Timeout:   time.Second * time.Duration(timeoutSecond),
 		Transport: &http.Transport{},
 	}
 

v2/pkg/handler/trade_logs.go

@@ -30,13 +30,11 @@ type TradeLogHandler struct {
 }
 
 type logMetadata struct {
-	blockNumber      uint64
-	blockHash        string
-	txHash           string
-	txIndex          int
-	timestamp        uint64
-	messageSender    string
-	interactContract string
+	blockNumber uint64
+	blockHash   string
+	txHash      string
+	txIndex     int
+	timestamp   uint64
 }
 
 func NewTradeLogHandler(l *zap.SugaredLogger, rpc *rpcnode.Client, storage *tradelogs.Manager, parsers []parser.Parser,
@@ -61,20 +59,23 @@ func (h *TradeLogHandler) ProcessBlock(blockHash string, blockNumber uint64, tim
 	for i, call := range calls {
 		logIndexStart = assignLogIndexes(&call.CallFrame, logIndexStart)
 		metadata := logMetadata{
-			blockNumber:      blockNumber,
-			blockHash:        blockHash,
-			txHash:           call.TxHash,
-			txIndex:          i,
-			timestamp:        timestamp,
-			messageSender:    call.CallFrame.From,
-			interactContract: call.CallFrame.To,
+			blockNumber: blockNumber,
+			blockHash:   blockHash,
+			txHash:      call.TxHash,
+			txIndex:     i,
+			timestamp:   timestamp,
 		}
 
 		tradeLogs := h.processCallFrame(call.CallFrame, metadata)
 		if len(tradeLogs) == 0 {
 			continue
 		}
 
+		for j := range tradeLogs {
+			tradeLogs[j].MessageSender = call.CallFrame.From
+			tradeLogs[j].InteractContract = call.CallFrame.To
+		}
+
 		err = h.storage.Insert(tradeLogs)
 		if err != nil {
 			return fmt.Errorf("write to storage error: %w", err)
@@ -83,7 +84,10 @@ func (h *TradeLogHandler) ProcessBlock(blockHash string, blockNumber uint64, tim
 
 		passCount, failCount := 0, 0
 		for _, log := range tradeLogs {
-			msgBytes, err := json.Marshal(log)
+			msgBytes, err := json.Marshal(kafka.Message{
+				Type: kafka.MessageTypeTradeLog,
+				Data: log,
+			})
 			if err != nil {
 				h.l.Errorw(" error when marshal trade log to json", "blockNumber", blockNumber, "log", log, "err", err)
 				failCount++
@@ -140,11 +144,7 @@ func (h *TradeLogHandler) processCallFrame(call types.CallFrame, metadata logMet
 		}
 
 		// parse trade log
-		tradeLogs, err := p.ParseWithCallFrame(
-			call, ethLog, metadata.timestamp,
-			storageTypes.WithMessageSender(metadata.messageSender),
-			storageTypes.WithInteractContract(metadata.interactContract),
-		)
+		tradeLogs, err := p.ParseWithCallFrame(call, ethLog, metadata.timestamp)
 		if err != nil {
 			h.l.Errorw("error when parse log", "log", ethLog, "err", err, "parser", p.Exchange())
 			continue
@@ -164,10 +164,28 @@ func (h *TradeLogHandler) findMatchingParser(log ethereumTypes.Log) parser.Parse
 }
 
 func (h *TradeLogHandler) RevertBlock(blocks []uint64) error {
+	if len(blocks) == 0 {
+		return nil
+	}
+
 	err := h.storage.Delete(blocks)
 	if err != nil {
 		return fmt.Errorf("delete blocks error: %w", err)
 	}
+
+	msgBytes, err := json.Marshal(kafka.Message{
+		Type: kafka.MessageTypeRevert,
+		Data: blocks,
+	})
+	if err != nil {
+		h.l.Errorw(" error when marshal revert message to json", "err", err)
+	}
+	err = h.publisher.Publish(h.kafkaTopic, msgBytes)
+	if err != nil {
+		h.l.Errorw("error when publish revert message", "err", err)
+	}
+	h.l.Infow("published revert message", "message", msgBytes)
+
 	return nil
 }
 

v2/pkg/handler/trade_logs_test.go

@@ -31,7 +31,7 @@ func TestTradeLogHandler_ProcessBlock(t *testing.T) {
 	client := rpcnode.NewClient(zap.S(), ethClient)
 
 	mockStorage := &mocks.MockStorage{}
-	mockStorage.On("Type").Return("zerox").
+	mockStorage.On("Exchange").Return("zerox").
 		On("Insert", mock.Anything).Return(nil)
 	s := tradelogs.NewManager(zap.S(), []types.Storage{mockStorage})
 

v2/pkg/kafka/types.go

@@ -0,0 +1,21 @@
+package kafka
+
+type Config struct {
+	Addresses []string
+
+	UseAuthentication bool
+	Username          string
+	Password          string
+}
+
+type Message struct {
+	Type MessageType `json:"type"`
+	Data interface{} `json:"data"`
+}
+
+type MessageType string
+
+const (
+	MessageTypeTradeLog MessageType = "trade_log"
+	MessageTypeRevert   MessageType = "revert"
+)

v2/pkg/parser/parser.go

@@ -8,36 +8,6 @@ import (
 	ethereumTypes "github.com/ethereum/go-ethereum/core/types"
 )
 
-const (
-	ExZeroX         = "zerox"
-	EXZeroXV3       = "zeroxV3"
-	ExParaswap      = "paraswap"
-	ExTokenlon      = "tokenlon"
-	ExZeroXRFQ      = "zerox_rfq"
-	ExKs            = "kyberswap"
-	ExHashflow      = "hashflow"
-	ExKsRFQ         = "kyberswapRFQ"
-	Ex1Inch         = "1inch"
-	Ex1InchV6       = "1inchV6"
-	ExHashflowV3    = "hashflowV3"
-	ExParaswapTaker = "paraswap_taker"
-	ExUniswapX      = "uniswapx"
-	ExNative        = "native"
-	ExBebop         = "bebop"
-	ExUniswapXV1    = "uniswapxV1"
-
-	Addr1InchV6      = "0x111111125421cA6dc452d289314280a0f8842A65"
-	AddrBebop        = "0xbbbbbBB520d69a9775E85b458C58c648259FAD5F"
-	AddrHashflowV3   = "0x24b9d98FABF4DA1F69eE10775F240AE3dA6856fd"
-	AddrKyberswap    = "0x6131B5fae19EA4f9D964eAc0408E4408b66337b5"
-	AddrKyberswapRFQ = "0x7A819Fa46734a49D0112796f9377E024c350FB26"
-	AddrParaswap     = "0xe92b586627ccA7a83dC919cc7127196d70f55a06"
-	AddrUniswapX     = "0x00000011F84B9aa48e5f8aA8B9897600006289Be"
-	AddrUniswapXV1   = "0x6000da47483062A0D734Ba3dc7576Ce6A0B645C4"
-	Addr0x           = "0xDef1C0ded9bec7F1a1670819833240f027b25EfF"
-	Deployer0xV3     = "0x00000000000004533Fe15556B1E086BB1A72cEae"
-)
-
 var (
 	ErrInvalidTopic  = errors.New("invalid order topic")
 	ErrNotFoundTrade = errors.New("not found log")
@@ -48,6 +18,6 @@ type Parser interface {
 	Topics() []string
 	Exchange() string
 	UseTraceCall() bool
-	ParseWithCallFrame(callFrame types.CallFrame, log ethereumTypes.Log, blockTime uint64, options ...storageTypes.Option) ([]storageTypes.TradeLog, error)
+	ParseWithCallFrame(callFrame types.CallFrame, log ethereumTypes.Log, blockTime uint64) ([]storageTypes.TradeLog, error)
 	LogFromExchange(log ethereumTypes.Log) bool
 }