Como Criar um Worker com Bull/BullMQ para Processar Consultas de CPF em Fila no Node.js

import { Queue } from "bullmq";
import express from "express";

const conexaoRedis = {
    host: process.env.REDIS_HOST || "127.0.0.1",
    port: parseInt(process.env.REDIS_PORT) || 6379
};

const filaCPF = new Queue("consulta-cpf", { connection: conexaoRedis });

const app = express();
app.use(express.json());

// Endpoint para enfileirar um único CPF
app.post("/api/consultar-cpf", async (req, res) => {
    const { cpf, callback_url } = req.body;

    const job = await filaCPF.add(
    "consultar",
    { cpf, callback_url },
    {
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: { age: 3600 },
    removeOnFail: { age: 86400 }
    }
    );

    res.status(202).json({
    mensagem: "Consulta enfileirada com sucesso",
    jobId: job.id,
    status: "pendente"
    });
});

// Endpoint para enfileirar lote de CPFs
app.post("/api/consultar-cpf/lote", async (req, res) => {
    const { cpfs, prioridade } = req.body;

    const jobs = await filaCPF.addBulk(
    cpfs.map((cpf, index) => ({
    name: "consultar",
    data: { cpf, loteId: Date.now().toString() },
    opts: {
    priority: prioridade || 5,
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
    delay: index * 100 // Espaçar requisições
    }
    }))
    );

    res.status(202).json({
    mensagem: `${jobs.length} CPFs enfileirados`,
    jobIds: jobs.map((j) => j.id)
    });
});

app.listen(3000, () => console.log("Servidor rodando na porta 3000"));

import { Worker } from "bullmq";

const workerCPF = new Worker(
    "consulta-cpf",
    async (job) => {
    const { cpf, callback_url } = job.data;

    console.log(
    `[Job ${job.id}] Processando CPF: ${cpf} ` +
    `(tentativa ${job.attemptsMade + 1})`
    );

    // Consultar a API de CPF
    const response = await fetch(`https://api.cpfhub.io/cpf/${cpf}`, {
    headers: { "x-api-key": process.env.CPFHUB_API_KEY }
    });

    if (!response.ok) {
    throw new Error(`API retornou status ${response.status}`);
    }

    const resultado = await response.json();

    // Atualizar progresso
    await job.updateProgress(100);

    // Notificar via callback se configurado
    if (callback_url && resultado.success) {
    await fetch(callback_url, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
    cpf,
    dados: resultado.data,
    jobId: job.id
    })
    });
    }

    return {
    cpf,
    sucesso: resultado.success,
    dados: resultado.data || null,
    processadoEm: new Date().toISOString()
    };
    },
    {
    connection: {
    host: process.env.REDIS_HOST || "127.0.0.1",
    port: parseInt(process.env.REDIS_PORT) || 6379
    },
    concurrency: 5,
    limiter: {
    max: 20,
    duration: 1000 // Máximo 20 jobs por segundo
    }
    }
);

workerCPF.on("completed", (job, resultado) => {
    console.log(`[Job ${job.id}] Concluído: ${resultado.cpf}`);
});

workerCPF.on("failed", (job, erro) => {
    console.error(`[Job ${job.id}] Falhou: ${erro.message}`);
});

workerCPF.on("error", (erro) => {
    console.error("Erro no worker:", erro);
});

import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter.js";
import { ExpressAdapter } from "@bull-board/express";

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath("/admin/filas");

createBullBoard({
    queues: [new BullMQAdapter(filaCPF)],
    serverAdapter
});

app.use("/admin/filas", serverAdapter.getRouter());

app.get("/api/consultar-cpf/status/:jobId", async (req, res) => {
    const job = await filaCPF.getJob(req.params.jobId);

    if (!job) {
    return res.status(404).json({ erro: "Job não encontrado" });
    }

    const estado = await job.getState();
    const progresso = job.progress;

    res.json({
    jobId: job.id,
    estado,
    progresso,
    dados: job.data,
    resultado: job.returnvalue || null,
    tentativas: job.attemptsMade,
    criadoEm: new Date(job.timestamp).toISOString(),
    processadoEm: job.processedOn
    ? new Date(job.processedOn).toISOString()
    : null,
    concluidoEm: job.finishedOn
    ? new Date(job.finishedOn).toISOString()
    : null
    });
});