DSL macro for defining Orkestra projectors.

A projector consumes domain events and maintains a queryable read model. Supports two backends: :postgres (default) and :elasticsearch.

Postgres projector (default)

A Postgres-backed projector uses an Ecto.Repo for both the read model and the projection checkpoint. Define event handlers with the project/2 macro; the module generates the dispatch, config, and OTP child_spec boilerplate automatically.

defmodule MyApp.OrderProjector do
  use Orkestra.Projector,
    repo: MyApp.OrderProjection.Repo,
    event_store: Orkestra.EventStore.InMemory

  project MyApp.Events.OrderPlaced, fn event, multi ->
    order = %{id: event.data.order_id, status: "placed"}
    Ecto.Multi.insert(multi, :read_model_insert, order)
  end

  project MyApp.Events.OrderCancelled, fn event, multi ->
    Ecto.Multi.update_all(multi, :read_model_update, ...)
  end
end

The multi parameter is a pre-built empty Ecto.Multi.new() that the handler chains operations onto. Step names must use the :read_model_ prefix to avoid name collisions with the GenServer's reserved steps (:checkpoint, :halted_checkpoint, :dead_letter).

Elasticsearch projector

An Elasticsearch-backed projector writes documents to an ES/OpenSearch index. The checkpoint is still stored in Postgres (:repo is always required). Use project_es/2 to declare event handlers; the handler must return {:ok, doc, id}, :skip, or {:error, reason}.

defmodule MyApp.OrderESProjector do
  use Orkestra.Projector,
    backend: :elasticsearch,
    repo: MyApp.OrderProjection.Repo,
    cluster: MyApp.ESCluster,
    index: "orders",
    event_store: Orkestra.EventStore.InMemory

  @impl true
  def index_mapping do
    %{
      "mappings" => %{
        "properties" => %{
          "order_id" => %{"type" => "keyword"},
          "status"   => %{"type" => "keyword"}
        }
      }
    }
  end

  project_es MyApp.Events.OrderPlaced, fn event, _position ->
    {:ok, %{"order_id" => event.data.order_id, "status" => "placed"},
     event.data.order_id}
  end
end

The GenServer calls Storage.Elasticsearch.init/1 at startup (via the :init_adapter message) to detect the engine and create the index before processing any events.

Elasticsearch projector with a schema

Instead of a raw index: name plus a hand-written index_mapping/0, an ES projector may declare an Orkestra.ES.Schema module. The projector then writes to the schema's alias (compatible with the alias + versioning index lifecycle) and index_mapping/0 is generated from the schema:

defmodule MyApp.OrderESProjector do
  use Orkestra.Projector,
    backend: :elasticsearch,
    repo: MyApp.OrderProjection.Repo,
    cluster: MyApp.ESCluster,
    schema: MyApp.Search.Order,
    culture: :it,
    event_store: Orkestra.EventStore.InMemory

  project_es MyApp.Events.OrderPlaced, fn event, _position ->
    {:ok, %MyApp.Search.Order{order_id: event.data.order_id, status: "placed"}}
  end
end

With schema:, a project_es/2 handler may return either the legacy {:ok, doc, id} tuple or {:ok, %SchemaStruct{}} — in the latter case the document and _id are derived from the schema (to_doc/1 and the primary-key field). :schema and :index are mutually exclusive, and defining index_mapping/0 manually alongside :schema is a compile error (the schema is the single source of truth).

:culture is only valid with :schema. For a multi-culture schema it defaults to the schema's default_culture; for a mono-culture schema it must be omitted.

Options for use Orkestra.Projector

• :repo (required) — the Ecto.Repo module for the projection checkpoint. For ES projectors this is the checkpoint Postgres repo; it does not store the read-model data.
• :backend (optional) — :postgres (default) or :elasticsearch.
• :cluster (required for ES) — the Snap.Cluster module.
• :index (ES, legacy path) — the Elasticsearch index name string. Mutually exclusive with :schema; requires a user-defined index_mapping/0.
• :schema (ES, schema path) — an Orkestra.ES.Schema module. The projector writes to the schema alias and index_mapping/0 is generated. Mutually exclusive with :index.
• :culture (ES, schema path only) — the culture atom; defaults to the schema's default_culture for multi-culture schemas, must be omitted for mono-culture schemas.
• :event_store (optional) — event store module; defaults to Orkestra.EventStore.
• :name (optional) — override the projector name string; defaults to inspect(__MODULE__).
• :max_retries (optional) — maximum retry attempts before halting; defaults to 5.
• :backoff_base_ms (optional) — base delay for exponential backoff in milliseconds; defaults to 500.
• :backoff_cap_ms (optional) — maximum backoff delay in milliseconds; defaults to 30_000.

The project/2 macro (Postgres backend)

Declares a handler for a specific event type:

project EventModule, fn event, multi -> multi end

The handler receives the event struct and an empty Ecto.Multi.new(). It must return an Ecto.Multi.t() — the multi is then wrapped in {:ok, multi} by the generated __handle__/3 bridge function.

The project_es/2 macro (Elasticsearch backend)

Declares a handler for a specific event type in an ES projector:

project_es EventModule, fn event, position ->
  {:ok, %{"field" => value}, document_id}
end

The handler receives (event, position) and must return one of:

• {:ok, doc, id} — index the document with deterministic _id
• :skip — skip this event (no ES write)
• {:error, reason} — signal failure

Generated functions

Postgres backend:

• __dispatch__/3 — routes by event type string; returns {:ok, Ecto.Multi.t()} for registered events or :skip for unknown ones.
• __handle__/3 — adapter-facing bridge: calls __dispatch__/3 and translates :skip into {:ok, Ecto.Multi.new()}.

Elasticsearch backend:

• __dispatch_es__/3 — routes by event type string; returns {:ok, doc, id} for registered events or :skip for unknown ones.
• __handle_es__/3 — adapter-facing bridge: calls __dispatch_es__/3 and passes through {:ok, doc, id}, :skip, or {:error, reason}.

Both backends:

• __projection_config__/0 — returns a map with :repo, :projector_name, :migrations_path, and :migration_source; used by mix tasks for discovery.
• child_spec/1 — returns a supervisor child spec targeting Orkestra.Projector.GenServer. For ES projectors the spec injects Storage.Elasticsearch and the necessary adapter_opts.

child_spec/1 and runtime overrides

child_spec/1 accepts a keyword list of overrides for runtime config:

# In your supervision tree
children = [
  {Orkestra.Projection.Supervisor, projectors: [
    MyApp.OrderProjector,
    {MyApp.CustomerProjector, repo: MyApp.CustomerProjection.TestRepo}
  ]}
]

projection_config/0 return shape

Used by mix tasks (e.g. mix projector.migrate, mix orkestra.projection.es.rebuild) to discover per-projection repos, migration paths, and backend-specific configuration:

%{
  repo: MyApp.OrderProjection.Repo,
  projector_name: "MyApp.OrderProjector",
  migrations_path: "priv/projections/myapp_order_projector/migrations",
  migration_source: "projection_myapp_order_projector_schema_migrations",
  backend: :postgres,
  cluster: nil,
  index: nil,
  projector_module: MyApp.OrderProjector
}

For Elasticsearch projectors the map additionally contains:

%{
  ...
  backend: :elasticsearch,
  cluster: MyApp.ESCluster,
  index: "orders",
  projector_module: MyApp.OrderESProjector
}

Per-Projection Repo Configuration

Each projector uses its own isolated Ecto.Repo. This keeps migrations, tables, and migration history fully independent across projections.

Example config.exs

config :my_app, MyApp.OrderProjection.Repo,
  database: "my_app_dev",
  hostname: "localhost",
  migration_source: "projection_myapp_order_projector_schema_migrations",
  priv: "priv/projections/myapp_order_projector"

The :migration_source key sets the migrations tracking table name so each projection's migration history is isolated from the app's main schema_migrations table. The :priv key points Mix to the correct migrations directory.

Defining the Repo

defmodule MyApp.OrderProjection.Repo do
  use Ecto.Repo,
    otp_app: :my_app,
    adapter: Ecto.Adapters.Postgres
end

Add the Repo to your supervision tree:

children = [
  MyApp.OrderProjection.Repo,
  {Orkestra.Projection.Supervisor, projectors: [MyApp.OrderProjector]}
]