feat(Ash.Reactor): Add load option to existing actions, and add new step type. (#1435)

documentation/dsls/DSL:-Ash.Reactor.md

@@ -429,6 +429,7 @@ Caveats/differences from `Ash.bulk_create/4`:
 
 ### Nested DSLs
  * [actor](#reactor-bulk_create-actor)
+ * [load](#reactor-bulk_create-load)
  * [tenant](#reactor-bulk_create-tenant)
  * [wait_for](#reactor-bulk_create-wait_for)
 
@@ -461,7 +462,6 @@ end
 | [`authorize_changeset_with`](#reactor-bulk_create-authorize_changeset_with){: #reactor-bulk_create-authorize_changeset_with } | `:filter \| :error` | `:filter` | If set to `:error`, instead of filtering unauthorized changes, unauthorized changes will raise an appropriate forbidden error |
 | [`authorize_query_with`](#reactor-bulk_create-authorize_query_with){: #reactor-bulk_create-authorize_query_with } | `:filter \| :error` | `:filter` | If set to `:error`, instead of filtering unauthorized query results, unauthorized query results will raise an appropriate forbidden error |
 | [`batch_size`](#reactor-bulk_create-batch_size){: #reactor-bulk_create-batch_size } | `nil \| pos_integer` |  | The number of records to include in each batch. Defaults to the `default_limit` or `max_page_size` of the action, or 100. |
-| [`load`](#reactor-bulk_create-load){: #reactor-bulk_create-load } | `atom \| list(atom)` | `[]` | A load statement to apply to records. Ignored if `return_records?` is not true. |
 | [`max_concurrency`](#reactor-bulk_create-max_concurrency){: #reactor-bulk_create-max_concurrency } | `non_neg_integer` | `0` | If set to a value greater than 0, up to that many tasks will be started to run batches asynchronously. |
 | [`notification_metadata`](#reactor-bulk_create-notification_metadata){: #reactor-bulk_create-notification_metadata } | `map \| Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` | `%{}` | Metadata to be merged into the metadata field for all notifications sent from this operation. |
 | [`notify?`](#reactor-bulk_create-notify?){: #reactor-bulk_create-notify? } | `boolean` | `false` | Whether or not to generate any notifications. This may be intensive for large bulk actions. |
@@ -519,6 +519,37 @@ Specifies the action actor
 
 Target: `Ash.Reactor.Dsl.Actor`
 
+## reactor.bulk_create.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-bulk_create-load-source){: #reactor-bulk_create-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-bulk_create-load-transform){: #reactor-bulk_create-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.bulk_create.tenant
 ```elixir
 tenant source
@@ -660,7 +691,6 @@ end
 | [`authorize_query?`](#reactor-bulk_update-authorize_query?){: #reactor-bulk_update-authorize_query? } | `boolean` | `true` | If a query is given, determines whether or not authorization is run on that query. |
 | [`batch_size`](#reactor-bulk_update-batch_size){: #reactor-bulk_update-batch_size } | `nil \| pos_integer` |  | The number of records to include in each batch. Defaults to the `default_limit` or `max_page_size` of the action, or 100. |
 | [`filter`](#reactor-bulk_update-filter){: #reactor-bulk_update-filter } | `map \| keyword` |  | A filter to apply to records. This is also applied to a stream of inputs. |
-| [`load`](#reactor-bulk_update-load){: #reactor-bulk_update-load } | `atom \| list(atom)` | `[]` | A load statement to apply to records. Ignored if `return_records?` is not true. |
 | [`lock`](#reactor-bulk_update-lock){: #reactor-bulk_update-lock } | `any` |  | A lock statement to add onto the query. |
 | [`max_concurrency`](#reactor-bulk_update-max_concurrency){: #reactor-bulk_update-max_concurrency } | `non_neg_integer` | `0` | If set to a value greater than 0, up to that many tasks will be started to run batches asynchronously. |
 | [`notification_metadata`](#reactor-bulk_update-notification_metadata){: #reactor-bulk_update-notification_metadata } | `map \| Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` | `%{}` | Metadata to be merged into the metadata field for all notifications sent from this operation. |
@@ -1010,6 +1040,7 @@ Declares a step that will call a create action on a resource.
 ### Nested DSLs
  * [actor](#reactor-create-actor)
  * [inputs](#reactor-create-inputs)
+ * [load](#reactor-create-load)
  * [tenant](#reactor-create-tenant)
  * [wait_for](#reactor-create-wait_for)
 
@@ -1128,6 +1159,37 @@ inputs(author: result(:get_user))
 
 Target: `Ash.Reactor.Dsl.Inputs`
 
+## reactor.create.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-create-load-source){: #reactor-create-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-create-load-transform){: #reactor-create-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.create.tenant
 ```elixir
 tenant source
@@ -1224,6 +1286,7 @@ Declares a step that will call a destroy action on a resource.
 ### Nested DSLs
  * [actor](#reactor-destroy-actor)
  * [inputs](#reactor-destroy-inputs)
+ * [load](#reactor-destroy-load)
  * [tenant](#reactor-destroy-tenant)
  * [wait_for](#reactor-destroy-wait_for)
 
@@ -1338,6 +1401,37 @@ inputs(author: result(:get_user))
 
 Target: `Ash.Reactor.Dsl.Inputs`
 
+## reactor.destroy.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-destroy-load-source){: #reactor-destroy-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-destroy-load-transform){: #reactor-destroy-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.destroy.tenant
 ```elixir
 tenant source
@@ -1413,6 +1507,149 @@ Target: `Ash.Reactor.Dsl.Destroy`
 
 
 
+## reactor.load
+```elixir
+load name, records, load
+```
+
+
+Declares a step that will load additional data on a resource.
+
+### Nested DSLs
+ * [actor](#reactor-load-actor)
+ * [tenant](#reactor-load-tenant)
+ * [wait_for](#reactor-load-wait_for)
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`name`](#reactor-load-name){: #reactor-load-name .spark-required} | `atom` |  | A unique name for the step. |
+| [`records`](#reactor-load-records){: #reactor-load-records .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | The records upon which to add extra loaded data |
+| [`load`](#reactor-load-load){: #reactor-load-load .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | An Ash load statement |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`domain`](#reactor-load-domain){: #reactor-load-domain } | `module` |  | The Domain to use when calling the action.  Defaults to the Domain set on the resource or in the `ash` section. |
+| [`async?`](#reactor-load-async?){: #reactor-load-async? } | `boolean` | `true` | When set to true the step will be executed asynchronously via Reactor's `TaskSupervisor`. |
+| [`authorize?`](#reactor-load-authorize?){: #reactor-load-authorize? } | `boolean \| nil` |  | Explicitly enable or disable authorization for the action. |
+| [`description`](#reactor-load-description){: #reactor-load-description } | `String.t` |  | A description for the step |
+| [`transform`](#reactor-load-transform){: #reactor-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load statement before it is passed to the load. |
+| [`lazy?`](#reactor-load-lazy?){: #reactor-load-lazy? } | `boolean` |  | If set to true, values will only be loaded if the related value isn't currently loaded. |
+| [`reuse_values?`](#reactor-load-reuse_values?){: #reactor-load-reuse_values? } | `boolean` |  | Whether calculations are allowed to reuse values that have already been loaded, or must refetch them from the data layer. |
+| [`strict?`](#reactor-load-strict?){: #reactor-load-strict? } | `boolean` |  | If set to true, only specified attributes will be loaded when passing a list of fields to fetch on a relationship, which allows for more optimized data-fetching. |
+
+
+## reactor.load.actor
+```elixir
+actor source
+```
+
+
+Specifies the action actor
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-load-actor-source){: #reactor-load-actor-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the actor. |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-load-actor-transform){: #reactor-load-actor-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the actor before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.Actor`
+
+## reactor.load.tenant
+```elixir
+tenant source
+```
+
+
+Specifies the action tenant
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-load-tenant-source){: #reactor-load-tenant-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the tenant. |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-load-tenant-transform){: #reactor-load-tenant-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the tenant before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.Tenant`
+
+## reactor.load.wait_for
+```elixir
+wait_for names
+```
+
+
+Wait for the named step to complete before allowing this one to start.
+
+Desugars to `argument :_, result(step_to_wait_for)`
+
+
+
+
+### Examples
+```
+wait_for :create_user
+```
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`names`](#reactor-load-wait_for-names){: #reactor-load-wait_for-names .spark-required} | `atom \| list(atom)` |  | The name of the step to wait for. |
+
+
+
+
+
+
+### Introspection
+
+Target: `Reactor.Dsl.WaitFor`
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.Load`
+
+
+
 ## reactor.read_one
 ```elixir
 read_one name, resource, action \\ nil
@@ -1424,6 +1661,7 @@ Declares a step that will call a read action on a resource returning a single re
 ### Nested DSLs
  * [actor](#reactor-read_one-actor)
  * [inputs](#reactor-read_one-inputs)
+ * [load](#reactor-read_one-load)
  * [tenant](#reactor-read_one-tenant)
  * [wait_for](#reactor-read_one-wait_for)
 
@@ -1533,6 +1771,37 @@ inputs(author: result(:get_user))
 
 Target: `Ash.Reactor.Dsl.Inputs`
 
+## reactor.read_one.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-read_one-load-source){: #reactor-read_one-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-read_one-load-transform){: #reactor-read_one-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.read_one.tenant
 ```elixir
 tenant source
@@ -1619,6 +1888,7 @@ Declares a step that will call a read action on a resource.
 ### Nested DSLs
  * [actor](#reactor-read-actor)
  * [inputs](#reactor-read-inputs)
+ * [load](#reactor-read-load)
  * [tenant](#reactor-read-tenant)
  * [wait_for](#reactor-read-wait_for)
 
@@ -1732,6 +2002,37 @@ inputs(author: result(:get_user))
 
 Target: `Ash.Reactor.Dsl.Inputs`
 
+## reactor.read.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-read-load-source){: #reactor-read-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-read-load-transform){: #reactor-read-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.read.tenant
 ```elixir
 tenant source
@@ -1900,6 +2201,7 @@ Declares a step that will call an update action on a resource.
 ### Nested DSLs
  * [actor](#reactor-update-actor)
  * [inputs](#reactor-update-inputs)
+ * [load](#reactor-update-load)
  * [tenant](#reactor-update-tenant)
  * [wait_for](#reactor-update-wait_for)
 
@@ -2016,6 +2318,37 @@ inputs(author: result(:get_user))
 
 Target: `Ash.Reactor.Dsl.Inputs`
 
+## reactor.update.load
+```elixir
+load source
+```
+
+
+Allows the addition of an Ash load statement to the action
+
+
+
+
+
+### Arguments
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`source`](#reactor-update-load-source){: #reactor-update-load-source .spark-required} | `Reactor.Template.Element \| Reactor.Template.Input \| Reactor.Template.Result \| Reactor.Template.Value` |  | What to use as the source of the load |
+### Options
+
+| Name | Type | Default | Docs |
+|------|------|---------|------|
+| [`transform`](#reactor-update-load-transform){: #reactor-update-load-transform } | `(any -> any) \| module \| nil` |  | An optional transformation function which can be used to modify the load before it is passed to the action. |
+
+
+
+
+
+### Introspection
+
+Target: `Ash.Reactor.Dsl.ActionLoad`
+
 ## reactor.update.tenant
 ```elixir
 tenant source

lib/ash/reactor/builders/action_load.ex

@@ -0,0 +1,6 @@
+defimpl Reactor.Argument.Build, for: Ash.Reactor.Dsl.ActionLoad do
+  @doc false
+  @impl true
+  def build(load),
+    do: {:ok, [%Reactor.Argument{name: :load, source: load.source, transform: load.transform}]}
+end

lib/ash/reactor/builders/bulk_create.ex

@@ -27,6 +27,7 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.BulkCreate do
         [initial, notification_metadata]
         |> maybe_append(bulk_create.actor)
         |> maybe_append(bulk_create.tenant)
+        |> maybe_append(bulk_create.load)
         |> Enum.concat(bulk_create.wait_for)
 
       action_options =

lib/ash/reactor/builders/bulk_update.ex

@@ -28,6 +28,7 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.BulkUpdate do
         arguments
         |> maybe_append(bulk_update.actor)
         |> maybe_append(bulk_update.tenant)
+        |> maybe_append(bulk_update.load)
         |> Enum.concat(bulk_update.wait_for)
         |> Enum.concat([initial, notification_metadata])
 
@@ -45,7 +46,6 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.BulkUpdate do
           :batch_size,
           :domain,
           :filter,
-          :load,
           :lock,
           :max_concurrency,
           :notify?,

lib/ash/reactor/builders/create.ex

@@ -29,6 +29,7 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.Create do
         arguments
         |> maybe_append(create.actor)
         |> maybe_append(create.tenant)
+        |> maybe_append(create.load)
         |> Enum.concat(create.wait_for)
         |> Enum.concat([initial])
 

lib/ash/reactor/builders/destroy.ex

@@ -16,6 +16,7 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.Destroy do
         arguments
         |> maybe_append(destroy.actor)
         |> maybe_append(destroy.tenant)
+        |> maybe_append(destroy.load)
         |> Enum.concat(destroy.wait_for)
         |> Enum.concat([%Argument{name: :initial, source: destroy.initial}])
 

lib/ash/reactor/builders/load.ex

@@ -0,0 +1,45 @@
+defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.Load do
+  @moduledoc false
+
+  alias Ash.Reactor.LoadStep
+  alias Reactor.{Argument, Builder}
+  import Ash.Reactor.BuilderUtils
+
+  @doc false
+  @impl true
+  def build(load, reactor) do
+    with {:ok, reactor} <- ensure_hooked(reactor) do
+      arguments =
+        [
+          Argument.from_template(:records, load.records),
+          Argument.from_template(:load, load.load, load.transform)
+        ]
+        |> maybe_append(load.actor)
+        |> maybe_append(load.tenant)
+        |> Enum.concat(load.wait_for)
+
+      load_options =
+        load
+        |> Map.take([:authorize?, :domain, :lazy?, :reuse_values?, :strict?])
+        |> Enum.reject(&is_nil(elem(&1, 1)))
+
+      step_options =
+        load
+        |> Map.take([:async?])
+        |> Map.put(:ref, :step_name)
+        |> Enum.to_list()
+
+      Builder.add_step(
+        reactor,
+        load.name,
+        {LoadStep, load_options},
+        arguments,
+        step_options
+      )
+    end
+  end
+
+  @doc false
+  @impl true
+  def verify(_load, _dsl_state), do: :ok
+end

lib/ash/reactor/builders/read.ex

@@ -6,11 +6,13 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.Read do
   @doc false
   @impl true
   def build(read, reactor) do
-    with {:ok, reactor, arguments} <- build_input_arguments(reactor, read) do
+    with {:ok, reactor} <- ensure_hooked(reactor),
+         {:ok, reactor, arguments} <- build_input_arguments(reactor, read) do
       arguments =
         arguments
         |> maybe_append(read.actor)
         |> maybe_append(read.tenant)
+        |> maybe_append(read.load)
         |> Enum.concat(read.wait_for)
 
       action_options =

lib/ash/reactor/builders/read_one.ex

@@ -6,11 +6,13 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.ReadOne do
   @doc false
   @impl true
   def build(read_one, reactor) do
-    with {:ok, reactor, arguments} <- build_input_arguments(reactor, read_one) do
+    with {:ok, reactor} <- ensure_hooked(reactor),
+         {:ok, reactor, arguments} <- build_input_arguments(reactor, read_one) do
       arguments =
         arguments
         |> maybe_append(read_one.actor)
         |> maybe_append(read_one.tenant)
+        |> maybe_append(read_one.load)
         |> Enum.concat(read_one.wait_for)
 
       action_options =

lib/ash/reactor/builders/update.ex

@@ -16,6 +16,7 @@ defimpl Reactor.Dsl.Build, for: Ash.Reactor.Dsl.Update do
         arguments
         |> maybe_append(update.actor)
         |> maybe_append(update.tenant)
+        |> maybe_append(update.load)
         |> Enum.concat(update.wait_for)
         |> Enum.concat([%Argument{name: :initial, source: update.initial}])
 

lib/ash/reactor/dsl/action.ex

@@ -25,7 +25,7 @@ defmodule Ash.Reactor.Dsl.Action do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
@@ -33,7 +33,7 @@ defmodule Ash.Reactor.Dsl.Action do
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
           name: atom,
           resource: module,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :action,
           undo_action: atom,
           undo: :always | :never | :outside_transaction,

lib/ash/reactor/dsl/action_load.ex

@@ -0,0 +1,43 @@
+defmodule Ash.Reactor.Dsl.ActionLoad do
+  @moduledoc """
+  Add a load statement to an action.
+  """
+
+  defstruct __identifier__: nil, source: nil, transform: nil
+
+  alias Reactor.Template
+  require Template
+
+  @type t :: %__MODULE__{
+          __identifier__: any,
+          source: Template.t(),
+          transform: nil | (any -> any) | {module, keyword} | mfa
+        }
+
+  @doc false
+  def __entity__ do
+    %Spark.Dsl.Entity{
+      name: :load,
+      describe: "Allows the addition of an Ash load statement to the action",
+      args: [:source],
+      imports: [Reactor.Dsl.Argument],
+      identifier: {:auto, :unique_integer},
+      target: __MODULE__,
+      schema: [
+        source: [
+          type: Template.type(),
+          required: true,
+          doc: "What to use as the source of the load"
+        ],
+        transform: [
+          type:
+            {:or, [{:spark_function_behaviour, Reactor.Step, {Reactor.Step.Transform, 1}}, nil]},
+          required: false,
+          default: nil,
+          doc:
+            "An optional transformation function which can be used to modify the load before it is passed to the action."
+        ]
+      ]
+    }
+  end
+end

lib/ash/reactor/dsl/action_transformer.ex

@@ -55,6 +55,11 @@ defmodule Ash.Reactor.Dsl.ActionTransformer do
     end
   end
 
+  defp transform_step(entity, dsl_state) when entity.type == :load do
+    default_domain = Transformer.get_option(dsl_state, [:ash], :default_domain)
+    {:ok, %{entity | domain: entity.domain || default_domain}, dsl_state}
+  end
+
   defp transform_step(_entity, _dsl_state), do: :ignore
 
   defp transform_nested_steps(entity, dsl_state) when is_list(entity.steps) do

lib/ash/reactor/dsl/actor.ex

@@ -9,7 +9,7 @@ defmodule Ash.Reactor.Dsl.Actor do
 
   @type t :: %__MODULE__{
           __identifier__: any,
-          source: Template.Input.t() | Template.Result.t() | Template.Value.t(),
+          source: Template.t(),
           transform: nil | (any -> any) | {module, keyword} | mfa
         }
 

lib/ash/reactor/dsl/bulk_create.ex

@@ -16,7 +16,7 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
             description: nil,
             domain: nil,
             initial: nil,
-            load: [],
+            load: nil,
             max_concurrency: 0,
             name: nil,
             notification_metadata: %{},
@@ -48,7 +48,7 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           assume_casted?: boolean,
           async?: boolean,
           authorize_changeset_with: :filter | :error,
@@ -58,7 +58,7 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
           description: String.t() | nil,
           domain: Ash.Domain.t(),
           initial: Reactor.Template.t(),
-          load: [atom],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           max_concurrency: non_neg_integer(),
           name: atom,
           notification_metadata: map,
@@ -74,7 +74,7 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
           sorted?: boolean,
           stop_on_error?: boolean,
           success_state: :success | :partial_success,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           timeout: nil | timeout,
           transaction: :all | :batch | false,
           type: :bulk_create,
@@ -119,10 +119,11 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
       imports: [Reactor.Dsl.Argument],
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema:
         [
@@ -159,13 +160,6 @@ defmodule Ash.Reactor.Dsl.BulkCreate do
             doc:
               "A collection of inputs to pass to the create action. Must implement the `Enumerable` protocol."
           ],
-          load: [
-            type: {:wrap_list, :atom},
-            doc:
-              "A load statement to apply to records. Ignored if `return_records?` is not true.",
-            required: false,
-            default: []
-          ],
           max_concurrency: [
             type: :non_neg_integer,
             doc:

lib/ash/reactor/dsl/bulk_update.ex

@@ -21,7 +21,7 @@ defmodule Ash.Reactor.Dsl.BulkUpdate do
             filter: %{},
             initial: nil,
             inputs: [],
-            load: [],
+            load: nil,
             lock: nil,
             max_concurrency: 0,
             name: nil,
@@ -56,7 +56,7 @@ defmodule Ash.Reactor.Dsl.BulkUpdate do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           allow_stream_with: :keyset | :offset | :full_read,
           assume_casted?: boolean,
           async?: boolean,
@@ -73,7 +73,7 @@ defmodule Ash.Reactor.Dsl.BulkUpdate do
             | Keyword.t(Keyword.t(String.t() | number | boolean)),
           initial: Reactor.Template.t(),
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
-          load: [atom],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           lock: nil | Ash.DataLayer.lock_type(),
           max_concurrency: non_neg_integer(),
           name: atom,
@@ -95,7 +95,7 @@ defmodule Ash.Reactor.Dsl.BulkUpdate do
           stream_batch_size: nil | pos_integer(),
           stream_with: nil | :keyset | :offset | :full_read,
           success_state: :success | :partial_success,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           timeout: nil | timeout,
           transaction: :all | :batch | false,
           type: :bulk_create,
@@ -204,13 +204,6 @@ defmodule Ash.Reactor.Dsl.BulkUpdate do
             doc:
               "A collection of inputs to pass to the create action. Must implement the `Enumerable` protocol."
           ],
-          load: [
-            type: {:wrap_list, :atom},
-            doc:
-              "A load statement to apply to records. Ignored if `return_records?` is not true.",
-            required: false,
-            default: []
-          ],
           lock: [
             type: :any,
             doc: "A lock statement to add onto the query.",

lib/ash/reactor/dsl/create.ex

@@ -13,6 +13,7 @@ defmodule Ash.Reactor.Dsl.Create do
             domain: nil,
             initial: nil,
             inputs: [],
+            load: nil,
             name: nil,
             resource: nil,
             tenant: [],
@@ -28,16 +29,17 @@ defmodule Ash.Reactor.Dsl.Create do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
           domain: Ash.Domain.t(),
           initial: nil | Ash.Resource.t() | Reactor.Template.t(),
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           name: atom,
           resource: module,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :create,
           undo_action: atom,
           undo: :always | :never | :outside_transaction,
@@ -75,10 +77,11 @@ defmodule Ash.Reactor.Dsl.Create do
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
         inputs: [Ash.Reactor.Dsl.Inputs.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema:
         [

lib/ash/reactor/dsl/destroy.ex

@@ -13,6 +13,7 @@ defmodule Ash.Reactor.Dsl.Destroy do
             domain: nil,
             initial: nil,
             inputs: [],
+            load: nil,
             name: nil,
             resource: nil,
             return_destroyed?: false,
@@ -27,17 +28,18 @@ defmodule Ash.Reactor.Dsl.Destroy do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
           domain: Ash.Domain.t(),
           initial: Reactor.Template.t(),
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           name: atom,
           resource: module,
           return_destroyed?: boolean,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :destroy,
           undo_action: atom,
           undo: :always | :never | :outside_transaction,
@@ -70,10 +72,11 @@ defmodule Ash.Reactor.Dsl.Destroy do
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
         inputs: [Ash.Reactor.Dsl.Inputs.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema:
         [

lib/ash/reactor/dsl/load.ex

@@ -0,0 +1,104 @@
+defmodule Ash.Reactor.Dsl.Load do
+  @moduledoc """
+  The `load` step entity for the `Ash.Reactor` reactor extension.
+  """
+
+  defstruct __identifier__: nil,
+            action_step?: false,
+            action: nil,
+            actor: nil,
+            async?: true,
+            authorize?: nil,
+            description: nil,
+            domain: nil,
+            lazy?: nil,
+            load: nil,
+            name: nil,
+            records: nil,
+            reuse_values?: nil,
+            strict?: nil,
+            tenant: nil,
+            transform: nil,
+            type: :load,
+            wait_for: []
+
+  @type t :: %__MODULE__{
+          __identifier__: any,
+          action: nil | atom,
+          action_step?: false,
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
+          async?: boolean,
+          authorize?: nil | boolean,
+          description: nil | String.t(),
+          domain: nil | Ash.Domain.t(),
+          lazy?: nil | boolean,
+          load: Reactor.Template.t(),
+          name: atom,
+          records: Reactor.Template.t(),
+          reuse_values?: nil | boolean,
+          strict?: nil | boolean,
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
+          transform: nil | (any -> any) | {module, keyword} | mfa,
+          type: :load,
+          wait_for: [Reactor.Dsl.WaitFor.t()]
+        }
+
+  @doc false
+  def __entity__,
+    do: %Spark.Dsl.Entity{
+      name: :load,
+      describe: "Declares a step that will load additional data on a resource.",
+      target: __MODULE__,
+      args: [:name, :records, :load],
+      imports: [Reactor.Dsl.Argument],
+      entities: [
+        actor: [Ash.Reactor.Dsl.Actor.__entity__()],
+        tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
+        wait_for: [Reactor.Dsl.WaitFor.__entity__()]
+      ],
+      singleton_entity_keys: [:actor, :tenant],
+      recursive_as: :steps,
+      schema:
+        false
+        |> Ash.Reactor.Dsl.Action.__shared_action_option_schema__()
+        |> Keyword.take([:domain, :async?, :authorize?, :description, :name])
+        |> Keyword.merge(
+          records: [
+            type: Reactor.Template.type(),
+            required: true,
+            doc: "The records upon which to add extra loaded data"
+          ],
+          transform: [
+            type:
+              {:or, [{:spark_function_behaviour, Reactor.Step, {Reactor.Step.Transform, 1}}, nil]},
+            required: false,
+            default: nil,
+            doc:
+              "An optional transformation function which can be used to modify the load statement before it is passed to the load."
+          ],
+          load: [
+            type: Reactor.Template.type(),
+            required: true,
+            doc: "An Ash load statement"
+          ],
+          lazy?: [
+            type: :boolean,
+            required: false,
+            doc:
+              "If set to true, values will only be loaded if the related value isn't currently loaded."
+          ],
+          reuse_values?: [
+            type: :boolean,
+            required: false,
+            doc:
+              "Whether calculations are allowed to reuse values that have already been loaded, or must refetch them from the data layer."
+          ],
+          strict?: [
+            type: :boolean,
+            required: false,
+            doc:
+              "If set to true, only specified attributes will be loaded when passing a list of fields to fetch on a relationship, which allows for more optimized data-fetching."
+          ]
+        )
+    }
+end

lib/ash/reactor/dsl/read.ex

@@ -6,15 +6,16 @@ defmodule Ash.Reactor.Dsl.Read do
   defstruct __identifier__: nil,
             action_step?: true,
             action: nil,
-            actor: [],
+            actor: nil,
             domain: nil,
             async?: true,
             authorize?: nil,
             description: nil,
             inputs: [],
+            load: nil,
             name: nil,
             resource: nil,
-            tenant: [],
+            tenant: nil,
             transform: nil,
             type: :read,
             wait_for: []
@@ -23,15 +24,16 @@ defmodule Ash.Reactor.Dsl.Read do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           domain: Ash.Domain.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           name: atom,
           resource: module,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :create,
           wait_for: [Reactor.Dsl.WaitFor.t()]
         }
@@ -59,10 +61,11 @@ defmodule Ash.Reactor.Dsl.Read do
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
         inputs: [Ash.Reactor.Dsl.Inputs.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema: Ash.Reactor.Dsl.Action.__shared_action_option_schema__(false)
     }

lib/ash/reactor/dsl/read_one.ex

@@ -13,6 +13,7 @@ defmodule Ash.Reactor.Dsl.ReadOne do
             domain: nil,
             fail_on_not_found?: nil,
             inputs: [],
+            load: nil,
             name: nil,
             resource: nil,
             tenant: [],
@@ -24,16 +25,17 @@ defmodule Ash.Reactor.Dsl.ReadOne do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
           domain: Ash.Domain.t(),
           fail_on_not_found?: boolean,
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           name: atom,
           resource: module,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :create,
           wait_for: [Reactor.Dsl.WaitFor.t()]
         }
@@ -59,10 +61,11 @@ defmodule Ash.Reactor.Dsl.ReadOne do
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
         inputs: [Ash.Reactor.Dsl.Inputs.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema:
         [

lib/ash/reactor/dsl/tenant.ex

@@ -1,6 +1,6 @@
 defmodule Ash.Reactor.Dsl.Tenant do
   @moduledoc """
-  Specify the actor used to execute an action.
+  Specify the tenant used to execute an action.
   """
 
   defstruct __identifier__: nil, source: nil, transform: nil
@@ -9,7 +9,7 @@ defmodule Ash.Reactor.Dsl.Tenant do
 
   @type t :: %__MODULE__{
           __identifier__: any,
-          source: Template.Input.t() | Template.Result.t() | Template.Value.t(),
+          source: Template.t(),
           transform: nil | (any -> any) | {module, keyword} | mfa
         }
 

lib/ash/reactor/dsl/update.ex

@@ -13,6 +13,7 @@ defmodule Ash.Reactor.Dsl.Update do
             domain: nil,
             initial: nil,
             inputs: [],
+            load: nil,
             name: nil,
             resource: nil,
             tenant: [],
@@ -26,16 +27,17 @@ defmodule Ash.Reactor.Dsl.Update do
           __identifier__: any,
           action_step?: true,
           action: atom,
-          actor: [Ash.Reactor.Dsl.Actor.t()],
+          actor: nil | Ash.Reactor.Dsl.Actor.t(),
           async?: boolean,
           authorize?: boolean | nil,
           description: String.t() | nil,
           domain: Ash.Domain.t(),
           initial: Reactor.Template.t(),
           inputs: [Ash.Reactor.Dsl.Inputs.t()],
+          load: nil | Ash.Reactor.Dsl.ActionLoad.t(),
           name: atom,
           resource: module,
-          tenant: [Ash.Reactor.Dsl.Tenant.t()],
+          tenant: nil | Ash.Reactor.Dsl.Tenant.t(),
           type: :update,
           undo_action: atom,
           undo: :always | :never | :outside_transaction,
@@ -71,10 +73,11 @@ defmodule Ash.Reactor.Dsl.Update do
       entities: [
         actor: [Ash.Reactor.Dsl.Actor.__entity__()],
         inputs: [Ash.Reactor.Dsl.Inputs.__entity__()],
+        load: [Ash.Reactor.Dsl.ActionLoad.__entity__()],
         tenant: [Ash.Reactor.Dsl.Tenant.__entity__()],
         wait_for: [Reactor.Dsl.WaitFor.__entity__()]
       ],
-      singleton_entity_keys: [:actor, :tenant],
+      singleton_entity_keys: [:actor, :tenant, :load],
       recursive_as: :steps,
       schema:
         [

lib/ash/reactor/reactor.ex

@@ -25,6 +25,7 @@ defmodule Ash.Reactor do
           | Ash.Reactor.Dsl.BulkUpdate.t()
           | Ash.Reactor.Dsl.Create.t()
           | Ash.Reactor.Dsl.Destroy.t()
+          | Ash.Reactor.Dsl.Load.t()
           | Ash.Reactor.Dsl.Read.t()
           | Ash.Reactor.Dsl.ReadOne.t()
           | Ash.Reactor.Dsl.Update.t()
@@ -41,6 +42,7 @@ defmodule Ash.Reactor do
         Ash.Reactor.Dsl.Change,
         Ash.Reactor.Dsl.Create,
         Ash.Reactor.Dsl.Destroy,
+        Ash.Reactor.Dsl.Load,
         Ash.Reactor.Dsl.ReadOne,
         Ash.Reactor.Dsl.Read,
         Ash.Reactor.Dsl.Transaction,

lib/ash/reactor/steps/action_step.ex

@@ -21,12 +21,12 @@ defmodule Ash.Reactor.ActionStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      []
+      [domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
 
     options[:resource]
     |> ActionInput.for_action(options[:action], arguments[:input], action_input_options)
-    |> options[:domain].run_action(action_options)
+    |> Ash.run_action(action_options)
   end
 
   @doc false
@@ -45,6 +45,7 @@ defmodule Ash.Reactor.ActionStep do
     action_options =
       []
       |> maybe_set_kw(:authorize?, options[:authorize?])
+      |> maybe_set_kw(:domain, options[:domain])
 
     inputs =
       arguments[:input]
@@ -52,7 +53,7 @@ defmodule Ash.Reactor.ActionStep do
 
     options[:resource]
     |> ActionInput.for_action(options[:action], inputs, action_input_options)
-    |> options[:domain].run_action(action_options)
+    |> Ash.run_action(action_options)
   end
 
   @doc false

lib/ash/reactor/steps/bulk_create_step.ex

@@ -19,7 +19,6 @@ defmodule Ash.Reactor.BulkCreateStep do
         :authorize?,
         :batch_size,
         :domain,
-        :load,
         :max_concurrency,
         :notify?,
         :read_action,
@@ -45,6 +44,7 @@ defmodule Ash.Reactor.BulkCreateStep do
       |> maybe_set_kw(:actor, arguments[:actor])
       |> maybe_set_kw(:tenant, arguments[:tenant])
       |> maybe_set_kw(:notification_metadata, arguments[:notification_metadata])
+      |> maybe_set_kw(:load, arguments[:load])
 
     success_states =
       options[:success_state]

lib/ash/reactor/steps/bulk_update_step.ex

@@ -23,7 +23,6 @@ defmodule Ash.Reactor.BulkUpdateStep do
         :batch_size,
         :domain,
         :filter,
-        :load,
         :lock,
         :max_concurrency,
         :notify?,
@@ -51,6 +50,7 @@ defmodule Ash.Reactor.BulkUpdateStep do
       |> maybe_set_kw(:actor, context[:actor])
       |> maybe_set_kw(:tenant, context[:tenant])
       |> maybe_set_kw(:tracer, context[:tracer])
+      |> maybe_set_kw(:load, arguments[:load])
 
     success_states =
       options[:success_state]

lib/ash/reactor/steps/create_step.ex

@@ -25,6 +25,7 @@ defmodule Ash.Reactor.CreateStep do
     action_options =
       [return_notifications?: true, domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
+      |> maybe_set_kw(:load, arguments[:load])
 
     changeset =
       case arguments.initial do

lib/ash/reactor/steps/destroy_step.ex

@@ -24,13 +24,14 @@ defmodule Ash.Reactor.DestroyStep do
       |> maybe_set_kw(:return_destroyed?, return_destroyed?)
 
     action_options =
-      [return_notifications?: true]
+      [return_notifications?: true, domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
       |> maybe_set_kw(:return_destroyed?, return_destroyed?)
+      |> maybe_set_kw(:load, arguments[:load])
 
     arguments[:initial]
     |> Changeset.for_destroy(options[:action], arguments[:input], changeset_options)
-    |> options[:domain].destroy(action_options)
+    |> Ash.destroy(action_options)
     |> case do
       :ok ->
         {:ok, :ok}
@@ -65,12 +66,12 @@ defmodule Ash.Reactor.DestroyStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      [return_notifications?: false]
+      [return_notifications?: false, domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
 
     options[:resource]
     |> Changeset.for_create(options[:undo_action], %{record: record}, changeset_options)
-    |> options[:domain].create(action_options)
+    |> Ash.create(action_options)
     |> case do
       {:ok, _record} -> :ok
       {:ok, _record, _notifications} -> :ok

lib/ash/reactor/steps/load_step.ex

@@ -0,0 +1,21 @@
+defmodule Ash.Reactor.LoadStep do
+  @moduledoc """
+  The Reactor step which is used to execute load steps.
+  """
+  use Reactor.Step
+  import Ash.Reactor.StepUtils
+
+  def run(arguments, context, options) do
+    load_options =
+      options
+      |> maybe_set_kw(:authorize?, context[:authorize?])
+      |> maybe_set_kw(:actor, context[:actor])
+      |> maybe_set_kw(:tenant, context[:tenant])
+      |> maybe_set_kw(:tracer, context[:tracer])
+      |> maybe_set_kw(:actor, arguments[:actor])
+      |> maybe_set_kw(:tenant, arguments[:tenant])
+
+    arguments.records
+    |> Ash.load(arguments.load, load_options)
+  end
+end

lib/ash/reactor/steps/read_one_step.ex

@@ -19,28 +19,13 @@ defmodule Ash.Reactor.ReadOneStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      []
+      [domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
+      |> maybe_set_kw(:load, arguments[:load])
+      |> maybe_set_kw(:not_found_error?, options[:fail_on_not_found?])
 
     options[:resource]
     |> Query.for_read(options[:action], arguments[:input], query_options)
-    |> options[:domain].read_one(action_options)
+    |> Ash.read_one(action_options)
-    |> case do
-      {:ok, nil} ->
-        if options[:fail_on_not_found?] do
-          raise Ash.Error.Query.NotFound, resource: options[:resource]
-        else
-          {:ok, nil}
-        end
-
-      {:ok, record} ->
-        {:ok, record}
-
-      {:ok, records, _} ->
-        {:ok, records}
-
-      {:error, reason} ->
-        {:error, reason}
-    end
   end
 end

lib/ash/reactor/steps/read_step.ex

@@ -19,16 +19,12 @@ defmodule Ash.Reactor.ReadStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      []
+      [domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
+      |> maybe_set_kw(:load, arguments[:load])
 
     options[:resource]
     |> Query.for_read(options[:action], arguments[:input], query_options)
-    |> options[:domain].read(action_options)
+    |> Ash.read(action_options)
-    |> case do
-      {:ok, records} -> {:ok, records}
-      {:ok, records, _} -> {:ok, records}
-      {:error, reason} -> {:error, reason}
-    end
   end
 end

lib/ash/reactor/steps/update_step.ex

@@ -21,15 +21,16 @@ defmodule Ash.Reactor.UpdateStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      [return_notifications?: true]
+      [return_notifications?: true, domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
+      |> maybe_set_kw(:load, arguments[:load])
 
     changeset =
       arguments[:initial]
       |> Changeset.for_update(options[:action], arguments[:input], changeset_options)
 
     changeset
-    |> options[:domain].update(action_options)
+    |> Ash.update(action_options)
     |> case do
       {:ok, record} ->
         {:ok, store_changeset_in_metadata(context.current_step.name, record, changeset)}
@@ -57,7 +58,7 @@ defmodule Ash.Reactor.UpdateStep do
       |> maybe_set_kw(:tenant, arguments[:tenant])
 
     action_options =
-      [return_notifications?: false]
+      [return_notifications?: false, domain: options[:domain]]
       |> maybe_set_kw(:authorize?, options[:authorize?])
 
     attributes =
@@ -65,7 +66,7 @@ defmodule Ash.Reactor.UpdateStep do
 
     record
     |> Changeset.for_update(options[:undo_action], attributes, changeset_options)
-    |> options[:domain].update(action_options)
+    |> Ash.update(action_options)
     |> case do
       {:ok, _record} -> :ok
       {:ok, _record, _notifications} -> :ok

test/reactor/load_test.exs

@@ -0,0 +1,86 @@
+defmodule Ash.Test.ReactorLoadTest do
+  @moduledoc false
+  use ExUnit.Case, async: true
+
+  alias __MODULE__, as: Self
+  alias Ash.Test.Domain
+
+  defmodule Post do
+    @moduledoc false
+    use Ash.Resource, data_layer: Ash.DataLayer.Ets, domain: Domain
+
+    ets do
+      private? true
+    end
+
+    attributes do
+      uuid_primary_key :id
+      attribute :title, :string, allow_nil?: false, public?: true
+    end
+
+    relationships do
+      has_many :comments, Self.Comment, public?: true
+    end
+
+    actions do
+      default_accept :*
+      defaults [:read, create: :*]
+    end
+
+    code_interface do
+      define :create
+    end
+  end
+
+  defmodule Comment do
+    @moduledoc false
+    use Ash.Resource, data_layer: Ash.DataLayer.Ets, domain: Domain
+
+    ets do
+      private? true
+    end
+
+    attributes do
+      uuid_primary_key :id
+      attribute :comment, :string, allow_nil?: false, public?: true
+    end
+
+    relationships do
+      belongs_to :post, Self.Post, public?: true
+    end
+
+    actions do
+      default_accept :*
+      defaults [:read, create: :*]
+    end
+
+    code_interface do
+      define :create
+    end
+  end
+
+  defmodule SimpleLoadReactor do
+    @moduledoc false
+    use Reactor, extensions: [Ash.Reactor]
+
+    ash do
+      default_domain(Domain)
+    end
+
+    input :post
+
+    load(:post_with_comments, input(:post), value(comments: :comment))
+  end
+
+  test "it performs loading" do
+    post = Post.create!(%{title: "Marty"})
+    comments = ["This is heavy", "You made a time machine... out of a Delorean?"]
+
+    for comment <- comments do
+      Comment.create!(%{post_id: post.id, comment: comment})
+    end
+
+    assert {:ok, post} = Reactor.run(SimpleLoadReactor, %{post: post}, %{}, async?: false)
+    assert Enum.sort(Enum.map(post.comments, & &1.comment)) == comments
+  end
+end

test/reactor/read_one_test.exs

@@ -2,6 +2,7 @@ defmodule Ash.Test.ReactorReadOneTest do
   @moduledoc false
   use ExUnit.Case, async: true
 
+  alias __MODULE__, as: Self
   alias Ash.Test.Domain
 
   defmodule Post do
@@ -17,6 +18,37 @@ defmodule Ash.Test.ReactorReadOneTest do
       attribute :title, :string, allow_nil?: false, public?: true
     end
 
+    relationships do
+      has_many :comments, Self.Comment, public?: true
+    end
+
+    actions do
+      default_accept :*
+      defaults [:read, create: :*]
+    end
+
+    code_interface do
+      define :create
+    end
+  end
+
+  defmodule Comment do
+    @moduledoc false
+    use Ash.Resource, data_layer: Ash.DataLayer.Ets, domain: Domain
+
+    ets do
+      private? true
+    end
+
+    attributes do
+      uuid_primary_key :id
+      attribute :comment, :string, allow_nil?: false, public?: true
+    end
+
+    relationships do
+      belongs_to :post, Self.Post, public?: true
+    end
+
     actions do
       default_accept :*
       defaults [:read, create: :*]
@@ -74,7 +106,9 @@ defmodule Ash.Test.ReactorReadOneTest do
     NotFoundReactor
     |> Reactor.run(%{}, %{}, async?: false)
     |> Ash.Test.assert_has_error(fn
-      %Reactor.Error.Invalid.RunStepError{error: %Ash.Error.Query.NotFound{}} ->
+      %Reactor.Error.Invalid.RunStepError{
+        error: %Ash.Error.Invalid{errors: [%Ash.Error.Query.NotFound{}]}
+      } ->
         true
 
       _ ->
@@ -88,4 +122,29 @@ defmodule Ash.Test.ReactorReadOneTest do
     assert {:ok, actual} = Reactor.run(SimpleReadOneReactor, %{}, %{}, async?: false)
     assert expected.id == actual.id
   end
+
+  test "it can load related data when asked" do
+    defmodule LoadRelatedReactor do
+      @moduledoc false
+      use Reactor, extensions: [Ash.Reactor]
+
+      ash do
+        default_domain(Domain)
+      end
+
+      read_one :read_one_post, Ash.Test.ReactorReadOneTest.Post, :read do
+        load value(comments: [:comment])
+      end
+    end
+
+    post = Post.create!(%{title: "Marty"})
+    comments = ["This is heavy", "You made a time machine... out of a Delorean?"]
+
+    for comment <- comments do
+      Comment.create!(%{post_id: post.id, comment: comment})
+    end
+
+    assert {:ok, post} = Reactor.run(LoadRelatedReactor, %{}, %{}, async?: false)
+    assert Enum.sort(Enum.map(post.comments, & &1.comment)) == comments
+  end
 end