diff --git a/docs/architecture/adr-003-unified-ai-job-engine.md b/docs/architecture/adr-003-unified-ai-job-engine.md index cb68826..42c7094 100644 --- a/docs/architecture/adr-003-unified-ai-job-engine.md +++ b/docs/architecture/adr-003-unified-ai-job-engine.md @@ -315,17 +315,115 @@ export class AiJobModule implements OnModuleInit { } ``` +#### 2.4 AiJobService — 统一 Job 创建入口与幂等性 + +```typescript +// AiJobService — 唯一 Job 创建入口(internal-only) +@Injectable() +export class AiJobService { + constructor( + private readonly prisma: PrismaService, + private readonly registry: JobDefinitionRegistry, + private readonly lifecycleRepo: AiJobLifecycleRepository, + private readonly outboxRepo: OutboxRepository, + private readonly snapshotBuilder: SnapshotBuilderService, + ) {} + + /** + * 创建 Job + Snapshot + Outbox(一个 Prisma Transaction 内)。 + * + * 幂等性: + * - AiJob 表有 @@unique([userId, jobType, idempotencyKey]) + * - 如果 (userId, jobType, idempotencyKey) 三元组匹配到已有 Job,直接返回已有 Job + * - 未提供 idempotencyKey 时每次创建新 Job + * + * 事务边界: + * - BEGIN TRANSACTION + * - createJob(tx) → INSERT AiJob (lifecycleStatus = 'queued') + * - createInTransaction(tx) → INSERT OutboxEvent (eventType = 'ai.job.enqueue') + * - COMMIT + * - OutboxDispatcher 异步投递到 BullMQ + */ + async createJob(input: CreateJobInput): Promise; +} + +interface CreateJobInput { + userId: string; + jobType: string; + targetType: string; + targetId: string; + /** 幂等 Key。提供时启用幂等;不提供时每次创建新 Job */ + idempotencyKey?: string; + /** 覆盖 Definition 默认值 */ + overrides?: { + priority?: number; + credentialMode?: 'platform_key' | 'user_deepseek_key'; + credentialId?: string; + promptVersion?: string; + outputSchemaVersion?: string; + }; +} +``` + +**幂等流程**: + +``` +AiJobService.createJob(input) + │ + ├─ 1. lookup JobDefinition via Registry.get(input.jobType) + │ + ├─ 2. if input.idempotencyKey is provided: + │ existing = prisma.aiJob.findUnique({ + │ where: { userId_jobType_idempotencyKey: { + │ userId: input.userId, + │ jobType: input.jobType, + │ idempotencyKey: input.idempotencyKey + │ }} + │ }) + │ if existing → return existing // 幂等返回,不创建新 Job + │ + ├─ 3. build snapshot via SnapshotBuilder + │ + ├─ 4. prisma.$transaction(async (tx) => { + │ job = lifecycleRepo.createJob(tx, input) + │ outboxRepo.createInTransaction(tx, { + │ eventType: 'ai.job.enqueue', + │ aggregateType: 'AiJob', + │ aggregateId: job.id, + │ dedupeKey: `${input.jobType}:${input.userId}:${input.idempotencyKey ?? job.id}`, + │ payload: { jobId: job.id, jobType: input.jobType, queueName: def.queue.queueName, priority: job.priority }, + │ availableAt: new Date(), + │ }) + │ return job + │ }) + │ + └─ 5. return job +``` + +**约束**: +- 不提供通用公开创建接口 —— 外部通过具体业务 Service(如 `ActiveRecallService.createJob()`)调用 +- `CreateJobInput` 的类型由 `JobDefinition` 的 `input.schemaVersion` 隐式约束 +- 并发相同 `(userId, jobType, idempotencyKey)` → `findUnique` 无记录 → 两个事务同时 INSERT → 一个成功,一个 `P2002` 唯一冲突 → catch 后 `findUnique` 返回已有(赢者)Job + --- ### 3. 队列架构 #### 3.1 队列配置 -| 队列名 | 用途 | Worker 进程 | 默认 concurrency | lockDuration | attempts | -|--------|------|------------|-----------------|-------------|----------| -| `ai-interactive` | 新统一 Engine — 交互式 Job(用户等待结果) | `WorkerModule` 新增 | 2 | 30s | 3 | -| `ai-background` | 新统一 Engine — 后台 Job(用户不等待) | `WorkerModule` 新增 | 3 | 60s | 3 | -| `ai-analysis` | 旧系统 A — active-recall + feynman-evaluation | `WorkerModule`(已有) | 1 | 30s | 3 | +> 以下值来自 Issue #288 原始规格。有意偏离需在对应 Issue 中注明原因。 + +| 队列名 | 用途 | Worker 进程 | concurrency | lockDuration | attempts | backoff | +|--------|------|------------|-------------|-------------|----------|---------| +| `ai-interactive` | 新统一 Engine — 交互式 Job(用户等待结果) | `WorkerModule` 新增 | 2 | 30s | 2 | exponential 2000ms | +| `ai-background` | 新统一 Engine — 后台 Job(用户不等待) | `WorkerModule` 新增 | 1 | 60s | 3 | exponential 5000ms | +| `ai-analysis` | 旧系统 A — active-recall + feynman-evaluation | `WorkerModule`(已有) | 1 | 30s | 3 | exponential 1000ms | + +**设计理由**: + +- `ai-interactive` attempts=2(非 3):用户等待结果,BullMQ retry 会延长感知延迟。Engine 内部已有一层重试,BullMQ 仅兜底 Worker 崩溃。 +- `ai-background` concurrency=1(非 3):后台 Job 以吞吐而非延迟为目标;单 Worker 串行消费避免 DeepSeek API 并发限流。若后续需要更大吞吐,通过水平扩展 Worker 实例而非提高 concurrency。 +- `ai-background` backoff=5000ms(非默认 1000ms):后台场景无实时性要求,更长退避减少无效重试。 **约束**: