浅念
81 lines
3.3 KiB
TypeScript
81 lines
3.3 KiB
TypeScript
// services/worker/src/watermark.processor.ts —— QUEUE_WATERMARK 消费者（P4a
|
||
// Task 3）。与 photo-process 消费者同构的 Worker 包装：
|
||
// · 照片级失败在 WatermarkRegenService 内部消化（failed 计数 + 继续），不抛错；
|
||
// · 任务级错误（完成事务乐观重试耗尽、DB 异常等）抛给 BullMQ 重试（3 次指数
|
||
// 退避 1s 起）；终败判定同 process.processor（BullMQ 在 moveToFailed 之后才
|
||
// 自增 attemptsMade，当前尝试 = attemptsMade + 1），耗尽置 job 行 failed +
|
||
// errorCode 兜底；服务级中止（配置缺失等）由服务自行落 failed，不进重试。
|
||
// jobId 即 watermark_jobs.id（cuid，无 ':'——BullMQ 保留分隔符约束天然满足）。
|
||
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_WATERMARK } from '@photo/contracts';
|
||
import { WatermarkRegenService, type WatermarkJobOutcome } from './watermark.service';
|
||
|
||
export interface WatermarkRegenJobData {
|
||
jobId: string;
|
||
}
|
||
|
||
/** job 名称（api 生产者同值）。 */
|
||
export const WATERMARK_REGEN_JOB_NAME = 'regen';
|
||
|
||
/** 与 api 生产者一致的重试策略。 */
|
||
export const WATERMARK_JOB_ATTEMPTS = 3;
|
||
export const WATERMARK_JOB_BACKOFF = { type: 'exponential', delay: 1000 } as const;
|
||
|
||
/** 终败 errorCode 兜底（非服务级中止码：网络/DB 等意外错误）。 */
|
||
export const WATERMARK_FALLBACK_ERROR_CODE = 'WATERMARK_REGEN_FAILED';
|
||
|
||
export interface WatermarkRegenWorkerOptions {
|
||
prisma: PrismaClient;
|
||
storage: StoragePort;
|
||
/** Worker 专用连接（阻塞命令），调用方负责 quit。 */
|
||
connection: IORedis;
|
||
concurrency?: number;
|
||
logger?: Logger;
|
||
}
|
||
|
||
export function createWatermarkRegenWorker(
|
||
options: WatermarkRegenWorkerOptions,
|
||
): Worker<WatermarkRegenJobData, WatermarkJobOutcome, string> {
|
||
const { prisma } = options;
|
||
const service = new WatermarkRegenService(prisma, options.storage, options.logger);
|
||
|
||
const processor = async (
|
||
job: Job<WatermarkRegenJobData>,
|
||
): Promise<WatermarkJobOutcome> => {
|
||
const jobId = job.data.jobId;
|
||
try {
|
||
const outcome = await service.processJob(jobId);
|
||
options.logger?.info({ jobId, outcome }, 'watermark job 完成');
|
||
return outcome;
|
||
} catch (err) {
|
||
const isFinalAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1);
|
||
if (isFinalAttempt) {
|
||
options.logger?.error(
|
||
{ jobId, err: err instanceof Error ? err.message : String(err) },
|
||
'watermark job 终败（重试耗尽）',
|
||
);
|
||
await prisma.watermark_jobs
|
||
.updateMany({
|
||
where: { id: jobId, state: { in: ['pending', 'running', 'failed'] } },
|
||
data: { state: 'failed', errorCode: WATERMARK_FALLBACK_ERROR_CODE },
|
||
})
|
||
.catch(() => undefined); // 落库失败仅丢审计位，不掩盖原始错误
|
||
}
|
||
throw err; // 交还 BullMQ：未耗尽则按退避重试，耗尽则入 failed set
|
||
}
|
||
};
|
||
|
||
return new Worker<WatermarkRegenJobData, WatermarkJobOutcome, string>(
|
||
QUEUE_WATERMARK,
|
||
processor,
|
||
{
|
||
connection: options.connection,
|
||
concurrency: options.concurrency ?? 1, // 相册级任务串行起步（多相册仍并行）
|
||
},
|
||
);
|
||
}
|