# Worker 队列迁移变更文档：DB 轮询 → Redis Stream 分发

> 状态：草案
> 日期：2026-07-10
> 范围：`internal/worker/`、`internal/app/`、`internal/config/`、部署配置
> 原则：Redis 负责消息分发（XADD/XREADGROUP），DB 只负责 job 元数据落地与审计追踪

---

## 1. 背景与动机

### 1.1 当前架构

Worker 子系统由 `internal/worker/worker.go` 的 `WorkerPool` 实现，核心行为：

- **入队**：`Enqueue()` 将 `BackgroundJob` 记录 INSERT 到 PostgreSQL `background_jobs` 表
- **消费**：每个 worker goroutine 每 500ms 轮询 DB，`SELECT ... FOR UPDATE SKIP LOCKED` claim 最早到期的 job
- **执行**：`perform()` 调用注册的 `JobHandler`，完成后 UPDATE 状态为 `completed`/`retrying`/`dead`
- **补偿**：`RequeueStaleJobs()` 扫描超过 `staleLockTimeout`（默认 15 分钟）的 `running` job，重置为 `retrying`

Redis 在项目中已用于三种场景，但 **均不涉及 job 队列**：

| 角色 | 实现 | 代码位置 |
|------|------|----------|
| WS 跨实例广播 | Redis Pub/Sub `PSubscribe` | `internal/ws/broadcast.go` |
| 事件总线 | Watermill Redis Streams (consumer group) | `internal/pubsub/redis_pubsub.go` |
| 通知投递 | Watermill Router + Redis Streams | `internal/service/notification_delivery_service.go` |

### 1.2 当前架构的问题

1. **轮询开销**：500ms 固定轮询，空闲时持续消耗 DB 连接和 CPU
2. **延迟下限**：job 从入队到执行至少 0–500ms 延迟
3. **DB 连接占用**：每个 worker goroutine 持续占用一个 DB 连接做轮询
4. **concurrency 配置未生效**：`Bootstrap()` 使用 `NewWorkerPool(db)` 而非 `NewWorkerPoolWithOptions(db, WithWorkerCount(n))`，实际只有 1 个 goroutine
5. **基础设施不统一**：Redis 已有 watermill-redisstream + go-redis，但 job 队列绕过 Redis 走 DB 轮询

### 1.3 迁移目标

```
Enqueue():
  1. INSERT into background_jobs (status=queued)     ← DB 落地，拿到 job ID
  2. XADD job stream {job_id, job_type, queue}       ← Redis 分发

Worker:
  1. XREADGROUP BLOCK → 拿到 job_id                   ← Redis 阻塞拉取（零轮询）
  2. SELECT * FROM background_jobs WHERE id=?         ← DB 加载 payload + 元数据
  3. UPDATE status=running, locked_by=...             ← DB 标记执行中
  4. handler(ctx, job)                                 ← 执行
  5. UPDATE status=completed/retrying/dead            ← DB 更新最终状态
  6. XACK                                              ← Redis 确认

Sweep (补偿):
  每 30s 扫描 DB 中 status=queued/retrying 且 scheduled_at <= now 的 job
  重新 XADD，兜底 Redis 投递失败的情况
```

---

## 2. 影响面分析

### 2.1 Job Type 全量清单（37 个）

迁移不改变任何 job type 定义，仅改变 `Enqueue` 和 `run` 的内部实现。

| Job Type | Queue | 调用方 | 是否延迟 | 幂等键 |
|----------|-------|--------|----------|--------|
| `message:send_reply` | `high` | `service.EnqueueSendReply` | 否 | `message:send_reply:{msgID}` |
| `event:dispatch_async` | `events` | `channel.Dispatcher.DispatchAsync` | 否 | 无 |
| `event:listener_dispatch` | `events` | `dispatch.EventDispatcher` | 否 | 无 |
| `automation:team_email_delivery` | `automation` | `automation.ActionService` | 否 | 无 |
| `automation:transcript_delivery` | `automation` | `automation.ActionService` | 否 | 无 |
| `automation:webhook_delivery` | `automation` | `automation.ActionService` | 否 | 无 |
| `automation:macro_execution` | `medium` | `automation.MacroService` | 否 | 无 |
| `csat:survey_send` | `automation` | `automation.CsatSurveyListener` | 否 | `csat-survey:{convID}` |
| `csat:template_create` | `automation` | `service.CsatTemplateService` | 否 | `csat-template:{tplID}:{ts}` |
| `scheduled:trigger_items` | `scheduled_jobs` | `service.EnqueueScheduledItemsTrigger` | **是**（1h 周期） | `scheduled:trigger_items:{bucket}` |
| `campaign:trigger_oneoff` | `low` | `conversationMaintenanceRunner` | 否 | `campaign:trigger_oneoff:{id}` |
| `conversation:reopen_snoozed` | `low` | `conversationMaintenanceRunner` | 否 | 无 |
| `account:conversations_resolution_scheduler` | `scheduled_jobs` | `conversationMaintenanceRunner` | 否 | 无 |
| `conversation:resolution` | `low` | `conversationMaintenanceRunner` | 否 | 无 |
| `conversation:update_message_status` | `deferred` | `service.EnqueueConversationMessageStatusUpdate` | **是**（可选） | `conversation:update_message_status:{convID}:{status}:{ts}` |
| `conversation:bulk_action` | `medium` | `service.EnqueueConversationBulkAction` | 否 | 无 |
| `contact:bulk_action` | `medium` | `service.EnqueueContactBulkAction` | 否 | 无 |
| `conversation:delete_object` | `low` | `service.ConversationService` | 否 | `conversation-delete:{acctID}:{convID}` |
| `contact:import` | `low` | `service.ContactService` | 否 | `contact-import:{id}` |
| `contact:export` | `low` | `service.ContactService` | 否 | `contact-export:{id}` |
| `search:index` | `search` | `service.DurableSearchIndexer` | 否 | 无 |
| `sla:trigger_accounts` | `scheduled_jobs` | `service.EnqueueSlaAccountsScan` | **是**（5min 周期） | `sla:trigger_accounts:{bucket}` |
| `sla:process_account` | `medium` | `slaProcessingJobRunner` | 否 | 无 |
| `sla:process_applied` | `medium` | `slaProcessingJobRunner` | 否 | 无 |
| `captain:copilot_response` | `default` | `service.CopilotService` | 否 | `captain:copilot_response:{msgID}` |
| `captain:conversation_response_builder` | `default` | `service.CaptainConversationService` | **是**（可选，附件等待） | `captain:conversation_response_builder:message:{msgID}` |
| `captain:document_sync` | `low` | `service.CaptainDocumentService` | 否 | `captain:document_sync:auto:...`（自动同步时） |
| `captain:document_crawl` | `low` | `service.CaptainDocumentService` | 否 | `captain:document_crawl:{acctID}:{id}` |
| `captain:document_page_crawl_parse` | `low` | `service.CaptainDocumentService` | 否 | `captain:document_page_crawl_parse:...` |
| `captain:document_response_builder` | `low` | `service.CaptainDocumentService` | 否 | 无 |
| `captain:documents_schedule_syncs` | `scheduled_jobs` | `service.EnqueueCaptainDocumentScheduleSyncs` | **是**（24h 周期） | `captain:documents_schedule_syncs:{bucket}` |
| `captain:llm_update_embedding` | `low` | `service.CaptainDocumentService` | 否 | 无 |
| `captain:article_translate` | `low` | `service.ArticleService` | 否 | `captain:article_translate:...` |
| `inbox:sync_templates` | `low` | `service.InboxService` | 否 | 无 |
| `webhook:incoming_message_persist` | `default`/`low` | `webhook.IncomingPersister` | 否 | `webhook:incoming_message:{inboxID}:{sourceID}` |
| `webhook:message_status_update` | `low` | `webhook.IncomingPersister` | 否 | `webhook:message_status:...` |
| `webhook:contact_messages_status_update` | `low` | `webhook.IncomingPersister` | 否 | `webhook:contact_messages_status:...` |
| `webhook:webwidget_triggered` | `default` | `service.WidgetService` | 否 | 无 |
| `reporting:rollup_day` | `low` | `service.EnqueueReportingRollupDay` | 否 | `reporting:rollup_day:{acctID}:{date}` |

### 2.2 Queue 名称汇总（8 个）

| Queue | 用途 | 延迟 job | 优先级特征 |
|-------|------|----------|------------|
| `default` | 通用 | 否 | 混杂 |
| `high` | 消息发送回复 | 否 | 最高优先 |
| `medium` | 批量操作、SLA 处理 | 否 | 中等 |
| `low` | 文档同步、爬取、导入导出、删除 | 否 | 可延迟 |
| `events` | 事件分发 | 否 | 需低延迟 |
| `automation` | 自动化规则执行 | 否 | 需及时 |
| `search` | 搜索索引 | 否 | 可延迟 |
| `scheduled_jobs` | 定时周期任务 | **是** | 按时间触发 |
| `deferred` | 延迟任务 | **是** | 按时间触发 |
| `purgable` | 可清理 | 否 | 低优先级 |

迁移后每个 queue 对应一个 Redis Stream key：`gochat:jobs:{queue}`。

### 2.3 延迟 Job（scheduled_at）的使用场景

5 个 job type 使用 `WithScheduledAt`，需要特殊处理：

1. **`scheduled:trigger_items`**：1 小时周期，自我重新入队
2. **`sla:trigger_accounts`**：5 分钟周期，自我重新入队
3. **`captain:documents_schedule_syncs`**：24 小时周期，自我重新入队
4. **`conversation:update_message_status`**：可选延迟
5. **`captain:conversation_response_builder`**：可选延迟（等待附件上传完成，1–5s）

延迟 job 在 Redis Stream 中不能直接投递（Stream 无原生延迟语义），需要延迟投递机制（见 §3.4）。

### 2.4 Priority 使用情况

`WithPriority` 已实现但**生产代码中无任何调用方**。仅 `worker_test.go` 测试中使用了 `WithPriority(10)`。

迁移后 priority 语义改为 queue 级别优先级（worker 先读 high queue 再读 low），不再需要 DB `ORDER BY priority DESC`。`BackgroundJob.Priority` 字段保留供 DB 审计，但不再影响消费顺序。

### 2.5 Idempotency Key 使用情况

19 个 job type 使用了 `WithIdempotencyKey`，依赖 DB 唯一索引 `idx_background_jobs_idempotency_key_unique` 做去重。迁移后 idempotency 仍在 DB 层保证，Redis 不参与去重。

---

## 3. 详细设计

### 3.1 WorkerPool 结构体变更

**文件**：`internal/worker/worker.go`

```go
type WorkerPool struct {
    db               *gorm.DB
    rdb              redis.UniversalClient   // 新增
    handlers         map[string]JobHandler
    queues           []string
    workerID         string
    workerCount      int
    pollInterval     time.Duration           // 保留，仅 fallback 模式使用
    staleLockTimeout time.Duration
    backoff          BackoffFunc
    now              func() time.Time

    // 新增：Redis 队列参数
    streamPrefix     string                  // 默认 "gochat:jobs"
    consumerGroup    string                  // 默认 "gochat-workers"
    blockTimeout     time.Duration           // XREADGROUP block 时长，默认 5s
    sweepInterval    time.Duration           // 补偿扫描间隔，默认 30s

    mu     sync.RWMutex
    ctx    context.Context
    cancel context.CancelFunc
    wg     sync.WaitGroup
}
```

新增 Option 函数：

```go
func WithRedisClient(rdb redis.UniversalClient) Option
func WithStreamPrefix(prefix string) Option
func WithConsumerGroup(group string) Option
func WithBlockTimeout(d time.Duration) Option
func WithSweepInterval(d time.Duration) Option
```

### 3.2 Enqueue 变更

**当前流程**：
1. 构建 `BackgroundJob` 结构
2. 处理 idempotency key 去重（DB 查询）
3. `db.Create(job)` 插入 DB
4. 返回 job

**迁移后流程**：
1. 构建 `BackgroundJob` 结构（不变）
2. 处理 idempotency key 去重（DB 查询，不变）
3. `db.Create(job)` 插入 DB（不变）
4. **新增**：如果 `scheduled_at <= now`，`XADD` 到 Redis Stream
5. **新增**：如果 `scheduled_at > now`，跳过 XADD（由 sweep 机制在到期时投递）
6. 返回 job（不变）

```go
func (wp *WorkerPool) Enqueue(ctx context.Context, jobType string, payload any, opts ...EnqueueOption) (*model.BackgroundJob, error) {
    // ... 现有逻辑：构建 job、idempotency、DB 插入（完全不变）...

    // 新增：DB 插入成功后投递到 Redis
    if wp.rdb != nil && !job.ScheduledAt.After(wp.now()) {
        if err := wp.pushToRedis(ctx, job); err != nil {
            // Redis 投递失败不影响 DB 已提交的 job
            // sweep 补偿机制会重新投递
            applogger.L().Warnf(
                "redis XAdd failed for job %d (queue=%s), will be picked up by sweep: %v",
                job.ID, job.Queue, err,
            )
        }
    }
    // scheduled_at > now 的 job 不立即投递
    // sweep goroutine 会定期扫描到期 job 并投递
    return job, nil
}

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":   fmt.Sprintf("%d", job.ID),
            "job_type": job.JobType,
            "queue":    job.Queue,
        },
    }).Err()
}

func (wp *WorkerPool) streamKeyFor(queue string) string {
    return fmt.Sprintf("%s:%s", wp.streamPrefix, queue)
}
```

**影响分析**：
- 所有 `Enqueue*` 辅助函数（`EnqueueSendReply`、`EnqueueScheduledItemsTrigger` 等）**不需要改动**，它们调用 `wp.Enqueue()` 的签名和返回值不变
- `SetWorkerPool` 调用方**不需要改动**
- idempotency 逻辑完全不变，仍在 DB 层

### 3.3 Worker 消费循环变更

**当前**：`run()` 每 500ms 调用 `ProcessOne()` → `claimNext()` → DB `SELECT ... FOR UPDATE SKIP LOCKED`

**迁移后**：`run()` 使用 `XREADGROUP BLOCK` 阻塞拉取

```go
func (wp *WorkerPool) run(ctx context.Context) {
    defer wp.wg.Done()

    if wp.rdb == nil {
        // fallback：Redis 不可用时回退到 DB 轮询模式
        wp.runDBPollLoop(ctx)
        return
    }

    streams := wp.streamKeys() // 监听所有已配置 queue 的 stream
    for {
        if ctx.Err() != nil {
            return
        }

        // XREADGROUP BLOCK
        results, err := wp.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    wp.consumerGroup,
            Consumer: wp.workerID,
            Streams:  streams,
            Count:    1,
            Block:    wp.blockTimeout, // 5s
        }).Result()

        if ctx.Err() != nil {
            return
        }
        if err != nil && !errors.Is(err, redis.Nil) {
            applogger.L().Errorf("XReadGroup error: %v", err)
            time.Sleep(time.Second) // 退避
            continue
        }

        for _, xstream := range results {
            for _, msg := range xstream.Messages {
                wp.processRedisMessage(ctx, msg)
            }
        }
    }
}

func (wp *WorkerPool) processRedisMessage(ctx context.Context, msg redis.XMessage) {
    jobIDStr, ok := msg.Values["job_id"]
    if !ok {
        // 格式错误，直接 ACK 丢弃
        wp.ackRedis(ctx, msg.Stream, msg.ID)
        return
    }

    jobID, err := strconv.ParseUint(jobIDStr.(string), 10, 64)
    if err != nil {
        wp.ackRedis(ctx, msg.Stream, msg.ID)
        return
    }

    // 从 DB 加载 job
    var job model.BackgroundJob
    if err := wp.db.WithContext(ctx).First(&job, jobID).Error; err != nil {
        if errors.Is(err, gorm.ErrRecordNotFound) {
            // job 已被删除，ACK 丢弃
            wp.ackRedis(ctx, msg.Stream, msg.ID)
            return
        }
        // DB 错误，不 ACK，让 Redis 重投给其他 consumer
        applogger.L().Errorf("load job %d from DB failed: %v", jobID, err)
        return
    }

    now := wp.now()
    if err := archiveExhaustedRetries(wp.db.WithContext(ctx).Where("id = ?", jobID), now); err != nil {
        applogger.L().Errorf("archive exhausted job %d failed: %v", jobID, err)
        return
    }

    // Claim 检查：防止多 consumer 竞争同一 job
    result := wp.db.WithContext(ctx).Model(&model.BackgroundJob{}).
        Where("id = ? AND status IN ? AND scheduled_at <= ? AND attempts < max_attempts",
            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 {
        // 已被其他 consumer claim 或尚未到期，ACK 跳过
        wp.ackRedis(ctx, msg.Stream, msg.ID)
        return
    }

    // 重新加载完整 job（含 attempts 等更新后的字段）
    wp.db.WithContext(ctx).First(&job, jobID)

    // 执行 handler
    if err := wp.perform(ctx, &job); err != nil {
        wp.fail(ctx, &job, err)
    }

    // 确认 Redis 消息
    wp.ackRedis(ctx, msg.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)
    }
}
```

**影响分析**：
- `perform()` 和 `fail()` 方法**不变**
- `ProcessOne()` 方法保留，仅用于测试和 fallback 模式
- `claimNext()` 保留，仅用于 DB 轮询 fallback 路径
- `JobHandler` 签名 `func(ctx, *model.BackgroundJob) error` **不变**

### 3.4 延迟 Job 投递机制

**问题**：Redis Stream 不支持延迟投递。`scheduled_at > now` 的 job 入队时不应立即 XADD。

**方案**：Sweep goroutine 定期扫描到期 job，投递到 Redis。

```go
func (wp *WorkerPool) sweepLoop(ctx context.Context) {
    defer wp.wg.Done()
    ticker := time.NewTicker(wp.sweepInterval) // 30s
    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)
            }
        }
    }
}

func (wp *WorkerPool) sweepDueJobs(ctx context.Context) error {
    now := wp.now()
    var jobs []model.BackgroundJob
    // 查找已到期但仍在 queued/retrying 状态的 job
    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 _, job := range jobs {
        // 幂等投递：重复 XADD 不会导致重复执行
        // 因为 processRedisMessage 中有 DB claim 检查
        if err := wp.pushToRedis(ctx, &job); err != nil {
            applogger.L().Warnf("sweep: XAdd failed for job %d: %v", job.ID, err)
        }
    }
    return nil
}
```

**幂等保证**：sweep 可能对同一个 job 多次 XADD（因为不知道 Redis 中是否已有该 job 的待处理消息）。这不会导致重复执行，因为 `processRedisMessage` 中 DB claim 检查保证只有一个 consumer 能成功将 status 从 `queued`/`retrying` 改为 `running`。

**影响分析**：
- 延迟 job 的 `Enqueue` 调用方**不需要改动**
- 延迟从最坏 500ms（轮询间隔）变为最坏 30s（sweep 间隔）
- 对于需要更精确延迟的 job（如 `captain:conversation_response_builder` 的 1–5s 附件等待），可配置更短的 sweep 间隔或单独处理

### 3.5 RequeueStaleJobs 变更

**当前**：扫描 `status=running AND locked_at < cutoff`，重置为 `retrying`。

**迁移后**：仍由 `Start()` 在启动时调用一次（回收上次崩溃时正在执行的 job），并在 sweep 循环中定期调用。stale job 未耗尽 `max_attempts` 时转为 `retrying`；已耗尽时原子转为 `dead` 并写入 `failed_at`，不再发生下一次 claim。

```go
// Start() 中调用（不变）
if _, err := wp.RequeueStaleJobs(ctx); err != nil {
    return err
}

// sweepLoop 中新增定期调用
func (wp *WorkerPool) sweepLoop(ctx context.Context) {
    // ... sweepDueJobs ...
    // 新增：定期回收 stale running jobs
    if _, err := wp.RequeueStaleJobs(ctx); err != nil {
        applogger.L().Warnf("requeue stale jobs failed: %v", err)
    }
}
```

**影响分析**：`RequeueStaleJobs` 方法签名不变。只有未耗尽的 stale job 会被下一轮 sweep 重新 XADD 到 Redis；耗尽的 job 保持 `dead` 终态。

### 3.6 Start/Stop 变更

**Start() 新增**：
- 启动 sweep goroutine
- 确保 consumer group 已创建（`XGROUP CREATE`，如果不存在）

```go
func (wp *WorkerPool) Start() error {
    if wp.db == nil {
        return nil
    }
    wp.mu.Lock()
    if wp.cancel != nil {
        wp.mu.Unlock()
        return nil
    }
    wp.ctx, wp.cancel = context.WithCancel(context.Background())
    ctx := wp.ctx
    workerCount := wp.workerCount
    wp.mu.Unlock()

    if _, err := wp.RequeueStaleJobs(ctx); err != nil {
        return err
    }

    // 新增：确保 Redis consumer group 存在
    if wp.rdb != nil {
        wp.ensureConsumerGroups(ctx)
    }

    for i := 0; i < workerCount; i++ {
        wp.wg.Add(1)
        go wp.run(ctx)
    }

    // 新增：启动 sweep goroutine
    if wp.rdb != nil {
        wp.wg.Add(1)
        go wp.sweepLoop(ctx)
    }

    return nil
}

func (wp *WorkerPool) ensureConsumerGroups(ctx context.Context) {
    for _, queue := range wp.allQueues() {
        streamKey := wp.streamKeyFor(queue)
        // XGROUP CREATE 幂等：如果 group 已存在则忽略 error
        err := wp.rdb.XGroupCreate(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)
        }
    }
}
```

**Stop() 不变**：`cancel()` + `wg.Wait()` 已经能正确停止所有 goroutine（包括新增的 sweep）。

### 3.7 Fallback 模式（Redis 不可用）

当 `rdb == nil`（Redis 未配置或初始化失败）时，worker 自动回退到当前 DB 轮询模式：

```go
func (wp *WorkerPool) run(ctx context.Context) {
    defer wp.wg.Done()
    if wp.rdb == nil {
        wp.runDBPollLoop(ctx) // 现有 run() 逻辑，保留
        return
    }
    // ... Redis XREADGROUP 逻辑 ...
}
```

这保证：
- SQLite 测试模式（不依赖 Redis）继续工作
- Redis 故障时 worker 不中断
- 渐进迁移：可以先在部分实例上启用 Redis 队列

### 3.8 Concurrency 配置修复

**当前 bug**：`Bootstrap()` 使用 `worker.NewWorkerPool(db)`，`cfg.Worker.Concurrency` 未生效。

**修复**：

```go
// internal/app/bootstrap.go:109
// 变更前：
workerPool := worker.NewWorkerPool(db)

// 变更后：
workerPool := worker.NewWorkerPoolWithOptions(db,
    worker.WithRedisClient(rdb),
    worker.WithWorkerCount(cfg.Worker.Concurrency),
)
```

注意：`rdb` 在 `Bootstrap()` 中的初始化位于 [bootstrap.go:113](../backend/internal/app/bootstrap.go:113)，在 `workerPool` 创建之前。但当前代码先创建 `workerPool`（line 109），后创建 `rdb`（line 113）。需要调整顺序：先创建 `rdb`，再创建 `workerPool`。

### 3.9 Queue 列表动态化

**当前**：`WorkerPool.queues` 通过 `WithQueues` option 设置，如果未设置则监听所有 queue（`claimNext` 中不追加 queue filter）。

**迁移后**：Redis 模式下必须显式列出要监听的 stream。如果 `queues` 为空，默认监听所有已知 queue。

```go
func (wp *WorkerPool) allQueues() []string {
    if len(wp.queues) > 0 {
        return wp.queues
    }
    // 默认监听所有 queue
    return []string{
        "default", "high", "medium", "low",
        "events", "automation", "search",
        "scheduled_jobs", "deferred", "purgable",
    }
}

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))
        keys = append(keys, ">") // > 表示只读取新消息
    }
    return keys
}
```

---

## 4. 配置变更

### 4.1 config.go 变更

**文件**：`internal/config/config.go`

```go
type WorkerConfig struct {
    Concurrency    int    `mapstructure:"concurrency"`
    // 新增
    StreamPrefix   string `mapstructure:"redis_stream_prefix"` // 默认 "gochat:jobs"
    ConsumerGroup  string `mapstructure:"redis_consumer_group"` // 默认 "gochat-workers"
    BlockTimeoutS  int    `mapstructure:"redis_block_timeout_s"` // 默认 5
    SweepIntervalS int    `mapstructure:"redis_sweep_interval_s"` // 默认 30
}
```

新增 viper 默认值：

```go
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)
```

新增环境变量映射：

```go
"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",
```

### 4.2 config.dev.yaml 变更

```yaml
worker:
  concurrency: 5
  redis_stream_prefix: "gochat:jobs"
  redis_consumer_group: "gochat-workers-dev"
  redis_block_timeout_s: 5
  redis_sweep_interval_s: 15  # 开发环境更短，便于测试延迟 job
```

### 4.3 backend/configs/config.production.yaml 变更

```yaml
worker:
  concurrency: 10
  redis_stream_prefix: "gochat:jobs"
  redis_consumer_group: "gochat-workers"
  redis_block_timeout_s: 5
  redis_sweep_interval_s: 30
```

### 4.4 validator.go 变更

```go
// 新增校验
if cfg.Worker.RedisBlockTimeoutS < 1 {
    return fmt.Errorf("worker.redis_block_timeout_s must be >= 1")
}
if cfg.Worker.RedisSweepIntervalS < 1 {
    return fmt.Errorf("worker.redis_sweep_interval_s must be >= 1")
}
```

### 4.5 hot-reload 支持

`worker.concurrency`、`worker.redis_block_timeout_s`、`worker.redis_sweep_interval_s` 加入 `ReloadableFields`：

```go
// config.go ReloadableFields
"worker.redis_block_timeout_s",
"worker.redis_sweep_interval_s",
```

注意：`worker.concurrency` 改变需要重启 worker goroutine 才能生效，标记为需要重启（已在不可变字段列表中）。`stream_prefix` 和 `consumer_group` 改变也需要重启。`block_timeout_s` 和 `sweep_interval_s` 可以热更新（下一轮循环自动使用新值）。

---

## 5. 迁移后受影响功能的验证计划

### 5.1 核心路径验证

| 验证项 | 验证方法 | 预期结果 | 影响的 job type |
|--------|----------|----------|-----------------|
| 基本入队+执行 | `go test ./internal/worker/ -run TestWorkerPoolProcessOne` | job 从 queued → running → completed | 任意 |
| 幂等去重 | `go test ./internal/worker/ -run TestWorkerPoolEnqueueStoresPayloadAndIdempotency` | 重复 idempotency_key 返回同一 job | 19 种带 key 的 job |
| 失败重试+死信 | `go test ./internal/worker/ -run TestWorkerPoolRetriesThenDeadLetters` | attempts 达到 max 后 dead | 任意 |
| 延迟调度 | `go test ./internal/worker/ -run TestWorkerPoolRespectsScheduleAndQueues` | scheduled_at 未到时不执行 | `scheduled:trigger_items` 等 5 种 |
| Stale 回收 | `go test ./internal/worker/ -run TestWorkerPoolRequeuesStaleRunningJobs` | running 超时 → retrying | 任意 |
| Start/Stop 循环 | `go test ./internal/worker/ -run TestWorkerPoolStartAndStop` | goroutine 正确启动和停止 | 任意 |

### 5.2 Redis 队列路径验证（新增测试）

| 验证项 | 验证方法 | 预期结果 |
|--------|----------|----------|
| XADD + XREADGROUP 端到端 | miniredis 创建 stream，Enqueue 后 XREADGROUP 拿到 job_id，DB claim + perform 后 completed | job 状态正确流转 |
| Redis 投递失败降级 | mock Redis 返回 error，Enqueue 仍成功返回 job，sweep 后重新投递 | job 不丢失 |
| Redis 不可用 fallback | rdb=nil，回退 DB 轮询，ProcessOne 正常工作 | 兼容现有测试 |
| 多 consumer 竞争 | 两个 WorkerPool 实例同一 consumer group，同一 job 只被一个执行 | DB claim 检查生效 |
| Sweep 补偿 | Enqueue 时 XAdd 失败，sweep 后 XAdd 成功，job 被执行 | 补偿机制有效 |
| 延迟 job sweep | Enqueue 时 scheduled_at 未来时间，sweep 到期后投递 | 延迟 job 被投递 |
| Consumer group 创建 | 首次 Start 时 XGROUP CREATE，已存在时不报错 | 幂等创建 |

### 5.3 各功能模块验证

#### 5.3.1 消息发送

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `message:send_reply` 正常投递 | 触发对话回复，检查 channel provider 被调用 | 消息通过 Redis 队列投递到 worker 执行 |
| 高优先级 queue 先消费 | 同时入队 `high` 和 `low` job，检查执行顺序 | `high` queue 先被消费 |

#### 5.3.2 事件分发

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `event:dispatch_async` 投递 | 触发 channel event，检查 listener 被调用 | 事件通过 Redis 队列分发 |
| `event:listener_dispatch` 投递 | dispatch.EventDispatcher 异步分发 | 正确投递和执行 |

#### 5.3.3 自动化规则

| 验证项 | 方法 | 预期 |
|--------|------|------|
| Automation rule 触发 action delivery | 触发规则条件，检查 email/webhook/transcript delivery job 入队 | job 入 Redis Stream 并被执行 |
| Macro 执行 | 调用 macro，检查 `automation:macro_execution` job | 正常执行 |
| CSAT survey 发送 | resolve 对话，检查 `csat:survey_send` job | 正常执行，幂等不重复 |

#### 5.3.4 定时任务

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `scheduled:trigger_items` 1h 周期 | 检查自我重新入队 | 延迟 1h 后被 sweep 投递 |
| `sla:trigger_accounts` 5min 周期 | 检查自我重新入队 | 延迟 5min 后被 sweep 投递 |
| `captain:documents_schedule_syncs` 24h 周期 | 检查自我重新入队 | 延迟 24h 后被 sweep 投递 |
| `conversation:update_message_status` 延迟 | 检查延迟投递 | 按指定时间延迟 |
| `captain:conversation_response_builder` 附件等待 | 检查 1–5s 延迟 | 按计算时间延迟 |

#### 5.3.5 Webhook 入站消息

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `webhook:incoming_message_persist` | 模拟 FB/WA/TG webhook 回调 | 消息持久化 job 通过 Redis 队列执行 |
| `webhook:message_status_update` | 模拟状态回调 | 状态更新 job 正常执行 |
| `webhook:contact_messages_status_update` | 模拟联系人状态回调 | 正常执行 |

#### 5.3.6 搜索索引

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `search:index` 正常投递 | 触发消息/对话/联系人变更 | 搜索索引同步 job 通过 Redis 队列执行 |

#### 5.3.7 Captain AI

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `captain:copilot_response` | 发送 copilot 消息 | 正常执行 |
| `captain:conversation_response_builder` | 入站消息触发 | 延迟后正确执行 |
| `captain:document_sync/crawl` | 文档同步操作 | 正常执行 |
| `captain:documents_schedule_syncs` | 24h 定时同步 | 延迟后正确执行 |

#### 5.3.8 Contact Import/Export

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `contact:import` | 导入 CSV | 正常执行 |
| `contact:export` | 导出联系人 | 正常执行 |

#### 5.3.9 Conversation Maintenance

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `conversation:reopen_snoozed` | snooze 到期 | 正常 reopen |
| `conversation:resolution` | auto-resolve 到期 | 正常 resolve |
| `conversation:bulk_action` | 批量操作 | 正常执行 |
| `contact:bulk_action` | 批量操作 | 正常执行 |
| `conversation:delete_object` | 删除对话 | 正常执行 |

#### 5.3.10 SLA 处理

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `sla:trigger_accounts` | 5min 周期触发 | 延迟后正确执行 |
| `sla:process_account` | account 级处理 | 正常执行 |
| `sla:process_applied` | applied SLA 处理 | 正常执行 |

#### 5.3.11 Reporting

| 验证项 | 方法 | 预期 |
|--------|------|------|
| `reporting:rollup_day` | 触发日汇总 | 正常执行 |

### 5.4 部署验证

| 验证项 | 方法 | 预期 |
|--------|------|------|
| Quickstart compose | `cd deploy/quickstart && docker compose up -d --build` | 单实例 API+worker 正常运行 |
| Prod compose worker 分离 | `docker-compose.prod.yml` worker service | `serve --worker-only` 正常工作（需实现） |
| CI postgres + redis | `go test -race ./internal/... ./pkg/... ./cmd/...` | 全部通过 |
| CI sqlite (无 redis) | `GOCHAT_TEST_DB=sqlite go test ./internal/... ./pkg/... ./cmd/...` | fallback 模式全部通过 |

### 5.5 回归测试清单

迁移后需要全部通过的现有测试文件（25 个）：

| 测试文件 | 依赖的 worker 行为 |
|----------|-------------------|
| `internal/worker/worker_test.go` | Enqueue、ProcessOne、retry、dead-letter、schedule、stale、start/stop |
| `internal/dispatch/dispatcher_worker_test.go` | EventDispatcher async job |
| `internal/channel/dispatcher_worker_test.go` | channel.Dispatcher async job |
| `internal/automation/action_service_test.go` | Action delivery job |
| `internal/automation/macro_service_test.go` | Macro execution job |
| `internal/automation/csat_survey_listener_test.go` | CSAT survey send job |
| `internal/service/conversation_maintenance_worker_test.go` | 全部 maintenance job |
| `internal/service/sla_worker_test.go` | SLA processing job |
| `internal/service/csat_template_worker_test.go` | CSAT template create job |
| `internal/service/copilot_response_worker_test.go` | Copilot response job |
| `internal/service/captain_conversation_worker_test.go` | Captain conversation builder job |
| `internal/service/captain_document_service_test.go` | Captain document sync/crawl job |
| `internal/service/search_indexer_worker_test.go` | Search index job |
| `internal/service/message_delivery_worker_test.go` | Message send reply job |
| `internal/service/contact_service_g3_test.go` | Contact import/export job |
| `internal/service/conversation_service_test.go` | Conversation delete job |
| `internal/service/inbox_service_test.go` | Inbox template sync job |
| `internal/service/widget_service_test.go` | Webwidget triggered job |
| `internal/service/article_service_test.go` | Article translate job |
| `internal/service/message_service_test.go` | Message service job |
| `internal/service/analytics_p513_test.go` | Analytics job |
| `internal/handler/webhook/webhook_lookup_test.go` | Webhook job |
| `internal/handler/api/v1/inbox_handler_test.go` | Inbox handler job |
| `internal/handler/api/v1/bulk_action_handler_test.go` | Bulk action job |
| `internal/handler/api/v1/article_handler_test.go` | Article handler job |

---

## 6. 实施步骤

### Phase 1：核心改造（不改变对外接口）

| 步骤 | 文件 | 改动 |
|------|------|------|
| 1.1 | `internal/worker/worker.go` | WorkerPool 结构体加 Redis 字段；新增 `WithRedisClient` 等 Option |
| 1.2 | `internal/worker/worker.go` | `Enqueue` 末尾加 `pushToRedis` |
| 1.3 | `internal/worker/worker.go` | `run` 改为 `XREADGROUP BLOCK`；保留 `runDBPollLoop` fallback |
| 1.4 | `internal/worker/worker.go` | 新增 `processRedisMessage`、`ackRedis`、`pushToRedis`、`streamKeyFor`、`streamKeys`、`allQueues` |
| 1.5 | `internal/worker/worker.go` | 新增 `sweepLoop`、`sweepDueJobs` |
| 1.6 | `internal/worker/worker.go` | `Start` 加 `ensureConsumerGroups` + sweep goroutine |
| 1.7 | `internal/worker/worker.go` | `ProcessOne`、`claimNext`、`perform`、`fail` 保留不变 |

### Phase 2：配置与接入

| 步骤 | 文件 | 改动 |
|------|------|------|
| 2.1 | `internal/config/config.go` | WorkerConfig 加 4 个 Redis 字段 |
| 2.2 | `internal/config/config.go` | viper SetDefault + 环境变量映射 |
| 2.3 | `internal/config/config.go` | applyZeroDefaults 补充新字段 |
| 2.4 | `internal/config/validator.go` | 新增字段校验 |
| 2.5 | `internal/app/bootstrap.go` | 调整 rdb 创建顺序（移到 workerPool 之前），改用 `NewWorkerPoolWithOptions` |
| 2.6 | `configs/config.dev.yaml` | 新增 worker Redis 配置 |
| 2.7 | `backend/configs/config.production.yaml` | 新增 worker Redis 配置 |

### Phase 3：测试

| 步骤 | 文件 | 改动 |
|------|------|------|
| 3.1 | `internal/worker/worker_test.go` | 新增 miniredis 测试：Redis 端到端、竞争、sweep、fallback |
| 3.2 | `internal/worker/worker_test.go` | 现有测试保持：rdb=nil 时 fallback 到 DB 轮询 |
| 3.3 | 全部 worker 相关测试 | `GOCHAT_TEST_DB=sqlite` 跑一遍确认 fallback 模式通过 |

### Phase 4：独立 worker 进程（可选，后续迭代）

| 步骤 | 文件 | 改动 |
|------|------|------|
| 4.1 | `cmd/gochat/main.go` | 新增 `serve --worker-only` 子命令 |
| 4.2 | `internal/app/bootstrap.go` | 拆出 `BootstrapWorker()` 函数，只初始化 DB + Redis + worker 相关依赖 |
| 4.3 | `deploy/docker/Dockerfile` | 更新注释，`gochat-worker` 二进制对应 `serve --worker-only` |
| 4.4 | `deploy/docker/docker-compose.prod.yml` | worker service command 改为实际可用命令 |

---

## 7. 风险与缓解

| 风险 | 影响 | 缓解 |
|------|------|------|
| Redis 宕机 | worker 无法拉取新 job | 自动 fallback 到 DB 轮询；已有 job 在 DB 中不丢失 |
| Redis Stream 消息积压 | 内存增长 | 配置 `MAXLEN` 限制 stream 长度；sweep + XACK 清理已处理消息 |
| Sweep 延迟过高 | 延迟 job 执行晚于预期 | 可配置 `sweep_interval`（默认 30s，开发环境 15s） |
| 多实例 worker 重复执行 | 同一 job 被执行两次 | DB claim 检查（`WHERE status IN ... AND ...` + `RowsAffected` 检查）保证只有一个成功 |
| Redis Stream 不保证顺序 | 同一 queue 内 job 执行顺序不保证 | 当前 DB 轮询也不保证多 worker 间的顺序；`priority` 改为 queue 级别优先级 |
| 迁移期间混用 | 部分 instance 用 DB 轮询，部分用 Redis | DB claim 检查兼容两种 consumer；Redis XADD + DB INSERT 同时存在不会冲突 |

---

## 8. 不改动的部分

以下内容在迁移中**完全不改动**：

- `BackgroundJob` model（字段、索引）
- `background_jobs` 表 schema 和 migration
- 所有 `Register*Jobs` 函数（`RegisterConversationMaintenanceJobs`、`RegisterActionDeliveryJobs` 等）
- 所有 `Enqueue*` 辅助函数（`EnqueueSendReply`、`EnqueueScheduledItemsTrigger` 等）
- 所有 `SetWorkerPool` 调用
- 所有 `JobHandler` 实现
- 所有 job type 常量
- 所有 queue 名称
- `EnqueueOption` / `WithQueue` / `WithScheduledAt` / `WithMaxAttempts` / `WithPriority` / `WithIdempotencyKey` 函数签名
- `perform()` 和 `fail()` 方法实现
- `RequeueStaleJobs()` 方法签名和 DB 操作
- `defaultBackoff()` 函数
- `marshalPayload()` 函数
- Redis Pub/Sub 广播（`internal/ws/broadcast.go`）
- Watermill EventBus（`internal/pubsub/event_bus.go`）
- Watermill RedisPubSub（`internal/pubsub/redis_pubsub.go`）
- NotificationDeliveryService（`internal/service/notification_delivery_service.go`）

---

## 9. 文件改动清单

| 文件 | 类型 | 改动范围 |
|------|------|----------|
| `internal/worker/worker.go` | 核心改动 | 结构体字段、Enqueue、run、Start、新增 processRedisMessage/sweep |
| `internal/worker/worker_test.go` | 新增测试 | miniredis Redis 队列路径测试 |
| `internal/config/config.go` | 配置扩展 | WorkerConfig 结构体、defaults、env 映射 |
| `internal/config/validator.go` | 校验 | 新字段校验 |
| `internal/app/bootstrap.go` | 接入 | rdb 顺序调整、NewWorkerPoolWithOptions |
| `configs/config.dev.yaml` | 配置 | worker Redis 参数 |
| `backend/configs/config.production.yaml` | 配置 | worker Redis 参数 |
| `deploy/docker/Dockerfile` | 注释更新 | gochat-worker 说明（Phase 4） |
| `deploy/docker/docker-compose.prod.yml` | 部署 | worker service command（Phase 4） |