Filas com BullMQ no Axxos Indica: O Que Aprendi Processando Jobs em Segundo Plano

A regra de negócio do Axxos Indica parecia bobinha: quando um lead indicado fecha contrato, o indicador ganha uma recompensa. Mas o pagamento da recompensa não é imediato: tem uma janela de 30 dias de carência (para cobrir cancelamento de contrato), depois entra uma esteira de aprovação fiscal, depois é creditado.

Minha primeira versão foi um setTimeout(processarRecompensa, 30 * dia) dentro do handler que registrava o contrato fechado. Tudo na memória do Node. Eu sabia que era ruim. Achei que ia "trocar depois". Como sempre que se acha isso, o "depois" só chegou depois do incidente.

O problema com setTimeout em Node não é a API. É que ele depende do processo continuar vivo pelos próximos 30 dias. Em uma startup que faz deploy várias vezes por semana, isso é uma piada. Qualquer reinício, deploy, OOM, node spot recolhido pela AWS, leva todos os timers junto. Nada disso aparece em log porque nada explodiu. Você só descobre quando alguém pergunta "ó, cadê minha indicação?".

Antes do BullMQ eu já tinha lido sobre filas. Sabia o que era um producer e um consumer. Mas continuava raciocinando como se fosse uma chamada de função: "quando eu chamar isso, ele executa". É raciocínio errado para fila.

O modelo certo é: o produtor registra uma intenção de que algo aconteça. A fila garante durabilidade dessa intenção. O worker realiza essa intenção, possivelmente muito tempo depois, possivelmente em outro processo, possivelmente várias vezes (se falhar e retentar). Tudo o que você escreve do lado do produtor precisa fazer sentido sem saber quem, quando ou quantas vezes o worker vai rodar.

Isso muda o que você coloca no payload. No setTimeout original eu passava uma closure com referência ao objeto contrato em memória. No BullMQ você passa um JSON serializável, então o payload precisa ter tudo o que o worker precisa para tomar decisão sozinho. ID do contrato, sim. Snapshot dos campos relevantes no momento da intenção, também sim, porque o contrato pode mudar entre a indicação e a execução do job.

// src/recompensas/recompensas.producer.ts
import { Injectable } from "@nestjs/common"
import { InjectQueue } from "@nestjs/bullmq"
import { Queue } from "bullmq"
 
interface AgendarRecompensaJob {
  contratoId: string
  indicadorId: string
  valorCentavos: number
  // snapshot do estado no momento da indicação
  // worker decide se ainda é válido na hora de executar
  snapshotEm: string
}
 
@Injectable()
export class RecompensasProducer {
  constructor(
    @InjectQueue("recompensas")
    private readonly fila: Queue<AgendarRecompensaJob>
  ) {}
 
  async agendarParaDaquiTrintaDias(payload: AgendarRecompensaJob) {
    await this.fila.add("processar-recompensa", payload, {
      // chave única evita duplicação se o produtor for chamado duas vezes
      jobId: `recompensa:${payload.contratoId}`,
      delay: 30 * 24 * 60 * 60 * 1000,
      attempts: 5,
      backoff: { type: "exponential", delay: 60_000 },
      removeOnComplete: { age: 7 * 24 * 60 * 60, count: 1000 },
      removeOnFail: false, // mantém para inspeção manual
    })
  }
}

A linha que mais importa é o jobId. Sem ela, se o handler de contrato fechado for chamado duas vezes (por retry de webhook, por exemplo), você agenda duas recompensas. Com ela, o BullMQ rejeita o segundo add silenciosamente porque já existe um job com aquele ID. É idempotência grátis no produtor.

A primeira versão do worker era inocente: pegava o job, conferia se o contrato ainda estava ativo, e creditava a recompensa. Rodou bem por uns dias. Aí o Redis teve uma falha transiente de rede, alguns jobs reentraram para retry, e descobri que tinha clientes recebendo recompensa duplicada.

O retry estava me mordendo. O BullMQ retenta o job inteiro se ele falhar: inclusive depois de eu já ter dado o INSERT na tabela de recompensa, mas antes de marcar o job como concluído. Em um cenário de falha de rede entre o worker e o Redis, o INSERT acontece, o worker tenta reportar sucesso ao Redis, falha, o BullMQ assume que o job não terminou e devolve para outro worker.

A solução foi tornar o worker idempotente em relação à recompensa, não só ao job:

// src/recompensas/recompensas.worker.ts
import { Processor, WorkerHost } from "@nestjs/bullmq"
import { Job } from "bullmq"
import { Injectable, Logger } from "@nestjs/common"
 
@Processor("recompensas")
@Injectable()
export class RecompensasWorker extends WorkerHost {
  private readonly logger = new Logger(RecompensasWorker.name)
 
  constructor(
    private readonly contratos: ContratosRepository,
    private readonly recompensas: RecompensasRepository
  ) {
    super()
  }
 
  async process(job: Job<AgendarRecompensaJob>): Promise<void> {
    const { contratoId, indicadorId, valorCentavos } = job.data
 
    // Chave de idempotência derivada do job, não do timestamp
    const idempotencyKey = `recompensa:${contratoId}`
 
    const jaExiste = await this.recompensas.findByIdempotencyKey(idempotencyKey)
    if (jaExiste) {
      this.logger.warn(`Job ${job.id} já processado, ignorando`, {
        idempotencyKey,
      })
      return
    }
 
    const contrato = await this.contratos.findById(contratoId)
    if (!contrato || contrato.status !== "ATIVO") {
      // Não é erro: o contrato foi cancelado na carência. Job concluído sem efeito.
      this.logger.log(`Contrato ${contratoId} não está ativo, sem recompensa`)
      return
    }
 
    await this.recompensas.creditar({
      idempotencyKey,
      indicadorId,
      valorCentavos,
      origem: { tipo: "INDICACAO", contratoId },
    })
  }
}

A coluna idempotencyKey na tabela tem UNIQUE constraint. Mesmo que dois workers processem o mesmo job em paralelo por algum bug bizarro, só um vai conseguir o INSERT. O outro estoura constraint violation, eu trato como sucesso, e o saldo do indicador continua correto.

Sem métrica, fila vira buraco negro. Você só fica sabendo que tem problema quando o cliente reclama, e nesse ponto o problema já está grande. Os três dashboards que olho toda manhã:

Profundidade da fila por estado: waiting, delayed, active, failed. Se delayed está crescendo, é esperado (jobs agendados acumulam). Se waiting está crescendo, meus workers não estão dando conta. Se failed está crescendo, algum bug subiu no último deploy.

Tempo médio de espera entre o job entrar na fila e começar a ser processado. É o número que diz se a fila está saudável. Quando ele começa a subir e não desce em 5 minutos, é hora de escalar workers ou investigar.

Taxa de retry. Job que vai pra retry uma vez é normal: Redis tossiu, terceiro deu timeout. Job que vai pra retry três vezes geralmente é bug determinístico no worker. Eu mando alerta quando passa de 2% de retry no agregado da hora.

Eu colocaria fila desde o primeiro dia para qualquer trabalho assíncrono. Não como over-engineering, como higiene. O custo de subir BullMQ em um projeto que já tem Redis é uma classe e meia. O custo de descobrir que perdeu eventos em produção é incalculável em confiança.

Também colocaria a dead letter queue desde o início. No começo eu deixei removeOnFail: true achando que ia simplificar. Resultado: quando um bug subiu e fez 200 jobs falharem, não tinha como reprocessar. Hoje todo job que esgota tentativas vai para uma fila separada que um cron varre toda noite e me notifica. Reprocessar manual é estranho, mas é melhor do que não saber.

E pararia de ver fila como "infra chata que processa em segundo plano". Filas são contrato entre partes do sistema. Quando você define que recompensa é uma fila com retry e dead letter, está dizendo: "esse trabalho pode demorar, pode falhar, pode precisar de atenção humana, e nada disso quebra o resto do produto". Esse contrato é o que separa software que aguenta produção de software que parece funcionar nos testes.