Event Sourcing Parte 5: CQRS

Olá!

Este é o quinto artigo de nossa série sobre Event Sourcing (ES) e, desta vez, vamos apresentar e discutir a integração com um pattern que torna possível a realização de consultas simples ou customizadas sobre nossos modelos: CQRS.

Mas, antes, uma observação: uma vez que CQRS tem se mostrado um padrão implementado frequentemente de forma questionável, e sujeito a falhas de interpretação, daremos a ele uma atenção especial, tanto ao definí-lo, quanto ao demonstrar o que consideramos mal-entendidos.

Vamos lá!

O que é CQRS?

CQRS (Command and Query Responsibility Segregation, segregação de responsabilidades entre comandos e consultas) é um padrão proposto pelo mesmo propositor do Event Sourcing, Greg Youg, para resolver um problema bastante específico: facilitar a criação de consultas customizadas e otimizadas em sistemas focados em domínio (Domain Centric).

Uma observação sobre o termo "customizadas". Muitas vezes nosso modelo de domínio possui diversas propriedades, algumas delas coleções ou um grafo de objetos, e, também muitas vezes, a interface com o usuário (ou endpoint, caso estejamos falando sobre uma Web API), demanda apenas um subconjunto dessas propriedades. Em casos como este, sem o emprego de CQRS, há um custo de recuperação de todo o modelo de domínio junto ao ORM, mais o custo da construção do modelo de leitura (Read Model) a ser retornado a partir das propriedades desejadas daquele modelo para retorno ao cliente. É um custo computacional considerável, ainda mais em cenários de demanda crescente ou de estresse eventual (como o famoso exemplo da Black Friday em e-commerces).

Pense por um momento em nossos exemplos anteriores de uso do ES: imagine o custo de recuperar todos os eventos relacionados ao nosso modelo e carregá-los em memória, e então usar este modelo para gerar um modelo de leitura que vai ser entregue ao cliente. Agora imagine, ainda seguindo estes exemplos, que precisamos retornar uma composição entre duas entidades distintas para atender ao cliente.

Muito custoso! Certo? É este o custo que pretendemos evitar com CQRS!

Ou seja, a proposta do CQRS, na prática, é extrair do domínio a responsabilidade de prover dados ao cliente (por isso o Responsibility Segregation), atribuindo-a a um agente especializado, capaz de realizar as consultas demandadas pelo cliente.

Nota: Greg Young, em seu CQRS Documents (em inglês), faz uso do padrão tático Aggregate (Agregado) do DDD ao explicar tanto ES quanto CQRS. Fizemos a escolha deliberada de não utilizá-lo na série para que conhecimento em DDD não fosse um requisito para a compreensão do padrão. Mas entendemos, assim como Young, que sistemas que empreguem DDD sejam os maiores beneficiários do uso de CQRS.

O que não é CQRS!

Repare que, na seção acima, utilizamos a expressão "sistemas focados em domínio". Isso significa que é premissa para o CQRS que um dado sistema contenha um modelo de domínio rico, que é exatamente o cenário que temos quando trabalhamos com ES.

Essa é uma observação importante porque, não raro, encontram-se tentativas de implementação de CQRS com modelos anêmicos! E é preciso deixar claro que a separação entre comandos e consultas em sistemas que se utilizam de modelos anêmicos não caracteriza CQRS, uma vez que não há a segregação de responsabilidades prevista no padrão. Ou seja, não existe um modelo de domínio responsável pelas regras de negócio, e um modelo de leitura responsável por prover dados ao cliente. O mesmo modelo é compartilhado nos dois cenários, o que torna uma tentativa de implementação do padrão apenas uma separação entre o objeto de acesso a dados responsável pela persistência, e o objeto de acesso a dados responsável pelas consultas, o que na prática, não traz à aplicação o ganho pretendido pelo padrão – já que não existe o problema que ele se propõe a resolver.

Essa separação também é comumente interpretada como um exemplo de CQS (Command and Query Separation), um padrão proposto por Bertrand Meyer em seu livro "Object-Oriented Software Construction" (em inglês), mas entendemos essa interpretação como uma extrapolação da ideia do padrão, tendo em vista que Meyer o propôs considerando operações em um mesmo objeto (em inglês), e não objetos independentes que ajam sobre um mesmo dado.

Importante! É muito comum o emprego de mecanismos de notificação (como o MediatR) para comunicar ao modelo de domínio a necessidade de executar um comando, ou ao objeto de acesso a dados (Data Access Object, DAO) para executar uma consulta dentro de um mesmo processo. Entretando entedemos essa solução como sub-ótima pois, dentro do mesmo processo, o modelo de domínio pode ser explicitamente invocado para realizar suas operações. Ou seja, há desperdício de recursos computacionais e aumento da complexidade técnica ao se introduzir mecanismos de notificação -- o mesmo vale para o DAO.

Implementando o CQRS

Agora que temos uma visão clara do resultado esperado com o uso do CQRS, vamos pensar em como implementá-lo.

Para entregarmos ao cliente um modelo de leitura que o atenda, primeiro precisamos criá-lo. E a criação deste modelo se dá de forma muito semelhante à manipulação de eventos de domínio apresentada no artigo anterior. Sempre que nosso modelo for persistido na Event Store, um agente precisará ser acionado para criar ou atualizar o modelo de leitura correspondente, e este agente é geralmente chamado de Projector (projetor). Ele recebe este nome porque, muitas vezes, existe a necessidade de se transformar dados a fim de apresentá-los ao cliente da forma esperada, criando uma projeção do nosso modelo de domínio na forma de nosso modelo de leitura -- uma forma bastante conhecida de projeção é o pattern Materialized View (em inglês).

Nota: é comum que se encontre implementações onde eventos de domínio sejam lançados quando há atualização no modelo de domínio, para que um handler realize a sincronização do modelo de leitura. Não consideramos esta solução pois entendemos que detalhes de implementação de infraestrutura, como a persistência dos modelos de leitura, não devem ser confundidos com o processo de negócio. Portanto, nossa sugestão é separar os dois procedimentos para aumentar a clareza de seu propósito no código.

Uma vez criado nosso modelo de leitura, é necessário armazená-lo e criar um agente para recuperá-lo quando demandado.

Existem algumas opções para essa persistência, como uma tabela no mesmo banco relacional em que nosso modelo de domínio é armazeado (ou uma nova coleção caso nosso modelo seja persistido em um banco NoSql), uma coleção em um banco NoSql separado do banco relacional onde o modelo de domínio é armazenado, um sistema de cache em memória (como Memcached ou Redis) etc. Esta é uma escolha que está sujeita a algumas variáveis, como o custo da contratação da solução de armazenamento, o quão ótima ela é para leitura, o quão familiarizado o time de Ops está com essa solução etc. Não há uma regra sobre o armazenamento do modelo de leitura, apenas as opções e restrições de sua organização.

Uma solução muito comum em cenários onde são empregados bancos de dados relacionais é o uso de uma tabela para cada modelo de leitura. Geralmente combina-se Micro-ORMs (como o Dapper) com Stored Procedures como forma de otimizar as consultas e a posterior construção do modelo de leitura. Também não é uma regra, mas entendemos como uma ótima sugestão pensando em simplicidade e desempenho.

Em nosso código de exemplo, foi usado um simulacro de mecanismo de cache baseado em pares chave-valor (como o Redis) para a persistência de nossos modelos de leitura.

Me mostre o código!

Para começarmos, precisamos conhecer o mecanismo responsável por invocar os projetores.

namespace Lab.EventSourcing.Core
{
    public interface IProjectorHost
    {
        void Add<TModel>(IProjector<TModel> projector)
            where TModel : EventSourcingModel<TModel>;

        void InvokeProjector<TModel>(TModel model)
            where TModel : EventSourcingModel<TModel>;
    }

    public class ProjectorHost : IProjectorHost
    {
        private readonly ConcurrentDictionary<Type, dynamic> _projectors =
                new ConcurrentDictionary<Type, dynamic>();

        public void Add<TModel>(IProjector<TModel> projector)
            where TModel : EventSourcingModel<TModel>
        {
            if (_projectors.ContainsKey(typeof(TModel)))
                throw new ArgumentException($"Projector for {projector.GetType()} already registered.", nameof(projector));

            _projectors.TryAdd(typeof(TModel), projector);
        }

        public void InvokeProjector<TModel>(TModel model) 
            where TModel : EventSourcingModel<TModel>
        {
            if (model is null)
                throw new ArgumentException("A domain model must be provided.");

            if (!_projectors.ContainsKey(typeof(TModel)))
                throw new InvalidOperationException($"There is no projector available for {typeof(TModel)}");

            ((IProjector<TModel>)_projectors[typeof(TModel)]).Execute(model);
        }
    }
}

Como podemos ver, há uma grande semelhança com o mecanismo de disparo de eventos de domínio. A exceção está no fato de que cada modelo, o equivalente a um tipo de evento no DomainEventDispatcher, está relacionado a um único tipo de projetor -- já que não faz sentido ter mais de um projetor para um mesmo modelo.

A seguir, temos um exemplo de implementação de projetor.

namespace Lab.EventSourcing.Core
{
    public interface IProjector<TModel> 
        where TModel : EventSourcingModel<TModel>
    {
        void Execute(TModel model);
    }
}
...
namespace Lab.EventSourcing.StockOrder
{
    public class OrderProjector : IProjector<Order>
    {
        private readonly MemoryCache _dtoStore;

        public OrderProjector(MemoryCache dtoStore) =>
            _dtoStore = dtoStore;

        public void Execute(Order model)
        {
            _dtoStore.AddOrUpdate(model.Id, new OrderDto
            {
                Id = model.Id,
                AccountId = model.AccountId,
                ExecutedQuantity = model.ExecutedQuantity,
                LeavesQuantity = model.LeavesQuantity,
                Price = model.Price,
                Quantity = model.Quantity,
                Side = model.Side,
                Status = model.Status,
                Symbol = model.Symbol
            });
        }
    }
}

No exemplo acima, nosso projetor cria um modelo de leitura para representar o modelo de domínio Order do artigo anterior. Na versão deste artigo incluímos duas propriedades, ExecutedQuantitye LeavesQuantity que indicam a quantidade já executada da ordem e sua quantidade restante. Essa pequena mudança foi introduzida apenas para ilustrar melhor a projeção, já que em nosso modelo de domínio essas propriedades são computadas a partir de uma coleção, e não atribuídas como em nosso modelo de leitura.

Veja a seguir:

namespace Lab.EventSourcing.StockOrder
{
    public class Order : EventSourcingModel<Order>
    {
        ...
        private Queue<Trade> _trades = new Queue<Trade>();
        public IReadOnlyCollection<Trade> Trades { get => _trades; }
        public uint ExecutedQuantity { get => (uint)_trades.Sum(t => t.Quantity); }
        public uint LeavesQuantity { get => Quantity - ExecutedQuantity; }
        ...
        public void Execute(Trade trade)
        {
            if (trade.Quantity > LeavesQuantity)
                throw new InvalidOperationException("This trade quantity overwhelms the order's quantity");

            RaiseEvent(new OrderExecuted(Id, NextVersion, trade));
        }
        ...
        protected override void Apply(IEvent pendingEvent)
        {
            switch (pendingEvent)
            {
                ...
                case OrderExecuted executed:
                    Apply(executed);
                    break;
                ...
            }
        }
        ...
        private void Apply(OrderExecuted executed) =>
            _trades.Enqueue(executed.Trade);
        ...
    }
}

Nosso repositório de eventos foi mais uma vez modificado, agora para comportar nosso host de projetores.

namespace Lab.EventSourcing.Core
{
    public class EventStore
    {
        private readonly EventStoreDbContext _eventStoreContext;
        private readonly IProjectorHost _projectorHost;
        private readonly IDomainEventDispatcher _domainEventDispatcher;

        public static EventStore Create(IProjectorHost projectorHost, IDomainEventDispatcher domainEventDispatcher) =>
            new EventStore(projectorHost, domainEventDispatcher);

        private EventStore(IProjectorHost projectorHost, IDomainEventDispatcher domainEventDispatcher)
        {
            _eventStoreContext = new EventStoreDbContext(new DbContextOptionsBuilder<EventStoreDbContext>()
                                                                .UseInMemoryDatabase(databaseName: "EventStore")
                                                                .EnableSensitiveDataLogging()
                                                                .Options);

            (_projectorHost, _domainEventDispatcher) = (projectorHost, domainEventDispatcher);
        }

        public void Commit<TModel>(TModel model) where TModel : EventSourcingModel<TModel>
        {
            var events = model.PendingEvents.Select(e => Event.Create(model.Id,
                                                                      ((ModelEventBase)e).ModelVersion,
                                                                      e.GetType().AssemblyQualifiedName,
                                                                      ((ModelEventBase)e).When,
                                                                      JsonConvert.SerializeObject(e)));

            _eventStoreContext.Events.AddRange(events);
            _eventStoreContext.SaveChanges();

            _projectorHost.InvokeProjector(model);
            _domainEventDispatcher.Dispatch(model.DomainEvents);

            model.Commit();
        }
        ...
    }
}

Repare que nosso host de projetores é injetado na criação da instância do repositório de eventos, da mesma forma que ocorre com o IDomainEventDispatcher, sendo invocado antes dele quando nosso modelo é persistido na base, a fim de encontrar o projetor correspondente ao modelo persistido para realizar a sincronização.

Nota: O projetor pode ser usado como um enfileirador de mudanças, tornando o processo de atualização dos modelos de leitura assíncrono. Não utilizamos esta opção em nosso código por simplicidade, mas entendemos que seja uma opção válida onde um intervalo maior para a consistência eventual seja tolerado e se deseje uma demanda menor de processamento na aplicação.

Acredite ou não, é só!

Pois é! Por incrível que pareça, CQRS é tão simples quanto o que foi demonstrado até aqui. Temos um modelo de domínio responsável por nossas regras de negócio, um modelo de leitura responsável por apresentar dados ao cliente, e um mecanismo de sincronização que pode atuar na mesma base de dados de nosso modelo de domínio, ou em uma base separada, como demonstrado em nosso exemplo do projetor de Order.

Para maiores detalhes, você pode clonar o repositório deste artigo, e verificar a partir do projeto de testes todo o fluxo apresentado. Sugerimos que os testes sejam debugados para que cada parte do processo fique o mais clara possível.

Gostou do artigo? Me deixe saber pelos indicadores. Ficou alguma dúvida? Me pergunte via comentários que respondo assim que puder!

Considerações finais

Com este artigo encerramos nossa série sobre Event Sourcing. Entendemos que todo o conteúdo essencial foi coberto e que, a partir daqui, com algumas considerações e ajustes, seja possível implementar uma aplicação real utilizando o padrão. Lembre-se de que este padrão possui um grau de complexidade elevado e que, portanto, deve ser empregado apenas quando os benefícios, ou a necessidade, superarem seus custos.

Espero que tenha gostado da série, e fico à disposição tanto por aqui quanto em meus demais contatos para trocarmos ideias a respeito. Feedbacks são sempre bem-vindos!

Até mais!