tx_indexer

api/plugin.go

@@ -18,12 +18,12 @@ import (
 	v1 "github.com/vultisig/commondata/go/vultisig/vault/v1"
 	"github.com/vultisig/mobile-tss-lib/tss"
 	vcommon "github.com/vultisig/verifier/common"
+	"github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	vtypes "github.com/vultisig/verifier/types"
 
 	"github.com/vultisig/plugin/common"
 	"github.com/vultisig/plugin/internal/sigutil"
 	"github.com/vultisig/plugin/internal/tasks"
-	"github.com/vultisig/plugin/internal/types"
 )
 
 type ErrorResponse struct {
@@ -99,21 +99,36 @@ func (s *Server) SignPluginMessages(c echo.Context) error {
 
 	req.Parties = []string{common.PluginPartyID, common.VerifierPartyID}
 
+	if s.txIndexerService != nil {
+		for i, msg := range req.Messages {
+			txToTrack, e := s.txIndexerService.CreateTx(
+				c.Request().Context(),
+				storage.CreateTxDto{
+					PluginID:      policy.PluginID,
+					ChainID:       msg.Chain,
+					PolicyID:      policy.ID,
+					FromPublicKey: req.PublicKey,
+					ProposedTxHex: req.Transaction,
+				},
+			)
+			if e != nil {
+				return fmt.Errorf("s.txIndexerService.CreateTx: %w", e)
+			}
+			req.Messages[i].TxIndexerID = txToTrack.ID.String()
+		}
+	}
+
 	buf, err := json.Marshal(req)
 	if err != nil {
 		return fmt.Errorf("fail to marshal to json, err: %w", err)
 	}
 
 	// TODO: check if this is relevant
 	// check that tx is done only once per period
+	// (by @webpiratt: period depends on particular dca/payroll strategy (monthly/daily tx send, etc),
+	// maybe consider to add some additional metadata there to enable tx replay protection correctly)
 	// should we also copy the db to the vultiserver, so that it can be used by the vultiserver (and use scheduler.go)? or query the blockchain?
 
-	txToSign, err := s.db.GetTransactionByHash(c.Request().Context(), txHash)
-	if err != nil {
-		s.logger.Errorf("Failed to get transaction by hash from database: %v", err)
-		return fmt.Errorf("fail to get transaction by hash: %w", err)
-	}
-
 	s.logger.Debug("PLUGIN SERVER: KEYSIGN TASK")
 
 	ti, err := s.client.EnqueueContext(c.Request().Context(),
@@ -122,22 +137,10 @@ func (s *Server) SignPluginMessages(c echo.Context) error {
 		asynq.Timeout(2*time.Minute),
 		asynq.Retention(5*time.Minute),
 		asynq.Queue(tasks.QUEUE_NAME))
-
 	if err != nil {
-		txToSign.Metadata["error"] = err.Error()
-		if updateErr := s.db.UpdateTransactionStatus(c.Request().Context(), txToSign.ID, types.StatusSigningFailed, txToSign.Metadata); updateErr != nil {
-			s.logger.Errorf("Failed to update transaction status: %v", updateErr)
-		}
 		return fmt.Errorf("fail to enqueue keysign task: %w", err)
 	}
 
-	txToSign.Metadata["task_id"] = ti.ID
-	if err := s.db.UpdateTransactionStatus(c.Request().Context(), txToSign.ID, types.StatusSigned, txToSign.Metadata); err != nil {
-		s.logger.Errorf("Failed to update transaction with task ID: %v", err)
-	}
-
-	s.logger.Infof("Created transaction history for tx from plugin: %s...", req.Transaction[:min(20, len(req.Transaction))])
-
 	return c.JSON(http.StatusOK, ti.ID)
 }
 

api/server.go

@@ -17,6 +17,7 @@ import (
 	"github.com/vultisig/mobile-tss-lib/tss"
 	vcommon "github.com/vultisig/verifier/common"
 	"github.com/vultisig/verifier/plugin"
+	"github.com/vultisig/verifier/tx_indexer"
 	vtypes "github.com/vultisig/verifier/types"
 	"github.com/vultisig/verifier/vault"
 
@@ -30,18 +31,19 @@ import (
 )
 
 type Server struct {
-	cfg           ServerConfig
-	db            storage.DatabaseStorage
-	redis         *storage.RedisStorage
-	vaultStorage  vault.Storage
-	client        *asynq.Client
-	inspector     *asynq.Inspector
-	sdClient      *statsd.Client
-	scheduler     *scheduler.SchedulerService
-	policyService service.Policy
-	plugin        plugin.Plugin
-	logger        *logrus.Logger
-	mode          string
+	cfg              ServerConfig
+	db               storage.DatabaseStorage
+	redis            *storage.RedisStorage
+	vaultStorage     vault.Storage
+	client           *asynq.Client
+	inspector        *asynq.Inspector
+	sdClient         *statsd.Client
+	scheduler        *scheduler.SchedulerService
+	policyService    service.Policy
+	plugin           plugin.Plugin
+	txIndexerService *tx_indexer.Service
+	logger           *logrus.Logger
+	mode             string
 }
 
 // NewServer returns a new server.
@@ -55,6 +57,7 @@ func NewServer(
 	inspector *asynq.Inspector,
 	sdClient *statsd.Client,
 	p plugin.Plugin,
+	txIndexerService *tx_indexer.Service,
 ) *Server {
 	logger := logrus.WithField("service", "plugin").Logger
 	schedulerService, err := scheduler.NewSchedulerService(
@@ -73,17 +76,18 @@ func NewServer(
 	}
 
 	return &Server{
-		cfg:           cfg,
-		redis:         redis,
-		client:        client,
-		inspector:     inspector,
-		sdClient:      sdClient,
-		vaultStorage:  vaultStorage,
-		plugin:        p,
-		db:            db,
-		scheduler:     schedulerService,
-		logger:        logger,
-		policyService: policyService,
+		cfg:              cfg,
+		redis:            redis,
+		client:           client,
+		inspector:        inspector,
+		sdClient:         sdClient,
+		vaultStorage:     vaultStorage,
+		plugin:           p,
+		db:               db,
+		scheduler:        schedulerService,
+		logger:           logger,
+		policyService:    policyService,
+		txIndexerService: txIndexerService,
 	}
 }
 

cmd/dca/server/main.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/verifier/tx_indexer"
+	tx_indexer_storage "github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
 
 	"github.com/vultisig/plugin/api"
@@ -16,6 +19,8 @@ import (
 )
 
 func main() {
+	ctx := context.Background()
+
 	cfg, err := GetConfigure()
 	if err != nil {
 		panic(err)
@@ -50,6 +55,17 @@ func main() {
 		panic(err)
 	}
 
+	txIndexerStore, err := tx_indexer_storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("tx_indexer_storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	txIndexerService := tx_indexer.NewService(
+		logger,
+		txIndexerStore,
+		tx_indexer.Chains(),
+	)
+
 	db, err := postgres.NewPostgresBackend(cfg.Database.DSN, nil)
 	if err != nil {
 		logger.Fatalf("Failed to connect to database: %v", err)
@@ -67,7 +83,9 @@ func main() {
 		client,
 		inspector,
 		sdClient,
-		p)
+		p,
+		txIndexerService,
+	)
 	if err := server.StartServer(); err != nil {
 		panic(err)
 	}

cmd/dca/worker/config.go

@@ -11,6 +11,9 @@ import (
 )
 
 type DCAWorkerConfig struct {
+	Database struct {
+		DSN string `mapstructure:"dsn" json:"dsn,omitempty"`
+	} `mapstructure:"database" json:"database,omitempty"`
 	Redis              storage.RedisConfig             `mapstructure:"redis" json:"redis,omitempty"`
 	BlockStorage       vault_config.BlockStorageConfig `mapstructure:"block_storage" json:"block_storage,omitempty"`
 	VaultServiceConfig vault_config.Config             `mapstructure:"vault_service" json:"vault_service,omitempty"`

cmd/dca/worker/main.go

@@ -1,17 +1,22 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/verifier/tx_indexer"
+	"github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
 
 	"github.com/vultisig/plugin/internal/tasks"
 )
 
 func main() {
+	ctx := context.Background()
+
 	cfg, err := GetConfigure()
 	if err != nil {
 		panic(err)
@@ -45,12 +50,24 @@ func main() {
 			},
 		},
 	)
+
+	txIndexerStore, err := storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	txIndexerService := tx_indexer.NewService(
+		logger,
+		txIndexerStore,
+		tx_indexer.Chains(),
+	)
+
 	vaultService, err := vault.NewManagementService(
 		cfg.VaultServiceConfig,
 		client,
 		sdClient,
 		vaultStorage,
-		nil, // not used in plugin
+		txIndexerService,
 	)
 	if err != nil {
 		panic(fmt.Errorf("failed to create vault service: %w", err))

cmd/payroll/server/main.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/verifier/tx_indexer"
+	tx_indexer_storage "github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
 
 	"github.com/vultisig/plugin/api"
@@ -16,6 +19,8 @@ import (
 )
 
 func main() {
+	ctx := context.Background()
+
 	cfg, err := GetConfigure()
 	if err != nil {
 		panic(err)
@@ -51,6 +56,17 @@ func main() {
 		panic(err)
 	}
 
+	txIndexerStore, err := tx_indexer_storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("tx_indexer_storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	txIndexerService := tx_indexer.NewService(
+		logger,
+		txIndexerStore,
+		tx_indexer.Chains(),
+	)
+
 	db, err := postgres.NewPostgresBackend(cfg.Database.DSN, nil)
 	if err != nil {
 		logger.Fatalf("Failed to connect to database: %v", err)
@@ -68,7 +84,9 @@ func main() {
 		client,
 		inspector,
 		sdClient,
-		p)
+		p,
+		txIndexerService,
+	)
 	if err := server.StartServer(); err != nil {
 		panic(err)
 	}

cmd/payroll/worker/main.go

@@ -1,11 +1,14 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/verifier/tx_indexer"
+	"github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
 
 	"github.com/vultisig/plugin/internal/scheduler"
@@ -15,6 +18,8 @@ import (
 )
 
 func main() {
+	ctx := context.Background()
+
 	cfg, err := GetConfigure()
 	if err != nil {
 		panic(err)
@@ -49,12 +54,24 @@ func main() {
 			},
 		},
 	)
+
+	txIndexerStore, err := storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	txIndexerService := tx_indexer.NewService(
+		logger,
+		txIndexerStore,
+		tx_indexer.Chains(),
+	)
+
 	vaultService, err := vault.NewManagementService(
 		cfg.VaultServiceConfig,
 		client,
 		sdClient,
 		vaultStorage,
-		nil, // not used in plugin
+		txIndexerService,
 	)
 	if err != nil {
 		panic(fmt.Errorf("failed to create vault service: %w", err))