Segregação de Responsabilidades (CQRS) e Event Sourcing: Consistência Transacional Desacoplada com DynamoDB e Aurora PostgreSQL

Arquiteturas baseadas em operações CRUD (Create, Read, Update, Delete) colapsam invariavelmente sob concorrência extrema. Quando uma corporação acopla fluxos transacionais de alta velocidade e consultas de agregação na mesma modelagem relacional, o motor de banco de dados sofre contenção severa. Bloqueios de linha para garantir a consistência de gravações paralisam as leituras, enquanto consultas analíticas bloqueiam as mutações essenciais de negócios. A adoção simultânea dos padrões CQRS e Event Sourcing dissolve este antipadrão estrutural. Ao desmantelar o modelo de dados unificado, engenheiros separam fisicamente as operações que alteram o estado (Commands) das operações que retornam informações (Queries). O DynamoDB absorve as gravações como uma série cronológica de eventos imutáveis (Event Sourcing), eliminando atualizações destrutivas. Paralelamente, uma arquitetura orientada a eventos projeta essa trilha de auditoria para um modelo relacional otimizado no Aurora PostgreSQL. Esta topologia entrega escalabilidade assimétrica, permitindo que a camada de gravação processe picos vertiginosos de transações enquanto a camada de leitura fornece resultados complexos com latência sub-milissegundo, preservando a integridade absoluta da corporação.

Pré-requisitos

A construção deste ecossistema desacoplado exige domínio avançado sobre sistemas distribuídos e consistência eventual. O provisionamento infraestrutural demanda o Terraform versão 1.7.0 ou superior, orquestrando o provedor HashiCorp AWS versão 5.40.0. A camada lógica necessita do Python 3.12, incorporando a biblioteca boto3 versão 1.34.0 para operações atômicas no DynamoDB, a biblioteca psycopg (versão 3) para atualizações de projeção no PostgreSQL e o aws-lambda-powertools versão 2.34.0. Privilégios administrativos no AWS IAM são vitais para delegar a integração direta (Zero-Compute) entre o DynamoDB Streams e o Amazon EventBridge Pipes.

Passo a Passo

O Repositório de Eventos Imutáveis (Event Store)

A fundação do CQRS inicia com a transformação da semântica de gravação. O domínio de negócios não atualiza mais o saldo de uma conta bancária de forma destrutiva; ele anexa um evento imutável documentando o que ocorreu (como FundsDeposited ou FundsWithdrawn). Provisionamos o Amazon DynamoDB para atuar como o Event Store. A justificativa técnica para essa seleção repousa na escalabilidade horizontal da chave de partição e no suporte a gravações condicionais atômicas (Conditional Puts). Modelamos a tabela utilizando o identificador do agregado (AggregateId) como chave de partição e o número da versão do evento (Version) como chave de ordenação. Esta modelagem garante que todos os eventos pertencentes à mesma entidade corporativa residam na mesma partição física, permitindo a reconstrução exata do estado em memória e prevenindo matematicamente condições de corrida.

resource "aws_dynamodb_table" "event_store" { name = "EnterpriseEventStore" billing_mode = "PAY_PER_REQUEST" hash_key = "AggregateId" range_key = "Version" attribute { name = "AggregateId" type = "S" } attribute { name = "Version" type = "N" } # O stream captura os eventos recém-adicionados para alimentar o modelo de leitura stream_enabled = true stream_view_type = "NEW_IMAGE" tags = { Domain = "Financial" Pattern = "EventSourcing" } }

Como o serviço de comando anexa novos eventos a este fluxo contínuo garantindo que duas transações simultâneas não corrompam a ordem cronológica da entidade e sobrescrevam eventos vitais?

Travamento Otimista e Adição Hexagonal de Comandos (Append)

Protegemos a ordem transacional implementando o mecanismo de Travamento Otimista (Optimistic Locking) no adaptador hexagonal primário. O serviço de comando instrui a persistência do evento especificando matematicamente a versão esperada. Se a conta bancária possui atualmente a versão 5, o próximo evento submetido deve, compulsoriamente, possuir a versão 6. O adaptador Python utiliza a ConditionExpression do DynamoDB (attribute_not_exists(Version)) para impor essa regra de negócio no nível do banco de dados. Se uma transação concorrente já tiver inserido a versão 6 frações de milissegundo antes, o DynamoDB rejeitará a gravação retornando uma exceção de falha condicional (ConditionalCheckFailedException). O código captura esta exceção e instrui o domínio a reiniciar o ciclo de reconstrução, recalcular a regra de negócios e tentar novamente. Esta mecânica assegura a serialização rigorosa sem a necessidade de bloqueios de tabela (Pessimistic Locks).

import json import boto3 from botocore.exceptions import ClientError from dataclasses import dataclass from typing import Dict, Any from aws_lambda_powertools import Logger logger = Logger(service="CommandAPI") dynamodb = boto3.resource("dynamodb") table = dynamodb.Table("EnterpriseEventStore") @dataclass class DepositCommand: account_id: str amount: float expected_version: int class ConcurrencyConflictError(Exception): pass class EventStoreAdapter: def append_event(self, aggregate_id: str, version: int, event_type: str, payload: dict) -> None: try: logger.info(f"Tentando anexar {event_type} na versão {version} para o agregado {aggregate_id}") table.put_item( Item={ "AggregateId": aggregate_id, "Version": version, "EventType": event_type, "Payload": payload, "Timestamp": "2026-11-07T14:58:00Z" # Simplificado para o exemplo  }, # Impõe matematicamente que esta versão não pode existir previamente  ConditionExpression="attribute_not_exists(Version)" ) except ClientError as e: if e.response['Error']['Code'] == 'ConditionalCheckFailedException': logger.warning(f"Conflito de concorrência no agregado {aggregate_id}. Versão {version} já existe.") raise ConcurrencyConflictError("O estado do agregado foi alterado por outra transação.") raise e event_store = EventStoreAdapter() @logger.inject_lambda_context def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: body = json.loads(event.get("body", "{}")) command = DepositCommand( account_id=body["account_id"], amount=body["amount"], expected_version=body["expected_version"] ) # Próxima versão cronológica do evento  target_version = command.expected_version + 1 try: event_store.append_event( aggregate_id=command.account_id, version=target_version, event_type="FundsDeposited", payload={"amount": command.amount} ) return {"statusCode": 201, "body": json.dumps({"status": "Event Appended"})} except ConcurrencyConflictError: return {"statusCode": 409, "body": json.dumps({"error": "Conflict. Please retry."})}

Uma vez que as gravações estão isoladas e protegidas, como o sistema reflete essas mutações de forma eficiente para as interfaces de usuário sem forçá-las a consultar o DynamoDB e agregar centenas de eventos em tempo real a cada leitura?

Coreografia e Projeção do Modelo de Leitura (Read Model)

Consolidamos o estado transacional materializando um Modelo de Leitura (Read Model) otimizado através de uma arquitetura projetora orientada a eventos. O DynamoDB Streams capta a inserção do evento e invoca o Amazon EventBridge Pipes. O EventBridge Pipe atua como um roteador e filtro severo (encaminhando apenas registros de inserção, ignorando modificações sistêmicas) e submete o lote de eventos a uma função AWS Lambda encarregada da projeção. Esta função traduz a série de eventos em um modelo de dados relacional clássico gravado no Amazon Aurora PostgreSQL. A justificativa técnica para projetar dados em um banco relacional reside no ganho brutal de performance analítica. A aplicação de leitura (Query API) executa comandos SELECT diretos na tabela consolidada do PostgreSQL, livre de bloqueios de concorrência, retornando o saldo final consolidado instantaneamente e assegurando que as aplicações clientes desconheçam inteiramente a complexidade do Event Store subjacente.

sequenceDiagram participant CommandAPI as API de Escrita participant DDB as Event Store (DynamoDB) participant Pipe as EventBridge Pipes participant Proj as Lambda (Projetor) participant Aurora as Read Model (PostgreSQL) participant QueryAPI as API de Leitura CommandAPI->>DDB: PutItem (FundsDeposited, Version 6) DDB->>DDB: Verifica Condição (Versão Inédita) DDB->>Pipe: Emite Registro via DynamoDB Streams Pipe->>Proj: Invoca Lote de Eventos Filtrados Proj->>Proj: Traduz Evento para SQL Update Proj->>Aurora: UPDATE accounts SET balance = balance + X QueryAPI->>Aurora: SELECT balance FROM accounts Aurora-->>QueryAPI: Retorna Saldo Consolidado (Sub-ms)

import os import json import psycopg from typing import Dict, Any from aws_lambda_powertools import Logger logger = Logger(service="ProjectionDomain") DB_CONN_STR = os.getenv("READ_MODEL_DB_CONN") class ReadModelProjectionAdapter: def __init__(self, connection_string: str): self.conn_str = connection_string def project_event(self, aggregate_id: str, version: int, event_type: str, payload: dict) -> None: with psycopg.connect(self.conn_str) as conn: with conn.transaction(): # A projeção requer o número da versão para assegurar a idempotência relacional  if event_type == "FundsDeposited": amount = float(payload.get("amount", 0.0)) conn.execute( """ UPDATE account_read_model SET balance = balance + %s, current_version = %s WHERE account_id = %s AND current_version < %s """, (amount, version, aggregate_id, version) ) # Outros tipos de evento são projetados através de padrões análogos  projector = ReadModelProjectionAdapter(DB_CONN_STR) @logger.inject_lambda_context def lambda_handler(event: Dict[str, Any], context: Any) -> None: # O payload chega estruturado pelo EventBridge Pipes  for record in event: dynamodb_data = record.get("dynamodb", {}).get("NewImage", {}) if dynamodb_data: aggregate_id = dynamodb_data.get("AggregateId", {}).get("S") version = int(dynamodb_data.get("Version", {}).get("N", 0)) event_type = dynamodb_data.get("EventType", {}).get("S") payload = json.loads(dynamodb_data.get("Payload", {}).get("S", "{}")) logger.info(f"Projetando evento {event_type} (Versão {version}) para o Read Model.") projector.project_event(aggregate_id, version, event_type, payload)

Solução de Problemas Comuns

A desintegração primária em sistemas CQRS orientados a eventos ocorre através de Projeções Fora de Ordem (Out-of-Order Projections). Embora o DynamoDB Streams garanta a ordem matemática dos registros no nível do estilhaço (Shard), falhas na função Lambda projetora ou redimensionamentos dinâmicos no fluxo podem causar reentregas. Se a projeção da versão 7 ocorrer antes da versão 6, o saldo da conta será corrompido no modelo relacional. A resolução arquitetônica definitiva está exemplificada na instrução SQL do adaptador projetor: WHERE account_id = %s AND current_version < %s. Ao condicionar o UPDATE no PostgreSQL para aceitar a mutação estritamente se a versão gravada no banco for inferior à versão do evento em trânsito, a infraestrutura descarta eventos duplicados e atrasados, imunizando o Read Model contra os artefatos temporais intrínsecos das redes assíncronas.

Outro defeito transacional paralisante é o Envenenamento de Lote (Poison Pill) na esteira do DynamoDB Streams. Se a Lambda projetora falhar persistentemente devido a um formato de evento não previsto (Parsing Error), o AWS Lambda reterá o lote inteiro de registros, executando repetições (Retries) infinitas ou até o limite de retenção do fluxo. Isso paralisa a projeção de todas as outras entidades que compartilham o mesmo lote. A resolução exige configurar explicitamente um Destino On-Failure (Dead-Letter Queue via Amazon SQS) no mapeamento da fonte de evento (Event Source Mapping) e reduzir o limite máximo de tentativas (MaximumRetryAttempts). Esta parametrização força o motor da AWS a descartar o lote problemático para a fila de auditoria e prosseguir imediatamente com a projeção dos registros saudáveis, preservando o fluxo analítico global.

Conclusão

A segregação do modelo transacional através da união de CQRS e Event Sourcing destitui os limites de performance endêmicos aos bancos de dados relacionais tradicionais. Ao utilizar o Amazon DynamoDB para absorver gravações elásticas garantidas por travamento otimista e propagar estados complexos para o Amazon Aurora, engenheiros de software arquitetam plataformas verdadeiramente independentes de escala. Esta dissociação eleva a resiliência corporativa. Em etapas de maturidade avançada, arquitetos devem aproveitar a riqueza histórica do Event Store, orquestrando fluxos secundários que projetam as mesmas séries cronológicas para repositórios especializados, como bancos de dados em grafo (Amazon Neptune) ou motores de busca textual (Amazon OpenSearch), multiplicando o valor analítico da inteligência sem adicionar a mínima latência ao núcleo de negócios.