payroll scheduler refactoring

.run/[git] payroll scheduler.run.xml

@@ -0,0 +1,15 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="[git] payroll scheduler" type="GoApplicationRunConfiguration" factoryName="Go Application">
+    <module name="plugin" />
+    <working_directory value="$PROJECT_DIR$" />
+    <envs>
+      <env name="DATABASE_DSN" value="postgres://myuser:mypassword@localhost:5431/vultisig-payroll?sslmode=disable" />
+      <env name="VS_CONFIG_NAME" value="payroll.scheduler.example" />
+    </envs>
+    <kind value="PACKAGE" />
+    <package value="github.com/vultisig/plugin/cmd/payroll/scheduler" />
+    <directory value="$PROJECT_DIR$" />
+    <filePath value="$PROJECT_DIR$/cmd/payroll/worker/main.go" />
+    <method v="2" />
+  </configuration>
+</component>

.run/[git] payroll worker.run.xml

@@ -3,7 +3,6 @@
     <module name="plugin" />
     <working_directory value="$PROJECT_DIR$" />
     <envs>
-      <env name="BASE_FILE_PATH" value="etc/vultisig" />
       <env name="DATABASE_DSN" value="postgres://myuser:mypassword@localhost:5431/vultisig-payroll?sslmode=disable" />
       <env name="VS_CONFIG_NAME" value="payroll.worker.example" />
     </envs>
@@ -13,4 +12,4 @@
     <filePath value="$PROJECT_DIR$/cmd/payroll/worker/main.go" />
     <method v="2" />
   </configuration>
-</component>
+</component>

.run/[git] plugin stack.run.xml

@@ -1,5 +1,6 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="[git] plugin stack" type="CompoundRunConfigurationType">
+    <toRun name="[git] payroll scheduler" type="GoApplicationRunConfiguration" />
     <toRun name="[git] payroll server" type="GoApplicationRunConfiguration" />
     <toRun name="[git] payroll worker" type="GoApplicationRunConfiguration" />
     <toRun name="[git] tx_indexer worker" type="GoApplicationRunConfiguration" />

Dockerfile.Payroll.scheduler

@@ -0,0 +1,43 @@
+# Use the official Go image as the base image
+FROM golang:1.24 as builder
+
+# Set the working directory
+WORKDIR /app
+
+# Copy go.mod and go.sum to install dependencies
+COPY go.mod go.sum ./
+RUN go mod download
+
+# Copy the source code
+COPY . .
+
+RUN wget https://github.com/vultisig/go-wrappers/archive/refs/heads/master.tar.gz
+RUN tar -xzf master.tar.gz && \
+    cd go-wrappers-master && \
+    mkdir -p /usr/local/lib/dkls && \
+    cp --recursive includes /usr/local/lib/dkls
+
+ENV LD_LIBRARY_PATH=/usr/local/lib/dkls/includes/linux/:${LD_LIBRARY_PATH:-}
+
+# Build the application
+RUN go build -o scheduler ./cmd/payroll/scheduler
+
+# Use a minimal base image for the final stage
+FROM ubuntu:24.04
+
+RUN apt-get update && \
+    apt-get install -y wget && \
+    rm -rf /var/lib/apt/lists/*
+
+# Set the working directory
+WORKDIR /app
+
+# Copy the built binary from the builder stage
+COPY --from=builder /app/scheduler .
+COPY --from=builder /usr/local/lib/dkls /usr/local/lib/dkls
+COPY payroll.scheduler.example.json config.json
+
+ENV LD_LIBRARY_PATH=/usr/local/lib/dkls/includes/linux/:${LD_LIBRARY_PATH:-}
+
+# Run the application
+CMD ["./scheduler"]

api/server.go

@@ -35,7 +35,6 @@ type Server struct {
 	client        *asynq.Client
 	inspector     *asynq.Inspector
 	sdClient      *statsd.Client
-	scheduler     *scheduler.SchedulerService
 	policyService service.Policy
 	plugin        plugin.Plugin
 	logger        *logrus.Logger
@@ -48,24 +47,15 @@ func NewServer(
 	db *postgres.PostgresBackend,
 	redis *storage.RedisStorage,
 	vaultStorage vault.Storage,
-	redisOpts asynq.RedisClientOpt,
 	client *asynq.Client,
 	inspector *asynq.Inspector,
 	sdClient *statsd.Client,
 	p plugin.Plugin,
+	scheduler scheduler.Service,
 ) *Server {
 	logger := logrus.WithField("service", "plugin").Logger
-	schedulerService, err := scheduler.NewSchedulerService(
-		db,
-		client,
-		redisOpts,
-	)
-	if err != nil {
-		logger.Fatalf("Failed to initialize scheduler service: %v", err)
-		return nil
-	}
 
-	policyService, err := service.NewPolicyService(db, schedulerService, logger.WithField("service", "policy").Logger)
+	policyService, err := service.NewPolicyService(db, scheduler, logger.WithField("service", "policy").Logger)
 	if err != nil {
 		logger.Fatalf("Failed to initialize policy service: %v", err)
 	}
@@ -79,7 +69,6 @@ func NewServer(
 		vaultStorage:  vaultStorage,
 		plugin:        p,
 		db:            db,
-		scheduler:     schedulerService,
 		logger:        logger,
 		policyService: policyService,
 	}

cmd/dca/server/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/plugin/internal/scheduler"
 	"github.com/vultisig/verifier/tx_indexer"
 	tx_indexer_storage "github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
@@ -80,11 +81,11 @@ func main() {
 		db,
 		redisStorage,
 		vaultStorage,
-		redisOptions,
 		client,
 		inspector,
 		sdClient,
 		p,
+		scheduler.NewNilService(),
 	)
 	if err := server.StartServer(); err != nil {
 		panic(err)

cmd/dca/worker/main.go

@@ -45,8 +45,7 @@ func main() {
 			Logger:      logger,
 			Concurrency: 10,
 			Queues: map[string]int{
-				tasks.QUEUE_NAME:         10,
-				"scheduled_plugin_queue": 10, // new queue
+				tasks.QUEUE_NAME: 10,
 			},
 		},
 	)

cmd/fees/server/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/plugin/internal/scheduler"
 	"github.com/vultisig/verifier/tx_indexer"
 	tx_indexer_storage "github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
@@ -100,11 +101,11 @@ func main() {
 		db,
 		redisStorage,
 		vaultStorage,
-		redisOptions,
 		client,
 		inspector,
 		sdClient,
 		feePlugin,
+		scheduler.NewNilService(),
 	)
 	if err := server.StartServer(); err != nil {
 		panic(err)

cmd/fees/worker/main.go

@@ -57,8 +57,7 @@ func main() {
 			Logger:      logger,
 			Concurrency: 10,
 			Queues: map[string]int{
-				tasks.QUEUE_NAME:         10,
-				"scheduled_plugin_queue": 10, // new queue
+				tasks.QUEUE_NAME: 10,
 			},
 		},
 	)

cmd/payroll/scheduler/config.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/spf13/viper"
+	"github.com/vultisig/plugin/storage"
+)
+
+type PayrollSchedulerConfig struct {
+	Redis    storage.RedisConfig `mapstructure:"redis" json:"redis,omitempty"`
+	Database struct {
+		DSN string `mapstructure:"dsn" json:"dsn,omitempty"`
+	} `mapstructure:"database" json:"database,omitempty"`
+}
+
+func GetConfigure() (*PayrollSchedulerConfig, error) {
+	configName := os.Getenv("VS_CONFIG_NAME")
+	if configName == "" {
+		configName = "config"
+	}
+	return ReadConfig(configName)
+}
+
+func ReadConfig(configName string) (*PayrollSchedulerConfig, error) {
+	viper.SetConfigName(configName)
+	viper.AddConfigPath(".")
+	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
+	viper.AutomaticEnv()
+
+	if err := viper.ReadInConfig(); err != nil {
+		return nil, fmt.Errorf("fail to reading config file, %w", err)
+	}
+	var cfg PayrollSchedulerConfig
+	err := viper.Unmarshal(&cfg)
+	if err != nil {
+		return nil, fmt.Errorf("unable to decode into struct, %w", err)
+	}
+	return &cfg, nil
+}

cmd/payroll/scheduler/main.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/hibiken/asynq"
+	"github.com/sirupsen/logrus"
+	"github.com/vultisig/plugin/internal/scheduler"
+	"github.com/vultisig/plugin/internal/tasks"
+	"github.com/vultisig/plugin/plugin/payroll"
+	"github.com/vultisig/plugin/storage/postgres"
+)
+
+func main() {
+	cfg, err := GetConfigure()
+	if err != nil {
+		panic(fmt.Errorf("failed to get config: %w", err))
+	}
+
+	client := asynq.NewClient(asynq.RedisClientOpt{
+		Addr:     net.JoinHostPort(cfg.Redis.Host, cfg.Redis.Port),
+		Username: cfg.Redis.User,
+		Password: cfg.Redis.Password,
+		DB:       cfg.Redis.DB,
+	})
+	defer func() {
+		_ = client.Close()
+	}()
+
+	db, err := postgres.NewPostgresBackend(cfg.Database.DSN, nil)
+	if err != nil {
+		panic(fmt.Errorf("failed to create db conn: %w", err))
+	}
+
+	worker := scheduler.NewWorker(
+		logrus.New(),
+		client,
+		tasks.TypePluginTransaction,
+		tasks.QUEUE_NAME,
+		scheduler.NewPostgresStorage(db.Pool()),
+		payroll.NewSchedulerInterval(),
+		db,
+	)
+
+	err = worker.Run()
+	if err != nil {
+		panic(fmt.Errorf("failed to run worker: %w", err))
+	}
+}

cmd/payroll/server/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
 	"github.com/sirupsen/logrus"
+	"github.com/vultisig/plugin/internal/scheduler"
 	"github.com/vultisig/verifier/tx_indexer"
 	tx_indexer_storage "github.com/vultisig/verifier/tx_indexer/pkg/storage"
 	"github.com/vultisig/verifier/vault"
@@ -83,16 +84,17 @@ func main() {
 	if err != nil {
 		logger.Fatalf("failed to create payroll plugin,err: %s", err)
 	}
+
 	server := api.NewServer(
 		cfg.Server,
 		db,
 		redisStorage,
 		vaultStorage,
-		redisOptions,
 		client,
 		inspector,
 		sdClient,
 		p,
+		payroll.NewSchedulerService(scheduler.NewPostgresStorage(db.Pool())),
 	)
 	if err := server.StartServer(); err != nil {
 		panic(err)

cmd/payroll/worker/main.go

@@ -14,14 +14,11 @@ import (
 	"github.com/vultisig/verifier/vault"
 	"github.com/vultisig/vultiserver/relay"
 
-	"github.com/vultisig/plugin/internal/scheduler"
 	"github.com/vultisig/plugin/internal/tasks"
 	"github.com/vultisig/plugin/plugin/payroll"
 	"github.com/vultisig/plugin/storage/postgres"
 )
 
-// Don't scale payroll.worker, it has scheduler which must be single instance running
-// Consider moving scheduler to separate worker
 func main() {
 	ctx := context.Background()
 
@@ -54,8 +51,7 @@ func main() {
 			Logger:      logger,
 			Concurrency: 10,
 			Queues: map[string]int{
-				tasks.QUEUE_NAME:         10,
-				"scheduled_plugin_queue": 10, // new queue
+				tasks.QUEUE_NAME: 10,
 			},
 		},
 	)
@@ -115,13 +111,6 @@ func main() {
 	if err != nil {
 		panic(fmt.Errorf("failed to create payroll plugin: %w", err))
 	}
-	schedulerSvc, err := scheduler.NewSchedulerService(postgressDB, client, redisOptions)
-	if err != nil {
-		panic(fmt.Errorf("failed to create scheduler service: %w", err))
-	}
-
-	schedulerSvc.Start()
-	defer schedulerSvc.Stop()
 
 	mux := asynq.NewServeMux()
 	mux.HandleFunc(tasks.TypePluginTransaction, p.HandleSchedulerTrigger)