From fcce35742f54c988f14529d633d8529182998596 Mon Sep 17 00:00:00 2001 From: wangdl Date: Sat, 20 Jun 2026 18:02:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M-AI-03=20=E6=B3=A8=E5=86=8C=E6=96=B0?= =?UTF-8?q?=E9=98=9F=E5=88=97=20ai-interactive/ai-background=20=E4=B8=8E?= =?UTF-8?q?=E8=96=84=20Worker=20Processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - queue.constants.ts: 新增 QUEUE_AI_INTERACTIVE / QUEUE_AI_BACKGROUND - queue-definitions.ts: 队列配置对齐 ADR-003 §3.1 ai-interactive: concurrency=2, attempts=2, backoff=2000ms ai-background: concurrency=1, attempts=3, backoff=5000ms - queue.service.ts: QueueService 注入并路由新队列 - queue.module.ts: BullModule.registerQueue 注册新队列 - AiJobExecutionEngine 接口 + AI_JOB_EXECUTION_ENGINE 注入 Token - AiInteractiveJobWorker + AiBackgroundJobWorker: 薄 Processor，仅接收 {jobId} 调用 Engine - worker.module.ts: 注册两个新 Processor（#291 提供 Engine 真实实现前为 no-op） Co-Authored-By: Claude --- src/infrastructure/queue/queue-definitions.ts | 51 +++++++++++++++ src/infrastructure/queue/queue.constants.ts | 3 + src/infrastructure/queue/queue.module.ts | 4 +- src/infrastructure/queue/queue.service.ts | 8 +++ .../ai-job-execution-engine.interface.ts | 34 ++++++++++ src/worker.module.ts | 4 ++ src/workers/ai-background.worker.ts | 62 +++++++++++++++++++ src/workers/ai-interactive.worker.ts | 62 +++++++++++++++++++ 8 files changed, 227 insertions(+), 1 deletion(-) create mode 100644 src/modules/ai-job/ai-job-execution-engine.interface.ts create mode 100644 src/workers/ai-background.worker.ts create mode 100644 src/workers/ai-interactive.worker.ts diff --git a/src/infrastructure/queue/queue-definitions.ts b/src/infrastructure/queue/queue-definitions.ts index dbfb7a2..4cb981d 100644 --- a/src/infrastructure/queue/queue-definitions.ts +++ b/src/infrastructure/queue/queue-definitions.ts @@ -5,6 +5,8 @@ import { QUEUE_DOMAIN_EVENTS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, + QUEUE_AI_INTERACTIVE, + QUEUE_AI_BACKGROUND, } from './queue.constants'; // ═══════════════════════════════════════════════════════════════════════ @@ -98,6 +100,55 @@ export const QUEUE_DEFINITIONS: Record = { [QUEUE_DOMAIN_EVENTS]: buildDef(QUEUE_DOMAIN_EVENTS), [QUEUE_AUDIT_LOG]: buildDef(QUEUE_AUDIT_LOG), [QUEUE_FILE_CLEANUP]: buildDef(QUEUE_FILE_CLEANUP), + // ── M-AI-03: 统一 Job Engine 队列（ADR-003 §3.1 规格，环境变量可覆盖）── + [QUEUE_AI_INTERACTIVE]: { + name: QUEUE_AI_INTERACTIVE, + workerOptions: { + concurrency: envInt('BULL_AI_INTERACTIVE_CONCURRENCY', 2, 1), + lockDuration: envInt('BULL_AI_INTERACTIVE_LOCK_DURATION', 30_000, 1), + stalledInterval: envInt('BULL_AI_INTERACTIVE_STALLED_INTERVAL', 30_000, 1), + maxStalledCount: envInt('BULL_AI_INTERACTIVE_MAX_STALLED', 1, 1), + }, + defaultJobOptions: { + attempts: envInt('BULL_AI_INTERACTIVE_ATTEMPTS', 2, 1), + backoff: { + type: 'exponential', + delay: envInt('BULL_AI_INTERACTIVE_BACKOFF_DELAY', 2000), + }, + removeOnComplete: { + count: envInt('BULL_COMPLETED_MAX_COUNT', 1000), + age: envInt('BULL_COMPLETED_MAX_AGE_SEC', 24 * 3600), + }, + removeOnFail: { + count: envInt('BULL_FAILED_MAX_COUNT', 5000), + age: envInt('BULL_FAILED_MAX_AGE_SEC', 7 * 24 * 3600), + }, + }, + }, + [QUEUE_AI_BACKGROUND]: { + name: QUEUE_AI_BACKGROUND, + workerOptions: { + concurrency: envInt('BULL_AI_BACKGROUND_CONCURRENCY', 1, 1), + lockDuration: envInt('BULL_AI_BACKGROUND_LOCK_DURATION', 60_000, 1), + stalledInterval: envInt('BULL_AI_BACKGROUND_STALLED_INTERVAL', 60_000, 1), + maxStalledCount: envInt('BULL_AI_BACKGROUND_MAX_STALLED', 1, 1), + }, + defaultJobOptions: { + attempts: envInt('BULL_AI_BACKGROUND_ATTEMPTS', 3, 1), + backoff: { + type: 'exponential', + delay: envInt('BULL_AI_BACKGROUND_BACKOFF_DELAY', 5000), + }, + removeOnComplete: { + count: envInt('BULL_COMPLETED_MAX_COUNT', 1000), + age: envInt('BULL_COMPLETED_MAX_AGE_SEC', 24 * 3600), + }, + removeOnFail: { + count: envInt('BULL_FAILED_MAX_COUNT', 5000), + age: envInt('BULL_FAILED_MAX_AGE_SEC', 7 * 24 * 3600), + }, + }, + }, }; export function queueDef(name: string): QueueDefinition { diff --git a/src/infrastructure/queue/queue.constants.ts b/src/infrastructure/queue/queue.constants.ts index 87307cd..5c988ff 100644 --- a/src/infrastructure/queue/queue.constants.ts +++ b/src/infrastructure/queue/queue.constants.ts @@ -4,3 +4,6 @@ export const QUEUE_NOTIFICATION = 'notification'; export const QUEUE_DOMAIN_EVENTS = 'domain-events'; export const QUEUE_AUDIT_LOG = 'audit-logs'; export const QUEUE_FILE_CLEANUP = 'file-cleanup'; +// M-AI-03: 统一 Job Engine 队列 +export const QUEUE_AI_INTERACTIVE = 'ai-interactive'; +export const QUEUE_AI_BACKGROUND = 'ai-background'; diff --git a/src/infrastructure/queue/queue.module.ts b/src/infrastructure/queue/queue.module.ts index dd30aec..923e0ce 100644 --- a/src/infrastructure/queue/queue.module.ts +++ b/src/infrastructure/queue/queue.module.ts @@ -2,7 +2,7 @@ import { Global, Module } from '@nestjs/common'; import { BullModule } from '@nestjs/bullmq'; import { ConfigService } from '@nestjs/config'; import { PrismaService } from '../database/prisma.service'; -import { QueueService, QUEUE_AI_ANALYSIS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, QUEUE_DOCUMENT_IMPORT, QUEUE_NOTIFICATION, QUEUE_DOMAIN_EVENTS } from './queue.service'; +import { QueueService, QUEUE_AI_ANALYSIS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, QUEUE_DOCUMENT_IMPORT, QUEUE_NOTIFICATION, QUEUE_DOMAIN_EVENTS, QUEUE_AI_INTERACTIVE, QUEUE_AI_BACKGROUND } from './queue.service'; @Global() @Module({ @@ -31,6 +31,8 @@ import { QueueService, QUEUE_AI_ANALYSIS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, Q { name: QUEUE_DOMAIN_EVENTS }, { name: QUEUE_AUDIT_LOG }, { name: QUEUE_FILE_CLEANUP }, + { name: QUEUE_AI_INTERACTIVE }, + { name: QUEUE_AI_BACKGROUND }, ), ], providers: [QueueService, PrismaService], diff --git a/src/infrastructure/queue/queue.service.ts b/src/infrastructure/queue/queue.service.ts index 52fd42c..868b56b 100644 --- a/src/infrastructure/queue/queue.service.ts +++ b/src/infrastructure/queue/queue.service.ts @@ -13,6 +13,8 @@ export { QUEUE_DOMAIN_EVENTS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, + QUEUE_AI_INTERACTIVE, + QUEUE_AI_BACKGROUND, } from './queue.constants'; import { QUEUE_AI_ANALYSIS, @@ -21,6 +23,8 @@ import { QUEUE_DOMAIN_EVENTS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, + QUEUE_AI_INTERACTIVE, + QUEUE_AI_BACKGROUND, } from './queue.constants'; @Injectable() @@ -35,6 +39,8 @@ export class QueueService { @InjectQueue(QUEUE_DOMAIN_EVENTS) private readonly domainEventsQueue: Queue, @InjectQueue(QUEUE_AUDIT_LOG) private readonly auditLogQueue: Queue, @InjectQueue(QUEUE_FILE_CLEANUP) private readonly fileCleanupQueue: Queue, + @InjectQueue(QUEUE_AI_INTERACTIVE) private readonly aiInteractiveQueue: Queue, + @InjectQueue(QUEUE_AI_BACKGROUND) private readonly aiBackgroundQueue: Queue, @Optional() private readonly eventBus?: any, ) {} @@ -70,6 +76,8 @@ export class QueueService { case QUEUE_DOMAIN_EVENTS: return this.domainEventsQueue; case QUEUE_AUDIT_LOG: return this.auditLogQueue; case QUEUE_FILE_CLEANUP: return this.fileCleanupQueue; + case QUEUE_AI_INTERACTIVE: return this.aiInteractiveQueue; + case QUEUE_AI_BACKGROUND: return this.aiBackgroundQueue; default: throw new Error(`Unknown queue: ${name}`); } } diff --git a/src/modules/ai-job/ai-job-execution-engine.interface.ts b/src/modules/ai-job/ai-job-execution-engine.interface.ts new file mode 100644 index 0000000..1560aa7 --- /dev/null +++ b/src/modules/ai-job/ai-job-execution-engine.interface.ts @@ -0,0 +1,34 @@ +/** + * M-AI-03: AiJobExecutionEngine 接口与注入 Token + * + * 薄 Worker Processor（AiInteractiveJobWorker / AiBackgroundJobWorker） + * 依赖此接口。M-AI-03-08（#291）提供真实实现。 + * + * 当前阶段（#288）提供 no-op 占位，使 Worker 可正常启动。 + * #291 将在 AiJobModule 中覆盖此 Provider。 + */ + +import type { Job } from 'bullmq'; + +/** 薄 Worker 传入 Engine 的 BullMQ Job 上下文（最小化） */ +export interface EngineJobContext { + /** BullMQ jobId */ + readonly jobId: string; + /** 当前尝试次数 */ + readonly attemptMade: number; + /** 更新进度（0–100） */ + readonly updateProgress: (progress: number) => Promise; +} + +export interface AiJobExecutionEngine { + /** + * 执行 Job 的完整管线（PREPARE → RESOLVE → EXECUTE → PROJECT → COMPLETE）。 + * + * @param aiJobId AiJob 数据库主键 + * @param context BullMQ Job 上下文 + */ + execute(aiJobId: string, context: EngineJobContext): Promise; +} + +/** NestJS 注入 Token */ +export const AI_JOB_EXECUTION_ENGINE = 'AI_JOB_EXECUTION_ENGINE'; diff --git a/src/worker.module.ts b/src/worker.module.ts index 6f940b0..a4669bb 100644 --- a/src/worker.module.ts +++ b/src/worker.module.ts @@ -12,6 +12,8 @@ import { LoggerModule } from './infrastructure/logger/logger.module'; import { AiAnalysisWorker } from './workers/ai-analysis.worker'; import { DocumentImportWorker } from './workers/document-import.worker'; import { NotificationWorker } from './workers/notification.worker'; +import { AiInteractiveJobWorker } from './workers/ai-interactive.worker'; +import { AiBackgroundJobWorker } from './workers/ai-background.worker'; import { AuditLogProcessor } from './modules/admin-audit-log/audit-log.processor'; import { FileCleanupProcessor } from './modules/files/file-cleanup.processor'; @@ -74,6 +76,8 @@ import appleConfig from './config/apple.config'; AiAnalysisWorker, DocumentImportWorker, NotificationWorker, + AiInteractiveJobWorker, + AiBackgroundJobWorker, AuditLogProcessor, FileCleanupProcessor, WorkerHeartbeatService, diff --git a/src/workers/ai-background.worker.ts b/src/workers/ai-background.worker.ts new file mode 100644 index 0000000..0c4cddd --- /dev/null +++ b/src/workers/ai-background.worker.ts @@ -0,0 +1,62 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Logger, Inject, Optional } from '@nestjs/common'; +import type { Job } from 'bullmq'; +import { + QUEUE_AI_BACKGROUND, +} from '../infrastructure/queue/queue.service'; +import { workerOptionsFor } from '../infrastructure/queue/queue-definitions'; +import { + AI_JOB_EXECUTION_ENGINE, +} from '../modules/ai-job/ai-job-execution-engine.interface'; + +/** + * M-AI-03-05: AiBackgroundJobWorker + * + * 薄 Processor — 职责仅限于： + * 1. 接收 BullMQ Job（payload 仅含 `{ jobId }`） + * 2. 注入 AiJobExecutionEngine + * 3. 调用 engine.execute(jobId, context) + * + * 禁止在 Processor 中：读取 Prompt、调用 Provider、写业务表、 + * 实现状态机、实现 Job Type switch。 + */ +@Processor(QUEUE_AI_BACKGROUND, workerOptionsFor(QUEUE_AI_BACKGROUND)) +export class AiBackgroundJobWorker extends WorkerHost { + private readonly logger = new Logger(AiBackgroundJobWorker.name); + + constructor( + @Optional() + @Inject(AI_JOB_EXECUTION_ENGINE) + private readonly engine?: any, + ) { + super(); + } + + async process(job: Job<{ jobId: string }>): Promise { + const { jobId } = job.data; + + if (!jobId) { + this.logger.error(`Invalid job payload: missing jobId`); + throw new Error('Invalid job payload: missing jobId'); + } + + if (!this.engine) { + // M-AI-03-05~07 阶段 Engine 尚未实现，暂记日志并标记完成 + this.logger.warn( + `Engine not available — skipping jobId=${jobId}. ` + + `Will be handled after M-AI-03-08 (#291).`, + ); + return; + } + + const context = { + jobId: job.id ?? '', + attemptMade: job.attemptsMade, + updateProgress: async (progress: number) => { + await job.updateProgress(progress); + }, + }; + + await this.engine.execute(jobId, context); + } +} diff --git a/src/workers/ai-interactive.worker.ts b/src/workers/ai-interactive.worker.ts new file mode 100644 index 0000000..9349f4a --- /dev/null +++ b/src/workers/ai-interactive.worker.ts @@ -0,0 +1,62 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Logger, Inject, Optional } from '@nestjs/common'; +import type { Job } from 'bullmq'; +import { + QUEUE_AI_INTERACTIVE, +} from '../infrastructure/queue/queue.service'; +import { workerOptionsFor } from '../infrastructure/queue/queue-definitions'; +import { + AI_JOB_EXECUTION_ENGINE, +} from '../modules/ai-job/ai-job-execution-engine.interface'; + +/** + * M-AI-03-05: AiInteractiveJobWorker + * + * 薄 Processor — 职责仅限于： + * 1. 接收 BullMQ Job（payload 仅含 `{ jobId }`） + * 2. 注入 AiJobExecutionEngine + * 3. 调用 engine.execute(jobId, context) + * + * 禁止在 Processor 中：读取 Prompt、调用 Provider、写业务表、 + * 实现状态机、实现 Job Type switch。 + */ +@Processor(QUEUE_AI_INTERACTIVE, workerOptionsFor(QUEUE_AI_INTERACTIVE)) +export class AiInteractiveJobWorker extends WorkerHost { + private readonly logger = new Logger(AiInteractiveJobWorker.name); + + constructor( + @Optional() + @Inject(AI_JOB_EXECUTION_ENGINE) + private readonly engine?: any, + ) { + super(); + } + + async process(job: Job<{ jobId: string }>): Promise { + const { jobId } = job.data; + + if (!jobId) { + this.logger.error(`Invalid job payload: missing jobId`); + throw new Error('Invalid job payload: missing jobId'); + } + + if (!this.engine) { + // M-AI-03-05~07 阶段 Engine 尚未实现，暂记日志并标记完成 + this.logger.warn( + `Engine not available — skipping jobId=${jobId}. ` + + `Will be handled after M-AI-03-08 (#291).`, + ); + return; + } + + const context = { + jobId: job.id ?? '', + attemptMade: job.attemptsMade, + updateProgress: async (progress: number) => { + await job.updateProgress(progress); + }, + }; + + await this.engine.execute(jobId, context); + } +}