photo-live/services/worker/src/process.processor.ts

// services/worker/src/process.processor.ts —— QUEUE_PHOTO_PROCESS 消费者（Task 21）。
// BullMQ Worker 包装 PublishService：
//   · 每次投递将 photo_jobs pending→active（attempt 自增），成功（含幂等 already/
//     跳过 skipped）置 done；失败原样抛给 BullMQ 按生产者配置重试（attempts:3，
//     指数退避 1s 起——与 api queue.provider.ts 一致）；
//   · 终败判定：BullMQ 在 moveToFailed/moveToCompleted 之后才自增 attemptsMade
//     （v6 classes/job.js），processor 内它恒为「已完成尝试数」（0 基），故当前
//     尝试 = attemptsMade + 1；等于 attempts 上限即终败 → 单事务置
//     photo_jobs.failed(errorCode) + photos.processing_failed（visibility 不动，
//     保持 unpublished——失败照片不公开），PipelineError 携带精确错误码；
//   · Worker 使用专用 ioredis 连接（阻塞命令），连接生命周期由调用方管理
//     （Task 19 gotcha：BullMQ 不接管外部连接；未监听的 error 事件会崩进程）。
import { Worker, type Job } from 'bullmq';
import type IORedis from 'ioredis';
import type { PrismaClient } from '@photo/db-cn';
import type { StoragePort } from '@photo/storage';
import type { Logger } from 'pino';
import { QUEUE_PHOTO_PROCESS } from '@photo/contracts';
import { PipelineError } from './pipeline';
import { PublishService, type ProcessOutcome } from './publish.service';

export type { ProcessOutcome } from './publish.service';

export interface PhotoProcessJobData {
  photoId: string;
}

/** job 名称（与 api 生产者/photo_jobs.type 同值）。 */
export const PHOTO_PROCESS_JOB_NAME = 'process';

/**
 * jobId 以 photoId 派生：并发投递/补偿扫描重复入队天然去重（BullMQ 同 jobId
 * 不重复执行）。不含 ':'（BullMQ 保留分隔符，Task 19 gotcha）。
 */
export function photoProcessJobId(photoId: string): string {
  return `${PHOTO_PROCESS_JOB_NAME}-${photoId}`;
}

/** 与 api 生产者一致的重试策略（sweep 补偿入队复用同一组值）。 */
export const PROCESS_JOB_ATTEMPTS = 3;
export const PROCESS_JOB_BACKOFF = { type: 'exponential', delay: 1000 } as const;

/** 终败errorCode 兜底（非流水线错误：对象缺失、网络等）。 */
export const FALLBACK_ERROR_CODE = 'PROCESS_FAILED';

export interface PhotoProcessWorkerOptions {
  prisma: PrismaClient;
  storage: StoragePort;
  /** Worker 专用连接（阻塞命令），调用方负责 quit。 */
  connection: IORedis;
  concurrency?: number;
  logger?: Logger;
}

export function createPhotoProcessWorker(
  options: PhotoProcessWorkerOptions,
): Worker<PhotoProcessJobData, ProcessOutcome, string> {
  const { prisma } = options;
  const publish = new PublishService(prisma, options.storage, options.logger);

  const processor = async (
    job: Job<PhotoProcessJobData>,
  ): Promise<ProcessOutcome> => {
    const photoId = job.data.photoId;
    // 投递即认领：pending→active（sweep 已置 active 的重复投递幂等），attempt 计数。
    await prisma.photo_jobs.updateMany({
      where: {
        photoId,
        type: PHOTO_PROCESS_JOB_NAME,
        state: { in: ['pending', 'active'] },
      },
      data: { state: 'active', attempt: { increment: 1 } },
    });
    try {
      const outcome = await publish.processPhoto(photoId);
      // published/already/skipped 对 job 而言都是「无需再投递」→ done。
      await prisma.photo_jobs.updateMany({
        where: {
          photoId,
          type: PHOTO_PROCESS_JOB_NAME,
          state: { in: ['pending', 'active'] },
        },
        data: { state: 'done', errorCode: null, nextRetryAt: null },
      });
      options.logger?.info({ photoId, jobId: job.id, outcome }, 'process job 完成');
      return outcome;
    } catch (err) {
      const isFinalAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1);
      if (isFinalAttempt) {
        const errorCode =
          err instanceof PipelineError ? err.code : FALLBACK_ERROR_CODE;
        options.logger?.error(
          { photoId, jobId: job.id, errorCode },
          'process job 终败（重试耗尽）',
        );
        await prisma.$transaction([
          prisma.photo_jobs.updateMany({
            where: { photoId, type: PHOTO_PROCESS_JOB_NAME },
            data: { state: 'failed', errorCode, nextRetryAt: null },
          }),
          prisma.photos.updateMany({
            where: {
              id: photoId,
              processingState: { in: ['uploaded', 'processing_failed'] },
            },
            data: { processingState: 'processing_failed', errorCode },
          }),
        ]);
      }
      throw err; // 交还 BullMQ：未耗尽则按退避重试，耗尽则入 failed set
    }
  };

  return new Worker<PhotoProcessJobData, ProcessOutcome, string>(
    QUEUE_PHOTO_PROCESS,
    processor,
    {
      connection: options.connection,
      concurrency: options.concurrency ?? 2,
    },
  );
}