Merge pull request #31 from KyberNetwork/feat/backfill-oneinch

feat: backfill oneinch

internal/bigquery/bigquery.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"cloud.google.com/go/bigquery"
-	"github.com/KyberNetwork/tradelogs/pkg/storage"
 	"github.com/KyberNetwork/tradelogs/pkg/parser"
+	"github.com/KyberNetwork/tradelogs/pkg/storage"
 	"go.uber.org/zap"
 	"google.golang.org/api/iterator"
 )
@@ -271,3 +271,13 @@ func (w *Worker) BackFillPartialData(fromTime, toTime int64, exchanges []string)
 	go w.run(minTime, maxTime, topics)
 	return nil
 }
+
+func (w *Worker) BackfillOneInchRFQ(tradeLogs []storage.TradeLog) error {
+	l := w.l.With("exchange", "1inch")
+	err := w.storage.Insert(tradeLogs)
+	if err != nil {
+		l.Errorw("Failed to insert the trade logs when backfilling the 1inch rfq trades", "err", err)
+		return err
+	}
+	return nil
+}

internal/server/backfill/server.go

@@ -50,7 +50,7 @@ func (s *Server) Run() error {
 func (s *Server) register() {
 	pprof.Register(s.r, "/debug")
 	s.r.POST("/backfill", s.backfill)
-
+	s.r.POST("/backfill-1inch", s.backfillOneinch)
 }
 
 func responseErr(c *gin.Context, err error) {
@@ -94,3 +94,26 @@ func (s *Server) backfill(c *gin.Context) {
 		"time":    time.Now().UnixMilli(),
 	})
 }
+
+func (s *Server) backfillOneinch(c *gin.Context) {
+	var (
+		query BackFillOneInchRequest
+	)
+	if err := c.ShouldBind(&query); err != nil {
+		responseErr(c, err)
+		return
+	}
+
+	tradeLogs := query.ToTradeLogs()
+
+	err := s.bq.BackfillOneInchRFQ(tradeLogs)
+	if err != nil {
+		responseErr(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{
+		"message": "Backfill 1inch rfq orders successfully",
+		"success": true,
+		"time":    time.Now().UnixMilli(),
+	})
+}

internal/server/backfill/types.go

@@ -0,0 +1,60 @@
+package server
+
+import (
+	"github.com/KyberNetwork/tradelogs/pkg/parser/oneinch"
+	"github.com/KyberNetwork/tradelogs/pkg/storage"
+)
+
+var (
+	oneinchAbi, _       = oneinch.OneinchMetaData.GetAbi()
+	oneinchRfqEvent, _  = oneinchAbi.Events[oneinch.FilledEvent]
+	oneinchRfqEventHash = oneinchRfqEvent.String()
+)
+
+type rfqOrder struct {
+	OrderHash        string `json:"order_hash" binding:"required"`
+	BlockNumber      uint64 `json:"block_number" binding:"required"`
+	LogIndex         uint64 `json:"log_index" binding:"required"`
+	EventHash        string `json:"event_hash" binding:"required"`
+	TxHash           string `json:"tx_hash" binding:"required"`
+	Maker            string `json:"maker"`
+	Taker            string `json:"taker"`
+	MakerToken       string `json:"maker_token"`
+	TakerToken       string `json:"taker_token"`
+	MakerTokenAmount string `json:"maker_token_amount"`
+	TakerTokenAmount string `json:"taker_token_amount"`
+	ContractAddress  string `json:"contract_address"`
+	Timestamp        uint64 `json:"timestamp"`
+}
+type BackFillOneInchRequest []rfqOrder
+
+func (rfqOrders BackFillOneInchRequest) ToTradeLogs() []storage.TradeLog {
+	tradeLogs := make([]storage.TradeLog, 0, len(rfqOrders))
+	for _, order := range rfqOrders {
+		if order.EventHash != oneinchRfqEventHash {
+			continue
+		}
+		if order.LogIndex < 0 || order.BlockNumber < 0 {
+			continue
+		}
+		if order.TxHash == "" {
+			continue
+		}
+		tradeLogs = append(tradeLogs, storage.TradeLog{
+			OrderHash:        order.OrderHash,
+			BlockNumber:      order.BlockNumber,
+			LogIndex:         order.LogIndex,
+			EventHash:        order.EventHash,
+			TxHash:           order.TxHash,
+			Maker:            order.Maker,
+			Taker:            order.Taker,
+			MakerToken:       order.MakerToken,
+			TakerToken:       order.TakerToken,
+			MakerTokenAmount: order.MakerTokenAmount,
+			TakerTokenAmount: order.TakerTokenAmount,
+			ContractAddress:  order.ContractAddress,
+			Timestamp:        order.Timestamp,
+		})
+	}
+	return tradeLogs
+}