# Orkestra A lightweight CQRS/ES toolkit for Elixir. Pluggable message bus, event store, and OpenTelemetry tracing built in. Orkestra gives you the building blocks without the framework lock-in. Define commands and events with a declarative DSL, wire up handlers that auto-subscribe, and swap between in-process PubSub and distributed RabbitMQ with a single config change. ## Installation ```elixir def deps do [ {:orkestra, "~> 0.1.0"} ] end ``` ## Quick start ### Define a command ```elixir defmodule MyApp.Commands.PlaceOrder do use Orkestra.Command param :product_id, :string, required: true param :quantity, :integer, required: true param :notes, :string, default: "" end ``` ### Define an event ```elixir defmodule MyApp.Events.OrderPlaced do use Orkestra.Event field :order_id, :string, required: true field :product_id, :string, required: true field :quantity, :integer, required: true field :placed_at, :string, required: true end ``` ### Handle the command ```elixir defmodule MyApp.Handlers.PlaceOrderHandler do use Orkestra.CommandHandler, command: MyApp.Commands.PlaceOrder @impl true def execute(command, metadata) do order_id = Ecto.UUID.generate() # ... your business logic ... {:ok, %{order_id: order_id}} end end ``` ### React to the event ```elixir defmodule MyApp.Handlers.SendOrderConfirmation do use Orkestra.EventHandler, event: MyApp.Events.OrderPlaced @impl true def handle_event(event, _metadata) do MyApp.Mailer.send_confirmation(event.data.order_id) :ok end end ``` ### Dispatch ```elixir alias Orkestra.{CommandEnvelope, MessageBus} {:ok, cmd} = MyApp.Commands.PlaceOrder.new(%{ product_id: "sku_42", quantity: 3 }, actor_id: "user_123", source: "web") bus = MessageBus.impl() :ok = bus.dispatch(CommandEnvelope.wrap(cmd, max_retries: 2)) ``` ### Supervision tree ```elixir children = [ {Phoenix.PubSub, name: MyApp.PubSub}, Orkestra.MessageBus.PubSub, MyApp.Handlers.PlaceOrderHandler, MyApp.Handlers.SendOrderConfirmation ] ``` ## Configuration ### Message bus adapter ```elixir # In-process (dev, test, single-node) config :orkestra, Orkestra.MessageBus, adapter: Orkestra.MessageBus.PubSub, app_prefix: MyApp config :orkestra, Orkestra.MessageBus.PubSub, pubsub: MyApp.PubSub # Distributed (production, multi-node) config :orkestra, Orkestra.MessageBus, adapter: Orkestra.MessageBus.RabbitMQ, app_prefix: MyApp config :orkestra, Orkestra.MessageBus.RabbitMQ, channel_provider: fn -> MyApp.RabbitMQ.Connection.channel() end ``` ### Topic derivation Topics are derived automatically from module names. The `app_prefix` is stripped: ``` MyApp.Orders.Commands.PlaceOrder -> "orders.commands.place_order" MyApp.Orders.Events.OrderPlaced -> "orders.events.order_placed" ``` ### Event store ```elixir # EventStoreDB (production) config :orkestra, Orkestra.EventStore, adapter: Orkestra.EventStore.EventStoreDB config :orkestra, Orkestra.EventStore.EventStoreDB, connection_string: "esdb://localhost:2113?tls=false" # In-memory (test) config :orkestra, Orkestra.EventStore, adapter: Orkestra.EventStore.InMemory ``` ## Core concepts ### Commands Commands represent an intent to change the system. They are validated, dispatched to a single handler, and either succeed or fail. ```elixir defmodule CreateAccount do use Orkestra.Command param :email, :string, required: true param :name, :string, required: true param :plan, :string, default: "free" @impl true def validate(%{email: email}) do if String.contains?(email, "@"), do: :ok, else: {:error, :invalid_email} end end {:ok, cmd} = CreateAccount.new(%{email: "a@b.com", name: "Alice"}) {:error, {:missing_params, [:email]}} = CreateAccount.new(%{name: "Bob"}) {:error, :invalid_email} = CreateAccount.new(%{email: "nope", name: "Eve"}) ``` ### Events Events represent something that happened. They are immutable facts, never rejected. Events can be derived from commands or other events, preserving the correlation chain. ```elixir defmodule AccountCreated do use Orkestra.Event field :account_id, :string, required: true field :email, :string, required: true field :plan, :string, required: true end # From a command (preserves correlation, sets causation) {:ok, event} = AccountCreated.from_command(cmd, %{ account_id: "acc_123", email: "a@b.com", plan: "free" }) event.metadata.correlation_id == cmd.metadata.correlation_id # true event.metadata.causation_id == cmd.id # true ``` ### Metadata Every command and event carries metadata that flows through the pipeline: ```elixir %Orkestra.Metadata{ correlation_id: "abc123", # links an entire chain of commands/events causation_id: "cmd_456", # what directly caused this actor_id: "user_789", # who initiated it actor_type: :user, # :user | :system | :expert | :scheduler source: "web", # where it originated issued_at: ~U[2026-03-27 12:00:00Z] } # Derive child metadata (preserves correlation, sets causation) child = Orkestra.Metadata.derive(parent_metadata, "parent_id") ``` ### Envelopes Envelopes wrap commands and events with dispatch context. **Command envelopes** track dispatch lifecycle and retries: ```elixir env = CommandEnvelope.wrap(cmd, max_retries: 3) env.status # :pending -> :dispatched -> :succeeded | :failed | :rejected CommandEnvelope.retryable?(env) # true if failed and attempts <= max_retries ``` **Event envelopes** track delivery to multiple handlers: ```elixir env = EventEnvelope.wrap(event) |> EventEnvelope.mark_published() env = EventEnvelope.register_handler(env, "SendEmail") env = EventEnvelope.register_handler(env, "UpdateIndex") env = EventEnvelope.mark_handler_succeeded(env, "SendEmail") env = EventEnvelope.mark_handler_failed(env, "UpdateIndex") env.status # :partially_handled ``` ### Aggregates Aggregates encapsulate domain invariants as pure functions. `Aggregate.Root` is the imperative shell that handles all I/O. ```elixir defmodule MyApp.BankAccount do @behaviour Orkestra.Aggregate @impl true def init_state, do: %{status: :new, balance: 0} @impl true def stream_id(command), do: "bank_account-#{command.params.account_id}" @impl true def evolve(state, %AccountOpened{} = e), do: %{state | status: :open, owner: e.data.owner} def evolve(state, %MoneyDeposited{} = e), do: %{state | balance: state.balance + e.data.amount} def evolve(state, _), do: state @impl true def decide(%{status: :new}, %OpenAccount{} = cmd) do {:ok, [AccountOpened.new!(%{account_id: cmd.params.account_id, owner: cmd.params.owner})]} end def decide(%{status: :open}, %Deposit{} = cmd) do {:ok, [MoneyDeposited.new!(%{amount: cmd.params.amount})]} end def decide(%{status: :new}, _), do: {:error, :account_not_opened} end # Execute via the imperative shell {:ok, events, new_state} = Orkestra.Aggregate.Root.execute(MyApp.BankAccount, open_cmd) ``` The `Root.execute/3` pipeline: load events from store → fold via `evolve/2` → call `decide/2` → append events → publish to bus → snapshot (if configured). Optimistic concurrency is handled automatically with retries on version conflicts. ### Event handlers Subscribe to one event, multiple events, or wildcard patterns: ```elixir # Single event use Orkestra.EventHandler, event: MyApp.Events.OrderPlaced # Multiple events use Orkestra.EventHandler, events: [MyApp.Events.OrderPlaced, MyApp.Events.OrderCancelled] # Wildcard pattern use Orkestra.EventHandler, topic: "orders.events.#" # With retry config use Orkestra.EventHandler, event: MyApp.Events.OrderPlaced, max_retries: 5 ``` ## Message bus adapters ### PubSub (in-process) Synchronous dispatch. Commands go to one handler, events broadcast to all subscribers. Retries are immediate (recursive). Dead-lettered messages are broadcast on `"orkestra:deadletter"`. ### RabbitMQ (distributed) Commands use exchange `orkestra.commands` with one queue per command type (competing consumers). Events use exchange `orkestra.events` with one queue per handler (fan-out). Features: - Retry tracking via `x-death` headers (native RabbitMQ) - Max retries via `x-max-retries` header - Dead letter exchange `orkestra.deadletter` with catch-all queue - All queues declared with DLX configuration - W3C trace context propagation in AMQP headers ## Observability Orkestra is instrumented with OpenTelemetry out of the box. ### Span hierarchy ``` orkestra.command.dispatch (message bus) orkestra.command.handle (command handler) orkestra.event.publish (message bus) orkestra.event.handle (event handler) ``` Additional spans: `orkestra.retry`, `orkestra.rabbitmq.publish` (kind: producer), `orkestra.rabbitmq.consume` (kind: consumer). Aggregate spans: `orkestra.aggregate.execute` → `orkestra.aggregate.load` → `orkestra.aggregate.fold` → `orkestra.aggregate.decide` → `orkestra.aggregate.append` → `orkestra.aggregate.publish` → `orkestra.aggregate.snapshot`. ### Span attributes All spans include: `orkestra.command.type`, `orkestra.command.id`, `orkestra.correlation_id`, `orkestra.causation_id`, `orkestra.actor_id`, `orkestra.handler`. ### Structured logging All log messages use Logger metadata instead of string interpolation: ``` [info] Command handler subscribed handler=MyApp.HandleOrder topic=orders.commands.place_order orkestra=command_handler [warning] Handler nack, requeuing handler=MyApp.HandleOrder attempt=2 max_retries=3 orkestra=rabbitmq [error] Dead letter recorded handler=MyApp.HandleOrder reason=:timeout orkestra=pubsub ``` Logger metadata set during handler execution: `correlation_id`, `causation_id`, `actor_id`, `trace_id`, `span_id`. ### Distributed tracing (RabbitMQ) Trace context is injected into AMQP message headers on publish and extracted on consume, creating linked spans across nodes. Uses `OpentelemetryProcessPropagator` for context propagation across BEAM processes. ## MCP server (orkestra_mcp) `orkestra_mcp` is a companion MCP server and CLI that scaffolds and introspects Orkestra projects. It exposes tools, resources, and prompts to any MCP-compatible AI assistant. ### Installation Build the escript from the `orkestra_mcp/` directory: ```bash cd orkestra_mcp mix deps.get mix escript.build ``` ### Usage Run the server, pointing it at your project: ```bash ./orkestra_mcp --project-dir /path/to/your/app ``` Register it in your MCP client configuration (e.g., `.mcp.json`): ```json { "mcpServers": { "orkestra": { "command": "/path/to/orkestra_mcp", "args": ["--project-dir", "/path/to/your/app"] } } } ``` ### Tools (code generators) | Tool | Description | |------|-------------| | `gen_command` | Scaffold a Command module with typed params | | `gen_event` | Scaffold an Event module with typed fields | | `gen_command_handler` | Scaffold a CommandHandler wired to a command | | `gen_event_handler` | Scaffold an EventHandler (single, multi, or wildcard) | | `gen_aggregate` | Scaffold an Aggregate with `decide`/`evolve` clauses | ### Resources (project introspection) | Resource URI | Description | |--------------|-------------| | `orkestra://commands` | JSON list of all Command modules discovered in the project | | `orkestra://events` | JSON list of all Event modules discovered in the project | | `orkestra://handlers` | JSON list of all handler modules discovered in the project | | `orkestra://aggregates` | JSON list of all Aggregate modules discovered in the project | | `orkestra://domain-map` | Cross-referenced domain map of commands, events, handlers, and aggregates | ### Prompts | Prompt | Description | |--------|-------------| | `conventions` | Orkestra CQRS/ES conventions and best practices | | `new_bounded_context` | Guided step-by-step workflow for adding a new bounded context | ## Testing ```bash mix test ``` Tests use ExUnit with `async: false` for bus-related tests (shared GenServer state). The in-memory PubSub and event store adapters are used throughout the test suite — no external services required. ## License MIT — see [LICENSE](LICENSE).