Lab 22 — Fila que absorve pico, com DLQ e idempotência

#!/usr/bin/env bash
# carga.sh — reproduz o pico de campanha contra as duas arquiteturas
set -euo pipefail

URL="https://$(terraform output -raw dominio)/api/pedidos"
TOTAL=3000   # 10x o normal por 10 minutos, comprimido para o laboratorio
CONCORRENCIA=20

# Cada requisicao manda um item e quantidade fixos, SEM chave de deduplicacao —
# e assim que o cliente real se comporta na versao sincrona.
seq 1 "$TOTAL" | xargs -P "$CONCORRENCIA" -I{} \
  curl -s -o /dev/null -w '%{http_code}\n' -X POST "$URL" \
    -H 'Content-Type: application/json' \
    -d '{"itemId":"sku-42","quantidade":1,"valorCentavos":15000}' \
  >> resultado_carga.txt

echo "200/202: $(grep -cE '^20[02]$' resultado_carga.txt)"
echo "5xx:     $(grep -cE '^5' resultado_carga.txt)"
echo "total:   $(wc -l < resultado_carga.txt)"
# Na arquitetura sincrona (medicao de exemplo): 3.000 requisicoes, 640 com
# 502/504 (21%) — o pool de conexao satura antes do fim do laco.
# Na arquitetura com fila: 3.000 requisicoes, 3.000 com 202 — a fila absorve
# o excesso; o que muda e quanto tempo o CONSUMIDOR leva para esvaziar.

{
  "pedidoId": "b3f1c2a4-7e21-4c9a-9f0e-1a2b3c4d5e6f",
  "itemId": "sku-42",
  "quantidade": 1,
  "valorCentavos": 15000,
  "lojaId": "loja-17",
  "recebidoEm": "2026-08-07T14:22:31.618Z"
}

// Program.cs (ANTES) — tudo dentro do mesmo request

app.MapPost("/api/pedidos", async (NovoPedidoRequest req, AppDb db, IGatewayPagamento gateway) =>
{
    // TRES operacoes, em serie, no MESMO request: gravar, decrementar estoque,
    // cobrar. O cliente so recebe resposta quando a ULTIMA termina — e ela e a
    // mais lenta das tres, porque depende de um servico que a Cadencia nao controla.
    var pedido = new Pedido(req.ItemId, req.Quantidade, req.ValorCentavos);
    db.Pedidos.Add(pedido);
    db.Estoque.Decrementar(req.ItemId, req.Quantidade);
    await db.SaveChangesAsync();

    // ~700 ms no p50, sem retry, sem timeout configurado. Sob carga normal
    // (0,5 pedido/s) isso e invisivel. Sob 10x, e o gargalo inteiro.
    await gateway.CobrarAsync(pedido.Id, req.ValorCentavos);

    return Results.Ok(pedido);
});

// Program.cs (DEPOIS) — a API registra a intencao e responde

app.MapPost("/api/pedidos", async (NovoPedidoRequest req, IAmazonSQS sqs, IOptions<FilaConfig> cfg) =>
{
    var pedidoId = Guid.NewGuid().ToString();

    var corpo = JsonSerializer.Serialize(new PedidoRecebido(
        PedidoId: pedidoId,
        ItemId: req.ItemId,
        Quantidade: req.Quantidade,
        ValorCentavos: req.ValorCentavos));

    // A API so registra a INTENCAO. Nenhum estoque e decrementado e nenhum
    // gateway e chamado aqui — isso passou a ser trabalho do consumidor,
    // minutos ou so segundos depois, no seu proprio ritmo.
    await sqs.SendMessageAsync(new SendMessageRequest
    {
        QueueUrl = cfg.Value.FilaPedidosUrl,
        MessageBody = corpo,
        MessageAttributes = new Dictionary<string, MessageAttributeValue>
        {
            ["PedidoId"] = new() { DataType = "String", StringValue = pedidoId },
        },
    });

    // 202: "recebi a intencao", nao "processei o pedido". E a mudanca de
    // contrato que o cliente da API precisa saber — nao e mais um 200 com o
    // pedido confirmado.
    return Results.Accepted(value: new { pedidoId, status = "recebido" });
});

# fila.tf — a fila principal e a DLQ, com os dois relogios explicitos

resource "aws_sqs_queue" "pedidos_dlq" {
  name = "${var.projeto}-pedidos-dlq"

  # Retencao MAXIMA do servico: 14 dias. Quem for investigar tem duas semanas,
  # nao os 4 dias padrao da fila principal. Mensagem que morre aqui sem alguem
  # olhar e o "lixo silencioso" que o alarme abaixo existe para evitar.
  message_retention_seconds = 1209600

  sqs_managed_sse_enabled = true
}

resource "aws_sqs_queue" "pedidos" {
  name = "${var.projeto}-pedidos"

  # RELOGIO 1: o teto de paciencia antes de a mensagem reaparecer para outro
  # consumidor. A recomendacao da AWS e pelo menos 6x o timeout da funcao —
  # aqui a funcao tem 10 s, entao o minimo seguro e 60 s. Abaixo disso, uma
  # reentrega pode colidir com o PROPRIO processamento ainda em andamento.
  visibility_timeout_seconds = 60

  # Quanto tempo uma mensagem nao consumida sobrevive antes de ser descartada.
  # O padrao (4 dias) cobre qualquer pico de campanha real; se o consumidor
  # ficar parado por mais que isso, o problema deixou de ser "fila".
  message_retention_seconds = 345600

  sqs_managed_sse_enabled = true

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.pedidos_dlq.arn
    # A AWS recomenda pelo menos 5: da ao consumidor algumas chances de
    # reentrega antes de desistir. Falha transitoria (o gateway de pagamento
    # cai por um instante) nao deveria virar alarme na PRIMEIRA tentativa.
    maxReceiveCount = 5
  })
}

# So a fila principal pode redirecionar para esta DLQ — nao qualquer fila da
# conta. Sem isto, qualquer engenheiro com permissao de SQS poderia apontar
# uma fila nao relacionada para a mesma DLQ, misturando causas de falha.
resource "aws_sqs_queue_redrive_allow_policy" "dlq_allow" {
  queue_url = aws_sqs_queue.pedidos_dlq.id

  redrive_allow_policy = jsonencode({
    redrivePermission = "byQueue"
    sourceQueueArns   = [aws_sqs_queue.pedidos.arn]
  })
}

# O alarme e a razao de a DLQ nao ser lixo silencioso. threshold = 0 significa
# "qualquer mensagem visivel ja dispara" — nao existe volume aceitavel de
# pedido que parou de ser processado.
resource "aws_cloudwatch_metric_alarm" "dlq_com_mensagem" {
  alarm_name          = "${var.projeto}-pedidos-dlq-nao-vazia"
  namespace           = "AWS/SQS"
  metric_name         = "ApproximateNumberOfMessagesVisible"
  statistic           = "Maximum"
  period              = 300
  evaluation_periods  = 1
  threshold           = 0
  comparison_operator = "GreaterThanThreshold"
  treat_missing_data  = "notBreaching"

  dimensions = {
    QueueName = aws_sqs_queue.pedidos_dlq.name
  }

  alarm_actions = [aws_sns_topic.alertas.arn]
}

output "fila_pedidos_url" {
  value       = aws_sqs_queue.pedidos.id
  description = "URL da fila principal; e para aqui que a API publica"
}

# consumidor.tf — o Lambda, o event source mapping e as permissoes minimas

resource "aws_lambda_function" "consumidor" {
  function_name = "${var.projeto}-consumidor-pedidos"
  runtime       = "dotnet8"
  handler       = "Consumidor::Cadencia.Pedidos.Consumidor.Function::FunctionHandler"
  role          = aws_iam_role.consumidor.arn

  # RELOGIO 2: o timeout da funcao. O visibility_timeout de 60 s da fila (em
  # fila.tf) tem de ser pelo menos 6x este valor — e 6 x 10 = 60, sem folga
  # sobrando. Subir este numero sem revisar o outro quebra a relacao.
  timeout     = 10
  memory_size = 512

  filename         = data.archive_file.consumidor.output_path
  source_code_hash = data.archive_file.consumidor.output_base64sha256

  environment {
    variables = {
      TABELA_IDEMPOTENCIA = aws_dynamodb_table.pedidos_processados.name
    }
  }
}

resource "aws_lambda_event_source_mapping" "consumidor_pedidos" {
  event_source_arn = aws_sqs_queue.pedidos.arn
  function_name    = aws_lambda_function.consumidor.arn

  batch_size = 10 # o padrao; e tambem o teto de uma fila FIFO. Standard aceita ate 10.000.

  # O MECANISMO central deste laboratorio, do lado da configuracao. Sem isto,
  # UMA mensagem com erro num lote de 10 devolve as OUTRAS NOVE para a fila
  # tambem — inclusive as que ja tiveram efeito aplicado. Com isto, a funcao
  # devolve so os IDs que de fato falharam.
  function_response_types = ["ReportBatchItemFailures"]

  maximum_batching_window_in_seconds = 5
}

data "aws_iam_policy_document" "consumidor_permissoes" {
  statement {
    effect  = "Allow"
    actions = ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes"]
    # So a fila PRINCIPAL. O consumidor nunca le da DLQ — quem le a DLQ e um
    # humano investigando, nao o codigo de producao.
    resources = [aws_sqs_queue.pedidos.arn]
  }

  statement {
    effect    = "Allow"
    actions   = ["dynamodb:PutItem"]
    resources = [aws_dynamodb_table.pedidos_processados.arn]
  }
}

resource "aws_iam_role" "consumidor" {
  name               = "${var.projeto}-consumidor-pedidos"
  assume_role_policy = data.aws_iam_policy_document.consumidor_assume.json
}

resource "aws_iam_role_policy" "consumidor_permissoes" {
  role   = aws_iam_role.consumidor.id
  policy = data.aws_iam_policy_document.consumidor_permissoes.json
}

resource "aws_iam_role_policy_attachment" "consumidor_logs" {
  role       = aws_iam_role.consumidor.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

# idempotencia.tf — a tabela onde a duplicata vira nao-operacao

resource "aws_dynamodb_table" "pedidos_processados" {
  name         = "${var.projeto}-pedidos-processados"
  billing_mode = "PAY_PER_REQUEST" # o trafego e o mesmo pico irregular da fila; provisionado sobraria ocioso
  hash_key     = "PedidoId"

  attribute {
    name = "PedidoId"
    type = "S"
  }

  # A marca de idempotencia nao precisa viver para sempre. 30 dias cobre
  # qualquer reentrega tardia plausivel — a retencao MAXIMA da propria fila e
  # 14 dias — sem acumular item indefinidamente.
  ttl {
    attribute_name = "ExpiraEm"
    enabled        = true
  }

  point_in_time_recovery {
    enabled = true # e o registro de "isto ja foi cobrado"; restaurar errado cobra duas vezes
  }
}

// Function.cs — consome em lote, decide por mensagem, devolve so quem falhou

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
using System.Text.Json;

[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace Cadencia.Pedidos.Consumidor;

public class Function
{
    private static readonly AmazonDynamoDBClient _dynamo = new();
    private static readonly HttpClient _gatewayPagamento = new() { Timeout = TimeSpan.FromSeconds(5) };
    private static readonly string _tabela = Environment.GetEnvironmentVariable("TABELA_IDEMPOTENCIA")!;

    public async Task<SQSBatchResponse> FunctionHandler(SQSEvent evento, ILambdaContext contexto)
    {
        var falhas = new List<SQSBatchResponse.BatchItemFailure>();

        // Cada mensagem do lote e tratada de forma INDEPENDENTE: uma excecao
        // aqui nao pode impedir que as outras sejam reportadas como sucesso —
        // e para isso que existe o ReportBatchItemFailures no event source
        // mapping. Sem essa configuracao (e sem este try/catch por mensagem),
        // uma unica falha devolveria o LOTE INTEIRO para a fila.
        foreach (var mensagem in evento.Records)
        {
            try
            {
                await ProcessarPedidoAsync(mensagem, contexto);
            }
            catch (Exception ex)
            {
                contexto.Logger.LogError($"Falha ao processar {mensagem.MessageId}: {ex.Message}");
                falhas.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = mensagem.MessageId });
            }
        }

        return new SQSBatchResponse(falhas);
    }

    private async Task ProcessarPedidoAsync(SQSEvent.SQSMessage mensagem, ILambdaContext contexto)
    {
        var pedido = JsonSerializer.Deserialize<PedidoRecebido>(mensagem.Body)!;

        // O MECANISMO de idempotencia mora INTEIRO nesta escrita condicional.
        // attribute_not_exists(PedidoId) so deixa o PutItem ter sucesso se
        // este for o PRIMEIRO PutItem com essa chave. SQS pode entregar a
        // mesma mensagem mais de uma vez (at-least-once, nunca exactly-once
        // numa fila standard); esta condicao e o que garante que o EFEITO de
        // negocio acontece uma vez so, nao a fila.
        try
        {
            await _dynamo.PutItemAsync(new PutItemRequest
            {
                TableName = _tabela,
                Item = new Dictionary<string, AttributeValue>
                {
                    ["PedidoId"] = new AttributeValue { S = pedido.PedidoId },
                    ["ProcessadoEm"] = new AttributeValue { S = DateTimeOffset.UtcNow.ToString("O") },
                    // TTL: a marca de idempotencia nao precisa viver para sempre.
                    ["ExpiraEm"] = new AttributeValue
                        { N = DateTimeOffset.UtcNow.AddDays(30).ToUnixTimeSeconds().ToString() },
                },
                ConditionExpression = "attribute_not_exists(PedidoId)",
            });
        }
        catch (ConditionalCheckFailedException)
        {
            // Chegou aqui de novo. NAO e um erro de aplicacao: e o contrato
            // at-least-once se manifestando. O efeito de negocio ja aconteceu
            // na entrega anterior — pular e devolver sucesso e a resposta
            // CORRETA, e e exatamente o que evita a cobranca duplicada.
            contexto.Logger.LogInformation(
                $"Pedido {pedido.PedidoId} ja processado — reentrega ignorada, sem duplicar efeito");
            return;
        }

        // So chega aqui quando o PutItem acima criou o item pela PRIMEIRA vez.
        // E aqui, e so aqui, que o efeito de negocio acontece.
        await DebitarEstoqueAsync(pedido);
        await CobrarCartaoAsync(pedido);
    }

    private static async Task DebitarEstoqueAsync(PedidoRecebido pedido)
    {
        // Decremento no PostgreSQL do L01/L03 — fora do escopo deste trecho.
        // O ponto deste laboratorio nao e ONDE o estoque mora, e sim QUANDO o
        // decremento acontece: depois da condicao de escrita ter passado.
        await Task.CompletedTask;
    }

    private static async Task CobrarCartaoAsync(PedidoRecebido pedido)
    {
        var resposta = await _gatewayPagamento.PostAsJsonAsync(
            "https://gateway.exemplo/cobrancas",
            new { pedido.PedidoId, pedido.ValorCentavos });
        resposta.EnsureSuccessStatusCode(); // erro aqui vira excecao e o item entra em batchItemFailures
    }
}

public record PedidoRecebido(string PedidoId, string ItemId, int Quantidade, int ValorCentavos);

# provas.sh — cinco medicoes; nenhuma conclusao vem de "parece que funcionou"

PROJETO=ffv-lab-fila
FILA_URL=$(terraform output -raw fila_pedidos_url)
DLQ_URL=$(aws sqs get-queue-url --queue-name "${PROJETO}-pedidos-dlq" --query QueueUrl --output text)

# ── Prova 1: a fila absorve o pico sem devolver erro ao cliente ──────────────
./carga.sh
# Esperado: 100% dos requests com 202. Qualquer 5xx aqui e regressao na API,
# nao na fila — a fila em si nao rejeita escrita por volume dentro dos limites.

# ── Prova 2: o backlog cresce e depois esvazia, sem perder mensagem ──────────
watch -n5 "aws sqs get-queue-attributes --queue-url $FILA_URL \
  --attribute-names ApproximateNumberOfMessages ApproximateAgeOfOldestMessage \
  --query 'Attributes'"
# Esperado: o numero sobe durante a carga e desce a zero em minutos. Se ele
# NUNCA desce, o consumidor esta mais lento que a chegada — e capacidade do
# Lambda, nao da fila.

# ── Prova 3: a mesma mensagem reentregue nao duplica o efeito ────────────────
# Reduza temporariamente o timeout da funcao Lambda para MENOR que o tempo
# real de processamento, force uma execucao, e mate o processo no meio
# (ou simplesmente reduza o visibility_timeout da fila para poucos segundos).
aws lambda update-function-configuration --function-name "${PROJETO}-consumidor-pedidos" --timeout 1
PEDIDO_ID=$(uuidgen)
aws sqs send-message --queue-url "$FILA_URL" \
  --message-body "{\"pedidoId\":\"$PEDIDO_ID\",\"itemId\":\"sku-42\",\"quantidade\":1,\"valorCentavos\":15000}"
sleep 90  # cobre pelo menos duas reentregas com visibility_timeout curto
aws dynamodb query --table-name "${PROJETO}-pedidos-processados" \
  --key-condition-expression "PedidoId = :p" \
  --expression-attribute-values "{\":p\":{\"S\":\"$PEDIDO_ID\"}}" \
  --query "length(Items)"
# Esperado: exatamente 1 item, mesmo com varias tentativas de entrega — e o
# PutItem condicional recusando as tentativas depois da primeira.
aws lambda update-function-configuration --function-name "${PROJETO}-consumidor-pedidos" --timeout 10

# ── Prova 4: falha parcial nao reprocessa quem ja teve sucesso ───────────────
# Envie um lote com 1 mensagem malformada (sem itemId) e 9 validas.
for i in $(seq 1 9); do
  aws sqs send-message --queue-url "$FILA_URL" \
    --message-body "{\"pedidoId\":\"$(uuidgen)\",\"itemId\":\"sku-42\",\"quantidade\":1,\"valorCentavos\":15000}"
done
aws sqs send-message --queue-url "$FILA_URL" --message-body '{"pedidoId":"quebrado"}'
sleep 15
aws cloudwatch get-metric-statistics --namespace AWS/Lambda --metric-name Invocations \
  --dimensions Name=FunctionName,Value="${PROJETO}-consumidor-pedidos" \
  --start-time "$(date -u -d '5 min ago' +%FT%TZ)" --end-time "$(date -u +%FT%TZ)" \
  --period 300 --statistics Sum
# Esperado: apenas 1 invocacao adicional na proxima rodada de poll (so a
# mensagem malformada reaparece) — nao 10. Sem ReportBatchItemFailures no
# event source mapping, esperar-se-ia todo o lote de volta.

# ── Prova 5: falha permanente chega na DLQ e dispara o alarme ────────────────
# A mensagem "quebrada" acima falha 5 vezes (deserializacao explode sempre).
sleep 400  # 5 tentativas x 60s de visibility timeout, com folga
aws sqs get-queue-attributes --queue-url "$DLQ_URL" \
  --attribute-names ApproximateNumberOfMessages --query 'Attributes.ApproximateNumberOfMessages'
aws cloudwatch describe-alarms --alarm-names "${PROJETO}-pedidos-dlq-nao-vazia" \
  --query 'MetricAlarms[0].StateValue'
# Esperado: 1 mensagem na DLQ, alarme em ALARM. Se o numero for 0, o
# maxReceiveCount ainda nao foi atingido — confira ApproximateReceiveCount.

# 1. Confirme que nao ha mensagem presa antes de derrubar a fila.
aws sqs get-queue-attributes --queue-url "$FILA_URL" \
  --attribute-names ApproximateNumberOfMessages --query "Attributes"
aws sqs get-queue-attributes --queue-url "$DLQ_URL" \
  --attribute-names ApproximateNumberOfMessages --query "Attributes"

# 2. Derrube o que o Terraform administra: filas, Lambda, event source
#    mapping, tabela DynamoDB, alarme.
terraform destroy -auto-approve

# 3. LOG GROUP DO LAMBDA: o Terraform so remove se voce declarou o
#    aws_cloudwatch_log_group explicitamente. Sem isso, ele sobrevive
#    e continua cobrando por retencao.
aws logs describe-log-groups \
  --log-group-name-prefix "/aws/lambda/ffv-lab-fila-consumidor-pedidos" \
  --query "logGroups[].logGroupName" --output table
aws logs delete-log-group \
  --log-group-name "/aws/lambda/ffv-lab-fila-consumidor-pedidos" 2>/dev/null || true

# 4. Prova final: nada com o nome do projeto de pe.
aws resourcegroupstaggingapi get-resources \
  --tag-filters Key=Projeto,Values=ffv-lab-fila \
  --query "ResourceTagMappingList[].ResourceARN" --output table