photo-live/services/worker/src/sweep.ts

// services/worker/src/sweep.ts —— photo_jobs 补偿扫描（Task 21，总纲 §5.6）。
// photo_jobs(pending) 是持久事实源：api complete 事务提交后入队失败（Redis 抖动/
// 宕机）留下孤儿 pending，sweep 周期性认领并重新入队。两类认领：
//   ① state='pending' AND (nextRetryAt IS NULL OR nextRetryAt < now)：
//      NULL 语义 = 立即可投递（api complete 新建的行不带 nextRetryAt）；
//   ② state='active' AND updatedAt < now-5min：worker 认领后崩溃/重启留下的
//      僵尸 active（BullMQ stalled 恢复之外的 DB 侧兜底），只重入队不改状态
//      （jobId 去重：BullMQ 里仍活跃的 job 不会重复执行）。
// 入队失败仅记日志（下一轮 sweep 自然重试）；done/failed 不再认领。
import type { Queue } from 'bullmq';
import type { PrismaClient } from '@photo/db-cn';
import type { Logger } from 'pino';
import {
  PHOTO_PROCESS_JOB_NAME,
  PROCESS_JOB_ATTEMPTS,
  PROCESS_JOB_BACKOFF,
  photoProcessJobId,
} from './process.processor';

/** active 僵尸阈值：超过该时长未更新的 active 视为 worker 已失联。 */
const STALE_ACTIVE_MS = 5 * 60_000;

export interface SweepDeps {
  prisma: PrismaClient;
  queue: Queue;
  logger?: Logger;
}

export interface SweepResult {
  /** 认领为 active 的到期 pending 行数。 */
  claimed: number;
  /** 因入队失败未补偿成功的行数（下一轮重试）。 */
  enqueueFailures: number;
  /** 僵尸 active 重新入队行数。 */
  staleReEnqueued: number;
}

async function enqueueProcess(deps: SweepDeps, photoId: string): Promise<boolean> {
  try {
    await deps.queue.add(
      PHOTO_PROCESS_JOB_NAME,
      { photoId },
      {
        jobId: photoProcessJobId(photoId),
        attempts: PROCESS_JOB_ATTEMPTS,
        backoff: PROCESS_JOB_BACKOFF,
      },
    );
    return true;
  } catch (err) {
    deps.logger?.error(
      { photoId, err: err instanceof Error ? err.message : String(err) },
      'sweep 入队失败（下一轮补偿）',
    );
    return false;
  }
}

/** 单轮扫描：可反复调用（e2e 直接驱动）；幂等——已入队 jobId 会被 BullMQ 去重。 */
export async function sweepOnce(deps: SweepDeps): Promise<SweepResult> {
  const now = new Date();

  // ① 到期 pending：IS NULL（立即可投递）或到期重试。
  const due = await deps.prisma.photo_jobs.findMany({
    where: {
      state: 'pending',
      type: PHOTO_PROCESS_JOB_NAME,
      OR: [{ nextRetryAt: null }, { nextRetryAt: { lt: now } }],
    },
    select: { photoId: true },
  });
  let claimed = 0;
  let enqueueFailures = 0;
  if (due.length > 0) {
    claimed = (
      await deps.prisma.photo_jobs.updateMany({
        where: {
          photoId: { in: due.map((row) => row.photoId) },
          type: PHOTO_PROCESS_JOB_NAME,
          state: 'pending', // 重检：并发 sweep 恰一方胜出
        },
        data: { state: 'active' },
      })
    ).count;
    for (const row of due) {
      if (!(await enqueueProcess(deps, row.photoId))) {
        enqueueFailures += 1;
      }
    }
  }

  // ② 僵尸 active：仅重入队（jobId 去重），不改 DB 状态。
  const stale = await deps.prisma.photo_jobs.findMany({
    where: {
      state: 'active',
      updatedAt: { lt: new Date(Date.now() - STALE_ACTIVE_MS) },
    },
    select: { photoId: true },
  });
  let staleReEnqueued = 0;
  for (const row of stale) {
    if (await enqueueProcess(deps, row.photoId)) {
      staleReEnqueued += 1;
    } else {
      enqueueFailures += 1;
    }
  }

  deps.logger?.info({ claimed, enqueueFailures, staleReEnqueued }, 'sweep 完成');
  return { claimed, enqueueFailures, staleReEnqueued };
}

/**
 * 周期扫描（main 以 30s 启动）：启动即扫一轮（进程重启后立刻补偿积压），随后
 * 固定间隔。返回 stop() 供优雅退出。
 */
export function startSweepInterval(
  intervalMs: number,
  deps: SweepDeps,
): { stop: () => void } {
  const run = () => {
    void sweepOnce(deps).catch((err: unknown) => {
      deps.logger?.error(
        { err: err instanceof Error ? err.message : String(err) },
        'sweep 轮次异常',
      );
    });
  };
  run();
  const timer = setInterval(run, intervalMs);
  return { stop: () => clearInterval(timer) };
}