feat(worker): publish transaction with outbox event and compensating sweep

Task 21: PublishService.processPhoto streams the original from storage, runs the sharp pipeline, puts thumb/browse variants under photo-live/dev/albums/{albumId}/variants/{photoId}/1/, then commits ONE Prisma transaction: photo_variants upserts (thumb+browse), a conditional photos update WHERE processingState IN (uploaded, processing_failed) that flips to ready+published, stamps publishedAt/width/height, clears errorCode and bumps entityVersion, and the outbox_events row (photo.published, payload = contracts PhotoPublishedPayload with the post-increment entityVersion). A zero-row conditional update means a concurrent winner already published -> idempotent 'already' with no second event and no extra version bump. The BullMQ worker wrapper (QUEUE_PHOTO_PROCESS, concurrency 2, jobId 'process-{photoId}' without the reserved ':') claims photo_jobs pending->active (attempt++), marks done for published/already/skipped alike, and rethrows pipeline/storage errors for the producer-configured 3-attempt exponential backoff; the final attempt (BullMQ increments attemptsMade only after moveToFinished, so in-processor it is 0-based: current = attemptsMade+1) marks photo_jobs failed + photo processing_failed with the PipelineError code while visibility stays unpublished. sweepOnce() claims due pending rows (nextRetryAt IS NULL OR < now) and re-enqueues stale (>5min) active rows with jobId dedupe; startSweepInterval(30s) boot-scans then loops, and main.ts wires pino logging, dedicated redis connections and graceful SIGINT/SIGTERM shutdown. Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-17 23:21:57 +08:00
parent 64a2cec788
commit 94357e9f03
5 changed files with 932 additions and 0 deletions
@@ -0,0 +1,73 @@
+// services/worker/src/main.ts —— photo worker 进程入口（Task 21）。构建产物
+// dist/main.js 由根 dev:worker / 生产启动命令运行；env 一律来自进程环境（根
+// dotenv-cli 注入），此处不读文件。启动顺序：Prisma + MinIO 适配器 → 两条专用
+// ioredis 连接（Worker 阻塞命令须独占连接；Queue/sweep 入队共用另一条）→
+// BullMQ Worker（并发 2）+ 30s 补偿扫描。优雅退出：SIGINT/SIGTERM 先停扫描，
+// worker.close()（等待在飞 job）后逐个关闭连接（Task 19 gotcha：BullMQ 不接管
+// 外部连接，须显式 quit；未监听的 error 事件会崩进程，全部挂日志监听）。
+import { Queue } from 'bullmq';
+import IORedis from 'ioredis';
+import { pino, type Logger } from 'pino';
+import { PrismaClient } from '@photo/db-cn';
+import { QUEUE_PHOTO_PROCESS } from '@photo/contracts';
+import { createStorageFromEnv } from '@photo/storage';
+import { createPhotoProcessWorker } from './process.processor';
+import { startSweepInterval } from './sweep';
+
+const SWEEP_INTERVAL_MS = 30_000;
+
+function createConnection(logger: Logger, label: string): IORedis {
+  const connection = new IORedis(process.env.REDIS_URL_CN ?? 'redis://localhost:6379', {
+    maxRetriesPerRequest: null, // BullMQ 阻塞命令要求：禁用 per-request 重试上限
+  });
+  connection.on('error', (err: Error) => {
+    logger.warn({ label, err: err.message }, 'redis 连接异常（ioredis 自动重连）');
+  });
+  return connection;
+}
+
+async function main(): Promise<void> {
+  const logger = pino({ level: process.env.LOG_LEVEL ?? 'info' });
+  const prisma = new PrismaClient();
+  const storage = createStorageFromEnv();
+  const workerConnection = createConnection(logger, 'worker');
+  const queueConnection = createConnection(logger, 'queue');
+  const sweepQueue = new Queue(QUEUE_PHOTO_PROCESS, { connection: queueConnection });
+  const worker = createPhotoProcessWorker({
+    prisma,
+    storage,
+    connection: workerConnection,
+    logger,
+  });
+  const sweep = startSweepInterval(SWEEP_INTERVAL_MS, { prisma, queue: sweepQueue, logger });
+  logger.info(
+    { queue: QUEUE_PHOTO_PROCESS, concurrency: 2, sweepIntervalMs: SWEEP_INTERVAL_MS },
+    'photo worker started',
+  );
+
+  let stopping = false;
+  const shutdown = async (signal: string): Promise<void> => {
+    if (stopping) return;
+    stopping = true;
+    logger.info({ signal }, 'photo worker shutting down');
+    sweep.stop();
+    try {
+      await worker.close(); // 等待在飞 job 完成
+      await sweepQueue.close();
+      await workerConnection.quit().catch(() => undefined);
+      await queueConnection.quit().catch(() => undefined);
+      await prisma.$disconnect();
+    } catch (err) {
+      logger.error(
+        { err: err instanceof Error ? err.message : String(err) },
+        'photo worker 关闭异常',
+      );
+    } finally {
+      process.exit(0);
+    }
+  };
+  process.once('SIGINT', () => void shutdown('SIGINT'));
+  process.once('SIGTERM', () => void shutdown('SIGTERM'));
+}
+
+void main();
@@ -0,0 +1,122 @@
+// services/worker/src/process.processor.ts —— QUEUE_PHOTO_PROCESS 消费者（Task 21）。
+// BullMQ Worker 包装 PublishService：
+//   · 每次投递将 photo_jobs pending→active（attempt 自增），成功（含幂等 already/
+//     跳过 skipped）置 done；失败原样抛给 BullMQ 按生产者配置重试（attempts:3，
+//     指数退避 1s 起——与 api queue.provider.ts 一致）；
+//   · 终败判定：BullMQ 在 moveToFailed/moveToCompleted 之后才自增 attemptsMade
+//     （v6 classes/job.js），processor 内它恒为「已完成尝试数」（0 基），故当前
+//     尝试 = attemptsMade + 1；等于 attempts 上限即终败 → 单事务置
+//     photo_jobs.failed(errorCode) + photos.processing_failed（visibility 不动，
+//     保持 unpublished——失败照片不公开），PipelineError 携带精确错误码；
+//   · Worker 使用专用 ioredis 连接（阻塞命令），连接生命周期由调用方管理
+//     （Task 19 gotcha：BullMQ 不接管外部连接；未监听的 error 事件会崩进程）。
+import { Worker, type Job } from 'bullmq';
+import type IORedis from 'ioredis';
+import type { PrismaClient } from '@photo/db-cn';
+import type { StoragePort } from '@photo/storage';
+import type { Logger } from 'pino';
+import { QUEUE_PHOTO_PROCESS } from '@photo/contracts';
+import { PipelineError } from './pipeline';
+import { PublishService, type ProcessOutcome } from './publish.service';
+
+export type { ProcessOutcome } from './publish.service';
+
+export interface PhotoProcessJobData {
+  photoId: string;
+}
+
+/** job 名称（与 api 生产者/photo_jobs.type 同值）。 */
+export const PHOTO_PROCESS_JOB_NAME = 'process';
+
+/**
+ * jobId 以 photoId 派生：并发投递/补偿扫描重复入队天然去重（BullMQ 同 jobId
+ * 不重复执行）。不含 ':'（BullMQ 保留分隔符，Task 19 gotcha）。
+ */
+export function photoProcessJobId(photoId: string): string {
+  return `${PHOTO_PROCESS_JOB_NAME}-${photoId}`;
+}
+
+/** 与 api 生产者一致的重试策略（sweep 补偿入队复用同一组值）。 */
+export const PROCESS_JOB_ATTEMPTS = 3;
+export const PROCESS_JOB_BACKOFF = { type: 'exponential', delay: 1000 } as const;
+
+/** 终败errorCode 兜底（非流水线错误：对象缺失、网络等）。 */
+export const FALLBACK_ERROR_CODE = 'PROCESS_FAILED';
+
+export interface PhotoProcessWorkerOptions {
+  prisma: PrismaClient;
+  storage: StoragePort;
+  /** Worker 专用连接（阻塞命令），调用方负责 quit。 */
+  connection: IORedis;
+  concurrency?: number;
+  logger?: Logger;
+}
+
+export function createPhotoProcessWorker(
+  options: PhotoProcessWorkerOptions,
+): Worker<PhotoProcessJobData, ProcessOutcome, string> {
+  const { prisma } = options;
+  const publish = new PublishService(prisma, options.storage, options.logger);
+
+  const processor = async (
+    job: Job<PhotoProcessJobData>,
+  ): Promise<ProcessOutcome> => {
+    const photoId = job.data.photoId;
+    // 投递即认领：pending→active（sweep 已置 active 的重复投递幂等），attempt 计数。
+    await prisma.photo_jobs.updateMany({
+      where: {
+        photoId,
+        type: PHOTO_PROCESS_JOB_NAME,
+        state: { in: ['pending', 'active'] },
+      },
+      data: { state: 'active', attempt: { increment: 1 } },
+    });
+    try {
+      const outcome = await publish.processPhoto(photoId);
+      // published/already/skipped 对 job 而言都是「无需再投递」→ done。
+      await prisma.photo_jobs.updateMany({
+        where: {
+          photoId,
+          type: PHOTO_PROCESS_JOB_NAME,
+          state: { in: ['pending', 'active'] },
+        },
+        data: { state: 'done', errorCode: null, nextRetryAt: null },
+      });
+      options.logger?.info({ photoId, jobId: job.id, outcome }, 'process job 完成');
+      return outcome;
+    } catch (err) {
+      const isFinalAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1);
+      if (isFinalAttempt) {
+        const errorCode =
+          err instanceof PipelineError ? err.code : FALLBACK_ERROR_CODE;
+        options.logger?.error(
+          { photoId, jobId: job.id, errorCode },
+          'process job 终败（重试耗尽）',
+        );
+        await prisma.$transaction([
+          prisma.photo_jobs.updateMany({
+            where: { photoId, type: PHOTO_PROCESS_JOB_NAME },
+            data: { state: 'failed', errorCode, nextRetryAt: null },
+          }),
+          prisma.photos.updateMany({
+            where: {
+              id: photoId,
+              processingState: { in: ['uploaded', 'processing_failed'] },
+            },
+            data: { processingState: 'processing_failed', errorCode },
+          }),
+        ]);
+      }
+      throw err; // 交还 BullMQ：未耗尽则按退避重试，耗尽则入 failed set
+    }
+  };
+
+  return new Worker<PhotoProcessJobData, ProcessOutcome, string>(
+    QUEUE_PHOTO_PROCESS,
+    processor,
+    {
+      connection: options.connection,
+      concurrency: options.concurrency ?? 2,
+    },
+  );
+}
@@ -0,0 +1,181 @@
+// services/worker/src/publish.service.ts —— 发布事务（Task 21，「上传即公开」的心脏）。
+// 处理成功 = 单个 Prisma 事务内原子完成三件事：
+//   ① photo_variants 两条 upsert（thumb/browse）；
+//   ② photos 条件更新（WHERE processingState IN ('uploaded','processing_failed')）：
+//      ready + published + publishedAt + 尺寸 + entityVersion+1（每次产生事件的
+//      状态迁移自增；失败重试成功也切 published——上传即公开语义）；
+//   ③ outbox_events 写入 photo.published（payload = 契约 PhotoPublishedPayload，
+//      entityVersion 与 ② 提交后一致；P2 消费按 eventId 去重、entityVersion 单调）。
+// 条件更新匹配 0 行（并发/重复投递已处理过）→ 幂等 no-op（'already'），不再
+// bump entityVersion、不再补 outbox——事务即幂等边界。
+// 变体对象先于事务写入 MinIO（键含 version 段，内容确定，重复写无害）；流式拉取
+// 原图整缓冲驻内存（50MB 上限内可接受，P5 再考虑流式）。
+import { Readable } from 'node:stream';
+import type { Prisma, PrismaClient } from '@photo/db-cn';
+import type { StoragePort } from '@photo/storage';
+import type { PhotoPublishedPayload } from '@photo/contracts';
+import type { Logger } from 'pino';
+import { runPipeline } from './pipeline';
+
+/** 对象键根（与 api src/uploads/support.ts 的 OBJECT_KEY_ROOT 一致：P1a 单环境）。 */
+const OBJECT_KEY_ROOT = 'photo-live/dev';
+
+/** 变体键版本段（对应 photo_variants.version；水印/规格升级时递增）。 */
+const VARIANT_VERSION = 1;
+
+/** 可处理态：首次上传（uploaded）与失败重试（processing_failed，终态仍 published）。 */
+const PROCESSABLE_STATES = ['uploaded', 'processing_failed'] as const;
+
+export type ProcessOutcome =
+  | 'published' // 本次处理并切 published（含失败重试成功）
+  | 'already' // 重复投递：已 ready（或并发赢家已处理），幂等 no-op
+  | 'skipped'; // 照片不存在 / 状态不可处理（pending_upload、processing 等）
+
+export function variantsObjectKey(
+  albumId: string,
+  photoId: string,
+  version: number,
+  variant: 'thumb' | 'browse',
+): string {
+  return `${OBJECT_KEY_ROOT}/albums/${albumId}/variants/${photoId}/${version}/${variant}.jpg`;
+}
+
+export class PublishService {
+  constructor(
+    private readonly prisma: PrismaClient,
+    private readonly storage: StoragePort,
+    private readonly logger?: Logger,
+  ) {}
+
+  /**
+   * 处理一张照片：拉原图 → sharp 流水线 → 上传变体 → 原子发布事务。
+   * PipelineError/存储错误原样抛出（processor 交给 BullMQ 重试，3 次耗尽置
+   * processing_failed）。不抛错路径绝不留下半发布状态。
+   */
+  async processPhoto(photoId: string): Promise<ProcessOutcome> {
+    const photo = await this.prisma.photos.findUnique({
+      where: { id: photoId },
+      select: { albumId: true, objectKey: true, processingState: true },
+    });
+    if (photo === null) {
+      this.logger?.warn({ photoId }, '照片不存在，跳过处理');
+      return 'skipped';
+    }
+    if (photo.processingState === 'ready') {
+      // 快路径：已发布（重复投递）。真正的幂等保障仍是事务内条件更新——
+      // 并发窗口内两方都读到 uploaded 时，恰一方 claim 成功。
+      return 'already';
+    }
+    if (!(PROCESSABLE_STATES as readonly string[]).includes(photo.processingState)) {
+      this.logger?.warn(
+        { photoId, processingState: photo.processingState },
+        '照片状态不可处理，跳过',
+      );
+      return 'skipped';
+    }
+
+    const input = await this.readOriginal(photo.objectKey);
+    const pipeline = await runPipeline(input); // 失败抛 PipelineError → 重试
+    const thumbKey = variantsObjectKey(photo.albumId, photoId, VARIANT_VERSION, 'thumb');
+    const browseKey = variantsObjectKey(photo.albumId, photoId, VARIANT_VERSION, 'browse');
+    await this.storage.put(thumbKey, pipeline.thumb, 'image/jpeg');
+    await this.storage.put(browseKey, pipeline.browse, 'image/jpeg');
+
+    const publishedAt = new Date();
+    const outcome = await this.prisma.$transaction(async (tx) => {
+      // 条件更新 = 处理权抢占：WHERE 仍处可处理态。匹配 0 → 已被处理，幂等返回。
+      const claimed = await tx.photos.updateMany({
+        where: { id: photoId, processingState: { in: [...PROCESSABLE_STATES] } },
+        data: {
+          processingState: 'ready',
+          visibility: 'published',
+          publishedAt,
+          width: pipeline.width,
+          height: pipeline.height,
+          errorCode: null, // processing_failed 重试成功后清除错误码
+          entityVersion: { increment: 1 },
+        },
+      });
+      if (claimed.count === 0) {
+        return 'already' as const;
+      }
+      const claimedPhoto = await tx.photos.findUniqueOrThrow({
+        where: { id: photoId },
+        select: { entityVersion: true },
+      });
+      await tx.photo_variants.upsert({
+        where: { photoId_variant: { photoId, variant: 'thumb' } },
+        update: {
+          objectKey: thumbKey,
+          bytes: pipeline.thumb.length,
+          version: VARIANT_VERSION,
+          width: pipeline.thumbWidth,
+          height: pipeline.thumbHeight,
+        },
+        create: {
+          photoId,
+          variant: 'thumb',
+          objectKey: thumbKey,
+          bytes: pipeline.thumb.length,
+          version: VARIANT_VERSION,
+          width: pipeline.thumbWidth,
+          height: pipeline.thumbHeight,
+        },
+      });
+      await tx.photo_variants.upsert({
+        where: { photoId_variant: { photoId, variant: 'browse' } },
+        update: {
+          objectKey: browseKey,
+          bytes: pipeline.browse.length,
+          version: VARIANT_VERSION,
+          width: pipeline.browseWidth,
+          height: pipeline.browseHeight,
+        },
+        create: {
+          photoId,
+          variant: 'browse',
+          objectKey: browseKey,
+          bytes: pipeline.browse.length,
+          version: VARIANT_VERSION,
+          width: pipeline.browseWidth,
+          height: pipeline.browseHeight,
+        },
+      });
+      const payload: PhotoPublishedPayload = {
+        photoId,
+        albumId: photo.albumId,
+        objectKey: photo.objectKey,
+        thumbKey,
+        browseKey,
+        width: pipeline.width,
+        height: pipeline.height,
+        publishedAt: publishedAt.toISOString(),
+        entityVersion: claimedPhoto.entityVersion,
+      };
+      await tx.outbox_events.create({
+        data: {
+          type: 'photo.published',
+          albumId: photo.albumId,
+          photoId,
+          entityVersion: claimedPhoto.entityVersion,
+          // 契约接口无 Prisma Json 输入所需的索引签名，结构化拷贝落 Json 列。
+          payload: payload as unknown as Prisma.InputJsonValue,
+          occurredAt: publishedAt,
+        },
+      });
+      return 'published' as const;
+    });
+    this.logger?.info({ photoId, outcome }, '照片处理完成');
+    return outcome;
+  }
+
+  /** 流式拉取原图并聚合为单缓冲（S3 Body → chunks → concat）。 */
+  private async readOriginal(objectKey: string): Promise<Buffer> {
+    const stream = (await this.storage.getStream(objectKey)) as Readable;
+    const chunks: Buffer[] = [];
+    for await (const chunk of stream) {
+      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array));
+    }
+    return Buffer.concat(chunks);
+  }
+}
@@ -0,0 +1,132 @@
+// 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) };
+}