src/server/jobs/runtime.ts

import type { WorkerOptions } from 'bullmq';
import type { PrismaClient } from '@/generated/prisma/client';
import { createRedis, createTaskQueue, queueName } from './queue';
import { createHeartbeat } from './heartbeat';
import { createTaskWorker } from './worker';
import { dispatchTasks } from './dispatcher';
import { createWorkerLogger } from './logger';

// 正式入口与子进程测试共用生命周期；测试可缩短时间预算，不替换信号或关闭流程。
export function runWorkerProcess(
  db: PrismaClient,
  connection: ReturnType<typeof createRedis>,
  options: {
    logger?: ReturnType<typeof createWorkerLogger>;
    workerOptions?: Omit<WorkerOptions, 'connection'>;
    heartbeatIntervalMs?: number;
    heartbeatTtlMs?: number;
    shutdownTimeoutMs?: number;
  } = {},
) {
  const logger = options.logger ?? createWorkerLogger();
  const heartbeat = createHeartbeat(
    connection,
    undefined,
    options.heartbeatTtlMs,
  );
  const queue = createTaskQueue(connection);
  const worker = createTaskWorker(
    db,
    connection,
    options.workerOptions,
    logger,
  );
  let stopping = false;
  let exitStatus = 0;
  let tickPromise: Promise<void> | undefined;
  let lastFailureLog = 0;
  let unhealthy = false;
  async function tick() {
    try {
      await dispatchTasks(db, queue);
      // SIGTERM 可能发生在扫描等待数据库时；结束的旧扫描不能把退出进程重新登记为健康。
      if (stopping) return;
      await heartbeat.renew();
      if (unhealthy)
        logger.info({ event: 'dispatcher.recovered' }, '任务投递已恢复');
      unhealthy = false;
    } catch {
      unhealthy = true;
      if (Date.now() - lastFailureLog > 30000) {
        logger.error(
          { event: 'dispatcher.failed' },
          '任务投递失败，请检查数据库和 Redis；待办保留并自动重试',
        );
        lastFailureLog = Date.now();
      }
    }
  }
  function triggerTick() {
    if (stopping || tickPromise) return;
    tickPromise = tick().finally(() => {
      tickPromise = undefined;
    });
  }
  const timer = setInterval(triggerTick, options.heartbeatIntervalMs ?? 3000);
  async function shutdown(signal: string, exitCode = 0) {
    exitStatus = Math.max(exitStatus, exitCode);
    if (stopping) return;
    stopping = true;
    clearInterval(timer);
    logger.info(
      { event: 'worker.stopping', signal },
      '停止接收新任务，等待当前任务完成',
    );
    // 默认读取事务最长 60 秒，关闭留 75 秒；超过预算以失败退出，不能把未完成任务改成功。
    const deadline = setTimeout(() => {
      logger.fatal(
        { event: 'worker.shutdown_timeout' },
        '关闭超时，未完成任务将在重启后恢复',
      );
      process.exit(1);
    }, options.shutdownTimeoutMs ?? 75000);
    deadline.unref();
    try {
      await Promise.all([
        // close 立即停止接新任务，但继续为在途任务续锁，直到消费结束。
        worker.close(),
        (async () => {
          // 一开始就撤销自己的登记；同伴的登记保留，在途任务仍可继续排空。
          await heartbeat.remove();
          await tickPromise;
        })(),
      ]);
      await queue.close();
      connection.disconnect();
      await db.$disconnect();
      clearTimeout(deadline);
      logger.info({ event: 'worker.stopped' }, 'Worker 已安全停止');
      process.exitCode = exitStatus;
    } catch {
      logger.fatal({ event: 'worker.shutdown_failed' }, '关闭资源失败');
      process.exit(1);
    }
  }
  process.on('SIGINT', () => void shutdown('SIGINT'));
  process.on('SIGTERM', () => void shutdown('SIGTERM'));
  process.on('unhandledRejection', () => {
    logger.fatal(
      { event: 'worker.unhandled_rejection' },
      '发生未处理异常，关闭进程以便守护程序重启',
    );
    void shutdown('unhandledRejection', 1);
  });
  process.on('uncaughtException', () => {
    logger.fatal(
      { event: 'worker.uncaught_exception' },
      '发生未捕获异常，关闭进程以便守护程序重启',
    );
    process.exit(1);
  });
  triggerTick();
  logger.info(
    {
      event: 'worker.started',
      instanceId: heartbeat.instanceId,
      queue: queueName,
      concurrency: options.workerOptions?.concurrency ?? 2,
    },
    'Worker 已启动',
  );
  return { worker, shutdown };
}