feat: M-AI-03 注册新队列 ai-interactive/ai-background 与薄 Worker Processor
Some checks failed
Deploy API Server / build-and-unit (push) Failing after 29s
Deploy API Server / current-integration (push) Has been skipped
Deploy API Server / backward-compat (push) Has been skipped
Deploy API Server / deploy (push) Has been skipped

- 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 <noreply@anthropic.com>
This commit is contained in:
wangdl 2026-06-20 18:02:26 +08:00
parent 373e5e09fa
commit fcce35742f
8 changed files with 227 additions and 1 deletions

View File

@ -5,6 +5,8 @@ import {
QUEUE_DOMAIN_EVENTS, QUEUE_DOMAIN_EVENTS,
QUEUE_AUDIT_LOG, QUEUE_AUDIT_LOG,
QUEUE_FILE_CLEANUP, QUEUE_FILE_CLEANUP,
QUEUE_AI_INTERACTIVE,
QUEUE_AI_BACKGROUND,
} from './queue.constants'; } from './queue.constants';
// ═══════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════
@ -98,6 +100,55 @@ export const QUEUE_DEFINITIONS: Record<string, QueueDefinition> = {
[QUEUE_DOMAIN_EVENTS]: buildDef(QUEUE_DOMAIN_EVENTS), [QUEUE_DOMAIN_EVENTS]: buildDef(QUEUE_DOMAIN_EVENTS),
[QUEUE_AUDIT_LOG]: buildDef(QUEUE_AUDIT_LOG), [QUEUE_AUDIT_LOG]: buildDef(QUEUE_AUDIT_LOG),
[QUEUE_FILE_CLEANUP]: buildDef(QUEUE_FILE_CLEANUP), [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 { export function queueDef(name: string): QueueDefinition {

View File

@ -4,3 +4,6 @@ export const QUEUE_NOTIFICATION = 'notification';
export const QUEUE_DOMAIN_EVENTS = 'domain-events'; export const QUEUE_DOMAIN_EVENTS = 'domain-events';
export const QUEUE_AUDIT_LOG = 'audit-logs'; export const QUEUE_AUDIT_LOG = 'audit-logs';
export const QUEUE_FILE_CLEANUP = 'file-cleanup'; 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';

View File

@ -2,7 +2,7 @@ import { Global, Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq'; import { BullModule } from '@nestjs/bullmq';
import { ConfigService } from '@nestjs/config'; import { ConfigService } from '@nestjs/config';
import { PrismaService } from '../database/prisma.service'; 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() @Global()
@Module({ @Module({
@ -31,6 +31,8 @@ import { QueueService, QUEUE_AI_ANALYSIS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, Q
{ name: QUEUE_DOMAIN_EVENTS }, { name: QUEUE_DOMAIN_EVENTS },
{ name: QUEUE_AUDIT_LOG }, { name: QUEUE_AUDIT_LOG },
{ name: QUEUE_FILE_CLEANUP }, { name: QUEUE_FILE_CLEANUP },
{ name: QUEUE_AI_INTERACTIVE },
{ name: QUEUE_AI_BACKGROUND },
), ),
], ],
providers: [QueueService, PrismaService], providers: [QueueService, PrismaService],

View File

@ -13,6 +13,8 @@ export {
QUEUE_DOMAIN_EVENTS, QUEUE_DOMAIN_EVENTS,
QUEUE_AUDIT_LOG, QUEUE_AUDIT_LOG,
QUEUE_FILE_CLEANUP, QUEUE_FILE_CLEANUP,
QUEUE_AI_INTERACTIVE,
QUEUE_AI_BACKGROUND,
} from './queue.constants'; } from './queue.constants';
import { import {
QUEUE_AI_ANALYSIS, QUEUE_AI_ANALYSIS,
@ -21,6 +23,8 @@ import {
QUEUE_DOMAIN_EVENTS, QUEUE_DOMAIN_EVENTS,
QUEUE_AUDIT_LOG, QUEUE_AUDIT_LOG,
QUEUE_FILE_CLEANUP, QUEUE_FILE_CLEANUP,
QUEUE_AI_INTERACTIVE,
QUEUE_AI_BACKGROUND,
} from './queue.constants'; } from './queue.constants';
@Injectable() @Injectable()
@ -35,6 +39,8 @@ export class QueueService {
@InjectQueue(QUEUE_DOMAIN_EVENTS) private readonly domainEventsQueue: Queue, @InjectQueue(QUEUE_DOMAIN_EVENTS) private readonly domainEventsQueue: Queue,
@InjectQueue(QUEUE_AUDIT_LOG) private readonly auditLogQueue: Queue, @InjectQueue(QUEUE_AUDIT_LOG) private readonly auditLogQueue: Queue,
@InjectQueue(QUEUE_FILE_CLEANUP) private readonly fileCleanupQueue: 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, @Optional() private readonly eventBus?: any,
) {} ) {}
@ -70,6 +76,8 @@ export class QueueService {
case QUEUE_DOMAIN_EVENTS: return this.domainEventsQueue; case QUEUE_DOMAIN_EVENTS: return this.domainEventsQueue;
case QUEUE_AUDIT_LOG: return this.auditLogQueue; case QUEUE_AUDIT_LOG: return this.auditLogQueue;
case QUEUE_FILE_CLEANUP: return this.fileCleanupQueue; 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}`); default: throw new Error(`Unknown queue: ${name}`);
} }
} }

View File

@ -0,0 +1,34 @@
/**
* M-AI-03: AiJobExecutionEngine Token
*
* Worker ProcessorAiInteractiveJobWorker / 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;
/** 更新进度0100 */
readonly updateProgress: (progress: number) => Promise<void>;
}
export interface AiJobExecutionEngine {
/**
* Job 线PREPARE RESOLVE EXECUTE PROJECT COMPLETE
*
* @param aiJobId AiJob
* @param context BullMQ Job
*/
execute(aiJobId: string, context: EngineJobContext): Promise<void>;
}
/** NestJS 注入 Token */
export const AI_JOB_EXECUTION_ENGINE = 'AI_JOB_EXECUTION_ENGINE';

View File

@ -12,6 +12,8 @@ import { LoggerModule } from './infrastructure/logger/logger.module';
import { AiAnalysisWorker } from './workers/ai-analysis.worker'; import { AiAnalysisWorker } from './workers/ai-analysis.worker';
import { DocumentImportWorker } from './workers/document-import.worker'; import { DocumentImportWorker } from './workers/document-import.worker';
import { NotificationWorker } from './workers/notification.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 { AuditLogProcessor } from './modules/admin-audit-log/audit-log.processor';
import { FileCleanupProcessor } from './modules/files/file-cleanup.processor'; import { FileCleanupProcessor } from './modules/files/file-cleanup.processor';
@ -74,6 +76,8 @@ import appleConfig from './config/apple.config';
AiAnalysisWorker, AiAnalysisWorker,
DocumentImportWorker, DocumentImportWorker,
NotificationWorker, NotificationWorker,
AiInteractiveJobWorker,
AiBackgroundJobWorker,
AuditLogProcessor, AuditLogProcessor,
FileCleanupProcessor, FileCleanupProcessor,
WorkerHeartbeatService, WorkerHeartbeatService,

View File

@ -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 Jobpayload `{ 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<void> {
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);
}
}

View File

@ -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 Jobpayload `{ 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<void> {
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);
}
}