Packages
ash
1.52.0-rc.18
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
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)
|> hydrate_flows()
{:ok,
%Flow{
steps: steps,
flow: flow,
returns: Ash.Flow.Info.returns(flow)
}}
end
defp to_steps(steps, input) do
Enum.map(steps, fn step ->
step =
case step do
%{steps: steps} = step ->
%{step | steps: to_steps(steps, input)}
step ->
step
end
%Step{step: step, input: input}
end)
end
defp hydrate_flows(steps) 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
}
} = run_flow_step ->
{run_flow_input, _} = Ash.Flow.handle_input_template(run_flow_input, input)
{:ok, %{steps: hydrated_steps}} = build(flow, run_flow_input, [])
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)
|> 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}}
])
%Step{step: %{steps: steps} = inner_step} = step ->
[%{step | step: %{inner_step | steps: hydrate_flows(steps)}}]
step ->
[step]
end)
end
@deps_keys [:input, :over, :record, :wait_for, :tenant]
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.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
steps
|> Enum.flat_map(&requests(steps, &1, opts))
|> Ash.Engine.run(verbose?: opts[:verbose?], timeout: opts[:timeout], name: inspect(flow))
|> case do
{:ok, %Ash.Engine{data: data, resource_notifications: resource_notifications}} ->
if opts[:return_notifications?] do
{:ok, return_value(steps, data, Ash.Flow.Info.returns(flow)),
%{notifications: resource_notifications}}
else
{:ok, return_value(steps, data, Ash.Flow.Info.returns(flow))}
end
{:error, %Ash.Engine{errors: errors}} ->
{:error, Ash.Error.to_error_class(errors)}
{:error, error} ->
{:error, error}
end
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.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: transaction_name} ->
Ash.Flow.do_get_in(data, [transaction_name, name])
step ->
Ash.Flow.do_get_in(data, data_path(step))
end
end
defp requests(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
}
} ->
{input, deps} = Ash.Flow.handle_input_template(input, %{})
{_, wait_for_deps} = Ash.Flow.handle_input_template(wait_for, %{})
dep_paths = get_dep_paths(all_steps, deps, transaction_name, wait_for_deps)
request_deps = dependable_request_paths(dep_paths)
[
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)
input =
input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
IO.puts("""
Debug Output for: #{inspect(name)}
#{inspect(input)}
""")
{:ok, input}
end)
)
]
%Step{
step:
%Ash.Flow.Step.Transaction{
steps: transaction_steps,
name: name,
resource: resource,
wait_for: wait_for,
touches_resources: touches_resources
} = transaction
} ->
depends_on_requests =
transaction_dependencies(name, transaction_steps, all_steps) |> Enum.uniq()
{_, wait_for_deps} = Ash.Flow.handle_input_template(wait_for, %{})
dep_paths = get_dep_paths(all_steps, [], name, wait_for_deps)
request_deps = dependable_request_paths(dep_paths)
[
Ash.Engine.Request.new(
path: [name],
name: "Transaction #{inspect(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 ->
transaction_steps
|> Enum.flat_map(
&requests(all_steps, &1, opts,
context: context,
transaction_name: transaction.name
)
)
|> 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 urn over all touched resources
# in all requests
[
%{
first
| touches_resources:
Enum.uniq(touches_resources ++ first.touches_resources)
}
| rest
]
|> Ash.Engine.run(
resource: resource,
name: "Transaction #{inspect(name)}",
verbose?: opts[:verbose?],
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)
)
]
%Step{
step: %Ash.Flow.Step.Map{
name: name,
steps: map_steps,
over: over,
output: output,
wait_for: wait_for
},
input: input
} ->
output = output || List.last(map_steps).step.name
{over, deps} = Ash.Flow.handle_input_template(over, input)
{_, wait_for_deps} = Ash.Flow.handle_input_template(wait_for, input)
dep_paths = get_dep_paths(all_steps, deps, transaction_name, wait_for_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)
elements =
over
|> Ash.Flow.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(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)
)
]
%Step{
step: %Ash.Flow.Step.Custom{
name: name,
input: custom_input,
custom: {mod, opts},
async?: async?,
wait_for: wait_for
},
input: input
} ->
{custom_input, deps} = Ash.Flow.handle_input_template(custom_input, input)
{_, wait_for_deps} = Ash.Flow.handle_input_template(wait_for, input)
dep_paths = get_dep_paths(all_steps, deps, transaction_name, wait_for_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)
custom_input =
custom_input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
context_arg = Map.take(context, [:actor, :authorize?, :async?, :verbose?])
mod.run(custom_input, opts, context_arg)
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?,
wait_for: wait_for
} = read
%{
action: action,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
action_input,
input,
transaction_name,
tenant,
wait_for
)
Ash.Actions.Read.as_requests([name], resource, api, action,
error_path: List.wrap(name),
authorize?: opts[:authorize?],
actor: opts[:actor],
query_dependencies: request_deps,
get?: get? || action.get?,
tenant: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
tenant
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
end,
query_input: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
action_input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
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
} = create
%{
action: action,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
action_input,
input,
transaction_name,
tenant,
wait_for
)
Ash.Actions.Create.as_requests([name], resource, api, action,
error_path: List.wrap(name),
authorize?: opts[:authorize?],
actor: opts[:actor],
changeset_dependencies: request_deps,
tenant: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
tenant
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
end,
changeset_input: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
action_input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
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
} = update
%{
action: action,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
action_input,
input,
transaction_name,
tenant,
wait_for
)
get_request =
get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
)
[
get_request
| Ash.Actions.Update.as_requests([name], resource, api, action,
error_path: List.wrap(name),
authorize?: opts[:authorize?],
actor: opts[:actor],
changeset_dependencies: [[name, :fetch, :data] | request_deps],
skip_on_nil_record?: true,
tenant: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
tenant
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
end,
record: fn context ->
Ash.Flow.do_get_in(context, [name, :fetch, :data])
end,
changeset_input: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
action_input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
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
} = destroy
%{
action: action,
action_input: action_input,
dep_paths: dep_paths,
tenant: tenant,
request_deps: request_deps
} =
action_request_info(
all_steps,
resource,
action,
action_input,
input,
transaction_name,
tenant,
wait_for
)
get_request =
get_request(
record,
input,
all_steps,
transaction_name,
name,
resource,
additional_context
)
[
get_request
| Ash.Actions.Destroy.as_requests([name], resource, api, action,
error_path: List.wrap(name),
authorize?: opts[:authorize?],
actor: opts[:actor],
changeset_dependencies: [[name, :fetch, :data] | request_deps],
skip_on_nil_record?: true,
record: fn context ->
Ash.Flow.do_get_in(context, [name, :fetch, :data])
end,
tenant: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
tenant
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
end,
changeset_input: fn context ->
context = Ash.Helpers.deep_merge_maps(context, additional_context)
results = results(dep_paths, context)
action_input
|> Ash.Flow.set_dependent_values(%{
results: results,
elements: Map.get(context, :_ash_engine_elements)
})
|> Ash.Flow.handle_modifiers()
end
)
]
end
end
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
],
record: [],
tenant: [],
wait_for: []
]
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.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.handle_input_template(record, input)
get_request_dep_paths = get_dep_paths(all_steps, record_deps, transaction_name, [])
get_request_deps =
Enum.map(get_request_dep_paths, fn {_, %{request_path: request_path}} -> request_path end)
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.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,
action_input,
input,
transaction_name,
tenant,
wait_for
) do
action =
Ash.Resource.Info.action(resource, action) ||
raise "No such action #{action} for #{resource}"
{action_input, deps} = Ash.Flow.handle_input_template(action_input, input)
{tenant, tenant_deps} = Ash.Flow.handle_input_template(tenant, input)
{_, wait_for_deps} = Ash.Flow.handle_input_template(wait_for, input)
dep_paths = get_dep_paths(all_steps, deps ++ tenant_deps, transaction_name, wait_for_deps)
request_deps = dependable_request_paths(dep_paths)
%{
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: ^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)
%Step{step: %Ash.Flow.Step.Transaction{steps: steps} = step} ->
case find_step_dep(steps, dep, transaction_name, stop_at_transaction_name?) 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?)
_ ->
nil
end)
end
defp result_path(%Ash.Flow.Step.Read{name: name}) do
[name, :data, :data]
end
defp result_path(%Ash.Flow.Step.Create{name: name}) do
[name, :commit, :data]
end
defp result_path(%Ash.Flow.Step.Update{name: name}) do
[name, :commit, :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.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, :data, :completion]
end
defp completion_path(%Ash.Flow.Step.Create{name: name}) do
[name, :commit, :completion]
end
defp completion_path(%Ash.Flow.Step.Update{name: name}) do
[name, :commit, :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.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, :data]
end
defp data_path(%Ash.Flow.Step.Create{name: name}) do
[name, :commit]
end
defp data_path(%Ash.Flow.Step.Update{name: name}) do
[name, :commit]
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.Map{name: name}) do
[name]
end
end