src/server/jobs/worker.ts

import { UnrecoverableError, Worker, type WorkerOptions } from 'bullmq';
import type { PrismaClient } from '@/generated/prisma/client';
import { createRedis, queueName, queuePrefix, type TaskJob } from './queue';
import { processTask } from './processor';
import { createWorkerLogger } from './logger';

// 启动脚本和集成测试复用同一处理器与监听器；测试只调整真实 Worker 的锁时长。
export function createTaskWorker(
  db: PrismaClient,
  connection: ReturnType<typeof createRedis>,
  options: Omit<WorkerOptions, 'connection'> = {},
  logger = createWorkerLogger(),
) {
  const worker = new Worker<TaskJob>(
    queueName,
    async (job) => {
      const startedAt = Date.now();
      const context = {
        taskId: job.data.taskId,
        run: job.data.run,
        attempt: job.attemptsMade + 1,
      };
      logger.info({ event: 'task.started', ...context }, '开始执行任务');
      try {
        await processTask(db, job);
        const task = await db.backgroundTask.findUnique({
          where: { id: job.data.taskId },
          select: { status: true, run: true },
        });
        logger.info(
          {
            event:
              task?.status === 'COMPLETED' && task.run === job.data.run
                ? 'task.completed'
                : 'task.skipped',
            ...context,
            durationMs: Date.now() - startedAt,
          },
          '任务处理结束',
        );
      } catch (error) {
        logger.warn(
          {
            event: 'task.attempt_failed',
            ...context,
            durationMs: Date.now() - startedAt,
            errorType:
              error instanceof Error && error.name === 'UnrecoverableError'
                ? 'permanent'
                : 'transient',
          },
          '本次执行失败，队列将决定重试或结束',
        );
        throw error;
      }
    },
    { connection, prefix: queuePrefix(), concurrency: 2, ...options },
  );
  worker.on('error', () =>
    logger.error(
      { event: 'worker.connection_error' },
      'Worker 连接异常，正在重连',
    ),
  );
  worker.on('stalled', (jobId) =>
    logger.warn(
      { event: 'task.stalled', jobId },
      '任务失去执行锁，队列将恢复处理',
    ),
  );
  worker.on('failed', (job, error) => {
    if (!job) return;
    const exhausted =
      job.attemptsMade >= (job.opts.attempts ?? 3) ||
      error instanceof UnrecoverableError;
    logger.warn(
      {
        event: 'task.failed',
        taskId: job.data.taskId,
        run: job.data.run,
        attempts: job.attemptsMade,
        exhausted,
      },
      '队列记录一次任务失败',
    );
    // failed 只有队列任务与轮次，没有失败那次执行的 executionId。
    // 不能据此清空 RUNNING / WAITING：回调迟到时可能已有新执行者接手。
    // 正常业务失败由处理器按租约写入；崩溃遗留由 dispatcher 恢复后核对队列终态。
  });
  return worker;
}