tests/tasks.integration.test.ts

import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { randomUUID } from 'node:crypto';
import { PrismaPg } from '@prisma/adapter-pg';
import { PrismaClient } from '@/generated/prisma/client';
import { appRouter } from '@/server/api/router';
import { processTask } from '@/server/jobs/processor';
import { downloadExport } from '@/server/services/tasks';
import { dispatchTasks } from '@/server/jobs/dispatcher';
import { createTaskQueue } from '@/server/jobs/queue';

// 通过数据库调用边界控制交错，不靠 sleep 猜测哪个 Worker 先执行。
function gate() {
  let open!: () => void;
  const promise = new Promise<void>((resolve) => {
    open = resolve;
  });
  return { promise, open };
}
async function reached(promise: Promise<void>) {
  let timer!: ReturnType<typeof setTimeout>;
  try {
    await Promise.race([
      promise,
      new Promise<never>((_, reject) => {
        timer = setTimeout(() => reject(new Error('并发同步点未到达')), 4000);
      }),
    ]);
  } finally {
    clearTimeout(timer);
  }
}

// 只清理本套测试创建的团队和账号，不重置已有业务数据。
describe.skipIf(!process.env.TEST_DATABASE_URL)(
  '后台任务权限与事务（真实数据库）',
  () => {
    const db = new PrismaClient({
      adapter: new PrismaPg({
        connectionString: process.env.TEST_DATABASE_URL ?? '',
        max: 5,
      }),
    });
    const admin = randomUUID(),
      customer = randomUUID(),
      other = randomUUID(),
      outsider = randomUUID();
    let teamId: string, foreignTeam: string;
    const api = (userId: string | null) =>
      appRouter.createCaller({ db, userId });
    const execute = async (id: string, attemptsMade = 0) => {
      const task = await db.backgroundTask.findUniqueOrThrow({ where: { id } });
      const values: number[] = [];
      await processTask(db, {
        data: { taskId: id, run: task.run },
        attemptsMade,
        updateProgress: async (value) => {
          values.push(value as number);
        },
      });
      return values;
    };
    const exportTask = async (ownerId = admin) =>
      db.backgroundTask.create({
        data: {
          teamId,
          ownerId,
          kind: 'EXPORT',
          filters: {},
          scopeCreatorId: ownerId === customer ? customer : null,
        },
      });
    const ticket = () =>
      api(customer).ticket.create({
        teamId,
        title: '=测试导出与提醒',
        description: '学习后台任务',
        priority: 'NORMAL',
      });
    beforeAll(async () => {
      for (const id of [admin, customer, other, outsider])
        await db.user.create({
          data: { id, email: `${id}@example.com`, name: id },
        });
      teamId = (await api(admin).team.create({ name: '任务测试团队' })).id;
      foreignTeam = (await api(outsider).team.create({ name: '任务隔离团队' }))
        .id;
      for (const userId of [customer, other])
        await db.membership.create({
          data: { teamId, userId, role: 'CUSTOMER' },
        });
      await ticket();
      await api(other).ticket.create({
        teamId,
        title: '其他客户的秘密',
        description: '不可导出给客户',
        priority: 'NORMAL',
      });
      await api(outsider).ticket.create({
        teamId: foreignTeam,
        title: '其他团队的秘密',
        description: '不可跨团队',
        priority: 'NORMAL',
      });
    });
    afterAll(async () => {
      await db.ticket.deleteMany({
        where: { teamId: { in: [teamId, foreignTeam].filter(Boolean) } },
      });
      await db.team.deleteMany({
        where: { id: { in: [teamId, foreignTeam].filter(Boolean) } },
      });
      await db.user.deleteMany({
        where: { id: { in: [admin, customer, other, outsider] } },
      });
      await db.$disconnect();
    });
    it('匿名与跨团队用户不能提交导出', async () => {
      await expect(
        api(null).task.createExport({ teamId, filters: {} }),
      ).rejects.toMatchObject({ code: 'UNAUTHORIZED' });
      await expect(
        api(outsider).task.createExport({ teamId, filters: {} }),
      ).rejects.toMatchObject({ code: 'FORBIDDEN' });
    });
    it('客户仅导出自己的匹配数据，进度最终 100%，CSV 保护公式', async () => {
      const task = await api(customer).task.createExport({
        teamId,
        filters: { q: '测试导出' },
      });
      expect(await execute(task.id)).toContain(100);
      const csv = await downloadExport(db, customer, task.id);
      expect(csv).toContain("'=测试导出与提醒");
      expect(csv).not.toContain('其他客户的秘密');
      expect(csv).not.toContain('其他团队的秘密');
      await expect(downloadExport(db, admin, task.id)).rejects.toMatchObject({
        code: 'NOT_FOUND',
      });
    });
    it('管理员导出也不能跨团队，筛选覆盖全部匹配结果', async () => {
      const task = await exportTask();
      await execute(task.id);
      const csv = await downloadExport(db, admin, task.id);
      expect(csv).toContain('其他客户的秘密');
      expect(csv).not.toContain('其他团队的秘密');
    });
    it('排队后被降权会终止执行，已生成文件也不能再下载', async () => {
      const pending = await exportTask();
      const finished = await exportTask();
      await execute(finished.id);
      const member = await db.membership.create({
        data: { teamId, userId: outsider, role: 'ADMIN' },
      });
      try {
        await api(outsider).team.changeRole({
          teamId,
          userId: admin,
          role: 'CUSTOMER',
        });
        await expect(execute(pending.id)).rejects.toThrow('成员权限已变化');
        await expect(
          downloadExport(db, admin, finished.id),
        ).rejects.toMatchObject({ code: 'FORBIDDEN' });
      } finally {
        await db.membership.update({
          where: { teamId_userId: { teamId, userId: admin } },
          data: { role: 'ADMIN' },
        });
        await db.membership.delete({ where: { id: member.id } });
      }
    });
    it('已移出团队的任务所有者不能继续执行或下载', async () => {
      const task = await exportTask(other);
      await db.membership.delete({
        where: { teamId_userId: { teamId, userId: other } },
      });
      try {
        await expect(execute(task.id)).rejects.toThrow('成员权限已变化');
      } finally {
        await db.membership.create({
          data: { teamId, userId: other, role: 'CUSTOMER' },
        });
      }
    });
    it('过期文件不能下载', async () => {
      const task = await exportTask();
      await execute(task.id);
      await db.backgroundTask.update({
        where: { id: task.id },
        data: { expiresAt: new Date(0) },
      });
      await expect(downloadExport(db, admin, task.id)).rejects.toMatchObject({
        code: 'NOT_FOUND',
      });
    });
    it('新工单和提醒任务原子创建，客户不能设置提醒', async () => {
      const item = await ticket();
      expect(
        await db.backgroundTask.count({ where: { ticketId: item.id } }),
      ).toBe(1);
      await expect(
        api(customer).task.schedule({
          teamId,
          id: item.id,
          version: item.version,
          minutes: 1,
        }),
      ).rejects.toMatchObject({ code: 'FORBIDDEN' });
      await expect(
        api(outsider).task.schedule({
          teamId,
          id: item.id,
          version: item.version,
          minutes: 1,
        }),
      ).rejects.toMatchObject({ code: 'FORBIDDEN' });
    });
    it('超时通知和系统记录幂等，未分配时提醒管理员', async () => {
      const item = await ticket();
      await db.ticket.update({
        where: { id: item.id },
        data: { dueAt: new Date(0) },
      });
      const task = await db.backgroundTask.findFirstOrThrow({
        where: { ticketId: item.id },
      });
      await execute(task.id);
      await execute(task.id);
      expect(
        await db.reminderNotification.count({
          where: { ticketId: item.id, userId: admin },
        }),
      ).toBe(1);
      expect(
        await db.ticketEvent.count({
          where: { ticketId: item.id, kind: 'TIMEOUT_REMINDED' },
        }),
      ).toBe(1);
      const notices = await api(admin).task.notifications({ teamId });
      const notice = notices.find((n) => n.ticket.id === item.id)!;
      await api(admin).task.read({ teamId, id: notice.id });
      expect(
        (await api(admin).task.notifications({ teamId })).find(
          (n) => n.id === notice.id,
        )?.readAt,
      ).toBeTruthy();
      expect(await api(customer).task.notifications({ teamId })).toEqual([]);
    });
    it.each(['新执行者认领', '恢复扫描清除租约'])(
      '提醒并发接手：%s 时旧事务撤销全部副作用',
      async (takeover) => {
        const item = await ticket();
        await db.ticket.update({
          where: { id: item.id },
          data: { dueAt: new Date(0) },
        });
        const task = await db.backgroundTask.findFirstOrThrow({
          where: { ticketId: item.id },
        });
        const readA = gate(),
          resumeA = gate(),
          claimedB = gate(),
          resumeB = gate();
        const oldProgress: number[] = [];
        const newProgress: number[] = [];
        const job = (values: number[]) => ({
          data: { taskId: task.id, run: task.run },
          attemptsMade: 0,
          updateProgress: async (value: string | number | object) => {
            values.push(value as number);
          },
        });
        const oldDb = db.$extends({
          query: {
            backgroundTask: {
              async findFirst({ args, query }) {
                const result = await query(args);
                // A 已在真实事务内持有团队锁，并读到了有效 E1。
                readA.open();
                await resumeA.promise;
                return result;
              },
            },
          },
        });
        const newDb = db.$extends({
          query: {
            backgroundTask: {
              async updateMany({ args, query }) {
                const result = await query(args);
                if (args.data.status === 'RUNNING') {
                  // B 的 E2 已真实提交，先不让 B 进入需要团队锁的事务。
                  claimedB.open();
                  await resumeB.promise;
                }
                return result;
              },
            },
          },
        });
        const assertEffects = async (count: number) => {
          const notices = await db.reminderNotification.findMany({
            where: { ticketId: item.id },
          });
          const events = await db.ticketEvent.findMany({
            where: { ticketId: item.id, kind: 'TIMEOUT_REMINDED' },
          });
          const current = await db.ticket.findUniqueOrThrow({
            where: { id: item.id },
          });
          expect(notices).toHaveLength(count);
          expect(events).toHaveLength(count);
          expect(current.version).toBe(item.version + count);
          expect(current.reminderRevision).toBe(item.reminderRevision);
          if (count) {
            expect(notices[0]).toMatchObject({
              teamId,
              userId: admin,
              reminderRevision: item.reminderRevision,
            });
            expect(events[0]).toMatchObject({
              teamId,
              version: current.version,
              actorName: '系统',
            });
          }
        };
        const running: Promise<unknown>[] = [];
        try {
          const a = processTask(
            oldDb as unknown as PrismaClient,
            job(oldProgress),
          );
          running.push(a);
          void a.catch(() => {});
          await reached(readA.promise);
          const firstLease = await db.backgroundTask.findUniqueOrThrow({
            where: { id: task.id },
          });
          let replacement;
          if (takeover === '新执行者认领') {
            const b = processTask(
              newDb as unknown as PrismaClient,
              job(newProgress),
            );
            running.push(b);
            void b.catch(() => {});
            await reached(claimedB.promise);
            replacement = await db.backgroundTask.findUniqueOrThrow({
              where: { id: task.id },
            });
            expect(replacement.executionId).not.toBe(firstLease.executionId);
            expect(replacement.executionId).not.toBeNull();
          } else {
            // 模拟恢复扫描的数据库状态迁移，限定本任务，不触碰其他夹具。
            await db.backgroundTask.updateMany({
              where: {
                id: task.id,
                executionId: firstLease.executionId,
                status: 'RUNNING',
              },
              data: {
                status: 'WAITING',
                executionId: null,
                dispatchedAt: null,
              },
            });
            replacement = await db.backgroundTask.findUniqueOrThrow({
              where: { id: task.id },
            });
          }
          resumeA.open();
          await a;
          // 检查交接中间态：旧事务必须整笔撤销，且不能把新租约标成失败。
          await assertEffects(0);
          expect(oldProgress).toEqual([]);
          expect(
            await db.backgroundTask.findUnique({ where: { id: task.id } }),
          ).toEqual(replacement);
          resumeB.open();
          if (takeover === '新执行者认领') await running[1];
          else await execute(task.id);
          await assertEffects(1);
          expect(
            await db.backgroundTask.findUnique({ where: { id: task.id } }),
          ).toMatchObject({
            status: 'COMPLETED',
            progress: 100,
            executionId: null,
            lastError: null,
            run: task.run,
            resultSummary: '已生成 1 条站内提醒',
          });
          if (takeover === '新执行者认领') expect(newProgress).toEqual([100]);
          await execute(task.id);
          await assertEffects(1);
        } finally {
          // 断言失败也放行所有执行者，避免留下占锁事务污染后续测试。
          resumeA.open();
          resumeB.open();
          await Promise.allSettled(running);
        }
      },
    );
    it('提醒写入事件后故障整笔回滚，重试只提交一份业务结果', async () => {
      const item = await ticket();
      await db.ticket.update({
        where: { id: item.id },
        data: { dueAt: new Date(0) },
      });
      const task = await db.backgroundTask.findFirstOrThrow({
        where: { ticketId: item.id },
      });
      const failingDb = db.$extends({
        query: {
          ticketEvent: {
            async create({ args, query }) {
              await query(args);
              // 通知、版本、事件均已写入事务，随后抛错检验真实数据库回滚。
              throw new Error('测试注入：事件写入后失败');
            },
          },
        },
      });
      await expect(
        processTask(failingDb as unknown as PrismaClient, {
          data: { taskId: task.id, run: task.run },
          attemptsMade: 0,
          updateProgress: async () => {},
        }),
      ).rejects.toThrow('后台任务执行失败');
      expect(
        await db.reminderNotification.count({ where: { ticketId: item.id } }),
      ).toBe(0);
      expect(
        await db.ticketEvent.count({
          where: { ticketId: item.id, kind: 'TIMEOUT_REMINDED' },
        }),
      ).toBe(0);
      expect(
        await db.ticket.findUnique({ where: { id: item.id } }),
      ).toMatchObject({ version: item.version });
      expect(
        await db.backgroundTask.findUnique({ where: { id: task.id } }),
      ).toMatchObject({
        status: 'WAITING',
        progress: 0,
        executionId: null,
        attempts: 1,
      });
      await execute(task.id, 1);
      expect(
        await db.reminderNotification.count({ where: { ticketId: item.id } }),
      ).toBe(1);
      const events = await db.ticketEvent.findMany({
        where: { ticketId: item.id, kind: 'TIMEOUT_REMINDED' },
      });
      expect(events).toHaveLength(1);
      expect(events[0].version).toBe(item.version + 1);
      expect(
        await db.ticket.findUnique({ where: { id: item.id } }),
      ).toMatchObject({ version: item.version + 1 });
      expect(
        await db.backgroundTask.findUnique({ where: { id: task.id } }),
      ).toMatchObject({
        status: 'COMPLETED',
        progress: 100,
        executionId: null,
        lastError: null,
        attempts: 2,
      });
    });
    it('已解决的工单不发送旧提醒；重新打开建立新提醒', async () => {
      let item = await ticket();
      const old = await db.backgroundTask.findFirstOrThrow({
        where: { ticketId: item.id },
      });
      item = await api(admin).ticket.updateStatus({
        teamId,
        id: item.id,
        version: item.version,
        status: 'IN_PROGRESS',
      });
      item = await api(admin).ticket.updateStatus({
        teamId,
        id: item.id,
        version: item.version,
        status: 'RESOLVED',
        reason: '完成',
      });
      await execute(old.id);
      expect(
        await db.reminderNotification.count({ where: { ticketId: item.id } }),
      ).toBe(0);
      item = await api(admin).ticket.updateStatus({
        teamId,
        id: item.id,
        version: item.version,
        status: 'OPEN',
        reason: '重新打开',
      });
      expect(
        await db.backgroundTask.count({ where: { ticketId: item.id } }),
      ).toBe(2);
    });
    it('取消和重新设置提醒后旧任务失效，版本冲突拒绝写入', async () => {
      const item = await ticket();
      const old = await db.backgroundTask.findFirstOrThrow({
        where: { ticketId: item.id },
      });
      await api(admin).task.schedule({
        teamId,
        id: item.id,
        version: item.version,
        minutes: null,
      });
      await execute(old.id);
      expect(
        await db.reminderNotification.count({ where: { ticketId: item.id } }),
      ).toBe(0);
      await expect(
        api(admin).task.schedule({
          teamId,
          id: item.id,
          version: item.version,
          minutes: 1,
        }),
      ).rejects.toMatchObject({ code: 'CONFLICT' });
    });
    it('手动重试只允许失败任务，生成新的执行轮次并防止旧任务覆盖', async () => {
      const task = await exportTask();
      await expect(
        api(admin).task.retry({ teamId, id: task.id }),
      ).rejects.toMatchObject({ code: 'CONFLICT' });
      await db.backgroundTask.update({
        where: { id: task.id },
        data: { status: 'FAILED' },
      });
      await expect(
        api(customer).task.retry({ teamId, id: task.id }),
      ).rejects.toMatchObject({ code: 'NOT_FOUND' });
      await api(admin).task.retry({ teamId, id: task.id });
      await processTask(db, {
        data: { taskId: task.id, run: 0 },
        attemptsMade: 0,
        updateProgress: async () => {},
      });
      expect(
        (await db.backgroundTask.findUniqueOrThrow({ where: { id: task.id } }))
          .status,
      ).toBe('WAITING');
      await execute(task.id);
      expect(
        (await db.backgroundTask.findUniqueOrThrow({ where: { id: task.id } }))
          .run,
      ).toBe(1);
    });
    it('Redis 投递失败时任务仍留在数据库待恢复', async () => {
      const task = await exportTask();
      const broken = {
        getJob: async () => null,
        add: async () => {
          throw new Error('Redis 离线');
        },
      } as unknown as ReturnType<typeof createTaskQueue>;
      await expect(dispatchTasks(db, broken)).rejects.toThrow('Redis 离线');
      expect(
        (await db.backgroundTask.findUniqueOrThrow({ where: { id: task.id } }))
          .status,
      ).toBe('WAITING');
    });
    it('中途进度写入故障保留任务，第三次失败后可手动重试恢复', async () => {
      const task = await exportTask();
      for (let attempt = 0; attempt < 3; attempt++) {
        await expect(
          processTask(db, {
            data: { taskId: task.id, run: 0 },
            attemptsMade: attempt,
            updateProgress: async () => {
              throw new Error('模拟队列进度写入故障');
            },
          }),
        ).rejects.toThrow('后台任务执行失败');
        const current = await db.backgroundTask.findUniqueOrThrow({
          where: { id: task.id },
        });
        expect(current.status).toBe(attempt === 2 ? 'FAILED' : 'WAITING');
        expect(current.attempts).toBe(attempt + 1);
        expect(current.resultCsv).toBeNull();
      }
      await api(admin).task.retry({ teamId, id: task.id });
      await execute(task.id);
      expect(
        (await db.backgroundTask.findUniqueOrThrow({ where: { id: task.id } }))
          .status,
      ).toBe('COMPLETED');
    });
    it('已分配工单仅提醒当前负责人', async () => {
      const member = await db.membership.create({
        data: { teamId, userId: outsider, role: 'AGENT' },
      });
      try {
        const item = await ticket();
        await db.ticket.update({
          where: { id: item.id },
          data: { dueAt: new Date(0), assigneeMembershipId: member.id },
        });
        const task = await db.backgroundTask.findFirstOrThrow({
          where: { ticketId: item.id },
        });
        await execute(task.id);
        expect(
          (
            await db.reminderNotification.findMany({
              where: { ticketId: item.id },
            })
          ).map((n) => n.userId),
        ).toEqual([outsider]);
      } finally {
        await db.ticket.updateMany({
          where: { teamId, assigneeMembershipId: member.id },
          data: { assigneeMembershipId: null },
        });
        await db.membership.delete({ where: { id: member.id } });
      }
    });
    it('导出分批读取不受列表分页限制，并报告中间进度', async () => {
      const title = `分页-${randomUUID()}`;
      await db.ticket.createMany({
        data: Array.from({ length: 205 }, () => ({
          teamId,
          creatorId: admin,
          title,
          description: '分页导出',
        })),
      });
      try {
        const task = await db.backgroundTask.create({
          data: {
            teamId,
            ownerId: admin,
            kind: 'EXPORT',
            filters: { q: title },
          },
        });
        const progress = await execute(task.id);
        expect(progress.some((p) => p > 0 && p < 95)).toBe(true);
        const csv = await downloadExport(db, admin, task.id);
        expect(csv.split(title).length - 1).toBe(205);
      } finally {
        await db.ticket.deleteMany({ where: { teamId, title } });
      }
    });
    it('超过导出条数限制时不生成不完整文件', async () => {
      const title = `上限-${randomUUID()}`;
      await db.ticket.createMany({
        data: Array.from({ length: 5001 }, () => ({
          teamId,
          creatorId: admin,
          title,
          description: '导出上限',
        })),
      });
      try {
        const task = await db.backgroundTask.create({
          data: {
            teamId,
            ownerId: admin,
            kind: 'EXPORT',
            filters: { q: title },
          },
        });
        await expect(execute(task.id)).rejects.toThrow('最多导出 5000 条');
        expect(
          (
            await db.backgroundTask.findUniqueOrThrow({
              where: { id: task.id },
            })
          ).resultCsv,
        ).toBeNull();
      } finally {
        await db.ticket.deleteMany({ where: { teamId, title } });
      }
    });
    const exportFixture = async (ownerId = admin) => {
      const marker = `导出竞争-${randomUUID()}`;
      const item = await ticket();
      await db.ticket.update({
        where: { id: item.id },
        data: { title: `${marker}-旧快照` },
      });
      const task = await exportTask(ownerId);
      await db.backgroundTask.update({
        where: { id: task.id },
        data: { filters: { q: marker } },
      });
      return { item, task, marker };
    };
    const exportJob = (
      task: { id: string; run: number },
      values: number[],
    ) => ({
      data: { taskId: task.id, run: task.run },
      attemptsMade: 0,
      updateProgress: async (value: string | number | object) => {
        values.push(value as number);
      },
    });
    it.each(['进度写入前', '最终授权后'])(
      '导出并发接手：%s 失去租约，旧执行者不能发布或报告完成',
      async (point) => {
        const { item, task, marker } = await exportFixture();
        const pausedA = gate(),
          resumeA = gate(),
          claimedB = gate(),
          resumeB = gate();
        const oldProgress: number[] = [],
          newProgress: number[] = [];
        let memberReads = 0;
        const oldDb = db.$extends({
          query: {
            ticket: {
              async findMany({ args, query }) {
                const result = await query(args);
                if (point === '进度写入前' && result.length) {
                  pausedA.open();
                  await resumeA.promise;
                }
                return result;
              },
            },
            membership: {
              async findUnique({ args, query }) {
                const result = await query(args);
                // 第一次是执行前授权，第二次是团队锁内的提交前授权。
                if (++memberReads === 2 && point === '最终授权后') {
                  pausedA.open();
                  await resumeA.promise;
                }
                return result;
              },
            },
          },
        });
        const newDb = db.$extends({
          query: {
            backgroundTask: {
              async updateMany({ args, query }) {
                const result = await query(args);
                if (args.data.status === 'RUNNING') {
                  claimedB.open();
                  await resumeB.promise;
                }
                return result;
              },
            },
          },
        });
        const running: Promise<unknown>[] = [];
        try {
          const a = processTask(
            oldDb as unknown as PrismaClient,
            exportJob(task, oldProgress),
          );
          running.push(a);
          void a.catch(() => {});
          await reached(pausedA.promise);
          const first = await db.backgroundTask.findUniqueOrThrow({
            where: { id: task.id },
          });
          // A 已读取旧标题，B 应发布新标题，避免两份相同 CSV 掩盖旧结果覆盖。
          await db.ticket.update({
            where: { id: item.id },
            data: { title: `${marker}-新快照` },
          });
          const b = processTask(
            newDb as unknown as PrismaClient,
            exportJob(task, newProgress),
          );
          running.push(b);
          void b.catch(() => {});
          await reached(claimedB.promise);
          const replacement = await db.backgroundTask.findUniqueOrThrow({
            where: { id: task.id },
          });
          expect(replacement.executionId).not.toBe(first.executionId);
          expect(replacement.executionId).not.toBeNull();
          resumeA.open();
          await a;
          expect(oldProgress).toEqual(point === '进度写入前' ? [] : [95]);
          expect(
            await db.backgroundTask.findUnique({ where: { id: task.id } }),
          ).toEqual(replacement);
          expect(replacement).toMatchObject({
            status: 'RUNNING',
            resultCsv: null,
            resultSummary: null,
            expiresAt: null,
          });
          resumeB.open();
          await b;
          const completed = await db.backgroundTask.findUniqueOrThrow({
            where: { id: task.id },
          });
          expect(completed).toMatchObject({
            status: 'COMPLETED',
            progress: 100,
            run: task.run,
            executionId: null,
            lastError: null,
            resultSummary: '已导出 1 条工单',
          });
          expect(completed.resultCsv).toContain(`${marker}-新快照`);
          expect(completed.resultCsv).not.toContain(`${marker}-旧快照`);
          expect(completed.expiresAt!.getTime()).toBeGreaterThan(Date.now());
          expect(newProgress).toEqual([95, 100]);
          expect(await downloadExport(db, admin, task.id)).toBe(
            completed.resultCsv,
          );
          // 已完成后重复执行不刷新文件内容、有效期或任务状态。
          await execute(task.id);
          expect(
            await db.backgroundTask.findUnique({ where: { id: task.id } }),
          ).toEqual(completed);
        } finally {
          resumeA.open();
          resumeB.open();
          await Promise.allSettled(running);
        }
      },
    );
    it.each(['降为客户', '移出团队'])(
      '导出读取中%s：提交前重查授权，不能发布已生成的团队文件',
      async (change) => {
        await db.membership.create({
          data: { teamId, userId: outsider, role: 'AGENT' },
        });
        const { task } = await exportFixture(outsider);
        const paused = gate(),
          resume = gate();
        const values: number[] = [];
        const job = exportJob(task, values);
        const running = processTask(db, {
          ...job,
          updateProgress: async (value) => {
            values.push(value as number);
            if (value === 95) {
              paused.open();
              await resume.promise;
            }
          },
        });
        void running.catch(() => {});
        try {
          await reached(paused.promise);
          if (change === '降为客户')
            await api(admin).team.changeRole({
              teamId,
              userId: outsider,
              role: 'CUSTOMER',
            });
          else await api(admin).team.removeMember({ teamId, userId: outsider });
          resume.open();
          await expect(running).rejects.toThrow('成员权限已变化');
          expect(values).toEqual([95]);
          expect(
            await db.backgroundTask.findUnique({ where: { id: task.id } }),
          ).toMatchObject({
            status: 'FAILED',
            executionId: null,
            resultCsv: null,
            resultSummary: null,
            expiresAt: null,
            lastError: '成员权限已变化，请重新创建导出',
          });
          await expect(
            downloadExport(db, outsider, task.id),
          ).rejects.toMatchObject({ code: 'FORBIDDEN' });
        } finally {
          resume.open();
          await Promise.allSettled([running]);
          await db.membership.deleteMany({
            where: { teamId, userId: outsider },
          });
        }
      },
    );
    it('导出旧执行者的权限拒绝晚到，不得报永久失败或破坏新租约', async () => {
      await db.membership.create({
        data: { teamId, userId: outsider, role: 'CUSTOMER' },
      });
      // 任务最初受理的是团队范围；A 读到降权状态，B 在合法恢复为客服后接手。
      const { task, marker } = await exportFixture(outsider);
      const readA = gate(),
        resumeA = gate(),
        claimedB = gate(),
        resumeB = gate();
      const oldProgress: number[] = [],
        newProgress: number[] = [];
      const oldDb = db.$extends({
        query: {
          membership: {
            async findUnique({ args, query }) {
              const result = await query(args);
              readA.open();
              await resumeA.promise;
              return result;
            },
          },
        },
      });
      const newDb = db.$extends({
        query: {
          backgroundTask: {
            async updateMany({ args, query }) {
              const result = await query(args);
              if (args.data.status === 'RUNNING') {
                claimedB.open();
                await resumeB.promise;
              }
              return result;
            },
          },
        },
      });
      const running: Promise<unknown>[] = [];
      try {
        const a = processTask(
          oldDb as unknown as PrismaClient,
          exportJob(task, oldProgress),
        );
        running.push(a);
        void a.catch(() => {});
        await reached(readA.promise);
        await api(admin).team.changeRole({
          teamId,
          userId: outsider,
          role: 'AGENT',
        });
        const b = processTask(
          newDb as unknown as PrismaClient,
          exportJob(task, newProgress),
        );
        running.push(b);
        void b.catch(() => {});
        await reached(claimedB.promise);
        const replacement = await db.backgroundTask.findUniqueOrThrow({
          where: { id: task.id },
        });
        resumeA.open();
        await a;
        expect(oldProgress).toEqual([]);
        expect(
          await db.backgroundTask.findUnique({ where: { id: task.id } }),
        ).toEqual(replacement);
        resumeB.open();
        await b;
        expect(
          await db.backgroundTask.findUnique({ where: { id: task.id } }),
        ).toMatchObject({
          status: 'COMPLETED',
          executionId: null,
          lastError: null,
          progress: 100,
        });
        expect(await downloadExport(db, outsider, task.id)).toContain(marker);
      } finally {
        resumeA.open();
        resumeB.open();
        await Promise.allSettled(running);
        await db.membership.deleteMany({ where: { teamId, userId: outsider } });
      }
    });
    it('导出最终写入后故障：正文、有效期与完成状态一起回滚，重试才发布', async () => {
      const { task, marker } = await exportFixture();
      const values: number[] = [];
      const failingDb = db.$extends({
        query: {
          backgroundTask: {
            async updateMany({ args, query }) {
              const result = await query(args);
              if (args.data.status === 'COMPLETED')
                throw new Error('测试注入：导出结果写入后失败');
              return result;
            },
          },
        },
      });
      await expect(
        processTask(
          failingDb as unknown as PrismaClient,
          exportJob(task, values),
        ),
      ).rejects.toThrow('后台任务执行失败');
      expect(values).toEqual([95]);
      expect(
        await db.backgroundTask.findUnique({ where: { id: task.id } }),
      ).toMatchObject({
        status: 'WAITING',
        progress: 95,
        executionId: null,
        resultCsv: null,
        resultSummary: null,
        expiresAt: null,
      });
      await expect(downloadExport(db, admin, task.id)).rejects.toMatchObject({
        code: 'NOT_FOUND',
      });
      await execute(task.id, 1);
      const completed = await db.backgroundTask.findUniqueOrThrow({
        where: { id: task.id },
      });
      expect(completed).toMatchObject({
        status: 'COMPLETED',
        progress: 100,
        executionId: null,
        attempts: 2,
        lastError: null,
        resultSummary: '已导出 1 条工单',
      });
      expect(completed.resultCsv).toContain(marker);
      expect(completed.expiresAt!.getTime()).toBeGreaterThan(Date.now());
      expect(await downloadExport(db, admin, task.id)).toBe(
        completed.resultCsv,
      );
    });
    it('导出最终授权与保存持有同一团队锁，降权必须等提交后生效', async () => {
      await db.membership.create({
        data: { teamId, userId: outsider, role: 'AGENT' },
      });
      const { task } = await exportFixture(outsider);
      const paused = gate(),
        resume = gate();
      let reads = 0;
      const committingDb = db.$extends({
        query: {
          membership: {
            async findUnique({ args, query }) {
              const result = await query(args);
              if (++reads === 2) {
                paused.open();
                await resume.promise;
              }
              return result;
            },
          },
        },
      });
      // 独立连接名只用于观察 PostgreSQL 的真实锁等待，不把“请求已发出”当作已阻塞。
      const connectionName = `export-permission-${randomUUID()}`;
      const permissionDb = new PrismaClient({
        adapter: new PrismaPg({
          connectionString: process.env.TEST_DATABASE_URL!,
          application_name: connectionName,
          max: 1,
        }),
      });
      const values: number[] = [];
      const running = processTask(
        committingDb as unknown as PrismaClient,
        exportJob(task, values),
      );
      void running.catch(() => {});
      let changing: Promise<unknown> | undefined;
      try {
        await reached(paused.promise);
        changing = appRouter
          .createCaller({ db: permissionDb, userId: admin })
          .team.changeRole({
            teamId,
            userId: outsider,
            role: 'CUSTOMER',
          });
        void changing.catch(() => {});
        const deadline = Date.now() + 4000;
        while (true) {
          const [state] = await db.$queryRaw<{ waiting: boolean }[]>`
            SELECT EXISTS (
              SELECT 1 FROM pg_stat_activity
              WHERE application_name = ${connectionName}
                AND cardinality(pg_blocking_pids(pid)) > 0
            ) AS waiting`;
          if (state.waiting) break;
          if (Date.now() >= deadline)
            throw new Error('降权没有等待最终提交的团队锁');
        }
        expect(
          await db.membership.findUnique({
            where: { teamId_userId: { teamId, userId: outsider } },
          }),
        ).toMatchObject({ role: 'AGENT' });
        expect(
          await db.backgroundTask.findUnique({ where: { id: task.id } }),
        ).toMatchObject({ status: 'RUNNING', resultCsv: null });
        resume.open();
        await running;
        await changing;
        const completed = await db.backgroundTask.findUniqueOrThrow({
          where: { id: task.id },
        });
        expect(completed).toMatchObject({
          status: 'COMPLETED',
          progress: 100,
          executionId: null,
          lastError: null,
        });
        expect(completed.resultCsv).not.toBeNull();
        expect(values).toEqual([95, 100]);
        // 保存发生在合法降权之前；降权生效后下载仍重新授权，账本不被倒改。
        await expect(
          downloadExport(db, outsider, task.id),
        ).rejects.toMatchObject({ code: 'FORBIDDEN' });
        expect(
          await db.backgroundTask.findUnique({ where: { id: task.id } }),
        ).toEqual(completed);
      } finally {
        resume.open();
        await Promise.allSettled([running, ...(changing ? [changing] : [])]);
        await permissionDb.$disconnect();
        await db.membership.deleteMany({ where: { teamId, userId: outsider } });
      }
    });
    it('导出已提交后汇报100失败，不把已保存文件退回等待或报永久失败', async () => {
      const { task, marker } = await exportFixture();
      await expect(
        processTask(db, {
          ...exportJob(task, []),
          updateProgress: async (value) => {
            if (value === 100) throw new Error('测试注入：队列进度汇报失败');
          },
        }),
      ).resolves.toBeUndefined();
      const completed = await db.backgroundTask.findUniqueOrThrow({
        where: { id: task.id },
      });
      expect(completed).toMatchObject({
        status: 'COMPLETED',
        progress: 100,
        executionId: null,
        lastError: null,
        resultSummary: '已导出 1 条工单',
      });
      expect(completed.expiresAt!.getTime()).toBeGreaterThan(Date.now());
      expect(await downloadExport(db, admin, task.id)).toContain(marker);
      await execute(task.id);
      expect(
        await db.backgroundTask.findUnique({ where: { id: task.id } }),
      ).toEqual(completed);
    });
    it('任务列表客户仅能查看自己的导出', async () => {
      const result = await api(customer).task.list({ teamId });
      expect(result.items.every((task) => task.kind === 'EXPORT')).toBe(true);
      await expect(api(outsider).task.list({ teamId })).rejects.toMatchObject({
        code: 'FORBIDDEN',
      });
    });
  },
);