[M-AI-03] 注册新队列与薄 Worker Processor #288

New Issue

wangdl · 2026-06-20T16:43:36+08:00

wangdl commented

2026-06-20 16:43:36 +08:00

类型 / 标签

backend
bullmq
worker
infrastructure

风险等级

P1

依赖

M-AI-03-02
M-AI-03-04

目标

新增两个统一 Engine 队列和对应的薄 Processor，但不迁移旧业务 Job。

新队列

ai-interactive
ai-background

配置按 ADR：

ai-interactive

concurrency: 2
attempts: 2
backoff: exponential 2000ms

ai-background

concurrency: 1
attempts: 3
backoff: exponential 5000ms

Processor

新增两个薄 Worker：

AiInteractiveJobWorker
AiBackgroundJobWorker

职责只能是：

接收 {jobId}
→ 调用 AiJobExecutionEngine.execute(jobId, bullJobContext)

禁止在 Processor 中：

读取 Prompt；
调用 Provider；
写业务表；
实现状态机；
实现 Job Type switch。

模块边界

只注册在 WorkerModule；
API AppModule 不得实例化 Processor；
旧 ai-analysis Processor 保持原样；
不监听新的 HTTP 端口。

非目标

不实现 Execution Engine（由 M-AI-03-07 负责）；
不迁移旧业务 Job。

验收标准

API 启动后新 Processor 数量为 0
Worker 启动后两个 Processor（AiInteractiveJobWorker / AiBackgroundJobWorker）存在
Queue 配置（concurrency、attempts、backoff）来自统一 Registry/Config
Redis Payload 只包含 {"jobId": "<AiJob.id>"}，不含任何其他字段
旧 M-AI-01 Worker Integration 测试继续通过

## 类型 / 标签 - backend - bullmq - worker - infrastructure ## 风险等级 P1 ## 依赖 - M-AI-03-02 - M-AI-03-04 ## 目标新增两个统一 Engine 队列和对应的薄 Processor，但不迁移旧业务 Job。 ## 新队列 ``` ai-interactive ai-background ``` 配置按 ADR： ### ai-interactive - concurrency: 2 - attempts: 2 - backoff: exponential 2000ms ### ai-background - concurrency: 1 - attempts: 3 - backoff: exponential 5000ms ## Processor 新增两个薄 Worker： ``` AiInteractiveJobWorker AiBackgroundJobWorker ``` 职责只能是： ``` 接收 {jobId} → 调用 AiJobExecutionEngine.execute(jobId, bullJobContext) ``` 禁止在 Processor 中： - 读取 Prompt； - 调用 Provider； - 写业务表； - 实现状态机； - 实现 Job Type switch。 ## 模块边界 - 只注册在 WorkerModule； - API AppModule 不得实例化 Processor； - 旧 `ai-analysis` Processor 保持原样； - 不监听新的 HTTP 端口。 ## 非目标 - 不实现 Execution Engine（由 M-AI-03-07 负责）； - 不迁移旧业务 Job。 ## 验收标准 - [ ] API 启动后新 Processor 数量为 0 - [ ] Worker 启动后两个 Processor（AiInteractiveJobWorker / AiBackgroundJobWorker）存在 - [ ] Queue 配置（concurrency、attempts、backoff）来自统一 Registry/Config - [ ] Redis Payload 只包含 `{"jobId": "<AiJob.id>"}`，不含任何其他字段 - [ ] 旧 M-AI-01 Worker Integration 测试继续通过

wangdl added this to the M-AI-03 统一 Job Engine、Registry 与 Outbox milestone 2026-06-20 16:44:54 +08:00

wangdl referenced this issue from a commit

2026-06-20 17:11:21 +08:00

fix: ADR-003 补充幂等流程 + 修正队列配置对齐 #288 规格

wangdl commented

2026-06-20 18:03:27 +08:00

开发完成评论

完成内容

queue.constants.ts：新增 QUEUE_AI_INTERACTIVE / QUEUE_AI_BACKGROUND
queue-definitions.ts：队列配置对齐 ADR-003 §3.1
- ai-interactive: concurrency=2, lockDuration=30s, attempts=2, backoff=exponential 2000ms
- ai-background: concurrency=1, lockDuration=60s, attempts=3, backoff=exponential 5000ms
- 环境变量可覆盖（BULL_AI_INTERACTIVE_* / BULL_AI_BACKGROUND_*）
queue.service.ts：QueueService 注入并路由两个新队列
queue.module.ts：BullModule.registerQueue 注册新队列
AiJobExecutionEngine 接口（ai-job-execution-engine.interface.ts）：execute(aiJobId, context) 签名 + EngineJobContext（jobId/attemptMade/updateProgress）+ AI_JOB_EXECUTION_ENGINE 字符串 Token
AiInteractiveJobWorker（ai-interactive.worker.ts）：薄 Processor — 仅接收 {jobId} payload → 注入 Engine → engine.execute(jobId, context)
AiBackgroundJobWorker（ai-background.worker.ts）：同上
worker.module.ts：注册两个新 Processor（Engine @Optional() 注入，#291 实现前为 no-op）

修改文件

src/infrastructure/queue/queue.constants.ts（+2 constants）
src/infrastructure/queue/queue-definitions.ts（+56 lines）
src/infrastructure/queue/queue.service.ts（+4 imports/injections/routes）
src/infrastructure/queue/queue.module.ts（+2 queue registrations）
src/modules/ai-job/ai-job-execution-engine.interface.ts（新增）
src/workers/ai-interactive.worker.ts（新增）
src/workers/ai-background.worker.ts（新增）
src/worker.module.ts（+2 imports/providers）

约束验证

薄 Processor 不读 Prompt、不调 Provider、不写业务表、不实现状态机、不实现 Job Type switch
仅注册在 WorkerModule；AppModule 不实例化
旧 ai-analysis Processor 保持原样
Payload 仅 {jobId}，无其他字段

测试情况

npx tsc --noEmit --skipLibCheck → 无错误
npx jest --testPathPatterns="ai-job" --no-coverage → 119/119 通过（4 suites）

是否建议进入 Review

是

## 开发完成评论 ### 完成内容 - **queue.constants.ts**：新增 `QUEUE_AI_INTERACTIVE` / `QUEUE_AI_BACKGROUND` - **queue-definitions.ts**：队列配置对齐 ADR-003 §3.1 - `ai-interactive`: concurrency=2, lockDuration=30s, attempts=2, backoff=exponential 2000ms - `ai-background`: concurrency=1, lockDuration=60s, attempts=3, backoff=exponential 5000ms - 环境变量可覆盖（`BULL_AI_INTERACTIVE_*` / `BULL_AI_BACKGROUND_*`） - **queue.service.ts**：QueueService 注入并路由两个新队列 - **queue.module.ts**：BullModule.registerQueue 注册新队列 - **AiJobExecutionEngine 接口**（`ai-job-execution-engine.interface.ts`）：`execute(aiJobId, context)` 签名 + `EngineJobContext`（jobId/attemptMade/updateProgress）+ `AI_JOB_EXECUTION_ENGINE` 字符串 Token - **AiInteractiveJobWorker**（`ai-interactive.worker.ts`）：薄 Processor — 仅接收 `{jobId}` payload → 注入 Engine → `engine.execute(jobId, context)` - **AiBackgroundJobWorker**（`ai-background.worker.ts`）：同上 - **worker.module.ts**：注册两个新 Processor（Engine `@Optional()` 注入，#291 实现前为 no-op） ### 修改文件 - `src/infrastructure/queue/queue.constants.ts`（+2 constants） - `src/infrastructure/queue/queue-definitions.ts`（+56 lines） - `src/infrastructure/queue/queue.service.ts`（+4 imports/injections/routes） - `src/infrastructure/queue/queue.module.ts`（+2 queue registrations） - `src/modules/ai-job/ai-job-execution-engine.interface.ts`（新增） - `src/workers/ai-interactive.worker.ts`（新增） - `src/workers/ai-background.worker.ts`（新增） - `src/worker.module.ts`（+2 imports/providers） ### 约束验证 - [x] 薄 Processor 不读 Prompt、不调 Provider、不写业务表、不实现状态机、不实现 Job Type switch - [x] 仅注册在 WorkerModule；AppModule 不实例化 - [x] 旧 ai-analysis Processor 保持原样 - [x] Payload 仅 `{jobId}`，无其他字段 ### 测试情况 - `npx tsc --noEmit --skipLibCheck` → 无错误 - `npx jest --testPathPatterns="ai-job" --no-coverage` → **119/119 通过**（4 suites） ### 是否建议进入 Review - 是

wangdl referenced this issue from a commit

2026-06-20 18:30:06 +08:00

feat: M-AI-03 统一 AiJob Worker Execution Engine

wangdl referenced this issue

2026-06-20 18:31:59 +08:00

[M-AI-03] 实现统一 AiJob Worker Execution Engine #291

wangdl closed this issue

2026-06-21 12:37:50 +08:00

Sign in to join this conversation.