tests/fixtures/worker-process.ts

import { PrismaPg } from '@prisma/adapter-pg';
import { PrismaClient } from '../../src/generated/prisma/client';
import { createRedis } from '../../src/server/jobs/queue';
import { createWorkerLogger } from '../../src/server/jobs/logger';
import { runWorkerProcess } from '../../src/server/jobs/runtime';

// 此入口只负责夹具和 IPC 关卡；信号、心跳、关闭期限均执行正式生命周期代码。
const config = JSON.parse(process.env.WORKER_PROCESS_FIXTURE!);
const waits = new Map<string, () => void>();
process.on('message', (message) => {
  const label = (message as { resume?: string }).resume;
  if (label) waits.get(label)?.();
});
async function checkpoint(label: string) {
  await new Promise<void>((resolve) => {
    waits.set(label, resolve);
    process.send?.({ checkpoint: label });
  });
}
const db = new PrismaClient({
  adapter: new PrismaPg({
    connectionString: process.env.DATABASE_URL!,
    max: 5,
  }),
});
let dispatches = 0;
const scoped = db.$extends({
  query: {
    backgroundTask: {
      async findMany({ args, query }) {
        args.where = { AND: [args.where ?? {}, { teamId: config.teamId }] };
        if (++dispatches === 2 && config.pauseDispatch)
          await checkpoint('dispatch');
        return query(args);
      },
      async updateMany({ args, query }) {
        args.where = { AND: [args.where ?? {}, { teamId: config.teamId }] };
        const result = await query(args);
        if (config.pause === 'claim' && args.data.status === 'RUNNING')
          await checkpoint('claim');
        return result;
      },
      async findFirst({ args, query }) {
        const result = await query(args);
        if (config.pause === 'read' && args.where?.id === config.taskId)
          await checkpoint('read');
        return result;
      },
    },
    ticket: {
      async findMany({ args, query }) {
        const result = await query(args);
        if (
          config.pause === 'read' &&
          result.some((item) => item.id === config.ticketId)
        )
          await checkpoint('read');
        return result;
      },
    },
  },
});
const logger = createWorkerLogger({
  write(line) {
    process.stdout.write(line);
    if (JSON.parse(String(line)).event === 'worker.stopped')
      process.disconnect?.();
  },
});
runWorkerProcess(scoped as unknown as PrismaClient, createRedis(true), {
  logger,
  heartbeatIntervalMs: 100,
  heartbeatTtlMs: config.heartbeatTtlMs ?? 600,
  shutdownTimeoutMs: config.shutdownTimeoutMs ?? 6000,
  workerOptions: { concurrency: 1, lockDuration: 1000, stalledInterval: 100 },
});