Packages
ash
2.21.7
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/flow/executor/ash_engine.ex
defmodule Ash.Flow.Executor.AshEngine do
@moduledoc """
Executes the requests using the Ash engine, which can parallelize individual steps when possible.
"""
use Ash.Flow.Executor
require Logger
require Ash.Tracer
require Ash.Flow.StepHelpers
defmodule Step do
@moduledoc false
defstruct [:step, :input]
end
defmodule Flow do
@moduledoc false
defstruct [:steps, :flow, :returns]
end
def build(flow, input, opts) do
steps =
flow
|> Ash.Flow.Info.steps()
|> to_steps(input, opts)
|> hydrate_flows(opts)
{:ok,
%Flow{
steps: steps,
flow: flow,
returns: Ash.Flow.Info.returns(flow)
}}
end
defp to_steps(steps, input, opts) do
Enum.map(steps, fn step ->
step =
case step do
%{steps: steps} = step ->
step =
if Map.has_key?(step, :output) && is_nil(step.output) do
if last_step = List.last(steps) do
%{step | output: last_step.name}
else
step
end
else
step
end
%{step | steps: to_steps(steps, input, opts)}
%{tenant: _tenant} = step ->
%{step | tenant: Keyword.get(opts, :tenant)}
step ->
step
end
%Step{step: step, input: input}
end)
end
defp hydrate_flows(steps, opts) do
Enum.flat_map(steps, fn
%Step{
input: input,
step: %Ash.Flow.Step.RunFlow{
name: name,
flow: flow,
input: run_flow_input,
wait_for: wait_for,
halt_if: halt_if
}
} = run_flow_step ->
{run_flow_input, _} = Ash.Flow.Template.handle_input_template(run_flow_input, input)
{:ok, %{steps: hydrated_steps}} = build(flow, run_flow_input, opts)
built_steps =
hydrated_steps
|> then(fn steps ->
if wait_for do
map_steps(steps, fn
%{wait_for: nil} = step ->
%{step | wait_for: wait_for}
%{wait_for: step_wait_for} = step
when is_list(step_wait_for) and is_list(wait_for) ->
%{step | wait_for: wait_for ++ step_wait_for}
%{wait_for: step_wait_for} = step when is_list(step_wait_for) ->
%{step | wait_for: [wait_for | step_wait_for]}
%{wait_for: step_wait_for} = step ->
%{step | wait_for: [wait_for, step_wait_for]}
end)
else
steps
end
end)
|> then(fn steps ->
if halt_if do
map_steps(steps, fn
%{halt_if: nil} = step ->
%{step | halt_if: halt_if}
%{halt_if: step_halt_if} = step ->
%{step | halt_if: Ash.Flow.StepHelpers.expr(^step_halt_if or ^halt_if)}
end)
else
steps
end
end)
|> remap_step_names(fn nested_name ->
List.wrap(name) ++ List.wrap(nested_name)
end)
|> handle_input_templates()
Enum.concat(built_steps, [
%{run_flow_step | step: %{run_flow_step.step | built: built_steps}}
])
|> handle_input_templates()
%Step{step: %{steps: steps} = inner_step} = step ->
[%{step | step: %{inner_step | steps: hydrate_flows(steps, opts)}}]
|> handle_input_templates()
step ->
[step] |> handle_input_templates()
end)
end
@deps_keys [
:input,
:over,
:record,
:wait_for,
:halt_if,
:tenant,
:condition,
:resource,
:api,
:action
]
defp handle_input_templates(run_flow_steps) do
run_flow_steps
|> map_outer_steps(fn %Step{step: step, input: input} = outer_step ->
new_step =
Enum.reduce(@deps_keys, step, fn key, step ->
case Map.fetch(step, key) do
{:ok, value} ->
{new_value, _} = Ash.Flow.Template.handle_input_template(value, input)
Map.put(step, key, new_value)
:error ->
step
end
end)
%{outer_step | step: new_step}
end)
end
def execute(%Flow{steps: steps, flow: flow}, _input, opts) do
tracer =
case opts[:tracer] do
nil ->
List.wrap(Application.get_env(:ash, :tracer))
tracer when is_list(tracer) ->
tracer ++ List.wrap(Application.get_env(:ash, :tracer))
tracer ->
[tracer | List.wrap(Application.get_env(:ash, :tracer))]
end
|> Enum.uniq()
steps
|> Enum.flat_map(&requests(flow, steps, &1, opts))
|> resume(opts[:resume])
|> case do
{:error, error} ->
{:error, error}
{:ok, requests} ->
requests
|> Ash.Engine.run(
actor: opts[:actor],
verbose?: opts[:verbose?],
timeout: opts[:timeout],
authorize?: Keyword.get(opts, :authorize?, true),
name: inspect(flow),
failure_mode: :continue,
tracer: tracer
)
|> case do
{:ok, %Ash.Engine{data: data, resource_notifications: resource_notifications}} ->
runner_metadata = %{data: data}
{:ok, return_value(steps, data, Ash.Flow.Info.returns(flow)),
%{notifications: resource_notifications, runner_metadata: runner_metadata}}
{:error,
%Ash.Engine{errors: errors, data: data, resource_notifications: resource_notifications}} ->
{:error, %{runner_metadata: %{data: data}, notifications: resource_notifications},
Ash.Error.to_error_class(errors)}
{:error, error} ->
{:error, error}
end
end
end
defp resume(requests, nil), do: {:ok, requests}
defp resume(requests, %Ash.Flow.Result{runner_metadata: %{data: data}}) do
{:ok,
Enum.map(requests, fn request ->
case Ash.Flow.do_fetch_in(data, request.path) do
{:ok, result} ->
%{request | completion: true, data: result}
:error ->
request
end
end)}
end
defp resume(_, _) do
{:error, "Could not resume flow because engine data was not present!"}
end
defp must_be_local?(%{resource: resource, touches_resources: touches_resources}) do
Ash.Engine.must_be_local?(%Ash.Engine.Request{
resource: resource,
async?: true,
touches_resources: touches_resources
})
end
defp must_be_local?(_), do: false
defp transaction_dependencies(
transaction_name,
transaction_steps,
all_steps
) do
Enum.flat_map(transaction_steps, fn step ->
deps =
step.step
|> get_all_deps()
|> Enum.flat_map(fn {:_result, dep} ->
case find_step_dep(all_steps, dep, transaction_name, true) do
nil ->
[]
step ->
[result_path(step)]
end
end)
case step do
%{step: %{steps: steps}} ->
deps ++ transaction_dependencies(transaction_name, steps, all_steps)
_ ->
deps
end
end)
end
def deps_keys, do: @deps_keys
defp get_all_deps(step) do
Enum.flat_map(@deps_keys, fn key ->
case Map.fetch(step, key) do
{:ok, value} ->
{_, deps} = Ash.Flow.Template.handle_input_template(value, %{})
deps
:error ->
[]
end
end)
end
defp return_value(steps, data, returns) do
case returns do
name when is_atom(name) ->
get_return_value(steps, data, name)
names when is_list(names) ->
Map.new(names, fn
{key, name} ->
{name, get_return_value(steps, data, key)}
name ->
{name, get_return_value(steps, data, name)}
end)
end
end
defp get_return_value(steps, data, name) do
case find_step_dep(steps, name, nil) do
nil ->
nil
%Ash.Flow.Step.Transaction{name: ^name, output: output} ->
get_return_value(steps, data, output)
%Ash.Flow.Step.Transaction{name: transaction_name} ->
Ash.Flow.do_get_in(data, [transaction_name, name])
step ->
Ash.Flow.do_get_in(data, data_path(step))
end
end
defp halt_if(halt_if, reason, step, results, context, otherwise) do
halt_if =
halt_if
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
if halt_if do
{:error, Ash.Error.Flow.Halted.exception(step: step, reason: reason)}
else
otherwise.()
end
end
defp requests(flow, all_steps, step, opts, request_opts \\ []) do
additional_context = request_opts[:context] || %{}
transaction_name = request_opts[:transaction_name]
case step do
%Step{
step: %Ash.Flow.Step.Debug{
name: name,
input: input,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
}
} ->
{input, deps} = Ash.Flow.Template.handle_input_template(input, %{})
{_, wait_for_deps} = Ash.Flow.Template.handle_input_template(wait_for, %{})
{_, halt_if_deps} = Ash.Flow.Template.handle_input_template(halt_if, %{})
dep_paths =
get_dep_paths(all_steps, deps, transaction_name, wait_for_deps ++ halt_if_deps)
request_deps = dependable_request_paths(dep_paths)
id = Ash.UUID.generate()
[
Ash.Engine.Request.new(
path: [name],
name: inspect(name),
async?: false,
error_path: [name],
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
input =
input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
output = """
Debug Output for: #{inspect(name)} | #{id}
#{inspect(input)}
"""
IO.puts(output)
{:ok, output}
end)
end)
)
]
%Step{
step:
%Ash.Flow.Step.Transaction{
steps: transaction_steps,
name: name,
resource: resource,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason,
touches_resources: touches_resources
} = transaction
} ->
request_name = "Transaction #{inspect(name)}"
maybe_dynamic(
[resource, touches_resources],
request_name,
[name],
%{},
all_steps,
transaction_name,
additional_context,
fn resource, touches_resources ->
depends_on_requests =
transaction_dependencies(name, transaction_steps, all_steps) |> Enum.uniq()
{_, wait_for_deps} = Ash.Flow.Template.handle_input_template(wait_for, %{})
{_, halt_if_deps} = Ash.Flow.Template.handle_input_template(halt_if, %{})
dep_paths = get_dep_paths(all_steps, [], name, wait_for_deps ++ halt_if_deps)
request_deps = dependable_request_paths(dep_paths)
touches_resources =
if resource do
[resource | touches_resources || []]
else
touches_resources || []
end
|> List.flatten()
|> Enum.filter(&(&1 && is_atom(&1)))
if touches_resources == [] do
Logger.error("""
No resources configured for transaction #{inspect(name)}.
""")
end
[
Ash.Engine.Request.new(
path: [name],
name: request_name,
async?: Enum.any?(transaction_steps, &must_be_local?/1),
touches_resources: touches_resources,
error_path: [name],
data:
Ash.Engine.Request.resolve(depends_on_requests ++ request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
transaction_steps
|> Enum.flat_map(
&requests(flow, all_steps, &1, opts,
context: context,
transaction_name: transaction.name
)
)
# TODO: Can we resume a transaction? For now, you cannot.
# Either the whole thing fails or the whole thing succeeds.
|> case do
[] ->
{:ok, %{}}
[first | rest] ->
# only one of the requests needs to be annotated as touching the resources
# the transaction claims to touch, since the transaction is run over all touched resources
# in all requests
[
%{
first
| touches_resources:
Enum.uniq(touches_resources ++ first.touches_resources)
}
| rest
]
|> Ash.Engine.run(
transaction_reason: %{
type: :flow_transaction,
metadata: %{
step_name: name,
flow: flow
}
},
authorize?: context[:authorize?],
resource: resource,
name: "Transaction #{inspect(name)}",
verbose?: context[:verbose?],
tracer: context[:tracer],
actor: context[:actor],
failure_mode: :stop,
transaction?: true
)
|> case do
{:ok, %{data: data, resource_notifications: notifications}} ->
if opts[:return_notifications?] do
{:ok,
Map.new(transaction_steps, fn step ->
{step.step.name,
Ash.Flow.do_get_in(data, data_path(step.step))}
end), %{notifications: notifications}}
else
{:ok,
Map.new(transaction_steps, fn step ->
{step.step.name,
Ash.Flow.do_get_in(data, data_path(step.step))}
end)}
end
{:error, %Ash.Engine{errors: errors}} ->
{:error, Ash.Error.to_error_class(errors)}
{:error, error} ->
{:error, error}
end
end
end)
end)
)
]
end
)
%Step{
step: %Ash.Flow.Step.Branch{
name: name,
steps: branch_steps,
output: output,
condition: condition,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
},
input: input
} ->
output = output || List.last(branch_steps).step.name
{condition, deps} = Ash.Flow.Template.handle_input_template(condition, input)
{_, wait_for_deps} = Ash.Flow.Template.handle_input_template(wait_for, input)
{_, halt_if_deps} = Ash.Flow.Template.handle_input_template(halt_if, input)
dep_paths =
get_dep_paths(all_steps, deps, transaction_name, wait_for_deps ++ halt_if_deps)
request_deps = dependable_request_paths(dep_paths)
[
Ash.Engine.Request.new(
path: [name],
async?: true,
name: "#{inspect(name)}",
error_path: List.wrap(name),
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
execute? =
condition
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
output_step = Enum.find(branch_steps, &(&1.step.name == output))
if is_nil(output_step) do
raise "#{inspect(output)} not found in #{inspect(step_names(branch_steps))}"
end
output_step = output_step.step
output_path = result_path(output_step)
result = Ash.Flow.do_fetch_in(context, output_path)
if execute? && match?({:ok, _}, result) do
result
else
if execute? do
all_steps_with_branch_steps =
all_steps
|> update_step(name, fn step ->
%{step | step: %{step.step | steps: branch_steps}}
end)
{:requests,
branch_steps
|> Enum.flat_map(
&requests(flow, all_steps_with_branch_steps, &1, opts,
context: context,
transaction_name: transaction_name
)
), [output_path]}
else
requests =
branch_steps
|> step_names()
|> Enum.map(fn name ->
Ash.Engine.Request.new(
authorize?: false,
async?: false,
name: "Set nil result for #{inspect(name)}",
path: [name],
data: {:ok, nil}
)
end)
{:ok, nil,
%{
requests: requests
}}
end
end
end)
end)
)
]
%Step{
step: %Ash.Flow.Step.Map{
name: name,
steps: map_steps,
over: over,
output: output,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
},
input: input
} ->
output = output || List.last(map_steps).step.name
{over, deps} = Ash.Flow.Template.handle_input_template(over, input)
{_, wait_for_deps} = Ash.Flow.Template.handle_input_template(wait_for, input)
{_, halt_if_deps} = Ash.Flow.Template.handle_input_template(halt_if, input)
dep_paths =
get_dep_paths(all_steps, deps, transaction_name, wait_for_deps ++ halt_if_deps)
request_deps = dependable_request_paths(dep_paths)
[
Ash.Engine.Request.new(
path: [name],
async?: true,
name: "#{inspect(name)}",
error_path: List.wrap(name),
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
elements =
over
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> Kernel.||([])
if Enum.empty?(elements) do
{:ok, []}
else
case Ash.Flow.do_get_in(context, [[name, :elements]]) do
nil ->
elements
|> Enum.with_index()
|> Enum.flat_map(fn {element, i} ->
new_additional_context =
additional_context
|> Map.put_new(:_ash_engine_elements, %{})
|> Map.update!(:_ash_engine_elements, &Map.put(&1, name, element))
|> Map.put_new(:_ash_engine_indices, %{})
|> Map.update!(:_ash_engine_indices, &Map.put(&1, name, i))
map_steps =
map_steps
|> remap_step_names(&set_index(&1, i))
output_name = set_index(output, i)
output_step = Enum.find(map_steps, &(&1.step.name == output_name))
if is_nil(output_step) do
raise "#{inspect(output_name)} not found in #{inspect(step_names(map_steps))}"
end
output_step = output_step.step
output_path = result_path(output_step)
all_steps_with_map_steps =
all_steps
|> update_step(name, fn step ->
%{step | step: %{step.step | steps: map_steps}}
end)
map_steps
|> Enum.flat_map(
&requests(flow, all_steps_with_map_steps, &1, opts,
context: new_additional_context,
transaction_name: transaction_name
)
)
|> Kernel.++([
{Ash.Engine.Request.new(
path: [[name, :elements], i],
error_path: [[name, :elements], i],
name: "Handle #{inspect(name)} index #{i}",
authorize?: false,
additional_context: new_additional_context,
data:
Ash.Engine.Request.resolve([output_path], fn context ->
{:ok, Ash.Flow.do_get_in(context, output_path)}
end),
async?: true
), :data}
])
end)
|> case do
[] ->
{:ok, []}
requests ->
{:requests, requests}
end
elements ->
{:ok,
elements
|> Enum.sort_by(&elem(&1, 0))
|> Enum.map(&elem(&1, 1))
|> Enum.map(&Map.get(&1 || %{}, :data))}
end
end
end)
end)
)
]
%Step{
step: %Ash.Flow.Step.Custom{
name: name,
tenant: tenant,
input: custom_input,
custom: {mod, opts},
async?: async?,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
},
input: input
} ->
{custom_input, deps} = Ash.Flow.Template.handle_input_template(custom_input, input)
{_, wait_for_deps} = Ash.Flow.Template.handle_input_template(wait_for, input)
{_, halt_if_deps} = Ash.Flow.Template.handle_input_template(halt_if, input)
dep_paths =
get_dep_paths(all_steps, deps, transaction_name, wait_for_deps ++ halt_if_deps)
request_deps = dependable_request_paths(dep_paths)
[
Ash.Engine.Request.new(
authorize?: false,
async?: async?,
name: "Run custom step #{inspect(name)}",
path: [name],
error_path: List.wrap(name),
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
custom_input =
custom_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
context_arg =
context
|> Map.take([:actor, :authorize?, :async?, :verbose?])
|> Map.put(:tenant, tenant)
Ash.Tracer.span :custom_flow_step,
"custom step #{inspect(mod)}",
opts[:tracer] do
metadata = %{
flow_short_name: Ash.Flow.Info.short_name(flow),
name: inspect(name),
step: inspect(mod)
}
Ash.Tracer.set_metadata(opts[:tracer], :custom_flow_step, metadata)
Ash.Tracer.telemetry_span [:ash, :flow, :custom_step], metadata do
mod.run(custom_input, opts, context_arg)
end
end
end)
end)
)
]
%Step{step: %Ash.Flow.Step.RunFlow{name: name, flow: flow, built: built}} ->
# No need to do anything with `wait_for` here, because
# `wait_for` is pushed down into all child steps when we hydrate
result_dependencies =
flow
|> Ash.Flow.Info.returns()
|> List.wrap()
|> Enum.map(fn
{key, _} ->
key
other ->
other
end)
|> Enum.map(fn key ->
{:_result, List.wrap(name) ++ List.wrap(key)}
end)
# If you want the result of the run flow step, you only get it
# when all parts of that flow are complete.
dep_paths =
get_dep_paths(
all_steps,
result_dependencies,
transaction_name,
step_names(built)
)
request_deps = dependable_request_paths(dep_paths)
[
Ash.Engine.Request.new(
path: [name],
name: "Build return value for #{inspect(name)}",
async?: false,
error_path: List.wrap(name),
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
case Ash.Flow.Info.returns(flow) do
key when is_atom(key) ->
{:ok,
get_flow_return_value(
all_steps,
List.wrap(name) ++ [key],
context,
transaction_name
)}
list when is_list(list) ->
list =
Enum.map(list, fn
{key, val} ->
{key, val}
key ->
{key, key}
end)
{:ok,
Map.new(list, fn {key, val} ->
{val,
get_flow_return_value(
all_steps,
List.wrap(name) ++ [key],
context,
transaction_name
)}
end)}
end
end)
)
]
%Step{step: %Ash.Flow.Step.Read{} = read, input: input} ->
%{
action: action,
api: api,
name: name,
resource: resource,
input: action_input,
tenant: tenant,
get?: get?,
not_found_error?: not_found_error?,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
} = read
List.wrap(
maybe_dynamic(
[resource, action],
name,
[name, :data],
input,
all_steps,
transaction_name,
additional_context,
fn resource, action ->
%{
action: action,
action_input: action_input,
resource: resource,
api: api,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
api,
action_input,
input,
transaction_name,
tenant,
[wait_for, halt_if]
)
Ash.Engine.Request.new(
resource: resource,
path: [name],
name: inspect([name]),
authorize?: false,
data:
Ash.Engine.Request.resolve(request_deps, fn data ->
data = Ash.Helpers.deep_merge_maps(data, additional_context)
results = results(dep_paths, data)
case halt_if(halt_if, halt_reason, name, results, data, fn ->
{:ok, Ash.Query.new(resource)}
end) do
{:error, error} ->
{:error, error}
{:ok, query} ->
tenant =
tenant
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
action_input =
action_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> Kernel.||(%{})
query
|> Ash.Query.for_read(action.name, action_input,
actor: data[:actor],
tenant: tenant,
authorize?: opts[:authorize?],
tracer: data[:tracer]
)
|> then(fn query ->
if get? do
case api.read_one(query) do
{:ok, nil} ->
if not_found_error? do
{:error, Ash.Error.Query.NotFound.exception(resource: resource)}
else
{:ok, nil}
end
{:ok, result} ->
{:ok, result}
other ->
other
end
else
api.read(query)
end
end)
end
end)
)
end
)
)
%Step{step: %Ash.Flow.Step.Create{} = create, input: input} ->
%{
action: action,
api: api,
name: name,
resource: resource,
input: action_input,
tenant: tenant,
wait_for: wait_for,
halt_if: halt_if,
upsert?: upsert?,
upsert_identity: upsert_identity,
halt_reason: halt_reason
} = create
List.wrap(
maybe_dynamic(
[resource, action],
name,
[name, :commit],
input,
all_steps,
transaction_name,
additional_context,
fn resource, action ->
%{
action: action,
resource: resource,
api: api,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
api,
action_input,
input,
transaction_name,
tenant,
[wait_for, halt_if]
)
Ash.Engine.Request.new(
resource: resource,
path: [name],
name: inspect([name]),
authorize?: false,
data:
Ash.Engine.Request.resolve(request_deps, fn data ->
data = Ash.Helpers.deep_merge_maps(data, additional_context)
results = results(dep_paths, data)
case halt_if(halt_if, halt_reason, name, results, data, fn ->
{:ok, Ash.Changeset.new(resource)}
end) do
{:error, error} ->
{:error, error}
{:ok, changeset} ->
tenant =
tenant
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
action_input =
action_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> Kernel.||(%{})
changeset
|> Ash.Changeset.set_tenant(tenant)
|> Ash.Changeset.for_create(action.name, action_input,
actor: data[:actor],
tenant: tenant,
authorize?: opts[:authorize?],
tracer: data[:tracer],
upsert?: upsert?,
upsert_identity: upsert_identity
)
|> api.create()
end
end)
)
end
)
)
%Step{step: %Ash.Flow.Step.Validate{only_keys: keys} = validate, input: input} ->
%{
action: action,
name: name,
resource: resource,
input: action_input,
record: record,
tenant: tenant,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
} = validate
%{
action: action,
resource: resource,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
nil,
action_input,
input,
transaction_name,
tenant,
[wait_for, halt_if]
)
{changeset_deps, get_request} =
if action.type in [:create, :read] do
get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
)
else
{[], nil}
end
validate_request =
Ash.Engine.Request.new(
path: [name, :validate],
name: "Fetch record for #{inspect(name)}",
error_path: List.wrap(name) ++ [:fetch],
async?: must_be_local?(%{resource: resource}),
resource: resource,
authorize?: false,
data:
Ash.Engine.Request.resolve(request_deps ++ changeset_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(changeset_deps ++ dep_paths, context)
halt_if(halt_if, halt_reason, name, results, context, fn ->
tenant =
tenant
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
input =
action_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
result =
case action.type do
:read ->
Ash.Query.for_read(resource, action.name, input,
tenant: tenant,
actor: context[:actor],
authorize?: context[:authorize?],
tracer: context[:tracer]
)
:create ->
Ash.Changeset.for_create(resource, action.name, input,
tenant: tenant,
actor: context[:actor],
authorize?: context[:authorize?],
tracer: context[:tracer]
)
type when type in [:update, :destroy] ->
record
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> case do
%^resource{} = record ->
Ash.Changeset.for_action(record, action.name, input,
tenant: tenant,
actor: context[:actor],
authorize?: context[:authorize?],
tracer: context[:tracer]
)
nil ->
{:error, "Cannot validate #{action.name} because input is nil"}
value ->
{:error, "Invalid record #{inspect(value)}"}
end
end
extract_changeset_or_query_errors(result, keys)
end)
end)
)
if get_request do
[get_request, validate_request]
else
[validate_request]
end
%Step{step: %Ash.Flow.Step.Update{} = update, input: input} ->
%{
action: action,
api: api,
name: name,
resource: resource,
input: action_input,
record: record,
tenant: tenant,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
} = update
List.wrap(
maybe_dynamic(
[resource, action],
name,
[name, :commit],
input,
all_steps,
transaction_name,
additional_context,
fn resource, action ->
%{
action: action,
action_input: action_input,
resource: resource,
api: api,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
api,
action_input,
input,
transaction_name,
tenant,
[wait_for, halt_if]
)
get_request =
get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
)
update_request =
Ash.Engine.Request.new(
resource: resource,
path: [name],
name: inspect([name]),
authorize?: false,
data:
Ash.Engine.Request.resolve([[name, :fetch, :data] | request_deps], fn data ->
case Ash.Flow.do_get_in(data, [name, :fetch, :data]) do
nil ->
{:ok, nil}
record ->
data = Ash.Helpers.deep_merge_maps(data, additional_context)
results = results(dep_paths, data)
case halt_if(halt_if, halt_reason, name, results, data, fn ->
{:ok, Ash.Changeset.new(record)}
end) do
{:error, error} ->
{:error, error}
{:ok, changeset} ->
tenant =
tenant
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
action_input =
action_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> Kernel.||(%{})
changeset
|> Ash.Changeset.set_tenant(tenant)
|> Ash.Changeset.for_update(action.name, action_input,
actor: data[:actor],
tenant: tenant,
authorize?: opts[:authorize?],
tracer: data[:tracer]
)
|> api.update()
end
end
end)
)
[
get_request,
update_request
]
end
)
)
%Step{step: %Ash.Flow.Step.Destroy{} = destroy, input: input} ->
%{
action: action,
api: api,
name: name,
resource: resource,
input: action_input,
record: record,
tenant: tenant,
wait_for: wait_for,
halt_if: halt_if,
halt_reason: halt_reason
} = destroy
List.wrap(
maybe_dynamic(
[resource, action],
name,
[name, :commit],
input,
all_steps,
transaction_name,
additional_context,
fn resource, action ->
%{
action: action,
action_input: action_input,
resource: resource,
api: api,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
api,
action_input,
input,
transaction_name,
tenant,
[wait_for, halt_if]
)
get_request =
get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
)
destroy_request =
Ash.Engine.Request.new(
resource: resource,
path: [name],
name: inspect([name]),
authorize?: false,
data:
Ash.Engine.Request.resolve([[name, :fetch, :data] | request_deps], fn data ->
case Ash.Flow.do_get_in(data, [name, :fetch, :data]) do
nil ->
{:ok, nil}
record ->
data = Ash.Helpers.deep_merge_maps(data, additional_context)
results = results(dep_paths, data)
case halt_if(halt_if, halt_reason, name, results, data, fn ->
{:ok, Ash.Changeset.new(record)}
end) do
{:error, error} ->
{:error, error}
{:ok, changeset} ->
tenant =
tenant
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
action_input =
action_input
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(data, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> Kernel.||(%{})
changeset
|> Ash.Changeset.set_tenant(tenant)
|> Ash.Changeset.for_destroy(
action.name,
action_input,
actor: data[:actor],
tenant: tenant,
authorize?: opts[:authorize?],
tracer: data[:tracer]
)
|> api.destroy()
|> case do
:ok ->
{:ok, record}
other ->
other
end
end
end
end)
)
[
get_request,
destroy_request
]
end
)
)
end
end
defp maybe_dynamic(
items,
name,
request_path,
input,
all_steps,
transaction_name,
additional_context,
fun
) do
if Ash.Flow.Template.is_template?(items) do
{items, deps} =
Enum.reduce(items, {[], []}, fn item, {items, deps} ->
{item, new_deps} = Ash.Flow.Template.handle_input_template(item, input)
{[item | items], deps ++ new_deps}
end)
dep_paths = get_dep_paths(all_steps, deps, transaction_name, [])
request_deps = dependable_request_paths(dep_paths)
Ash.Engine.Request.new(
authorize?: false,
async?: false,
name: "Run dynamic step #{inspect(name)}",
path: [:dynamic, request_path],
error_path: [],
data:
Ash.Engine.Request.resolve(request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
items =
items
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
requests =
fun
|> apply(Enum.reverse(items))
|> List.wrap()
{:ok, nil, %{requests: requests}}
end)
)
else
apply(fun, items)
end
end
defp extract_changeset_or_query_errors(result, keys) do
keys =
Enum.map(keys, fn key ->
if is_list(key) do
{:lists.droplast(key), List.last(key)}
else
{[], key}
end
end)
result.errors
|> expand_errors()
|> Enum.filter(&error_applies?(&1, keys))
end
defp error_applies?(error, keys) do
Enum.any?(keys, fn {path, key} ->
path_matches?(error.path, path ++ [key]) ||
(path_matches?(error.path || [], path) && field_matches?(error, key))
end)
end
defp field_matches?(error, key) do
key in List.wrap(Map.get(error, :fields)) || key == Map.get(error, :field)
end
defp path_matches?(nil, []), do: true
defp path_matches?(path, path), do: true
defp path_matches?(_, _), do: false
defp expand_errors(errors) when is_list(errors) do
Enum.flat_map(errors, &expand_errors/1)
end
defp expand_errors(%class_mod{} = error)
when class_mod in [
Ash.Error.Forbidden,
Ash.Error.Framework,
Ash.Error.Invalid,
Ash.Error.Unkonwn
] do
Enum.flat_map(error.errors, &expand_errors/1)
end
defp expand_errors(other), do: List.wrap(other)
defp set_index(name, i) do
List.wrap(i) ++ List.wrap(name)
end
defp results(dep_paths, context) do
Map.new(dep_paths, fn {name, %{fetch_path: fetch_path}} ->
{name, Ash.Flow.do_get_in(context, fetch_path)}
end)
end
defp map_steps(steps, fun) do
steps
|> Enum.map(fn %Step{step: inner_step} = step ->
%{step | step: fun.(inner_step)}
end)
|> Enum.map(fn
%Step{step: %{steps: inner_steps}} = step ->
%{step | step: %{step.step | steps: map_steps(inner_steps, fun)}}
step ->
step
end)
end
defp map_outer_steps(steps, fun) do
steps
|> Enum.map(fn step ->
fun.(step)
end)
|> Enum.map(fn
%Step{step: %{steps: inner_steps}} = step ->
%{step | step: %{step.step | steps: map_outer_steps(inner_steps, fun)}}
step ->
step
end)
end
defp remap_step_names(steps, fun, step_names \\ nil) do
step_names = step_names || step_names(steps)
steps
|> Enum.map(fn
%Step{step: %{name: name} = inner_step} = step ->
%{step | step: %{inner_step | name: fun.(name)}}
|> remap_step(step_names, fun)
end)
|> Enum.map(fn
%Step{step: %{steps: inner_steps}} = step ->
%{step | step: %{step.step | steps: remap_step_names(inner_steps, fun, step_names)}}
other ->
other
end)
end
defp step_names(steps) do
Enum.flat_map(steps, fn step ->
case step.step do
%{name: name, steps: steps} ->
[name | step_names(steps)]
%{name: name} ->
[name]
end
end)
end
defp update_step(steps, name, func) do
Enum.map(steps, fn step ->
if step.step.name == name do
func.(step)
else
case step do
%{step: %{steps: steps}} ->
%{step | step: %{step.step | steps: update_step(steps, name, func)}}
step ->
step
end
end
end)
end
# output needs a config
@remapped_step_keys [
input: [],
over: [],
output: [
raw?: true
],
condition: [],
record: [],
tenant: [],
wait_for: [],
halt_if: []
]
defp remap_step(step, step_names, fun) do
Enum.reduce(@remapped_step_keys, step, fn {key, config}, step ->
remap_key(step, step_names, key, config, fun)
end)
end
defp remap_key(%{step: step} = outer_step, step_names, key, config, fun) do
case Map.fetch(step, key) do
{:ok, value} ->
new_value =
if config[:raw?] do
if value do
fun.(value)
end
else
Ash.Flow.Template.remap_result_references(value, fn name ->
if name in step_names do
fun.(name)
else
name
end
end)
end
%{outer_step | step: Map.put(step, key, new_value)}
:error ->
outer_step
end
end
defp get_flow_return_value(
steps,
name,
data,
transaction_name,
in_transaction? \\ false,
default \\ nil
) do
case find_step_dep(steps, name, transaction_name) do
nil ->
default
%Ash.Flow.Step.Transaction{} = transaction ->
data =
if in_transaction? do
Ash.Flow.do_get_in(data, [transaction.name])
else
Ash.Flow.do_get_in(data, [transaction.name, :data])
end
get_flow_return_value(transaction.steps, name, data, transaction.name, true, data)
step ->
if in_transaction? do
Ash.Flow.do_get_in(data, [step.name])
else
Ash.Flow.do_get_in(data, result_path(%{step | name: name}))
end
end
end
defp get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
) do
{record, record_deps} = Ash.Flow.Template.handle_input_template(record, input)
get_request_dep_paths =
all_steps
|> get_dep_paths(record_deps, transaction_name, [])
get_request_deps = dependable_request_paths(get_request_dep_paths)
Ash.Engine.Request.new(
path: [name, :fetch],
name: "Fetch record for #{inspect(name)}",
error_path: List.wrap(name) ++ [:fetch],
async?: must_be_local?(%{resource: resource}),
resource: resource,
authorize?: false,
data:
Ash.Engine.Request.resolve(get_request_deps, fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(get_request_dep_paths, context)
record
|> Ash.Flow.Template.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
|> case do
%^resource{} = record ->
{:ok, record}
nil ->
{:ok, nil}
value ->
{:error, "Invalid record #{inspect(value)}"}
end
end)
)
end
defp action_request_info(
all_steps,
resource,
action,
api,
action_input,
input,
transaction_name,
tenant,
additional
) do
{action_input, deps} = Ash.Flow.Template.handle_input_template(action_input, input)
{tenant, tenant_deps} = Ash.Flow.Template.handle_input_template(tenant, input)
{_, additional_deps} = Ash.Flow.Template.handle_input_template(additional, input)
{resource, resource_deps} = Ash.Flow.Template.handle_input_template(resource, input)
{api, api_deps} = Ash.Flow.Template.handle_input_template(api, input)
{action, action_deps} = Ash.Flow.Template.handle_input_template(action, input)
action =
Ash.Resource.Info.action(resource, action) ||
raise "No such action #{action} for #{resource}"
dep_paths =
get_dep_paths(
all_steps,
deps ++ tenant_deps ++ api_deps ++ resource_deps ++ action_deps,
transaction_name,
additional_deps
)
request_deps = dependable_request_paths(dep_paths)
%{
resource: resource,
api: api,
action: action,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
}
end
defp dependable_request_paths(dep_paths) do
for {_, %{request_path: request_path, depend?: true}} <- dep_paths do
request_path
end
end
defp get_dep_paths(all_steps, deps, transaction_name, wait_for) do
# We favor the result dependencies over the completion dependencies
Map.merge(
dep_paths(wait_for, all_steps, transaction_name, :completion),
dep_paths(deps, all_steps, transaction_name)
)
end
defp dep_paths(deps, steps, transaction_name, result_type \\ :result) do
deps
|> Enum.reduce(%{}, fn
{:_result, dep}, acc ->
case find_step_dep(steps, dep, transaction_name, false, true) do
nil ->
acc
{step, :no_depend} ->
add_dep(step, dep, acc, result_type, false)
step ->
add_dep(step, dep, acc, result_type, true)
end
_, acc ->
acc
end)
end
defp add_dep(step, dep, acc, type, depend?) do
case step do
%Ash.Flow.Step.Transaction{name: name} ->
request_path =
if type == :completion do
[name, :completion]
else
[name, :data]
end
Map.put(acc, dep, %{
fetch_path: [name, :data, dep],
request_path: request_path,
depend?: depend?
})
step ->
request_path =
if type == :completion do
completion_path(step)
else
result_path(step)
end
Map.put(acc, dep, %{
fetch_path: request_path,
request_path: request_path,
depend?: depend?
})
end
end
defp find_step_dep(
steps,
dep,
transaction_name,
stop_at_transaction_name? \\ false,
no_depend? \\ false
) do
Enum.find_value(steps, fn
%Step{step: %Ash.Flow.Step.Transaction{name: ^dep} = step} ->
if transaction_name && no_depend? do
{step, :no_depend}
else
step
end
%Step{step: %Ash.Flow.Step.Transaction{name: ^transaction_name}}
when stop_at_transaction_name? ->
nil
%Step{step: %Ash.Flow.Step.Transaction{name: ^transaction_name, steps: steps}} ->
find_step_dep(steps, dep, nil, stop_at_transaction_name?, no_depend?)
%Step{step: %Ash.Flow.Step.Transaction{steps: steps} = step} ->
case find_step_dep(steps, dep, transaction_name, stop_at_transaction_name?, no_depend?) do
nil ->
nil
inner_step ->
if transaction_name do
inner_step
else
step
end
end
%Step{step: %{name: ^dep} = step} ->
if transaction_name && no_depend? do
{step, :no_depend}
else
step
end
%Step{step: %{steps: steps}} ->
find_step_dep(steps, dep, transaction_name, stop_at_transaction_name?, no_depend?)
_ ->
nil
end)
end
defp result_path(%Ash.Flow.Step.Read{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Create{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Destroy{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Validate{name: name}) do
[name, :validate, :data]
end
defp result_path(%Ash.Flow.Step.Update{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Transaction{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.RunFlow{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Branch{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Map{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Custom{name: name}) do
[name, :data]
end
defp result_path(%Ash.Flow.Step.Debug{name: name}) do
[name, :data]
end
defp completion_path(%Ash.Flow.Step.Read{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Create{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Validate{name: name}) do
[name, :commit, :completion]
end
defp completion_path(%Ash.Flow.Step.Update{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Destroy{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Transaction{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.RunFlow{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Branch{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Map{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Custom{name: name}) do
[name, :completion]
end
defp completion_path(%Ash.Flow.Step.Debug{name: name}) do
[name, :completion]
end
defp data_path(%Ash.Flow.Step.Debug{name: name}) do
[name, :data]
end
defp data_path(%Ash.Flow.Step.Transaction{name: name}) do
[name, :data]
end
defp data_path(%Ash.Flow.Step.Read{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Create{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Destroy{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Validate{name: name}) do
[name, :commit]
end
defp data_path(%Ash.Flow.Step.Update{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.RunFlow{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Custom{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Branch{name: name}) do
[name]
end
defp data_path(%Ash.Flow.Step.Map{name: name}) do
[name]
end
end