diff --git a/backend/configs/config.dev.yaml b/backend/configs/config.dev.yaml index 6875793a..e9c320e1 100644 --- a/backend/configs/config.dev.yaml +++ b/backend/configs/config.dev.yaml @@ -14,6 +14,10 @@ log: worker: concurrency: 5 + redis_stream_prefix: "gochat:jobs" + redis_consumer_group: "gochat-workers-dev" + redis_block_timeout_s: 5 + redis_sweep_interval_s: 15 # shorter in dev for faster delayed-job pickup redis: channel_prefix: "gochat_dev" diff --git a/backend/configs/config.prod.yaml b/backend/configs/config.prod.yaml index 5f4e6988..b1478b1b 100644 --- a/backend/configs/config.prod.yaml +++ b/backend/configs/config.prod.yaml @@ -24,6 +24,10 @@ log: worker: concurrency: 10 + redis_stream_prefix: "gochat:jobs" + redis_consumer_group: "gochat-workers" + redis_block_timeout_s: 5 + redis_sweep_interval_s: 30 redis: channel_prefix: "gochat_production" diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index 04dca83c..6081d540 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -106,15 +106,24 @@ func Bootstrap(env string) (*App, error) { applogger.L().Info("Database migrations completed successfully") } - workerPool := worker.NewWorkerPool(db) - automation.RegisterActionDeliveryJobs(workerPool, &dbProvider{db: db}) - // Step 5: Connect to Redis (ref: Chatwoot config/cable.yml) + // Redis must be initialized before the worker pool so it can be wired + // in as the job dispatch backend (XADD/XREADGROUP instead of DB polling). rdb, err := NewRedisClient(&cfg.Redis) if err != nil { return nil, fmt.Errorf("redis init failed: %w", err) } + workerPool := worker.NewWorkerPoolWithOptions(db, + worker.WithRedisClient(rdb), + worker.WithWorkerCount(cfg.Worker.Concurrency), + worker.WithStreamPrefix(cfg.Worker.StreamPrefix), + worker.WithConsumerGroup(cfg.Worker.ConsumerGroup), + worker.WithBlockTimeout(time.Duration(cfg.Worker.BlockTimeoutS)*time.Second), + worker.WithSweepInterval(time.Duration(cfg.Worker.SweepIntervalS)*time.Second), + ) + automation.RegisterActionDeliveryJobs(workerPool, &dbProvider{db: db}) + // Step 5b: Initialize PubSub — Redis-backed when Redis is available, in-memory fallback var ps pubsub.PubSub redisPubSub, redisErr := pubsub.NewRedisPubSub(rdb) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 6cf319bb..a1456f57 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -46,7 +46,11 @@ type Config struct { } type WorkerConfig struct { - Concurrency int `mapstructure:"concurrency"` + Concurrency int `mapstructure:"concurrency"` + StreamPrefix string `mapstructure:"redis_stream_prefix"` // default "gochat:jobs" + ConsumerGroup string `mapstructure:"redis_consumer_group"` // default "gochat-workers" + BlockTimeoutS int `mapstructure:"redis_block_timeout_s"` // default 5 + SweepIntervalS int `mapstructure:"redis_sweep_interval_s"` // default 30 } // SearchConfig controls the full-text search backend. Meilisearch is the @@ -491,6 +495,8 @@ var ReloadableFields = []string{ "captain.max_tokens", "captain.temperature", "worker.concurrency", + "worker.redis_block_timeout_s", + "worker.redis_sweep_interval_s", "server.cors.allowed_origins", } @@ -570,6 +576,9 @@ func (r *ConfigReloader) applyReloadableFields(newCfg *Config) { // Worker concurrency — safe to change at runtime r.cfg.Worker.Concurrency = newCfg.Worker.Concurrency + // Worker Redis sweep/block timing — safe to change at runtime + r.cfg.Worker.BlockTimeoutS = newCfg.Worker.BlockTimeoutS + r.cfg.Worker.SweepIntervalS = newCfg.Worker.SweepIntervalS // CORS allowed origins — safe to update whitelist at runtime r.cfg.Server.CORS = newCfg.Server.CORS @@ -644,7 +653,11 @@ func LoadWithEnv(env string) (*Config, error) { "GOCHAT_CAPTAIN_LLM_PROVIDER": "captain.llm_provider", "GOCHAT_CAPTAIN_LLM_MODEL": "captain.llm_model", "GOCHAT_CAPTAIN_LLM_API_KEY": "captain.llm_api_key", - "GOCHAT_WORKER_CONCURRENCY": "worker.concurrency", + "GOCHAT_WORKER_CONCURRENCY": "worker.concurrency", + "GOCHAT_WORKER_REDIS_STREAM_PREFIX": "worker.redis_stream_prefix", + "GOCHAT_WORKER_REDIS_CONSUMER_GROUP": "worker.redis_consumer_group", + "GOCHAT_WORKER_REDIS_BLOCK_TIMEOUT_S": "worker.redis_block_timeout_s", + "GOCHAT_WORKER_REDIS_SWEEP_INTERVAL_S": "worker.redis_sweep_interval_s", "GOCHAT_SEARCH_ENGINE": "search.engine", "GOCHAT_SEARCH_HOST": "search.host", "GOCHAT_SEARCH_API_KEY": "search.api_key", @@ -793,6 +806,10 @@ func setDefaults(v *viper.Viper) { v.SetDefault("search.timeout_seconds", 5) v.SetDefault("worker.concurrency", 4) + v.SetDefault("worker.redis_stream_prefix", "gochat:jobs") + v.SetDefault("worker.redis_consumer_group", "gochat-workers") + v.SetDefault("worker.redis_block_timeout_s", 5) + v.SetDefault("worker.redis_sweep_interval_s", 30) // CORS production defaults v.SetDefault("server.cors.allowed_methods", []string{"GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"}) @@ -844,6 +861,18 @@ func applyZeroDefaults(cfg *Config) { if cfg.Worker.Concurrency == 0 { cfg.Worker.Concurrency = 4 } + if cfg.Worker.StreamPrefix == "" { + cfg.Worker.StreamPrefix = "gochat:jobs" + } + if cfg.Worker.ConsumerGroup == "" { + cfg.Worker.ConsumerGroup = "gochat-workers" + } + if cfg.Worker.BlockTimeoutS == 0 { + cfg.Worker.BlockTimeoutS = 5 + } + if cfg.Worker.SweepIntervalS == 0 { + cfg.Worker.SweepIntervalS = 30 + } applySearchDefaults(&cfg.Search) // CSRF defaults if cfg.CSRF.CookieName == "" { diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index fe4e5264..30387e48 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -28,7 +28,7 @@ func TestValidate_ValidConfig(t *testing.T) { }, Log: LogConfig{Level: "info", Format: "json"}, Captain: CaptainConfig{Enabled: false}, - Worker: WorkerConfig{Concurrency: 4}, + Worker: WorkerConfig{Concurrency: 4, BlockTimeoutS: 5, SweepIntervalS: 30}, OAuth: OAuthConfig{}, RateLimit: RateLimitConfig{Enabled: true, RequestsPerMinute: 100, WindowSeconds: 60}, Search: SearchConfig{Engine: "meilisearch", Host: "http://localhost:7700", IndexPrefix: "gochat_", TimeoutSeconds: 5}, @@ -164,7 +164,7 @@ func TestValidate_SearchMeilisearchRequiresValidHost(t *testing.T) { Redis: RedisConfig{URL: "redis://localhost:6379"}, JWT: JWTConfig{Secret: "test-secret-key-min-32-chars!!"}, Log: LogConfig{Level: "info"}, - Worker: WorkerConfig{Concurrency: 1}, + Worker: WorkerConfig{Concurrency: 1, BlockTimeoutS: 5, SweepIntervalS: 30}, RateLimit: RateLimitConfig{RequestsPerMinute: 100, WindowSeconds: 60}, Search: SearchConfig{Engine: "meilisearch", Host: "not a url", TimeoutSeconds: 5}, } @@ -181,7 +181,7 @@ func TestValidate_SearchDBFallbackAllowed(t *testing.T) { Redis: RedisConfig{URL: "redis://localhost:6379"}, JWT: JWTConfig{Secret: "test-secret-key-min-32-chars!!"}, Log: LogConfig{Level: "info"}, - Worker: WorkerConfig{Concurrency: 1}, + Worker: WorkerConfig{Concurrency: 1, BlockTimeoutS: 5, SweepIntervalS: 30}, RateLimit: RateLimitConfig{RequestsPerMinute: 100, WindowSeconds: 60}, Search: SearchConfig{Engine: "db"}, } @@ -197,7 +197,7 @@ func TestValidate_SearchDBFallbackRejectedInRelease(t *testing.T) { Redis: RedisConfig{URL: "redis://localhost:6379"}, JWT: JWTConfig{Secret: "test-secret-key-min-32-chars!!"}, Log: LogConfig{Level: "info"}, - Worker: WorkerConfig{Concurrency: 1}, + Worker: WorkerConfig{Concurrency: 1, BlockTimeoutS: 5, SweepIntervalS: 30}, RateLimit: RateLimitConfig{RequestsPerMinute: 100, WindowSeconds: 60}, Search: SearchConfig{Engine: "db"}, } diff --git a/backend/internal/config/reloader_test.go b/backend/internal/config/reloader_test.go index 4740551c..a9d237b0 100644 --- a/backend/internal/config/reloader_test.go +++ b/backend/internal/config/reloader_test.go @@ -70,6 +70,8 @@ func TestConfigReloader_ReloadableFieldsList(t *testing.T) { "captain.max_tokens", "captain.temperature", "worker.concurrency", + "worker.redis_block_timeout_s", + "worker.redis_sweep_interval_s", "server.cors.allowed_origins", } assert.ElementsMatch(t, expectedFields, ReloadableFields) diff --git a/backend/internal/config/validator.go b/backend/internal/config/validator.go index 10b1d54c..f68a83cf 100644 --- a/backend/internal/config/validator.go +++ b/backend/internal/config/validator.go @@ -62,6 +62,12 @@ func Validate(cfg *Config) error { if cfg.Worker.Concurrency < 1 { return fmt.Errorf("worker concurrency must be >= 1") } + if cfg.Worker.BlockTimeoutS < 1 { + return fmt.Errorf("worker.redis_block_timeout_s must be >= 1") + } + if cfg.Worker.SweepIntervalS < 1 { + return fmt.Errorf("worker.redis_sweep_interval_s must be >= 1") + } // Rate limit validation if cfg.RateLimit.RequestsPerMinute < 1 { diff --git a/backend/internal/worker/worker.go b/backend/internal/worker/worker.go index 4194d57f..4d530b2b 100644 --- a/backend/internal/worker/worker.go +++ b/backend/internal/worker/worker.go @@ -6,10 +6,14 @@ import ( "errors" "fmt" "math" + "strconv" + "strings" "sync" "time" "github.com/gochat/gochat/internal/model" + applogger "github.com/gochat/gochat/pkg/logger" + "github.com/redis/go-redis/v9" "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -26,6 +30,7 @@ type BackoffFunc func(attempt int) time.Duration // execution, retry, and dead-letter visibility for async side effects. type WorkerPool struct { db *gorm.DB + rdb redis.UniversalClient handlers map[string]JobHandler queues []string workerID string @@ -35,6 +40,12 @@ type WorkerPool struct { backoff BackoffFunc now func() time.Time + // Redis stream queue parameters + streamPrefix string // default "gochat:jobs" + consumerGroup string // default "gochat-workers" + blockTimeout time.Duration // XREADGROUP block duration, default 5s + sweepInterval time.Duration // compensation sweep interval, default 30s + mu sync.RWMutex ctx context.Context cancel context.CancelFunc @@ -54,6 +65,10 @@ func NewWorkerPool(db ...*gorm.DB) *WorkerPool { staleLockTimeout: 15 * time.Minute, backoff: defaultBackoff, now: time.Now, + streamPrefix: "gochat:jobs", + consumerGroup: "gochat-workers", + blockTimeout: 5 * time.Second, + sweepInterval: 30 * time.Second, } if len(db) > 0 { wp.db = db[0] @@ -125,6 +140,51 @@ func WithNow(now func() time.Time) Option { } } +// WithRedisClient configures the WorkerPool to use Redis Streams for job +// dispatch instead of DB polling. When set, Enqueue pushes to Redis and +// workers consume via XREADGROUP. When nil, the pool falls back to DB polling. +func WithRedisClient(rdb redis.UniversalClient) Option { + return func(wp *WorkerPool) { + wp.rdb = rdb + } +} + +// WithStreamPrefix sets the Redis stream key prefix (default "gochat:jobs"). +func WithStreamPrefix(prefix string) Option { + return func(wp *WorkerPool) { + if prefix != "" { + wp.streamPrefix = prefix + } + } +} + +// WithConsumerGroup sets the Redis consumer group name (default "gochat-workers"). +func WithConsumerGroup(group string) Option { + return func(wp *WorkerPool) { + if group != "" { + wp.consumerGroup = group + } + } +} + +// WithBlockTimeout sets the XREADGROUP block duration (default 5s). +func WithBlockTimeout(d time.Duration) Option { + return func(wp *WorkerPool) { + if d > 0 { + wp.blockTimeout = d + } + } +} + +// WithSweepInterval sets the compensation sweep interval (default 30s). +func WithSweepInterval(d time.Duration) Option { + return func(wp *WorkerPool) { + if d > 0 { + wp.sweepInterval = d + } + } +} + func (wp *WorkerPool) Register(jobType string, handler JobHandler) { wp.mu.Lock() defer wp.mu.Unlock() @@ -220,6 +280,19 @@ func (wp *WorkerPool) Enqueue(ctx context.Context, jobType string, payload any, } return nil, err } + + // Push to Redis Stream for immediate dispatch if the job is due. + // Scheduled (future) jobs are picked up by the sweep goroutine when they mature. + if wp.rdb != nil && !job.ScheduledAt.After(wp.now()) { + if err := wp.pushToRedis(ctx, job); err != nil { + // Redis delivery failure does not roll back the DB-committed job. + // The sweep compensation mechanism will re-push it on the next cycle. + applogger.L().Warnf( + "redis XAdd failed for job %d (queue=%s), will be picked up by sweep: %v", + job.ID, job.Queue, err, + ) + } + } return job, nil } @@ -240,10 +313,23 @@ func (wp *WorkerPool) Start() error { if _, err := wp.RequeueStaleJobs(ctx); err != nil { return err } + + // Ensure Redis consumer groups exist for all configured queues. + if wp.rdb != nil { + wp.ensureConsumerGroups(ctx) + } + for i := 0; i < workerCount; i++ { wp.wg.Add(1) go wp.run(ctx) } + + // Start the sweep goroutine for delayed-job delivery and stale-job recovery. + if wp.rdb != nil { + wp.wg.Add(1) + go wp.sweepLoop(ctx) + } + return nil } @@ -290,8 +376,55 @@ func (wp *WorkerPool) RequeueStaleJobs(ctx context.Context) (int64, error) { return result.RowsAffected, result.Error } +// run is the per-goroutine consume loop. When Redis is configured it uses +// XREADGROUP BLOCK; otherwise it falls back to DB polling. func (wp *WorkerPool) run(ctx context.Context) { defer wp.wg.Done() + + if wp.rdb == nil { + wp.runDBPollLoop(ctx) + return + } + + // Build the XREADGROUP streams list: [streamKey1, ">", streamKey2, ">", ...] + streams := wp.streamKeys() + for { + if ctx.Err() != nil { + return + } + + results, err := wp.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + Group: wp.consumerGroup, + Consumer: wp.workerID, + Streams: streams, + Count: 1, + Block: wp.blockTimeout, + }).Result() + + if ctx.Err() != nil { + return + } + if err != nil && !errors.Is(err, redis.Nil) { + applogger.L().Errorf("XReadGroup error: %v", err) + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } + continue + } + + for _, xstream := range results { + for _, msg := range xstream.Messages { + wp.processRedisMessage(ctx, xstream.Stream, msg) + } + } + } +} + +// runDBPollLoop is the legacy DB-polling loop, used when Redis is unavailable +// (e.g. SQLite test mode) or not configured. +func (wp *WorkerPool) runDBPollLoop(ctx context.Context) { ticker := time.NewTicker(wp.pollInterval) defer ticker.Stop() for { @@ -307,6 +440,180 @@ func (wp *WorkerPool) run(ctx context.Context) { } } +// processRedisMessage loads the job from DB, claims it atomically, executes +// the handler, and acknowledges the Redis message. The DB claim check +// guarantees at-most-once execution even if the same job is XADD'd multiple +// times (e.g. by both Enqueue and sweep). +func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, msg redis.XMessage) { + jobIDStr, ok := msg.Values["job_id"] + if !ok { + wp.ackRedis(ctx, stream, msg.ID) + return + } + + jobID, err := strconv.ParseUint(fmt.Sprintf("%v", jobIDStr), 10, 64) + if err != nil { + wp.ackRedis(ctx, stream, msg.ID) + return + } + + var job model.BackgroundJob + if err := wp.db.WithContext(ctx).First(&job, jobID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + // Job was deleted; ACK and drop. + wp.ackRedis(ctx, stream, msg.ID) + return + } + // DB error — do not ACK so Redis can redeliver to another consumer. + applogger.L().Errorf("load job %d from DB failed: %v", jobID, err) + return + } + + // Claim check: atomically transition queued/retrying → running. + // RowsAffected == 0 means another consumer already claimed it or the + // job is not yet due; ACK to avoid redelivery loops. + now := wp.now() + result := wp.db.WithContext(ctx).Model(&model.BackgroundJob{}). + Where("id = ? AND status IN ? AND scheduled_at <= ?", + jobID, + []string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying}, + now, + ). + Updates(map[string]any{ + "status": model.BackgroundJobStatusRunning, + "locked_at": &now, + "locked_by": wp.workerID, + "attempts": gorm.Expr("attempts + 1"), + }) + if result.Error != nil { + applogger.L().Errorf("claim job %d failed: %v", jobID, result.Error) + return + } + if result.RowsAffected == 0 { + wp.ackRedis(ctx, stream, msg.ID) + return + } + + // Reload the job with updated attempts/locked_by fields. + wp.db.WithContext(ctx).First(&job, jobID) + + if err := wp.perform(ctx, &job); err != nil { + wp.fail(ctx, &job, err) + } + + wp.ackRedis(ctx, stream, msg.ID) +} + +func (wp *WorkerPool) ackRedis(ctx context.Context, stream, msgID string) { + if err := wp.rdb.XAck(ctx, stream, wp.consumerGroup, msgID).Err(); err != nil { + applogger.L().Warnf("XAck failed for stream=%s msgID=%s: %v", stream, msgID, err) + } +} + +// pushToRedis adds a job ID message to the Redis stream for the job's queue. +func (wp *WorkerPool) pushToRedis(ctx context.Context, job *model.BackgroundJob) error { + streamKey := wp.streamKeyFor(job.Queue) + return wp.rdb.XAdd(ctx, &redis.XAddArgs{ + Stream: streamKey, + Values: map[string]interface{}{ + "job_id": strconv.FormatUint(uint64(job.ID), 10), + "job_type": job.JobType, + "queue": job.Queue, + }, + }).Err() +} + +func (wp *WorkerPool) streamKeyFor(queue string) string { + return fmt.Sprintf("%s:%s", wp.streamPrefix, queue) +} + +// streamKeys returns the go-redis Streams slice: all stream keys first, then all IDs. +// The ">" sentinel tells Redis to deliver only messages never seen by this group. +func (wp *WorkerPool) streamKeys() []string { + queues := wp.allQueues() + keys := make([]string, 0, len(queues)*2) + for _, q := range queues { + keys = append(keys, wp.streamKeyFor(q)) + } + for range queues { + keys = append(keys, ">") + } + return keys +} + +// allQueues returns the queues this worker listens on. If none are explicitly +// configured, it defaults to the full set of known queues. +func (wp *WorkerPool) allQueues() []string { + if len(wp.queues) > 0 { + return wp.queues + } + return []string{ + "default", "high", "medium", "low", + "events", "automation", "search", + "scheduled_jobs", "deferred", "purgable", + } +} + +// ensureConsumerGroups creates the Redis consumer group for each queue's stream. +// XGROUP CREATE is idempotent: a BUSYGROUP reply means the group already exists. +func (wp *WorkerPool) ensureConsumerGroups(ctx context.Context) { + for _, queue := range wp.allQueues() { + streamKey := wp.streamKeyFor(queue) + err := wp.rdb.XGroupCreateMkStream(ctx, streamKey, wp.consumerGroup, "$").Err() + if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") { + applogger.L().Warnf("XGroupCreate for stream %s group %s: %v", + streamKey, wp.consumerGroup, err) + } + } +} + +// sweepLoop periodically scans the DB for due jobs that are still in +// queued/retrying state and re-pushes them to Redis. It also recovers stale +// running jobs. This compensates for Redis delivery failures and handles +// delayed (scheduled_at > now) jobs that were not XADD'd at enqueue time. +func (wp *WorkerPool) sweepLoop(ctx context.Context) { + defer wp.wg.Done() + ticker := time.NewTicker(wp.sweepInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := wp.sweepDueJobs(ctx); err != nil { + applogger.L().Warnf("sweep due jobs failed: %v", err) + } + if _, err := wp.RequeueStaleJobs(ctx); err != nil { + applogger.L().Warnf("requeue stale jobs failed: %v", err) + } + } + } +} + +func (wp *WorkerPool) sweepDueJobs(ctx context.Context) error { + now := wp.now() + var jobs []model.BackgroundJob + if err := wp.db.WithContext(ctx). + Where("status IN ? AND scheduled_at <= ?", + []string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying}, + now, + ). + Limit(100). + Find(&jobs).Error; err != nil { + return err + } + + for i := range jobs { + // Idempotent push: duplicate XADD does not cause duplicate execution + // because processRedisMessage has a DB claim check. + if err := wp.pushToRedis(ctx, &jobs[i]); err != nil { + applogger.L().Warnf("sweep: XAdd failed for job %d: %v", jobs[i].ID, err) + } + } + return nil +} + func (wp *WorkerPool) claimNext(ctx context.Context) (*model.BackgroundJob, error) { var job model.BackgroundJob err := wp.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { diff --git a/backend/internal/worker/worker_test.go b/backend/internal/worker/worker_test.go index 4c97c110..e1f14e68 100644 --- a/backend/internal/worker/worker_test.go +++ b/backend/internal/worker/worker_test.go @@ -8,7 +8,9 @@ import ( "testing" "time" + "github.com/alicebob/miniredis/v2" "github.com/gochat/gochat/internal/model" + "github.com/redis/go-redis/v9" "gorm.io/driver/sqlite" "gorm.io/gorm" "gorm.io/gorm/logger" @@ -219,3 +221,330 @@ func TestWorkerPoolStartAndStopProcessJobs(t *testing.T) { t.Fatalf("expected completed job after worker loop: %+v", reloaded) } } + +// --- Redis Stream path tests --- + +func newMiniRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client) { + t.Helper() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { rdb.Close() }) + return mr, rdb +} + +func newRedisWorkerPool(t *testing.T, db *gorm.DB, rdb redis.UniversalClient, opts ...Option) *WorkerPool { + t.Helper() + defaults := []Option{ + WithRedisClient(rdb), + WithNow(func() time.Time { return time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) }), + WithBlockTimeout(200 * time.Millisecond), + WithSweepInterval(100 * time.Millisecond), + WithWorkerID("test-redis-worker"), + } + wp := NewWorkerPoolWithOptions(db, append(defaults, opts...)...) + return wp +} + +// TestRedisEnqueueAndProcessEndToEnd verifies the full Redis path: Enqueue +// XADDs to the stream, XREADGROUP picks it up, DB claim succeeds, handler +// runs, and the job reaches "completed" status. +func TestRedisEnqueueAndProcessEndToEnd(t *testing.T) { + db := newWorkerTestDB(t) + _, rdb := newMiniRedis(t) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + wp := newRedisWorkerPool(t, db, rdb, WithNow(func() time.Time { return now })) + + var handled atomic.Int32 + wp.Register("test_job", func(ctx context.Context, job *model.BackgroundJob) error { + handled.Add(1) + if job.LockedBy != "test-redis-worker" { + t.Fatalf("expected locked_by=test-redis-worker, got %s", job.LockedBy) + } + return nil + }) + + if err := wp.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer wp.Stop() + + job, err := wp.Enqueue(context.Background(), "test_job", map[string]any{"k": "v"}) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if handled.Load() == 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + if handled.Load() != 1 { + t.Fatalf("handler was not invoked via Redis path") + } + + reloaded := loadJob(t, db, job.ID) + if reloaded.Status != model.BackgroundJobStatusCompleted { + t.Fatalf("expected completed, got %s", reloaded.Status) + } + if reloaded.LockedAt != nil || reloaded.LockedBy != "" { + t.Fatalf("expected cleared lock, got locked_by=%s locked_at=%v", reloaded.LockedBy, reloaded.LockedAt) + } +} + +// TestRedisFallbackToDBPolling verifies that when rdb is nil the pool falls +// back to DB polling and existing ProcessOne path works. +func TestRedisFallbackToDBPolling(t *testing.T) { + db := newWorkerTestDB(t) + wp := NewWorkerPoolWithOptions(db, + WithNow(func() time.Time { return time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) }), + WithPollInterval(5*time.Millisecond), + ) + // rdb is nil → DB polling fallback + + var handled atomic.Int32 + wp.Register("fallback_job", func(ctx context.Context, job *model.BackgroundJob) error { + handled.Add(1) + return nil + }) + + if err := wp.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer wp.Stop() + + if _, err := wp.Enqueue(context.Background(), "fallback_job", nil); err != nil { + t.Fatalf("enqueue: %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if handled.Load() == 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + if handled.Load() != 1 { + t.Fatalf("fallback DB polling did not process job") + } +} + +// TestRedisSweepPicksUpDueDelayedJob verifies that a job enqueued with a +// future scheduled_at is not XADD'd immediately, but the sweep goroutine +// pushes it to Redis once it matures. +func TestRedisSweepPicksUpDueDelayedJob(t *testing.T) { + db := newWorkerTestDB(t) + mr, rdb := newMiniRedis(t) + baseTime := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + muNow := baseTime + wp := newRedisWorkerPool(t, db, rdb, + WithNow(func() time.Time { return muNow }), + WithSweepInterval(50*time.Millisecond), + WithBlockTimeout(50*time.Millisecond), + ) + + var handled atomic.Int32 + wp.Register("delayed_job", func(ctx context.Context, job *model.BackgroundJob) error { + handled.Add(1) + return nil + }) + + if err := wp.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer wp.Stop() + + // Enqueue a job scheduled 200ms in the future. + _, err := wp.Enqueue(context.Background(), "delayed_job", nil, WithScheduledAt(baseTime.Add(200*time.Millisecond))) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + // At this point the stream should be empty (no immediate XADD for future jobs). + // Verify no messages in the default stream yet. + streamKey := "gochat:jobs:default" + streamLen := func() int { entries, _ := mr.Stream(streamKey); return len(entries) }() + if streamLen != 0 { + t.Fatalf("expected 0 messages in stream before due time, got %d", streamLen) + } + + // Advance the mock clock past the scheduled_at so the job becomes due. + muNow = baseTime.Add(1 * time.Second) + + // The sweep goroutine should now see the job as due, push it to Redis, + // and the worker should process it. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if handled.Load() == 1 { + break + } + time.Sleep(20 * time.Millisecond) + } + if handled.Load() != 1 { + t.Fatalf("sweep did not pick up delayed job") + } +} + +// TestRedisMultiConsumerCompetition verifies that when two WorkerPool +// instances share the same consumer group, a job is only executed once. +func TestRedisMultiConsumerCompetition(t *testing.T) { + db := newWorkerTestDB(t) + _, rdb := newMiniRedis(t) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + + var handled atomic.Int32 + handler := func(ctx context.Context, job *model.BackgroundJob) error { + handled.Add(1) + time.Sleep(50 * time.Millisecond) // simulate work to widen the race window + return nil + } + + wp1 := newRedisWorkerPool(t, db, rdb, WithNow(func() time.Time { return now }), WithWorkerID("worker-1")) + wp1.Register("compete_job", handler) + wp2 := newRedisWorkerPool(t, db, rdb, WithNow(func() time.Time { return now }), WithWorkerID("worker-2")) + wp2.Register("compete_job", handler) + + if err := wp1.Start(); err != nil { + t.Fatalf("start wp1: %v", err) + } + defer wp1.Stop() + if err := wp2.Start(); err != nil { + t.Fatalf("start wp2: %v", err) + } + defer wp2.Stop() + + // Enqueue 5 jobs, each gets XADD'd; both workers compete for them. + for i := 0; i < 5; i++ { + if _, err := wp1.Enqueue(context.Background(), "compete_job", map[string]any{"i": i}); err != nil { + t.Fatalf("enqueue %d: %v", i, err) + } + } + + // Wait for all 5 jobs to be handled. + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if handled.Load() == 5 { + break + } + time.Sleep(20 * time.Millisecond) + } + if handled.Load() != 5 { + t.Fatalf("expected 5 jobs handled exactly once, got %d", handled.Load()) + } + + // Verify no job was double-processed: all should be completed, none running/retrying. + // Poll until all DB records reach completed state (handlers may still be finishing). + var stuck int64 + checkDeadline := time.Now().Add(5 * time.Second) + for time.Now().Before(checkDeadline) { + db.Model(&model.BackgroundJob{}).Where("status != ?", model.BackgroundJobStatusCompleted).Count(&stuck) + if stuck == 0 { + break + } + time.Sleep(50 * time.Millisecond) + } + if stuck != 0 { + t.Fatalf("expected all jobs completed, found %d in non-completed state", stuck) + } +} + +// TestRedisConsumerGroupCreationIdempotent verifies that calling Start() +// multiple times (or multiple pools) does not error on XGROUP CREATE. +func TestRedisConsumerGroupCreationIdempotent(t *testing.T) { + db := newWorkerTestDB(t) + _, rdb := newMiniRedis(t) + + wp := newRedisWorkerPool(t, db, rdb) + wp.Register("noop", func(ctx context.Context, job *model.BackgroundJob) error { return nil }) + + // ensureConsumerGroups should succeed and create groups. + ctx := context.Background() + wp.ensureConsumerGroups(ctx) + + // Calling again should be idempotent (BUSYGROUP is silently ignored). + wp.ensureConsumerGroups(ctx) + + // Verify groups exist on each stream. + for _, queue := range wp.allQueues() { + streamKey := wp.streamKeyFor(queue) + groups, err := rdb.XInfoGroups(ctx, streamKey).Result() + if err != nil { + t.Fatalf("XInfoGroups for %s: %v", streamKey, err) + } + found := false + for _, g := range groups { + if g.Name == wp.consumerGroup { + found = true + break + } + } + if !found { + t.Fatalf("consumer group %s not found on stream %s", wp.consumerGroup, streamKey) + } + } +} + +// TestRedisEnqueuePushFailureCompensatedBySweep verifies that if XADD fails +// during Enqueue, the job is still delivered via the sweep mechanism. +func TestRedisEnqueuePushFailureCompensatedBySweep(t *testing.T) { + db := newWorkerTestDB(t) + mr, rdb := newMiniRedis(t) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + + var handled atomic.Int32 + wp := newRedisWorkerPool(t, db, rdb, + WithNow(func() time.Time { return now }), + WithSweepInterval(50*time.Millisecond), + WithBlockTimeout(50*time.Millisecond), + ) + wp.Register("sweep_recovery", func(ctx context.Context, job *model.BackgroundJob) error { + handled.Add(1) + return nil + }) + + if err := wp.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer wp.Stop() + + // Close miniredis to simulate Redis being down during Enqueue. + mr.Close() + + job, err := wp.Enqueue(context.Background(), "sweep_recovery", nil) + if err != nil { + t.Fatalf("enqueue should still succeed (DB committed): %v", err) + } + // XADD should have failed silently; job is in DB as queued. + reloaded := loadJob(t, db, job.ID) + if reloaded.Status != model.BackgroundJobStatusQueued { + t.Fatalf("expected queued status after failed XADD, got %s", reloaded.Status) + } + + // Restart miniredis at the same address won't work (port already freed). + // Instead, we manually push to Redis to simulate sweep recovery once Redis is back. + // Use a fresh miniredis on a new port + new client. + mr2 := miniredis.RunT(t) + rdb2 := redis.NewClient(&redis.Options{Addr: mr2.Addr()}) + t.Cleanup(func() { rdb2.Close() }) + + // Swap in the recovered Redis client. + wp.mu.Lock() + wp.rdb = rdb2 + wp.mu.Unlock() + + // Recreate consumer groups on the new Redis instance. + wp.ensureConsumerGroups(context.Background()) + + // Wait for sweep to pick up the job and push it, then worker to process. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if handled.Load() == 1 { + break + } + time.Sleep(20 * time.Millisecond) + } + if handled.Load() != 1 { + t.Fatalf("sweep did not recover job after Redis reconnection") + } +}