photo-live/services/worker/src/main.ts

// services/worker/src/main.ts —— photo worker 进程入口（Task 21；P4a Task 3
// 增补水印重生成消费者）。构建产物 dist/main.js 由根 dev:worker / 生产启动
// 命令运行；env 一律来自进程环境（根 dotenv-cli 注入），此处不读文件。
// 启动顺序：Prisma + MinIO 适配器 → 三条专用 ioredis 连接（Worker 阻塞命令
// 须独占连接——photo 与 watermark 各一条；Queue/sweep 入队共用另一条）→
// BullMQ Worker ×2（photo 并发 2 / watermark 并发 1）+ 30s 补偿扫描。
// 优雅退出：SIGINT/SIGTERM 先停扫描，worker.close()（等待在飞 job）后逐个
// 关闭连接（Task 19 gotcha：BullMQ 不接管外部连接，须显式 quit；未监听的
// error 事件会崩进程，全部挂日志监听）。
import { Queue } from 'bullmq';
import IORedis from 'ioredis';
import { pino, type Logger } from 'pino';
import { PrismaClient } from '@photo/db-cn';
import { QUEUE_PHOTO_PROCESS, QUEUE_WATERMARK } from '@photo/contracts';
import { createStorageFromEnv } from '@photo/storage';
import { createPhotoProcessWorker } from './process.processor';
import { createWatermarkRegenWorker } from './watermark.processor';
import { startSweepInterval } from './sweep';

const SWEEP_INTERVAL_MS = 30_000;

function createConnection(logger: Logger, label: string): IORedis {
  const connection = new IORedis(process.env.REDIS_URL_CN ?? 'redis://localhost:6379', {
    maxRetriesPerRequest: null, // BullMQ 阻塞命令要求：禁用 per-request 重试上限
  });
  connection.on('error', (err: Error) => {
    logger.warn({ label, err: err.message }, 'redis 连接异常（ioredis 自动重连）');
  });
  return connection;
}

async function main(): Promise<void> {
  const logger = pino({ level: process.env.LOG_LEVEL ?? 'info' });
  const prisma = new PrismaClient();
  const storage = createStorageFromEnv();
  const workerConnection = createConnection(logger, 'worker');
  const watermarkConnection = createConnection(logger, 'watermark-worker');
  const queueConnection = createConnection(logger, 'queue');
  const sweepQueue = new Queue(QUEUE_PHOTO_PROCESS, { connection: queueConnection });
  const worker = createPhotoProcessWorker({
    prisma,
    storage,
    connection: workerConnection,
    logger,
  });
  const watermarkWorker = createWatermarkRegenWorker({
    prisma,
    storage,
    connection: watermarkConnection,
    logger,
  });
  const sweep = startSweepInterval(SWEEP_INTERVAL_MS, { prisma, queue: sweepQueue, logger });
  logger.info(
    {
      queue: QUEUE_PHOTO_PROCESS,
      watermarkQueue: QUEUE_WATERMARK,
      concurrency: 2,
      sweepIntervalMs: SWEEP_INTERVAL_MS,
    },
    'photo worker started',
  );

  let stopping = false;
  const shutdown = async (signal: string): Promise<void> => {
    if (stopping) return;
    stopping = true;
    logger.info({ signal }, 'photo worker shutting down');
    sweep.stop();
    try {
      await worker.close(); // 等待在飞 job 完成
      await watermarkWorker.close();
      await sweepQueue.close();
      await workerConnection.quit().catch(() => undefined);
      await watermarkConnection.quit().catch(() => undefined);
      await queueConnection.quit().catch(() => undefined);
      await prisma.$disconnect();
    } catch (err) {
      logger.error(
        { err: err instanceof Error ? err.message : String(err) },
        'photo worker 关闭异常',
      );
    } finally {
      process.exit(0);
    }
  };
  process.once('SIGINT', () => void shutdown('SIGINT'));
  process.once('SIGTERM', () => void shutdown('SIGTERM'));
}

void main();