import { Injectable, Logger, Inject, Optional } from '@nestjs/common'; import { PrismaService } from '../../infrastructure/database/prisma.service'; import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; import { AiJobStateMachine } from './ai-job-state-machine'; import { ResultProjector, ProjectionContext, ArtifactReference, RESULT_PROJECTORS } from './result-projector.interface'; import { JobAlreadyTerminalError } from './ai-job.errors'; /** * M-AI-03-09: Projection Executor * * 在 Prisma Transaction 内原子执行: * 1. ResultProjector.project(tx, context) → 业务实体 * 2. AiJobArtifact 写入(project 内部已做,此处验证) * 3. AiJobLifecycleRepository.markSucceeded(tx) → 原子标记成功 * * 幂等:Job 已 succeeded → 直接返回(不重复执行 Projector)。 */ @Injectable() export class ProjectionExecutor { private readonly logger = new Logger(ProjectionExecutor.name); private readonly projectorMap = new Map(); constructor( private readonly prisma: PrismaService, private readonly lifecycleRepo: AiJobLifecycleRepository, private readonly stateMachine: AiJobStateMachine, @Optional() @Inject(RESULT_PROJECTORS) projectors?: ResultProjector[], ) { if (projectors) { for (const p of projectors) { if (this.projectorMap.has(p.key)) { this.logger.warn(`Duplicate projector key "${p.key}" — overwriting`); } this.projectorMap.set(p.key, p); } } this.logger.log( `ProjectionExecutor initialized with ${this.projectorMap.size} projector(s): ` + `[${Array.from(this.projectorMap.keys()).join(', ') || 'none'}]`, ); } /** * 执行 Projector。 * * @param projectorKey JobDefinition.projectorKey * @param context Projection 上下文 * @returns Artifact 引用列表 */ async execute( projectorKey: string | undefined, context: ProjectionContext, ): Promise { if (!projectorKey) { this.logger.log(`No projectorKey for jobType=${context.job.jobType} — marking succeeded without projection`); await this.lifecycleRepo.markSucceeded(context.job.id); return []; } const projector = this.projectorMap.get(projectorKey); if (!projector) { // Projector 未注册 — 仍应标记 Job 成功(Engine 的 AI 调用已完成) this.logger.warn(`Projector "${projectorKey}" not registered — marking succeeded without projection for job=${context.job.id}`); await this.lifecycleRepo.markSucceeded(context.job.id); return []; } // 幂等检查:Job 已终态则跳过 const currentJob = await this.prisma.aiJob.findUnique({ where: { id: context.job.id }, select: { lifecycleStatus: true }, }); if (currentJob && this.stateMachine.isTerminal(currentJob.lifecycleStatus as any)) { this.logger.warn( `Job ${context.job.id} already terminal (${currentJob.lifecycleStatus}) — skipping projector`, ); // 返回已有 Artifact 引用 const existing = await this.prisma.aiJobArtifact.findMany({ where: { jobId: context.job.id }, orderBy: { ordinal: 'asc' }, }); return existing.map((a) => ({ artifactType: a.artifactType, artifactId: a.artifactId, ordinal: a.ordinal, })); } // 事务内:Projector + markSucceeded const artifacts = await this.prisma.$transaction(async (tx) => { const result = await projector.project(tx, context); // 验证 Artifact 引用的完整性 if (result.length > 0) { const dbArtifacts = await tx.aiJobArtifact.findMany({ where: { jobId: context.job.id }, }); if (dbArtifacts.length < result.length) { throw new Error( `Artifact mismatch: projector returned ${result.length} but only ${dbArtifacts.length} in DB`, ); } } // 标记成功(事务内,与 Projector 原子提交) await this.lifecycleRepo.markSucceeded(context.job.id, tx); return result; }); this.logger.log( `Projection complete: job=${context.job.id} projector=${projectorKey} artifacts=${artifacts.length}`, ); return artifacts; } }