diff --git a/src/modules/ai-job/ai-job-creation.service.spec.ts b/src/modules/ai-job/ai-job-creation.service.spec.ts index cf97d41..7a25e88 100644 --- a/src/modules/ai-job/ai-job-creation.service.spec.ts +++ b/src/modules/ai-job/ai-job-creation.service.spec.ts @@ -138,58 +138,44 @@ describe('AiJobCreationService', () => { }); describe('幂等性', () => { - it('相同 (userId, jobType, idempotencyKey) 返回已有 Job，不创建新记录', async () => { + it('并发重复创建相同 (userId, jobType, idempotencyKey) → P2002 → 返回已有 Job', async () => { registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue({}); - const existingJob = { - id: 'existing-job', - lifecycleStatus: 'succeeded', - jobType: 'test_job', - }; - prisma.aiJob.findUnique.mockResolvedValue(existingJob); + // First request creates successfully + lifecycleRepo.createJob.mockResolvedValueOnce({ id: 'job-001' }); + + await service.createJob({ + userId: 'u1', jobType: 'test_job', triggerType: 'user_api', + targetType: 'kb', targetId: 'kb-1', idempotencyKey: 'idem-001', + }); + expect(lifecycleRepo.createJob).toHaveBeenCalledTimes(1); + + // Second request: transaction fails with P2002 (duplicate unique key) + const p2002Error = new Error('Unique constraint failed') as any; + p2002Error.code = 'P2002'; + prisma.$transaction.mockRejectedValueOnce(p2002Error); + + const existingJob = { id: 'job-001', lifecycleStatus: 'queued', jobType: 'test_job' }; + prisma.aiJob.findUniqueOrThrow = jest.fn().mockResolvedValue(existingJob); const result = await service.createJob({ - userId: 'u1', - jobType: 'test_job', - triggerType: 'user_api', - targetType: 'kb', - targetId: 'kb-1', - idempotencyKey: 'idem-001', + userId: 'u1', jobType: 'test_job', triggerType: 'user_api', + targetType: 'kb', targetId: 'kb-1', idempotencyKey: 'idem-001', }); expect(result).toBe(existingJob); - // 不应调用 snapshot、lifecycleRepo、transaction - expect(snapshotBuilder.buildSnapshot).not.toHaveBeenCalled(); - expect(lifecycleRepo.createJob).not.toHaveBeenCalled(); - expect(prisma.$transaction).not.toHaveBeenCalled(); + // Snapshot was still built (pre-transaction), but no new Job created }); - it('未提供 idempotencyKey 时每次创建新 Job', async () => { + it('未提供 idempotencyKey 时每次创建新 Job（无幂等保护）', async () => { registry.get.mockReturnValue(validDef()); snapshotBuilder.buildSnapshot.mockResolvedValue({}); const mockJob = { id: 'job-002' }; lifecycleRepo.createJob.mockResolvedValue(mockJob); - // 第一次创建 - prisma.aiJob.findUnique.mockResolvedValue(null); - await service.createJob({ - userId: 'u1', - jobType: 'test_job', - triggerType: 'user_api', - targetType: 'kb', - targetId: 'kb-1', - // no idempotencyKey - }); - - // 第二次创建（无幂等键，应再次创建） - await service.createJob({ - userId: 'u1', - jobType: 'test_job', - triggerType: 'user_api', - targetType: 'kb', - targetId: 'kb-1', - // no idempotencyKey - }); + await service.createJob({ userId: 'u1', jobType: 'test_job', triggerType: 'user_api', targetType: 'kb', targetId: 'kb-1' }); + await service.createJob({ userId: 'u1', jobType: 'test_job', triggerType: 'user_api', targetType: 'kb', targetId: 'kb-1' }); expect(lifecycleRepo.createJob).toHaveBeenCalledTimes(2); }); diff --git a/src/modules/ai-job/ai-job-creation.service.ts b/src/modules/ai-job/ai-job-creation.service.ts index 454e033..3478e5c 100644 --- a/src/modules/ai-job/ai-job-creation.service.ts +++ b/src/modules/ai-job/ai-job-creation.service.ts @@ -1,6 +1,7 @@ import { Injectable, Logger, BadRequestException } from '@nestjs/common'; +import * as crypto from 'crypto'; import { PrismaService } from '../../infrastructure/database/prisma.service'; -import { OutboxRepository, CreateOutboxInput } from '../../infrastructure/outbox/outbox.repository'; +import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; import { SnapshotBuilderService } from '../ai-runtime/snapshot-builder.service'; import { AiJobLifecycleRepository, CreateJobInput } from './ai-job-lifecycle.repository'; import { JobDefinitionRegistry } from './job-definition-registry'; @@ -63,32 +64,12 @@ export class AiJobCreationService { // 1. Resolve Definition（fail-fast：未知 JobType 在此拦截） const def = this.registry.get(input.jobType); - // 2. Validate input schema version consistency + // 2. Validate triggerType（TypeScript 编译期 + 运行时双重保护） if (input.triggerType && !['user_api', 'admin_manual', 'system_scheduled', 'parent_job'].includes(input.triggerType)) { throw new BadRequestException(`Invalid triggerType: ${input.triggerType}`); } - // 3. Idempotency check（fast-return before snapshot build） - if (input.idempotencyKey) { - const existing = await this.prisma.aiJob.findUnique({ - where: { - userId_jobType_idempotencyKey: { - userId: input.userId, - jobType: input.jobType, - idempotencyKey: input.idempotencyKey, - }, - }, - }); - if (existing) { - this.logger.log( - `Idempotent return: jobId=${existing.id} ` + - `key=${input.idempotencyKey}`, - ); - return existing; - } - } - - // 4. Build snapshot（事务外：依赖外部 DB 读取，不应占用事务时间） + // 3. Build snapshot（事务外：依赖外部 DB 读取，不应占用事务时间） const snapshot = await this.snapshotBuilder.buildSnapshot( input.userId, input.targetType, @@ -97,9 +78,9 @@ export class AiJobCreationService { const contentHash = this.computeHash(JSON.stringify(snapshot)); - // 5. Transaction: Job + Snapshot + Outbox - const job = await this.prisma.$transaction(async (tx) => { - // 5a. Create AiJob + // 4. Transaction: Job + Snapshot + Outbox（含幂等保护） + const createFn = async (tx: Prisma.TransactionClient) => { + // 4a. Create AiJob const jobInput: CreateJobInput = { userId: input.userId, jobType: input.jobType, @@ -114,6 +95,7 @@ export class AiJobCreationService { credentialId: input.overrides?.credentialId, modelProvider: def.model.modelProvider, modelName: def.model.modelName, + promptKey: def.prompt.promptKey, promptVersion: def.prompt.promptVersion, outputSchemaVersion: def.output.schemaVersion, inputSchemaVersion: def.input.schemaVersion, @@ -123,7 +105,7 @@ export class AiJobCreationService { const job = await this.lifecycleRepo.createJob(tx, jobInput); - // 5b. Create AiJobSnapshot + // 4b. Create AiJobSnapshot await tx.aiJobSnapshot.create({ data: { jobId: job.id, @@ -135,7 +117,7 @@ export class AiJobCreationService { }, }); - // 5c. Create OutboxEvent + // 4c. Create OutboxEvent const dedupeKey = `ai.job.enqueue:${job.id}`; await this.outboxRepo.createInTransaction(tx, { eventType: 'ai.job.enqueue', @@ -146,21 +128,41 @@ export class AiJobCreationService { availableAt: new Date(), }); + return job; // job 在事务成功路径中使用，P2002 catch 路径中不用 + }; + + try { + const job = await this.prisma.$transaction(createFn); + this.logger.log( + `Job created: id=${job.id} type=${input.jobType} ` + + `queue=${def.queue.queueName} snapshotHash=${contentHash}`, + ); return job; - }); - - this.logger.log( - `Job created: id=${job.id} type=${input.jobType} ` + - `queue=${def.queue.queueName} snapshotHash=${contentHash}`, - ); - - return job; + } catch (err: any) { + // 幂等：并发重复创建 → UNIQUE 冲突（P2002）→ 返回已有 Job + if (err?.code === 'P2002' && input.idempotencyKey) { + const existing = await this.prisma.aiJob.findUniqueOrThrow({ + where: { + userId_jobType_idempotencyKey: { + userId: input.userId, + jobType: input.jobType, + idempotencyKey: input.idempotencyKey, + }, + }, + }); + this.logger.log( + `Idempotent return (P2002): jobId=${existing.id} ` + + `key=${input.idempotencyKey}`, + ); + return existing; + } + throw err; + } } // ── helpers ── private computeHash(content: string): string { - const crypto = require('crypto'); return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16); } } diff --git a/src/modules/ai-job/ai-job-lifecycle.repository.ts b/src/modules/ai-job/ai-job-lifecycle.repository.ts index aeee9cc..d35b44c 100644 --- a/src/modules/ai-job/ai-job-lifecycle.repository.ts +++ b/src/modules/ai-job/ai-job-lifecycle.repository.ts @@ -31,6 +31,7 @@ export interface CreateJobInput { credentialId?: string; modelProvider?: string; modelName?: string; + promptKey?: string; promptVersion?: string; outputSchemaVersion?: string; inputSchemaVersion?: string; @@ -98,6 +99,7 @@ export class AiJobLifecycleRepository { // 若未传入，依赖 Prisma schema @default 保证列非空。 ...(input.modelProvider ? { modelProvider: input.modelProvider } : {}), ...(input.modelName ? { modelName: input.modelName } : {}), + promptKey: input.promptKey ?? null, promptVersion: input.promptVersion ?? null, outputSchemaVersion: input.outputSchemaVersion ?? null, inputSchemaVersion: input.inputSchemaVersion ?? null,