init service to maintain promotees list

cmd/promotees/main.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+
+	libapp "github.com/KyberNetwork/tradelogs/internal/app"
+	"github.com/KyberNetwork/tradelogs/internal/dbutil"
+	"github.com/KyberNetwork/tradelogs/internal/promotionworker"
+	"github.com/KyberNetwork/tradelogs/pkg/evmlistenerclient"
+
+	server "github.com/KyberNetwork/tradelogs/internal/server/promotees"
+	"github.com/KyberNetwork/tradelogs/pkg/promotionparser"
+	"github.com/KyberNetwork/tradelogs/pkg/promotionparser/oneinchv2"
+	"github.com/KyberNetwork/tradelogs/pkg/promotionstorage"
+	"github.com/go-redis/redis/v8"
+	"github.com/jmoiron/sqlx"
+	"github.com/urfave/cli"
+	"go.uber.org/zap"
+)
+
+func main() {
+	app := libapp.NewApp()
+	app.Name = "Promotees fetcher service"
+	app.Action = run
+	app.Flags = append(app.Flags, libapp.PostgresSQLFlags("tradelogs")...)
+	app.Flags = append(app.Flags, libapp.RedisFlags()...)
+	app.Flags = append(app.Flags, libapp.EvmListenerFlags()...)
+	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("app starting ..")
+
+	db, err := initDB(c)
+	if err != nil {
+		l.Panicw("cannot init DB", "err", err)
+	}
+	redis, err := initRedis(c)
+	if err != nil {
+		l.Panicw("cannot init redis", "err", err)
+	}
+	s := promotionstorage.New(l, db)
+	listener := evmlistenerclient.New(l, libapp.EvmlistenerConfigFromCli(c), redis)
+	err = listener.Init(context.Background())
+	if err != nil {
+		l.Errorw("Error while init listener service")
+		return err
+	}
+
+	parsers := []promotionparser.Parser{oneinchv2.MustNewParser()}
+
+	w := promotionworker.New(l, s, listener, parsers)
+
+	apiServer := server.New(s, c.String(libapp.HTTPPromoteeServerFlag.Name))
+	go func() {
+		if err := apiServer.Run(); err != nil {
+			panic(err)
+		}
+	}()
+
+	return w.Run(context.Background())
+}
+
+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
+}
+
+func initRedis(c *cli.Context) (redis.UniversalClient, error) {
+	redis := redis.NewUniversalClient(libapp.RedisOptionFromFlags(c))
+	if _, err := redis.Ping(context.Background()).Result(); err != nil {
+		return nil, err
+	}
+	return redis, nil
+}

internal/app/server.go

@@ -29,11 +29,19 @@ var DuneKeyFlag = cli.StringFlag{
 	EnvVar: "DUNE_KEY",
 }
 
+var HTTPPromoteeServerFlag = cli.StringFlag{
+	Name:   "promotee-server-address",
+	Usage:  "Run the rest for promotees server",
+	EnvVar: "PROMOTEE_SERVER_ADDRESS",
+	Value:  "localhost:8082",
+}
+
 func HTTPServerFlags() []cli.Flag {
 	return []cli.Flag{
 		HTTPServerFlag,
 		HTTPBackfillServerFlag,
 		DuneURLFlag,
 		DuneKeyFlag,
+		HTTPPromoteeServerFlag,
 	}
 }

internal/promotionworker/proworker.go

@@ -0,0 +1,97 @@
+package promotionworker
+
+import (
+	"context"
+
+	"github.com/KyberNetwork/tradelogs/pkg/convert"
+	"github.com/KyberNetwork/tradelogs/pkg/evmlistenerclient"
+	"github.com/KyberNetwork/tradelogs/pkg/promotionparser"
+	"github.com/KyberNetwork/tradelogs/pkg/promotionstorage"
+	ethTypes "github.com/ethereum/go-ethereum/core/types"
+	"go.uber.org/zap"
+)
+
+type Proworker struct {
+	listener *evmlistenerclient.Client
+	l        *zap.SugaredLogger
+	s        *promotionstorage.Storage
+	p        []promotionparser.Parser
+}
+
+func New(l *zap.SugaredLogger, s *promotionstorage.Storage, listener *evmlistenerclient.Client,
+	parsers []promotionparser.Parser) *Proworker {
+	return &Proworker{
+		listener: listener,
+		l:        l,
+		s:        s,
+		p:        parsers,
+	}
+}
+
+func (w *Proworker) Run(ctx context.Context) error {
+	for {
+		m, err := w.listener.GConsume(ctx)
+		if err != nil {
+			w.l.Errorw("Error while consume in group")
+			return err
+		}
+		w.l.Infow("process msg", "count", len(m))
+		if len(m) == 0 {
+			continue
+		}
+		if err := w.processMessages(m); err != nil {
+			w.l.Errorw("Error when proccess msg", "error", err)
+			return err
+		}
+		if err := w.listener.Ack(ctx, m); err != nil {
+			w.l.Errorw("Error when ack msg", "error", err)
+			return err
+		}
+	}
+}
+
+func (w *Proworker) processMessages(m []evmlistenerclient.Message) error {
+	for _, message := range m {
+		var (
+			insertPromotees []promotionstorage.Promotee
+			deleteBlocks    []uint64
+		)
+		for _, block := range message.NewBlocks {
+			for _, log := range block.Logs {
+				ethLog := convert.ToETHLog(log)
+				ps := w.findMatchingParser(ethLog)
+				if ps == nil {
+					continue
+				}
+				promotee, err := ps.Parse(ethLog, block.Timestamp)
+				if err != nil {
+					continue
+				}
+				insertPromotees = append(insertPromotees, promotee)
+			}
+		}
+		for _, block := range message.RevertedBlocks {
+			deleteBlocks = append(deleteBlocks, block.Number.Uint64())
+		}
+
+		if err := w.s.Delete(deleteBlocks); err != nil {
+			return err
+		}
+		if err := w.s.Insert(insertPromotees); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *Proworker) findMatchingParser(log ethTypes.Log) promotionparser.Parser {
+	var ps promotionparser.Parser
+	for _, p := range w.p {
+		if !p.LogFromContract(log) {
+			continue
+		}
+		ps = p
+		break
+	}
+	return ps
+}

internal/server/promotees/server.go

@@ -0,0 +1,80 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/KyberNetwork/tradelogs/pkg/promotionstorage"
+	"github.com/gin-contrib/pprof"
+	"github.com/gin-gonic/gin"
+)
+
+type Server struct {
+	r        *gin.Engine
+	bindAddr string
+	s        *promotionstorage.Storage
+}
+
+func New(s *promotionstorage.Storage, bindAddr string) *Server {
+	engine := gin.New()
+	engine.Use(gin.Recovery())
+
+	server := &Server{
+		r:        engine,
+		bindAddr: bindAddr,
+		s:        s,
+	}
+
+	gin.SetMode(gin.DebugMode)
+	server.register()
+
+	return server
+}
+
+func (s *Server) Run() error {
+	if err := s.r.Run(s.bindAddr); err != nil {
+		return fmt.Errorf("run server: %w", err)
+	}
+
+	return nil
+}
+
+func (s *Server) register() {
+	pprof.Register(s.r, "/debug")
+	s.r.GET("/promotees", s.getPromotees)
+}
+
+func responseErr(c *gin.Context, status int, err error) {
+	c.JSON(http.StatusBadRequest, gin.H{
+		"success": false,
+		"error":   err.Error(),
+		"status":  status,
+	})
+}
+
+func (s *Server) getPromotees(c *gin.Context) {
+	var (
+		query promotionstorage.PromoteesQuery
+		err   = c.ShouldBind(&query)
+	)
+	if query.Promotee != "" {
+		query.Promotee = strings.ToLower(query.Promotee)
+	}
+	if query.Promoter != "" {
+		query.Promoter = strings.ToLower(query.Promoter)
+	}
+	if err != nil {
+		responseErr(c, http.StatusBadRequest, err)
+		return
+	}
+	data, err := s.s.Get(query)
+	if err != nil {
+		responseErr(c, http.StatusBadRequest, err)
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{
+		"success": true,
+		"data":    data,
+	})
+}

pkg/evmlistenerclient/client.go

@@ -38,7 +38,7 @@ func New(l *zap.SugaredLogger, cfg Config, client redis.UniversalClient) *Client
 		client:    client,
 		config:    cfg,
 		l:         l,
-		groupName: "trading-tradelogs",
+		groupName: cfg.GroupName,
 	}
 }