Idempotência e Retries: o antídoto pra rede que quebra

# idempotency.py — pattern completo com locking e TTL
from fastapi import FastAPI, Header, HTTPException, Request
from sqlalchemy import text
import json, hashlib

app = FastAPI()

IDEMPOTENCY_TTL_HOURS = 24  # Stripe guarda 24h, ajuste por caso de uso

@app.post("/payments")
async def create_payment(
    request: Request,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
):
    body = await request.json()
    body_hash = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()

    async with db.begin() as tx:
        # Lock da key — evita duas réplicas da API processarem o mesmo key em paralelo
        row = await tx.execute(
            text("""
                SELECT status_code, response_body, request_hash
                FROM idempotency_keys
                WHERE key = :k AND created_at > NOW() - INTERVAL '24 hours'
                FOR UPDATE
            """),
            {"k": idempotency_key},
        )
        existing = row.fetchone()

        if existing:
            # Proteção: mesma key com body diferente = erro 422 (Stripe faz isso)
            if existing.request_hash != body_hash:
                raise HTTPException(422, "Idempotency-Key reuse with different payload")
            return {"status": existing.status_code, "body": existing.response_body}

        # Executa a lógica real do pagamento
        payment = await create_payment_logic(body)
        response = {"id": payment.id, "status": "created"}

        # Grava a resposta pra próxima chamada com a mesma key
        await tx.execute(
            text("""
                INSERT INTO idempotency_keys
                  (key, request_hash, status_code, response_body, created_at)
                VALUES (:k, :h, 201, :r, NOW())
            """),
            {"k": idempotency_key, "h": body_hash, "r": json.dumps(response)},
        )
        return response

# retry_with_jitter.py
from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(5),                       # até 5 tentativas
    wait=wait_random_exponential(                     # full jitter
        multiplier=1,                                 # base em segundos
        max=30,                                       # cap em 30s
    ),
    retry=retry_if_exception_type(httpx.TransportError),
    reraise=True,
)
async def call_payment_gateway(payload: dict):
    async with httpx.AsyncClient(timeout=5.0) as client:
        resp = await client.post(
            "https://gateway.example.com/charge",
            json=payload,
            headers={"Idempotency-Key": payload["op_id"]},  # sempre junto!
        )
        resp.raise_for_status()
        return resp.json()

// backoff.ts
export async function retryWithBackoff<T>(
  fn: () => Promise<T>,
  opts: { maxAttempts?: number; baseMs?: number; capMs?: number } = {},
): Promise<T> {
  const { maxAttempts = 5, baseMs = 200, capMs = 30_000 } = opts;
  let lastErr: unknown;

  for (let attempt = 0; attempt < maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      if (!isRetryable(err)) throw err;              // 4xx não-429: não retria
      if (attempt === maxAttempts - 1) break;

      const exp = Math.min(capMs, baseMs * 2 ** attempt);
      const jitter = Math.random() * exp;            // full jitter
      await new Promise((r) => setTimeout(r, jitter));
    }
  }
  throw lastErr;
}

function isRetryable(err: unknown): boolean {
  // network + 5xx + 429 = retriáveis. 4xx normais = não.
  if (err instanceof TypeError) return true;           // network
  const status = (err as { status?: number })?.status;
  return status === 429 || (!!status && status >= 500);
}

# circuit_breaker.py — versão educativa
import time
from enum import Enum

class State(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, fail_threshold=5, reset_timeout_s=60):
        self.state = State.CLOSED
        self.fail_count = 0
        self.fail_threshold = fail_threshold
        self.reset_timeout_s = reset_timeout_s
        self.opened_at: float | None = None

    def call(self, fn, *args, **kwargs):
        if self.state == State.OPEN:
            if time.time() - self.opened_at > self.reset_timeout_s:
                self.state = State.HALF_OPEN   # deixa 1 tentativa passar
            else:
                raise CircuitOpenError("circuit is open")

        try:
            result = fn(*args, **kwargs)
        except Exception:
            self._on_failure()
            raise
        self._on_success()
        return result

    def _on_success(self):
        self.fail_count = 0
        self.state = State.CLOSED

    def _on_failure(self):
        self.fail_count += 1
        if self.fail_count >= self.fail_threshold:
            self.state = State.OPEN
            self.opened_at = time.time()

class CircuitOpenError(Exception): ...

# consumer_dedup.py — at-least-once do broker + dedup local = effectively-once
async def process_message(msg: dict):
    msg_id = msg["id"]  # broker garante que este id é estável

    async with db.begin() as tx:
        # Tenta inserir — se duplicata, constraint falha e a gente ignora
        result = await tx.execute(
            text("""
                INSERT INTO processed_messages (id, processed_at)
                VALUES (:id, NOW())
                ON CONFLICT (id) DO NOTHING
            """),
            {"id": msg_id},
        )
        if result.rowcount == 0:
            log.info("duplicate message, skipping", msg_id=msg_id)
            return

        # Primeira vez vendo essa msg — processa na mesma transação
        await apply_side_effects(msg, tx)

    # Só ACK o broker depois do commit bem-sucedido
    await broker.ack(msg)