feat: M-AI-03 Outbox Dispatcher — 可靠投递 ai.job.enqueue 到 BullMQ

- OutboxDispatcher: 每 5s 轮询 OutboxEvent 表，CAS 领取 → 验证 → BullMQ 投递 → 标记 published
- 采用 ADR-003 §4.3 先 CAS 后投递算法：markProcessing(CAS) → queue.add() → markPublished
- BullMQ jobId = AiJob.id（幂等重复 add）
- 崩溃窗口保护：bullmq jobId 幂等 + releaseExpiredLocks(60s)
- 错误处理：永久失败（未知eventType/Job不存在/非法queueName/超过最大尝试）
  vs 暂时失败（exponential backoff 回退 pending）
- 仅在 WorkerModule 注册（API 进程不运行 Dispatcher）

Co-Authored-By: Claude <noreply@anthropic.com>

src/modules/ai-job/ai-job.module.ts

@@ -11,7 +11,8 @@ import { AiJobCreationService } from './ai-job-creation.service';
  * M-AI-03: AiJob 统一 Job Engine 模块
  *
  * 当前包含：状态机、生命周期仓储、Registry、CreationService。
- * 后续 Issue（#290–#293）逐步新增：OutboxDispatcher、Engine、Projector。
+ * OutboxDispatcher 仅在 WorkerModule 注册（API 进程不运行）。
+ * 后续 Issue（#291–#293）逐步新增：Engine、Projector。
  */
 @Module({
   imports: [PrismaModule, AiRuntimeModule],

src/modules/ai-job/outbox-dispatcher.service.spec.ts

@@ -0,0 +1,278 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { OutboxDispatcher } from './outbox-dispatcher.service';
+import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository';
+import { QueueService } from '../../infrastructure/queue/queue.service';
+import { PrismaService } from '../../infrastructure/database/prisma.service';
+
+describe('OutboxDispatcher', () => {
+  let dispatcher: OutboxDispatcher;
+  let outboxRepo: any;
+  let queueService: any;
+  let prisma: any;
+
+  function makeEvent(overrides?: any) {
+    return {
+      id: 'evt-001',
+      eventType: 'ai.job.enqueue',
+      aggregateType: 'AiJob',
+      aggregateId: 'job-001',
+      dedupeKey: 'ai.job.enqueue:job-001',
+      payload: { jobId: 'job-001' },
+      status: 'pending',
+      attemptCount: 0,
+      availableAt: new Date(),
+      lockedAt: null,
+      lockedBy: null,
+      lastErrorCode: null,
+      lastErrorMessage: null,
+      ...overrides,
+    };
+  }
+
+  beforeEach(async () => {
+    outboxRepo = {
+      findDispatchable: jest.fn(),
+      markProcessing: jest.fn(),
+      markPublished: jest.fn(),
+      markFailed: jest.fn(),
+      releaseExpiredLocks: jest.fn(),
+    };
+
+    queueService = { add: jest.fn() };
+
+    prisma = {
+      aiJob: { findUnique: jest.fn() },
+      outboxEvent: { update: jest.fn() },
+    };
+
+    // Suppress NestJS logger noise in tests
+    jest.spyOn(require('@nestjs/common').Logger.prototype, 'log').mockImplementation(() => {});
+    jest.spyOn(require('@nestjs/common').Logger.prototype, 'warn').mockImplementation(() => {});
+    jest.spyOn(require('@nestjs/common').Logger.prototype, 'error').mockImplementation(() => {});
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        OutboxDispatcher,
+        { provide: OutboxRepository, useValue: outboxRepo },
+        { provide: QueueService, useValue: queueService },
+        { provide: PrismaService, useValue: prisma },
+      ],
+    }).compile();
+
+    dispatcher = module.get(OutboxDispatcher);
+    // Don't start the interval
+  });
+
+  describe('正常分发', () => {
+    it('领取 → CAS 抢锁 → 验证 → 投递 BullMQ → 标记 published', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing', lockedBy: 'x' }));
+      prisma.aiJob.findUnique.mockResolvedValue({
+        id: 'job-001',
+        queueName: 'ai-interactive',
+        lifecycleStatus: 'queued',
+      });
+      queueService.add.mockResolvedValue({ id: 'job-001' });
+      outboxRepo.markPublished.mockResolvedValue(undefined);
+
+      const result = await dispatcher.cycle();
+      expect(result.published).toBe(1);
+      expect(result.skipped).toBe(0);
+
+      // Step order: CAS first, then enqueue
+      expect(outboxRepo.markProcessing).toHaveBeenCalledWith('evt-001', expect.any(String));
+      expect(queueService.add).toHaveBeenCalledWith(
+        'ai-interactive',
+        { jobId: 'job-001' },
+        { jobId: 'job-001' }, // BullMQ jobId = AiJob.id
+      );
+      expect(outboxRepo.markPublished).toHaveBeenCalledWith('evt-001');
+    });
+  });
+
+  describe('CAS 并发安全', () => {
+    it('markProcessing 返回 null → skip（被其他 Dispatcher 抢走）', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(null); // CAS failed
+
+      const result = await dispatcher.cycle();
+      expect(result.skipped).toBe(1);
+      expect(result.published).toBe(0);
+      // 不应投递 BullMQ
+      expect(queueService.add).not.toHaveBeenCalled();
+    });
+  });
+
+  describe('幂等 BullMQ 入队', () => {
+    it('相同 jobId 重复 add → BullMQ 视为 no-op（不抛错）', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing' }));
+      prisma.aiJob.findUnique.mockResolvedValue({
+        id: 'job-001',
+        queueName: 'ai-interactive',
+        lifecycleStatus: 'queued',
+      });
+      // BullMQ 同 jobId repeat add → no error
+      queueService.add.mockResolvedValue({ id: 'job-001' });
+      outboxRepo.markPublished.mockResolvedValue(undefined);
+
+      // 第一次
+      await dispatcher.cycle();
+      // 模拟第二次（相同 event 被 releaseExpiredLocks 恢复）
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing' }));
+      await dispatcher.cycle();
+
+      // 两次都成功 published
+      expect(outboxRepo.markPublished).toHaveBeenCalledTimes(2);
+    });
+  });
+
+  describe('锁超时恢复', () => {
+    it('releaseExpiredLocks 在每轮开始调用', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(3); // 3 expired locks released
+      outboxRepo.findDispatchable.mockResolvedValue([]);
+
+      await dispatcher.cycle();
+      expect(outboxRepo.releaseExpiredLocks).toHaveBeenCalledWith(60_000);
+    });
+  });
+
+  describe('未知 eventType', () => {
+    it('非 ai.job.enqueue → 永久 failed', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([
+        makeEvent({ eventType: 'unknown.event', id: 'evt-bad' }),
+      ]);
+      outboxRepo.markProcessing.mockResolvedValue(
+        makeEvent({ eventType: 'unknown.event', id: 'evt-bad' }),
+      );
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+      expect(outboxRepo.markFailed).toHaveBeenCalledWith(
+        'evt-bad',
+        'UNKNOWN_EVENT_TYPE',
+        expect.stringContaining('unknown.event'),
+      );
+      expect(queueService.add).not.toHaveBeenCalled();
+    });
+  });
+
+  describe('Job 不存在', () => {
+    it('AiJob 已删除 → 永久 failed', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent());
+      prisma.aiJob.findUnique.mockResolvedValue(null); // Job not found
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+      expect(outboxRepo.markFailed).toHaveBeenCalledWith(
+        'evt-001',
+        'JOB_NOT_FOUND',
+        expect.any(String),
+      );
+    });
+  });
+
+  describe('非法 queueName', () => {
+    it('queueName 不在允许列表 → 永久 failed', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent());
+      prisma.aiJob.findUnique.mockResolvedValue({
+        id: 'job-001',
+        queueName: 'ai-analysis', // 旧队列，不允许
+        lifecycleStatus: 'queued',
+      });
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+      expect(outboxRepo.markFailed).toHaveBeenCalledWith(
+        'evt-001',
+        'ILLEGAL_QUEUE_NAME',
+        expect.stringContaining('ai-analysis'),
+      );
+    });
+  });
+
+  describe('BullMQ 入队失败回退', () => {
+    it('入队抛错 → 回退到 pending（未达最大尝试）', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ attemptCount: 0 }));
+      prisma.aiJob.findUnique.mockResolvedValue({
+        id: 'job-001',
+        queueName: 'ai-interactive',
+        lifecycleStatus: 'queued',
+      });
+      queueService.add.mockRejectedValue(new Error('Redis connection lost'));
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+
+      // 应重置为 pending（退避重试）
+      expect(prisma.outboxEvent.update).toHaveBeenCalledWith(
+        expect.objectContaining({
+          where: { id: 'evt-001' },
+          data: expect.objectContaining({
+            status: 'pending',
+            attemptCount: { increment: 1 },
+          }),
+        }),
+      );
+    });
+
+    it('入队抛错 → 已达最大尝试 → 永久 failed', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ attemptCount: 2 })); // 已达 2 次
+      prisma.aiJob.findUnique.mockResolvedValue({
+        id: 'job-001',
+        queueName: 'ai-interactive',
+        lifecycleStatus: 'queued',
+      });
+      queueService.add.mockRejectedValue(new Error('Redis down'));
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+
+      // 应调用 markFailed（永久）
+      expect(outboxRepo.markFailed).toHaveBeenCalledWith(
+        'evt-001',
+        expect.any(String),
+        expect.any(String),
+      );
+    });
+  });
+
+  describe('空轮询', () => {
+    it('无可分发事件 → 返回全零', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([]);
+
+      const result = await dispatcher.cycle();
+      expect(result).toEqual({ processed: 0, published: 0, failed: 0, skipped: 0 });
+    });
+  });
+
+  describe('无效 payload', () => {
+    it('payload 缺少 jobId → 永久 failed', async () => {
+      outboxRepo.releaseExpiredLocks.mockResolvedValue(0);
+      outboxRepo.findDispatchable.mockResolvedValue([makeEvent({ payload: {} })]);
+      outboxRepo.markProcessing.mockResolvedValue(makeEvent({ payload: {} }));
+
+      const result = await dispatcher.cycle();
+      expect(result.failed).toBe(1);
+      expect(outboxRepo.markFailed).toHaveBeenCalledWith(
+        'evt-001',
+        'INVALID_PAYLOAD',
+        expect.any(String),
+      );
+    });
+  });
+});

src/modules/ai-job/outbox-dispatcher.service.ts

@@ -0,0 +1,206 @@
+import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
+import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository';
+import { QueueService } from '../../infrastructure/queue/queue.service';
+import { PrismaService } from '../../infrastructure/database/prisma.service';
+
+/**
+ * M-AI-03-07: Outbox Dispatcher
+ *
+ * 将 OutboxEvent（eventType = 'ai.job.enqueue'）可靠投递到 BullMQ。
+ *
+ * 运行位置：仅 Worker Process（API 不启动 Dispatcher）。
+ *
+ * 算法（ADR-003 §4.3 — 先 CAS 后投递）：
+ *   1. findDispatchable(N) — 读取 pending 事件
+ *   2. FOR EACH:
+ *      a. markProcessing (CAS: pending → processing)
+ *      b. IF locked IS NULL → SKIP（被其他 Dispatcher 抢走）
+ *      c. 验证 eventType、加载 AiJob、验证 queueName
+ *      d. queue.add() — BullMQ jobId = AiJob.id
+ *      e. markPublished / markFailed
+ *
+ * 崩溃窗口保护：
+ *   BullMQ jobId = AiJob.id → 相同 jobId 重复 add 视为 no-op。
+ *   markProcessing 锁超时（60s）→ releaseExpiredLocks 重置为 pending。
+ */
+
+const DISPATCH_INTERVAL_MS = 5_000;  // 每 5s 轮询
+const DISPATCH_BATCH_SIZE = 10;      // 每次最多领取 10 个
+const LOCK_TIMEOUT_MS = 60_000;      // processing 锁超时
+const MAX_ATTEMPTS = 3;
+
+@Injectable()
+export class OutboxDispatcher implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(OutboxDispatcher.name);
+  private readonly instanceId = `dispatcher-${process.pid}-${Date.now()}`;
+  private timer: ReturnType<typeof setInterval> | null = null;
+  private running = false;
+
+  constructor(
+    private readonly outboxRepo: OutboxRepository,
+    private readonly queueService: QueueService,
+    private readonly prisma: PrismaService,
+  ) {}
+
+  async onModuleInit(): Promise<void> {
+    this.logger.log(`OutboxDispatcher starting, instance=${this.instanceId}`);
+    // 启动时立即运行一次，然后定时轮询
+    await this.cycle().catch((err) =>
+      this.logger.error(`Initial dispatch cycle failed`, err?.stack),
+    );
+    this.timer = setInterval(
+      () => this.cycle().catch((err) => this.logger.error(`Dispatch cycle failed`, err?.stack)),
+      DISPATCH_INTERVAL_MS,
+    );
+  }
+
+  onModuleDestroy(): void {
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = null;
+    }
+    this.logger.log(`OutboxDispatcher stopped, instance=${this.instanceId}`);
+  }
+
+  /**
+   * 单次调度周期。
+   */
+  async cycle(): Promise<{ processed: number; published: number; failed: number; skipped: number }> {
+    if (this.running) {
+      // 防止并发周期重叠（上一批尚未处理完）
+      return { processed: 0, published: 0, failed: 0, skipped: 0 };
+    }
+    this.running = true;
+
+    let published = 0;
+    let failed = 0;
+    let skipped = 0;
+    let events: any[] = [];
+
+    try {
+      // 1. 释放过期锁（先清理上一轮崩溃的 processing 事件）
+      const released = await this.outboxRepo.releaseExpiredLocks(LOCK_TIMEOUT_MS);
+      if (released > 0) {
+        this.logger.warn(`Released ${released} expired lock(s)`);
+      }
+
+      // 2. 查找可分发事件
+      events = await this.outboxRepo.findDispatchable(DISPATCH_BATCH_SIZE);
+
+      if (events.length === 0) {
+        this.running = false;
+        return { processed: 0, published: 0, failed: 0, skipped: 0 };
+      }
+
+      // 3. 逐个处理
+      for (const event of events) {
+        try {
+          const result = await this.dispatchOne(event.id);
+          if (result === 'published') published++;
+          else if (result === 'failed') failed++;
+          else skipped++; // CAS locked by another dispatcher
+        } catch (err: any) {
+          this.logger.error(`Unexpected error dispatching event=${event.id}: ${err.message}`, err?.stack);
+          failed++;
+          await this.outboxRepo.markFailed(event.id, 'DISPATCH_INTERNAL', err.message?.slice(0, 500)).catch(() => {});
+        }
+      }
+    } finally {
+      this.running = false;
+    }
+
+    if (published > 0 || failed > 0) {
+      this.logger.log(
+        `Dispatch cycle: ${events.length} found, ${published} published, ` +
+        `${failed} failed, ${skipped} skipped`,
+      );
+    }
+
+    return { processed: published + failed + skipped, published, failed, skipped };
+  }
+
+  // ═══════════════════════════════════════════════════════════
+  // 单个事件分发
+  // ═══════════════════════════════════════════════════════════
+
+  private async dispatchOne(eventId: string): Promise<'published' | 'failed' | 'skipped'> {
+    // Step 1: CAS 抢锁（先 CAS 后投递 — ADR-003 §4.3 修正）
+    const locked = await this.outboxRepo.markProcessing(eventId, this.instanceId);
+    if (!locked) return 'skipped'; // 被其他 Dispatcher 抢走
+
+    try {
+      // Step 2: 验证 eventType
+      if (locked.eventType !== 'ai.job.enqueue') {
+        this.logger.warn(`Unknown eventType="${locked.eventType}" for event=${eventId} — marking failed`);
+        await this.outboxRepo.markFailed(eventId, 'UNKNOWN_EVENT_TYPE', `Unsupported eventType: ${locked.eventType}`);
+        return 'failed';
+      }
+
+      // Step 3: 加载并验证 AiJob
+      const payload = locked.payload as any;
+      const jobId = payload?.jobId;
+      if (!jobId) {
+        await this.outboxRepo.markFailed(eventId, 'INVALID_PAYLOAD', 'Missing jobId in Outbox payload');
+        return 'failed';
+      }
+
+      const job = await this.prisma.aiJob.findUnique({ where: { id: jobId } });
+      if (!job) {
+        await this.outboxRepo.markFailed(eventId, 'JOB_NOT_FOUND', `AiJob ${jobId} not found`);
+        return 'failed';
+      }
+
+      // Step 4: 验证 Job 状态（因幂等检查可能已处于终态，仍应投递）
+      // 若 Job 已不在 queued，不拒绝（幂等场景下可能已处理过）
+
+      // Step 5: 验证 queueName
+      const allowedQueues = ['ai-interactive', 'ai-background'];
+      if (job.queueName && !allowedQueues.includes(job.queueName)) {
+        await this.outboxRepo.markFailed(eventId, 'ILLEGAL_QUEUE_NAME', `Queue "${job.queueName}" not allowed`);
+        return 'failed';
+      }
+
+      const targetQueue = job.queueName || 'ai-interactive';
+
+      // Step 6: 投递 BullMQ（jobId = AiJob.id 保证幂等）
+      // 入队失败 → 抛出，由外层 catch 统一处理退避/重试
+      await this.queueService.add(targetQueue, { jobId: job.id }, { jobId: job.id });
+
+      // Step 7: 标记 published
+      await this.outboxRepo.markPublished(eventId);
+
+      this.logger.log(`Published: event=${eventId} job=${job.id} → queue=${targetQueue}`);
+      return 'published';
+    } catch (err: any) {
+      const errorCode = err?.code || 'DISPATCH_ERROR';
+      const message = err?.message?.slice(0, 500) || 'Unknown error';
+      const currentAttempt = (locked.attemptCount || 0) + 1;
+
+      if (currentAttempt >= MAX_ATTEMPTS) {
+        // 永久失败
+        await this.outboxRepo.markFailed(eventId, errorCode, message);
+        this.logger.error(`Event ${eventId} permanently failed after ${currentAttempt} attempts`);
+        return 'failed';
+      }
+
+      // 暂时失败：回退到 pending（exponential backoff），不经过 markFailed
+      await this.prisma.outboxEvent.update({
+        where: { id: eventId },
+        data: {
+          status: 'pending',
+          availableAt: new Date(Date.now() + Math.pow(2, currentAttempt) * 1000),
+          lockedAt: null,
+          lockedBy: null,
+          attemptCount: { increment: 1 },
+          lastErrorCode: errorCode,
+          lastErrorMessage: message,
+        },
+      });
+      this.logger.warn(
+        `Event ${eventId} transient failure: attempt=${currentAttempt}/${MAX_ATTEMPTS}, ` +
+        `backoff=${Math.pow(2, currentAttempt)}s, error=${errorCode}`,
+      );
+      return 'failed'; // 本轮失败，将重试
+    }
+  }
+}

src/worker.module.ts

@@ -18,6 +18,7 @@ import { AiBackgroundJobWorker } from './workers/ai-background.worker';
 import { AuditLogProcessor } from './modules/admin-audit-log/audit-log.processor';
 import { FileCleanupProcessor } from './modules/files/file-cleanup.processor';
 import { WorkerHeartbeatService } from './workers/worker-heartbeat.service';
+import { OutboxDispatcher } from './modules/ai-job/outbox-dispatcher.service';
 
 import { AiAnalysisModule } from './modules/ai-analysis/ai-analysis.module';
 import { DocumentImportModule } from './modules/document-import/document-import.module';
@@ -81,6 +82,7 @@ import appleConfig from './config/apple.config';
     AuditLogProcessor,
     FileCleanupProcessor,
     WorkerHeartbeatService,
+    OutboxDispatcher,
   ],
 })
 export class WorkerModule {}