feat: M-AI-03 事务化 AiJob 创建、Snapshot 与 Enqueue Outbox

- 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 <noreply@anthropic.com>

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>): 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',
+          }),
+        }),
+      );
+    });
+  });
+});

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);
+  }
+}

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。
+ * 当前包含：状态机、生命周期仓储、Registry、CreationService。
- * 后续 Issue（#288–#293）在此模块中逐步新增：
+ * 后续 Issue（#290–#293）逐步新增：OutboxDispatcher、Engine、Projector。
- *   - AiJobService
- *   - AiJobExecutionEngine
- *   - OutboxDispatcher
- *   - ResultProjector
  */
 @Module({
-  imports: [PrismaModule],
+  imports: [PrismaModule, AiRuntimeModule],
-  providers: [AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry],
+  providers: [
-  exports: [AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry],
+    AiJobStateMachine,
+    AiJobLifecycleRepository,
+    JobDefinitionRegistry,
+    OutboxRepository,
+    AiJobCreationService,
+  ],
+  exports: [
+    AiJobStateMachine,
+    AiJobLifecycleRepository,
+    JobDefinitionRegistry,
+    AiJobCreationService,
+  ],
 })
 export class AiJobModule {}