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

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'; // 本轮失败，将重试
    }
  }
}