Por que Você Precisa de uma Camada de Busca Dedicada para Cognito

O Amazon Cognito é excelente para autenticação e gerenciamento básico de usuários. Mas quando sua equipe de suporte precisa encontrar um usuário com apenas parte do e-mail, um sobrenome e um nível de acesso específico entre milhares de contas, a API ListUsers padrão não dá conta.

O problema? A API do Cognito só suporta correspondência exata em atributos padrão e a paginação fica lenta em escala. Para cenários avançados — como busca fuzzy em atributos personalizados, segmentação em tempo real por grupos ou auditoria — você precisa de uma camada de busca dedicada.

Este guia mostra como construir uma arquitetura orientada a eventos que sincroniza automaticamente os dados de usuários do Cognito para o OpenSearch Serverless via DynamoDB Streams, garantindo buscas em milissegundos em qualquer escala.

AWS architecture diagram showing Cognito, Lambda, DynamoDB, and OpenSearch integration for user search Development Concept Image

Arquitetura: Dois Fluxos de Ingestão + Um Fluxo de Busca

A solução usa três serviços AWS em um pipeline serverless:

  • Amazon DynamoDB como armazenamento de perfis
  • AWS Lambda para transformação e ingestão de dados
  • Amazon OpenSearch Serverless como índice de busca

Fluxo de Ingestão 1: Triggers do Cognito Lambda

Captura dados do usuário durante eventos de autenticação (cadastro e login).

# Trigger pós-confirmação do Cognito (Python)
import json
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('UserProfiles')

def lambda_handler(event, context):
    # Extrai atributos do usuário do evento Cognito
    user_attrs = event['request']['userAttributes']
    user_id = event['userName']
    
    # Cria registro inicial no DynamoDB
    item = {
        'userId': user_id,
        'email': user_attrs.get('email', ''),
        'name': user_attrs.get('name', ''),
        'groups': [],  # preenchido depois no trigger de geração de token
        'lastLogin': None,
        'createdAt': event['request']['userAttributes'].get('created_at', '')
    }
    
    table.put_item(Item=item)
    return event

Fluxo de Ingestão 2: CloudTrail para Ações Administrativas

Ações administrativas (criar usuários via CLI, por exemplo) não disparam os triggers do Cognito. O CloudTrail + EventBridge preenche essa lacuna.

# Lambda consumidora de eventos CloudTrail (Python)
import json
import boto3
from datetime import datetime

cognito = boto3.client('cognito-idp')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('UserProfiles')

def lambda_handler(event, context):
    # Analisa evento CloudTrail para API admin do Cognito
    detail = event['detail']
    if detail['eventSource'] != 'cognito-idp.amazonaws.com':
        return
    
    user_pool_id = detail['requestParameters']['userPoolId']
    username = detail['requestParameters']['username']
    
    # Busca estado atual do usuário no Cognito
    response = cognito.admin_get_user(
        UserPoolId=user_pool_id,
        Username=username
    )
    
    # Upsert no DynamoDB
    item = {
        'userId': username,
        'email': next((a['Value'] for a in response['UserAttributes'] if a['Name'] == 'email'), ''),
        'name': next((a['Value'] for a in response['UserAttributes'] if a['Name'] == 'name'), ''),
        'lastModified': datetime.utcnow().isoformat()
    }
    table.put_item(Item=item)
    return event

Fluxo de Busca: Consultando o Índice

# Lambda de busca (Python)
import json
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

host = 'seu-endpoint-do-domínio-opensearch'
region = 'us-east-1'
service = 'aoss'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key,
                   region, service, session_token=credentials.token)

client = OpenSearch(
    hosts=[{'host': host, 'port': 443}],
    http_auth=awsauth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

def lambda_handler(event, context):
    # Extrai parâmetros de busca do evento API Gateway
    query = event.get('queryStringParameters', {})
    search_term = query.get('q', '')
    field = query.get('field', 'email')  # padrão: e-mail
    
    # Monta consulta OpenSearch com busca fuzzy
    search_body = {
        'query': {
            'multi_match': {
                'query': search_term,
                'fields': [field, 'name^2', 'groups'],
                'fuzziness': 'AUTO',
                'operator': 'and'
            }
        },
        'size': query.get('size', 20)
    }
    
    response = client.search(index='cognito-users', body=search_body)
    
    # Formata resultados
    results = []
    for hit in response['hits']['hits']:
        results.append(hit['_source'])
    
    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({'results': results, 'total': response['hits']['total']['value']})
    }

Serverless data pipeline flow for user indexing with DynamoDB Streams and Lambda triggers Coding Session Visual

Pontos de Atenção e Cuidados Essenciais

1. Timeout dos Triggers do Cognito

O Cognito impõe um limite de 5 segundos para funções trigger. Se você já tem outra lógica nesses triggers (claims personalizados, analytics), garanta que o tempo total de execução fique bem abaixo desse limite. Considere usar concorrência provisionada para evitar cold starts.

2. Consistência de Dados

Eventos do CloudTrail são entregues pelo menos uma vez, mas podem chegar fora de ordem. Use escritas condicionais no DynamoDB com um campo de versão para evitar que dados antigos sobrescrevam registros novos.

3. Custos do OpenSearch

O OpenSearch Serverless cobra por OCU (OpenSearch Capacity Units). Para volumes altos de escrita, considere fazer batch dos eventos do DynamoDB Streams para reduzir escritas no índice.

4. Segurança

  • Use o Cognito Authorizer no API Gateway para validar tokens JWT antes de executar a busca.
  • A Lambda de busca deve assumir uma role IAM somente leitura para o OpenSearch.
  • Criptografe os dados do usuário em repouso tanto no DynamoDB quanto no OpenSearch.

Próximos Passos

  1. Implante o stack completo usando o repositório AWS CDK — ele inclui um frontend React e toda a infraestrutura.
  2. Integre com agentes de IA: Combine esta camada de busca com o Amazon Bedrock para criar um assistente conversacional de diretório de usuários. Para uma visão mais ampla sobre estratégias de dados unificados, veja este guia de migração de Oracle para PostgreSQL no Azure.
  3. Monitore a performance: Use métricas do CloudWatch para latência da Lambda, capacidade de leitura/escrita do DynamoDB e latência de consulta do OpenSearch para ajustar sua arquitetura.

Developer querying user directory with fuzzy search and complex filters on AWS cloud IT Technology Image

Conclusão

Combinando triggers do Cognito Lambda, DynamoDB Streams e OpenSearch Serverless, você pode construir uma camada de busca de usuários em tempo real, escalável, que suporta busca fuzzy, filtros complexos e respostas em milissegundos — sem jobs manuais ou código de indexação personalizado.

Esse padrão permite que sua equipe de suporte encontre usuários em milhares de contas instantaneamente, que administradores segmentem usuários por grupo para campanhas direcionadas, e que times de compliance auditem atributos com filtros complexos.

Para um mergulho mais profundo em como as tendências de banco de dados estão evoluindo para dados unificados e agentes de IA, confira a análise sobre a visão de banco de dados da Microsoft para 2026.

Leitura Recomendada

Este conteúdo foi elaborado com o auxílio de ferramentas de IA, com base em fontes confiáveis, e revisado pela nossa equipe editorial antes da publicação. Não substitui o aconselhamento de um profissional especializado.