Lab 19 — Busca no catálogo: LIKE até onde?

// Requisicao de busca que a API monta para o OpenSearch, e a resposta que volta.
// O campo "fuzziness" e a diferenca central em relacao a consulta no RDS.
{
  "requisicao": {
    "query": {
      "match": {
        "descricao": {
          "query": "tenes azul",
          "fuzziness": "AUTO",
          "operator": "and"
        }
      }
    },
    "size": 20,
    "from": 0
  },
  "resposta_resumida": {
    "took_ms": 18,
    "hits": {
      "total": { "value": 34 },
      "hits": [
        {
          "_id": "sku-00481932",
          // _score vem do BM25: frequencia do termo NO documento,
          // penalizada pela frequencia do termo NO corpus inteiro.
          "_score": 11.284,
          "_source": { "nome": "Tenis azul corrida", "preco": 289.9 }
        }
      ]
    }
  }
}

-- migration_gin_busca.sql — o índice invertido dentro do próprio RDS

-- unaccent normaliza "tênis" e "tenis" para o mesmo lexema. Sem isto, a busca
-- textual nativa continuaria sensível a acento, que não é o mesmo problema
-- que curinga à esquerda, mas frustra o usuário do mesmo jeito.
CREATE EXTENSION IF NOT EXISTS unaccent;

-- Coluna GERADA: recalculada pelo próprio Postgres a cada escrita, nunca pela
-- aplicação. "simple" em vez de "portuguese" evita stemming agressivo demais
-- para nome de produto (stemming juntaria "tênis" e "tenista", por exemplo).
ALTER TABLE produtos ADD COLUMN busca tsvector
  GENERATED ALWAYS AS (
    setweight(to_tsvector('simple', unaccent(coalesce(nome, ''))), 'A') ||
    setweight(to_tsvector('simple', unaccent(coalesce(descricao, ''))), 'B')
  ) STORED;

-- CONCURRENTLY é a linha que mais importa nesta migration: sem ela, criar o
-- índice pede ACCESS EXCLUSIVE na tabela inteira, e a integração de
-- fornecedores para de escrever até o índice terminar de ser construído.
-- Em 2,1 milhões de linhas isso não é segundos — é minutos de escrita travada.
CREATE INDEX CONCURRENTLY idx_produtos_busca ON produtos USING GIN (busca);

-- Sem isto o planejador decide com estatística desatualizada e pode preferir
-- Seq Scan mesmo com o índice pronto — é a causa mais comum de "criei o
-- índice e continua lento".
ANALYZE produtos;

// ProdutoBuscaRepository.cs — a consulta que usa o índice invertido do RDS

public sealed class ProdutoBuscaRepository
{
    private readonly NpgsqlDataSource _dataSource; // pool; nunca uma conexão por requisição

    public ProdutoBuscaRepository(NpgsqlDataSource dataSource) => _dataSource = dataSource;

    public async Task<IReadOnlyList<ProdutoResultado>> BuscarAsync(string termo, int limite = 20)
    {
        // websearch_to_tsquery aceita entrada de usuário sem quebrar: trata
        // aspas, "e"/"ou" e hífen sem lançar erro de sintaxe, ao contrário de
        // to_tsquery, que espera operadores já formatados.
        const string sql = """
            SELECT id, nome, preco,
                   -- ts_rank_cd pondera DENSIDADE dos termos no documento, não
                   -- só a contagem — dois termos próximos pesam mais que dois
                   -- termos espalhados no fim de uma descrição longa.
                   ts_rank_cd(busca, websearch_to_tsquery('simple', @termo)) AS relevancia
            FROM produtos
            WHERE busca @@ websearch_to_tsquery('simple', @termo)
            ORDER BY relevancia DESC
            LIMIT @limite;
            """;

        await using var cmd = _dataSource.CreateCommand(sql);
        cmd.Parameters.AddWithValue("termo", unaccentar(termo));
        cmd.Parameters.AddWithValue("limite", limite);

        var resultado = new List<ProdutoResultado>();
        await using var reader = await cmd.ExecuteReaderAsync();
        while (await reader.ReadAsync())
        {
            resultado.Add(new ProdutoResultado(
                reader.GetGuid(0), reader.GetString(1), reader.GetDecimal(2), reader.GetFloat(3)));
        }
        return resultado;
    }

    // A normalização acontece nos dois lados: a coluna gerada já passa por
    // unaccent(), e o termo de busca precisa do mesmo tratamento, senão
    // "tênis" no banco nunca encontra "tenis" digitado sem acento.
    private static string unaccentar(string termo) =>
        termo.Normalize(NormalizationForm.FormD)
             .Where(c => CharUnicodeInfo.GetUnicodeCategory(c) != UnicodeCategory.NonSpacingMark)
             .Aggregate(new StringBuilder(), (sb, c) => sb.Append(c)).ToString();
}

# cdc.tf — a ponte entre RDS e OpenSearch, sem escrita dupla no código

# O interruptor que liga a captura de mudança. É parâmetro ESTÁTICO: exige
# reboot da instância para valer. Ao aplicá-lo, o próprio RDS ajusta
# wal_level, max_wal_senders, max_replication_slots e max_connections — não
# defina esses quatro manualmente, ou a mudança automática pode conflitar.
resource "aws_db_parameter_group" "postgres_cdc" {
  name   = "${var.projeto}-postgres-cdc"
  family = "postgres16"

  parameter {
    name         = "rds.logical_replication"
    value        = "1"
    apply_method = "pending-reboot"
  }
}

resource "aws_dms_replication_subnet_group" "principal" {
  replication_subnet_group_id          = "${var.projeto}-dms"
  replication_subnet_group_description = "Sub-redes privadas para a instância de replicação"
  subnet_ids                           = aws_subnet.privada[*].id
}

resource "aws_dms_replication_instance" "principal" {
  replication_instance_id   = "${var.projeto}-dms"
  replication_instance_class = "dms.t3.medium" # dimensione pelo volume de escrita, não pelo de leitura
  allocated_storage          = 50
  vpc_security_group_ids     = [aws_security_group.dms.id]
  replication_subnet_group_id = aws_dms_replication_subnet_group.principal.id
  publicly_accessible         = false
}

resource "aws_dms_endpoint" "origem_rds" {
  endpoint_id   = "${var.projeto}-origem-rds"
  endpoint_type = "source"
  engine_name   = "postgres"
  server_name   = aws_db_instance.principal.address
  port          = 5432
  database_name = "cadencia"
  username      = "dms_replicador" # usuário dedicado, só com REPLICATION e SELECT em produtos
  password      = data.aws_secretsmanager_secret_version.dms_credencial.secret_string
}

resource "aws_dms_endpoint" "destino_opensearch" {
  endpoint_id   = "${var.projeto}-destino-opensearch"
  endpoint_type = "target"
  engine_name   = "opensearch"

  elasticsearch_settings {
    endpoint_uri            = "https://${aws_opensearch_domain.busca.endpoint}"
    service_access_role_arn = aws_iam_role.dms_para_opensearch.arn
    # Percentual de erro de FULL LOAD tolerado antes de a tarefa parar sozinha
    # — protege contra indexar um catálogo pela metade sem ninguém notar.
    full_load_error_percentage = 10
    error_retry_duration       = 300
  }
}

resource "aws_dms_replication_task" "sincronizar_produtos" {
  replication_task_id      = "${var.projeto}-sync-produtos"
  replication_instance_arn = aws_dms_replication_instance.principal.replication_instance_arn
  source_endpoint_arn      = aws_dms_endpoint.origem_rds.endpoint_arn
  target_endpoint_arn      = aws_dms_endpoint.destino_opensearch.endpoint_arn
  # cdc-only pressupõe que o catálogo atual já foi carregado uma vez; para
  # popular do zero, use "full-load-and-cdc" na primeira execução.
  migration_type = "cdc"

  table_mappings = jsonencode({
    rules = [{
      rule-type = "selection"
      rule-id   = "1"
      rule-name = "produtos"
      object-locator = { schema-name = "public", table-name = "produtos" }
      rule-action = "include"
    }]
  })
}

resource "aws_opensearch_domain" "busca" {
  domain_name    = "${var.projeto}-busca"
  engine_version = "OpenSearch_2.15"

  cluster_config {
    instance_type  = "r6g.large.search"
    instance_count = 2 # 2 nós, 1 réplica por shard — sobrevive à perda de 1 nó
  }

  vpc_options {
    subnet_ids         = [aws_subnet.privada[0].id]
    security_group_ids = [aws_security_group.opensearch.id]
  }

  encrypt_at_rest { enabled = true }
  node_to_node_encryption { enabled = true }

  # Sem isto o domínio aceita requisição sem autenticação de usuário dentro da
  # própria VPC — o security group limita QUEM alcança a porta, não QUEM pode
  # ler cada documento depois de alcançá-la.
  advanced_security_options {
    enabled                        = true
    internal_user_database_enabled = false
    master_user_options {
      master_user_arn = aws_iam_role.opensearch_admin.arn
    }
  }
}

# O papel que o DMS assume para escrever no OpenSearch. Ação restrita ao
# domínio específico — não existe "ESHttpPost em qualquer domínio" aqui.
resource "aws_iam_role_policy" "dms_para_opensearch" {
  role = aws_iam_role.dms_para_opensearch.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect   = "Allow"
      Action   = ["es:ESHttpPost", "es:ESHttpPut", "es:ESHttpGet", "es:ESHttpHead"]
      Resource = "${aws_opensearch_domain.busca.arn}/*"
    }]
  })
}

// BuscaOpenSearchService.cs — a mesma pergunta, motor diferente, resposta que pode divergir em ordem

public sealed class BuscaOpenSearchService
{
    private readonly OpenSearchClient _cliente;

    public BuscaOpenSearchService(OpenSearchClient cliente) => _cliente = cliente;

    public async Task<IReadOnlyList<ProdutoResultado>> BuscarAsync(string termo, int limite = 20)
    {
        var resposta = await _cliente.SearchAsync<ProdutoDocumento>(s => s
            .Index("produtos")
            .Size(limite)
            .Query(q => q
                .Match(m => m
                    .Field(f => f.Descricao)
                    .Query(termo)
                    // AUTO: 0-2 caracteres exige igualdade exata; 3-5 tolera
                    // 1 edição; acima de 5, tolera 2. É a distância de edição
                    // Damerau-Levenshtein, e é o que o tsvector do RDS não tem.
                    .Fuzziness(Fuzziness.Auto)
                    .Operator(Operator.And))));

        if (!resposta.IsValid)
        {
            // Falha aqui NÃO significa catálogo vazio — pode ser domínio
            // indisponível. Devolver lista vazia silenciosamente faria o
            // lojista concluir "não temos esse produto" por um erro de infra.
            throw new BuscaIndisponivelException(resposta.DebugInformation);
        }

        // _score vem do BM25 padrão do OpenSearch: frequência do termo NO
        // documento, saturada, contra a frequência do termo NO corpus inteiro
        // — por isso a ordem pode diferir do ts_rank do RDS sem ser bug.
        return resposta.Hits
            .Select(h => new ProdutoResultado(h.Source.Id, h.Source.Nome, h.Source.Preco, h.Score ?? 0))
            .ToList();
    }
}

# provas.sh — cinco medições; nenhuma conclusão vem de "parece mais rápido"

PROJETO=ffv-lab-busca

# ── Prova 1: o plano muda de Seq Scan para Bitmap Index Scan ─────────────────
psql -c "EXPLAIN ANALYZE SELECT id FROM produtos WHERE descricao ILIKE '%tenis azul%';"
# Esperado ANTES: "Seq Scan on produtos ... rows=2100000 ... Execution Time: ~1800 ms"
psql -c "EXPLAIN ANALYZE SELECT id FROM produtos
         WHERE busca @@ websearch_to_tsquery('simple', 'tenis azul');"
# Esperado DEPOIS: "Bitmap Heap Scan ... Bitmap Index Scan on idx_produtos_busca
#                    ... Execution Time: ~8 ms"
# Reprova: se o segundo comando ainda mostrar Seq Scan, rode ANALYZE produtos
# — estatística desatualizada faz o planejador ignorar um índice existente.

# ── Prova 2: tolerância a erro de digitação, RDS não tem; OpenSearch tem ─────
psql -c "SELECT count(*) FROM produtos
         WHERE busca @@ websearch_to_tsquery('simple', 'tenes');"
# Esperado: 0 — "tenes" não é lexema de "tênis", é erro de digitação, não
# variação gramatical, e o tsvector não corrige isso.
curl -s -X GET "https://$OS_ENDPOINT/produtos/_search" -H 'Content-Type: application/json' -d '
  {"query":{"match":{"descricao":{"query":"tenes","fuzziness":"AUTO"}}}}' \
  | jq '.hits.total.value'
# Esperado: > 0 — o mesmo termo com erro encontra resultado no OpenSearch.

# ── Prova 3: o atraso de sincronização é mensurável, não hipotético ──────────
psql -c "UPDATE produtos SET estoque = estoque - 1 WHERE id = '$SKU_TESTE';"
T0=$(date +%s%3N)
until curl -s "https://$OS_ENDPOINT/produtos/_doc/$SKU_TESTE" | jq -e '.found' >/dev/null; do
  sleep 0.2
done
echo "atraso observado: $(( $(date +%s%3N) - T0 )) ms"
# Na base de exemplo, sob carga normal: 400-900 ms. Sob carga em lote do
# fornecedor: até alguns segundos. MEÇA no seu ambiente — depende do volume
# de escrita concorrente e da classe da instância de replicação.

# ── Prova 4: a leitura de busca saiu do caminho do RDS ───────────────────────
aws cloudwatch get-metric-statistics --namespace AWS/RDS \
  --metric-name CPUUtilization --statistics Average --period 300 \
  --start-time "$(date -u -d '-1 hour' +%FT%TZ)" --end-time "$(date -u +%FT%TZ)" \
  --dimensions Name=DBInstanceIdentifier,Value=${PROJETO} --output table
# Esperado: CPU do RDS estável durante um teste de carga de busca — se subir
# junto com o teste, a busca ainda está batendo no banco relacional.

# ── Prova 5: o índice do OpenSearch reconstrói a partir do RDS ───────────────
# Prova de que OpenSearch é derivado, não fonte da verdade: apague o índice e
# recarregue com uma tarefa full-load-and-cdc nova. Se o RDS não bastasse para
# reconstruir, a arquitetura teria um problema mais sério que atraso.
curl -s -X DELETE "https://$OS_ENDPOINT/produtos"
aws dms start-replication-task --replication-task-arn "$TASK_ARN" \
  --start-replication-task-type reload-target
# Esperado: contagem de documentos volta a bater com "SELECT count(*) FROM produtos"

# 1. Pare a tarefa do DMS ANTES de tocar em qualquer outra coisa.
aws dms stop-replication-task --replication-task-arn "$TASK_ARN"
aws dms wait replication-task-stopped --replication-task-arn "$TASK_ARN"

# 2. Confirme e derrube o slot de replicacao logica no PROPRIO Postgres.
#    O Terraform NAO administra isto — o slot e criado pelo DMS dentro do banco,
#    nao e um recurso da AWS que o destroy enxergue.
psql -c "SELECT slot_name, active FROM pg_replication_slots;"
psql -c "SELECT pg_drop_replication_slot('nome_do_slot');"

# 3. So agora derrube a infraestrutura administrada pelo Terraform.
terraform destroy -auto-approve

# 4. Confirme que o dominio OpenSearch realmente sumiu — ele NAO tem
#    "protecao contra delecao" ligada por padrao, mas confirme mesmo assim,
#    porque e a linha mais cara desta lista.
aws opensearch describe-domain --domain-name ffv-lab-busca 2>&1 | head -5

# 5. Reverta o parametro no grupo de parametros do RDS, se o L14 nao for
#    continuar usando replicacao logica para outra coisa.
aws rds describe-db-parameters --db-parameter-group-name ffv-lab-postgres-cdc \
  --query "Parameters[?ParameterName=='rds.logical_replication']"

# 6. Prova final: nada com o nome do projeto de pe.
aws resourcegroupstaggingapi get-resources \
  --tag-filters Key=Projeto,Values=ffv-lab-busca --output table