src/server/jobs/dispatcher.ts

import type { PrismaClient } from '@/generated/prisma/client';
import { createTaskQueue, taskJobId } from './queue';

type Queue = ReturnType<typeof createTaskQueue>;
export async function dispatchTasks(db: PrismaClient, queue: Queue) {
  // 重复投递使用同一个 jobId，BullMQ 自动去重。每 30 秒还会核对一次旧的待执行记录，弥补 Redis 数据丢失。
  const tasks = await db.backgroundTask.findMany({
    where: {
      status: 'WAITING',
      OR: [
        { dispatchedAt: null },
        { dispatchedAt: { lt: new Date(Date.now() - 30000) } },
      ],
    },
    // 优先投递新任务，再轮询最久未核对的旧任务，避免大量延迟提醒饿死新导出。
    orderBy: [
      { dispatchedAt: { sort: 'asc', nulls: 'first' } },
      { runAt: 'asc' },
    ],
    take: 100,
    select: { id: true, run: true, runAt: true },
  });
  for (const task of tasks) {
    const existing = await queue.getJob(taskJobId(task.id, task.run));
    const state = await existing?.getState();
    // failed 事件没有执行租约，不能直接改账本。这里只处理已回到等待、无租约的同轮任务；
    // 若扫描期间新执行者已认领为 RUNNING，条件更新会失效，不会清掉它的租约。
    if (state === 'failed') {
      await db.backgroundTask.updateMany({
        where: {
          id: task.id,
          run: task.run,
          status: 'WAITING',
          executionId: null,
        },
        data: { status: 'FAILED', lastError: '队列执行失败，请手动重试' },
      });
      continue;
    }
    if (state === 'completed') await existing?.remove();
    await queue.add(
      'execute',
      { taskId: task.id, run: task.run },
      {
        jobId: taskJobId(task.id, task.run),
        delay: Math.max(0, task.runAt.getTime() - Date.now()),
      },
    );
    await db.backgroundTask.updateMany({
      where: { id: task.id, run: task.run, status: 'WAITING' },
      data: { dispatchedAt: new Date() },
    });
  }
  // Worker 崩溃或队列数据丢失后，过期执行租约回到待处理；旧执行者的 executionId 会失效。
  // updatedAt 也是条件的一部分：新认领或进度已刷新时间的任务不会被这轮恢复覆盖。
  await db.backgroundTask.updateMany({
    where: {
      status: 'RUNNING',
      updatedAt: { lt: new Date(Date.now() - 120000) },
    },
    data: { status: 'WAITING', executionId: null, dispatchedAt: null },
  });
  // 导出只保留一天，清理正文但保留任务账本；下载入口也会检查过期时间。
  await db.backgroundTask.updateMany({
    where: {
      kind: 'EXPORT',
      expiresAt: { lt: new Date() },
      resultCsv: { not: null },
    },
    data: { resultCsv: null },
  });
}