146 أسطر
3.8 KiB
TypeScript
146 أسطر
3.8 KiB
TypeScript
import {
|
|
Injectable,
|
|
OnApplicationBootstrap,
|
|
OnModuleDestroy,
|
|
OnModuleInit,
|
|
} from '@nestjs/common';
|
|
import { ConfigService } from '@nestjs/config';
|
|
import { JobsOptions, Queue, Worker } from 'bullmq';
|
|
import { AppLoggerService } from '../logging/app-logger.service';
|
|
import { RedisService } from '../redis/redis.service';
|
|
|
|
type JobProcessor = (payload: Record<string, unknown>) => Promise<void>;
|
|
|
|
@Injectable()
|
|
export class AppQueueService
|
|
implements OnModuleInit, OnApplicationBootstrap, OnModuleDestroy
|
|
{
|
|
private readonly processors = new Map<string, JobProcessor>();
|
|
private queue: Queue | null = null;
|
|
private worker: Worker | null = null;
|
|
|
|
constructor(
|
|
private readonly configService: ConfigService,
|
|
private readonly redisService: RedisService,
|
|
private readonly logger: AppLoggerService,
|
|
) {}
|
|
|
|
onModuleInit(): void {
|
|
// Intentionally empty. Processors are usually registered by other providers before bootstrap.
|
|
}
|
|
|
|
onApplicationBootstrap(): void {
|
|
if (!this.isQueueEnabled() || !this.redisService.isEnabled()) {
|
|
return;
|
|
}
|
|
|
|
const queueName = this.getQueueName();
|
|
const queueConnection = this.redisService.createQueueClient();
|
|
const workerConnection = this.redisService.createQueueClient();
|
|
if (!queueConnection || !workerConnection) {
|
|
return;
|
|
}
|
|
|
|
this.queue = new Queue(queueName, {
|
|
connection: queueConnection,
|
|
defaultJobOptions: this.getDefaultJobOptions(),
|
|
});
|
|
|
|
this.worker = new Worker(
|
|
queueName,
|
|
async (job) => {
|
|
const processor = this.processors.get(job.name);
|
|
if (!processor) {
|
|
throw new Error(`No processor registered for job "${job.name}"`);
|
|
}
|
|
|
|
await processor(job.data as Record<string, unknown>);
|
|
},
|
|
{
|
|
connection: workerConnection,
|
|
concurrency:
|
|
this.configService.get<number>('queue.workerConcurrency', { infer: true }) ?? 5,
|
|
},
|
|
);
|
|
|
|
this.worker.on('failed', (job, error) => {
|
|
this.logger.error(
|
|
{
|
|
queue: queueName,
|
|
jobName: job?.name,
|
|
jobId: job?.id,
|
|
error: error.message,
|
|
},
|
|
undefined,
|
|
AppQueueService.name,
|
|
);
|
|
});
|
|
}
|
|
|
|
registerProcessor(jobName: string, processor: JobProcessor): void {
|
|
this.processors.set(jobName, processor);
|
|
}
|
|
|
|
async enqueue(
|
|
jobName: string,
|
|
payload: Record<string, unknown>,
|
|
options: JobsOptions = {},
|
|
): Promise<void> {
|
|
if (this.queue) {
|
|
await this.queue.add(jobName, payload, {
|
|
...this.getDefaultJobOptions(),
|
|
...options,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const processor = this.processors.get(jobName);
|
|
if (!processor) {
|
|
return;
|
|
}
|
|
|
|
queueMicrotask(() => {
|
|
void processor(payload).catch((error: Error) => {
|
|
this.logger.error(
|
|
{
|
|
jobName,
|
|
payload,
|
|
error: error.message,
|
|
},
|
|
error.stack,
|
|
AppQueueService.name,
|
|
);
|
|
});
|
|
});
|
|
}
|
|
|
|
async onModuleDestroy(): Promise<void> {
|
|
await this.worker?.close();
|
|
await this.queue?.close();
|
|
this.worker = null;
|
|
this.queue = null;
|
|
}
|
|
|
|
private isQueueEnabled(): boolean {
|
|
return this.configService.get<boolean>('queue.enabled', { infer: true }) ?? false;
|
|
}
|
|
|
|
private getQueueName(): string {
|
|
return this.configService.get<string>('queue.name', { infer: true }) ?? 'app-jobs';
|
|
}
|
|
|
|
private getDefaultJobOptions(): JobsOptions {
|
|
return {
|
|
attempts:
|
|
this.configService.get<number>('queue.defaultJobAttempts', { infer: true }) ?? 3,
|
|
backoff: {
|
|
type: 'exponential',
|
|
delay:
|
|
this.configService.get<number>('queue.defaultJobBackoffMs', { infer: true }) ?? 1000,
|
|
},
|
|
removeOnComplete:
|
|
this.configService.get<boolean>('queue.removeOnComplete', { infer: true }) ?? true,
|
|
};
|
|
}
|
|
}
|