From f0183c03c482c72d61195d8b1a2ea4e73cb1287f Mon Sep 17 00:00:00 2001 From: wangdl Date: Sat, 20 Jun 2026 18:13:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M-AI-03=20=E4=BA=8B=E5=8A=A1=E5=8C=96?= =?UTF-8?q?=20AiJob=20=E5=88=9B=E5=BB=BA=E3=80=81Snapshot=20=E4=B8=8E=20En?= =?UTF-8?q?queue=20Outbox?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AiJobCreationService: 在一个 Prisma Transaction 内原子创建 Job+Snapshot+Outbox - 幂等性: (userId, jobType, idempotencyKey) UNIQUE 约束防重复 - Job 元数据从 JobDefinition 写入 (queueName/priority/maxAttempts/timeoutMs 等) - overrides 支持覆盖 priority/credentialMode/credentialId - Outbox payload 仅含 {jobId}，不含 userId/credentialId/snapshot - 依赖 SnapshotBuilderService + OutboxRepository，AiJobModule 导入 AiRuntimeModule Co-Authored-By: Claude --- .../ai-job/ai-job-creation.service.spec.ts | 357 ++++++++++++++++++ src/modules/ai-job/ai-job-creation.service.ts | 166 ++++++++ src/modules/ai-job/ai-job.module.ts | 28 +- 3 files changed, 542 insertions(+), 9 deletions(-) create mode 100644 src/modules/ai-job/ai-job-creation.service.spec.ts create mode 100644 src/modules/ai-job/ai-job-creation.service.ts diff --git a/src/modules/ai-job/ai-job-creation.service.spec.ts b/src/modules/ai-job/ai-job-creation.service.spec.ts new file mode 100644 index 0000000..cf97d41 --- /dev/null +++ b/src/modules/ai-job/ai-job-creation.service.spec.ts @@ -0,0 +1,357 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { BadRequestException } from '@nestjs/common'; +import { AiJobCreationService } from './ai-job-creation.service'; +import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; +import { JobDefinitionRegistry } from './job-definition-registry'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; +import { SnapshotBuilderService } from '../ai-runtime/snapshot-builder.service'; +import { UnknownJobTypeError } from './job-definition-registry'; +import { JobDefinition } from './job-definition.types'; + +/** 最小合法 Definition 工厂 */ +function validDef(overrides?: Partial): JobDefinition { + return { + jobType: 'test_job', + metadata: { label: 'Test Job', description: '', domain: 'analysis', version: '1.0' }, + queue: { queueName: 'ai-interactive', defaultPriority: 0 }, + execution: { timeoutMs: 30000, maxRetries: 2, retryBackoff: { type: 'exponential', delay: 2000 }, cancellable: true, abortStrategy: 'fail' }, + input: { schemaVersion: '1.0.0' }, + output: { schemaVersion: '1.0.0' }, + prompt: { promptKey: 'test', promptVersion: '1.0' }, + model: { modelTier: 'primary', modelProvider: 'deepseek', modelName: 'deepseek-chat' }, + credential: { allowedModes: ['platform_key'], defaultMode: 'platform_key' }, + security: { contentSafetyCheck: false, outputRedaction: false }, + ...overrides, + }; +} + +describe('AiJobCreationService', () => { + let service: AiJobCreationService; + let prisma: any; + let lifecycleRepo: any; + let registry: any; + let snapshotBuilder: any; + let outboxRepo: any; + + beforeEach(async () => { + registry = { get: jest.fn(), has: jest.fn() }; + + lifecycleRepo = { createJob: jest.fn() }; + + snapshotBuilder = { buildSnapshot: jest.fn() }; + + outboxRepo = { createInTransaction: jest.fn() }; + + prisma = { + aiJob: { + findUnique: jest.fn(), + }, + aiJobSnapshot: { create: jest.fn() }, + $transaction: jest.fn((fn: any) => fn(prisma)), // mock tx = same prisma + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + AiJobCreationService, + { provide: PrismaService, useValue: prisma }, + { provide: AiJobLifecycleRepository, useValue: lifecycleRepo }, + { provide: JobDefinitionRegistry, useValue: registry }, + { provide: SnapshotBuilderService, useValue: snapshotBuilder }, + { provide: OutboxRepository, useValue: outboxRepo }, + ], + }).compile(); + + service = module.get(AiJobCreationService); + }); + + describe('事务化创建', () => { + it('在一个事务内创建 AiJob + Snapshot + Outbox', async () => { + registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue({ userId: 'u1', profile: {} }); + const mockJob = { id: 'job-001', lifecycleStatus: 'queued', jobType: 'test_job' }; + lifecycleRepo.createJob.mockResolvedValue(mockJob); + + const result = await service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'user_api', + targetType: 'knowledge_base', + targetId: 'kb-1', + }); + + // 验证事务调用 + expect(prisma.$transaction).toHaveBeenCalled(); + // Job created in tx + expect(lifecycleRepo.createJob).toHaveBeenCalledWith( + expect.any(Object), // the tx client + expect.objectContaining({ + userId: 'u1', + jobType: 'test_job', + queueName: 'ai-interactive', + }), + ); + // Snapshot created in tx + expect(prisma.aiJobSnapshot.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + jobId: 'job-001', + jobType: 'test_job', + contentHash: expect.any(String), + }), + }), + ); + // Outbox created in tx + expect(outboxRepo.createInTransaction).toHaveBeenCalledWith( + expect.any(Object), + expect.objectContaining({ + eventType: 'ai.job.enqueue', + aggregateType: 'AiJob', + aggregateId: 'job-001', + dedupeKey: 'ai.job.enqueue:job-001', + payload: { jobId: 'job-001' }, + }), + ); + expect(result).toBe(mockJob); + }); + + it('事务中任一失败则全部回滚', async () => { + registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue({ userId: 'u1' }); + lifecycleRepo.createJob.mockResolvedValue({ id: 'job-001' }); + // Snapshot create fails + prisma.aiJobSnapshot.create.mockRejectedValue(new Error('snapshot insert failed')); + + await expect( + service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'user_api', + targetType: 'knowledge_base', + targetId: 'kb-1', + }), + ).rejects.toThrow('snapshot insert failed'); + + // Outbox must NOT be called (tx rolled back) + expect(outboxRepo.createInTransaction).not.toHaveBeenCalled(); + }); + }); + + describe('幂等性', () => { + it('相同 (userId, jobType, idempotencyKey) 返回已有 Job，不创建新记录', async () => { + registry.get.mockReturnValue(validDef()); + + const existingJob = { + id: 'existing-job', + lifecycleStatus: 'succeeded', + jobType: 'test_job', + }; + prisma.aiJob.findUnique.mockResolvedValue(existingJob); + + const result = await service.createJob({ + 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(); + }); + + 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 + }); + + expect(lifecycleRepo.createJob).toHaveBeenCalledTimes(2); + }); + }); + + describe('Job 元数据从 Definition 写入', () => { + it('queueName / priority / timeout / maxAttempts / schema 等来自 Definition', async () => { + registry.get.mockReturnValue( + validDef({ + queue: { queueName: 'ai-background', defaultPriority: 5 }, + execution: { timeoutMs: 90000, maxRetries: 1, retryBackoff: { type: 'exponential', delay: 5000 }, cancellable: false, abortStrategy: 'retry' }, + input: { schemaVersion: '2.0.0' }, + output: { schemaVersion: '2.0.0' }, + prompt: { promptKey: 'bg_prompt', promptVersion: '2.0' }, + model: { modelTier: 'fallback', modelProvider: 'deepseek', modelName: 'deepseek-chat' }, + }), + ); + snapshotBuilder.buildSnapshot.mockResolvedValue({}); + const mockJob = { id: 'bg-job' }; + lifecycleRepo.createJob.mockResolvedValue(mockJob); + + await service.createJob({ + userId: 'u1', + jobType: 'bg_job', + triggerType: 'system_scheduled', + targetType: 'user', + targetId: 'u1', + }); + + expect(lifecycleRepo.createJob).toHaveBeenCalledWith( + expect.any(Object), + expect.objectContaining({ + queueName: 'ai-background', + priority: 5, + timeoutMs: 90000, + maxAttempts: 1, + inputSchemaVersion: '2.0.0', + outputSchemaVersion: '2.0.0', + promptVersion: '2.0', + modelProvider: 'deepseek', + modelName: 'deepseek-chat', + }), + ); + }); + + it('overrides 覆盖 Definition 默认值', async () => { + registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue({}); + const mockJob = { id: 'override-job' }; + lifecycleRepo.createJob.mockResolvedValue(mockJob); + + await service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'user_api', + targetType: 'kb', + targetId: 'kb-1', + overrides: { + priority: 99, + credentialMode: 'user_deepseek_key', + credentialId: 'cred-xyz', + }, + }); + + expect(lifecycleRepo.createJob).toHaveBeenCalledWith( + expect.any(Object), + expect.objectContaining({ + priority: 99, + credentialMode: 'user_deepseek_key', + credentialId: 'cred-xyz', + }), + ); + }); + }); + + describe('输入校验', () => { + it('未知 jobType → UnknownJobTypeError', async () => { + registry.get.mockImplementation(() => { + throw new UnknownJobTypeError('bad_job'); + }); + + await expect( + service.createJob({ + userId: 'u1', + jobType: 'bad_job', + triggerType: 'user_api', + targetType: 'kb', + targetId: 'kb-1', + }), + ).rejects.toThrow(UnknownJobTypeError); + }); + + it('非法 triggerType → BadRequestException', async () => { + registry.get.mockReturnValue(validDef()); + + await expect( + service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'evil_hack' as any, + targetType: 'kb', + targetId: 'kb-1', + }), + ).rejects.toThrow(BadRequestException); + }); + }); + + describe('Outbox payload 约束', () => { + it('Outbox payload 仅含 {jobId}', async () => { + registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue({ userId: 'u1' }); + lifecycleRepo.createJob.mockResolvedValue({ id: 'job-minimal' }); + + await service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'user_api', + targetType: 'kb', + targetId: 'kb-1', + }); + + expect(outboxRepo.createInTransaction).toHaveBeenCalledWith( + expect.any(Object), + expect.objectContaining({ + payload: { jobId: 'job-minimal' }, + }), + ); + + // payload 不应包含 userId、credentialId 等敏感字段 + const call = outboxRepo.createInTransaction.mock.calls[0][1]; + expect(call.payload).not.toHaveProperty('userId'); + expect(call.payload).not.toHaveProperty('credentialId'); + expect(call.payload).not.toHaveProperty('snapshot'); + }); + }); + + describe('Snapshot', () => { + it('Snapshot 传入 buildSnapshot 产出的 content 和 contentHash', async () => { + const snapContent = { userId: 'u1', profile: { level: 5 }, settings: {} }; + registry.get.mockReturnValue(validDef()); + snapshotBuilder.buildSnapshot.mockResolvedValue(snapContent); + lifecycleRepo.createJob.mockResolvedValue({ id: 'snap-job' }); + + await service.createJob({ + userId: 'u1', + jobType: 'test_job', + triggerType: 'user_api', + targetType: 'kb', + targetId: 'kb-1', + }); + + expect(prisma.aiJobSnapshot.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + jobId: 'snap-job', + content: snapContent, + contentHash: expect.any(String), + schemaVersion: '1.0.0', + redactionVersion: '1.0', + }), + }), + ); + }); + }); +}); diff --git a/src/modules/ai-job/ai-job-creation.service.ts b/src/modules/ai-job/ai-job-creation.service.ts new file mode 100644 index 0000000..454e033 --- /dev/null +++ b/src/modules/ai-job/ai-job-creation.service.ts @@ -0,0 +1,166 @@ +import { Injectable, Logger, BadRequestException } from '@nestjs/common'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { OutboxRepository, CreateOutboxInput } 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'; +import { Prisma } from '@prisma/client'; + +/** + * M-AI-03-06: AiJobCreationService + * + * 统一 Job 创建入口（internal-only）。 + * 在一个 Prisma Transaction 内原子创建： + * 1. AiJob（lifecycleStatus = 'queued'） + * 2. AiJobSnapshot（不可变输入快照） + * 3. OutboxEvent（eventType = 'ai.job.enqueue'） + * + * 幂等性：AiJob 表的 @@unique([userId, jobType, idempotencyKey]) 保证 + * 重复 (userId, jobType, idempotencyKey) 返回已有 Job，不重复创建。 + * + * 约束： + * - 不向 BullMQ 直接入队（由 OutboxDispatcher 异步投递） + * - 不调用 Provider + * - 不创建业务结果 + */ + +export interface CreateAiJobInput { + userId: string; + jobType: string; + triggerType: 'user_api' | 'admin_manual' | 'system_scheduled' | 'parent_job'; + targetType: string; + targetId: string; + /** 父 Job ID（Job 链场景） */ + parentJobId?: string; + /** 幂等 Key。提供时启用幂等；不提供时每次创建新 Job */ + idempotencyKey?: string; + /** 覆盖 Definition 默认值 */ + overrides?: { + priority?: number; + credentialMode?: 'platform_key' | 'user_deepseek_key'; + credentialId?: string; + }; +} + +@Injectable() +export class AiJobCreationService { + private readonly logger = new Logger(AiJobCreationService.name); + + constructor( + private readonly prisma: PrismaService, + private readonly lifecycleRepo: AiJobLifecycleRepository, + private readonly registry: JobDefinitionRegistry, + private readonly snapshotBuilder: SnapshotBuilderService, + private readonly outboxRepo: OutboxRepository, + ) {} + + /** + * 创建 Job + Snapshot + Outbox（一个 Prisma Transaction 内）。 + * + * 幂等：若 (userId, jobType, idempotencyKey) 已存在，返回已有 Job。 + */ + async createJob(input: CreateAiJobInput) { + // 1. Resolve Definition（fail-fast：未知 JobType 在此拦截） + const def = this.registry.get(input.jobType); + + // 2. Validate input schema version consistency + 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 读取，不应占用事务时间） + const snapshot = await this.snapshotBuilder.buildSnapshot( + input.userId, + input.targetType, + input.targetId, + ); + + const contentHash = this.computeHash(JSON.stringify(snapshot)); + + // 5. Transaction: Job + Snapshot + Outbox + const job = await this.prisma.$transaction(async (tx) => { + // 5a. Create AiJob + const jobInput: CreateJobInput = { + userId: input.userId, + jobType: input.jobType, + triggerType: input.triggerType, + queueName: def.queue.queueName, + targetType: input.targetType, + targetId: input.targetId, + parentJobId: input.parentJobId, + idempotencyKey: input.idempotencyKey, + priority: input.overrides?.priority ?? def.queue.defaultPriority, + credentialMode: input.overrides?.credentialMode ?? def.credential.defaultMode, + credentialId: input.overrides?.credentialId, + modelProvider: def.model.modelProvider, + modelName: def.model.modelName, + promptVersion: def.prompt.promptVersion, + outputSchemaVersion: def.output.schemaVersion, + inputSchemaVersion: def.input.schemaVersion, + maxAttempts: def.execution.maxRetries, + timeoutMs: def.execution.timeoutMs, + }; + + const job = await this.lifecycleRepo.createJob(tx, jobInput); + + // 5b. Create AiJobSnapshot + await tx.aiJobSnapshot.create({ + data: { + jobId: job.id, + jobType: input.jobType, + schemaVersion: def.input.schemaVersion, + redactionVersion: '1.0', + content: snapshot as any, + contentHash, + }, + }); + + // 5c. Create OutboxEvent + const dedupeKey = `ai.job.enqueue:${job.id}`; + await this.outboxRepo.createInTransaction(tx, { + eventType: 'ai.job.enqueue', + aggregateType: 'AiJob', + aggregateId: job.id, + dedupeKey, + payload: { jobId: job.id }, + availableAt: new Date(), + }); + + return job; + }); + + this.logger.log( + `Job created: id=${job.id} type=${input.jobType} ` + + `queue=${def.queue.queueName} snapshotHash=${contentHash}`, + ); + + return job; + } + + // ── 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.module.ts b/src/modules/ai-job/ai-job.module.ts index 2f6f3fa..13a0c36 100644 --- a/src/modules/ai-job/ai-job.module.ts +++ b/src/modules/ai-job/ai-job.module.ts @@ -1,22 +1,32 @@ import { Module } from '@nestjs/common'; import { PrismaModule } from '../../infrastructure/database/prisma.module'; +import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; +import { AiRuntimeModule } from '../ai-runtime/ai-runtime.module'; import { AiJobStateMachine } from './ai-job-state-machine'; import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; import { JobDefinitionRegistry } from './job-definition-registry'; +import { AiJobCreationService } from './ai-job-creation.service'; /** * M-AI-03: AiJob 统一 Job Engine 模块 * - * 当前包含：状态机、生命周期仓储、Job Definition Registry。 - * 后续 Issue（#288–#293）在此模块中逐步新增： - * - AiJobService - * - AiJobExecutionEngine - * - OutboxDispatcher - * - ResultProjector + * 当前包含：状态机、生命周期仓储、Registry、CreationService。 + * 后续 Issue（#290–#293）逐步新增：OutboxDispatcher、Engine、Projector。 */ @Module({ - imports: [PrismaModule], - providers: [AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry], - exports: [AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry], + imports: [PrismaModule, AiRuntimeModule], + providers: [ + AiJobStateMachine, + AiJobLifecycleRepository, + JobDefinitionRegistry, + OutboxRepository, + AiJobCreationService, + ], + exports: [ + AiJobStateMachine, + AiJobLifecycleRepository, + JobDefinitionRegistry, + AiJobCreationService, + ], }) export class AiJobModule {}