Como implementar dead letter queue para consultas de CPF que falharam

import requests
import json
import time
import redis
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, db=0)

FILA_PRINCIPAL = 'fila:consulta_cpf'
FILA_DLQ = 'fila:consulta_cpf:dlq'
MAX_TENTATIVAS = 3

headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
}

def enfileirar_consulta(cpf, contexto=None):
    mensagem = {
    'cpf': cpf,
    'tentativa': 0,
    'contexto': contexto or {},
    'criado_em': datetime.utcnow().isoformat()
    }
    r.lpush(FILA_PRINCIPAL, json.dumps(mensagem))

def processar_fila():
    while True:
    item = r.brpop(FILA_PRINCIPAL, timeout=5)
    if not item:
    continue

    mensagem = json.loads(item[1])
    cpf = mensagem['cpf']
    tentativa = mensagem['tentativa'] + 1

    try:
    response = requests.get(
    f'https://api.cpfhub.io/cpf/{cpf}',
    headers=headers,
    timeout=15
    )

    if response.status_code == 200:
    resultado = response.json()
    salvar_resultado(cpf, resultado, mensagem['contexto'])
    print(f'OK: {cpf} (tentativa {tentativa})')
    continue

    elif response.status_code in [400, 401]:
    # Erros que nao melhoram com retry
    mover_para_dlq(mensagem, f'http_{response.status_code}', tentativa)
    continue

    except requests.exceptions.Timeout:
    pass
    except requests.exceptions.ConnectionError:
    time.sleep(5)

    # Retentativa ou DLQ
    if tentativa >= MAX_TENTATIVAS:
    mover_para_dlq(mensagem, 'max_tentativas', tentativa)
    else:
    mensagem['tentativa'] = tentativa
    mensagem['ultima_tentativa'] = datetime.utcnow().isoformat()
    r.lpush(FILA_PRINCIPAL, json.dumps(mensagem))
    # Backoff exponencial
    time.sleep(2 ** tentativa)

def mover_para_dlq(mensagem, motivo, tentativa):
    mensagem['dlq_motivo'] = motivo
    mensagem['dlq_tentativas'] = tentativa
    mensagem['dlq_timestamp'] = datetime.utcnow().isoformat()
    r.lpush(FILA_DLQ, json.dumps(mensagem))
    print(f'DLQ: {mensagem["cpf"]} - {motivo} (apos {tentativa} tentativas)')

def salvar_resultado(cpf, resultado, contexto):
    r.setex(
    f'cpf:resultado:{cpf}',
    86400,
    json.dumps(resultado)
    )

const Queue = require('bull');

const filaConsulta = new Queue('consulta-cpf', {
    redis: { host: 'localhost', port: 6379 },
    defaultJobOptions: {
    attempts: 3,
    backoff: {
    type: 'exponential',
    delay: 2000
    },
    removeOnComplete: true,
    removeOnFail: false
    }
});

// Fila DLQ
const filaDLQ = new Queue('consulta-cpf-dlq', {
    redis: { host: 'localhost', port: 6379 }
});

// Processador principal
filaConsulta.process(async (job) => {
    const { cpf, contexto } = job.data;

    const response = await fetch(`https://api.cpfhub.io/cpf/${cpf}`, {
    method: 'GET',
    headers: {
    'x-api-key': process.env.CPFHUB_API_KEY,
    'Accept': 'application/json'
    },
    signal: AbortSignal.timeout(15000)
    });

    if (response.status === 401) {
    // Mover direto para DLQ (sem retry)
    await filaDLQ.add({
    cpf,
    contexto,
    motivo: 'api_key_invalida',
    tentativas: job.attemptsMade + 1
    });
    return;
    }

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

    return response.json();
});

// Listener de falha final
filaConsulta.on('failed', async (job, err) => {
    if (job.attemptsMade >= job.opts.attempts) {
    await filaDLQ.add({
    cpf: job.data.cpf,
    contexto: job.data.contexto,
    motivo: err.message,
    tentativas: job.attemptsMade
    });
    console.log(`DLQ: ${job.data.cpf} - ${err.message}`);
    }
});

// Enfileirar consulta
async function solicitarConsulta(cpf, contexto = {}) {
    await filaConsulta.add({ cpf, contexto });
}

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/api/dlq/status')
def dlq_status():
    tamanho = r.llen(FILA_DLQ)
    mensagens = []
    for i in range(min(tamanho, 20)): # ultimas 20
    msg = r.lindex(FILA_DLQ, i)
    if msg:
    mensagens.append(json.loads(msg))

    # Agrupar por motivo
    motivos = {}
    for msg in mensagens:
    motivo = msg.get('dlq_motivo', 'desconhecido')
    motivos[motivo] = motivos.get(motivo, 0) + 1

    return jsonify({
    'total_dlq': tamanho,
    'por_motivo': motivos,
    'ultimas_mensagens': mensagens[:10]
    })

def reprocessar_dlq(filtro_motivo=None):
    tamanho = r.llen(FILA_DLQ)
    reprocessados = 0
    mantidos = 0

    mensagens_manter = []

    for _ in range(tamanho):
    item = r.rpop(FILA_DLQ)
    if not item:
    break

    mensagem = json.loads(item)

    if filtro_motivo and mensagem.get('dlq_motivo') != filtro_motivo:
    mensagens_manter.append(item)
    mantidos += 1
    continue

    # Resetar tentativas e reenviar para fila principal
    mensagem['tentativa'] = 0
    mensagem['reprocessado_em'] = datetime.utcnow().isoformat()
    r.lpush(FILA_PRINCIPAL, json.dumps(mensagem))
    reprocessados += 1

    # Devolver mensagens que nao foram reprocessadas
    for msg in mensagens_manter:
    r.lpush(FILA_DLQ, msg)

    print(f'Reprocessados: {reprocessados} | Mantidos: {mantidos}')

# Reprocessar apenas falhas por tentativas esgotadas
reprocessar_dlq('max_tentativas')