Packages
ash
3.1.0
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/actions/destroy/bulk.ex
defmodule Ash.Actions.Destroy.Bulk do
@moduledoc false
require Ash.Query
@spec run(Ash.Domain.t(), Enumerable.t() | Ash.Query.t(), atom(), input :: map, Keyword.t()) ::
Ash.BulkResult.t()
def run(domain, resource, action, input, opts, not_atomic_reason \\ nil)
def run(domain, stream, action, input, opts, not_atomic_reason)
when is_atom(action) and not is_nil(action) do
resource = opts[:resource]
opts =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
case Ash.Helpers.get_action(resource, opts, :destroy, action) do
{:ok, action} ->
run(
domain,
stream,
action,
input,
opts,
not_atomic_reason
)
{:error, error} ->
%Ash.BulkResult{status: :error, errors: [Ash.Error.to_error_class(error)]}
end
end
def run(domain, stream, nil, input, opts, not_atomic_reason) do
resource = opts[:resource]
run(
domain,
stream,
Ash.Resource.Info.primary_action!(resource, :destroy),
input,
opts,
not_atomic_reason
)
end
def run(domain, stream, %{soft?: true} = action, input, opts, not_atomic_reason) do
Ash.Actions.Update.Bulk.run(domain, stream, action, input, opts, not_atomic_reason)
end
def run(domain, resource, action, input, opts, not_atomic_reason) when is_atom(resource) do
run(domain, Ash.Query.new(resource), action, input, opts, not_atomic_reason)
end
def run(domain, %Ash.Query{} = query, action, input, opts, not_atomic_reason) do
opts = set_strategy(opts, query.resource)
opts = select(opts, query.resource)
opts =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
query =
if query.__validated_for_action__ do
query
else
query =
Ash.Query.for_read(
query,
Ash.Resource.Info.primary_action!(query.resource, :read).name,
%{},
actor: opts[:actor],
tenant: opts[:tenant]
)
{query, _opts} = Ash.Actions.Helpers.set_context_and_get_opts(domain, query, opts)
query
end
query = %{query | domain: domain}
fully_atomic_changeset =
cond do
not_atomic_reason ->
{:not_atomic, not_atomic_reason}
:atomic not in opts[:strategy] ->
{:not_atomic, "Not in requested strategies"}
changeset = opts[:atomic_changeset] ->
changeset
Ash.DataLayer.data_layer_can?(query.resource, :destroy_query) ->
private_context = Map.new(Keyword.take(opts, [:actor, :tenant, :authorize]))
opts =
Keyword.update(
opts,
:context,
%{private: private_context},
&Map.put(&1, :private, private_context)
)
Ash.Changeset.fully_atomic_changeset(query.resource, action, input, opts)
true ->
{:not_atomic, "data layer does not support destroying a query"}
end
case fully_atomic_changeset do
{:not_atomic, reason} ->
case Ash.Actions.Read.Stream.stream_strategy(
query,
nil,
opts[:allow_stream_with] || :keyset
) do
{:error, %Ash.Error.Invalid.NonStreamableAction{} = exception} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [
Ash.Error.to_error_class(
Ash.Error.Invalid.NoMatchingBulkStrategy.exception(
resource: query.resource,
action: query.action.name,
requested_strategies: opts[:strategy],
not_stream_reason: "could not stream the query",
footer: "Non stream reason:\n\n" <> Exception.message(exception)
)
)
]
}
_ ->
read_opts =
opts
|> then(fn read_opts ->
if opts[:batch_size] do
Keyword.put(read_opts, :batch_size, opts[:stream_batch_size])
else
read_opts
end
end)
|> Keyword.put(:authorize?, opts[:authorize?] && opts[:authorize_query?])
|> Keyword.put(:domain, domain)
|> Keyword.take(Ash.stream_opt_keys())
query =
Ash.Query.do_filter(query, opts[:filter])
run(
domain,
Ash.stream!(
query,
read_opts
),
action,
input,
Keyword.merge(opts, resource: query.resource, input_was_stream?: false),
reason
)
end
%Ash.Changeset{valid?: false, errors: errors} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [Ash.Error.to_error_class(errors)]
}
atomic_changeset ->
{atomic_changeset, opts} =
Ash.Actions.Helpers.set_context_and_get_opts(domain, atomic_changeset, opts)
atomic_changeset = Ash.Actions.Helpers.apply_opts_load(atomic_changeset, opts)
atomic_changeset = %{atomic_changeset | domain: domain}
atomic_changeset =
if opts[:context] do
Ash.Changeset.set_context(atomic_changeset, opts[:context])
else
atomic_changeset
end
notify? =
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
try do
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
has_after_batch_hooks? =
Enum.any?(
action.changes ++ Ash.Resource.Info.changes(atomic_changeset.resource, :destroy),
fn
%{change: {module, change_opts}} ->
function_exported?(module, :after_batch, 3) &&
module.batch_callbacks?(query, change_opts, context)
_ ->
false
end
)
opts =
if has_after_batch_hooks? || !Enum.empty?(atomic_changeset.after_action) ||
opts[:notify?] do
Keyword.put(opts, :return_records?, true)
else
opts
end
context_key =
case atomic_changeset.action.type do
:update ->
:bulk_update
:destroy ->
:bulk_destroy
end
if (has_after_batch_hooks? || !Enum.empty?(atomic_changeset.after_action)) &&
Keyword.get(opts, :transaction, true) do
Ash.DataLayer.transaction(
List.wrap(atomic_changeset.resource) ++ action.touches_resources,
fn ->
do_atomic_destroy(query, atomic_changeset, has_after_batch_hooks?, input, opts)
end,
opts[:timeout],
%{
type: context_key,
metadata: %{
resource: query.resource,
action: atomic_changeset.action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || %{}
}
)
else
{:ok, do_atomic_destroy(query, atomic_changeset, has_after_batch_hooks?, input, opts)}
end
|> case do
{:ok, bulk_result} ->
if opts[:return_notifications?] do
bulk_result
else
if notify? do
notifications =
List.wrap(Process.delete(:ash_notifications)) ++
List.wrap(bulk_result.notifications)
if opts[:notify?] do
remaining_notifications = Ash.Notifier.notify(notifications)
Ash.Actions.Helpers.warn_missed!(atomic_changeset.resource, action, %{
resource_notifications: remaining_notifications
})
%{bulk_result | notifications: notifications}
else
%{bulk_result | notifications: []}
end
else
process_notifications = List.wrap(Process.get(:ash_notifications, []))
Process.put(
:ash_notifications,
process_notifications ++ List.wrap(bulk_result.notifications)
)
%{bulk_result | notifications: []}
end
end
{:error, error} ->
%Ash.BulkResult{
status: :error,
errors: [Ash.Error.to_ash_error(error)],
error_count: 1
}
end
after
if notify? do
Process.delete(:ash_started_transaction)
end
end
end
end
def run(domain, stream, action, input, opts, not_atomic_reason) do
resource = opts[:resource]
opts = select(opts, resource)
opts =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
opts = set_strategy(opts, resource, Keyword.get(opts, :input_was_stream?, true))
not_atomic_reason =
not_atomic_reason ||
if :atomic_batches not in opts[:strategy],
do: "Cannot perform atomic destroys on an enumerable of inputs"
action =
case action do
nil ->
Ash.Resource.Info.primary_action!(resource, :update)
name when is_atom(name) ->
action = Ash.Resource.Info.action(resource, action)
if !action do
raise Ash.Error.Invalid.NoSuchAction, resource: resource, action: name, type: :update
end
action
action ->
action
end
if opts[:transaction] == :all && opts[:return_stream?] do
raise ArgumentError,
"Cannot specify `transaction: :all` and `return_stream?: true` together"
end
if opts[:return_stream?] && opts[:sorted?] do
raise ArgumentError, "Cannot specify `sorted?: true` and `return_stream?: true` together"
end
if opts[:transaction] == :all &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
notify? =
if opts[:notify?] do
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
else
false
end
try do
Ash.DataLayer.transaction(
List.wrap(resource) ++ action.touches_resources,
fn ->
do_run(
domain,
stream,
action,
input,
Keyword.merge(opts, notify?: opts[:notify?], return_notifications?: opts[:notify?]),
not_atomic_reason
)
end,
opts[:timeout],
%{
type: :bulk_destroy,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || %{}
}
)
|> case do
{:ok, bulk_result} ->
bulk_result =
if notify? do
notifications =
if opts[:return_notifications?] do
bulk_result.notifications ++ List.wrap(Process.delete(:ash_notifications))
else
if opts[:notify?] do
remaining_notifications =
Ash.Notifier.notify(
bulk_result.notifications ++
List.wrap(Process.delete(:ash_notifications))
)
Ash.Actions.Helpers.warn_missed!(resource, action, %{
resource_notifications: remaining_notifications
})
else
[]
end
end
%{
bulk_result
| notifications: notifications
}
else
Process.put(
:ash_notifications,
List.wrap(Process.get(:ash_notifications)) ++
List.wrap(bulk_result.notifications)
)
bulk_result
end
handle_bulk_result(bulk_result, resource, action, opts)
{:error, error} ->
[{:error, error}]
end
after
if notify? do
Process.delete(:ash_started_transaction?)
end
end
else
domain
|> do_run(stream, action, input, opts, not_atomic_reason)
|> handle_bulk_result(resource, action, opts)
end
end
defp select(opts, resource) do
if opts[:select] do
opts
else
all_attributes =
resource
|> Ash.Resource.Info.public_attributes()
|> Enum.map(& &1.name)
Keyword.put(opts, :select, all_attributes)
end
end
defp do_atomic_destroy(query, atomic_changeset, has_after_batch_hooks?, input, opts) do
atomic_changeset =
if atomic_changeset.context[:data_layer][:use_atomic_destroy_data?] do
atomic_changeset
else
%{
atomic_changeset
| data: %Ash.Changeset.OriginalDataNotAvailable{reason: :atomic_query_destroy}
}
end
context =
struct(Ash.Resource.Change.Context, %{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
{all_changes, conditional_after_batch_hooks, calculations} =
Ash.Actions.Update.Bulk.hooks_and_calcs_for_update_query(
atomic_changeset,
context,
query,
opts
)
destroy_query_opts =
opts
|> Keyword.take([:return_records?, :tenant, :select])
|> Map.new()
|> Map.put(:calculations, calculations)
with {:ok, query} <-
authorize_bulk_query(query, atomic_changeset, opts),
{:ok, atomic_changeset, query} <-
authorize_atomic_changeset(query, atomic_changeset, opts),
{query, atomic_changeset} <- add_changeset_filters(query, atomic_changeset),
{:ok, data_layer_query} <- Ash.Query.data_layer_query(query) do
case Ash.DataLayer.destroy_query(
data_layer_query,
atomic_changeset,
destroy_query_opts
)
|> Ash.Actions.Helpers.rollback_if_in_transaction(query.resource, nil) do
:ok ->
%Ash.BulkResult{
status: :success
}
{:ok, results} ->
results = List.wrap(results)
{results, notifications} =
if has_after_batch_hooks? do
Ash.Actions.Update.Bulk.run_atomic_after_batch_hooks(
results,
atomic_changeset,
all_changes,
conditional_after_batch_hooks,
context
)
else
{results, []}
end
{results, errors, error_count} =
case load_data(
results,
atomic_changeset.domain,
atomic_changeset.resource,
atomic_changeset,
opts
) do
{:ok, results} ->
{results, [], 0}
{:error, error} ->
{[], List.wrap(error), Enum.count(List.wrap(error))}
end
notifications =
if opts[:notify?] do
notifications ++
Enum.map(results, fn result ->
notification(atomic_changeset, result, opts)
end)
else
notifications
end
status =
case {error_count, results} do
{0, []} ->
:success
{0, _results} ->
:success
{_error_count, []} ->
:error
end
%Ash.BulkResult{
status: status,
error_count: error_count,
notifications: notifications,
errors: errors,
records:
if opts[:return_records?] do
results
else
[]
end
}
{:error, :no_rollback,
%Ash.Error.Forbidden.Placeholder{
authorizer: authorizer
}} ->
error =
Ash.Authorizer.exception(
authorizer,
:forbidden,
query.context[:private][:authorizer_state][authorizer]
)
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [Ash.Error.to_error_class(error)]
}
{:error,
%Ash.Error.Forbidden.Placeholder{
authorizer: authorizer
}} ->
error =
Ash.Authorizer.exception(
authorizer,
:forbidden,
query.context[:private][:authorizer_state][authorizer]
)
if Ash.DataLayer.in_transaction?(atomic_changeset.resource) do
Ash.DataLayer.rollback(atomic_changeset.resource, Ash.Error.to_error_class(error))
else
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [Ash.Error.to_error_class(error)]
}
end
{:error, :no_rollback, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [Ash.Error.to_error_class(error)]
}
{:error, error} ->
if Ash.DataLayer.in_transaction?(atomic_changeset.resource) do
Ash.DataLayer.rollback(atomic_changeset.resource, Ash.Error.to_error_class(error))
else
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [Ash.Error.to_error_class(error)]
}
end
end
else
{:error, %Ash.Error.Forbidden.InitialDataRequired{}} ->
case Ash.Actions.Read.Stream.stream_strategy(
query,
nil,
opts[:allow_stream_with] || :keyset
) do
{:error, %Ash.Error.Invalid.NonStreamableAction{} = exception} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [
Ash.Error.to_error_class(
Ash.Error.Invalid.NoMatchingBulkStrategy.exception(
resource: atomic_changeset.resource,
action: atomic_changeset.action.name,
requested_strategies: opts[:strategy],
not_stream_reason: "could not stream the query, see message below for more",
not_atomic_batches_reason: "authorization requires initial data",
not_atomic_reason: "authorization requires initial data",
footer: "Non stream reason:\n\n" <> Exception.message(exception)
)
)
]
}
_strategy ->
run(
atomic_changeset.domain,
query,
atomic_changeset.action,
input,
Keyword.put(opts, :strategy, [:stream]),
"authorization requires initial data"
)
end
{:error, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [Ash.Error.to_error_class(error)]
}
end
end
defp set_strategy(opts, resource, inputs_is_enumerable? \\ false) do
opts =
if Ash.DataLayer.data_layer_can?(resource, :update_query) do
opts
else
Keyword.put(opts, :strategy, [:stream])
end
if inputs_is_enumerable? && :atomic in List.wrap(opts[:strategy]) do
Keyword.put(opts, :strategy, Enum.uniq([:atomic_batches | List.wrap(opts[:strategy])]))
else
opts
end
end
defp add_changeset_filters(query, changeset) do
{Ash.Query.do_filter(query, changeset.filter), %{changeset | filter: nil}}
end
defp do_run(domain, stream, action, input, opts, not_atomic_reason) do
resource = opts[:resource]
opts = Ash.Actions.Helpers.set_opts(opts, domain)
{context_cs, opts} =
Ash.Actions.Helpers.set_context_and_get_opts(domain, Ash.Changeset.new(resource), opts)
fully_atomic_changeset =
cond do
:atomic_batches not in opts[:strategy] ->
{:not_atomic, "Not in requested strategies"}
Enum.empty?(Ash.Resource.Info.primary_key(resource)) ->
{:not_atomic, "cannot atomically destroy a stream without a primary key"}
!Ash.Resource.Info.primary_action(resource, :read) ->
{:not_atomic, "cannot atomically destroy a stream without a primary read action"}
Ash.DataLayer.data_layer_can?(resource, :destroy_query) ->
opts =
Keyword.update(
opts,
:context,
%{private: context_cs.context.private},
&Map.put(&1, :private, context_cs.context.private)
)
Ash.Changeset.fully_atomic_changeset(resource, action, input, opts)
true ->
{:not_atomic, "data layer does not support destroying a query"}
end
case fully_atomic_changeset do
%Ash.Changeset{} = atomic_changeset ->
query =
resource
|> Ash.Query.new()
|> Map.put(:action, Ash.Resource.Info.primary_action!(resource, :read))
case Ash.Actions.Read.Stream.stream_strategy(
query,
nil,
opts[:allow_stream_with] || :keyset
) do
{:error, exception} ->
if :stream in opts[:strategy] do
do_stream_batches(domain, stream, action, input, opts)
else
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [
Ash.Error.to_error_class(
Ash.Error.Invalid.NoMatchingBulkStrategy.exception(
resource: resource,
action: action.name,
requested_strategies: opts[:strategy],
not_atomic_batches_reason: "could not stream the query",
not_atomic_reason: not_atomic_reason,
not_stream_reason: "could not stream the query",
footer: "Non stream reason:\n\n" <> Exception.message(exception)
)
)
]
}
end
_ ->
do_atomic_batches(
atomic_changeset,
domain,
stream,
action,
input,
opts,
not_atomic_reason
)
end
{:not_atomic, not_atomic_batches_reason} ->
if :stream in opts[:strategy] do
do_stream_batches(domain, stream, action, input, opts)
else
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [
Ash.Error.to_error_class(
Ash.Error.Invalid.NoMatchingBulkStrategy.exception(
resource: resource,
action: action.name,
requested_strategies: opts[:strategy],
not_atomic_batches_reason: not_atomic_batches_reason,
not_atomic_reason: not_atomic_reason
)
)
]
}
end
end
end
defp do_atomic_batches(atomic_changeset, domain, stream, action, input, opts, not_atomic_reason) do
batch_size = opts[:batch_size] || 100
resource = opts[:resource]
ref = make_ref()
pkey = Ash.Resource.Info.primary_key(resource)
stream
|> Stream.chunk_every(batch_size)
|> map_batches(
resource,
opts,
ref,
fn batch ->
pkeys = [or: Enum.map(batch, &Map.take(&1, pkey))]
resource
|> Ash.Query.for_read(Ash.Resource.Info.primary_action!(resource, :read).name, %{},
actor: opts[:actor],
authorize?: false,
context: atomic_changeset.context,
tenant: atomic_changeset.tenant,
tracer: opts[:tracer]
)
|> Ash.Query.set_context(%{private: %{internal?: true}})
|> Ash.Query.filter(^pkeys)
|> Ash.Query.select([])
|> then(fn query ->
run(
domain,
query,
action.name,
input,
[
actor: opts[:actor],
authorize_query?: false,
authorize?: opts[:authorize?],
tenant: atomic_changeset.tenant,
tracer: opts[:tracer],
atomic_changeset: atomic_changeset,
return_errors?: opts[:return_errors?],
filter: opts[:filter],
load: opts[:load],
resource: opts[:resource],
return_notifications?: opts[:return_notifications?],
notify?: opts[:notify?],
return_records?: opts[:return_records?],
strategy: [:atomic]
],
not_atomic_reason
)
|> case do
%Ash.BulkResult{error_count: 0, records: records, notifications: notifications} ->
store_notification(ref, notifications, opts)
List.wrap(records)
%Ash.BulkResult{
errors: errors,
notifications: notifications,
error_count: error_count
} ->
store_notification(ref, notifications, opts)
store_error(ref, errors, opts, error_count)
[]
end
end)
end
)
|> run_batches(ref, opts)
end
defp do_stream_batches(domain, stream, action, input, opts) do
resource = opts[:resource]
manual_action_can_bulk? =
case action.manual do
{mod, _opts} ->
function_exported?(mod, :bulk_destroy, 3)
_ ->
false
end
batch_size =
if manual_action_can_bulk? do
opts[:batch_size] || 100
else
1
end
ref = make_ref()
base_changeset = base_changeset(resource, domain, opts, action, input)
all_changes =
pre_template_all_changes(action, resource, action.type, base_changeset, opts[:actor])
argument_names = Enum.map(action.arguments, & &1.name)
stream
|> Stream.with_index()
|> Stream.chunk_every(batch_size)
|> map_batches(
resource,
opts,
ref,
fn batch ->
try do
batch
|> Stream.map(
&setup_changeset(
&1,
action,
opts,
input,
argument_names,
domain
)
)
|> handle_batch(domain, resource, action, all_changes, opts, ref, base_changeset)
after
if opts[:notify?] && !opts[:return_notifications?] do
Ash.Notifier.notify(Process.delete({:bulk_destroy_notifications, ref}))
end
end
end
)
|> run_batches(ref, opts)
end
defp run_batches(changeset_stream, ref, opts) do
if opts[:return_stream?] do
Stream.concat(changeset_stream)
else
try do
records =
if opts[:return_records?] do
Enum.to_list(Stream.concat(changeset_stream))
else
Stream.run(changeset_stream)
[]
end
notifications =
if opts[:notify?] && opts[:return_notifications?] do
Process.delete({:bulk_destroy_notifications, ref})
else
if opts[:notify?] do
Ash.Notifier.notify(Process.delete({:bulk_destroy_notifications, ref}))
end
[]
end
{errors, error_count} = Process.get({:bulk_destroy_errors, ref}) || {[], 0}
bulk_result = %Ash.BulkResult{
records: records,
errors: errors,
notifications: notifications,
error_count: error_count
}
case bulk_result do
%{records: _, error_count: 0} -> %{bulk_result | status: :success}
%{records: [], error_count: _} -> %{bulk_result | status: :error}
_ -> %{bulk_result | status: :partial_success}
end
catch
{:error, error, batch_number} ->
status =
if batch_number > 1 do
:partial_success
else
:error
end
result = %Ash.BulkResult{
status: status,
notifications: Process.delete({:bulk_destroy_notifications, ref})
}
{error_count, errors} = errors(result, error, opts)
%{result | errors: errors, error_count: error_count}
after
Process.delete({:bulk_destroy_errors, ref})
Process.delete({:bulk_destroy_notifications, ref})
end
end
end
defp base_changeset(resource, domain, opts, action, input) do
arguments =
Enum.reduce(input, %{}, fn {key, value}, acc ->
argument =
if is_binary(key) do
Enum.find(action.arguments, fn arg ->
to_string(arg.name) == key
end)
else
Enum.find(action.arguments, fn arg ->
arg.name == key
end)
end
if argument do
Map.put(acc, argument.name, value)
else
acc
end
end)
resource
|> Ash.Changeset.new()
|> Ash.Changeset.filter(opts[:filter])
|> Map.put(:domain, domain)
|> Ash.Actions.Helpers.add_context(opts)
|> Ash.Changeset.set_context(opts[:context] || %{})
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
|> Ash.Changeset.set_arguments(arguments)
end
defp authorize_bulk_query(query, atomic_changeset, opts) do
if opts[:authorize?] && opts[:authorize_query?] do
case Ash.can(query, opts[:actor],
return_forbidden_error?: true,
pre_flight?: false,
atomic_changeset: atomic_changeset,
filter_with: opts[:authorize_query_with] || :filter,
run_queries?: false,
maybe_is: false,
alter_source?: true
) do
{:ok, true} ->
{:ok, query}
{:ok, true, query} ->
{:ok, query}
{:ok, false, error} ->
{:error, error}
{:error, error} ->
{:error, error}
end
else
{:ok, query}
end
end
defp authorize_atomic_changeset(query, changeset, opts) do
if opts[:authorize?] do
case Ash.can(changeset, opts[:actor],
return_forbidden_error?: true,
pre_flight?: false,
maybe_is: false,
atomic_changeset: changeset,
no_check?: true,
on_must_pass_strict_check:
{:error, %Ash.Error.Forbidden.InitialDataRequired{source: changeset}},
run_queries?: false,
filter_with: opts[:authorize_changeset_with] || :filter,
alter_source?: true,
base_query: query
) do
{:ok, true} ->
{:ok, changeset, query}
{:ok, true, %Ash.Query{} = query} ->
{:ok, changeset, query}
{:ok, true, %Ash.Changeset{} = changeset} ->
{:ok, changeset, query}
{:ok, true, %Ash.Changeset{} = changeset, %Ash.Query{} = query} ->
{:ok, changeset, query}
{:ok, false, error} ->
{:error, error}
{:error, error} ->
{:error, error}
end
else
{:ok, changeset, query}
end
end
defp pre_template_all_changes(action, resource, :destroy, base, actor) do
action.changes
|> Enum.concat(Ash.Resource.Info.validations(resource, action.type))
|> Enum.concat(Ash.Resource.Info.changes(resource, action.type))
|> Enum.map(fn
%{change: {module, opts}} = change ->
%{change | change: {module, pre_template(opts, base, actor)}}
%{validation: {module, opts}} = validation ->
%{validation | validation: {module, pre_template(opts, base, actor)}}
end)
|> Enum.map(fn
%{where: where} = change ->
new_where =
if where do
where
|> List.wrap()
|> Enum.map(fn {module, opts} -> {module, pre_template(opts, base, actor)} end)
end
%{change | where: new_where}
other ->
other
end)
|> Enum.with_index()
end
defp pre_template(opts, changeset, actor) do
if Ash.Expr.template_references_context?(opts) do
opts
else
{:templated,
Ash.Expr.fill_template(
opts,
actor,
changeset.arguments,
changeset.context
)}
end
end
defp error_stream(ref) do
Stream.resource(
fn -> Process.delete({:bulk_destroy_errors, ref}) end,
fn
{errors, _count} ->
{Stream.map(errors || [], &{:error, &1}), []}
_ ->
{:halt, []}
end,
fn _ -> :ok end
)
end
defp notification_stream(ref) do
Stream.resource(
fn -> Process.delete({:bulk_destroy_notifications, ref}) end,
fn
[] ->
{:halt, []}
notifications ->
{Stream.map(notifications || [], &{:notification, &1}), []}
end,
fn _ -> :ok end
)
end
defp handle_batch(batch, domain, resource, action, all_changes, opts, ref, base_changeset) do
%{
must_return_records?: must_return_records_for_changes?,
batch: batch,
changes: changes
} =
run_action_changes(
batch,
all_changes,
action,
opts[:actor],
opts[:authorize?],
opts[:tracer],
opts[:tenant]
)
{batch, must_be_simple} =
Enum.reduce(batch, {[], []}, fn changeset, {batch, must_be_simple} ->
if changeset.after_transaction in [[], nil] do
changeset = Ash.Changeset.run_before_transaction_hooks(changeset)
{[changeset | batch], must_be_simple}
else
{batch, [%{changeset | __validated_for_action__: action.name} | must_be_simple]}
end
end)
must_be_simple_results =
Enum.flat_map(must_be_simple, fn changeset ->
case Ash.Actions.Destroy.run(
domain,
changeset,
action,
Keyword.put(opts, :return_destroyed?, opts[:return_records?])
) do
:ok ->
[]
{:ok, result} when not is_list(result) ->
[
Ash.Resource.set_metadata(result, %{
bulk_destroy_index: changeset.context.bulk_destroy.index
})
]
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
if opts[:transaction] == :batch &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
context = batch |> Enum.at(0) |> Kernel.||(%{}) |> Map.get(:context)
notify? =
if opts[:notify?] do
if Process.get(:ash_started_transaction?) do
false
else
Process.put(:ash_started_transaction?, true)
true
end
else
false
end
try do
Ash.DataLayer.transaction(
List.wrap(resource) ++ action.touches_resources,
fn ->
tmp_ref = make_ref()
result =
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
tmp_ref,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results
)
{new_errors, new_error_count} =
Process.delete({:bulk_destroy_errors, tmp_ref}) || {[], 0}
store_error(ref, new_errors, opts, new_error_count)
notifications = Process.get({:bulk_destroy_notifications, tmp_ref}) || []
store_notification(ref, notifications, opts)
result
end,
opts[:timeout],
%{
type: :bulk_destroy,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || context
}
)
|> case do
{:ok, result} ->
result
{:error, error} ->
[{:error, error}]
end
after
if notify? do
notifications = Process.get(:ash_notifications, [])
remaining_notifications = Ash.Notifier.notify(notifications)
Process.delete(:ash_notifications) || []
Ash.Actions.Helpers.warn_missed!(resource, action, %{
resource_notifications: remaining_notifications
})
Process.delete(:ash_started_transaction?)
end
end
else
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
ref,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results
)
end
end
defp do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
ref,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results
) do
must_return_records? =
opts[:notify?] ||
Enum.any?(batch, fn item ->
item.after_action != []
end)
batch =
batch
|> authorize(opts)
|> Enum.to_list()
|> run_bulk_before_batches(
changes,
all_changes,
opts,
ref
)
changesets_by_index = index_changesets(batch)
run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
domain,
ref
)
|> run_after_action_hooks(opts, domain, ref, changesets_by_index)
|> process_results(
changes,
all_changes,
opts,
ref,
changesets_by_index,
batch,
domain,
resource,
base_changeset
)
|> Stream.concat(must_be_simple_results)
|> then(fn stream ->
if opts[:return_stream?] do
stream
|> Stream.map(&{:ok, &1})
|> Stream.concat(error_stream(ref))
|> Stream.concat(notification_stream(ref))
else
stream
end
end)
end
defp setup_changeset(
{record, index},
action,
opts,
input,
argument_names,
domain
) do
record
|> Ash.Changeset.new()
|> Map.put(:domain, domain)
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
|> Ash.Changeset.put_context(:bulk_destroy, %{index: index})
|> Ash.Changeset.set_context(opts[:context] || %{})
|> handle_params(
Keyword.get(opts, :assume_casted?, false),
action,
opts,
input,
argument_names
)
end
defp handle_params(changeset, false, action, opts, input, _argument_names) do
Ash.Changeset.handle_params(changeset, action, input, opts)
end
defp handle_params(changeset, true, action, opts, input, argument_names) do
{args, attrs} =
Map.split(input, argument_names)
%{changeset | arguments: args, attributes: attrs}
|> Ash.Changeset.handle_params(action, input, Keyword.put(opts, :cast_params?, false))
end
defp map_batches(stream, resource, opts, ref, callback) do
max_concurrency = opts[:max_concurrency]
max_concurrency =
if max_concurrency && max_concurrency > 1 && Ash.DataLayer.can?(:async_engine, resource) do
max_concurrency
else
0
end
if max_concurrency && max_concurrency > 1 do
Task.async_stream(
stream,
fn batch ->
try do
Process.put(:ash_started_transaction?, true)
batch_result = callback.(batch)
{errors, _} = Process.get({:bulk_destroy_errors, ref}) || {[], 0}
notifications =
if opts[:notify?] do
process_notifications = Process.get(:ash_notifications, [])
bulk_notifications = Process.get({:bulk_destroy_notifications, ref}) || []
if opts[:return_notifications?] do
process_notifications ++ bulk_notifications
else
if opts[:transaction] && opts[:transaction] != :all do
Ash.Notifier.notify(bulk_notifications)
Ash.Notifier.notify(process_notifications)
end
[]
end
end
{batch_result, notifications, errors}
after
Process.delete(:ash_started_transaction?)
end
end,
timeout: :infinity,
max_concurrency: max_concurrency
)
|> Stream.map(fn
{:ok, {:throw, value}} ->
throw(value)
{:ok, {result, notifications, errors}} ->
store_notification(ref, notifications, opts)
store_error(ref, errors, opts)
result
{:exit, error} ->
store_error(ref, error, opts)
[]
end)
else
Stream.map(stream, callback)
end
end
defp index_changesets(batch) do
Enum.reduce(batch, %{}, fn changeset, changesets_by_index ->
Map.put(
changesets_by_index,
changeset.context.bulk_destroy.index,
changeset
)
end)
end
defp errors(result, invalid, opts) when is_list(invalid) do
Enum.reduce(invalid, {result.error_count, result.errors}, fn invalid, {error_count, errors} ->
errors(%{result | error_count: error_count, errors: errors}, invalid, opts)
end)
end
defp errors(result, nil, _opts) do
{result.error_count + 1, []}
end
defp errors(result, {:error, error}, opts) do
if opts[:return_errors?] do
{result.error_count + 1, [error | List.wrap(result.errors)]}
else
{result.error_count + 1, []}
end
end
defp errors(result, invalid, opts) do
if Enumerable.impl_for(invalid) do
invalid = Enum.to_list(invalid)
errors(result, invalid, opts)
else
errors(result, {:error, invalid}, opts)
end
end
defp run_bulk_before_batches(
batch,
changes,
all_changes,
opts,
ref
) do
all_changes
|> Enum.filter(fn
{%{change: {module, _opts}}, _} ->
function_exported?(module, :before_batch, 3)
_ ->
false
end)
|> Enum.reduce(batch, fn {%{change: {module, change_opts}}, index}, batch ->
if changes[index] == :all do
module.before_batch(
batch,
change_opts,
struct(Ash.Resource.Change.Context, %{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
)
else
{matches, non_matches} =
batch
|> Enum.split_with(fn
%{valid?: false} ->
false
changeset ->
changeset.context.bulk_destroy.index in List.wrap(changes[index])
end)
before_batch_results =
module.before_batch(
matches,
change_opts,
struct(Ash.Resource.Change.Context, %{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
)
Enum.concat([before_batch_results, non_matches])
end
end)
|> Enum.reject(fn
%Ash.Notifier.Notification{} = notification ->
store_notification(ref, notification, opts)
true
_changeset ->
false
end)
end
defp store_error(ref, errors, opts, count \\ nil)
defp store_error(ref, empty, _opts, error_count) when empty in [[], nil] do
{errors, count} = Process.get({:bulk_destroy_errors, ref}) || {[], 0}
Process.put({:bulk_destroy_errors, ref}, {errors, count + error_count})
end
defp store_error(ref, error, opts, count) do
add = count || Enum.count(List.wrap(error))
if opts[:stop_on_error?] && !opts[:return_stream?] do
throw({:error, Ash.Error.to_error_class(error), 0, []})
else
if opts[:return_errors?] do
{errors, count} = Process.get({:bulk_destroy_errors, ref}) || {[], 0}
new_errors =
error
|> List.wrap()
|> Enum.map(&Ash.Error.to_ash_error/1)
Process.put(
{:bulk_destroy_errors, ref},
{new_errors ++ errors, count + add}
)
else
{errors, count} = Process.get({:bulk_destroy_errors, ref}) || {[], 0}
Process.put({:bulk_destroy_errors, ref}, {errors, count + add})
end
end
end
defp store_notification(_ref, empty, _opts) when empty in [[], nil], do: :ok
defp store_notification(ref, notification, opts) do
if opts[:notify?] || opts[:return_notifications?] do
notifications = Process.get({:bulk_destroy_notifications, ref}) || []
new_notifications =
if is_list(notification) do
notification ++ List.wrap(notifications)
else
[notification | List.wrap(notifications)]
end
Process.put({:bulk_destroy_notifications, ref}, new_notifications)
end
end
defp authorize(batch, opts) do
if opts[:authorize?] do
batch
|> Enum.map(fn changeset ->
if changeset.valid? do
case Ash.can(
changeset,
opts[:actor],
return_forbidden_error?: true,
pre_flight?: false,
maybe_is: false
) do
{:ok, true} ->
changeset
{:ok, false, error} ->
Ash.Changeset.add_error(changeset, error)
{:error, error} ->
Ash.Changeset.add_error(changeset, error)
end
else
changeset
end
end)
else
batch
end
end
defp handle_bulk_result(%Ash.BulkResult{} = bulk_result, _resource, _action, opts) do
bulk_result
|> sort(opts)
|> ensure_records_return_type(opts)
|> ensure_errors_return_type(opts)
end
# for when we return a stream
defp handle_bulk_result(stream, _, _, _), do: stream
defp ensure_records_return_type(result, opts) do
if opts[:return_records?] do
%{result | records: result.records || []}
else
%{result | records: nil}
end
end
defp ensure_errors_return_type(result, opts) do
if opts[:return_errors?] do
%{result | errors: result.errors || []}
else
%{result | errors: nil}
end
end
defp sort(%{records: records} = result, opts) when is_list(records) do
if opts[:sorted?] do
%{result | records: Enum.sort_by(records, & &1.__metadata__.bulk_destroy_index)}
else
result
end
end
defp sort(result, _), do: result
defp run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
domain,
ref
) do
batch
|> Enum.map(fn changeset ->
if changeset.valid? do
{changeset, %{notifications: new_notifications}} =
Ash.Changeset.run_before_actions(changeset)
new_notifications = store_notification(ref, new_notifications, opts)
{changeset, manage_notifications} =
if changeset.valid? do
case Ash.Actions.ManagedRelationships.setup_managed_belongs_to_relationships(
changeset,
opts[:actor],
authorize?: opts[:authorize?],
actor: opts[:actor],
tenant: opts[:tenant]
) do
{:error, error} ->
{Ash.Changeset.add_error(changeset, error), new_notifications}
{changeset, manage_instructions} ->
{changeset, manage_instructions.notifications}
end
else
{changeset, []}
end
store_notification(ref, manage_notifications, opts)
changeset
else
changeset
end
end)
|> Enum.reject(fn
%{valid?: false} = changeset ->
store_error(ref, changeset, opts)
true
_changeset ->
false
end)
|> case do
[] ->
[]
batch ->
batch
|> Enum.group_by(&{&1.atomics, &1.filter})
|> Enum.flat_map(fn {_atomics, batch} ->
result =
case action.manual do
{mod, manual_opts} ->
if function_exported?(mod, :bulk_destroy, 3) do
mod.bulk_destroy(batch, manual_opts, %{
actor: opts[:actor],
select: opts[:select],
batch_size: opts[:batch_size],
authorize?: opts[:authorize?],
tracer: opts[:tracer],
domain: domain,
return_records?:
opts[:return_records?] || must_return_records? ||
must_return_records_for_changes?,
tenant: Ash.ToTenant.to_tenant(opts[:tenant], resource)
})
|> Enum.flat_map(fn
{:ok, result} ->
[result]
{:error, error} ->
store_error(ref, error, opts)
[]
{:notifications, notifications} ->
store_notification(ref, notifications, opts)
[]
end)
else
[changeset] = batch
result =
mod.destroy(changeset, opts, %Ash.Resource.ManualDestroy.Context{
select: opts[:select],
actor: opts[:actor],
tenant: opts[:tenant],
authorize?: opts[:authorize?],
tracer: opts[:tracer],
domain: domain
})
case result do
{:ok, result} ->
{:ok,
[
Ash.Resource.put_metadata(
result,
:bulk_destroy_index,
changeset.context.bulk_destroy.index
)
]}
{:error, error} ->
{:error, error}
end
end
_ ->
[changeset] = batch
result =
resource
|> Ash.DataLayer.destroy(changeset)
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
case result do
:ok ->
{:ok, data} = Ash.Changeset.apply_attributes(changeset, force?: true)
{:ok,
[
data
|> Ash.Resource.set_meta(%Ecto.Schema.Metadata{
state: :deleted,
schema: changeset.resource
})
|> Ash.Resource.put_metadata(
:bulk_destroy_index,
changeset.context.bulk_destroy.index
)
]}
{:error, error} ->
{:error, error}
end
end
case result do
{:error, %Ash.Error.Changes.StaleRecord{}} ->
[]
{:ok, result} ->
result
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
end
end
defp manage_relationships(destroyed, domain, changeset, engine_opts) do
with {:ok, loaded} <-
Ash.Actions.ManagedRelationships.load(domain, destroyed, changeset, engine_opts),
{:ok, with_relationships, new_notifications} <-
Ash.Actions.ManagedRelationships.manage_relationships(
loaded,
changeset,
engine_opts[:actor],
engine_opts
) do
{:ok, with_relationships, %{notifications: new_notifications, new_changeset: changeset}}
end
end
defp run_after_action_hooks(
batch_results,
opts,
domain,
ref,
changesets_by_index
) do
Enum.flat_map(batch_results, fn result ->
changeset = changesets_by_index[result.__metadata__.bulk_destroy_index]
case manage_relationships(result, domain, changeset,
actor: opts[:actor],
authorize?: opts[:authorize?]
) do
{:ok, result, %{notifications: new_notifications, new_changeset: changeset}} ->
store_notification(ref, new_notifications, opts)
case Ash.Changeset.run_after_actions(result, changeset, []) do
{:error, error} ->
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(changeset.resource) do
Ash.DataLayer.rollback(
changeset.resource,
error
)
end
end
store_error(ref, error, opts)
[]
{:ok, result, _changeset, %{notifications: more_new_notifications}} ->
store_notification(ref, more_new_notifications, opts)
[result]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
end
defp process_results(
batch,
changes,
all_changes,
opts,
ref,
changesets_by_index,
changesets,
domain,
resource,
base_changeset
) do
changes
|> run_bulk_after_changes(
all_changes,
batch,
changesets_by_index,
changesets,
opts,
resource,
ref
)
|> Enum.flat_map(fn result ->
changeset = changesets_by_index[result.__metadata__[:bulk_destroy_index]]
if opts[:notify?] || opts[:return_notifications?] do
store_notification(ref, notification(changeset, result, opts), opts)
end
try do
case Ash.Changeset.run_after_transactions(
{:ok, result},
changeset
) do
{:ok, result} ->
if opts[:return_records?] do
[result]
else
[]
end
{:error, error} ->
store_error(ref, error, opts)
[]
end
rescue
e ->
store_error(ref, e, opts)
[]
end
end)
|> load_data(domain, resource, base_changeset, opts)
|> case do
{:ok, records} ->
Enum.reject(records, & &1.__metadata__[:private][:missing_from_data_layer])
{:error, error} ->
store_error(ref, error, opts)
[]
end
end
defp load_data(records, domain, resource, changeset, opts) do
select =
if changeset.select do
List.wrap(changeset.select)
else
resource |> Ash.Resource.Info.public_attributes() |> Enum.map(& &1.name)
end
case Ash.load(records, select,
reuse_values?: true,
domain: domain,
tenant: opts[:tenant],
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
) do
{:ok, records} ->
Ash.load(
records,
List.wrap(changeset.load),
reuse_values?: true,
tenant: opts[:tenant],
domain: domain,
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
)
|> Ash.Actions.Helpers.select(changeset)
other ->
other
end
end
defp run_bulk_after_changes(
changes,
all_changes,
results,
changesets_by_index,
changesets,
opts,
resource,
ref
) do
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
all_changes
|> Enum.filter(fn
{%{change: {module, change_opts}}, _} ->
function_exported?(module, :after_batch, 3) &&
function_exported?(module, :batch_change, 3) &&
module.batch_callbacks?(changesets, change_opts, context)
_ ->
false
end)
|> Enum.reduce(results, fn {%{change: {module, change_opts}}, index}, results ->
if changes[index] == :all do
records = results
results =
Enum.map(results, fn result ->
{changesets_by_index[result.__metadata__.bulk_destroy_index], result}
end)
module.after_batch(
results,
change_opts,
context
)
|> handle_after_batch_results(records, ref, resource, opts)
else
{matches, non_matches} =
results
|> Enum.split_with(fn
{:ok, result} ->
result.__metadata__.bulk_destroy_index in List.wrap(changes[index])
_ ->
false
end)
match_records = matches
matches =
Enum.map(matches, fn match ->
{changesets_by_index[match.__metadata__.bulk_destroy_index], match}
end)
after_batch_results =
module.after_batch(
matches,
change_opts,
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
)
|> handle_after_batch_results(match_records, ref, resource, opts)
Enum.concat([after_batch_results, non_matches])
end
end)
end
defp handle_after_batch_results(:ok, match_records, _ref, _resource, _options) do
match_records
end
defp handle_after_batch_results(results, _match_records, ref, resource, opts) do
Enum.flat_map(
results,
fn
%Ash.Notifier.Notification{} = notification ->
store_notification(ref, notification, opts)
{:ok, result} ->
[result]
{:error, error} ->
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(resource) do
Ash.DataLayer.rollback(
resource,
error
)
end
end
store_error(ref, error, opts)
[]
end
)
end
defp notification(changeset, result, opts) do
%Ash.Notifier.Notification{
resource: changeset.resource,
domain: changeset.domain,
actor: opts[:actor],
action: changeset.action,
for: Ash.Resource.Info.notifiers(changeset.resource) ++ changeset.action.notifiers,
data: result,
changeset: changeset
}
end
defp run_action_changes(batch, all_changes, _action, actor, authorize?, tracer, tenant) do
context = %{
actor: actor,
authorize?: authorize? || false,
tracer: tracer,
tenant: tenant
}
Enum.reduce(
all_changes,
%{must_return_records?: false, batch: batch, changes: %{}},
fn
{%{validation: {module, opts}} = validation, _change_index}, %{batch: batch} = state ->
batch =
Enum.map(batch, fn changeset ->
if Enum.all?(validation.where || [], fn {module, opts} ->
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end) do
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
case module.validate(
changeset,
opts,
struct(
Ash.Resource.Validation.Context,
Map.put(context, :message, validation.message)
)
) do
:ok ->
changeset
{:error, error} ->
Ash.Changeset.add_error(changeset, error)
end
else
changeset
end
end)
%{
state
| must_return_records?: state.must_return_records?,
batch: batch,
changes: state.changes
}
{%{change: {module, change_opts}} = change, change_index}, %{batch: batch} = state ->
# could track if any element in the batch is invalid, and if not use the fast version
if Enum.empty?(change.where) && !change.only_when_valid? do
batch = batch_change(module, batch, change_opts, context, actor)
must_return_records? =
state.must_return_records? ||
Enum.any?(batch, fn item ->
item.relationships not in [nil, %{}] || !Enum.empty?(item.after_action)
end) ||
(function_exported?(module, :after_batch, 3) &&
function_exported?(module, :batch_change, 3) &&
module.batch_callbacks?(batch, change_opts, context))
%{
state
| must_return_records?: must_return_records?,
batch: batch,
changes: Map.put(state.changes, change_index, :all)
}
else
{matches, non_matches} =
batch
|> Enum.split_with(fn changeset ->
applies_from_where? =
Enum.all?(change.where || [], fn {module, opts} ->
opts = templated_opts(opts, actor, changeset.arguments, changeset.context)
{:ok, opts} = module.init(opts)
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end)
applies_from_only_when_valid? =
if change.only_when_valid? do
changeset.valid?
else
true
end
applies_from_where? and applies_from_only_when_valid?
end)
if Enum.empty?(matches) do
%{
state
| must_return_records?: state.must_return_records?,
batch: non_matches,
changes: state.changes
}
else
matches = batch_change(module, matches, change_opts, context, actor)
must_return_records? =
state.must_return_records? ||
Enum.any?(batch, fn item ->
item.relationships not in [nil, %{}] || !Enum.empty?(item.after_action)
end) ||
(function_exported?(module, :after_batch, 3) &&
function_exported?(module, :batch_change, 3) &&
module.batch_callbacks?(batch, change_opts, context))
%{
state
| must_return_records?: must_return_records?,
batch: Enum.concat(matches, non_matches),
changes:
Map.put(
state.changes,
change_index,
Enum.map(matches, & &1.context.bulk_destroy.index)
)
}
end
end
end
)
end
defp batch_change(module, batch, change_opts, context, actor) do
case change_opts do
{:templated, change_opts} ->
if function_exported?(module, :batch_change, 4) do
module.batch_change(
batch,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true),
actor
)
else
Enum.map(batch, fn changeset ->
{:ok, change_opts} = module.init(change_opts)
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
change_opts ->
Enum.map(batch, fn changeset ->
change_opts = templated_opts(change_opts, actor, changeset.arguments, changeset.context)
{:ok, change_opts} = module.init(change_opts)
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
end
|> Enum.to_list()
end
defp templated_opts({:templated, opts}, _actor, _arguments, _context), do: opts
defp templated_opts(opts, actor, arguments, context) do
Ash.Expr.fill_template(
opts,
actor,
arguments,
context
)
end
end