Lab 23 — Fanout: um evento, vários interessados

# Quantas integracoes de saida vivem dentro do metodo de checkout, hoje.
grep -c "SendMessageAsync" src/Cadencia.Checkout/PedidosController.cs
# Saida na Cadencia: 2. Cada consumidor novo soma 1 aqui — E soma 1 na
# superficie de teste do metodo mais sensivel do sistema.

# O raio de explosao: se a chamada para a fila de email lancar excecao sem
# tratamento, o que acontece com a resposta ao cliente?
grep -A3 "SendMessageAsync.*filaEmail" src/Cadencia.Checkout/PedidosController.cs
# Na Cadencia: nao ha try/catch ao redor da segunda chamada. Uma excecao ali
# propaga e o POST /pedidos retorna 500 — mesmo com o pedido ja gravado.

// Corpo publicado no topico. Com RawMessageDelivery=true, este e o CORPO
// exato que cada fila recebe — sem envelope do SNS por cima.
{
  "pedidoId": "b3f1e8c2-4a9d-4e11-9c2a-7d5e0f1a2b3c",
  "lojaId": 17,
  "valorTotal": 249.90,
  "itens": [
    { "sku": "TENIS-42-AZ", "quantidade": 1 }
  ],
  "criadoEm": "2026-08-07T14:32:00Z"
}

// Atributos de mensagem do Publish (NAO vao no corpo — sao metadado do SNS,
// e sao o que o filtro de assinatura compara):
// tipoEvento = "pedido.criado"            (String)
// origem     = "checkout-cliente"         (String) — ou "reprocessamento-interno",
//              ou "correcao-manual", conforme quem disparou a criacao do pedido.

# fanout.tf — um topico, N filas que decidem se querem

resource "aws_sns_topic" "pedido_criado" {
  name              = "${var.projeto}-pedido-criado"
  kms_master_key_id = "alias/aws/sns" # chave gerenciada basta: o payload nao carrega segredo
}

# Cada consumidor tem seu proprio ritmo e sua propria causa de falha — por isso
# fila e DLQ sao POR CONSUMIDOR, nunca compartilhadas. O filtro mora aqui, ao
# lado do que ele protege: quem le a fila de email entende por que ela existe.
locals {
  consumidores = {
    estoque = {
      # Recebe TUDO. Baixa de estoque tem de acontecer mesmo quando o evento
      # veio de um reprocessamento — o saldo tem de refletir a realidade.
      filtro = null
    }
    email = {
      # So evento de cliente de verdade. Reprocessamento interno e correcao
      # manual nao podem gerar confirmacao duplicada para quem ja recebeu a
      # primeira. "anything-but" e o operador certo aqui: nega uma lista curta
      # em vez de enumerar todos os valores validos, que podem crescer.
      filtro = jsonencode({
        origem = [{ "anything-but" : ["reprocessamento-interno", "correcao-manual"] }]
      })
    }
    analytics = {
      # Fila NOVA deste laboratorio. Recebe tudo, inclusive reprocessamento —
      # para quem analisa dado, isso tambem e sinal de negocio.
      filtro = null
    }
  }
}

resource "aws_sqs_queue" "dlq" {
  for_each                   = local.consumidores
  name                       = "${var.projeto}-${each.key}-dlq"
  message_retention_seconds  = 1209600 # 14 dias — o teto, para dar tempo de investigar
}

resource "aws_sqs_queue" "fila" {
  for_each                    = local.consumidores
  name                        = "${var.projeto}-${each.key}"
  visibility_timeout_seconds  = 30 # ponto de partida; ajuste por consumidor e' o L22

  # Esta e' a DLQ de PROCESSAMENTO: recebe mensagem que a fila ENTREGOU ao
  # consumidor e ele nao confirmou apos 5 tentativas. E' diferente da DLQ de
  # assinatura abaixo, que e' de ENTREGA — a distincao central deste modulo.
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq[each.key].arn
    maxReceiveCount      = 5 # herdado do L22
  })
}

# A fila nao aceita mensagem de ninguem por padrao. Esta politica autoriza
# ESTE topico — e so ele — a chamar SendMessage. Sem ela, a assinatura existe
# e fica "confirmed", mas toda entrega falha do lado de dentro: o Publish no
# topico continua retornando sucesso, porque publicar != entregar.
resource "aws_sqs_queue_policy" "permite_sns" {
  for_each  = local.consumidores
  queue_url = aws_sqs_queue.fila[each.key].id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Principal = { Service = "sns.amazonaws.com" }
      Action    = "sqs:SendMessage"
      Resource  = aws_sqs_queue.fila[each.key].arn
      Condition = {
        # Sem isto, QUALQUER topico SNS de QUALQUER conta poderia escrever
        # nesta fila: o principal "sns.amazonaws.com" sozinho nao distingue
        # topicos. O ArnEquals com o SourceArn e' a fronteira real.
        ArnEquals = { "aws:SourceArn" = aws_sns_topic.pedido_criado.arn }
      }
    }]
  })
}

# DLQ de ENTREGA — da ASSINATURA, nao da fila. So recebe mensagem se o SNS
# nao conseguir ENTREGAR a fila (ex.: politica revogada, fila apagada). E' um
# erro de lado de fora; a DLQ da fila acima e' erro de lado de dentro.
resource "aws_sqs_queue" "dlq_entrega" {
  name                       = "${var.projeto}-fanout-entrega-dlq"
  message_retention_seconds  = 1209600
}

resource "aws_sns_topic_subscription" "assinatura" {
  for_each  = local.consumidores
  topic_arn = aws_sns_topic.pedido_criado.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.fila[each.key].arn

  # Sem isto, o corpo da mensagem SQS vira o envelope do SNS (Type, MessageId,
  # TopicArn, Message, Timestamp, assinatura de verificacao...) e o consumidor
  # precisaria desserializar duas vezes. Com RawMessageDelivery, o corpo da
  # mensagem SQS E' o JSON publicado, e os atributos do Publish viram
  # atributos de mensagem do SQS diretamente.
  raw_message_delivery = true

  filter_policy       = each.value.filtro
  filter_policy_scope = "MessageAttributes" # e' o padrao; explicito porque a alternativa (MessageBody) exige corpo bem-formado

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq_entrega.arn
  })

  depends_on = [aws_sqs_queue_policy.permite_sns]
}

output "topico_pedido_criado" {
  value       = aws_sns_topic.pedido_criado.arn
  description = "unico ponto de publicacao; a lista de quem recebe mora nas assinaturas, nao aqui"
}

# consumo.tf — cada consumidor le so a fila dele, e nada alem disso

data "aws_iam_policy_document" "publicar_evento" {
  statement {
    effect    = "Allow"
    actions   = ["sns:Publish"]
    resources = [aws_sns_topic.pedido_criado.arn] # so este topico; a task nunca fala com SQS direto
  }
}

resource "aws_iam_role_policy" "checkout_publica" {
  role   = aws_iam_role.task.id # o mesmo papel de aplicacao do L01/L22 — nao o de execucao
  policy = data.aws_iam_policy_document.publicar_evento.json
}

# Um papel POR consumidor. O de email nao consegue ler a fila de estoque, e
# vice-versa — isolamento que a arquitetura de filas separadas so entrega se
# o IAM acompanhar.
data "aws_iam_policy_document" "consumir_fila" {
  for_each = local.consumidores
  statement {
    effect = "Allow"
    actions = [
      "sqs:ReceiveMessage",
      "sqs:DeleteMessage",
      "sqs:GetQueueAttributes",
    ]
    resources = [aws_sqs_queue.fila[each.key].arn]
  }
}

resource "aws_iam_role" "consumidor" {
  for_each           = local.consumidores
  name               = "${var.projeto}-${each.key}-consumidor"
  assume_role_policy = data.aws_iam_policy_document.lambda_assume.json
}

resource "aws_iam_role_policy" "consumidor_le_fila" {
  for_each = local.consumidores
  role     = aws_iam_role.consumidor[each.key].id
  policy   = data.aws_iam_policy_document.consumir_fila[each.key].json
}

resource "aws_lambda_function" "consumidor" {
  for_each      = local.consumidores
  function_name = "${var.projeto}-${each.key}"
  role          = aws_iam_role.consumidor[each.key].arn
  handler       = "Consumidor::Consumidor.Funcao::Executar"
  runtime       = "dotnet8"
  filename      = "build/${each.key}.zip"
  timeout       = 30
}

resource "aws_lambda_event_source_mapping" "consumo" {
  for_each                 = local.consumidores
  event_source_arn         = aws_sqs_queue.fila[each.key].arn
  function_name            = aws_lambda_function.consumidor[each.key].arn
  batch_size                = 10
  function_response_types  = ["ReportBatchItemFailures"] # falha de UM item do lote nao reprocessa o lote inteiro
}

// PedidosController.cs — trecho: publica UMA vez, nao sabe quem esta ouvindo

[HttpPost]
public async Task<IActionResult> Criar(CriarPedidoRequest req)
{
    var pedido = await _pedidos.CriarAsync(req); // grava no Postgres, mesma transacao de negocio

    // Ate a semana passada, este metodo tinha DUAS chamadas SendMessageAsync
    // explicitas — uma por fila. Agora ele nao sabe, e nao precisa saber,
    // quantos sistemas se interessam pelo pedido criado.
    var evento = new
    {
        pedidoId = pedido.Id,
        lojaId = pedido.LojaId,
        valorTotal = pedido.ValorTotal,
        itens = pedido.Itens.Select(i => new { sku = i.Sku, quantidade = i.Quantidade }),
        criadoEm = pedido.CriadoEm
    };

    await _sns.PublishAsync(new PublishRequest
    {
        TopicArn = _config["Sns:TopicoPedidoCriado"],
        Message = JsonSerializer.Serialize(evento),
        MessageAttributes = new Dictionary<string, MessageAttributeValue>
        {
            ["tipoEvento"] = new() { DataType = "String", StringValue = "pedido.criado" },
            // "origem" e' o atributo que a assinatura da fila de email filtra.
            // Uma reexecucao em lote usa "reprocessamento-interno" e nao gera
            // e-mail duplicado para quem ja recebeu o primeiro.
            ["origem"] = new()
            {
                DataType = "String",
                StringValue = req.OrigemInterna ? "reprocessamento-interno" : "checkout-cliente"
            }
        }
    });

    // Se o Publish falhar (rede, throttling), a exceção propaga daqui — e e'
    // a UNICA chamada externa no caminho critico. No desenho anterior, eram
    // duas, e uma terceira estava para chegar.
    return Ok(pedido);
}

// FuncaoEmailConfirmacao.cs — a mensagem pode chegar mais de uma vez (SNS e
// SQS entregam "pelo menos uma vez"); a idempotencia e' a mesma tatica do L22.

public async Task FunctionHandler(SQSEvent evento, ILambdaContext contexto)
{
    foreach (var msg in evento.Records)
    {
        // O corpo JA' e' o JSON puro do pedido: RawMessageDelivery=true tirou
        // o envelope do SNS. Sem isso, este Deserialize leria {Type, MessageId,
        // TopicArn, Message: "..."} em vez do pedido.
        var pedido = JsonSerializer.Deserialize<PedidoCriado>(msg.Body)!;

        try
        {
            // A trava de idempotencia: so um PutItem por pedidoId vence. E' o
            // MESMO padrao do consumidor de estoque do L22 — reaproveitado,
            // nao reinventado.
            await _dynamo.PutItemAsync(new PutItemRequest
            {
                TableName = "idempotencia-email",
                Item = new Dictionary<string, AttributeValue>
                {
                    ["pedidoId"] = new AttributeValue { S = pedido.PedidoId },
                    ["expiraEm"] = new AttributeValue
                    {
                        N = DateTimeOffset.UtcNow.AddDays(7).ToUnixTimeSeconds().ToString()
                    },
                },
                ConditionExpression = "attribute_not_exists(pedidoId)",
            });
        }
        catch (ConditionalCheckFailedException)
        {
            // Segunda entrega do mesmo pedidoId: ja processamos, nao reenvia
            // e-mail. Nao e' erro — e' o comportamento esperado de "pelo
            // menos uma vez".
            continue;
        }

        await _ses.SendEmailAsync(MontarEmailConfirmacao(pedido));
    }
}

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

PROJETO=ffv-lab; REGIAO=us-east-1
TOPICO=$(terraform output -raw topico_pedido_criado)

# ── Prova 1: um Publish, tres filas recebem cada uma sua copia ───────────────
aws sns publish --topic-arn "$TOPICO" --region "$REGIAO" \
  --message '{"pedidoId":"prova-001","lojaId":17,"valorTotal":99.9,"itens":[],"criadoEm":"2026-08-07T15:00:00Z"}' \
  --message-attributes '{"tipoEvento":{"DataType":"String","StringValue":"pedido.criado"},"origem":{"DataType":"String","StringValue":"checkout-cliente"}}'

for f in estoque email analytics; do
  n=$(aws sqs get-queue-attributes \
        --queue-url "$(terraform output -json filas | jq -r .$f)" \
        --attribute-names ApproximateNumberOfMessages \
        --query 'Attributes.ApproximateNumberOfMessages' --output text)
  echo "fila $f: $n mensagem(ns)"
done
# Esperado: as tres filas mostram pelo menos 1. Na medicao de referencia deste
# modulo, as tres chegaram em menos de 2 s do Publish.

# ── Prova 2: o filtro funciona — reprocessamento NAO gera e-mail ─────────────
aws sns publish --topic-arn "$TOPICO" --region "$REGIAO" \
  --message '{"pedidoId":"prova-002","lojaId":17,"valorTotal":10,"itens":[],"criadoEm":"2026-08-07T15:01:00Z"}' \
  --message-attributes '{"tipoEvento":{"DataType":"String","StringValue":"pedido.criado"},"origem":{"DataType":"String","StringValue":"reprocessamento-interno"}}'
sleep 3
aws sqs get-queue-attributes \
  --queue-url "$(terraform output -json filas | jq -r .email)" \
  --attribute-names ApproximateNumberOfMessages --query 'Attributes.ApproximateNumberOfMessages'
# Esperado: 0 mensagens NOVAS na fila de email (a da prova 1 ja deve ter sido
# consumida). Estoque e analytics, ao contrario, recebem esta tambem.

# ── Prova 3: o terceiro consumidor nao tocou o produtor ──────────────────────
git diff --stat origin/main -- src/Cadencia.Checkout/
# Esperado: zero linhas no diretorio do servico de checkout. A fila e a
# Lambda de analytics aparecem so no diff do Terraform.

# ── Prova 4: durabilidade — consumidor fora do ar nao perde mensagem ─────────
aws lambda put-function-concurrency \
  --function-name "${PROJETO}-estoque" --reserved-concurrent-executions 0
aws sns publish --topic-arn "$TOPICO" --region "$REGIAO" \
  --message '{"pedidoId":"prova-004","lojaId":17,"valorTotal":10,"itens":[],"criadoEm":"2026-08-07T15:02:00Z"}' \
  --message-attributes '{"tipoEvento":{"DataType":"String","StringValue":"pedido.criado"},"origem":{"DataType":"String","StringValue":"checkout-cliente"}}'
sleep 5
aws sqs get-queue-attributes \
  --queue-url "$(terraform output -json filas | jq -r .estoque)" \
  --attribute-names ApproximateNumberOfMessagesNotVisible,ApproximateAgeOfOldestMessage
aws lambda delete-function-concurrency --function-name "${PROJETO}-estoque"
# Esperado: a mensagem fica RETIDA na fila (idade > 0, visivel apos o
# visibility timeout) em vez de se perder. E' a diferenca central em relacao
# a assinar a Lambda direto no topico.

# ── Prova 5: mensagem malformada cai na DLQ de PROCESSAMENTO, nao na de entrega
aws sqs send-message \
  --queue-url "$(terraform output -json filas | jq -r .email)" \
  --message-body '{"pedidoId":"prova-005-sem-campos-esperados"}'
sleep 2
aws sqs get-queue-attributes \
  --queue-url "$(terraform output -raw fila_email_dlq)" \
  --attribute-names ApproximateNumberOfMessages
# Esperado: 1 mensagem na DLQ da FILA apos 5 tentativas (redrive_policy da
# fila) — nao na DLQ da assinatura, que so trata falha de ENTREGA do SNS.

# 1. Derrube o que o Terraform administra.
terraform destroy -auto-approve

# 2. MENSAGENS PRESAS NAS DLQ: o destroy remove a fila, mas se voce quer
#    investigar antes de perder o conteudo, drene primeiro.
for f in estoque email analytics fanout-entrega; do
  aws sqs receive-message --queue-url "$(aws sqs get-queue-url \
    --queue-name ffv-lab-${f}-dlq --query QueueUrl --output text 2>/dev/null)" \
    --max-number-of-messages 10 2>/dev/null || true
done

# 3. ASSINATURAS: normalmente saem com o destroy, mas se a fila foi apagada
#    manualmente ANTES do topico, a assinatura pode ficar orfa ("confirmed"
#    apontando para um ARN que nao existe mais). Confira.
aws sns list-subscriptions-by-topic --topic-arn "$(terraform output -raw topico_pedido_criado 2>/dev/null)" 2>/dev/null || \
  echo "topico ja removido — confira subscricoes orfas manualmente no console"

# 4. GRUPOS DE LOGS de cada Lambda: retencao propria, nao pertencem ao ciclo
#    da funcao.
for f in estoque email analytics; do
  aws logs delete-log-group --log-group-name "/aws/lambda/ffv-lab-${f}" 2>/dev/null || true
done

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