Lab 28 — Processar o arquivo que acabou de chegar

{
  "version": "0",
  "source": "aws.s3",
  "detail-type": "Object Created",
  "account": "111122223333",
  "region": "sa-east-1",
  "time": "2026-08-07T13:04:12Z",
  "detail": {
    "bucket": { "name": "ffv-lab-planilhas" },
    "object": {
      "key": "entrada/fornecedor-08/9c2f-precos-agosto.csv",
      "size": 184320,
      "etag": "d41d8cd98f00b204e9800998ecf8427e",
      "sequencer": "0062E99A88DC407460"
    },
    "request-id": "8N2J5M3R7K1QZP4W",
    "requester": "111122223333",
    "reason": "PutObject"
  }
}

# eventbridge.tf — o bucket fala com o EventBridge, a regra filtra

# Habilita a entrega de TODOS os eventos do bucket ao EventBridge. Diferente da
# notificacao direta, aqui nao se escolhe tipo de evento nem destino — isso fica
# a cargo das regras abaixo, e e por isso que multiplos consumidores cabem sem
# reconfigurar este bloco de novo.
resource "aws_s3_bucket_notification" "planilhas" {
  bucket      = aws_s3_bucket.planilhas.id
  eventbridge = true
}

# A regra e quem decide o que importa: bucket certo, evento de criacao, e
# SOMENTE o prefixo de entrada. Sem o filtro de prefixo, um objeto movido para
# processado/ ou rejeitado/ pelo proprio pipeline gera um NOVO ObjectCreated —
# e o validador dispararia a si mesmo. Este e o loop de execucao que a
# documentacao da AWS avisa para Lambda-com-S3, e a causa e sempre a mesma:
# saida do pipeline caindo dentro do alcance do proprio gatilho.
resource "aws_cloudwatch_event_rule" "planilha_recebida" {
  name = "${var.projeto}-planilha-recebida"

  event_pattern = jsonencode({
    source      = ["aws.s3"]
    detail-type = ["Object Created"]
    detail = {
      bucket = { name = [aws_s3_bucket.planilhas.id] }
      object = { key = [{ prefix = "entrada/" }] }
    }
  })
}

resource "aws_cloudwatch_event_target" "invoca_validador" {
  rule      = aws_cloudwatch_event_rule.planilha_recebida.name
  target_id = "validador"
  arn       = aws_lambda_function.validador.arn
}

resource "aws_lambda_permission" "eventbridge_invoca_validador" {
  statement_id  = "AllowEventBridgeInvoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.validador.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.planilha_recebida.arn
}

# A tabela e a fonte de verdade do estado de cada arquivo. O prefixo do S3
# (entrada/processado/rejeitado) e so conveniencia para quem navega pelo
# console — se um move falhar no meio, e esta tabela que continua confiavel.
resource "aws_dynamodb_table" "arquivos" {
  name         = "${var.projeto}-arquivos"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "ArquivoId"

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

  point_in_time_recovery {
    enabled = true
  }
}

output "tabela_arquivos" {
  value       = aws_dynamodb_table.arquivos.name
  description = "fonte de verdade do estado de cada arquivo recebido"
}

// Validador/Function.cs — decide rapido, sem processar linha nenhuma

public class Function
{
    private readonly IAmazonS3 _s3;
    private readonly IAmazonDynamoDB _db;
    private readonly IAmazonSQS _sqs;
    private const string TabelaArquivos = "ffv-lab-arquivos";
    private const string FilaProcessamento = "https://sqs.sa-east-1.amazonaws.com/.../ffv-lab-planilhas-processamento";

    public async Task FunctionHandler(EventBridgeEvent<S3ObjectCreatedDetail> evento, ILambdaContext ctx)
    {
        var bucket = evento.Detail.Bucket.Name;
        var chave = evento.Detail.Object.Key;
        // O ETag do proprio evento e parte da identidade: duas ENTREGAS do
        // mesmo objeto (retry do S3/EventBridge) tem o mesmo ETag; um objeto
        // REENVIADO pelo fornecedor com o mesmo nome, nao — e por isso ele
        // conta como arquivo novo, o que e o comportamento correto.
        var arquivoId = $"{bucket}/{chave}#{evento.Detail.Object.ETag}";

        // Escrita condicional: so tem sucesso na PRIMEIRA vez que este
        // ArquivoId e visto. E a mesma tecnica do L22 (PutItem condicional
        // por PedidoId), aplicada aqui por ARQUIVO. Entrega "ao menos uma
        // vez" e o contrato documentado do S3 e do EventBridge — nao uma
        // falha rara a ignorar.
        try
        {
            await _db.PutItemAsync(new PutItemRequest
            {
                TableName = TabelaArquivos,
                Item = new Dictionary<string, AttributeValue>
                {
                    ["ArquivoId"] = new() { S = arquivoId },
                    ["Bucket"] = new() { S = bucket },
                    ["Chave"] = new() { S = chave },
                    ["Estado"] = new() { S = "RECEBIDO" },
                    ["RecebidoEm"] = new() { S = DateTimeOffset.UtcNow.ToString("O") },
                },
                ConditionExpression = "attribute_not_exists(ArquivoId)",
            });
        }
        catch (ConditionalCheckFailedException)
        {
            // Entrega repetida do MESMO evento. Nao e erro: e o contrato "ao
            // menos uma vez" se manifestando. Registrar e sair e a resposta
            // correta — reprocessar aqui duplicaria o efeito la na frente.
            ctx.Logger.LogInformation($"entrega repetida ignorada: {arquivoId}");
            return;
        }

        var cabecalho = await LerPrimeirasLinhasAsync(bucket, chave);
        var (valido, motivo) = ValidarEstrutura(cabecalho);

        if (!valido)
        {
            var chaveRejeitada = chave.Replace("entrada/", $"rejeitado/{Sanear(motivo)}/");
            await _s3.CopyObjectAsync(bucket, chave, bucket, chaveRejeitada);
            await _s3.DeleteObjectAsync(bucket, chave);
            await AtualizarEstadoAsync(arquivoId, "REJEITADO", motivo);
            return;
        }

        await AtualizarEstadoAsync(arquivoId, "VALIDADO", motivo: null);

        // So enfileira o trabalho pesado depois de a condicao ter sido nova
        // E a estrutura ter passado. O validador nunca fica preso lendo
        // milhares de linhas — isso e trabalho do Processador, em outra funcao.
        await _sqs.SendMessageAsync(new SendMessageRequest
        {
            QueueUrl = FilaProcessamento,
            MessageBody = JsonSerializer.Serialize(new { arquivoId, bucket, chave }),
        });
    }

    private static (bool valido, string motivo) ValidarEstrutura(string primeirasLinhas)
    {
        // Confere SO estrutura: cabecalho esperado e codificacao. Nenhuma
        // linha de dado e interpretada aqui — isso e responsabilidade do
        // Processador, deliberadamente separado.
        var colunasEsperadas = new[] { "sku", "preco", "estoque" };
        var primeiraLinha = primeirasLinhas.Split('\n').FirstOrDefault() ?? "";
        var colunas = primeiraLinha.Trim().ToLowerInvariant().Split(',');

        if (!colunasEsperadas.All(colunas.Contains))
            return (false, "cabecalho-inesperado");

        // Protege quem eventualmente abrir o arquivo rejeitado no Excel: uma
        // celula comecando com =, +, - ou @ e interpretada como formula pelo
        // Excel, e pode executar comando. Nao processamos o arquivo, mas
        // marcamos o risco no motivo para quem for investigar manualmente.
        if (primeirasLinhas.Contains(",=") || primeirasLinhas.Contains(",+") || primeirasLinhas.Contains(",@"))
            return (false, "possivel-injecao-de-formula");

        return (true, motivo: null);
    }
}

# fila.tf — fila e DLQ do processamento pesado, o mesmo padrao do L22

resource "aws_sqs_queue" "planilhas_dlq" {
  name                       = "${var.projeto}-planilhas-dlq"
  message_retention_seconds  = 1209600 # 14 dias — teto do servico, para dar tempo de investigar
  sqs_managed_sse_enabled    = true
}

resource "aws_sqs_queue" "planilhas_processamento" {
  name = "${var.projeto}-planilhas-processamento"

  # 6x o timeout da funcao processadora (60 s) — cobre reentrega por
  # throttling sem multiplicar demais o atraso caso a mensagem precise
  # mesmo ser reentregue.
  visibility_timeout_seconds = 360

  message_retention_seconds = 345600 # 4 dias cobrem qualquer pico de inicio de mes
  sqs_managed_sse_enabled   = true

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.planilhas_dlq.arn
    # 5, como no L22: da margem a falha transitoria (ex.: pool do Postgres
    # momentaneamente cheio) sem deixar mensagem ruim reentregando para sempre.
    maxReceiveCount = 5
  })
}

resource "aws_sqs_queue_redrive_allow_policy" "dlq_allow" {
  queue_url = aws_sqs_queue.planilhas_dlq.id
  redrive_allow_policy = jsonencode({
    redrivePermission = "byQueue"
    sourceQueueArns   = [aws_sqs_queue.planilhas_processamento.arn]
  })
}

resource "aws_cloudwatch_metric_alarm" "dlq_com_mensagem" {
  alarm_name          = "${var.projeto}-planilhas-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.planilhas_dlq.name
  }

  alarm_actions = [aws_sns_topic.alertas.arn]
}

resource "aws_lambda_event_source_mapping" "processador_consome_fila" {
  event_source_arn = aws_sqs_queue.planilhas_processamento.arn
  function_name    = aws_lambda_function.processador.arn
  batch_size       = 10

  # Sem isto, UM item com erro no lote de 10 devolve os 10 para a fila —
  # inclusive os 9 que ja tiveram efeito aplicado. O mesmo mecanismo do L22.
  function_response_types = ["ReportBatchItemFailures"]
}

# O validador so pode gravar mensagem NESTA fila — nao em qualquer fila da
# conta. GetAuthorizationToken-like: sqs:SendMessage aceita recurso
# especifico, entao aqui o ARN e restrito, sem "*" nenhum.
data "aws_iam_policy_document" "validador_envia_fila" {
  statement {
    effect    = "Allow"
    actions   = ["sqs:SendMessage"]
    resources = [aws_sqs_queue.planilhas_processamento.arn]
  }
}

// Processador/Function.cs — aplica linha a linha, e sabe quando desistir

public class Function
{
    private readonly NpgsqlDataSource _db;
    private readonly IAmazonS3 _s3;
    private readonly IAmazonDynamoDB _dynamo;

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

        foreach (var msg in evento.Records)
        {
            try
            {
                var tarefa = JsonSerializer.Deserialize<TarefaProcessamento>(msg.Body)!;
                var linhas = await LerPlanilhaAsync(tarefa.Bucket, tarefa.Chave);

                await using var conexao = await _db.OpenConnectionAsync();
                await using var transacao = await conexao.BeginTransactionAsync();

                // UPSERT por SKU com o valor ABSOLUTO da planilha — nao um
                // incremento. E o que torna reaplicar a mesma linha duas
                // vezes inofensivo: o resultado final e o mesmo preco e o
                // mesmo estoque, nao um estoque somado duas vezes.
                var comando = new NpgsqlCommand(
                    "INSERT INTO catalogo (sku, preco_centavos, estoque, atualizado_em) " +
                    "VALUES (@sku, @preco, @estoque, now()) " +
                    "ON CONFLICT (sku) DO UPDATE SET " +
                    "  preco_centavos = EXCLUDED.preco_centavos, " +
                    "  estoque = EXCLUDED.estoque, " +
                    "  atualizado_em = now()",
                    conexao, transacao);

                int aplicadas = 0;
                foreach (var linha in linhas)
                {
                    comando.Parameters.Clear();
                    comando.Parameters.AddWithValue("sku", linha.Sku);
                    comando.Parameters.AddWithValue("preco", linha.PrecoCentavos);
                    comando.Parameters.AddWithValue("estoque", linha.Estoque);
                    await comando.ExecuteNonQueryAsync();
                    aplicadas++;
                }

                await transacao.CommitAsync();

                var chaveFinal = tarefa.Chave.Replace("entrada/", "processado/");
                await _s3.CopyObjectAsync(tarefa.Bucket, tarefa.Chave, tarefa.Bucket, chaveFinal);
                await _s3.DeleteObjectAsync(tarefa.Bucket, tarefa.Chave);

                await AtualizarEstadoAsync(tarefa.ArquivoId, "PROCESSADO", linhasAplicadas: aplicadas);
            }
            catch (Exception ex)
            {
                ctx.Logger.LogError($"falha ao processar {msg.MessageId}: {ex.Message}");
                // Devolve SO este item — os demais mensagens do lote, se
                // ja tiverem tido sucesso, nao voltam para a fila.
                falhas.Add(new SQSBatchResponse.BatchItemFailure { ItemIdentifier = msg.MessageId });
            }
        }

        return new SQSBatchResponse { BatchItemFailures = falhas };
    }
}

# ── Prova 1: do upload ao veredito, sem toque humano ────────────────────────
aws s3 cp planilha-valida.csv \
  "s3://ffv-lab-planilhas/entrada/fornecedor-08/$(uuidgen)-precos.csv"
INICIO=$(date +%s)
watch -n2 "aws dynamodb scan --table-name ffv-lab-arquivos \
  --query 'Items[?Estado.S==\`PROCESSADO\`]' --output json | jq length"
# Esperado: o estado chega a PROCESSADO sem nenhum comando manual entre o cp e o scan.
# Medicao de exemplo, arquivo de 800 linhas: 8 a 14 s do upload ao veredito final,
# dominados pelo tempo de leitura e UPSERT das linhas, nao pelo roteamento do evento.

# ── Prova 2: arquivo malformado nao desaparece — vai para rejeitado/ com motivo ──
echo "coluna1,coluna2" > planilha-invalida.csv
aws s3 cp planilha-invalida.csv "s3://ffv-lab-planilhas/entrada/fornecedor-08/teste-invalido.csv"
sleep 10
aws s3 ls s3://ffv-lab-planilhas/rejeitado/ --recursive | grep teste-invalido
aws dynamodb get-item --table-name ffv-lab-arquivos \
  --key '{"ArquivoId":{"S":"ffv-lab-planilhas/entrada/fornecedor-08/teste-invalido.csv#..."}}' \
  --query "Item.Motivo.S"
# Esperado: "cabecalho-inesperado", e ZERO linhas tocadas no catalogo.

# ── Prova 3: entrega duplicada nao processa duas vezes ──────────────────────
# Publica o MESMO evento de amostra duas vezes, simulando o retry documentado.
for i in 1 2; do
  aws events put-events --entries file://evento-de-amostra.json
done
aws logs filter-log-events --log-group-name /aws/lambda/ffv-lab-validador \
  --filter-pattern "entrega repetida ignorada" --query "length(events)"
# Esperado: 1 — a segunda entrega foi reconhecida e ignorada, nao reprocessada.

# ── Prova 4: falha persistente vai para a DLQ, com alarme ───────────────────
# Force um erro no Processador (ex.: aponte para uma tabela Postgres inexistente)
# e envie um arquivo valido. Depois de 5 tentativas:
aws sqs get-queue-attributes \
  --queue-url "$(terraform output -raw fila_dlq_url)" \
  --attribute-names ApproximateNumberOfMessagesVisible
# Esperado: 1 mensagem visivel na DLQ, e o alarme ffv-lab-planilhas-dlq-nao-vazia em ALARM.

# ── Prova 5: a notificacao direta NAO aceita dois destinos sobrepostos ──────
# Prova negativa: mostra por que este modulo nao usou notificacao direta.
aws s3api put-bucket-notification-configuration \
  --bucket ffv-lab-planilhas \
  --notification-configuration file://duas-notificacoes-mesmo-prefixo.json
# Esperado: erro da API — configuracao ambigua para o mesmo tipo de evento e prefixo
# sobreposto. E o limite que a decisao desta secao evita, na pratica, nao so na teoria.

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

# 2. OBJETOS NO BUCKET: o destroy nao apaga o CONTEUDO de um bucket que ja
#    existia (herdado do L17). Planilhas em entrada/, processado/ e rejeitado/
#    continuam la, cobrando armazenamento.
aws s3 ls s3://ffv-lab-planilhas --recursive --summarize | tail -2
aws s3 rm s3://ffv-lab-planilhas/entrada/ --recursive
aws s3 rm s3://ffv-lab-planilhas/processado/ --recursive
aws s3 rm s3://ffv-lab-planilhas/rejeitado/ --recursive

# 3. TABELA DYNAMODB: guarda o historico de estado de cada arquivo. Apagar a
#    tabela apaga a TRILHA inteira, nao so o "estado atual" — confirme que
#    ninguem precisa dela para auditoria antes de seguir.
aws dynamodb delete-table --table-name ffv-lab-arquivos

# 4. FILA COM MENSAGEM: se voce interrompeu no meio de um teste de falha, a
#    DLQ pode ainda ter mensagem — o destroy de uma fila SQS nao falha por
#    causa disso, mas confirme que nao ha nada relevante antes de seguir.
aws sqs get-queue-attributes \
  --queue-url "$(terraform output -raw fila_dlq_url 2>/dev/null || echo "")" \
  --attribute-names ApproximateNumberOfMessagesVisible 2>/dev/null || true

# 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