🎉 Basic treasury structure intro (no functionality, just basic hello world structure)

cmd/treasury/server/main.go

@@ -0,0 +1,119 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net"
+
+	"github.com/DataDog/datadog-go/statsd"
+	"github.com/hibiken/asynq"
+	"github.com/sirupsen/logrus"
+	"github.com/vultisig/plugin/internal/scheduler"
+	"github.com/vultisig/plugin/internal/verifierapi"
+	"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"
+	feeconfig "github.com/vultisig/plugin/cmd/fees/config"
+	"github.com/vultisig/plugin/plugin/treasury"
+	"github.com/vultisig/plugin/storage"
+	"github.com/vultisig/plugin/storage/postgres"
+)
+
+func main() {
+	ctx := context.Background()
+
+	cfg, err := feeconfig.GetConfigure()
+	if err != nil {
+		panic(err)
+	}
+
+	logger := logrus.New()
+	sdClient, err := statsd.New(net.JoinHostPort(cfg.Datadog.Host, cfg.Datadog.Port))
+	if err != nil {
+		panic(err)
+	}
+	redisStorage, err := storage.NewRedisStorage(cfg.Redis)
+	if err != nil {
+		panic(err)
+	}
+	redisOptions := asynq.RedisClientOpt{
+		Addr:     net.JoinHostPort(cfg.Redis.Host, cfg.Redis.Port),
+		Username: cfg.Redis.User,
+		Password: cfg.Redis.Password,
+		DB:       cfg.Redis.DB,
+	}
+
+	client := asynq.NewClient(redisOptions)
+	defer func() {
+		if err := client.Close(); err != nil {
+			fmt.Println("fail to close asynq client,", err)
+		}
+	}()
+
+	inspector := asynq.NewInspector(redisOptions)
+
+	vaultStorage, err := vault.NewBlockStorageImp(cfg.BlockStorage)
+	if err != nil {
+		panic(err)
+	}
+
+	txIndexerStore, err := tx_indexer_storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("tx_indexer_storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	_ = 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)
+	}
+
+	treasuryPluginConfig, err := treasury.NewTreasuryConfig(
+		treasury.WithChainIdRaw(1),
+		treasury.WithEthProviderRaw("https://eth.public-rpc.com"),
+		treasury.WithEncryptionSecret(cfg.Server.EncryptionSecret))
+
+	if err != nil {
+		logger.Fatalf("failed to create treasury config,err: %s", err)
+	}
+
+	verifierApi := verifierapi.NewVerifierApi(
+		cfg.Verifier.URL,
+		cfg.Verifier.Token,
+		logger.WithField("pkg", "verifierapi.VerifierApi").Logger,
+	)
+
+	treasuryPlugin, err := treasury.NewTreasuryPlugin(
+		treasuryPluginConfig,
+		vaultStorage,
+		nil,
+		db,
+		logger,
+		verifierApi,
+	)
+	if err != nil {
+		logger.Fatalf("failed to create fee plugin,err: %s", err)
+	}
+
+	server := api.NewServer(
+		cfg.Server,
+		db,
+		redisStorage,
+		vaultStorage,
+		client,
+		inspector,
+		sdClient,
+		treasuryPlugin,
+		scheduler.NewNilService(),
+	)
+	if err := server.StartServer(); err != nil {
+		panic(err)
+	}
+}

cmd/treasury/worker/main.go

@@ -0,0 +1,195 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/DataDog/datadog-go/statsd"
+	"github.com/hibiken/asynq"
+	"github.com/robfig/cron/v3"
+	"github.com/sirupsen/logrus"
+	"github.com/vultisig/verifier/plugin/keysign"
+	"github.com/vultisig/verifier/plugin/tasks"
+	"github.com/vultisig/verifier/plugin/tx_indexer"
+	"github.com/vultisig/verifier/plugin/tx_indexer/pkg/storage"
+	"github.com/vultisig/verifier/vault"
+	"github.com/vultisig/vultiserver/relay"
+
+	feeconfig "github.com/vultisig/plugin/cmd/fees/config"
+	"github.com/vultisig/plugin/internal/verifierapi"
+	"github.com/vultisig/plugin/plugin/treasury"
+	"github.com/vultisig/plugin/storage/postgres"
+)
+
+func main() {
+	ctx := context.Background()
+
+	cfg, err := feeconfig.GetConfigure()
+	if err != nil {
+		panic(err)
+	}
+
+	sdClient, err := statsd.New(cfg.Datadog.Host + ":" + cfg.Datadog.Port)
+	if err != nil {
+		panic(err)
+	}
+
+	vaultStorage, err := vault.NewBlockStorageImp(cfg.BlockStorage)
+	if err != nil {
+		panic(fmt.Sprintf("failed to initialize vault storage: %v", err))
+	}
+
+	redisOptions := asynq.RedisClientOpt{
+		Addr:     cfg.Redis.Host + ":" + cfg.Redis.Port,
+		Username: cfg.Redis.User,
+		Password: cfg.Redis.Password,
+		DB:       cfg.Redis.DB,
+	}
+	logger := logrus.StandardLogger()
+	logger.SetFormatter(&logrus.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: time.RFC3339,
+	})
+
+	asynqClient := asynq.NewClient(redisOptions)
+
+	srv := asynq.NewServer(
+		redisOptions,
+		asynq.Config{
+			Logger:      logger,
+			Concurrency: 10,
+			Queues: map[string]int{
+				tasks.QUEUE_NAME: 10,
+			},
+		},
+	)
+
+	postgressDB, err := postgres.NewPostgresBackend(cfg.Database.DSN, nil)
+	if err != nil {
+		panic(fmt.Errorf("failed to create postgres backend: %w", err))
+	}
+
+	txIndexerStore, err := storage.NewPostgresTxIndexStore(ctx, cfg.Database.DSN)
+	if err != nil {
+		panic(fmt.Errorf("storage.NewPostgresTxIndexStore: %w", err))
+	}
+
+	chains, err := tx_indexer.Chains()
+	if err != nil {
+		panic(fmt.Errorf("failed to initialize tx indexer chains: %w", err))
+	}
+
+	txIndexerService := tx_indexer.NewService(
+		logger,
+		txIndexerStore,
+		chains,
+	)
+
+	vaultService, err := vault.NewManagementService(
+		cfg.VaultServiceConfig,
+		asynqClient,
+		sdClient,
+		vaultStorage,
+		txIndexerService,
+	)
+
+	if err != nil {
+		panic(fmt.Errorf("failed to create vault service: %w", err))
+	}
+
+	treasuryPluginConfig, err := treasury.NewTreasuryConfig(
+		treasury.WithChainIdRaw(1),
+		treasury.WithEthProviderRaw("https://eth.public-rpc.com"),
+		treasury.WithEncryptionSecret(cfg.VaultServiceConfig.EncryptionSecret),
+	)
+
+	if err != nil {
+		logger.Fatalf("failed to create fees config,err: %s", err)
+	}
+
+	signer := keysign.NewSigner(
+		logger.WithField("pkg", "keysign.Signer").Logger,
+		relay.NewRelayClient(cfg.VaultServiceConfig.Relay.Server),
+		[]keysign.Emitter{
+			keysign.NewVerifierEmitter(cfg.Verifier.URL, cfg.Verifier.Token),
+			keysign.NewPluginEmitter(asynqClient, tasks.TypeKeySignDKLS, tasks.QUEUE_NAME),
+		},
+		[]string{
+			cfg.Verifier.PartyPrefix,
+			cfg.VaultServiceConfig.LocalPartyPrefix,
+		},
+	)
+
+	verifierApi := verifierapi.NewVerifierApi(
+		cfg.Verifier.URL,
+		cfg.Verifier.Token,
+		logger.WithField("pkg", "verifierapi.VerifierApi").Logger,
+	)
+
+	treasuryPlugin, err := treasury.NewTreasuryPlugin(
+		treasuryPluginConfig,
+		vaultStorage,
+		signer,
+		postgressDB,
+		logger,
+		verifierApi,
+	)
+	if err != nil {
+		logger.Fatalf("failed to create fee plugin,err: %s", err)
+	}
+
+	mux := asynq.NewServeMux()
+	//	mux.HandleFunc(tasks.TypePluginTransaction, vaultService.HandlePluginTransaction)
+
+	//Core functions, every plugin should have these functions
+	mux.HandleFunc(tasks.TypeKeySignDKLS, vaultService.HandleKeySignDKLS)
+	mux.HandleFunc(tasks.TypeReshareDKLS, vaultService.HandleReshareDKLS)
+
+	//Plugin specific functions.
+	mux.HandleFunc(treasury.TypeTreasuryLoad, treasuryPlugin.LoadTreasuryPayments)
+	mux.HandleFunc(treasury.TypeTreasuryTransact, treasuryPlugin.TransactTreasuryPayments)
+	mux.HandleFunc(treasury.TypeTreasuryPostTx, treasuryPlugin.PostTreasuryPayments)
+
+	loadTreasuryPaymentsCron := cron.New()
+	loadTreasuryPaymentsCron.AddFunc(treasuryPluginConfig.Jobs.Load.Cronexpr, func() {
+		payload := make([]byte, 0)
+		asynqClient.Enqueue(
+			asynq.NewTask(treasury.TypeTreasuryLoad, payload),
+			asynq.MaxRetry(0),
+			asynq.Timeout(2*time.Minute),
+			asynq.Retention(5*time.Minute),
+			asynq.Queue(tasks.QUEUE_NAME))
+	})
+	loadTreasuryPaymentsCron.Start()
+
+	transactTreasuryPaymentsCron := cron.New()
+	transactTreasuryPaymentsCron.AddFunc(treasuryPluginConfig.Jobs.Transact.Cronexpr, func() {
+		payload := make([]byte, 0)
+		asynqClient.Enqueue(
+			asynq.NewTask(treasury.TypeTreasuryTransact, payload),
+			asynq.MaxRetry(0),
+			asynq.Timeout(2*time.Minute),
+			asynq.Retention(5*time.Minute),
+			asynq.Queue(tasks.QUEUE_NAME))
+	})
+	transactTreasuryPaymentsCron.Start()
+
+	postTreasuryPaymentsCron := cron.New()
+	postTreasuryPaymentsCron.AddFunc(treasuryPluginConfig.Jobs.Post.Cronexpr, func() {
+		payload := make([]byte, 0)
+		asynqClient.Enqueue(
+			asynq.NewTask(treasury.TypeTreasuryPostTx, payload),
+			asynq.MaxRetry(0),
+			asynq.Timeout(2*time.Minute),
+			asynq.Retention(5*time.Minute),
+			asynq.Queue(tasks.QUEUE_NAME))
+	})
+	postTreasuryPaymentsCron.Start()
+
+	logger.Info("Starting asynq listener")
+
+	if err := srv.Run(mux); err != nil {
+		panic(fmt.Errorf("could not run server: %w", err))
+	}
+}