config.go

package senna

import (
	"crypto/tls"
	"time"

	"github.com/redis/go-redis/v9"
)

// Config configures shared Senna settings used by clients and workers.
type Config struct {
	Redis      RedisConfig
	Namespace  string
	Worker     WorkerSettings
	Client     ClientSettings
	Encryption EncryptionSettings
}

// RedisConfig configures the Redis connection used by Senna.
type RedisConfig struct {
	Addr         string
	Password     string
	DB           int
	PoolSize     int
	MinIdleConns int
	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	TLSConfig    *tls.Config
}

// Options converts the RedisConfig into go-redis client options.
func (c RedisConfig) Options() *redis.Options {
	opts := &redis.Options{
		Addr:                  c.Addr,
		Password:              c.Password,
		DB:                    c.DB,
		TLSConfig:             c.TLSConfig,
		ContextTimeoutEnabled: true,
	}
	if c.PoolSize > 0 {
		opts.PoolSize = c.PoolSize
	}
	if c.MinIdleConns > 0 {
		opts.MinIdleConns = c.MinIdleConns
	}
	if c.DialTimeout > 0 {
		opts.DialTimeout = c.DialTimeout
	}
	if c.ReadTimeout > 0 {
		opts.ReadTimeout = c.ReadTimeout
	}
	if c.WriteTimeout > 0 {
		opts.WriteTimeout = c.WriteTimeout
	}
	return opts
}

// WorkerSettings configures worker runtime behavior.
type WorkerSettings struct {
	Concurrency                 int
	Queues                      []QueueConfig
	ShutdownTimeout             time.Duration
	PollInterval                time.Duration
	BlockTimeout                time.Duration
	ScheduledPollInterval       time.Duration
	HeartbeatRate               time.Duration
	ReaperOperationTimeout      time.Duration
	ReaperInterval              time.Duration
	SequentialLockTTL           time.Duration
	SequentialLockRenewInterval time.Duration
	IterableMaxRuntime          time.Duration
	PeriodicPollInterval        time.Duration
	PeriodicEnabled             bool
	StrictPriority              bool // If true, always process higher priority queues first (can starve lower priority)
}

const (
	// DefaultQueueName is the default queue used by clients and workers.
	DefaultQueueName = "default"
	// DefaultRetryCount is the default retry count assigned to new jobs.
	DefaultRetryCount = 25
	// DefaultBulkChunkSize bounds bulk enqueue Redis command size by default.
	DefaultBulkChunkSize = 1000
)

// DefaultWorkerSettings returns the default worker settings.
func DefaultWorkerSettings() WorkerSettings {
	return WorkerSettings{
		Concurrency:                 10,
		Queues:                      []QueueConfig{{Name: DefaultQueueName, Priority: 1}},
		ShutdownTimeout:             30 * time.Second,
		PollInterval:                100 * time.Millisecond,
		BlockTimeout:                2 * time.Second,
		ScheduledPollInterval:       5 * time.Second,
		HeartbeatRate:               5 * time.Second,
		ReaperOperationTimeout:      5 * time.Second,
		ReaperInterval:              30 * time.Second,
		SequentialLockTTL:           30 * time.Second,
		SequentialLockRenewInterval: 10 * time.Second,
		IterableMaxRuntime:          30 * time.Second,
		PeriodicPollInterval:        15 * time.Second,
	}
}

// QueueConfig configures a worker queue and its scheduling behavior.
type QueueConfig struct {
	Name       string
	Priority   int
	Paused     bool
	Sequential bool // Only one worker globally processes this queue at a time
}

// ClientSettings configures enqueue defaults for a client.
type ClientSettings struct {
	DefaultQueue  string
	DefaultRetry  int
	BulkChunkSize int
}

// DefaultClientSettings returns the default client settings.
func DefaultClientSettings() ClientSettings {
	return ClientSettings{
		DefaultQueue:  DefaultQueueName,
		DefaultRetry:  DefaultRetryCount,
		BulkChunkSize: DefaultBulkChunkSize,
	}
}

// EncryptionSettings configures job argument encryption.
type EncryptionSettings struct {
	Enabled bool
	Key     []byte
}

// Option mutates a Config.
type Option func(*Config)

// WithNamespace sets the Redis key namespace.
func WithNamespace(ns string) Option {
	return func(c *Config) {
		c.Namespace = ns
	}
}

// WithConcurrency sets the worker concurrency.
func WithConcurrency(n int) Option {
	return func(c *Config) {
		c.Worker.Concurrency = n
	}
}

// WithQueues sets the worker queue configuration.
func WithQueues(queues ...QueueConfig) Option {
	return func(c *Config) {
		c.Worker.Queues = queues
	}
}

// WithShutdownTimeout sets the worker shutdown timeout.
func WithShutdownTimeout(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.ShutdownTimeout = d
	}
}

// WithReaperOperationTimeout sets the per-Redis-command timeout for orphan recovery.
func WithReaperOperationTimeout(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.ReaperOperationTimeout = d
	}
}

// WithReaperInterval sets how often workers scan for orphaned in-flight jobs.
func WithReaperInterval(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.ReaperInterval = d
	}
}

// WithSequentialLockTTL sets how long a sequential queue lock remains valid.
func WithSequentialLockTTL(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.SequentialLockTTL = d
	}
}

// WithSequentialLockRenewInterval sets how often workers renew held sequential queue locks.
func WithSequentialLockRenewInterval(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.SequentialLockRenewInterval = d
	}
}

// WithIterableMaxRuntime sets how long an iterable job can run before yielding.
func WithIterableMaxRuntime(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.IterableMaxRuntime = d
	}
}

// WithPeriodicPollInterval sets how often workers check periodic job schedules.
func WithPeriodicPollInterval(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.PeriodicPollInterval = d
	}
}

// WithBlockTimeout sets how long a worker can block in Redis waiting for jobs.
func WithBlockTimeout(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.BlockTimeout = d
	}
}

// WithEncryptionConfig enables encryption with the provided key.
func WithEncryptionConfig(key []byte) Option {
	return func(c *Config) {
		c.Encryption.Enabled = true
		c.Encryption.Key = key
	}
}

// WithPeriodicEnabled enables periodic job scheduling for workers.
func WithPeriodicEnabled() Option {
	return func(c *Config) {
		c.Worker.PeriodicEnabled = true
	}
}

// WithScheduledPollInterval sets how often a worker checks scheduled jobs.
func WithScheduledPollInterval(d time.Duration) Option {
	return func(c *Config) {
		c.Worker.ScheduledPollInterval = d
	}
}

// WithStrictPriority makes workers always prefer higher-priority queues.
func WithStrictPriority() Option {
	return func(c *Config) {
		c.Worker.StrictPriority = true
	}
}