TRD-684 Price Filler & Tradelogs Query Server for Tradelogs v2 (#90)

* TRD-684 price filler & tradelogs query server

* TRD-684 update unit test

* TRD-684 remove state column

Dockerfile-v2

@@ -8,6 +8,8 @@ COPY . .
 RUN go build -o parse_log ./v2/cmd/parse_log
 RUN go build -o backfill ./v2/cmd/backfill
 RUN go build -o broadcast ./v2/cmd/broadcast
+RUN go build -o price_filler ./v2/cmd/price_filler
+RUN go build -o tradelogs ./v2/cmd/tradelogs
 
 
 ## DEPLOY
@@ -23,6 +25,8 @@ WORKDIR /v2
 COPY --from=builder /src/parse_log /v2/parse_log
 COPY --from=builder /src/backfill /v2/backfill
 COPY --from=builder /src/broadcast /v2/broadcast
+COPY --from=builder /src/price_filler /v2/price_filler
+COPY --from=builder /src/tradelogs /v2/tradelogs
 
 COPY v2/cmd/migrations /v2/migrations
 

v2/cmd/migrations/00004_update_price_column.up.sql

@@ -0,0 +1,25 @@
+alter table tradelogs_zerox
+    alter column maker_token_price drop not null;
+
+alter table tradelogs_zerox
+    alter column maker_token_price drop default;
+
+alter table tradelogs_zerox
+    alter column taker_token_price drop not null;
+
+alter table tradelogs_zerox
+    alter column taker_token_price drop default;
+
+alter table tradelogs_zerox
+    alter column maker_usd_amount drop not null;
+
+alter table tradelogs_zerox
+    alter column maker_usd_amount drop default;
+
+alter table tradelogs_zerox
+    alter column taker_usd_amount drop not null;
+
+alter table tradelogs_zerox
+    alter column taker_usd_amount drop default;
+
+alter table tradelogs_zerox drop column state;

v2/cmd/price_filler/main.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/KyberNetwork/go-binance/v2"
+	libapp "github.com/KyberNetwork/tradelogs/v2/pkg/app"
+	"github.com/KyberNetwork/tradelogs/v2/pkg/price_filler"
+	storageTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/types"
+	zxotcStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/zxotc"
+	"github.com/KyberNetwork/tradinglib/pkg/dbutil"
+	"github.com/jmoiron/sqlx"
+	"github.com/urfave/cli"
+	"go.uber.org/zap"
+)
+
+func main() {
+	app := libapp.NewApp()
+	app.Name = "trade logs crawler service"
+	app.Action = run
+	app.Flags = append(app.Flags, libapp.PostgresSQLFlags("tradelogs_v2")...)
+	app.Flags = append(app.Flags, libapp.PriceFillerFlags()...)
+
+	if err := app.Run(os.Args); err != nil {
+		log.Panic(err)
+	}
+}
+
+func run(c *cli.Context) error {
+	logger, _, flush, err := libapp.NewLogger(c)
+	if err != nil {
+		return fmt.Errorf("new logger: %w", err)
+	}
+
+	defer flush()
+
+	zap.ReplaceGlobals(logger)
+	l := logger.Sugar()
+	l.Infow("Starting price filler service")
+
+	db, err := initDB(c)
+	l.Infow("init db successfully")
+	if err != nil {
+		return fmt.Errorf("cannot init DB: %w", err)
+	}
+
+	// trade log storages
+	s := []storageTypes.Storage{
+		zxotcStorage.New(l, db),
+	}
+
+	binanceClient := binance.NewClient(c.String(libapp.BinanceAPIKeyFlag.Name), c.String(libapp.BinanceSecretKeyFlag.Name))
+	priceFiller, err := pricefiller.NewPriceFiller(l, binanceClient, s)
+	if err != nil {
+		l.Errorw("Error while init price filler")
+		return err
+	}
+	priceFiller.Run()
+	return nil
+}
+
+func initDB(c *cli.Context) (*sqlx.DB, error) {
+	db, err := libapp.NewDB(map[string]interface{}{
+		"host":     c.String(libapp.PostgresHost.Name),
+		"port":     c.Int(libapp.PostgresPort.Name),
+		"user":     c.String(libapp.PostgresUser.Name),
+		"password": c.String(libapp.PostgresPassword.Name),
+		"dbname":   c.String(libapp.PostgresDatabase.Name),
+		"sslmode":  "disable",
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = dbutil.RunMigrationUp(db.DB, c.String(libapp.PostgresMigrationPath.Name),
+		c.String(libapp.PostgresDatabase.Name))
+	if err != nil {
+		return nil, err
+	}
+	return db, nil
+}

v2/cmd/tradelogs/main.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/KyberNetwork/tradelogs/v2/internal/server"
+	libapp "github.com/KyberNetwork/tradelogs/v2/pkg/app"
+	storageTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/types"
+	zxotcStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/zxotc"
+	"github.com/KyberNetwork/tradinglib/pkg/dbutil"
+	"github.com/jmoiron/sqlx"
+	"github.com/urfave/cli"
+	"go.uber.org/zap"
+)
+
+func main() {
+	app := libapp.NewApp()
+	app.Name = "trade logs crawler service"
+	app.Action = run
+	app.Flags = append(app.Flags, libapp.PostgresSQLFlags("tradelogs_v2")...)
+	app.Flags = append(app.Flags, libapp.HTTPServerFlags()...)
+
+	if err := app.Run(os.Args); err != nil {
+		log.Panic(err)
+	}
+}
+
+func run(c *cli.Context) error {
+	logger, _, flush, err := libapp.NewLogger(c)
+	if err != nil {
+		return fmt.Errorf("new logger: %w", err)
+	}
+
+	defer flush()
+
+	zap.ReplaceGlobals(logger)
+	l := logger.Sugar()
+	l.Infow("Starting trade logs server")
+
+	db, err := initDB(c)
+	l.Infow("init db successfully")
+	if err != nil {
+		return fmt.Errorf("cannot init DB: %w", err)
+	}
+
+	// trade log storages
+	storage := []storageTypes.Storage{
+		zxotcStorage.New(l, db),
+	}
+
+	s := server.NewTradeLogs(l, storage, c.String(libapp.HTTPTradeLogsServerFlag.Name))
+	return s.Run()
+}
+
+func initDB(c *cli.Context) (*sqlx.DB, error) {
+	db, err := libapp.NewDB(map[string]interface{}{
+		"host":     c.String(libapp.PostgresHost.Name),
+		"port":     c.Int(libapp.PostgresPort.Name),
+		"user":     c.String(libapp.PostgresUser.Name),
+		"password": c.String(libapp.PostgresPassword.Name),
+		"dbname":   c.String(libapp.PostgresDatabase.Name),
+		"sslmode":  "disable",
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = dbutil.RunMigrationUp(db.DB, c.String(libapp.PostgresMigrationPath.Name),
+		c.String(libapp.PostgresDatabase.Name))
+	if err != nil {
+		return nil, err
+	}
+	return db, nil
+}

v2/internal/server/tradelogs.go

@@ -0,0 +1,91 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	storageTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/types"
+	"github.com/gin-contrib/pprof"
+	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+)
+
+var (
+	maxTimeRange = uint64(7 * 24 * time.Hour.Milliseconds())
+)
+
+type TradeLogs struct {
+	r        *gin.Engine
+	bindAddr string
+	l        *zap.SugaredLogger
+	storage  []storageTypes.Storage
+}
+
+func NewTradeLogs(l *zap.SugaredLogger, s []storageTypes.Storage, bindAddr string) *TradeLogs {
+	engine := gin.New()
+	engine.Use(gin.Recovery())
+
+	server := &TradeLogs{
+		r:        engine,
+		bindAddr: bindAddr,
+		l:        l,
+		storage:  s,
+	}
+
+	gin.SetMode(gin.ReleaseMode)
+	server.register()
+
+	return server
+}
+
+// Run runs server.
+func (s *TradeLogs) Run() error {
+	if err := s.r.Run(s.bindAddr); err != nil {
+		return fmt.Errorf("run server: %w", err)
+	}
+
+	return nil
+}
+
+func (s *TradeLogs) register() {
+	pprof.Register(s.r, "/debug")
+	s.r.GET("/tradelogs", s.getTradeLogs)
+}
+
+func (s *TradeLogs) getTradeLogs(c *gin.Context) {
+	var (
+		query storageTypes.TradeLogsQuery
+		err   = c.ShouldBind(&query)
+	)
+
+	if err != nil {
+		responseErr(c, http.StatusBadRequest, err)
+		return
+	}
+
+	if query.ToTime < query.FromTime {
+		responseErr(c, http.StatusBadRequest, fmt.Errorf("to_time cannot smaller than from_time"))
+		return
+	}
+
+	if query.ToTime-query.FromTime > maxTimeRange {
+		responseErr(c, http.StatusBadRequest, fmt.Errorf("max time range: %v", maxTimeRange))
+		return
+	}
+
+	var data []storageTypes.TradeLog
+	for _, storage := range s.storage {
+		tradeLogs, err := storage.Get(query)
+		if err != nil {
+			responseErr(c, http.StatusBadRequest, err)
+			return
+		}
+		data = append(data, tradeLogs...)
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"success": true,
+		"data":    data,
+	})
+}