Como Construir um Sistema de Retry Inteligente com Dead Letter Queue para Consultas de CPF

const axios = require("axios");

class RetryConfig {
    constructor({
    maxRetries = 3,
    baseDelay = 1000,
    maxDelay = 30000,
    jitterFactor = 0.5,
    retryableStatuses = [408, 429, 500, 502, 503, 504],
    } = {}) {
    this.maxRetries = maxRetries;
    this.baseDelay = baseDelay;
    this.maxDelay = maxDelay;
    this.jitterFactor = jitterFactor;
    this.retryableStatuses = retryableStatuses;
    }

    calculateDelay(attempt) {
    const exponential = this.baseDelay * Math.pow(2, attempt);
    const capped = Math.min(exponential, this.maxDelay);
    const jitter = capped * this.jitterFactor * Math.random();
    return capped + jitter;
    }

    isRetryable(error) {
    if (error.code === "ECONNABORTED" || error.code === "ETIMEDOUT") {
    return true;
    }
    if (error.response) {
    return this.retryableStatuses.includes(error.response.status);
    }
    return true; // Erros de rede são retryable
    }
}

async function consultarCpfComRetry(cpf, config = new RetryConfig()) {
    let lastError;

    for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
    try {
    const response = await axios.get(
    `https://api.cpfhub.io/cpf/${cpf}`,
    {
    headers: { "x-api-key": process.env.CPFHUB_API_KEY },
    timeout: 10000,
    }
    );
    return response.data;
    } catch (error) {
    lastError = error;

    if (!config.isRetryable(error) || attempt === config.maxRetries) {
    break;
    }

    const delay = config.calculateDelay(attempt);
    console.log(`Tentativa ${attempt + 1} falhou. Retry em ${delay}ms...`);
    await new Promise((resolve) => setTimeout(resolve, delay));
    }
    }

    throw lastError;
}

const { Queue, Worker } = require("bullmq");
const Redis = require("ioredis");

const connection = new Redis({ host: "localhost", port: 6379 });

// Fila principal de consultas
const cpfQueue = new Queue("cpf-consultas", { connection });

// Dead Letter Queue
const cpfDLQ = new Queue("cpf-dlq", { connection });

// Worker que processa consultas
const worker = new Worker(
    "cpf-consultas",
    async (job) => {
    const { cpf, requestId } = job.data;
    try {
    const resultado = await consultarCpfComRetry(cpf);
    return resultado;
    } catch (error) {
    // Após esgotar retries, envia para DLQ
    await cpfDLQ.add("consulta-falhada", {
    cpf,
    requestId,
    error: error.message,
    originalJobId: job.id,
    failedAt: new Date().toISOString(),
    attempts: job.attemptsMade,
    });
    throw error;
    }
    },
    {
    connection,
    concurrency: 10,
    limiter: { max: 100, duration: 60000 },
    }
);

// Enfileirar uma consulta
async function enfileirarConsulta(cpf, requestId) {
    await cpfQueue.add("consultar", { cpf, requestId }, {
    attempts: 3,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: 1000,
    removeOnFail: false,
    });
}

// Worker da DLQ - reprocessa periodicamente
const dlqWorker = new Worker(
    "cpf-dlq",
    async (job) => {
    const { cpf, requestId, failedAt } = job.data;

    // Verifica se a consulta não é muito antiga
    const horasDesdeAFalha =
    (Date.now() - new Date(failedAt).getTime()) / 3600000;

    if (horasDesdeAFalha > 24) {
    console.log(`Job ${requestId} expirado (${horasDesdeAFalha}h)`);
    return { status: "expired" };
    }

    // Tenta reprocessar com configuração mais tolerante
    const resultado = await consultarCpfComRetry(
    cpf,
    new RetryConfig({ maxRetries: 5, baseDelay: 5000 })
    );

    return resultado;
    },
    {
    connection,
    concurrency: 2,
    limiter: { max: 10, duration: 60000 },
    }
);

// Métricas da DLQ
async function obterMetricasDLQ() {
    const waiting = await cpfDLQ.getWaitingCount();
    const failed = await cpfDLQ.getFailedCount();
    return { pendentes: waiting, falhados: failed };
}