diff --git a/.run/[git] payroll scheduler.run.xml b/.run/[git] payroll scheduler.run.xml new file mode 100644 index 0000000..4fb296e --- /dev/null +++ b/.run/[git] payroll scheduler.run.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.run/[git] payroll worker.run.xml b/.run/[git] payroll worker.run.xml index 6b9974f..37fb0bf 100644 --- a/.run/[git] payroll worker.run.xml +++ b/.run/[git] payroll worker.run.xml @@ -3,7 +3,6 @@ - @@ -13,4 +12,4 @@ - + \ No newline at end of file diff --git a/.run/[git] plugin stack.run.xml b/.run/[git] plugin stack.run.xml index af63e95..27e8141 100644 --- a/.run/[git] plugin stack.run.xml +++ b/.run/[git] plugin stack.run.xml @@ -1,5 +1,6 @@ + diff --git a/Dockerfile.Payroll.scheduler b/Dockerfile.Payroll.scheduler new file mode 100644 index 0000000..32adedb --- /dev/null +++ b/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"] diff --git a/api/server.go b/api/server.go index 28613ed..e8ecab4 100644 --- a/api/server.go +++ b/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, } diff --git a/cmd/dca/server/main.go b/cmd/dca/server/main.go index 114fc85..e427485 100644 --- a/cmd/dca/server/main.go +++ b/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) diff --git a/cmd/dca/worker/main.go b/cmd/dca/worker/main.go index 849043a..5fb7b2e 100644 --- a/cmd/dca/worker/main.go +++ b/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, }, }, ) diff --git a/cmd/fees/server/main.go b/cmd/fees/server/main.go index 6a2c153..e0d69b9 100644 --- a/cmd/fees/server/main.go +++ b/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) diff --git a/cmd/fees/worker/main.go b/cmd/fees/worker/main.go index 64dd50a..7da9b84 100644 --- a/cmd/fees/worker/main.go +++ b/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, }, }, ) diff --git a/cmd/payroll/scheduler/config.go b/cmd/payroll/scheduler/config.go new file mode 100644 index 0000000..73a5f4d --- /dev/null +++ b/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 +} diff --git a/cmd/payroll/scheduler/main.go b/cmd/payroll/scheduler/main.go new file mode 100644 index 0000000..66b9a83 --- /dev/null +++ b/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)) + } +} diff --git a/cmd/payroll/server/main.go b/cmd/payroll/server/main.go index 4aa20a4..8dc4a22 100644 --- a/cmd/payroll/server/main.go +++ b/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) diff --git a/cmd/payroll/worker/main.go b/cmd/payroll/worker/main.go index 9e9aa36..0cbb6c7 100644 --- a/cmd/payroll/worker/main.go +++ b/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) diff --git a/docker-compose.yaml b/docker-compose.yaml index e6d940f..b4aa408 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -83,7 +83,7 @@ services: BLOCK_STORAGE_HOST: http://minio-plugin:9000 DATABASE_DSN: postgres://myuser:mypassword@plugin-db:5432/vultisig-payroll?sslmode=disable ports: - - "8081:8080" + - "8081:8080" depends_on: plugin-db: condition: service_healthy @@ -93,6 +93,7 @@ services: condition: service_started networks: - shared_network + payroll-worker: build: context: . @@ -114,6 +115,27 @@ services: networks: - shared_network + payroll-scheduler: + build: + context: . + dockerfile: Dockerfile.Payroll.scheduler + platforms: + - linux/amd64 + environment: + REDIS_HOST: redis-payroll + REDIS_PORT: 6379 + BLOCK_STORAGE_HOST: http://minio-plugin:9000 + DATABASE_DSN: postgres://myuser:mypassword@plugin-db:5432/vultisig-payroll?sslmode=disable + depends_on: + plugin-db: + condition: service_healthy + redis-payroll: + condition: service_healthy + minio-plugin: + condition: service_started + networks: + - shared_network + fee-server: build: context: . @@ -126,7 +148,7 @@ services: BLOCK_STORAGE_HOST: http://minio-plugin:9000 DATABASE_DSN: postgres://myuser:mypassword@plugin-db:5432/vultisig-fee?sslmode=disable ports: - - "8082:8080" + - "8082:8080" depends_on: plugin-db: condition: service_healthy @@ -136,6 +158,7 @@ services: condition: service_started networks: - shared_network + fee-worker: build: context: . @@ -159,7 +182,7 @@ services: volumes: db_data: - minio_data: + minio_data: redis_payroll: redis_fees: networks: diff --git a/go.mod b/go.mod index 2833375..a7a37fc 100644 --- a/go.mod +++ b/go.mod @@ -20,8 +20,8 @@ require ( github.com/stretchr/testify v1.10.0 github.com/vultisig/commondata v0.0.0-20250710214228-61d9ed8f7778 github.com/vultisig/mobile-tss-lib v0.0.0-20250316003201-2e7e570a4a74 - github.com/vultisig/recipes v0.0.0-20250725112632-c546f5f66830 - github.com/vultisig/verifier v0.0.0-20250725121724-4a7f10cc3689 + github.com/vultisig/recipes v0.0.0-20250729120802-9b1d07f8262a + github.com/vultisig/verifier v0.0.0-20250723150319-a51c59a884bf github.com/vultisig/vultiserver v0.0.0-20250715212748-4b23f9849e4b golang.org/x/sync v0.14.0 google.golang.org/protobuf v1.36.6 diff --git a/go.sum b/go.sum index eaf2534..3c0ef31 100644 --- a/go.sum +++ b/go.sum @@ -759,14 +759,18 @@ github.com/vultisig/mobile-tss-lib v0.0.0-20250316003201-2e7e570a4a74 h1:goqwk4n github.com/vultisig/mobile-tss-lib v0.0.0-20250316003201-2e7e570a4a74/go.mod h1:nOykk4nOy1L3yXtLSlYvVsgizBnCQ3tR2N5uwGPdvaM= github.com/vultisig/recipes v0.0.0-20250723142134-153c7f486070 h1:AcE6x2dxkl/4P/TxaMJ70pa2BcZ3+FJWnVmlPVAq4Dk= github.com/vultisig/recipes v0.0.0-20250723142134-153c7f486070/go.mod h1:30NOW5y2BnMCmEYFeVls3NhxkbMvuv7BNyRQfoFhGIM= -github.com/vultisig/recipes v0.0.0-20250724185247-fde1a4c02314 h1:817gEvSIW4cNqmQxZIpwttnuzaw3RYIGbHFtIsm9Eyk= -github.com/vultisig/recipes v0.0.0-20250724185247-fde1a4c02314/go.mod h1:30NOW5y2BnMCmEYFeVls3NhxkbMvuv7BNyRQfoFhGIM= -github.com/vultisig/recipes v0.0.0-20250725112632-c546f5f66830 h1:r1Gw32G3yE1/w9Sxtq9fR53xixGu/r41JRl1HAqGGnY= -github.com/vultisig/recipes v0.0.0-20250725112632-c546f5f66830/go.mod h1:30NOW5y2BnMCmEYFeVls3NhxkbMvuv7BNyRQfoFhGIM= +github.com/vultisig/recipes v0.0.0-20250724111913-675e7fdcbc24 h1:NKkcuAtSSQfaYTQ0Fo96mnTsTDAoTh+XrbrUHLEliNY= +github.com/vultisig/recipes v0.0.0-20250724111913-675e7fdcbc24/go.mod h1:Ot3lrUnnSw67Hep+MelclVPgNLDxJP01Ezixw/1RYRE= +github.com/vultisig/recipes v0.0.0-20250724145933-37c12287503c h1:PFaQfDNMSBymgwu+6Jr9+wYF/tFpv3jk97i8VE2+bSY= +github.com/vultisig/recipes v0.0.0-20250724145933-37c12287503c/go.mod h1:Ot3lrUnnSw67Hep+MelclVPgNLDxJP01Ezixw/1RYRE= +github.com/vultisig/recipes v0.0.0-20250724150617-ab960a120ce8 h1:aMVowo/fdQGizJf7lFDzh07t7GmKvLyddVOWeFm7dkw= +github.com/vultisig/recipes v0.0.0-20250724150617-ab960a120ce8/go.mod h1:Ot3lrUnnSw67Hep+MelclVPgNLDxJP01Ezixw/1RYRE= +github.com/vultisig/recipes v0.0.0-20250724151024-afb75caf8884 h1:ObxhyUAEQYfmuqH2mnvw/fidEmaDPIEbxdp3oIpM9WY= +github.com/vultisig/recipes v0.0.0-20250724151024-afb75caf8884/go.mod h1:Ot3lrUnnSw67Hep+MelclVPgNLDxJP01Ezixw/1RYRE= +github.com/vultisig/recipes v0.0.0-20250729120802-9b1d07f8262a h1:KoAwytLj092KNgWHh0a5tcupsSm5HtentT0yUbDGWFQ= +github.com/vultisig/recipes v0.0.0-20250729120802-9b1d07f8262a/go.mod h1:Ot3lrUnnSw67Hep+MelclVPgNLDxJP01Ezixw/1RYRE= github.com/vultisig/verifier v0.0.0-20250723150319-a51c59a884bf h1:3VhxFQR0oJEieaajNDvKlOGBDyTBbSMJSxow3zwUlmw= github.com/vultisig/verifier v0.0.0-20250723150319-a51c59a884bf/go.mod h1:9f3yGSZWKZeXn30mU+/brufkbkLoGlFmNLKOpRbQgeI= -github.com/vultisig/verifier v0.0.0-20250725121724-4a7f10cc3689 h1:o/eXinXl4jgR2Km+zYP/TcmId+Za9rWzb9XvS2NX9Rk= -github.com/vultisig/verifier v0.0.0-20250725121724-4a7f10cc3689/go.mod h1:Q+Dp/lISXGVTNALJCpFTi+YhxsCd2UDQaA75IQkevOY= github.com/vultisig/vultiserver v0.0.0-20250715212748-4b23f9849e4b h1:Ed2DOWo8fA0KG6e36rzUmGpxcOQjmWTbxWyvUbI5by8= github.com/vultisig/vultiserver v0.0.0-20250715212748-4b23f9849e4b/go.mod h1:HwP2IgW6Mcu/gX8paFuKvfibrGE9UmPgkOFTub6dskM= github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q= diff --git a/internal/scheduler/interval.go b/internal/scheduler/interval.go new file mode 100644 index 0000000..45c18d6 --- /dev/null +++ b/internal/scheduler/interval.go @@ -0,0 +1,11 @@ +package scheduler + +import ( + "time" + + "github.com/vultisig/verifier/types" +) + +type Interval interface { + FromNowWhenNext(policy types.PluginPolicy) (time.Time, error) +} diff --git a/internal/scheduler/policy.go b/internal/scheduler/policy.go new file mode 100644 index 0000000..9e1ebfa --- /dev/null +++ b/internal/scheduler/policy.go @@ -0,0 +1,12 @@ +package scheduler + +import ( + "context" + + "github.com/google/uuid" + "github.com/vultisig/verifier/types" +) + +type PolicyFetcher interface { + GetPluginPolicy(ctx context.Context, id uuid.UUID) (*types.PluginPolicy, error) +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go deleted file mode 100644 index cf990bf..0000000 --- a/internal/scheduler/scheduler.go +++ /dev/null @@ -1,467 +0,0 @@ -package scheduler - -import ( - "context" - "encoding/base64" - "encoding/json" - "fmt" - "time" - - "github.com/hibiken/asynq" - "github.com/jackc/pgx/v5" - "github.com/robfig/cron/v3" - "github.com/sirupsen/logrus" - "github.com/vultisig/plugin/internal/tasks" - rtypes "github.com/vultisig/recipes/types" - vtypes "github.com/vultisig/verifier/types" - "google.golang.org/protobuf/proto" - - "github.com/vultisig/plugin/internal/types" - "github.com/vultisig/plugin/storage" -) - -const ( - secondsInDay = 24 * 60 * 60 - secondsInWeek = 7 * 24 * 60 * 60 -) - -type SchedulerService struct { - db storage.DatabaseStorage - logger *logrus.Logger - client *asynq.Client - inspector *asynq.Inspector - done chan struct{} -} - -func NewSchedulerService(db storage.DatabaseStorage, - client *asynq.Client, - redisOpts asynq.RedisClientOpt) (*SchedulerService, error) { - if db == nil { - return nil, fmt.Errorf("failed to initialize scheduler: db is nil") - } - - // create inspector using the same Redis configuration as the client - inspector := asynq.NewInspector(redisOpts) - return &SchedulerService{ - db: db, - logger: logrus.WithField("component", "scheduler").Logger, - client: client, - inspector: inspector, - done: make(chan struct{}), - }, nil -} - -func (s *SchedulerService) Start() { - go s.run() -} - -func (s *SchedulerService) Stop() { - close(s.done) -} - -func (s *SchedulerService) run() { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - if err := s.checkAndEnqueueTasks(); err != nil { - s.logger.Errorf("Failed to check and enqueue tasks: %v", err) - } - case <-s.done: - return - } - } -} - -func (s *SchedulerService) checkAndEnqueueTasks() error { - ctx := context.Background() - triggers, err := s.db.GetPendingTimeTriggers(ctx) - if err != nil { - return fmt.Errorf("failed to get pending triggers: %w", err) - } - s.logger.Infof("Found %d active triggers: %+v: ", len(triggers), triggers) - - for _, trigger := range triggers { - s.logger.WithFields(logrus.Fields{ - "policy_id": trigger.PolicyID, - "last_exec": trigger.LastExecution, - }).Info("Processing trigger") - - // Parse cron expression - schedule, err := createSchedule(trigger.CronExpression, trigger.Frequency, trigger.StartTime, trigger.Interval) - if err != nil { - s.logger.Errorf("Failed to create schedule: %v", err) - err := s.db.DeleteTimeTrigger(ctx, trigger.PolicyID) - if err != nil { - return fmt.Errorf("failed to delete time trigger: %w", err) - } - continue - } - - // Check if it's time to execute - var nextTime time.Time - if trigger.LastExecution != nil { - nextTime = schedule.Next(*trigger.LastExecution) - } else { - nextTime = trigger.StartTime - } - - nextTime = nextTime.UTC() - endTime := trigger.EndTime - - if endTime != nil && time.Now().UTC().After(*endTime) { - // TODO: Check if this end time was ever set anywhere. - s.logger.WithFields(logrus.Fields{ - "policy_id": trigger.PolicyID, - "end_time": *endTime, - }).Info("Trigger end time reached") - err := s.db.DeleteTimeTrigger(ctx, trigger.PolicyID) - if err != nil { - return fmt.Errorf("failed to delete time trigger: %w", err) - } - continue - } - - triggerStatus, err := s.db.GetTriggerStatus(ctx, trigger.PolicyID) - if err != nil { - s.logger.Errorf("Failed to get trigger status: %v", err) - continue - } - - if time.Now().UTC().Before(nextTime) || triggerStatus == types.StatusTimeTriggerRunning { - s.logger.WithFields(logrus.Fields{ - "policy_id": trigger.PolicyID, - "next_time": nextTime, - "state": triggerStatus, - }).Info("Trigger have not reached next time or it's in running status") - continue - } - - if err := s.db.UpdateTriggerStatus(ctx, trigger.PolicyID, types.StatusTimeTriggerRunning); err != nil { - s.logger.Errorf("Failed to update trigger status: %v", err) - continue - } - - buf, err := json.Marshal(trigger) - if err != nil { - s.logger.Errorf("Failed to marshal trigger event: %v", err) - continue - } - ti, err := s.client.Enqueue( - asynq.NewTask(tasks.TypePluginTransaction, buf), - asynq.MaxRetry(0), - asynq.Timeout(5*time.Minute), - asynq.Retention(10*time.Minute), - asynq.Queue(tasks.QUEUE_NAME), - ) - if err != nil { - s.logger.Errorf("Failed to enqueue trigger task: %v", err) - continue - - } - - s.logger.WithFields(logrus.Fields{ - "task_id": ti.ID, - "policy_id": trigger.PolicyID, - }).Info("Enqueued trigger task") - } - - return nil -} - -func (s *SchedulerService) CreateTimeTrigger(ctx context.Context, policy vtypes.PluginPolicy, dbTx pgx.Tx) error { - if s.db == nil { - return fmt.Errorf("failed to create time trigger: database backend is nil") - } - - trigger, err := s.GetTriggerFromPolicy(policy) - if err != nil { - return fmt.Errorf("failed to get trigger from policy: %w", err) - } - - return s.db.CreateTimeTriggerTx(ctx, dbTx, *trigger) -} - -func (s *SchedulerService) GetScheduleFromPolicy(policy vtypes.PluginPolicy) (*rtypes.Schedule, error) { - recipe, err := base64.StdEncoding.DecodeString(policy.Recipe) - if err != nil { - return nil, fmt.Errorf("base64.StdEncoding.DecodeString: %w", err) - } - - var p rtypes.Policy - if err := proto.Unmarshal(recipe, &p); err != nil { - return nil, fmt.Errorf("failed to parse policy schedule: %w", err) - } - return p.Schedule, nil -} - -func (s *SchedulerService) GetTriggerFromPolicy(policy vtypes.PluginPolicy) (*types.TimeTrigger, error) { - recipe, err := base64.StdEncoding.DecodeString(policy.Recipe) - if err != nil { - return nil, fmt.Errorf("failed to decode policy recipe: %w", err) - } - - var p rtypes.Policy - if err := proto.Unmarshal(recipe, &p); err != nil { - return nil, fmt.Errorf("failed to parse policy schedule: %w", err) - } - if p.Schedule == nil { - return nil, fmt.Errorf("failed to get schedule from policy: policy schedule is nil") - } - interval := 1 - if p.Schedule.Interval > 0 { - interval = int(p.Schedule.Interval) - } - startTime := time.Now().UTC() - if p.Schedule.GetStartTime() != nil { - startTime = p.Schedule.GetStartTime().AsTime() - } - var endTime *time.Time - if p.Schedule.GetEndTime() != nil { - tmpTime := p.Schedule.GetEndTime().AsTime() - endTime = &tmpTime - } - cronExpr := frequencyToCron(p.Schedule.GetFrequency(), startTime, interval) - trigger := types.TimeTrigger{ - PolicyID: policy.ID, - CronExpression: cronExpr, - StartTime: startTime, - EndTime: endTime, - Frequency: p.Schedule.Frequency, - Interval: interval, - Status: types.StatusTimeTriggerPending, - } - - return &trigger, nil -} - -func createSchedule(cronExpr string, frequency rtypes.ScheduleFrequency, startTime time.Time, interval int) (cron.Schedule, error) { - // Use our custom schedule implementation for intervals > 1 and when frequency is daily, weekly, monthly - if interval > 1 && - (frequency == rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY || - frequency == rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY || - frequency == rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_MONTHLY) { - return NewIntervalSchedule(frequency, startTime, interval) - } - - // For standard cron - schedule, err := cron.ParseStandard(cronExpr) - if err != nil { - return nil, fmt.Errorf("failed to parse cron expression: %w", err) - } - - return schedule, nil -} - -func frequencyToCron(frequency rtypes.ScheduleFrequency, startTime time.Time, interval int) string { - switch frequency { - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_UNSPECIFIED: - return "" - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY: - return fmt.Sprintf("%d %d * * *", startTime.Minute(), startTime.Hour()) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_HOURLY: - if interval == 1 { - return fmt.Sprintf("%d * * * *", startTime.Minute()) - } - return fmt.Sprintf("%d */%d * * *", startTime.Minute(), interval) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY: - return fmt.Sprintf("%d %d * * %d", startTime.Minute(), startTime.Hour(), startTime.Weekday()) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_MONTHLY: - return fmt.Sprintf("%d %d %d * *", startTime.Minute(), startTime.Hour(), startTime.Day()) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_BIWEEKLY: - return fmt.Sprintf("%d %d */14 * %d", startTime.Minute(), startTime.Hour(), startTime.Weekday()) - default: - return "" - } -} - -type IntervalSchedule struct { - Frequency rtypes.ScheduleFrequency - Interval int - StartTime time.Time - Minute int - Hour int - Day int - Weekday time.Weekday - Location *time.Location -} - -func NewIntervalSchedule(frequency rtypes.ScheduleFrequency, startTime time.Time, interval int) (*IntervalSchedule, error) { - if interval < 1 { - return nil, fmt.Errorf("failed to create interval schedule: interval must be at least 1") - } - - return &IntervalSchedule{ - Frequency: frequency, - Interval: interval, - StartTime: startTime, - Minute: startTime.Minute(), - Hour: startTime.Hour(), - Day: startTime.Day(), - Weekday: startTime.Weekday(), - Location: startTime.Location(), - }, nil -} - -func (s *IntervalSchedule) ToRangeFrom(from time.Time) (time.Time, time.Time) { - next := s.Next(from) // Compute the next scheduled execution time after `from`. - - // Backtrack to find the previous execution time - // by walking back from `from` until `Next(prev)` == `next` - // Start with candidate time far in the past - prev := s.StartTime - for { - n := s.Next(prev) - if !n.Before(next) || n.After(from) { - break - } - prev = n - } - return prev, next -} - -func (s *IntervalSchedule) Next(t time.Time) time.Time { - t = t.In(s.Location) - - switch s.Frequency { - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY: - return s.nextDaily(t) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY: - return s.nextWeekly(t) - case rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_MONTHLY: - return s.nextMonthly(t) - default: - return time.Time{} - } -} - -func (s *IntervalSchedule) nextDaily(t time.Time) time.Time { - // Create candidate time with the correct hour and minute on the current day - candidate := time.Date(t.Year(), t.Month(), t.Day(), s.Hour, s.Minute, 0, 0, s.Location) - - // If the candidate is in the past, move to the next day - if !candidate.After(t) { - candidate = candidate.AddDate(0, 0, 1) - } - - // Calculate the absolute number of days from the epoch for both start time and candidate - // This ensures proper alignment regardless of month boundaries - startDays := int(s.StartTime.Unix() / secondsInDay) - candidateDays := int(candidate.Unix() / secondsInDay) - - // Calculate how many days past the start time - daysPastStart := candidateDays - startDays - - // If we're already on a valid day, return the candidate - if daysPastStart >= 0 && daysPastStart%s.Interval == 0 { - return candidate - } - - // Otherwise, calculate days to add to reach the next valid day - daysToAdd := s.Interval - (daysPastStart % s.Interval) - if daysPastStart < 0 { - // Special handling if we're before the start time - daysToAdd = -daysPastStart - } - - return candidate.AddDate(0, 0, daysToAdd) -} - -// nextWeekly calculates the next execution for weekly intervals > 1 -func (s *IntervalSchedule) nextWeekly(t time.Time) time.Time { - // First find the next occurrence of the correct weekday - daysUntilWeekday := int(s.Weekday - t.Weekday()) - if daysUntilWeekday <= 0 { - daysUntilWeekday += 7 - } - - // Create the candidate time with the correct weekday, hour, and minute - candidate := time.Date( - t.Year(), t.Month(), t.Day()+daysUntilWeekday, - s.Hour, s.Minute, 0, 0, s.Location, - ) - - // If the candidate is in the past, move to the next week - if !candidate.After(t) { - candidate = candidate.AddDate(0, 0, 7) - } - - // Calculate absolute number of weeks from epoch for proper alignment - // Using Monday as the start of the week for consistent calculations - startWeeks := int(timeToMondayMidnight(s.StartTime).Unix() / secondsInWeek) - candidateWeeks := int(timeToMondayMidnight(candidate).Unix() / secondsInWeek) - - // Calculate how many weeks past the start time - weeksPastStart := candidateWeeks - startWeeks - - // If we're already on a valid week, return the candidate - if weeksPastStart >= 0 && weeksPastStart%s.Interval == 0 { - return candidate - } - - // Otherwise, calculate weeks to add to reach the next valid week - weeksToAdd := s.Interval - (weeksPastStart % s.Interval) - if weeksPastStart < 0 { - // Special handling if we're before the start time - weeksToAdd = -weeksPastStart - } - - return candidate.AddDate(0, 0, 7*weeksToAdd) -} - -func (s *IntervalSchedule) nextMonthly(t time.Time) time.Time { - // Always start from at least the schedule's start time - if t.Before(s.StartTime) { - t = s.StartTime - } - - // Calculate total months since the epoch (or any fixed reference point) - startMonths := s.StartTime.Year()*12 + int(s.StartTime.Month()) - 1 - currentMonths := t.Year()*12 + int(t.Month()) - 1 - - // Calculate how many intervals have passed since start - intervalsPassed := (currentMonths - startMonths) / s.Interval - - // Calculate the last interval month - lastIntervalMonth := startMonths + intervalsPassed*s.Interval - - // Calculate the next interval month - nextIntervalMonth := lastIntervalMonth - - // If we're already past the day/time in the current interval month, - // or if we're exactly at the current interval month but before the start date, - // move to the next interval - if currentMonths > lastIntervalMonth || - (currentMonths == lastIntervalMonth && - (t.Day() > s.Day || (t.Day() == s.Day && (t.Hour() > s.Hour || (t.Hour() == s.Hour && t.Minute() >= s.Minute))))) { - nextIntervalMonth = lastIntervalMonth + s.Interval - } - - // Convert back to year and month - nextYear := nextIntervalMonth / 12 - nextMonth := time.Month(nextIntervalMonth%12 + 1) - - // Create the candidate time - candidate := time.Date(nextYear, nextMonth, s.Day, s.Hour, s.Minute, 0, 0, s.Location) - - // Handle months with fewer days than our target day - if candidate.Day() != s.Day { - // We got bumped to the next month due to day overflow, go back to last day of previous month - candidate = time.Date(nextYear, nextMonth, 0, s.Hour, s.Minute, 0, 0, s.Location) - } - - return candidate -} - -func timeToMondayMidnight(t time.Time) time.Time { - daysFromMonday := int(t.Weekday()) - if daysFromMonday == 0 { // Sunday - daysFromMonday = 6 - } else { - daysFromMonday-- - } - - return time.Date(t.Year(), t.Month(), t.Day()-daysFromMonday, 0, 0, 0, 0, t.Location()) -} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go deleted file mode 100644 index 0f89be1..0000000 --- a/internal/scheduler/scheduler_test.go +++ /dev/null @@ -1,69 +0,0 @@ -package scheduler - -import ( - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "github.com/vultisig/recipes/types" -) - -func TestIntervalSchedule_ToRangeFrom(t *testing.T) { - tests := []struct { - name string - startTime time.Time - freq types.ScheduleFrequency - interval int - ref time.Time - wantLast time.Time - wantNext time.Time - }{ - { - name: "freq daily", - freq: types.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY, - interval: 1, - startTime: parseTime("2025-01-01T00:00:00Z"), - ref: parseTime("2025-01-02T10:00:00Z"), - wantLast: parseTime("2025-01-02T00:00:00Z"), - wantNext: parseTime("2025-01-03T00:00:00Z"), - }, - { - name: "freq weekly", - freq: types.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY, - interval: 1, - startTime: parseTime("2025-01-01T00:00:00Z"), - ref: parseTime("2025-01-01T10:00:00Z"), - wantLast: parseTime("2025-01-01T00:00:00Z"), - wantNext: parseTime("2025-01-08T00:00:00Z"), - }, - { - name: "freq monthly", - freq: types.ScheduleFrequency_SCHEDULE_FREQUENCY_MONTHLY, - interval: 1, - startTime: parseTime("2025-01-01T00:00:00Z"), - ref: parseTime("2025-01-01T10:00:00Z"), - wantLast: parseTime("2025-01-01T00:00:00Z"), - wantNext: parseTime("2025-02-01T00:00:00Z"), - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - schedule, err := NewIntervalSchedule(tt.freq, tt.startTime, tt.interval) - require.NoError(t, err) - - last, next := schedule.ToRangeFrom(tt.ref) - assert.Equal(t, tt.wantLast, last) - assert.Equal(t, tt.wantNext, next) - }) - } -} - -func parseTime(s string) time.Time { - t, err := time.Parse(time.RFC3339Nano, s) - if err != nil { - panic(err) - } - return t -} diff --git a/internal/scheduler/service.go b/internal/scheduler/service.go new file mode 100644 index 0000000..d99c3cc --- /dev/null +++ b/internal/scheduler/service.go @@ -0,0 +1,15 @@ +package scheduler + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/vultisig/verifier/types" +) + +type Service interface { + Create(ctx context.Context, tx pgx.Tx, policy types.PluginPolicy) error + Update(ctx context.Context, tx pgx.Tx, oldPolicy, newPolicy types.PluginPolicy) error + Delete(ctx context.Context, tx pgx.Tx, policyID uuid.UUID) error +} diff --git a/internal/scheduler/service_nil.go b/internal/scheduler/service_nil.go new file mode 100644 index 0000000..9a8aa42 --- /dev/null +++ b/internal/scheduler/service_nil.go @@ -0,0 +1,28 @@ +package scheduler + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/vultisig/verifier/types" +) + +// NilService implements the scheduler.Service for plugins where scheduling not required +type NilService struct{} + +func NewNilService() *NilService { + return &NilService{} +} + +func (s *NilService) Create(_ context.Context, _ pgx.Tx, _ types.PluginPolicy) error { + return nil +} + +func (s *NilService) Update(_ context.Context, _ pgx.Tx, _, _ types.PluginPolicy) error { + return nil +} + +func (s *NilService) Delete(_ context.Context, _ pgx.Tx, _ uuid.UUID) error { + return nil +} diff --git a/internal/scheduler/storage.go b/internal/scheduler/storage.go new file mode 100644 index 0000000..8cf838f --- /dev/null +++ b/internal/scheduler/storage.go @@ -0,0 +1,23 @@ +package scheduler + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +type Scheduler struct { + PolicyID uuid.UUID `json:"policy_id"` + NextExecution time.Time `json:"next_execution"` +} + +type Storage interface { + GetByPolicy(ctx context.Context, policyID uuid.UUID) (Scheduler, error) + CreateWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID, next time.Time) error + DeleteWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID) error + GetPending(ctx context.Context) ([]Scheduler, error) + SetNext(ctx context.Context, policyID uuid.UUID, next time.Time) error + SetNextWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID, next time.Time) error +} diff --git a/internal/scheduler/storage_pg.go b/internal/scheduler/storage_pg.go new file mode 100644 index 0000000..e8d8619 --- /dev/null +++ b/internal/scheduler/storage_pg.go @@ -0,0 +1,110 @@ +package scheduler + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type PostgresStorage struct { + pool *pgxpool.Pool +} + +func NewPostgresStorage(pool *pgxpool.Pool) *PostgresStorage { + return &PostgresStorage{ + pool: pool, + } +} + +func (s *PostgresStorage) CreateWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID, next time.Time) error { + _, err := tx.Exec(ctx, ` + INSERT INTO scheduler (policy_id, next_execution) + VALUES ($1, $2) + `, policyID, next) + if err != nil { + return fmt.Errorf("failed to create scheduler entry: %w", err) + } + return nil +} + +func (s *PostgresStorage) GetByPolicy(ctx context.Context, policyID uuid.UUID) (Scheduler, error) { + var scheduler Scheduler + err := s.pool.QueryRow(ctx, ` + SELECT policy_id, next_execution + FROM scheduler + WHERE policy_id = $1 + LIMIT 1 + `, policyID).Scan(&scheduler.PolicyID, &scheduler.NextExecution) + if err != nil { + return Scheduler{}, fmt.Errorf("failed to query scheduler by policy: %w", err) + } + + return scheduler, nil +} + +func (s *PostgresStorage) GetPending(ctx context.Context) ([]Scheduler, error) { + rows, err := s.pool.Query(ctx, ` + SELECT policy_id, next_execution + FROM scheduler + WHERE next_execution <= NOW() + ORDER BY next_execution + `) + if err != nil { + return nil, fmt.Errorf("failed to query pending scheduler entries: %w", err) + } + defer rows.Close() + + var schedulers []Scheduler + for rows.Next() { + var scheduler Scheduler + if err := rows.Scan(&scheduler.PolicyID, &scheduler.NextExecution); err != nil { + return nil, fmt.Errorf("failed to scan scheduler entry: %w", err) + } + schedulers = append(schedulers, scheduler) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate over scheduler entries: %w", err) + } + + return schedulers, nil +} + +func (s *PostgresStorage) SetNext(ctx context.Context, policyID uuid.UUID, next time.Time) error { + _, err := s.pool.Exec(ctx, ` + UPDATE scheduler + SET next_execution = $2 + WHERE policy_id = $1 + `, policyID, next) + if err != nil { + return fmt.Errorf("failed to update next execution time: %w", err) + } + return nil +} + +func (s *PostgresStorage) SetNextWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID, next time.Time) error { + _, err := tx.Exec(ctx, ` + UPDATE scheduler + SET next_execution = $2 + WHERE policy_id = $1 + `, policyID, next) + if err != nil { + return fmt.Errorf("failed to update next execution time: %w", err) + } + return nil +} + +func (s *PostgresStorage) DeleteWithTx(ctx context.Context, tx pgx.Tx, policyID uuid.UUID) error { + _, err := tx.Exec(ctx, ` + DELETE FROM scheduler + WHERE policy_id = $1 + `, policyID) + if err != nil { + return fmt.Errorf("failed to delete scheduler entry: %w", err) + } + return nil +} diff --git a/internal/scheduler/worker.go b/internal/scheduler/worker.go new file mode 100644 index 0000000..c8f8457 --- /dev/null +++ b/internal/scheduler/worker.go @@ -0,0 +1,142 @@ +package scheduler + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/hibiken/asynq" + "github.com/sirupsen/logrus" + "github.com/vultisig/verifier/tx_indexer/pkg/graceful" + "golang.org/x/sync/errgroup" +) + +type Worker struct { + logger *logrus.Logger + + client *asynq.Client + task string + queue string + + repo Storage + interval Interval + policy PolicyFetcher + + pollInterval time.Duration + iterationTimeout time.Duration +} + +func NewWorker( + logger *logrus.Logger, + client *asynq.Client, + task, + queue string, + repo Storage, + interval Interval, + policy PolicyFetcher, +) *Worker { + return &Worker{ + logger: logger.WithField("pkg", "scheduler.Worker").Logger, + client: client, + task: task, + queue: queue, + repo: repo, + interval: interval, + policy: policy, + pollInterval: 30 * time.Second, + iterationTimeout: 30 * time.Second, + } +} + +func (w *Worker) Run() error { + ctx, stop := context.WithCancel(context.Background()) + + go func() { + graceful.HandleSignals(stop) + w.logger.Info("got exit signal, will stop after current processing step finished...") + }() + + err := w.start(ctx) + if err != nil { + return fmt.Errorf("failed to start: %w", err) + } + return nil +} + +func (w *Worker) start(aliveCtx context.Context) error { + err := w.enqueue() + if err != nil { + return fmt.Errorf("failed to enqueue: %w", err) + } + + for { + select { + case <-aliveCtx.Done(): + w.logger.Infof("context done & no processing: stop worker") + return nil + case <-time.After(w.pollInterval): + er := w.enqueue() + if er != nil { + w.logger.Errorf("processing error, continue loop: %v", er) + } + } + } +} + +func (w *Worker) enqueue() error { + ctx, cancel := context.WithTimeout(context.Background(), w.iterationTimeout) + defer cancel() + + w.logger.Info("worker tick") + + tasks, err := w.repo.GetPending(ctx) + if err != nil { + return fmt.Errorf("failed to get pending tasks: %w", err) + } + + eg := &errgroup.Group{} + for _, _task := range tasks { + task := _task + eg.Go(func() error { + policy, er := w.policy.GetPluginPolicy(ctx, task.PolicyID) + if er != nil { + return fmt.Errorf("failed to fetch policy: %w", er) + } + + next, er := w.interval.FromNowWhenNext(*policy) + if er != nil { + return fmt.Errorf("failed to compute next: %w", er) + } + + buf, er := json.Marshal(task) + if er != nil { + return fmt.Errorf("failed to marshal task: %w", er) + } + + _, er = w.client.EnqueueContext( + ctx, + asynq.NewTask(w.task, buf), + asynq.MaxRetry(0), + asynq.Timeout(5*time.Minute), + asynq.Retention(10*time.Minute), + asynq.Queue(w.queue), + ) + if er != nil { + return fmt.Errorf("failed to enqueue task: %w", er) + } + + er = w.repo.SetNext(ctx, task.PolicyID, next) + if er != nil { + return fmt.Errorf("failed to set next: %w", er) + } + return nil + }) + } + err = eg.Wait() + if err != nil { + return fmt.Errorf("failed to process tasks: %w", err) + } + + return nil +} diff --git a/internal/tasks/tasks.go b/internal/tasks/tasks.go index 6989733..314a3bb 100644 --- a/internal/tasks/tasks.go +++ b/internal/tasks/tasks.go @@ -9,7 +9,6 @@ import ( const QUEUE_NAME = "plugin_queue" const ( TypePluginTransaction = "plugin:transaction" - TypeKeyGenerationDKLS = "key:generationDKLS" TypeKeySignDKLS = "key:signDKLS" TypeReshareDKLS = "key:reshareDKLS" ) diff --git a/internal/types/time_trigger.go b/internal/types/time_trigger.go deleted file mode 100644 index f0e8de6..0000000 --- a/internal/types/time_trigger.go +++ /dev/null @@ -1,26 +0,0 @@ -package types - -import ( - "time" - - "github.com/google/uuid" - rtypes "github.com/vultisig/recipes/types" -) - -type TimeTriggerStatus string - -const ( - StatusTimeTriggerPending TimeTriggerStatus = "PENDING" - StatusTimeTriggerRunning TimeTriggerStatus = "RUNNING" -) - -type TimeTrigger struct { - PolicyID uuid.UUID `json:"policy_id"` - CronExpression string `json:"cron_expression"` - StartTime time.Time `json:"start_time"` - EndTime *time.Time `json:"end_time"` - Frequency rtypes.ScheduleFrequency `json:"frequency"` - Interval int `json:"interval"` - LastExecution *time.Time `json:"last_execution"` - Status TimeTriggerStatus `json:"status"` -} diff --git a/payroll.scheduler.example.json b/payroll.scheduler.example.json new file mode 100644 index 0000000..b4bafcd --- /dev/null +++ b/payroll.scheduler.example.json @@ -0,0 +1,10 @@ +{ + "redis": { + "host": "localhost", + "port": "6378", + "password": "password" + }, + "database": { + "dsn": "postgres://myuser:mypassword@plugin-db:5432/vultisig-payroll?sslmode=disable" + } +} diff --git a/payroll.server.example.json b/payroll.server.example.json index 15947b9..4f74b60 100644 --- a/payroll.server.example.json +++ b/payroll.server.example.json @@ -7,7 +7,6 @@ "database": { "dsn": "postgres://myuser:mypassword@plugin-db:5432/vultisig-payroll?sslmode=disable" }, - "base_config_path": "../../../etc/vultisig", "redis": { "host": "localhost", "port": "6378", diff --git a/plugin/dca/dca.go b/plugin/dca/dca.go index 22e1ffb..39d9cb8 100644 --- a/plugin/dca/dca.go +++ b/plugin/dca/dca.go @@ -816,15 +816,6 @@ func (p *DCAPlugin) GetRecipeSpecification() (*rtypes.RecipeSchema, error) { Required: true, }, }, - Scheduling: &rtypes.SchedulingCapability{ - SupportsScheduling: true, - SupportedFrequencies: []rtypes.ScheduleFrequency{ - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_HOURLY, - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY, - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY, - }, - MaxScheduledExecutions: 1000, // DCA can have many small orders - }, Requirements: &rtypes.PluginRequirements{ MinVultisigVersion: 1, SupportedChains: []string{"ethereum"}, diff --git a/plugin/payroll/interval.go b/plugin/payroll/interval.go new file mode 100644 index 0000000..db0c7ed --- /dev/null +++ b/plugin/payroll/interval.go @@ -0,0 +1,40 @@ +package payroll + +import ( + "fmt" + "time" + + "github.com/vultisig/verifier/types" +) + +type Interval struct { +} + +func NewSchedulerInterval() *Interval { + return &Interval{} +} + +func (i *Interval) FromNowWhenNext(policy types.PluginPolicy) (time.Time, error) { + recipe, err := policy.GetRecipe() + if err != nil { + return time.Time{}, fmt.Errorf("failed to unpack recipe: %w", err) + } + + cfg := recipe.GetConfiguration().GetFields() + + var next time.Time + freq := cfg[frequency].GetStringValue() + switch freq { + case daily: + next = time.Now().AddDate(0, 0, 1) + case weekly: + next = time.Now().AddDate(0, 0, 7) + case biWeekly: + next = time.Now().AddDate(0, 0, 14) + case monthly: + next = time.Now().AddDate(0, 1, 0) + default: + return time.Time{}, fmt.Errorf("unknown frequency: %s", freq) + } + return next, nil +} diff --git a/plugin/payroll/scheduler.go b/plugin/payroll/scheduler.go new file mode 100644 index 0000000..b6dfef6 --- /dev/null +++ b/plugin/payroll/scheduler.go @@ -0,0 +1,71 @@ +package payroll + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/vultisig/plugin/internal/scheduler" + "github.com/vultisig/verifier/types" +) + +type SchedulerService struct { + repo scheduler.Storage +} + +func NewSchedulerService(repo scheduler.Storage) *SchedulerService { + return &SchedulerService{ + repo: repo, + } +} + +func (s *SchedulerService) Create(ctx context.Context, tx pgx.Tx, policy types.PluginPolicy) error { + start, err := startDateFromPolicy(policy) + if err != nil { + return fmt.Errorf("failed to unpack start date from policy: %w", err) + } + return s.repo.CreateWithTx(ctx, tx, policy.ID, start) +} + +func startDateFromPolicy(policy types.PluginPolicy) (time.Time, error) { + recipe, err := policy.GetRecipe() + if err != nil { + return time.Time{}, fmt.Errorf("failed to unpack recipe: %w", err) + } + + cfg := recipe.GetConfiguration().GetFields() + + start := time.Now() + cfgStartDate := cfg[startDate].GetStringValue() + if cfgStartDate != "" { + t, er := time.Parse(time.RFC3339, cfgStartDate) + if er != nil { + return time.Time{}, fmt.Errorf("failed to parse start date (%s): %w", cfgStartDate, er) + } + start = t + } + return start, nil +} + +func (s *SchedulerService) Update(ctx context.Context, tx pgx.Tx, oldPolicy, newPolicy types.PluginPolicy) error { + oldStart, err := startDateFromPolicy(oldPolicy) + if err != nil { + return fmt.Errorf("failed to unpack start date from policy: %w", err) + } + newStart, err := startDateFromPolicy(newPolicy) + if err != nil { + return fmt.Errorf("failed to unpack start date from policy: %w", err) + } + if oldStart.Equal(newStart) { + // no changes + return nil + } + + return s.repo.SetNextWithTx(ctx, tx, oldPolicy.ID, newStart) +} + +func (s *SchedulerService) Delete(ctx context.Context, tx pgx.Tx, policyID uuid.UUID) error { + return s.repo.DeleteWithTx(ctx, tx, policyID) +} diff --git a/plugin/payroll/transaction.go b/plugin/payroll/transaction.go index 36f73e7..c7d8786 100644 --- a/plugin/payroll/transaction.go +++ b/plugin/payroll/transaction.go @@ -17,6 +17,7 @@ import ( "github.com/vultisig/mobile-tss-lib/tss" "github.com/vultisig/plugin/common" "github.com/vultisig/plugin/internal/plugin" + "github.com/vultisig/plugin/internal/scheduler" "github.com/vultisig/recipes/ethereum" "github.com/vultisig/recipes/sdk/evm" rtypes "github.com/vultisig/recipes/types" @@ -26,8 +27,6 @@ import ( vtypes "github.com/vultisig/verifier/types" "github.com/vultisig/vultiserver/contexthelper" "golang.org/x/sync/errgroup" - - "github.com/vultisig/plugin/internal/types" ) func (p *Plugin) HandleSchedulerTrigger(c context.Context, t *asynq.Task) error { @@ -38,7 +37,7 @@ func (p *Plugin) HandleSchedulerTrigger(c context.Context, t *asynq.Task) error p.logger.WithError(err).Warn("Context cancelled, skipping scheduler trigger") return err } - var trigger types.TimeTrigger + var trigger scheduler.Scheduler if err := json.Unmarshal(t.Payload(), &trigger); err != nil { p.logger.WithError(err).Error("Failed to unmarshal trigger payload") return fmt.Errorf("failed to unmarshal trigger payload: %s, %w", err, asynq.SkipRetry) @@ -339,14 +338,6 @@ func (p *Plugin) GetRecipeSpecification() (*rtypes.RecipeSchema, error) { Required: true, }, }, - Scheduling: &rtypes.SchedulingCapability{ - SupportsScheduling: true, - SupportedFrequencies: []rtypes.ScheduleFrequency{ - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_WEEKLY, - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_BIWEEKLY, - rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_MONTHLY, - }, - }, Configuration: cfg, Requirements: &rtypes.PluginRequirements{ MinVultisigVersion: 1, diff --git a/scripts/dev/create_fee_policy/dummy_server.go b/scripts/dev/create_fee_policy/dummy_server.go index 1070f49..1781797 100644 --- a/scripts/dev/create_fee_policy/dummy_server.go +++ b/scripts/dev/create_fee_policy/dummy_server.go @@ -10,6 +10,7 @@ import ( "github.com/sirupsen/logrus" "github.com/spf13/viper" "github.com/vultisig/plugin/api" + "github.com/vultisig/plugin/internal/scheduler" "github.com/vultisig/plugin/plugin/fees" "github.com/vultisig/plugin/storage" "github.com/vultisig/plugin/storage/postgres" @@ -116,11 +117,11 @@ func createDummyServer() (*api.Server, vault.Storage, *fees.FeePlugin) { db, redisStorage, vaultStorage, - redisOptions, client, inspector, sdClient, nil, + scheduler.NewNilService(), ) return server, vaultStorage, &p diff --git a/service/policy.go b/service/policy.go index 1619dd2..3b8eeb9 100644 --- a/service/policy.go +++ b/service/policy.go @@ -26,11 +26,11 @@ var _ Policy = (*PolicyService)(nil) type PolicyService struct { db storage.DatabaseStorage - scheduler *scheduler.SchedulerService + scheduler scheduler.Service logger *logrus.Logger } -func NewPolicyService(db storage.DatabaseStorage, scheduler *scheduler.SchedulerService, logger *logrus.Logger) (*PolicyService, error) { +func NewPolicyService(db storage.DatabaseStorage, scheduler scheduler.Service, logger *logrus.Logger) (*PolicyService, error) { if db == nil { return nil, fmt.Errorf("database storage cannot be nil") } @@ -61,12 +61,11 @@ func (s *PolicyService) CreatePolicy(ctx context.Context, policy vtypes.PluginPo return nil, fmt.Errorf("failed to insert policy: %w", err) } - // Handle trigger if scheduler exists - if s.scheduler != nil { - if err := s.scheduler.CreateTimeTrigger(ctx, policy, tx); err != nil { - return nil, fmt.Errorf("failed to create time trigger: %w", err) - } + err = s.scheduler.Create(ctx, tx, policy) + if err != nil { + return nil, fmt.Errorf("failed to create policy in scheduler: %w", err) } + if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("failed to commit transaction: %w", err) } @@ -82,21 +81,20 @@ func (s *PolicyService) UpdatePolicy(ctx context.Context, policy vtypes.PluginPo } defer s.handleRollback(ctx, tx) - // Update policy with tx - updatedPolicy, err := s.db.UpdatePluginPolicyTx(ctx, tx, policy) + oldPolicy, err := s.db.GetPluginPolicy(ctx, policy.ID) if err != nil { - return nil, fmt.Errorf("failed to update policy: %w", err) + return nil, fmt.Errorf("failed to get plugin policy: %w", err) } - if s.scheduler != nil { - trigger, err := s.scheduler.GetTriggerFromPolicy(policy) - if err != nil { - return nil, fmt.Errorf("failed to get trigger from policy: %w", err) - } + err = s.scheduler.Update(ctx, tx, *oldPolicy, policy) + if err != nil { + return nil, fmt.Errorf("failed to update policy in scheduler: %w", err) + } - if err := s.db.UpdateTimeTriggerTx(ctx, policy.ID, *trigger, tx); err != nil { - return nil, fmt.Errorf("failed to update trigger execution tx: %w", err) - } + // Update policy with tx + updatedPolicy, err := s.db.UpdatePluginPolicyTx(ctx, tx, policy) + if err != nil { + return nil, fmt.Errorf("failed to update policy: %w", err) } if err := tx.Commit(ctx); err != nil { @@ -119,6 +117,11 @@ func (s *PolicyService) DeletePolicy(ctx context.Context, policyID uuid.UUID, si return fmt.Errorf("failed to delete policy: %w", err) } + err = s.scheduler.Delete(ctx, tx, policyID) + if err != nil { + return fmt.Errorf("failed to delete policy from scheduler: %w", err) + } + if err := tx.Commit(ctx); err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } diff --git a/storage/db.go b/storage/db.go index 8259256..114578d 100644 --- a/storage/db.go +++ b/storage/db.go @@ -21,15 +21,6 @@ type DatabaseStorage interface { InsertPluginPolicyTx(ctx context.Context, dbTx pgx.Tx, policy vtypes.PluginPolicy) (*vtypes.PluginPolicy, error) UpdatePluginPolicyTx(ctx context.Context, dbTx pgx.Tx, policy vtypes.PluginPolicy) (*vtypes.PluginPolicy, error) - CreateTimeTriggerTx(ctx context.Context, dbTx pgx.Tx, trigger types.TimeTrigger) error - GetPendingTimeTriggers(ctx context.Context) ([]types.TimeTrigger, error) - UpdateTimeTriggerLastExecution(ctx context.Context, policyID uuid.UUID) error - UpdateTimeTriggerTx(ctx context.Context, policyID uuid.UUID, trigger types.TimeTrigger, dbTx pgx.Tx) error - - DeleteTimeTrigger(ctx context.Context, policyID uuid.UUID) error - UpdateTriggerStatus(ctx context.Context, policyID uuid.UUID, status types.TimeTriggerStatus) error - GetTriggerStatus(ctx context.Context, policyID uuid.UUID) (types.TimeTriggerStatus, error) - CreateFeeRun(ctx context.Context, policyId uuid.UUID, state types.FeeRunState, fees []verifierapi.FeeDto) (*types.FeeRun, error) SetFeeRunSent(ctx context.Context, runId uuid.UUID, txId uuid.UUID) error diff --git a/storage/postgres/migrations/plugin/20241211182120_add_time_triggers.sql b/storage/postgres/migrations/plugin/20241211182120_add_time_triggers.sql deleted file mode 100644 index d0b3667..0000000 --- a/storage/postgres/migrations/plugin/20241211182120_add_time_triggers.sql +++ /dev/null @@ -1,27 +0,0 @@ --- +goose Up --- +goose StatementBegin -CREATE TYPE trigger_status AS ENUM ( - 'PENDING', - 'RUNNING' -); - -CREATE TABLE time_triggers ( - id SERIAL PRIMARY KEY, - policy_id UUID NOT NULL REFERENCES plugin_policies(id), - cron_expression TEXT NOT NULL, - start_time TIMESTAMP NOT NULL, - end_time TIMESTAMP, - frequency INT NOT NULL, - interval INTEGER NOT NULL, - last_execution TIMESTAMP, - status trigger_status NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP -); --- Add indexes for time_triggers -CREATE INDEX idx_time_triggers_policy_id ON time_triggers(policy_id); -CREATE INDEX idx_time_triggers_start_time ON time_triggers(start_time); --- +goose StatementEnd --- +goose Down --- +goose StatementBegin -DROP TABLE IF EXISTS time_triggers; --- +goose StatementEnd \ No newline at end of file diff --git a/storage/postgres/migrations/plugin/20250721165241_scheduler.sql b/storage/postgres/migrations/plugin/20250721165241_scheduler.sql new file mode 100644 index 0000000..0496160 --- /dev/null +++ b/storage/postgres/migrations/plugin/20250721165241_scheduler.sql @@ -0,0 +1,11 @@ +-- +goose Up +-- +goose StatementBegin +BEGIN; +CREATE TABLE IF NOT EXISTS scheduler ( + policy_id UUID PRIMARY KEY, + next_execution TIMESTAMP NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_scheduler_next_execution ON scheduler(next_execution); +END; +-- +goose StatementEnd diff --git a/storage/postgres/schema/schema.sql b/storage/postgres/schema/schema.sql index 53ac1d3..f8ead9e 100644 --- a/storage/postgres/schema/schema.sql +++ b/storage/postgres/schema/schema.sql @@ -5,11 +5,6 @@ CREATE TYPE "plugin_id" AS ENUM ( 'vultisig-fees-feee' ); -CREATE TYPE "trigger_status" AS ENUM ( - 'PENDING', - 'RUNNING' -); - CREATE TYPE "tx_indexer_status" AS ENUM ( 'PROPOSED', 'VERIFIED', @@ -22,39 +17,6 @@ CREATE TYPE "tx_indexer_status_onchain" AS ENUM ( 'FAIL' ); -CREATE FUNCTION "prevent_insert_if_policy_deleted"() RETURNS "trigger" - LANGUAGE "plpgsql" - AS $$ -BEGIN - IF NEW.deleted = true THEN - RAISE EXCEPTION 'Cannot insert a deleted policy'; - END IF; - RETURN NEW; -END; -$$; - -CREATE FUNCTION "prevent_update_if_policy_deleted"() RETURNS "trigger" - LANGUAGE "plpgsql" - AS $$ -BEGIN - IF OLD.deleted = true THEN - RAISE EXCEPTION 'Cannot update a deleted policy'; - END IF; - RETURN NEW; -END; -$$; - -CREATE FUNCTION "set_policy_inactive_on_delete"() RETURNS "trigger" - LANGUAGE "plpgsql" - AS $$ -BEGIN - IF NEW.deleted = true THEN - NEW.active := false; - END IF; - RETURN NEW; -END; -$$; - CREATE FUNCTION "update_updated_at_column"() RETURNS "trigger" LANGUAGE "plpgsql" AS $$ @@ -104,33 +66,14 @@ CREATE TABLE "plugin_policies" ( "recipe" "text" NOT NULL, "active" boolean DEFAULT true NOT NULL, "created_at" timestamp with time zone DEFAULT "now"() NOT NULL, - "updated_at" timestamp with time zone DEFAULT "now"() NOT NULL, - "deleted" boolean DEFAULT false NOT NULL + "updated_at" timestamp with time zone DEFAULT "now"() NOT NULL ); -CREATE TABLE "time_triggers" ( - "id" integer NOT NULL, +CREATE TABLE "scheduler" ( "policy_id" "uuid" NOT NULL, - "cron_expression" "text" NOT NULL, - "start_time" timestamp without time zone NOT NULL, - "end_time" timestamp without time zone, - "frequency" integer NOT NULL, - "interval" integer NOT NULL, - "last_execution" timestamp without time zone, - "status" "trigger_status" NOT NULL, - "created_at" timestamp without time zone DEFAULT CURRENT_TIMESTAMP + "next_execution" timestamp without time zone NOT NULL ); -CREATE SEQUENCE "time_triggers_id_seq" - AS integer - START WITH 1 - INCREMENT BY 1 - NO MINVALUE - NO MAXVALUE - CACHE 1; - -ALTER SEQUENCE "time_triggers_id_seq" OWNED BY "public"."time_triggers"."id"; - CREATE TABLE "tx_indexer" ( "id" "uuid" DEFAULT "gen_random_uuid"() NOT NULL, "plugin_id" character varying(255) NOT NULL, @@ -149,8 +92,6 @@ CREATE TABLE "tx_indexer" ( "updated_at" timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL ); -ALTER TABLE ONLY "time_triggers" ALTER COLUMN "id" SET DEFAULT "nextval"('"public"."time_triggers_id_seq"'::"regclass"); - ALTER TABLE ONLY "fee" ADD CONSTRAINT "fee_pkey" PRIMARY KEY ("id"); @@ -160,8 +101,8 @@ ALTER TABLE ONLY "fee_run" ALTER TABLE ONLY "plugin_policies" ADD CONSTRAINT "plugin_policies_pkey" PRIMARY KEY ("id"); -ALTER TABLE ONLY "time_triggers" - ADD CONSTRAINT "time_triggers_pkey" PRIMARY KEY ("id"); +ALTER TABLE ONLY "scheduler" + ADD CONSTRAINT "scheduler_pkey" PRIMARY KEY ("policy_id"); ALTER TABLE ONLY "tx_indexer" ADD CONSTRAINT "tx_indexer_pkey" PRIMARY KEY ("id"); @@ -178,20 +119,12 @@ CREATE INDEX "idx_plugin_policies_plugin_id" ON "plugin_policies" USING "btree" CREATE INDEX "idx_plugin_policies_public_key" ON "plugin_policies" USING "btree" ("public_key"); -CREATE INDEX "idx_time_triggers_policy_id" ON "time_triggers" USING "btree" ("policy_id"); - -CREATE INDEX "idx_time_triggers_start_time" ON "time_triggers" USING "btree" ("start_time"); +CREATE INDEX "idx_scheduler_next_execution" ON "scheduler" USING "btree" ("next_execution"); CREATE INDEX "idx_tx_indexer_key" ON "tx_indexer" USING "btree" ("chain_id", "plugin_id", "policy_id", "token_id", "to_public_key", "created_at"); CREATE INDEX "idx_tx_indexer_status_onchain_lost" ON "tx_indexer" USING "btree" ("status_onchain", "lost"); -CREATE TRIGGER "trg_prevent_insert_if_policy_deleted" BEFORE INSERT ON "plugin_policies" FOR EACH ROW EXECUTE FUNCTION "public"."prevent_insert_if_policy_deleted"(); - -CREATE TRIGGER "trg_prevent_update_if_policy_deleted" BEFORE UPDATE ON "plugin_policies" FOR EACH ROW WHEN (("old"."deleted" = true)) EXECUTE FUNCTION "public"."prevent_update_if_policy_deleted"(); - -CREATE TRIGGER "trg_set_policy_inactive_on_delete" BEFORE INSERT OR UPDATE ON "plugin_policies" FOR EACH ROW WHEN (("new"."deleted" = true)) EXECUTE FUNCTION "public"."set_policy_inactive_on_delete"(); - CREATE TRIGGER "update_fee_run_updated_at" BEFORE UPDATE ON "fee_run" FOR EACH ROW EXECUTE FUNCTION "public"."update_updated_at_column"(); ALTER TABLE ONLY "fee" @@ -203,6 +136,3 @@ ALTER TABLE ONLY "fee_run" ALTER TABLE ONLY "fee_run" ADD CONSTRAINT "fee_run_tx_id_fkey" FOREIGN KEY ("tx_id") REFERENCES "tx_indexer"("id") ON DELETE SET NULL; -ALTER TABLE ONLY "time_triggers" - ADD CONSTRAINT "time_triggers_policy_id_fkey" FOREIGN KEY ("policy_id") REFERENCES "plugin_policies"("id"); - diff --git a/storage/postgres/time_trigger.go b/storage/postgres/time_trigger.go deleted file mode 100644 index 404957f..0000000 --- a/storage/postgres/time_trigger.go +++ /dev/null @@ -1,180 +0,0 @@ -package postgres - -import ( - "context" - "errors" - "fmt" - "time" - - "github.com/google/uuid" - "github.com/jackc/pgx/v5" - - "github.com/vultisig/plugin/internal/types" -) - -// checkContext returns ctx.Err() if the context is done (timeout or canceled), otherwise nil. -func checkContext(ctx context.Context) error { - if ctx == nil { - return fmt.Errorf("context is nil") - } - return ctx.Err() -} - -func (p *PostgresBackend) CreateTimeTriggerTx(ctx context.Context, tx pgx.Tx, trigger types.TimeTrigger) error { - if err := checkContext(ctx); err != nil { - return err - } - - query := ` - INSERT INTO time_triggers - (policy_id, cron_expression, start_time, end_time, frequency, interval, status) - VALUES ($1, $2, $3, $4, $5, $6, $7) - ` - - _, err := tx.Exec(ctx, query, - trigger.PolicyID, - trigger.CronExpression, - trigger.StartTime, - trigger.EndTime, - trigger.Frequency, - trigger.Interval, - trigger.Status, - ) - - return err -} - -func (p *PostgresBackend) DeleteTimeTrigger(ctx context.Context, policyID uuid.UUID) error { - if err := checkContext(ctx); err != nil { - return err - } - - query := `DELETE FROM time_triggers WHERE policy_id = $1` - _, err := p.pool.Exec(ctx, query, policyID) - - return err -} - -func (p *PostgresBackend) GetPendingTimeTriggers(ctx context.Context) ([]types.TimeTrigger, error) { - if err := checkContext(ctx); err != nil { - return nil, err - } - - // TODO: add limit and proper index - query := ` - WITH active_triggers AS ( - SELECT t.policy_id, t.cron_expression, t.start_time, t.end_time, t.frequency, t.interval, t.last_execution, t.status - FROM time_triggers t - INNER JOIN plugin_policies p ON t.policy_id = p.id - WHERE t.start_time <= $1 - AND (t.end_time IS NULL OR t.end_time > $1) - AND p.active = true - AND t.status = 'PENDING' - AND (t.last_execution IS NULL OR t.last_execution < $1) - ) - SELECT * FROM active_triggers - ORDER BY start_time ASC - ` - - rows, err := p.pool.Query(ctx, query, time.Now().UTC()) - if err != nil { - return nil, err - } - defer rows.Close() - - var triggers []types.TimeTrigger - for rows.Next() { - var t types.TimeTrigger - err := rows.Scan( - &t.PolicyID, - &t.CronExpression, - &t.StartTime, - &t.EndTime, - &t.Frequency, - &t.Interval, - &t.LastExecution, - &t.Status) - if err != nil { - return nil, err - } - triggers = append(triggers, t) - } - - return triggers, nil -} - -func (p *PostgresBackend) UpdateTimeTriggerLastExecution(ctx context.Context, policyID uuid.UUID) error { - if err := checkContext(ctx); err != nil { - return err - } - - query := ` - UPDATE time_triggers - SET last_execution = $2 - WHERE policy_id = $1 - ` - - _, err := p.pool.Exec(ctx, query, policyID, time.Now().UTC()) - return err -} - -func (p *PostgresBackend) UpdateTimeTriggerTx(ctx context.Context, policyID uuid.UUID, trigger types.TimeTrigger, tx pgx.Tx) error { - if err := checkContext(ctx); err != nil { - return err - } - - query := ` - UPDATE time_triggers - SET start_time = $2, - frequency = $3, - interval = $4, - cron_expression = $5 - WHERE policy_id = $1 - ` - _, err := tx.Exec(ctx, query, - policyID, - trigger.StartTime, - trigger.Frequency, - trigger.Interval, - trigger.CronExpression, - ) - return err -} - -func (p *PostgresBackend) GetTriggerStatus(ctx context.Context, policyID uuid.UUID) (types.TimeTriggerStatus, error) { - if err := checkContext(ctx); err != nil { - return "", err - } - - query := ` - SELECT status - FROM time_triggers - WHERE policy_id = $1 - ` - - var status types.TimeTriggerStatus - err := p.pool.QueryRow(ctx, query, policyID).Scan(&status) - if err != nil { - if errors.Is(err, pgx.ErrNoRows) { - return "", fmt.Errorf("trigger not found for policy_id: %s", policyID) - } - return "", err - } - - return status, nil -} - -func (p *PostgresBackend) UpdateTriggerStatus(ctx context.Context, policyID uuid.UUID, status types.TimeTriggerStatus) error { - if err := checkContext(ctx); err != nil { - return err - } - - query := ` - UPDATE time_triggers - SET status = $2 - WHERE policy_id = $1 - ` - - _, err := p.pool.Exec(ctx, query, policyID, status) - return err -} diff --git a/storage/postgres/time_trigger_test.go b/storage/postgres/time_trigger_test.go deleted file mode 100644 index 2059a00..0000000 --- a/storage/postgres/time_trigger_test.go +++ /dev/null @@ -1,83 +0,0 @@ -package postgres - -import ( - "context" - "testing" - "time" - - "github.com/google/uuid" - rtypes "github.com/vultisig/recipes/types" - vtypes "github.com/vultisig/verifier/types" - - "github.com/vultisig/plugin/internal/types" -) - -func TestTimeTrigger(t *testing.T) { - t.SkipNow() - // Initialize Postgres backend - backend, err := NewPostgresBackend("postgres://myuser:mypassword@localhost:5432/vultisig-plugin?sslmode=disable", nil) - if err != nil { - t.Fatalf("Failed to create Postgres backend: %v", err) - } - defer backend.Close() - policyID := uuid.New() - - // Create a new time trigger - trigger := types.TimeTrigger{ - PolicyID: policyID, - CronExpression: "0 0 * * *", - StartTime: time.Now().UTC(), - EndTime: nil, - Frequency: rtypes.ScheduleFrequency_SCHEDULE_FREQUENCY_DAILY, - Interval: 1, - Status: "PENDING", - } - - ctx := context.Background() - tx, err := backend.pool.Begin(ctx) - if err != nil { - t.Fatalf("Failed to begin transaction: %v", err) - } - if _, err := backend.InsertPluginPolicyTx(ctx, tx, vtypes.PluginPolicy{ - ID: policyID, - PublicKey: "4a7b9c2f8e1d3a5b6c4f9e2d7a1b3c5f8d4e6a2b9c1f7e3d5a6b4c2f8e1d3a5b", - PluginID: "vultisig-payroll-0000", - PluginVersion: "1", - PolicyVersion: 1, - Signature: "whatever", - Recipe: "k3uaL47TpWtsT54tfhO7K6Gbyxf4H71jWma0zC2e3kQ=", - Active: true, - }); err != nil { - tx.Rollback(ctx) - t.Fatalf("Failed to insert plugin policy: %v", err) - } - if err := backend.CreateTimeTriggerTx(ctx, tx, trigger); err != nil { - tx.Rollback(ctx) - t.Fatalf("Failed to create time trigger: %v", err) - } - - if err := tx.Commit(ctx); err != nil { - t.Fatalf("Failed to commit transaction: %v", err) - } - - t.Logf("Successfully created time trigger with ID: %s", trigger.PolicyID) - - // Test getting pending time triggers - triggers, err := backend.GetPendingTimeTriggers(ctx) - if err != nil { - t.Fatalf("Failed to get pending time triggers: %v", err) - } - - if len(triggers) == 0 { - t.Fatal("Expected at least one pending time trigger") - } - - t.Logf("Found %d pending time triggers", len(triggers)) - - // Test deleting the time trigger - if err := backend.DeleteTimeTrigger(ctx, trigger.PolicyID); err != nil { - t.Fatalf("Failed to delete time trigger: %v", err) - } - - t.Logf("Successfully deleted time trigger with ID: %s", trigger.PolicyID) -}