Módulo BullMQ

O módulo BullMQ de @grupodiariodaregiao/bunstone fornece uma forma de processar tarefas em segundo plano usando BullMQ.

Instalação

Primeiro, garanta que você tenha as dependências necessárias:

bun add bullmq ioredis

Configuração

Para usar BullMQ, você precisa registrar o BullMqModule no seu AppModule.

import { BullMqModule, Module } from "@grupodiariodaregiao/bunstone";

@Module({
  imports: [
    BullMqModule.register({
      host: 'localhost',
      port: 6379,
    }),
  ],
})
export class AppModule {}

Criando um processor

Um processor é uma classe decorada com @Processor(). Ela contém métodos decorados com @Process() que lidam com jobs de uma fila específica.

import { Process, Processor } from "@grupodiariodaregiao/bunstone";
import { Job } from "bullmq";

@Processor('email-queue')
export class EmailProcessor {
  @Process('send-welcome')
  async handleSendWelcome(job: Job) {
    console.log('Enviando e-mail de boas-vindas para:', job.data.email);
    // Simula trabalho
    await new Promise(resolve => setTimeout(resolve, 1000));
    return { sent: true };
  }

  @Process()
  async handleDefault(job: Job) {
    console.log('Processando tipo de job desconhecido:', job.name);
  }
}

Não se esqueça de adicionar seu processor aos providers de um módulo.

Produzindo jobs

Você pode injetar o QueueService nos seus controllers ou services para adicionar jobs a uma fila.

import {
  Controller,
  Get,
  Query,
  QueueService,
} from "@grupodiariodaregiao/bunstone";

@Controller('/email')
export class EmailController {
  constructor(private readonly queueService: QueueService) {}

  @Get('/send')
  async sendEmail(@Query('email') email: string) {
    await this.queueService.add('email-queue', 'send-welcome', { email });
    return { status: 'Job adicionado à fila' };
  }
}

Exemplo completo

Explore um fluxo completo com producer + processor:

import {
	AppStartup,
	BullMqModule,
	Controller,
	Get,
	Module,
	Process,
	Processor,
	QueueService,
} from "../../index";
import { Job } from "bullmq";

// 1. Define a Job Processor
@Processor({
	queueName: "mail-queue",
	concurrency: 2,
})
export class MailProcessor {
	@Process("welcome-email")
	async handleWelcomeEmail(job: Job) {
		console.log(`[Worker] Processing welcome email for: ${job.data.email}`);
		// Simulate some work
		await new Promise((resolve) => setTimeout(resolve, 1000));
		console.log(`[Worker] Welcome email sent to ${job.data.email}`);
		return { success: true, recipient: job.data.email };
	}

	@Process()
	async handleGenericJob(job: Job) {
		console.log(`[Worker] Processing generic job: ${job.name}`);
	}
}

// 2. Define a Controller to produce jobs
@Controller("/jobs")
export class JobsController {
	constructor(private readonly queueService: QueueService) {}

	@Get("/add")
	async addJob() {
		const email = `user-${Math.floor(Math.random() * 1000)}@example.com`;
		console.log(`[Controller] Adding welcome-email job for ${email}`);

		await this.queueService.add("mail-queue", "welcome-email", { email });

		return {
			message: "Job added to queue",
			email,
		};
	}
}

// 3. Setup the Application Module
@Module({
	imports: [
		BullMqModule.register({
			host: process.env.REDIS_HOST || "localhost",
			port: Number(process.env.REDIS_PORT) || 6379,
		}),
	],
	controllers: [JobsController],
	providers: [MailProcessor],
})
class AppModule {}

// 4. Start the app
console.log("Starting BullMQ example app...");
console.log("Make sure you have a Redis instance running at localhost:6379");
const app = await AppStartup.create(AppModule);
app.listen(3000);

Veja no GitHub

Opções do processor

O decorator @Processor aceita um objeto de opções:

@Processor({
  queueName: 'heavy-tasks',
  concurrency: 5,
})
export class HeavyTaskProcessor {
  // ...
}