Agentes-Inc/Skills

Padrões BullMQ

Guia rápido: Use BullMQ (v5.x) para processamento de tarefas em segundo plano, agendamento de tarefas e orquestração de fluxos de trabalho sobre Redis. Classes principais: Queue (adiciona tarefas), Worker (processa tarefas), QueueEvents (ouvinte de eventos globais), FlowProducer (árvores de tarefas pai-filho). Sempre passe um objeto connection para cada construtor (obrigatório no v5). Defina maxRetriesPerRequest: null nas conexões ioredis para Workers. Use upsertJobScheduler para tarefas repetitivas/cron (substitui a API repetitiva descontinuada). QueueScheduler foi removido no v4 — suas responsabilidades agora são tratadas automaticamente pelos Workers.

---

<critical_requirements>

CRÍTICO: Antes de Usar Este Skill

Todo código deve seguir as convenções do projeto em CLAUDE.md (kebab-case, named exports, ordenação de imports, import type, constantes nomeadas)

(Você DEVE passar um objeto connection para cada construtor de Queue, Worker, QueueEvents e FlowProducer — BullMQ v5 lança exceção se connection estiver faltando)

(Você DEVE definir maxRetriesPerRequest: null nas conexões ioredis usadas por Workers — BullMQ requer tentativas infinitas e lança exceção sem essa configuração)

(Você DEVE chamar await worker.close() no SIGTERM/SIGINT para desligamento gracioso — sem isso, tarefas em andamento ficam paradas)

(Você DEVE usar upsertJobScheduler para tarefas repetitivas/cron — a opção antiga repeat no queue.add está obsoleta desde v5.16.0)

</critical_requirements>

---

Exemplos

• Padrões Principais — Configuração de Queue, processamento Worker, opções de tarefa, fábrica de conexão, desligamento gracioso, tarefas tipadas
• Padrões Avançados — FlowProducer, limite de taxa, agendamento de tarefas, QueueEvents, concorrência, processadores em sandbox

Recursos adicionais:

• reference.md — Estruturas de decisão, referência de opções de tarefa, antipadrões, lista de verificação de produção

---

Detecção automática: BullMQ, bullmq, Queue, Worker, QueueEvents, FlowProducer, fila de tarefas, tarefa em segundo plano, processo worker, agendador de tarefas, upsertJobScheduler, limitador de taxa, prioridade de tarefa, atraso de tarefa, processador em sandbox, tarefa repetitiva, tarefa cron, produtor de fluxo, tarefas pai-filho

Quando usar:

• Processamento em segundo plano (envio de e-mail, processamento de imagens, geração de PDF)
• Tarefas agendadas/cron (relatórios noturnos, limpeza periódica)
• Orquestração de fluxo de trabalho com dependências de tarefas pai-filho (FlowProducer)
• Consumo de API com limite de taxa (limitação de requisições de saída)
• Processamento de tarefas baseado em prioridade (tarefas urgentes antes de operações em lote)
• Distribuição de trabalho intensivo de CPU entre múltiplos workers ou máquinas

Principais padrões abordados:

• Configuração de Queue e Worker com dados de tarefa e valores de retorno tipados
• Fábrica de conexão com maxRetriesPerRequest: null para Workers
• Opções de tarefa: atraso, prioridade, tentativas, backoff, removeOnComplete/Fail
• Desligamento gracioso com worker.close() em sinais de processo
• FlowProducer para árvores de tarefas pai-filho com rastreamento de dependência
• Agendadores de tarefas para tarefas repetitivas/cron (upsertJobScheduler)
• Limitação de taxa (limitador global e Worker.RateLimitError manual)
• Controle de concorrência (local por worker e global)
• QueueEvents para monitoramento global de eventos entre todos os workers
• Processadores em sandbox para trabalho intensivo de CPU

Quando NÃO usar:

• Temporizadores simples no processo ou setTimeout (sem necessidade de persistência)
• Mensageria pub/sub em tempo real sem persistência (use sua solução pub/sub)
• Dados que devem ser processados sincronamente dentro de um ciclo requisição-resposta
• Filas que não precisam de persistência, tentativas ou agendamento

---

Filosofia

BullMQ é uma fila de tarefas baseada em Redis para Node.js que fornece processamento confiável em segundo plano com garantias de entrega pelo menos uma vez. O princípio central: separar a produção de tarefas do consumo de tarefas para que sua aplicação permaneça responsiva enquanto o trabalho ocorre assincronamente.

Princípios principais:

1. Produtores e consumidores são desacoplados — Qualquer processo pode adicionar tarefas a uma fila; qualquer Worker pode processá-las. Isso permite escalabilidade horizontal adicionando mais Workers.
2. Tarefas são persistentes — Tarefas sobrevivem a reinicializações de processo porque ficam no Redis. Tarefas de um Worker que falhou são retomadas por outros Workers (ou pelo mesmo Worker após reinicialização).
3. Entrega pelo menos uma vez — BullMQ garante que cada tarefa seja processada pelo menos uma vez. Use processadores idempotentes para lidar com o (raro) caso de processamento duplicado após uma parada.
4. Falhe graciosamente com tentativas — Configure estratégias de attempts e backoff para que falhas transitórias sejam resolvidas automaticamente. Tarefas com falha permanente vão para o conjunto de falha para inspeção.
5. Cada Queue/Worker/QueueEvents precisa de sua própria conexão — BullMQ gerencia o estado da conexão internamente. Nunca compartilhe uma única instância ioredis entre múltiplas classes BullMQ (exceto Queues que atuam apenas como produtoras).

---

Padrões Principais

Padrão 1: Fábrica de Conexão

BullMQ v5 requer uma conexão Redis explícita em cada construtor. Workers precisam de maxRetriesPerRequest: null para que o ioredis tente novamente indefinidamente em vez de desistir.

import Redis from "ioredis";

function createBullMQConnection(): Redis {
  const url = process.env.REDIS_URL;
  if (!url) throw new Error("REDIS_URL é obrigatória");
  return new Redis(url, { maxRetriesPerRequest: null });
}
export { createBullMQConnection };

Por que é bom: maxRetriesPerRequest: null satisfaz o requisito do BullMQ, a fábrica garante configuração consistente, a variável de ambiente mantém credenciais fora do código

// Ruim — faltando maxRetriesPerRequest
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ lança: "maxRetriesPerRequest deve ser null"

Por que é ruim: BullMQ requer tentativas infinitas em conexões de Worker e lançará exceção na inicialização sem null

Consulte examples/core.md para a fábrica de conexão completa com separação produtor vs consumidor.

---

Padrão 2: Configuração de Queue e Worker

Queue adiciona tarefas; Worker as processa. Ambos aceitam genéricos TypeScript para dados de tarefa e valores de retorno com segurança de tipos.

import { Queue, Worker, type Job } from "bullmq";

interface EmailJobData {
  to: string;
  subject: string;
  body: string;
}

const QUEUE_NAME = "emails";

const emailQueue = new Queue<EmailJobData>(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

const emailWorker = new Worker<EmailJobData>(
  QUEUE_NAME,
  async (job: Job<EmailJobData>) => {
    await sendEmail(job.data.to, job.data.subject, job.data.body);
  },
  { connection: createBullMQConnection() },
);

Por que é bom: Genéricos impõem segurança de tipos em job.data, conexões separadas para Queue e Worker, constante nomeada para o nome da fila

Consulte examples/core.md para configuração completa com eventos, tratamento de erros e valores de retorno tipados.

---

Padrão 3: Opções de Tarefa

Controle o comportamento da tarefa com opções: atraso, prioridade, tentativas, backoff e remoção automática.

const MAX_ATTEMPTS = 5;
const BACKOFF_DELAY_MS = 1000;
const DELAY_MS = 60_000;
const KEEP_COMPLETED_COUNT = 100;

await emailQueue.add(
  "send-welcome",
  { to: "[email protected]", subject: "Bem-vindo", body: "..." },
  {
    delay: DELAY_MS,
    priority: 1, // 1 = maior prioridade, números maiores = menor prioridade
    attempts: MAX_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
    removeOnComplete: { count: KEEP_COMPLETED_COUNT },
    removeOnFail: false,
  },
);

Por que é bom: Constantes nomeadas para todos os valores numéricos, backoff exponencial para falhas transitórias, removeOnComplete com count evita crescimento ilimitado da memória Redis, removeOnFail: false preserva tarefas com falha para depuração

Consulte examples/core.md para defaultJobOptions, addBulk e padrões LIFO.

---

Padrão 4: Desligamento Gracioso

Chame worker.close() em sinais de processo para finalizar tarefas em andamento antes de sair. Sem isso, as tarefas ficam paradas e são reprocessadas por outros Workers.

async function shutdown(workers: Worker[]): Promise<void> {
  await Promise.all(workers.map((w) => w.close()));
  process.exit(0);
}

process.on("SIGTERM", () => shutdown([emailWorker]));
process.on("SIGINT", () => shutdown([emailWorker]));

Por que é bom: close() para de aceitar novas tarefas e aguarda a conclusão das tarefas em andamento, Promise.all lida com múltiplos workers, os manipuladores de sinal cobrem SIGTERM e SIGINT

Pegadinha: worker.close() não tem timeout embutido. Se um processador travar, o desligamento trava. Envolva com seu próprio timeout se necessário.

Consulte examples/core.md para desligamento com padrão de timeout.

---

Padrão 5: FlowProducer (Tarefas Pai-Filho)

FlowProducer cria árvores de tarefas onde tarefas pai aguardam que todas as filhas sejam concluídas. Toda a árvore é adicionada atomicamente.

import { FlowProducer } from "bullmq";

const flowProducer = new FlowProducer({ connection: createBullMQConnection() });

await flowProducer.add({
  name: "publish-video",
  queueName: "videos",
  children: [
    { name: "extract-audio", queueName: "media", data: { videoId: "abc" } },
    {
      name: "generate-thumbnail",
      queueName: "media",
      data: { videoId: "abc" },
    },
    {
      name: "transcode",
      queueName: "media",
      data: { videoId: "abc", format: "mp4" },
    },
  ],
});

Por que é bom: O pai aguarda até que todos os filhos sejam concluídos, adição atômica (tudo ou nada), os filhos podem estar em filas diferentes, o processador pai pode acessar os resultados dos filhos via job.getChildrenValues()

Consulte examples/advanced.md para acessar valores dos filhos e fluxos aninhados.

---

Padrão 6: Agendadores de Tarefas (Tarefas Repetitivas/Cron)

Use upsertJobScheduler (v5.16.0+) para tarefas repetitivas. Substitui a opção obsoleta repeat.

const REPORT_INTERVAL_MS = 3_600_000; // 1 hora

// Intervalo fixo
await emailQueue.upsertJobScheduler(
  "hourly-digest",
  { every: REPORT_INTERVAL_MS },
  {
    name: "send-digest",
    data: { type: "hourly" },
  },
);

// Expressão Cron — diariamente às 3:15 AM
await emailQueue.upsertJobScheduler(
  "nightly-cleanup",
  { pattern: "0 15 3 * * *" },
  {
    name: "cleanup",
    data: { type: "nightly" },
    opts: { removeOnComplete: true },
  },
);

Por que é bom: upsertJobScheduler é idempotente (seguro chamar em cada inicialização), o modelo define nome/dados/opções da tarefa herdadas por cada tarefa produzida, sintaxe cron via pattern

Pegadinha: Intervalos every alinham-se ao relógio, não ao momento em que você chamou upsertJobScheduler. Um intervalo de 2000ms dispara em 0s, 2s, 4s, etc.

Consulte examples/advanced.md para remover agendadores e listar agendadores ativos.

---

Padrão 7: Limitação de Taxa

Limite quantas tarefas um Worker processa por janela de tempo. O limitador é global entre todos os Workers na mesma fila.

const RATE_LIMIT_MAX = 10;
const RATE_LIMIT_DURATION_MS = 1000;

const worker = new Worker("api-calls", processor, {
  connection: createBullMQConnection(),
  limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
});

Por que é bom: Limite de taxa global (10 workers com esta configuração ainda processam no máximo 10 tarefas/segundo no total), constantes nomeadas para limites

Para limitação de taxa dinâmica baseada em respostas de API externa, use worker.rateLimit(duration) + throw Worker.RateLimitError(). Consulte examples/advanced.md.

---

Padrão 8: QueueEvents para Monitoramento Global

QueueEvents usa Redis Streams (não Pub/Sub), então os eventos são confiáveis mesmo em reconexões. Requer sua própria conexão dedicada.

import { QueueEvents } from "bullmq";

const queueEvents = new QueueEvents(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

queueEvents.on("completed", ({ jobId, returnvalue }) => {
  console.log(`Tarefa ${jobId} concluída com: ${returnvalue}`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.error(`Tarefa ${jobId} falhou: ${failedReason}`);
});

Por que é bom: Monitora todos os Workers em uma fila a partir de um único ouvinte, entrega confiável via Redis Streams, conexão separada conforme necessário

Consulte examples/advanced.md para rastreamento de progresso e espera por conclusão de tarefa específica.

---

Otimização de Performance

• Concorrência — Defina concurrency nas opções do Worker para processar múltiplas tarefas em paralelo dentro de uma única instância Worker: { concurrency: 10 }. Só é eficaz para trabalho vinculado a I/O. Trabalho vinculado a CPU bloqueia o loop de eventos e causa tarefas paradas — use processadores em sandbox.
• Processadores em sandbox — Passe um caminho de arquivo em vez de uma função para o construtor Worker para executar processadores em um processo separado (ou thread de worker com useWorkerThreads: true). Impede que trabalho intensivo de CPU bloqueie a renovação de bloqueio.
• addBulk — Use queue.addBulk([...]) para adicionar muitas tarefas em uma única ida e volta ao Redis em vez de chamar queue.add() em um loop.
• removeOnComplete/Fail — Configure a remoção automática para evitar crescimento ilimitado da memória Redis. Use { count: N } para manter as últimas N tarefas para depuração.
• Conexões separadas — Produtores (Queue) podem compartilhar uma conexão. Workers e QueueEvents precisam de sua própria conexão devido a comandos Redis bloqueantes.

---

<decision_framework>

Estrutura de Decisão

Quando Usar BullMQ

Você precisa de processamento de tarefas em segundo plano?
|-- NÃO -> Não use BullMQ
+-- SIM -> Você precisa de persistência, tentativas ou agendamento?
    |-- NÃO -> Uma fila simples no processo ou setTimeout pode ser suficiente
    +-- SIM -> Você precisa de dependências de tarefas pai-filho?
        |-- SIM -> BullMQ com FlowProducer
        +-- NÃO -> Você precisa de limitação de taxa ou prioridade?
            |-- SIM -> BullMQ com opções de limitador/prioridade
            +-- NÃO -> BullMQ com Queue + Worker básicos

Qual Padrão de Tarefa?

Que tipo de agendamento de tarefas você precisa?
|-- Tarefa única com atraso -> queue.add() com opção de atraso

…

[truncado para tradução]