src/server/services/tasks.ts

import { TRPCError } from '@trpc/server';
import type { PrismaClient } from '@/generated/prisma/client';
import type { ExportFilters } from '@/shared/tasks';
import { requireMembership, requireTicketAccess } from './access';
import { recordTicketEvent } from './ticket-events';
import { scheduleReminder } from '@/server/jobs/outbox';

export function createTaskService(
  db: PrismaClient,
  userId: string,
  teamId: string,
) {
  return {
    async createExport(filters: ExportFilters) {
      return db.$transaction(async (tx) => {
        await tx.$queryRaw`SELECT id FROM "Team" WHERE id = ${teamId}::uuid FOR UPDATE`;
        const member = await requireMembership(tx, userId, teamId);
        const pending = await tx.backgroundTask.count({
          where: {
            teamId,
            ownerId: userId,
            kind: 'EXPORT',
            OR: [
              { status: { in: ['WAITING', 'RUNNING'] } },
              { createdAt: { gt: new Date(Date.now() - 60000) } },
            ],
          },
        });
        if (pending >= 3)
          throw new TRPCError({
            code: 'TOO_MANY_REQUESTS',
            message: '导出任务较多，请一分钟后或任务完成后再试',
          });
        // 这里只写待投递记录，因此 Redis 暂时离线也不会丢失已受理的导出。
        const task = await tx.backgroundTask.create({
          data: {
            teamId,
            ownerId: userId,
            kind: 'EXPORT',
            filters,
            scopeCreatorId: member.role === 'CUSTOMER' ? userId : null,
          },
        });
        return { id: task.id };
      });
    },
    async list(input: { page: number; pageSize: number }) {
      return db.$transaction(
        async (tx) => {
          const member = await requireMembership(tx, userId, teamId);
          const where = {
            teamId,
            OR: [
              { kind: 'EXPORT' as const, ownerId: userId },
              ...(member.role !== 'CUSTOMER'
                ? [{ kind: 'REMINDER' as const }]
                : []),
            ],
          };
          const items = await tx.backgroundTask.findMany({
            where,
            orderBy: { createdAt: 'desc' },
            skip: (input.page - 1) * input.pageSize,
            take: input.pageSize,
            select: {
              id: true,
              kind: true,
              status: true,
              progress: true,
              attempts: true,
              run: true,
              runAt: true,
              createdAt: true,
              lastError: true,
              resultSummary: true,
              expiresAt: true,
              scopeCreatorId: true,
              ticket: { select: { number: true } },
            },
          });
          return {
            items: items.map((task) => ({
              ...task,
              runAt: task.runAt.toISOString(),
              createdAt: task.createdAt.toISOString(),
              expiresAt: task.expiresAt?.toISOString() ?? null,
              canDownload:
                task.kind === 'EXPORT' &&
                task.status === 'COMPLETED' &&
                !!task.expiresAt &&
                task.expiresAt > new Date() &&
                (!!task.scopeCreatorId || member.role !== 'CUSTOMER'),
            })),
            total: await tx.backgroundTask.count({ where }),
          };
        },
        { isolationLevel: 'RepeatableRead' },
      );
    },
    async retry(id: string) {
      return db.$transaction(async (tx) => {
        await tx.$queryRaw`SELECT id FROM "Team" WHERE id = ${teamId}::uuid FOR UPDATE`;
        const member = await requireMembership(tx, userId, teamId);
        const task = await tx.backgroundTask.findFirst({
          where: {
            id,
            teamId,
            OR: [
              { kind: 'EXPORT', ownerId: userId },
              ...(member.role !== 'CUSTOMER'
                ? [{ kind: 'REMINDER' as const }]
                : []),
            ],
          },
        });
        if (!task)
          throw new TRPCError({
            code: 'NOT_FOUND',
            message: '任务不存在或无权访问',
          });
        if (task.status !== 'FAILED')
          throw new TRPCError({
            code: 'CONFLICT',
            message: '只有失败的任务可以重试',
          });
        if (
          task.kind === 'EXPORT' &&
          !task.scopeCreatorId &&
          member.role === 'CUSTOMER'
        )
          throw new TRPCError({
            code: 'FORBIDDEN',
            message: '权限已变化，请重新创建导出',
          });
        // 新一轮使用新的 jobId；保留旧的失败队列记录，也不会被 BullMQ 去重挡住。
        await tx.backgroundTask.update({
          where: { id },
          data: {
            run: { increment: 1 },
            status: 'WAITING',
            progress: 0,
            attempts: 0,
            lastError: null,
            executionId: null,
            dispatchedAt: null,
            runAt: new Date(),
            resultCsv: null,
            resultSummary: null,
            expiresAt: null,
          },
        });
        return { id };
      });
    },
    async schedule(input: {
      id: string;
      version: number;
      minutes: number | null;
    }) {
      return db.$transaction(async (tx) => {
        await tx.$queryRaw`SELECT id FROM "Team" WHERE id = ${teamId}::uuid FOR UPDATE`;
        const { ticket, member } = await requireTicketAccess(
          tx,
          userId,
          teamId,
          input.id,
        );
        if (member.role === 'CUSTOMER')
          throw new TRPCError({
            code: 'FORBIDDEN',
            message: '仅团队工作人员可以设置提醒',
          });
        if (['RESOLVED', 'CLOSED'].includes(ticket.status))
          throw new TRPCError({
            code: 'CONFLICT',
            message: '已解决或关闭的工单无需提醒',
          });
        if (ticket.version !== input.version)
          throw new TRPCError({
            code: 'CONFLICT',
            message: '工单已更新，请刷新后重试',
          });
        const updated = await tx.ticket.update({
          where: { id: ticket.id },
          data: {
            dueAt:
              input.minutes === null
                ? null
                : new Date(Date.now() + input.minutes * 60000),
            reminderRevision: { increment: 1 },
            version: { increment: 1 },
          },
        });
        await recordTicketEvent(tx, {
          teamId,
          ticketId: ticket.id,
          version: updated.version,
          kind: 'REMINDER_SCHEDULED',
          actorUserId: userId,
          actorRole: member.role,
          reason:
            input.minutes === null
              ? '取消超时提醒'
              : `设置 ${input.minutes} 分钟后超时提醒`,
        });
        await scheduleReminder(tx, updated);
        return { id: updated.id };
      });
    },
    async notifications() {
      const member = await requireMembership(db, userId, teamId);
      if (member.role === 'CUSTOMER') return [];
      return (
        await db.reminderNotification.findMany({
          where: { teamId, userId },
          orderBy: { createdAt: 'desc' },
          take: 30,
          include: {
            ticket: { select: { id: true, number: true, title: true } },
          },
        })
      ).map((item) => ({
        id: item.id,
        ticket: item.ticket,
        readAt: item.readAt?.toISOString() ?? null,
        createdAt: item.createdAt.toISOString(),
      }));
    },
    async read(id: string) {
      const member = await requireMembership(db, userId, teamId);
      if (member.role === 'CUSTOMER')
        throw new TRPCError({ code: 'FORBIDDEN' });
      await db.reminderNotification.updateMany({
        where: { id, teamId, userId },
        data: { readAt: new Date() },
      });
      return { ok: true };
    },
  };
}
export async function downloadExport(
  db: PrismaClient,
  userId: string,
  id: string,
) {
  return db.$transaction(
    async (tx) => {
      const task = await tx.backgroundTask.findFirst({
        where: { id, kind: 'EXPORT', ownerId: userId },
      });
      if (!task)
        throw new TRPCError({
          code: 'NOT_FOUND',
          message: '文件不存在或无权访问',
        });
      const member = await requireMembership(tx, userId, task.teamId);
      if (!task.scopeCreatorId && member.role === 'CUSTOMER')
        throw new TRPCError({
          code: 'FORBIDDEN',
          message: '权限已变化，请重新导出',
        });
      if (
        task.status !== 'COMPLETED' ||
        !task.expiresAt ||
        task.expiresAt < new Date() ||
        task.resultCsv === null
      )
        throw new TRPCError({
          code: 'NOT_FOUND',
          message: '导出尚未完成或文件已过期',
        });
      return task.resultCsv;
    },
    { isolationLevel: 'RepeatableRead' },
  );
}