From 94357e9f0362788a5d7d68aa658b1f3617d1071b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 17 Sep 2026 23:21:57 +0800 Subject: [PATCH] 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 --- services/worker/src/main.ts | 73 ++++ services/worker/src/process.processor.ts | 122 +++++++ services/worker/src/publish.service.ts | 181 ++++++++++ services/worker/src/sweep.ts | 132 +++++++ services/worker/test/publish.e2e-spec.ts | 424 +++++++++++++++++++++++ 5 files changed, 932 insertions(+) create mode 100644 services/worker/src/main.ts create mode 100644 services/worker/src/process.processor.ts create mode 100644 services/worker/src/publish.service.ts create mode 100644 services/worker/src/sweep.ts create mode 100644 services/worker/test/publish.e2e-spec.ts diff --git a/services/worker/src/main.ts b/services/worker/src/main.ts new file mode 100644 index 0000000..cb2486a --- /dev/null +++ b/services/worker/src/main.ts @@ -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 { + 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 => { + 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(); diff --git a/services/worker/src/process.processor.ts b/services/worker/src/process.processor.ts new file mode 100644 index 0000000..81bb336 --- /dev/null +++ b/services/worker/src/process.processor.ts @@ -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 { + const { prisma } = options; + const publish = new PublishService(prisma, options.storage, options.logger); + + const processor = async ( + job: Job, + ): Promise => { + 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( + QUEUE_PHOTO_PROCESS, + processor, + { + connection: options.connection, + concurrency: options.concurrency ?? 2, + }, + ); +} diff --git a/services/worker/src/publish.service.ts b/services/worker/src/publish.service.ts new file mode 100644 index 0000000..db8ee4d --- /dev/null +++ b/services/worker/src/publish.service.ts @@ -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 { + 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 { + 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); + } +} diff --git a/services/worker/src/sweep.ts b/services/worker/src/sweep.ts new file mode 100644 index 0000000..ee18a1d --- /dev/null +++ b/services/worker/src/sweep.ts @@ -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 { + 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 { + 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) }; +} diff --git a/services/worker/test/publish.e2e-spec.ts b/services/worker/test/publish.e2e-spec.ts new file mode 100644 index 0000000..9768a03 --- /dev/null +++ b/services/worker/test/publish.e2e-spec.ts @@ -0,0 +1,424 @@ +// services/worker/test/publish.e2e-spec.ts —— Task 21 发布事务/补偿扫描 e2e（先写后实现）。 +// 真实 MySQL（Prisma 直种 album/photos/photo_jobs）+ 真实 MinIO（storage.put 真 +// JPEG 原图，变体 head 复核）+ 真实 Redis（BullMQ）。 +// 分层驱动以规避 BullMQ 时序抖动：核心断言直接调 PublishService/sweepOnce； +// Worker 包装经一组真实入队用例覆盖（happy / 终败 / 跳过）。 +// 覆盖： +// ① 发布事务：变体对象 + photo_variants(2) + photos(ready/published/尺寸/ +// entityVersion+1) + outbox(photo.published，payload=契约字段全集) 单事务一致； +// ② 重复处理幂等：无二次变体/事件，entityVersion 不再自增； +// ③ 状态不可处理（pending_upload）→ 跳过； +// ④ sweepOnce：到期 pending（nextRetryAt 过期 / NULL）认领入队，done 不再认领， +// 僵尸 active（updatedAt>5min）重入队； +// ⑤ 真实 Worker：happy 路径 done；坏文件 3 次指数退避后 photo processing_failed +// （visibility 仍 unpublished）+ job failed(attempt=3)；pending_upload 跳过置 done。 +import { randomBytes } from 'node:crypto'; +import { Queue, type Worker } from 'bullmq'; +import IORedis from 'ioredis'; +import type { Logger } from 'pino'; +import sharp from 'sharp'; +import { PrismaClient } from '@photo/db-cn'; +import type { PhotoPublishedPayload } from '@photo/contracts'; +import { + createStorageFromEnv, + type MinioStorageAdapter, +} from '@photo/storage'; +import { + PROCESS_JOB_ATTEMPTS, + PROCESS_JOB_BACKOFF, + PHOTO_PROCESS_JOB_NAME, + createPhotoProcessWorker, + photoProcessJobId, + type PhotoProcessJobData, + type ProcessOutcome, +} from '../src/process.processor'; +import { PublishService } from '../src/publish.service'; +import { sweepOnce } from '../src/sweep'; +import { resetDb } from './reset'; + +jest.setTimeout(120_000); + +/** 轮询等待：e2e 对 BullMQ 异步路径的断言前置。 */ +async function waitFor( + check: () => Promise, + timeoutMs = 30_000, + stepMs = 100, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (await check()) return; + if (Date.now() > deadline) { + throw new Error(`waitFor 超时（${timeoutMs}ms）`); + } + await new Promise((resolve) => setTimeout(resolve, stepMs)); + } +} + +/** cuid 形状 ID（对象键含 photoId，须先有 ID 再一次 insert 成形，同 api 约定）。 */ +function cuidLike(alias: string): string { + return `c${Date.now().toString(36)}${alias}${randomBytes(6).toString('hex')}`; +} + +/** 测试内生成真实 JPEG（无二进制入库）。 */ +async function makeJpeg(width: number, height: number): Promise { + return sharp({ + create: { width, height, channels: 3, background: { r: 180, g: 120, b: 60 } }, + }) + .jpeg({ quality: 85 }) + .toBuffer(); +} + +describe('publish (e2e)', () => { + let prisma: PrismaClient; + let storage: MinioStorageAdapter; + let publish: PublishService; + let queue: Queue; + let queueConnection: IORedis; + + const KEY_ROOT = 'photo-live/dev'; + + interface SeedResult { + albumId: string; + photoId: string; + objectKey: string; + } + + /** 直种 album + photo(+可选原图对象) + photo_jobs(pending)。 */ + async function seedPhoto( + alias: string, + options: { + state?: string; + jobState?: string; + nextRetryAt?: Date | null; + updatedAt?: Date; + original?: Buffer | null; // null = 不写对象（或调用方自写） + } = {}, + ): Promise { + const suffix = randomBytes(4).toString('hex'); + const albumId = ( + await prisma.albums.create({ + data: { + publicToken: `pt-${alias}-${suffix}`, + ownerId: `u-${alias}`, + startAt: new Date(), + defaultLocale: 'zh-CN', + }, + select: { id: true }, + }) + ).id; + const photoId = cuidLike(alias); + const objectKey = `${KEY_ROOT}/albums/${albumId}/originals/${photoId}.jpg`; + const original = + options.original === undefined + ? await makeJpeg(800, 500) + : options.original; + if (original !== null) { + await storage.put(objectKey, original, 'image/jpeg'); + } + await prisma.photos.create({ + data: { + id: photoId, + albumId, + uploaderId: `u-${alias}`, + objectKey, + bytes: original?.length ?? 0, + processingState: options.state ?? 'uploaded', + }, + }); + await prisma.photo_jobs.create({ + data: { + photoId, + type: PHOTO_PROCESS_JOB_NAME, + state: options.jobState ?? 'pending', + nextRetryAt: options.nextRetryAt ?? null, + ...(options.updatedAt !== undefined ? { updatedAt: options.updatedAt } : {}), + }, + }); + return { albumId, photoId, objectKey }; + } + + /** 变体对象键（与 src/publish.service.ts 的 variantsObjectKey 同布局的断言字面量）。 */ + const variantKey = (albumId: string, photoId: string, variant: string) => + `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/${variant}.jpg`; + + beforeAll(async () => { + await resetDb(); + prisma = new PrismaClient(); + storage = createStorageFromEnv(); + publish = new PublishService(prisma, storage); + queueConnection = new IORedis( + process.env.REDIS_URL_CN ?? 'redis://localhost:6379', + { maxRetriesPerRequest: null }, + ); + queueConnection.on('error', () => undefined); // 测试环境不崩进程 + queue = new Queue('photo-process', { connection: queueConnection }); + }); + + afterAll(async () => { + await queue.obliterate({ force: true }).catch(() => undefined); + await queue.close().catch(() => undefined); + await queueConnection.quit().catch(() => undefined); + await prisma.$disconnect(); + }); + + describe('PublishService.processPhoto（直接驱动）', () => { + test('发布事务：变体上传 + 三表单事务一致（photos/photo_variants/outbox）', async () => { + const { albumId, photoId, objectKey } = await seedPhoto('pub1'); + const outcome = await publish.processPhoto(photoId); + expect(outcome).toBe('published'); + + // MinIO：thumb/browse 真实存在（head 复核，jpeg + 非零字节）。 + const thumbHead = await storage.head(variantKey(albumId, photoId, 'thumb')); + const browseHead = await storage.head(variantKey(albumId, photoId, 'browse')); + expect(thumbHead).not.toBeNull(); + expect(browseHead).not.toBeNull(); + expect(thumbHead?.contentType).toBe('image/jpeg'); + expect(thumbHead?.bytes).toBeGreaterThan(0); + expect(browseHead?.bytes).toBeGreaterThan(0); + + // photos：ready/published/publishedAt/尺寸/entityVersion 1→2。 + const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.processingState).toBe('ready'); + expect(photo.visibility).toBe('published'); + expect(photo.publishedAt).not.toBeNull(); + expect(photo.width).toBe(800); + expect(photo.height).toBe(500); + expect(photo.entityVersion).toBe(2); + expect(photo.errorCode).toBeNull(); + + // photo_variants：两行，键/尺寸/版本一致（800x500 → thumb 480x300， + // browse 不放大 800x500）。 + const variants = await prisma.photo_variants.findMany({ + where: { photoId }, + orderBy: { variant: 'asc' }, // browse < thumb（字典序） + }); + expect(variants).toHaveLength(2); + const [browseRow, thumbRow] = variants; + expect(thumbRow).toMatchObject({ + variant: 'thumb', + objectKey: variantKey(albumId, photoId, 'thumb'), + version: 1, + width: 480, + height: 300, + }); + expect(thumbRow.bytes).toBe(thumbHead?.bytes); + expect(browseRow).toMatchObject({ + variant: 'browse', + objectKey: variantKey(albumId, photoId, 'browse'), + version: 1, + width: 800, + height: 500, + }); + expect(browseRow.bytes).toBe(browseHead?.bytes); + + // outbox：类型/entityVersion 对齐 photo，payload = 契约 PhotoPublishedPayload 全集。 + const event = await prisma.outbox_events.findFirstOrThrow({ + where: { photoId }, + }); + expect(event.type).toBe('photo.published'); + expect(event.albumId).toBe(albumId); + expect(event.entityVersion).toBe(2); + expect(event.occurredAt.getTime()).toBe(photo.publishedAt?.getTime()); + const expectedPayload: PhotoPublishedPayload = { + photoId, + albumId, + objectKey, + thumbKey: variantKey(albumId, photoId, 'thumb'), + browseKey: variantKey(albumId, photoId, 'browse'), + width: 800, + height: 500, + publishedAt: (photo.publishedAt as Date).toISOString(), + entityVersion: 2, + }; + expect(event.payload).toEqual(expectedPayload); + // 全库仅此一条事件（无重复/孤儿）。 + expect(await prisma.outbox_events.count()).toBe(1); + }); + + test('重复处理幂等：无二次变体/事件，entityVersion 不再自增', async () => { + const { photoId } = await seedPhoto('pub2'); + expect(await publish.processPhoto(photoId)).toBe('published'); + const before = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + + const again = await publish.processPhoto(photoId); + expect(again).toBe('already'); + + const after = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(after.entityVersion).toBe(before.entityVersion); + expect(after.publishedAt?.getTime()).toBe(before.publishedAt?.getTime()); + expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(2); + expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(1); + }); + + test('状态不可处理（pending_upload）→ 跳过，不留任何处理痕迹', async () => { + const { photoId } = await seedPhoto('pub3', { state: 'pending_upload' }); + expect(await publish.processPhoto(photoId)).toBe('skipped'); + const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.processingState).toBe('pending_upload'); + expect(photo.visibility).toBe('unpublished'); + expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0); + expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0); + }); + }); + + describe('sweepOnce（直接驱动，无 Worker 在跑）', () => { + test('认领到期 pending（过期/NULL）并重排，done 不认领，僵尸 active 重排', async () => { + // 前置：直驱用例留下的 pending job 行（照片其实已处理）先置 done—— + // 既贴近真实（api 侧 job 随处理闭环），也让本用例的认领计数可精确断言。 + await prisma.photo_jobs.updateMany({ + where: { state: 'pending' }, + data: { state: 'done' }, + }); + + const past = await seedPhoto('sw-a', { + state: 'pending_upload', + nextRetryAt: new Date(Date.now() - 60_000), + }); // ① nextRetryAt 过期 + const nullRetry = await seedPhoto('sw-b', { state: 'pending_upload' }); // ② NULL + const done = await seedPhoto('sw-c', { + state: 'pending_upload', + jobState: 'done', + }); // done：不再认领 + const stale = await seedPhoto('sw-d', { + state: 'pending_upload', + jobState: 'active', + updatedAt: new Date(Date.now() - 6 * 60_000), + }); // 僵尸 active（>5min） + + const result = await sweepOnce({ prisma, queue }); + expect(result.claimed).toBe(2); + expect(result.enqueueFailures).toBe(0); + expect(result.staleReEnqueued).toBe(1); + + // DB：pending→active（过期与 NULL 都认领）；done 不动；active 保持。 + const jobs = await prisma.photo_jobs.findMany({ + where: { photoId: { in: [past.photoId, nullRetry.photoId, done.photoId, stale.photoId] } }, + }); + const stateOf = (photoId: string) => + jobs.find((j) => j.photoId === photoId)?.state; + expect(stateOf(past.photoId)).toBe('active'); + expect(stateOf(nullRetry.photoId)).toBe('active'); + expect(stateOf(done.photoId)).toBe('done'); + expect(stateOf(stale.photoId)).toBe('active'); + + // BullMQ：认领/重排的 jobId 已入队；done 的 jobId 不存在。 + expect(await queue.getJob(photoProcessJobId(past.photoId))).toBeDefined(); + expect(await queue.getJob(photoProcessJobId(nullRetry.photoId))).toBeDefined(); + expect(await queue.getJob(photoProcessJobId(stale.photoId))).toBeDefined(); + expect(await queue.getJob(photoProcessJobId(done.photoId))).toBeUndefined(); + + // 清空队列，后续真实 Worker 用例从干净状态开始。 + await queue.obliterate({ force: true }); + }); + }); + + describe('真实 BullMQ Worker', () => { + let worker: Worker; + let workerConnection: IORedis; + const logger = console as unknown as Logger; // e2e 直观日志 + + beforeAll(async () => { + workerConnection = new IORedis( + process.env.REDIS_URL_CN ?? 'redis://localhost:6379', + { maxRetriesPerRequest: null }, + ); + workerConnection.on('error', () => undefined); + worker = createPhotoProcessWorker({ + prisma, + storage, + connection: workerConnection, + logger, + }); + }); + + afterAll(async () => { + await worker.close(); + await workerConnection.quit().catch(() => undefined); + }); + + /** 与 api 生产者同参入队（jobId 去重 + attempts:3 指数退避 1s 起）。 */ + const enqueue = (photoId: string) => + queue.add( + PHOTO_PROCESS_JOB_NAME, + { photoId }, + { + jobId: photoProcessJobId(photoId), + attempts: PROCESS_JOB_ATTEMPTS, + backoff: PROCESS_JOB_BACKOFF, + }, + ); + + test('happy：真实入队 → 处理 → 发布事务 + photo_jobs done', async () => { + const { albumId, photoId } = await seedPhoto('wk-ok'); + await enqueue(photoId); + + await waitFor(async () => + (await prisma.photos.findUnique({ where: { id: photoId } }))?.processingState === 'ready', + ); + const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.visibility).toBe('published'); + expect(photo.entityVersion).toBe(2); + expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(2); + expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(1); + expect( + await storage.head(variantKey(albumId, photoId, 'thumb')), + ).not.toBeNull(); + + const job = await prisma.photo_jobs.findUniqueOrThrow({ + where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } }, + }); + expect(job.state).toBe('done'); + expect(job.attempt).toBe(1); + }); + + test('坏文件：3 次退避重试后终败，photo processing_failed 且不公开', async () => { + const corrupt = Buffer.concat([ + Buffer.from([0xff, 0xd8, 0xff, 0xe0]), + randomBytes(8192), + ]); + const { photoId } = await seedPhoto('wk-bad', { original: corrupt }); + await enqueue(photoId); + + await waitFor( + async () => + (await prisma.photo_jobs.findUnique({ + where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } }, + }))?.state === 'failed', + 60_000, + ); + + const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.processingState).toBe('processing_failed'); + expect(photo.errorCode).toBe('DECODE_FAILED'); + expect(photo.visibility).toBe('unpublished'); // 失败不公开 + expect(photo.publishedAt).toBeNull(); + + const job = await prisma.photo_jobs.findUniqueOrThrow({ + where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } }, + }); + expect(job.state).toBe('failed'); + expect(job.errorCode).toBe('DECODE_FAILED'); + expect(job.attempt).toBe(3); + + expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0); + expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0); + }); + + test('pending_upload 跳过：job 置 done，照片零处理', async () => { + const { photoId } = await seedPhoto('wk-skip', { state: 'pending_upload' }); + await enqueue(photoId); + + await waitFor( + async () => + (await prisma.photo_jobs.findUnique({ + where: { photoId_type: { photoId, type: PHOTO_PROCESS_JOB_NAME } }, + }))?.state === 'done', + ); + const photo = await prisma.photos.findUniqueOrThrow({ where: { id: photoId } }); + expect(photo.processingState).toBe('pending_upload'); + expect(photo.visibility).toBe('unpublished'); + expect(await prisma.photo_variants.count({ where: { photoId } })).toBe(0); + expect(await prisma.outbox_events.count({ where: { photoId } })).toBe(0); + }); + }); +});