src/server/jobs/processor.ts

import { randomUUID } from 'node:crypto';
import { UnrecoverableError, type Job } from 'bullmq';
import type { BackgroundTask, PrismaClient } from '@/generated/prisma/client';
import {
  exportFiltersSchema,
  MAX_EXPORT_BYTES,
  MAX_EXPORT_ROWS,
  MAX_TASK_ATTEMPTS,
} from '@/shared/tasks';
import { ticketWhere } from '@/server/ticket-filter';
import { csvLine } from './csv';
import type { TaskJob } from './queue';

export class TaskRejected extends Error {}
// 丢失租约表示工作已交给别人，不是业务失败；旧执行者只能撤销自己的事务。
class LeaseLost extends Error {}
async function exportMember(db: PrismaClient, task: BackgroundTask) {
  const member = task.ownerId
    ? await db.membership.findUnique({
        where: { teamId_userId: { teamId: task.teamId, userId: task.ownerId } },
      })
    : null;
  // 排队时有权限，不代表执行/下载时仍有权限。客户创建的任务即使后来升职，也保留最初的数据范围。
  if (!member || (!task.scopeCreatorId && member.role === 'CUSTOMER'))
    throw new TaskRejected('成员权限已变化，请重新创建导出');
  return member;
}
export async function processTask(
  db: PrismaClient,
  job: Pick<Job<TaskJob>, 'data' | 'attemptsMade' | 'updateProgress'>,
) {
  const task = await db.backgroundTask.findUnique({
    where: { id: job.data.taskId },
  });
  if (
    !task ||
    task.run !== job.data.run ||
    task.status === 'COMPLETED' ||
    task.status === 'FAILED'
  )
    return;
  // 每次执行拿到新的“租约”。崩溃后接手的 Worker 会换掉它，旧执行者便不能再覆盖结果。
  const executionId = randomUUID();
  const lease = { id: task.id, run: task.run, executionId };
  const claimed = await db.backgroundTask.updateMany({
    where: {
      id: task.id,
      run: task.run,
      status: { in: ['WAITING', 'RUNNING'] },
    },
    data: {
      status: 'RUNNING',
      executionId,
      attempts: job.attemptsMade + 1,
      progress: 0,
      lastError: null,
    },
  });
  if (!claimed.count) return;
  const progress = async (value: number) => {
    const updated = await db.backgroundTask.updateMany({
      where: { ...lease, status: 'RUNNING' },
      data: { progress: value },
    });
    if (!updated.count) throw new LeaseLost();
    await job.updateProgress(value);
  };
  try {
    if (task.kind === 'EXPORT') {
      const member = await exportMember(db, task);
      const parsed = exportFiltersSchema.safeParse(task.filters);
      if (!parsed.success) throw new TaskRejected('导出筛选条件无效');
      const where = ticketWhere(
        task.teamId,
        member,
        parsed.data,
        task.scopeCreatorId,
      );
      let csv =
        '\uFEFF' +
        csvLine([
          '编号',
          '标题',
          '描述',
          '状态',
          '优先级',
          '负责人',
          '创建时间',
        ]);
      let rows = 0;
      // 可重复读让分页读取看到同一个快照，避免导出到一半时数据变化造成漏行/重复。
      await db.$transaction(
        async (tx) => {
          const total = await tx.ticket.count({ where });
          if (total > MAX_EXPORT_ROWS)
            throw new TaskRejected('最多导出 5000 条，请缩小筛选范围');
          let cursor = 0;
          while (true) {
            const batch = await tx.ticket.findMany({
              where: { ...where, number: { gt: cursor } },
              orderBy: { number: 'asc' },
              take: 200,
              include: {
                assignee: { include: { user: { select: { name: true } } } },
              },
            });
            if (!batch.length) break;
            for (const ticket of batch)
              csv += csvLine([
                ticket.number,
                ticket.title,
                ticket.description,
                ticket.status,
                ticket.priority,
                ticket.assignee?.user.name,
                ticket.createdAt.toISOString(),
              ]);
            if (Buffer.byteLength(csv, 'utf8') > MAX_EXPORT_BYTES)
              throw new TaskRejected('导出超过 5 MiB，请缩小筛选范围');
            rows += batch.length;
            cursor = batch[batch.length - 1].number;
            await progress(
              Math.min(95, Math.round((rows / Math.max(total, 1)) * 95)),
            );
          }
        },
        { isolationLevel: 'RepeatableRead', timeout: 60000 },
      );
      await db.$transaction(async (tx) => {
        // 和成员权限修改共用团队锁，保证“检查权限”和“保存文件”之间不会被降权插入。
        await tx.$queryRaw`SELECT id FROM "Team" WHERE id = ${task.teamId}::uuid FOR UPDATE`;
        const current = task.ownerId
          ? await tx.membership.findUnique({
              where: {
                teamId_userId: { teamId: task.teamId, userId: task.ownerId },
              },
            })
          : null;
        if (!current || (!task.scopeCreatorId && current.role === 'CUSTOMER'))
          throw new TaskRejected('成员权限已变化，请重新创建导出');
        const completed = await tx.backgroundTask.updateMany({
          where: { ...lease, status: 'RUNNING' },
          data: {
            status: 'COMPLETED',
            progress: 100,
            resultCsv: csv,
            resultSummary: `已导出 ${rows} 条工单`,
            expiresAt: new Date(Date.now() + 86400000),
            executionId: null,
          },
        });
        // 条件写入失败只代表旧文件未保存，不能继续向队列报告 100%。
        if (completed.count !== 1) throw new LeaseLost();
      });
    } else {
      await db.$transaction(async (tx) => {
        await tx.$queryRaw`SELECT id FROM "Team" WHERE id = ${task.teamId}::uuid FOR UPDATE`;
        const current = await tx.backgroundTask.findFirst({
          where: { ...lease, status: 'RUNNING' },
        });
        if (!current) return;
        const ticket = task.ticketId
          ? await tx.ticket.findFirst({
              where: { id: task.ticketId, teamId: task.teamId },
            })
          : null;
        let resultSummary = '已跳过：提醒已取消、被新设置替代或工单已解决';
        if (
          ticket &&
          ticket.dueAt &&
          ticket.reminderRevision === task.reminderRevision &&
          !['RESOLVED', 'CLOSED'].includes(ticket.status)
        ) {
          if (ticket.dueAt.getTime() > Date.now())
            throw new Error('提醒尚未到期');
          const recipients = await tx.membership.findMany({
            where: {
              teamId: task.teamId,
              role: { in: ['ADMIN', 'AGENT'] },
              ...(ticket.assigneeMembershipId
                ? { id: ticket.assigneeMembershipId }
                : { role: 'ADMIN' }),
            },
          });
          // 通知、工单版本、操作记录与任务完成一起提交。重试不会发出两次通知或留下半笔历史。
          await tx.reminderNotification.createMany({
            data: recipients.map((member) => ({
              teamId: task.teamId,
              ticketId: ticket.id,
              userId: member.userId,
              reminderRevision: ticket.reminderRevision,
            })),
            skipDuplicates: true,
          });
          const updated = await tx.ticket.update({
            where: { id: ticket.id },
            data: { version: { increment: 1 } },
          });
          await tx.ticketEvent.create({
            data: {
              teamId: task.teamId,
              ticketId: ticket.id,
              version: updated.version,
              kind: 'TIMEOUT_REMINDED',
              actorName: '系统',
              reason: `工单已超时，已生成 ${recipients.length} 条站内提醒`,
            },
          });
          resultSummary = `已生成 ${recipients.length} 条站内提醒`;
        }
        const completed = await tx.backgroundTask.updateMany({
          where: { ...lease, status: 'RUNNING' },
          data: {
            status: 'COMPLETED',
            progress: 100,
            resultSummary,
            executionId: null,
          },
        });
        // 认领在团队锁外，读过租约后仍可能被接手。更新 0 行必须抛错，
        // 让同一事务中的通知、事件和版本一起回滚，不能提交半份业务结果。
        if (completed.count !== 1) throw new LeaseLost();
      });
    }
    await job.updateProgress(100);
  } catch (error) {
    // 新执行者持有的任务状态不归旧执行者修改，也不应被标记为永久失败。
    if (error instanceof LeaseLost) return;
    const permanent = error instanceof TaskRejected;
    const final = permanent || job.attemptsMade + 1 >= MAX_TASK_ATTEMPTS;
    // 内部异常可能包含连接串/SQL，前端只接收可读的固定提示，实际重试由 BullMQ 负责。
    const failed = await db.backgroundTask.updateMany({
      where: { ...lease, status: 'RUNNING' },
      data: {
        status: final ? 'FAILED' : 'WAITING',
        executionId: null,
        lastError: permanent
          ? error.message
          : '执行暂时失败，系统将按退避策略重试；达到上限后可手动重试',
      },
    });
    // 发生业务/网络错误时也可能已经被接手。只有仍持有租约的执行者
    // 才能向队列报告失败，避免旧错误触发失败回调、误伤同 run 的新任务。
    if (!failed.count) return;
    if (permanent) throw new UnrecoverableError(error.message);
    throw new Error('后台任务执行失败');
  }
}