87 lines
3.4 KiB
TypeScript
87 lines
3.4 KiB
TypeScript
|
// services/api/src/uploads/queue.provider.ts —— photo-process 队列生产者（Task 19
|
|||
|
|
// 入队；队列名常量 QUEUE_PHOTO_PROCESS 由 @photo/contracts 导出，worker（Task 20/21）
|
|||
|
|
// 消费同一队列）。lazy-safe 约定：
|
|||
|
|
// · Queue 实例（含 ioredis 连接）在首次 enqueue 时才创建——Redis 不可用时
|
|||
|
|
// 应用启动/请求管线均不受影响（BullMQ 连接仅在入队时使用）；
|
|||
|
|
// · 入队失败（连接拒绝/超时等）只记日志不抛错：complete 事务已提交，
|
|||
|
|
// photo_jobs(pending) 行是持久事实源，Task 21 的补偿扫描会认领孤儿任务
|
|||
|
|
// （总纲 §5.6「complete 后入队失败」的补偿路径）；
|
|||
|
|
// · ioredis 的 'error' 事件挂 no-op 监听：重连风暴期间未监听的 error 事件
|
|||
|
|
// 会让进程崩溃，这里吞掉并交给 enqueue 的 catch 记日志。
|
|||
|
|
import { Injectable, Logger, type OnModuleDestroy } from '@nestjs/common';
|
|||
|
|
import { Queue } from 'bullmq';
|
|||
|
|
import IORedis from 'ioredis';
|
|||
|
|
import { QUEUE_PHOTO_PROCESS } from '@photo/contracts';
|
|||
|
|
|
|||
|
|
/** DI token。 */
|
|||
|
|
export const PHOTO_PROCESS_QUEUE = Symbol('PHOTO_PROCESS_QUEUE');
|
|||
|
|
|
|||
|
|
/** 生产者窄接口：服务只依赖 enqueue，便于测试替身。 */
|
|||
|
|
export interface PhotoProcessQueueProducer {
|
|||
|
|
enqueuePhotoProcess(photoId: string): Promise<void>;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** BullMQ job 名称（与 photo_jobs.type 同值）。 */
|
|||
|
|
const PHOTO_PROCESS_JOB_NAME = 'process';
|
|||
|
|
|
|||
|
|
@Injectable()
|
|||
|
|
export class BullMqPhotoProcessQueue
|
|||
|
|
implements PhotoProcessQueueProducer, OnModuleDestroy
|
|||
|
|
{
|
|||
|
|
private readonly logger = new Logger(BullMqPhotoProcessQueue.name);
|
|||
|
|
private queue: Queue | null = null;
|
|||
|
|
|
|||
|
|
/** 首次使用时构造；连接错误经 no-op 监听吞掉（见文件头注释）。 */
|
|||
|
|
private getQueue(): Queue {
|
|||
|
|
if (this.queue === null) {
|
|||
|
|
const url = process.env.REDIS_URL_CN ?? 'redis://localhost:6379';
|
|||
|
|
const connection = new IORedis(url, {
|
|||
|
|
// BullMQ 对阻塞命令的要求：禁用 per-request 重试上限（等待 Worker 语义）。
|
|||
|
|
maxRetriesPerRequest: null,
|
|||
|
|
});
|
|||
|
|
connection.on('error', (err: Error) => {
|
|||
|
|
this.logger.warn(
|
|||
|
|
`photo-process 队列连接异常（将由 ioredis 自动重连）: ${err.message}`,
|
|||
|
|
);
|
|||
|
|
});
|
|||
|
|
this.queue = new Queue(QUEUE_PHOTO_PROCESS, { connection });
|
|||
|
|
}
|
|||
|
|
return this.queue;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async enqueuePhotoProcess(photoId: string): Promise<void> {
|
|||
|
|
try {
|
|||
|
|
await this.getQueue().add(
|
|||
|
|
PHOTO_PROCESS_JOB_NAME,
|
|||
|
|
{ photoId },
|
|||
|
|
{
|
|||
|
|
// jobId 以 photoId 派生：并发 complete/补偿扫描重复入队天然去重（A03）。
|
|||
|
|
jobId: `${PHOTO_PROCESS_JOB_NAME}:${photoId}`,
|
|||
|
|
// Task 21 消费侧重试策略：3 次指数退避。
|
|||
|
|
attempts: 3,
|
|||
|
|
backoff: { type: 'exponential', delay: 1000 },
|
|||
|
|
},
|
|||
|
|
);
|
|||
|
|
} catch (err) {
|
|||
|
|
// 不抛错：photo_jobs(pending) 已是持久事实源，Task 21 sweep 补偿。
|
|||
|
|
this.logger.error(
|
|||
|
|
`photo-process 入队失败（sweep 将补偿）photoId=${photoId}: ${
|
|||
|
|
err instanceof Error ? err.message : String(err)
|
|||
|
|
}`,
|
|||
|
|
);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async onModuleDestroy(): Promise<void> {
|
|||
|
|
if (this.queue !== null) {
|
|||
|
|
await this.queue.close().catch(() => undefined);
|
|||
|
|
this.queue = null;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
export const photoProcessQueueProvider = {
|
|||
|
|
provide: PHOTO_PROCESS_QUEUE,
|
|||
|
|
useClass: BullMqPhotoProcessQueue,
|
|||
|
|
};
|