Segregate Responsibilities with Elixir Commanded

Explore how Elixir's Commanded library revolutionizes application development through CQRS and Event Sourcing, offering powerful capabilities for audit, time travel, and seamless event-driven architecture.

Responsibility segregation in Elixir programming language

CQRS

Command Query Responsibility Segregation (CQRS) at a conceptual level, emphasizes that actions leading to state changes are designated as commands, while data retrieval actions are termed queries. Due to the distinct operational demands of executing commands and queries, developers are encouraged to employ diverse persistence strategies for handling each, thus segregating their responsibilities.

Event Sourcing pattern

Event Sourcing is a pattern for storing data as events in an append-only log called event store. While this definition may seem straightforward, it overlooks a crucial aspect: by storing events, the context surrounding each event is retained. For instance, from a single piece of information, you can discern not only that an email was sent but also the reason behind it. In contrast, in alternative storage patterns, the context of business logic operations is often lost or stored separately. The current state of an entity can be reconstructed by sequentially replaying all events in the order they occurred. The information within the system is derived directly from these events.

The proposal of the CQRS pattern coincided with the introduction of Event Sourcing to the public intentionally. Unlike state-based persistence, where circumventing the use of the domain model for queries may be feasible, such an approach becomes challenging, if not impossible, in event-sourced systems. This is due to the absence of a singular location where the complete state of a domain object is stored.

Pros of event sourced system

What are the benefits of those? Apart from others:

• Audit: Data is stored as immutable events, providing a robust audit event log.
• Time Travel: All state changes are retained, allowing for backward and forward time traversal, beneficial for debugging or analysis, and enabling the rebuilding of downstream projections.
• Root Cause Analysis: Business events can be traced back to their origins, offering visibility into entire workflows.
• Event-Driven Architecture: Immediate reaction to new events enhances efficiency and enables real-time business workflow modeling.
• Asynchronous First: Minimizing synchronous interactions leads to responsive, high-performance, scalable systems.
• One-Way Data Flow: Data flows one way through independent models, aiding in reasoning and debugging.

Eventual consistency

Eventual consistency entails the concept that reads or queries may not be instantly synchronized with writes. This phenomenon can also arise in systems that do not utilize event sourcing, often due to database replication delays. In essence, the data in your read storage eventually catches up with the latest writes, which could take milliseconds, seconds, minutes, etc. Given this, event sourcing systems frequently encounter eventual consistency challenges, necessitating readiness to address the trade-offs between eventual and strong consistency.

In the context of Elixir programming, the Commanded package provides a powerful framework for implementing CQRS and Event Sourcing patterns. Let's explore how Commanded facilitates the implementation of these patterns in Elixir applications.

Commanded in practice

Let's explore the application code and check how to implement event sourcing in our own elixir applications

As always when adding a new package we need to adjust our mix.exs file:

def deps do
  [
      {:commanded, "~> 1.4"},
      {:jason, "~> 1.3"} # Optional
  ]
end

Event store

Next, we must decide what store to use to persist our events. We have two primary options to choose from:

• PostgreSQL-based EventStore
• Greg Young's EventStoreDB

Both options are great and have their pros and cons but to make things simpler I’ll suggest using PostgreSQL adapter in the beginning.

We will add another dependency for that:

def deps do
  [
    {:eventstore, "~> 1.4"}
  ]
end

as well as MyApp.EventStore module which we will include in our supervision tree:

# event_store.ex
defmodule MyApp.EventStore do
  use EventStore, otp_app: :my_app

  def init(config) do
    {:ok, config}
  end
end

# application.ex
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      MyApp.EventStore
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

Finally, a bunch of configurations to set up our store:

config :my_app, MyApp.EventStore,
  serializer: EventStore.JsonSerializer,
  username: "postgres",
  password: "postgres",
  database: "eventstore",
  hostname: "localhost"

config :my_app, event_stores: [MyApp.EventStore]

And now we are good to go and run mix tasks for our event store initialization:

mix do event_store.create, event_store.init

By Your Command! - Commanded library quickstart

Commands & Events

In the simplest words, commands represent intentions to perform actions, while events represent the outcomes of those actions.

A command should include a field that uniquely identifies the aggregate instance, such as "competition_id." Because commands are standard Elixir structs, you can improve the process of defining structs, enforcing mandatory keys, or performing validations by employing libraries like typed_struct or domo. This approach reduces the need for excessive boilerplate code.

defmodule MyApp.Commands.CreateCompetition do
  @moduledoc """
  Create a new competition command
  """

  use Domo

  @type t() :: %__MODULE__{
          competition_id: nil | binary(),
          name: binary(),
          url: url(),
          user_id: binary()
        }

  defstruct [
    :competition_id,
    :name,
    :url,
    :user_id
  ]
end

Domain events signify significant occurrences within the scope of an aggregate. They are conventionally named in the past tense, such as "competition created".

For each domain event, create a separate module and define its fields using defstruct. Ensure that each event also includes a field to uniquely identify the aggregate instance, such as “competition_id”

Additionally, remember to implement the Jason.Encoder protocol for the event struct to enable JSON serialization, as illustrated below.

defmodule MyApp.Events.CompetitionCreated do
  @moduledoc """
  Competition created event
  """

  @derive [Jason.Encoder]
  @type t() :: %__MODULE__{
          competition_id: binary(),
          name: binary(),
          url: binary(),
          user_id: binary()
        }
  defstruct [
    :competition_id,
    :name,
    :url,
    :user_id
  ]
end

Aggregates

Aggregates play a crucial role in DDD by defining consistency boundaries within a domain model and encapsulating domain logic. A single aggregate is a cluster of domain objects that are treated as a single unit for the purpose of data changes. In Commanded, aggregate is comprised of its state, public command functions, command handlers functions, and state mutators. As we store events, we have access to a detailed event log of past events handled in our application.

defmodule MyApp.Aggregates.Competition do
  @moduledoc """
  Competition write model aggregate
  """

  use Domo

  @type t() :: %__MODULE__{
          id: binary(),
          name: binary(),
          url: binary()
        }
  defstruct [
    :id,
    :name,
    :url
  ]

  alias MyApp.Aggregates.Competition
  alias MyApp.Commands.CreateCompetition
  alias MyApp.Events.CompetitionCreated

# Public API
  def create_competition(%Competition{id: nil}, uid, name, url, user_id) do
    event = %CompetitionCreated{
            competition_id: uid,
      name: name,
      url: url,
      user_id: user_id
    }

    {:ok, event}
  end

# Command handler
  def execute(%Competition{id: nil}, %CreateCompetition{} = command) do
    %CompetitionCreated{
            competition_id: command.id,
      name: command.name,
      url: command.url,
      user_id: command.user_id
    }
  end

# State mutator
  def apply(%Competition{} = competition, %CompetitionCreated{} = event) do
    %Competition{
      competition
      | competition_id: event.id,
                name: event.name,
        url: event.url
    }
  end
end

Dispatchers & Handlers

At the opposite of handling command execution directly inside aggregate, we can use a separate module. A handler accepts both the aggregate and the command undergoing execution. This handler provides the opportunity to validate, authorize, and/or augment the command with supplementary data before executing the relevant function within the aggregate module.

defmodule MyApp.Handlers.CreateCompetitionHandler do
  @behaviour Commanded.Commands.Handler

    alias MyApp.Commands.CreateCompetition
  alias MyApp.Aggregates.CompetitionAggregate

  def handle(%CompetitionAggregate{} = aggregate, %CreateCompetition{} = command) do
    %CreateCompetition{id: uid, name: name, url: url, user_id: user_id} = command

    CompetitionAggregate.create_competition(aggregate, uid, name, url, user_id)
  end
end

A Router module serves the purpose of dispatching commands to their designated command handler and/or aggregate module.

To create a router module, utilize Commanded.Commands.Router, and proceed to register each command along with its corresponding handler.

defmodule MyApp.Router do
  use Commanded.Commands.Router

    alias MyApp.Commands.CreateCompetition
  alias MyApp.Aggregates.CompetitionAggregate

  dispatch CreateCompetition, to: CompetitionAggregate, identity: :competition_id
end

Projectors & Projections

The concept involves software components known as Projections and Projectors, which subscribe to the real-time event stream from the events database. Upon receiving an event, the Projection can then translate the data contained within that event into a view model within a designated reporting database.

You have the flexibility to opt for an SQL or NoSQL database, a document store, a filesystem, a full-text search index, or any other storage mechanism. Moreover, you can employ multiple storage providers, tailored to optimize the querying requirements they need to fulfill.

The simplest solution will be using the Commanded Ecto Projections library.

This part will be our read model and if we don't change the default behaviour by ourselves it will be updated asynchronously so after the write model finishes saving events to the event store we can't be sure read models are also updated.

defmodule MyApp.Projectors.Competition do
  use Commanded.Projections.Ecto,
    application: MyApp.Application,
    repo: MyApp.Projections.Repo,
    name: "MyApp.Projectors.Competition"

    alias MyApp.Events.CompetitionCreated
    alias MyApp.Projections.Competition

  project %CompetitionCreated{} = event, _metadata, fn multi ->
    %CompetitionCreated{name: name, url: url, user_id: user_id} = event

    projection = %Competition{name: name, url: url, user_id: user_id}

    Ecto.Multi.insert(multi, :competition, projection)
  end
end

FAQ

What is Command Query Responsibility Segregation (CQRS) in Elixir?

CQRS is a design pattern that separates the responsibilities of command execution and data querying in applications. In Elixir, this means using different models for updating state (commands) and for reading data (queries), enhancing performance and scalability.

How does Event Sourcing complement CQRS in Elixir applications?

Event Sourcing involves storing changes to application state as a sequence of events. This approach, combined with CQRS, enables powerful features like audit trails, time travel, and the simplification of complex business transactions in Elixir applications.

What are the benefits of using Event Sourcing in Elixir?

Event Sourcing offers several advantages, including robust audit logs, the ability to replay events for debugging or analysis (time travel), simplified root cause analysis, and support for event-driven architectures which can improve application responsiveness and scalability.

What challenges does Eventual Consistency introduce in Elixir systems?

Eventual consistency refers to the delay between executing a command and having the updated state available for queries. This delay can affect system design and user experience, requiring strategies to handle the temporary inconsistency between command execution and data availability.

How does the Commanded library facilitate CQRS and Event Sourcing in Elixir?

The Commanded library provides Elixir developers with tools to implement CQRS and Event Sourcing patterns efficiently. It helps in organizing commands, events, and aggregates, facilitating the development of event-driven, scalable applications.

How do you set up an event store in Elixir using Commanded?

Setting up an event store in Elixir involves choosing a storage backend (e.g., PostgreSQL), adding the necessary dependencies, configuring the event store module, and initializing it to persist events.

What role do Aggregates play in an Elixir CQRS/Event Sourcing system?

Aggregates define consistency boundaries and encapsulate business logic in CQRS/Event Sourcing systems. They manage state changes by processing commands and applying events to maintain the integrity of transactional operations.

How are commands and events defined and processed in Commanded-based Elixir applications?

Commands in Commanded represent actions to change the system state, while events record these changes. Developers define commands and events using Elixir structs, handling them with aggregates or processors to implement business logic and state transitions.

What is the purpose of Projectors and Projections in Elixir's CQRS/Event Sourcing systems?

Projectors and Projections manage the read model in CQRS/Event Sourcing systems. They listen to events and update view models or databases to reflect the current application state, supporting efficient data queries.