feat: migrate worker job dispatch from DB polling to Redis Streams

- worker.go: add Redis fields, XADD on Enqueue, XREADGROUP BLOCK consumer
  loop, sweep compensation for delayed/failed jobs, DB-polling fallback
  when rdb is nil, XGroupCreateMkStream for idempotent consumer groups
- config.go: extend WorkerConfig with stream_prefix, consumer_group,
  block_timeout_s, sweep_interval_s + defaults, env bindings, reloadable
- validator.go: validate new worker Redis fields
- bootstrap.go: fix concurrency bug (NewWorkerPoolWithOptions with Redis
  + cfg.Worker.Concurrency), reorder rdb init before worker pool
- config.dev.yaml/config.prod.yaml: add worker Redis params
- worker_test.go: 6 new miniredis tests (end-to-end, fallback, sweep,
  multi-consumer competition, group creation, Redis failure recovery)
- config_test.go/reloader_test.go: adapt fixtures for new fields

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"

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"

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)

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 == "" {

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"},
 	}

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)

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 {

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 {

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")
+	}
+}