Preserve pipe_through declaration order in actions

Author: nallwhy

documentation/dsls/DSL-Ash.Resource.md

@@ -13,7 +13,6 @@ defmodule Ash.Resource.Actions.Action do
     constraints: [],
     touches_resources: [],
     skip_unknown_inputs: [],
-    pipe_through: [],
     arguments: [],
     preparations: [],
     allow_nil?: false,
@@ -31,7 +30,6 @@ defmodule Ash.Resource.Actions.Action do
           description: String.t() | nil,
           arguments: [Ash.Resource.Actions.Argument.t()],
           skip_unknown_inputs: list(atom() | String.t()),
-          pipe_through: list(Ash.Resource.Actions.PipeThrough.t()),
           allow_nil?: boolean,
           touches_resources: [Ash.Resource.t()],
           constraints: Keyword.t(),

lib/ash/resource/actions/action/action.ex

@@ -20,7 +20,6 @@ defmodule Ash.Resource.Actions.Create do
     delay_global_validations?: false,
     skip_global_validations?: false,
     skip_unknown_inputs: [],
-    pipe_through: [],
     upsert?: false,
     upsert_identity: nil,
     upsert_fields: nil,
@@ -47,7 +46,6 @@ defmodule Ash.Resource.Actions.Create do
           multitenancy: atom,
           upsert?: boolean,
           skip_unknown_inputs: list(atom | String.t()),
-          pipe_through: list(Ash.Resource.Actions.PipeThrough.t()),
           return_skipped_upsert?: boolean(),
           notifiers: [module()],
           delay_global_validations?: boolean,

lib/ash/resource/actions/create.ex

@@ -15,7 +15,6 @@ defmodule Ash.Resource.Actions.Destroy do
     manual: nil,
     require_atomic?: Application.compile_env(:ash, :require_atomic_by_default?, true),
     skip_unknown_inputs: [],
-    pipe_through: [],
     atomic_upgrade?: true,
     atomic_upgrade_with: nil,
     action_select: nil,
@@ -46,7 +45,6 @@ defmodule Ash.Resource.Actions.Destroy do
           arguments: list(Ash.Resource.Actions.Argument.t()),
           atomic_upgrade?: boolean(),
           skip_unknown_inputs: list(atom() | String.t()),
-          pipe_through: list(Ash.Resource.Actions.PipeThrough.t()),
           atomic_upgrade_with: atom() | nil,
           require_atomic?: boolean,
           accept: nil | list(atom),

lib/ash/resource/actions/destroy.ex

@@ -14,7 +14,6 @@ defmodule Ash.Resource.Actions.Read do
             manual: nil,
             metadata: [],
             skip_unknown_inputs: [],
-            pipe_through: [],
             skip_global_validations?: false,
             modify_query: nil,
             multitenancy: nil,
@@ -40,7 +39,6 @@ defmodule Ash.Resource.Actions.Read do
           manual: atom | {atom, Keyword.t()} | nil,
           metadata: [Ash.Resource.Actions.Metadata.t()],
           skip_unknown_inputs: list(atom | String.t()),
-          pipe_through: list(Ash.Resource.Actions.PipeThrough.t()),
           skip_global_validations?: boolean,
           modify_query: nil | mfa,
           multitenancy: atom,

lib/ash/resource/actions/read.ex

@@ -15,7 +15,6 @@ defmodule Ash.Resource.Actions.Update do
     require_attributes: [],
     allow_nil_input: [],
     skip_unknown_inputs: [],
-    pipe_through: [],
     manual: nil,
     manual?: false,
     require_atomic?: Application.compile_env(:ash, :require_atomic_by_default?, true),
@@ -43,7 +42,6 @@ defmodule Ash.Resource.Actions.Update do
           manual: module | nil,
           multitenancy: atom,
           skip_unknown_inputs: list(atom | String.t()),
-          pipe_through: list(Ash.Resource.Actions.PipeThrough.t()),
           atomic_upgrade?: boolean(),
           action_select: list(atom) | nil,
           atomic_upgrade_with: atom() | nil,

lib/ash/resource/actions/update.ex

@@ -560,9 +560,7 @@ defmodule Ash.Resource.Dsl do
     entities: [
       changes: [
         @action_change,
-        @action_validate
-      ],
-      pipe_through: [
+        @action_validate,
         @pipe_through
       ],
       arguments: [
@@ -622,9 +620,6 @@ defmodule Ash.Resource.Dsl do
       arguments: [
         @action_argument
       ],
-      pipe_through: [
-        @pipe_through
-      ],
       preparations: [
         @prepare,
         %{
@@ -633,7 +628,8 @@ defmodule Ash.Resource.Dsl do
               @validate.schema
               |> Keyword.delete(:always_atomic?)
               |> Keyword.delete(:on)
-        }
+        },
+        @pipe_through
       ]
     ],
     args: [:name, {:optional, :returns}]
@@ -674,9 +670,6 @@ defmodule Ash.Resource.Dsl do
       arguments: [
         @action_argument
       ],
-      pipe_through: [
-        @pipe_through
-      ],
       preparations: [
         @prepare,
         %{
@@ -685,7 +678,8 @@ defmodule Ash.Resource.Dsl do
               @validate.schema
               |> Keyword.delete(:always_atomic?)
               |> Keyword.delete(:on)
-        }
+        },
+        @pipe_through
       ],
       pagination: [
         @pagination
@@ -716,9 +710,7 @@ defmodule Ash.Resource.Dsl do
     entities: [
       changes: [
         @action_change,
-        @action_validate
-      ],
-      pipe_through: [
+        @action_validate,
         @pipe_through
       ],
       metadata: [
@@ -764,9 +756,7 @@ defmodule Ash.Resource.Dsl do
     entities: [
       changes: [
         @action_change,
-        @action_validate
-      ],
-      pipe_through: [
+        @action_validate,
         @pipe_through
       ],
       metadata: [

lib/ash/resource/dsl.ex

@@ -3,9 +3,10 @@
 # SPDX-License-Identifier: MIT
 
 defmodule Ash.Resource.Transformers.ResolvePipelines do
-  @moduledoc "Resolves `pipe_through` on actions by injecting pipeline changes/preparations."
+  @moduledoc "Resolves `pipe_through` on actions by replacing PipeThrough entities with pipeline contents in-place."
   use Spark.Dsl.Transformer
 
+  alias Ash.Resource.Actions.PipeThrough
   alias Spark.Dsl.Transformer
   alias Spark.Error.DslError
 
@@ -23,90 +24,80 @@ defmodule Ash.Resource.Transformers.ResolvePipelines do
 
     dsl_state
     |> Transformer.get_entities([:actions])
-    |> Enum.filter(&(&1.pipe_through != []))
     |> Enum.reduce_while({:ok, dsl_state}, fn action, {:ok, dsl_state} ->
-      case resolve_pipe_through(action, pipelines, dsl_state) do
-        {:ok, updated_action} ->
-          new_state =
-            Transformer.replace_entity(
-              dsl_state,
-              [:actions],
-              updated_action,
-              &(&1.name == action.name && &1.type == action.type)
-            )
-
-          {:cont, {:ok, new_state}}
-
-        {:error, error} ->
-          {:halt, {:error, error}}
+      field = entity_field(action.type)
+      entities = Map.get(action, field, [])
+
+      if Enum.any?(entities, &match?(%PipeThrough{}, &1)) do
+        updated_action = resolve_action(action, field, entities, pipelines, dsl_state)
+
+        new_state =
+          Transformer.replace_entity(
+            dsl_state,
+            [:actions],
+            updated_action,
+            &(&1.name == action.name && &1.type == action.type)
+          )
+
+        {:cont, {:ok, new_state}}
+      else
+        {:cont, {:ok, dsl_state}}
       end
     end)
   end
 
-  defp resolve_pipe_through(action, pipelines, dsl_state) do
-    # Flatten all pipe_through declarations into {name, where_conditions} pairs
-    pairs =
-      Enum.flat_map(action.pipe_through, fn %{names: names, where: where} ->
-        Enum.map(names, &{&1, where})
+  defp resolve_action(action, field, entities, pipelines, dsl_state) do
+    {expanded, arguments, accept} =
+      Enum.reduce(entities, {[], [], []}, fn
+        %PipeThrough{names: names, where: where}, {expanded, arguments, accept} ->
+          {new_entities, new_arguments, new_accept} =
+            expand_pipe_through(names, where, action, pipelines, dsl_state)
+
+          {expanded ++ new_entities, arguments ++ new_arguments, accept ++ new_accept}
+
+        entity, {expanded, arguments, accept} ->
+          {expanded ++ [entity], arguments, accept}
       end)
 
-    pairs
-    |> Enum.reduce_while({:ok, {[], [], []}}, fn {name, where},
-                                                 {:ok, {entities, arguments, accept}} ->
+    action
+    |> Map.put(field, expanded)
+    |> merge_arguments(arguments, dsl_state)
+    |> merge_accept(merge_pipeline_accepts(accept))
+  end
+
+  defp expand_pipe_through(names, where, action, pipelines, dsl_state) do
+    Enum.reduce(names, {[], [], []}, fn name, {entities, arguments, accept} ->
       case Map.fetch(pipelines, name) do
         {:ok, pipeline} ->
-          new_entities = collect_entities(action.type, pipeline, where, dsl_state)
+          pipeline_entities = collect_entities(action.type, pipeline, where, dsl_state)
 
-          {:cont,
-           {:ok,
-            {[new_entities | entities], [pipeline.arguments | arguments],
-             [pipeline.accept | accept]}}}
+          {entities ++ pipeline_entities, arguments ++ pipeline.arguments,
+           accept ++ List.wrap(if pipeline.accept == :*, do: [:*], else: pipeline.accept)}
 
         :error ->
-          {:halt,
-           {:error,
-            DslError.exception(
-              module: Transformer.get_persisted(dsl_state, :module),
-              path: [:actions, action.type],
-              message:
-                "Action `#{action.name}` references pipeline `#{name}` via `pipe_through`, but no pipeline named `#{name}` exists."
-            )}}
+          raise DslError,
+            module: Transformer.get_persisted(dsl_state, :module),
+            path: [:actions, action.type],
+            message:
+              "Action `#{action.name}` references pipeline `#{name}` via `pipe_through`, but no pipeline named `#{name}` exists."
       end
     end)
-    |> case do
-      {:ok, {entities, arguments, accept}} ->
-        merged_accept = merge_pipeline_accepts(accept)
-
-        {:ok,
-         apply_to_action(
-           action,
-           entities |> Enum.reverse() |> Enum.concat(),
-           arguments |> Enum.reverse() |> Enum.concat(),
-           merged_accept,
-           dsl_state
-         )}
-
-      {:error, error} ->
-        {:error, error}
-    end
   end
 
-  defp apply_to_action(action, entities, arguments, accept, dsl_state) do
-    action
-    |> prepend_entities(entities)
-    |> merge_arguments(arguments, dsl_state)
-    |> merge_accept(accept)
-  end
+  defp collect_entities(type, pipeline, where, dsl_state) do
+    subject = subject_for_type(type)
+    entities = source_entities(pipeline, type)
+
+    Enum.each(entities, &validate_supports!(&1, subject, pipeline, dsl_state))
 
-  defp prepend_entities(action, entities) do
-    field = entity_field(action.type)
-    Map.update!(action, field, &(entities ++ &1))
+    Enum.map(entities, fn entity ->
+      Map.update!(entity, :where, &(where ++ &1))
+    end)
   end
 
   defp merge_arguments(action, [], _dsl_state), do: action
 
   defp merge_arguments(action, pipeline_arguments, dsl_state) do
-    # Dedup pipeline arguments (multiple pipelines may define the same arg)
     {deduped_pipeline_args, pipeline_conflicts} =
       Enum.reduce(pipeline_arguments, {%{}, []}, fn arg, {seen, conflicts} ->
         case Map.fetch(seen, arg.name) do
@@ -115,7 +106,6 @@ defmodule Ash.Resource.Transformers.ResolvePipelines do
 
           {:ok, existing} ->
             if existing.type == arg.type do
-              # same type from different pipelines — keep first
               {seen, conflicts}
             else
               {seen, [{arg.name, existing.type, arg.type} | conflicts]}
@@ -135,7 +125,6 @@ defmodule Ash.Resource.Transformers.ResolvePipelines do
         message: "Pipelines define argument(s) with conflicting types: #{conflict_details}"
     end
 
-    # Check pipeline args against action args
     action_args_by_name = Map.new(action.arguments, &{&1.name, &1})
 
     {new_args, action_conflicts} =
@@ -172,10 +161,10 @@ defmodule Ash.Resource.Transformers.ResolvePipelines do
   end
 
   defp merge_pipeline_accepts(accept_lists) do
-    if Enum.any?(accept_lists, &(&1 == :*)) do
+    if :* in accept_lists do
       :*
     else
-      accept_lists |> Enum.reverse() |> Enum.concat() |> Enum.uniq()
+      Enum.uniq(accept_lists)
     end
   end
 
@@ -195,17 +184,6 @@ defmodule Ash.Resource.Transformers.ResolvePipelines do
     end
   end
 
-  defp collect_entities(type, pipeline, where, dsl_state) do
-    subject = subject_for_type(type)
-    entities = source_entities(pipeline, type)
-
-    Enum.each(entities, &validate_supports!(&1, subject, pipeline, dsl_state))
-
-    Enum.map(entities, fn entity ->
-      Map.update!(entity, :where, &(where ++ &1))
-    end)
-  end
-
   defp validate_supports!(
          %Ash.Resource.Validation{module: module, opts: opts},
          subject,

lib/ash/resource/transformers/resolve_pipelines.ex

@@ -160,6 +160,39 @@ defmodule Ash.Test.Resource.Actions.PipelinesTest do
     end
   end
 
+  describe "pipe_through placement ordering" do
+    test "preserves declaration order — change, pipe_through, change" do
+      defresource PlacementBetween do
+        pipelines do
+          pipeline :p do
+            change set_attribute(:score, 1)
+          end
+        end
+
+        actions do
+          create :ordered do
+            accept [:name]
+            change set_attribute(:state, :before)
+            pipe_through [:p]
+            change set_attribute(:name, "after")
+          end
+        end
+      end
+
+      action = Info.action(PlacementBetween, :ordered)
+      [first, second, third] = action.changes
+
+      assert first.change ==
+               {Ash.Resource.Change.SetAttribute, value: :before, attribute: :state}
+
+      assert second.change ==
+               {Ash.Resource.Change.SetAttribute, value: 1, attribute: :score}
+
+      assert third.change ==
+               {Ash.Resource.Change.SetAttribute, value: "after", attribute: :name}
+    end
+  end
+
   describe "multiple pipe_through declarations" do
     test "each declaration's entities are appended in order with separate where" do
       defresource MultiPipeThrough do