Home Posts Post Search Tag Search

Real World Event Sourcing - 06 - Building Resilient Applications with Event Stores
Published on: 2026-07-02 Tags: elixir, Ecto, Supervisor Tree, event-sourcing, cqrs, aggregates, Real World Event Sourcing, Commanded, EventHandler, ProcessManager, EventStoreDB, Redis

6. Building Resilient Applications with Event Stores

This is where we learn to store the data that we have built. We have only worked with GenServers that lose all state once we turn them off. We will need to build storage that is easily accessed and then build the current state from those.



Evaluating Event and Projection Stores

There should be three things within any data store:

• Aggregate state (snapshots)
• Projections (materialized views, read model)
• Event Log (write model)

The aggregate should be an O(1) call time and should be a snapshot. You should use some products that are designed to work with Snapshots. These will make the read and write of them a lot faster.


Projections are a bit different. This is where we might have a few rows or even projections for a single entity. We want to have O(1) read for any query that we might want to make.


The event log is the source of truth. This is going to be a series of logged events that everything else is based off of. You want to be able to destroy the others and build from this. A time series DB is pretty normal for this. One of the things to pull from this is that you should choose what you want to store separately from the different types of storage.



Replaying Events

This is your recovery or you changed an aggregate and need to rerun the state catch all. There are some issues to keep in mind but as long as you adhere to them you will be fine.


Verifying Models for Replay

You must be able to build the same state no matter how many times you replay. You must be able to use past events in the future, if there is a need to update an entity you must do that with a new event, not a modification to an older event. If you get a different state from 2 different replays your model is “effectful”.



Capacity Planning

One of the things that you need to worry about here is that you are storing “more” information that a standard one line per user DB, there will be growth for the db as you continue to add more users, and your users use the system. One of the ways to estimate the growth is to: take the largest event, take the average event per hour that you receive, then multiply it by 720(30 x 24) and that is the rough estimate of the growth of the DB over 1 month.



Exploring Event Store

The EventStoreDB is an open-source database/tool that specializes in storing events. You can simply run this docker command to get it up and running. Then head to http://localhost:2113

docker run --name esdb-node -it -p 2113:2113 -p 1113:1113 \
eventstore/eventstore:latest --insecure --run-projections=All

Within the browser you can add in a stream within the stream browser tab. Add this to the stream with: Stream ID account-ledger, and Event Type amountDeposited.

{
  "accountNumber": "ABC",
  "amount": 100
}

You can now head to the leader and see the new stream.


Creating an EventStoreDB Projection

This will be a state produced by JavaScript that runs against the stream, and optionally emits events to output streams.


Head to the Projections tab and hit the New Projection button. The name will be account-balances it will run in continuous mode, then enter the following into the source field:

function getBalance(balances, accountNumber) {
  if (accountNumber in balances) {
    return balances[accountNumber];
  } else {
    return 0;
  }
}

options({
  $includeLinks: false,
  reorderEvents: false,
  processingLag: 0
})

fromStream('account-ledger')
  .when({
    $init: function() {
      return {
        balances: {}
      }
    },
    amountDeposited: function(state, event) {
      const evt = JSON.parse(event.bodyRaw);
      const newbalance =
        getBalance(state.balances, evt.accountNumber) + evt.amount;
      state.balances[evt.accountNumber] = newbalance;
    },
    amountWithdrawn: function(state, event) {
      const evt = JSON.parse(event.bodyRaw);
      const newbalance =
        getBalance(state.balances, evt.accountNumber) - evt.amount;
      state.balances[evt.accountNumber] = newbalance;
    }
  })
  .outputState()

This might look like a lot but you can see that once we set some event types and make them standard we will be able to see the current state of the account with the current state. The events are amountDeposited and amountWithdrawn. This is a powerful tool and using it right will make EventStorage much easier.



Upgrading Lunar Frontiers

Okay make a copy of the lunar_frontiers_1 folder _2 and we can work with this for now. Also we will need to work with PostgresSQL adapter.


We will need to add this dependency to the mix.ex file.

{:commanded_eventstore_adapter, "~> 1.4"},

Now we need to edit the config.exs file in order to use the new adapter.

import Config

config :lunar_frontiers, LunarFrontiers.App.Application,
  event_store: [
    adapter: Commanded.EventStore.Adapters.EventStore,
    event_store: LunarFrontiers.EventStore
  ],
  pubsub: :local,
  registry: :local

config :lunar_frontiers, event_stores: [LunarFrontiers.EventStore]

config :lunar_frontiers, LunarFrontiers.EventStore,
  serializer: Commanded.Serialization.JsonSerializer,
  username: "postgres",
  password: "postgres",
  database: "eventstore",
  hostname: "localhost"

You can see that this references a event_store module you can add that now to lib/lunar_frontiers directory as event_store.ex

defmodule LunarFrontiers.EventStore do
  use EventStore, otp_app: :lunar_frontiers

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

You might now need to install Postgres on your system so do so now. Check out the install steps from the Postgres site. Once that is done you should be able to run the following command from your lunar_frontiers_2 folder.

mix do event_store.create, event_store.init

For linux you will need to access the correct plsql database and that looks something like:

psql -h localhost -U postgres -d eventstore
# Then inside

\dt public.*

Now that its up and running within the same folder start up and iex session we will advance the game and then check the psql session for changes.

iex -S mix
# in the iex session
iex(1)> LunarFrontiers.App.Application.dispatch(\
...(1)> %LunarFrontiers.App.Commands.AdvanceGameloop{game_id: 1, tick: 1})

# In the psql session
eventstore=# select event_type, metadata from events;
                    event_type                     | metadata 
---------------------------------------------------+----------
 Elixir.LunarFrontiers.App.Events.GameloopAdvanced | \x7b7d
(1 row)

# In the iex session
iex(2)> LunarFrontiers.App.Application.dispatch(
%LunarFrontiers.App.Commands.SpawnSite{
completion_ticks: 2, location: 1, player_id: 1,
site_id: 1, site_type: 1, tick: 1
})
iex(3)> LunarFrontiers.App.Application.dispatch(
%LunarFrontiers.App.Commands.AdvanceGameloop{
game_id: 1, tick: 2})
iex(4)> LunarFrontiers.App.Application.dispatch(
%LunarFrontiers.App.Commands.AdvanceGameloop{
game_id: 1, tick: 3})
iex(5)> :ets.tab2list(:buildings)
[
{1,
%{
complete: 100.0,
location: 1,
player_id: 1,
ready: true,
site_id: 1,
site_type: 1
}}
]

# In the psql session
eventstore=# select event_type from events; event_type
                       event_type                        
---------------------------------------------------------
 Elixir.LunarFrontiers.App.Events.GameloopAdvanced
 Elixir.LunarFrontiers.App.Events.SiteSpawned
 Elixir.LunarFrontiers.App.Events.GameloopAdvanced
 Elixir.LunarFrontiers.App.Events.ConstructionProgressed
 Elixir.LunarFrontiers.App.Events.GameloopAdvanced
 Elixir.LunarFrontiers.App.Events.ConstructionProgressed
 Elixir.LunarFrontiers.App.Events.ConstructionCompleted
 Elixir.LunarFrontiers.App.Events.BuildingSpawned
(8 rows)

Okay so we now have all the events within the psql but the projection will not be the same once we quit and run it again. Fear not as we can reset the run tasks with mix commanded.reset or within the terminal 

iex(1)> :ets.tab2list(:buildings)
[]
iex(2)> Mix.Tasks.Commanded.Reset.run(["--app",
...(2)> "LunarFrontiers.App.Application", "--handler",
...(2)> "LunarFrontiers.App.Projectors.Building"])
Resetting "LunarFrontiers.App.Projectors.Building"
iex(3)> :ets.tab2list(:buildings)
[
  {1,
   %{
     complete: 100.0,
     location: 1,
     site_id: 1,
     site_type: 1,
     player_id: 1,
     ready: true
   }}
]

So now that we have a way to reset the projections we are now able to build from the events we have stored.



Adding Durable Projections to Lunar Frontiers

We will now go through and take care of some unfinished business.


Fixing a Weak Link

Right now we have a SystemsTrigger that listens for the game loop to advance. This will take events and then issue commands but its not a real Process Manager we need to upgrade it one. We also have an issue as it also reads from the Projections. That is a big no no as “Process Managers Must Not Read from Projections.” This might feel easy as they are there to make an O(1) read but this will get you into trouble if at any point they are not update or even if you no longer use that projection.


The book wants you to swap to lunar_frontiers_3 because there are a lot of changes that need to take place.

cp -r "/mnt/c/Users/*user*/path/to/folder/lunar_frontiers_3" ~/Projects/real_world_event_sourcing/

Head to the file game_loop_manager.ex and check out the new code.

#---
# Excerpted from "Real-World Event Sourcing",
# published by The Pragmatic Bookshelf.
# Copyrights apply to this code. It may not be used to create training material,
# courses, books, articles, and the like. Contact us if you are in doubt.
# We make no guarantees that this code is fit for any purpose.
# Visit https://pragprog.com/titles/khpes for more book information.
#---
defmodule LunarFrontiers.App.ProcessManagers.GameLoopManager do
  require Logger
  alias LunarFrontiers.App.Commands.AdvanceConstruction

  alias LunarFrontiers.App.Events.{
    GameloopAdvanced,
    GameStarted,
    ConstructionCompleted,
    SiteSpawned,
    GameStopped
  }

  alias __MODULE__

  use Commanded.ProcessManagers.ProcessManager,
    application: LunarFrontiers.App.Application,
    name: __MODULE__

  @derive Jason.Encoder
  defstruct [
    :current_tick,
    :active_construction_sites,
    :game_id
  ]

  def interested?(%GameStarted{game_id: gid}), do: {:start, gid}
  def interested?(%SiteSpawned{game_id: gid}), do: {:continue, gid}
  def interested?(%ConstructionCompleted{game_id: gid}), do: {:continue, gid}
  def interested?(%GameloopAdvanced{game_id: gid}), do: {:continue, gid}
  def interested?(%GameStopped{game_id: gid}), do: {:stop, gid}
  def interested?(_event), do: false

  def handle(%__MODULE__{} = state, %GameloopAdvanced{tick: tick}) do
    sites = state.active_construction_sites || []

    construction_cmds = sites
      |> Enum.map(fn site_id ->
        %AdvanceConstruction{
          site_id: site_id,
          tick: tick,
          game_id: state.game_id,
          advance_ticks: 1
        }
      end)

    construction_cmds
  end

  def apply(%GameLoopManager{} = state, %GameloopAdvanced{tick: tick}) do
    %GameLoopManager{
      state
      | current_tick: tick
    }
  end

  def apply(%GameLoopManager{} = _state,
    %GameStarted{game_id: gid} = _evt) do
    %GameLoopManager{
      current_tick: 0,
      game_id: gid,
      active_construction_sites: []
    }
  end

  def apply(%GameLoopManager{} = state,
    %SiteSpawned{site_id: sid, tick: t} = _evt) do
    %GameLoopManager{
      current_tick: t,
      game_id: state.game_id,
      active_construction_sites: state.active_construction_sites ++ [sid]
    }
  end

  def apply(%GameLoopManager{} = state,
    %ConstructionCompleted{site_id: sid, tick: t} = _evt) do
    %GameLoopManager{
      current_tick: t,
      game_id: state.game_id,
      active_construction_sites: state.active_construction_sites -- [sid]
    }
  end

  # By default skip any problematic events
  def error(error, _command_or_event, _failure_context) do
    Logger.error(fn ->
      "#{__MODULE__} encountered an error: #{inspect(error)}"
    end)

    :skip
  end
end

This now no longer uses the projections and its monitored by the internal state of the process manager.


The game loop process manager is initialized by a new event, GameStarted, we should add that to the GameLoop aggregate:

  ### This is a new function that will pattern match.
  def execute(%Gameloop{} = _loop, %StartGame{game_id: gid}) do
    event = %GameStarted{game_id: gid}
    {:ok, event}
  end

Just a few more tweaks and we are done. Let’s update the router.ex by adding in the StartGame to the commands.

#---
# Excerpted from "Real-World Event Sourcing",
# published by The Pragmatic Bookshelf.
# Copyrights apply to this code. It may not be used to create training material,
# courses, books, articles, and the like. Contact us if you are in doubt.
# We make no guarantees that this code is fit for any purpose.
# Visit https://pragprog.com/titles/khpes for more book information.
#---
defmodule LunarFrontiers.App.Router do
  alias LunarFrontiers.App.Commands.{
    AdvanceGameloop,
    StartGame,
    AdvanceConstruction,
    SpawnSite,
    SpawnBuilding
  }

  alias LunarFrontiers.App.Aggregates.{
    Gameloop,
    ConstructionSite,
    Building
  }

  use Commanded.Commands.Router

  identify(Gameloop,
    by: :game_id,
    prefix: "game-"
  )

  identify(ConstructionSite,
    by: :site_id,
    prefix: "site-"
  )

  identify(Building,
    by: :site_id,
    prefix: "bldg-"
  )

  dispatch([AdvanceGameloop, StartGame], to: Gameloop)
  dispatch([SpawnSite, AdvanceConstruction], to: ConstructionSite)
  dispatch([SpawnBuilding], to: Building)
end

Lastly we need to update the OTP Supervisor to include the new GameLoopManager within the Supervisor Tree.

defmodule LunarFrontiers.App.Supervisor do
  use Supervisor

  alias LunarFrontiers.App

  def start_link(arg) do
    Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
  end

  @impl true
  def init(_arg) do
    children = [
      # Application
      App.Application,

      # Process Managers
      App.ProcessManagers.Construction,
      App.ProcessManagers.GameLoopManager,

      # Projectors (read model)
      App.Projectors.Building
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

You may not start up the game. Run these commands to clear the database and then run the game. You might get an error saying that the database couldn’t be dropped but that is because you might still have a psql session attached to the database in the background; closing that terminal will free it up.

mix event_store.drop
mix do event_store.create, event_store.init

Projecting with Redis

We now need a way to keep our projections up-to-date, and Amazon’s RDS does this. First let’s add the dependencies to the mix.exs

{:redix, "~> 1.1"},

## Also check out the supervisor.ex
  @impl true
  def start(_type, _args) do
    children = [
      {App.Supervisor, nil},
      {Redix, name: :projections}
    ]

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

Now we can look at the building.ex data that will be responsible for keeping the projection up-to-date for us.

#---
# Excerpted from "Real-World Event Sourcing",
# published by The Pragmatic Bookshelf.
# Copyrights apply to this code. It may not be used to create training material,
# courses, books, articles, and the like. Contact us if you are in doubt.
# We make no guarantees that this code is fit for any purpose.
# Visit https://pragprog.com/titles/khpes for more book information.
#---
defmodule LunarFrontiers.App.Projectors.Building do
  alias LunarFrontiers.App.Events.BuildingSpawned
  alias LunarFrontiers.App.Events.SiteSpawned
  alias LunarFrontiers.App.Events.ConstructionProgressed

  use Commanded.Event.Handler,
    application: LunarFrontiers.App.Application,
    name: __MODULE__


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

  def handle(
        %SiteSpawned{
          site_id: site_id,
          site_type: site_type,
          location: location,
          player_id: player_id
        },
        _metadata
      ) do

    building = %{
      complete: 0.0,
      site_id: site_id,
      site_type: site_type,
      player_id: player_id,
      location: location,
      ready: false
    } |> Jason.encode!()

    Redix.command(:projections,
      ["SET", projection_key(site_id), building])
    Redix.command(:projections,
      ["SADD", player_sites_key(player_id), site_id])

    :ok
  end

  def handle(
        %BuildingSpawned{
          site_id: site_id,
          site_type: site_type,
          location: location,
          player_id: player_id,
          tick: t
        },
        _metadata
      ) do

    building = %{
      complete: 100.0,
      site_id: site_id,
      spawned_tick: t,
      site_type: site_type,
      location: location,
      player_id: player_id,
      ready: true
    } |> Jason.encode!()

    Redix.command(:projections, ["SET", projection_key(site_id), building])

    :ok
  end

  def handle(
        %ConstructionProgressed{
          site_id: site_id,
          site_type: site_type,
          location: loc,
          player_id: player_id,
          progressed_ticks: p,
          required_ticks: r
        },
        _metadata
      ) do
    building = %{
      player_id: player_id,
      complete: Float.round(r / p * 100, 1),
      site_id: site_id,
      site_type: site_type,
      location: loc,
      ready: false
    } |> Jason.encode!()

    Redix.command(:projections, ["SET", projection_key(site_id), building])

    :ok
  end

  defp projection_key(id) do
    "building:#{id}"
  end

  defp player_sites_key(player_id) do
    "sites:#{player_id}"
  end

end

We now have some clean code to work from. We could start up an iex session with iex -S mix but we have a script that we can run created by the book creators. This is a bit of what we will be doing in future chapters when we test Event-Sourced Systems.

alias LunarFrontiers.App.Commands
import LunarFrontiers.App.Application

player_id = "px42"

gid = UUID.uuid4()
sid = UUID.uuid4()

IO.puts "New game #{gid}, going to build site #{sid}"

dispatch(%Commands.StartGame{game_id: gid})
dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 1})

dispatch(%Commands.SpawnSite{
  completion_ticks: 5,
  location: 1,
  player_id: player_id,
  site_id: sid,
  site_type: 1,
  tick: 1,
  game_id: gid
})

dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 2})
dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 3})
dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 4})
dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 5})
dispatch(%Commands.AdvanceGameloop{game_id: gid, tick: 6})

Now from the main project folder run

iex -S mix run scripts/single_site.exs

Then from the psql session that is attached to the database

eventstore=# select stream_uuid from streams;
stream_uuid

Projecting with Ecto

The Commanded Library also has some built-in functionality for projectors. We won’t be using it but here is some code that could run it.

defmodule Projector do
  use Commanded.Projections.Ecto,
    application: MyApp.Application,
    name: "my-projection",
    repo: MyApp.Repo,
    schema_prefix: "my-prefix",
    timeout: :infinity

  project(%Event{}, _metadata, fn multi ->
    Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
  end)

  project(%AnotherEvent{}, fn multi ->
    Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
  end)
end


Wrapping Up

We now have stored memory for the project. This is durable and we have shown that just having the events will allow a game to come back to its full played state no matter what happens as long as the events are logged.

Back to posts