Event Sourcing
Architecture

What is Event Sourcing?

Traditional systems store current state — a row that gets overwritten.
Event Sourcing stores what happened — an append-only log of immutable events.

  Traditional:   Entity { name: "Acme Corp", status: "active" }  ← overwrites

  Event Sourced:
    1  TenantRegistered  { name: "Acme" }
    2  TenantUpdated     { name: "Acme Corp" }
    3  TenantSuspended   { reason: "..." }
                ↑ nothing is ever deleted or overwritten
  

How It Helps Us

Our Goals

1. Write side — all mutations through Commanded aggregates. No direct DB writes.
2. Read side — Ash resources projected from events, optimised per query pattern.
3. Multiple read models — same event stream fans out to PostgreSQL, Cassandra, and beyond.
4. Compute pipeline — formula / rollup / lookup values computed reactively from events.
5. Workflow engine — task execution driven by events, not polling.
6. Multi-region — event log is source of truth; read models rebuilt anywhere.

The Model — Write Side (Aggregates)

Each aggregate is a consistency boundary — handles its own commands, emits events, never reads another aggregate's state.

The Model — Read Side (Ash Resources)

Read models are projections — built from events, never written to directly. Each is optimised for its access pattern.

Client sends a GraphQL request

# Mutation → write path → command dispatched to Commanded
mutation {
  createEntity(collectionId: "...", values: { name: "Acme" }) {
    id
  }
}

# Query → read path → Ash resource, no aggregates involved
query {
  entities(collectionId: "...") { id values }
}
  

Resolver translates mutation → Command

# apps/core_web/lib/core_web/graphql/resolvers/data_resolver.ex

def create_entity(%{input: input}, %{context: %{core_context: ctx}}) do
  cmd = %Core.Data.Commands.CreateEntity{
    id:            Ecto.UUID.generate(),
    tenant_id:     ctx.tenant_id,
    collection_id: input.collection_id,
    values:        input.values || %{}
  }
  Core.CommandedApp.dispatch(cmd)
end
  

Router routes Command to the Aggregate

# apps/core/lib/core/router.ex

identify(Core.Data.Aggregates.Entity, by: :id)

dispatch(
  [Core.Data.Commands.CreateEntity, ...],
  to: Core.Data.Aggregates.Entity
)
  

Aggregate validates → returns an Event

# apps/core/lib/core/data/aggregates/entity.ex

def execute(%__MODULE__{id: nil}, %CreateEntity{} = cmd) do
  # ✓ valid — emit the event
  %EntityCreated{
    id:            cmd.id,
    tenant_id:     cmd.tenant_id,
    collection_id: cmd.collection_id,
    values:        cmd.values
  }
end

def execute(%__MODULE__{id: id}, %CreateEntity{}) when not is_nil(id),
  do: {:error, :entity_already_exists}
  

Event persisted to the EventStore

  EventStore (PostgreSQL, append-only)
  ┌──────────────────────────────────────────────────────────┐
  │ stream: "entity-550e8400-..."                            │
  │                                                          │
  │  seq │ event_type        │ data                          │
  │    1 │ EntityCreated     │ { id, tenant_id, values… }    │
  │    2 │ AttributeValueSet │ { attribute_id, value… }      │
  │    3 │ EntitiesLinked    │ { target_id… }                │
  └──────────────────────────────────────────────────────────┘
  

Aggregate state updated in memory

def apply(%__MODULE__{} = e, %EntityCreated{} = ev) do
  %__MODULE__{e |
    id:            ev.id,
    tenant_id:     ev.tenant_id,
    collection_id: ev.collection_id,
    values:        ev.values,
    status:        :active
  }
end
  

Event broadcast to Projectors

Projector writes to the Read Model

defmodule Core.Data.Projectors.EntityProjector do
  use Commanded.Projections.Ecto,
    application: Core.CommandedApp,
    repo:        Core.Repo,
    name:        "entity_projector"

  project(%EntityCreated{} = e, _meta, fn multi ->
    Ecto.Multi.insert(multi, :entity, %Core.Data.Resources.Entity{
      id:            e.id,
      tenant_id:     e.tenant_id,
      collection_id: e.collection_id,
      values:        e.values,
      status:        :active
    })
  end)
end
  

Query reads from Read Model via Ash

# Resolver:
Ash.read(Core.Data.Resources.Entity,
  action: :in_collection,
  actor:  ctx,
  params: %{collection_id: collection_id}
)

# Ash resource:
defmodule Core.Data.Resources.Entity do
  use Ash.Resource, domain: Core.Data.Domain,
                    data_layer: AshPostgres.DataLayer

  read :in_collection do
    argument :collection_id, :uuid, allow_nil?: false
    filter expr(collection_id == ^arg(:collection_id) and status == :active)
  end
end
  

Full Flow Summary

Adding new things to the model

Example: adding Comment on an Entity

1. Command struct — AddComment{ id, entity_id, author_id, body }
2. Event struct — CommentAdded{ id, entity_id, author_id, body }
3. Aggregate — execute/2 + apply/2 (~20 lines)
4. Router entry — identify + dispatch (3 lines)
5. Migration — create comments table
6. Ash resource — use Ash.Resource with attributes and read actions
7. Projector — project(%CommentAdded{}, ...) writes to table
8. Resolver — translate GraphQL ↔ Command / Ash

New projection for a novel database

Example: Dgraph (property graph database) — relevant because our Entity link / backlink attributes are literally a graph. Multi-hop traversals are expensive in PostgreSQL, trivial in Dgraph.

defmodule Core.Data.Projectors.EntityProjector.Dgraph do
  use Commanded.EventHandler, application: Core.CommandedApp,
                              name: "entity_dgraph_projector"

  def handle(%EntityCreated{} = e, _meta) do
    Dlex.mutate(conn(), %{"uid" => "_:#{e.id}",
      "dgraph.type" => "Entity", "tenant_id" => e.tenant_id})
  end

  def handle(%EntitiesLinked{} = e, _meta) do
    Dlex.mutate(conn(), %{"uid" => e.from_id,
      "links" => [%{"uid" => e.to_id}]})
  end

  def handle(_, _), do: :ok
end
  

How do we scale?

Failure recovery

Multi-region deployment

All command dispatch routes to the primary region. Reads served from any region. Tenant affinity (consistent hash on tenant_id) avoids distributed transaction complexity.