Packages
ash
3.5.26
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/update/bulk.ex
defmodule Ash.Actions.Update.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, 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 =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
{query, opts} =
if query.__validated_for_action__ do
{query, opts}
else
{query, opts} = Ash.Actions.Helpers.set_context_and_get_opts(domain, query, opts)
query =
Ash.Query.for_read(
query,
get_read_action(query.resource, action, opts).name,
%{},
actor: opts[:actor],
tenant: opts[:tenant],
context: %{query_for: :bulk_update}
)
{query, opts}
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"}
query.action.manual ->
{:not_atomic, "Manual read actions cannot be updated atomically"}
!Enum.empty?(query.before_action) ->
{:not_atomic, "cannot atomically update a query if it has `before_action` hooks"}
!Enum.empty?(query.after_action) ->
{:not_atomic, "cannot atomically update a query if it has `after_action` hooks"}
changeset = opts[:atomic_changeset] ->
changeset
Ash.DataLayer.data_layer_can?(query.resource, :update_query) ->
private_context = Map.new(Keyword.take(opts, [:actor, :tenant, :authorize]))
opts =
Keyword.update(
opts,
:context,
%{private: private_context},
fn context ->
Map.update(context, :private, private_context, &Map.merge(&1, private_context))
end
)
Ash.Changeset.fully_atomic_changeset(query.resource, action, input, opts)
true ->
{:not_atomic, "data layer does not support updating 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: action,
requested_strategies: opts[:strategy],
not_atomic_reason: reason,
not_stream_reason: "could not stream the query",
footer: "Non stream reason:\n\n" <> Exception.message(exception)
)
)
]
}
_ ->
query =
Ash.Query.do_filter(query, opts[:filter])
read_opts =
opts
|> then(fn read_opts ->
if opts[:stream_batch_size] do
Keyword.put(read_opts, :batch_size, opts[:stream_batch_size])
else
Keyword.delete(read_opts, :batch_size)
end
end)
|> Keyword.put(:authorize?, opts[:authorize?] && opts[:authorize_query?])
|> Keyword.update(
:context,
%{query_for: :bulk_update},
&Map.put(&1, :query_for, :bulk_update)
)
|> Keyword.put(:domain, domain)
|> Keyword.delete(:load)
if query.limit && query.limit < (opts[:batch_size] || 100) do
read_opts = Keyword.take(read_opts, Keyword.keys(Ash.read_opts()))
case Ash.Actions.Read.unpaginated_read(query, query.action, read_opts) do
{:ok, results} ->
run(
domain,
results,
action,
input,
Keyword.merge(
opts,
[
resource: query.resource,
input_was_stream?: false,
context:
opts[:atomic_changeset] &&
Map.delete(opts[:atomic_changeset].context, :private)
],
fn
:context, v1, v2 ->
Ash.Helpers.deep_merge_maps(v1, v2)
_k, _v1, v2 ->
v2
end
),
reason
)
{:error, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [Ash.Error.to_error_class(error)]
}
end
else
read_opts =
Keyword.put(
Keyword.take(read_opts, Ash.stream_opt_keys()),
:batch_size,
opts[:batch_size] || 100
)
# We need to figure out a way to capture errors raised by the stream when picking items off somehow
# for now, we only go this route if there are potentially more records in the result set than
# in the batch size, to solve this problem for atomic upgrades.
# we can likely make the stream throw something instead of raising something
# like `{:stream_error, ...}` if a specific option is passed in.
# once we figure this out, we may be able to remove the branch above
run(
domain,
Ash.stream!(
query,
read_opts
),
action,
input,
Keyword.merge(
opts,
[
resource: query.resource,
input_was_stream?: false,
context:
opts[:atomic_changeset] &&
Map.delete(opts[:atomic_changeset].context, :private)
],
fn
:context, v1, v2 ->
Ash.Helpers.deep_merge_maps(v1, v2)
_k, _v1, v2 ->
v2
end
),
reason
)
end
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 =
Ash.Changeset.filter(atomic_changeset, opts[:filter])
{atomic_changeset, opts} =
Ash.Actions.Helpers.set_context_and_get_opts(domain, atomic_changeset, opts)
atomic_changeset = %{atomic_changeset | domain: domain}
notify? = !Process.put(:ash_started_transaction?, true)
try do
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
source_context: atomic_changeset.context,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
has_after_batch_hooks? =
Enum.any?(
atomic_changeset.action.changes ++
Ash.Resource.Info.changes(atomic_changeset.resource, atomic_changeset.action_type),
fn
%{change: {module, change_opts}} ->
module.has_after_batch?() &&
module.batch_callbacks?(query, change_opts, context)
_ ->
false
end
)
context_key =
case atomic_changeset.action.type do
:update ->
:bulk_update
:destroy ->
:bulk_destroy
end
prefer_transaction? =
has_after_batch_hooks? || !Enum.empty?(atomic_changeset.after_action) ||
Ash.DataLayer.prefer_transaction_for_atomic_updates?(atomic_changeset.resource)
if prefer_transaction? &&
Keyword.get(opts, :transaction, true) do
Ash.DataLayer.in_transaction?(atomic_changeset.resource)
Ash.DataLayer.transaction(
atomic_changeset.resource,
fn ->
do_atomic_update(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_update(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
rescue
e ->
action_name =
case action do
%{name: name} -> name
name -> name
end
reraise Ash.Error.to_error_class(e,
stacktrace: __STACKTRACE__,
bread_crumbs: [
"Exception raised in bulk update: #{inspect(opts[:resource])}.#{action_name}"
]
),
__STACKTRACE__
end
def run(domain, stream, action, input, opts, not_atomic_reason) do
resource = opts[:resource]
opts = set_strategy(opts, resource, Keyword.get(opts, :input_was_stream?, true))
opts =
if opts[:return_notifications?] do
Keyword.put(opts, :notify?, true)
else
opts
end
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
{context_key, metadata_key} =
case action.type do
:update ->
{:bulk_update, :bulk_update_index}
:destroy ->
{:bulk_destroy, :bulk_destroy_index}
end
if opts[:transaction] == :all &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
notify? = !Process.put(:ash_started_transaction?, true)
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?]),
metadata_key,
context_key,
not_atomic_reason
)
end,
opts[:timeout],
%{
type: context_key,
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, metadata_key, opts)
{:error, error} ->
handle_bulk_result(
%Ash.BulkResult{errors: [error], status: :error},
metadata_key,
opts
)
end
after
if notify? do
Process.delete(:ash_started_transaction?)
end
end
else
domain
|> do_run(stream, action, input, opts, metadata_key, context_key, not_atomic_reason)
|> handle_bulk_result(metadata_key, opts)
end
rescue
e ->
action_name =
case action do
%{name: name} -> name
name -> name
end
reraise Ash.Error.to_error_class(e,
stacktrace: __STACKTRACE__,
bread_crumbs: [
"Exception raised in bulk update: #{inspect(opts[:resource])}.#{action_name}"
]
),
__STACKTRACE__
end
defp do_atomic_update(query, atomic_changeset, has_after_batch_hooks?, input, opts) do
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
source_context: atomic_changeset.context,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
atomic_changeset = Ash.Actions.Helpers.apply_opts_load(atomic_changeset, opts)
atomic_changeset =
if opts[:select] do
Ash.Changeset.select(atomic_changeset, opts[:select])
else
atomic_changeset
end
atomic_changeset =
if atomic_changeset.context[:data_layer][:use_atomic_update_data?] do
atomic_changeset
else
%{
atomic_changeset
| data: %Ash.Changeset.OriginalDataNotAvailable{reason: :atomic_query_update}
}
end
{all_changes, conditional_after_batch_hooks, calculations} =
hooks_and_calcs_for_update_query(atomic_changeset, context, query, opts)
action_select =
if Ash.DataLayer.data_layer_can?(atomic_changeset.resource, :action_select) do
Enum.uniq(
Enum.concat(
Ash.Resource.Info.action_select(atomic_changeset.resource, atomic_changeset.action),
List.wrap(
opts[:select] ||
MapSet.to_list(
Ash.Resource.Info.selected_by_default_attribute_names(atomic_changeset.resource)
)
)
)
)
else
MapSet.to_list(Ash.Resource.Info.attribute_names(atomic_changeset.resource))
end
update_query_opts =
opts
|> Keyword.take([:tenant])
|> Map.new()
|> Map.put(
:return_records?,
has_after_batch_hooks? || opts[:notify?] || opts[:return_records?] ||
!Enum.empty?(atomic_changeset.after_action)
)
|> Map.put(:calculations, calculations)
|> Map.put(
:action_select,
action_select
)
with :ok <- validate_multitenancy(atomic_changeset.resource, opts),
{: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),
query =
Ash.Actions.Read.add_calc_context_to_query(
query,
opts[:actor],
opts[:authorize?],
query.tenant,
opts[:tracer],
query.domain,
expand?: true,
source_context: query.context
),
%Ash.Changeset{valid?: true} = atomic_changeset <-
Ash.Changeset.handle_allow_nil_atomics(atomic_changeset, opts[:actor]),
atomic_changeset <- sort_atomic_changes(atomic_changeset),
{:ok, query} <- Ash.Actions.Read.handle_multitenancy(query),
{:ok, data_layer_query} <-
Ash.Query.data_layer_query(query) do
update_query_result =
if atomic_changeset.context.changed? do
Ash.DataLayer.update_query(
data_layer_query,
atomic_changeset,
update_query_opts
)
else
atomic_validation_expr =
with [field | _] <- Ash.Resource.Info.primary_key(atomic_changeset.resource),
{:ok, v} <- Keyword.fetch(atomic_changeset.atomics, field) do
v
else
_ ->
nil
end
if is_nil(atomic_validation_expr) && is_nil(query.filter) &&
!update_query_opts[:return_records?] do
:ok
else
Ash.DataLayer.update_query(
data_layer_query,
atomic_changeset,
update_query_opts
)
end
end
case update_query_result do
:ok ->
%Ash.BulkResult{
status: :success
}
{:ok, results} ->
results =
Ash.Actions.Helpers.select(results, %{
resource: query.resource,
select: action_select
})
results =
case results do
[result] ->
if atomic_changeset.context[:data_layer][:use_atomic_update_data?] do
Map.put(result, :__metadata__, atomic_changeset.data.__metadata__)
else
[result]
end
other ->
other
end
results = List.wrap(results)
{results, notifications} =
if has_after_batch_hooks? do
run_atomic_after_batch_hooks(
results,
atomic_changeset,
all_changes,
conditional_after_batch_hooks,
context
)
else
{results, []}
end
{results, errors, error_count, notifications} =
if Enum.empty?(atomic_changeset.after_action) do
{results, [], 0, notifications}
else
Enum.reduce(
results,
{[], [], 0, notifications},
fn result, {results, errors, error_count, notifications} ->
# we can't actually know if the changeset changed or not when doing atomics
# so we just have to set it to statically true here.
case Ash.Changeset.run_after_actions(result, atomic_changeset, []) do
{:error, error} ->
if opts[:transaction] && opts[:rollback_on_error?] do
if Ash.DataLayer.in_transaction?(atomic_changeset.resource) do
Ash.DataLayer.rollback(
atomic_changeset.resource,
error
)
end
end
{results, errors ++ List.wrap(error),
error_count + Enum.count(List.wrap(error)), notifications}
{:ok, result, _changeset, %{notifications: more_new_notifications}} ->
{[result | results], errors, error_count,
notifications ++ more_new_notifications}
end
end
)
|> then(fn {results, errors, error_count, notifications} ->
{Enum.reverse(results), errors, error_count, notifications}
end)
end
{results, errors, error_count} =
case load_data(
results,
atomic_changeset.domain,
atomic_changeset.resource,
atomic_changeset,
opts
) do
{:ok, results} ->
{results, errors, error_count}
{:error, error} ->
{[], errors ++ List.wrap(error), error_count + 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,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
{: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,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
)
else
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [
Ash.Error.to_error_class(error,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
end
{:error, :no_rollback, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [
Ash.Error.to_error_class(error,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
{: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,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
end
end
else
%Ash.Changeset{valid?: false, errors: 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,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
end
{:error, %Ash.Error.Forbidden.InitialDataRequired{} = e} ->
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 ->
input_stream =
if atomic_changeset.context[:data_layer][:use_atomic_update_data?] do
[atomic_changeset.data]
else
query
end
run(
atomic_changeset.domain,
input_stream,
atomic_changeset.action,
input,
Keyword.merge(opts,
authorize_query?: false,
input_was_stream?: false
),
e.source
)
end
{:error, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
notifications: [],
errors: [
Ash.Error.to_error_class(error,
bread_crumbs: [
"Returned from bulk query update: #{inspect(atomic_changeset.resource)}.#{atomic_changeset.action.name}"
]
)
]
}
end
end
defp validate_multitenancy(resource, opts) do
if Ash.Resource.Info.multitenancy_strategy(resource) &&
!Ash.Resource.Info.multitenancy_global?(resource) && !opts[:tenant] do
{:error, Ash.Error.Invalid.TenantRequired.exception(resource: resource)}
else
:ok
end
end
defp sort_atomic_changes(atomic_changeset) do
primary_key = Ash.Resource.Info.primary_key(atomic_changeset.resource)
Map.update!(atomic_changeset, :atomics, fn atomics ->
Enum.sort_by(atomics, fn {key, _} ->
key not in primary_key
end)
end)
end
@doc false
def run_atomic_after_batch_hooks(
results,
atomic_changeset,
all_changes,
conditional_after_batch_hooks,
context
) do
results =
results
|> Stream.with_index()
|> Enum.map(fn {result, index} ->
{atomic_changeset, Ash.Resource.set_metadata(result, %{atomic_index: index})}
end)
{records_with_changesets, notifications} =
conditional_after_batch_hooks
|> Enum.reduce(
{results, []},
fn {where, indices}, {records, additional_notifications} ->
{to_apply, to_not_apply} =
case where do
true ->
{records, []}
false ->
{[], records}
where ->
Enum.split_with(records, fn {_, record} ->
record.calculations[{:run_after_batch, where}]
end)
end
if Enum.empty?(to_apply) do
{records, additional_notifications}
else
{applied, notifications} =
Enum.reduce(
indices,
{to_apply, additional_notifications},
fn index, {records, additional_notifications} ->
change = Enum.at(all_changes, index)
{module, opts} = change.change
case module.after_batch(records, opts, context) do
:ok ->
{to_apply, additional_notifications}
instructions ->
Enum.reduce(
instructions,
{[], additional_notifications},
fn
{:ok, record}, {records, additional_notifications} ->
{[{atomic_changeset, record} | records], additional_notifications}
{:error, error}, _ ->
raise Ash.Error.to_ash_error(error)
%Ash.Notifier.Notification{} = notification,
{records, additional_notifications} ->
{records, [notification | additional_notifications]}
end
)
end
end
)
{applied ++ to_not_apply, notifications ++ additional_notifications}
end
end
)
final_results =
records_with_changesets
# would be nice to have `Enum.map_sort_by`
|> Stream.map(&elem(&1, 1))
|> Enum.sort_by(& &1.__metadata__.atomic_index)
{final_results, notifications}
end
@doc false
def hooks_and_calcs_for_update_query(atomic_changeset, context, query, opts) do
all_changes =
atomic_changeset.action.changes
|> Enum.concat(
Ash.Resource.Info.changes(atomic_changeset.resource, atomic_changeset.action_type)
)
conditional_after_batch_hooks =
all_changes
|> Stream.with_index()
|> Enum.filter(fn
{%{change: {module, change_opts}}, _index} ->
module.has_after_batch?() &&
module.batch_callbacks?(query, change_opts, context)
_ ->
false
end)
|> Enum.reduce(%{}, fn {%{where: where}, index}, acc ->
{:atomic, condition} =
Ash.Changeset.atomic_condition(where, atomic_changeset, context)
Map.update(acc, condition, [index], &[index | &1])
end)
calculations =
Enum.flat_map(conditional_after_batch_hooks, fn
{static, _indices} when static in [true, false] ->
[]
{where, _indices} ->
{:ok, calculation} =
Ash.Query.Calculation.new(
{:run_after_batch, where},
Ash.Resource.Calculation.Expression,
[expression: where],
:boolean,
[],
arguments: atomic_changeset.arguments,
source_context: atomic_changeset.context
)
calculation =
Ash.Actions.Read.add_calc_context(
calculation,
opts[:actor],
opts[:authorize?],
opts[:tenant],
opts[:tracer],
atomic_changeset.domain,
atomic_changeset.resource,
source_context: atomic_changeset.context
)
[{calculation, where}]
end)
{all_changes, conditional_after_batch_hooks, calculations}
end
defp set_strategy(opts, resource, inputs_is_enumerable? \\ false) do
opts =
if Ash.DataLayer.data_layer_can?(resource, :update_query) do
if Ash.DataLayer.data_layer_can?(resource, :expr_error) do
opts
else
Keyword.put_new(opts, :strategy, [:stream, :atomic_batches, :atomic])
end
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 do_run(domain, stream, action, input, opts, metadata_key, context_key, not_atomic_reason) do
resource = opts[:resource]
opts = Ash.Actions.Helpers.set_opts(opts, domain)
read_action = get_read_action(resource, action, opts)
{context_cs, opts} =
Ash.Actions.Helpers.set_context_and_get_opts(domain, Ash.Changeset.new(resource), opts)
case validate_multitenancy(resource, opts) do
{:error, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [
Ash.Error.to_error_class(error)
]
}
:ok ->
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 update a stream without a primary key"}
!read_action ->
{:not_atomic, "cannot atomically update a stream without a primary read action"}
read_action.manual ->
{:not_atomic, "Manual read actions cannot be updated atomically"}
Ash.DataLayer.data_layer_can?(resource, :update_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 updating a query"}
end
case fully_atomic_changeset do
%Ash.Changeset{} = atomic_changeset ->
query =
resource
|> Ash.Query.new()
|> Map.put(:action, read_action)
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,
metadata_key,
context_key
)
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
)
end
{:not_atomic, not_atomic_batches_reason} ->
if :stream in opts[:strategy] do
do_stream_batches(domain, stream, action, input, opts, metadata_key, context_key)
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
end
defp do_atomic_batches(
atomic_changeset,
domain,
stream,
action,
input,
opts
) 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))]
read_action = get_read_action(resource, action, opts).name
resource
|> Ash.Query.for_read(read_action, %{},
actor: opts[:actor],
authorize?: false,
context: Map.put(atomic_changeset.context, :query_for, :bulk_update),
tenant: atomic_changeset.tenant,
tracer: opts[:tracer]
)
|> Ash.Query.set_context(%{private: %{internal?: true}})
|> Ash.Query.filter(^pkeys)
|> Ash.Query.filter(^atomic_changeset.filter)
|> 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,
context: opts[:context] || %{},
tracer: opts[:tracer],
atomic_changeset: atomic_changeset,
load: opts[:load],
resource: opts[:resource],
input_was_stream?: false,
return_errors?: opts[:return_errors?],
filter: opts[:filter],
return_notifications?: opts[:return_notifications?],
notify?: opts[:notify?],
return_records?: opts[:return_records?],
allow_stream_with: opts[:allow_stream_with],
read_action: read_action,
strategy: [:atomic]
)
|> case do
%Ash.BulkResult{
error_count: 0,
records: records,
notifications: notifications,
status: status
} ->
Process.put({:any_success?, ref}, status != :error)
store_notification(ref, notifications, opts)
List.wrap(records)
%Ash.BulkResult{
errors: errors,
notifications: notifications,
error_count: error_count,
status: status
} ->
Process.put({:any_success?, ref}, status != :error)
store_notification(ref, notifications, opts)
store_error(ref, errors, opts, error_count)
[]
end
end)
end
)
|> run_batches(ref, atomic_changeset.resource, atomic_changeset.action.name, opts)
end
defp do_stream_batches(domain, stream, action, input, opts, metadata_key, context_key) do
resource = opts[:resource]
action_select =
Enum.uniq(
Enum.concat(
Ash.Resource.Info.action_select(resource, action.name),
List.wrap(
opts[:select] ||
MapSet.to_list(Ash.Resource.Info.selected_by_default_attribute_names(resource))
)
)
)
manual_action_can_bulk? =
case action.manual do
{mod, _opts} ->
function_exported?(mod, :bulk_update, 3)
_ ->
false
end
batch_size =
if action.manual == nil || 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],
base_changeset.to_tenant
)
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
|> Enum.map(
&setup_changeset(
&1,
action,
opts,
input,
argument_names,
domain,
context_key
)
)
|> handle_batch(
domain,
resource,
action,
all_changes,
opts,
ref,
context_key,
metadata_key,
base_changeset,
action_select
)
after
if opts[:notify?] && !opts[:return_notifications?] do
Process.put(
{:bulk_notifications, ref},
Ash.Notifier.notify(Process.delete({:bulk_notifications, ref}) || [])
)
end
end
end
)
|> run_batches(ref, resource, action.name, opts)
end
defp run_batches(changeset_stream, ref, resource, action_name, 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)
nil
end
notifications =
if opts[:notify?] do
Process.delete({:bulk_notifications, ref}) || []
else
[]
end
# Unless return_notifications? is true, we always want to end up with nil.
notifications =
cond do
Enum.empty?(notifications) ->
if opts[:return_notifications?], do: [], else: nil
opts[:return_notifications?] ->
notifications
true ->
notifications =
if opts[:notify?] do
Ash.Notifier.notify(notifications)
else
notifications
end
if Process.get(:ash_started_transaction?) do
Process.put(
:ash_notifications,
Process.get(:ash_notifications, []) ++ notifications
)
end
nil
end
{errors, error_count} = Process.get({:bulk_errors, ref}) || {[], 0}
errors =
if opts[:return_errors?] do
Enum.map(
errors,
&Ash.Error.to_ash_error(&1, [],
bread_crumbs: [
"Returned from bulk update: #{inspect(resource)}.#{action_name}"
]
)
)
else
nil
end
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_notifications, ref})
}
{error_count, errors} = errors(result, error, opts)
%{result | errors: errors, error_count: error_count}
after
Process.delete({:bulk_errors, ref})
Process.delete({:bulk_notifications, ref})
end
end
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,
maybe_is: false,
pre_flight?: false,
atomic_changeset: atomic_changeset,
filter_with: opts[:authorize_query_with] || opts[:authorize_with] || :filter,
run_queries?: false,
alter_source?: true,
no_check?: 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,
maybe_is: false,
atomic_changeset: changeset,
pre_flight?: false,
no_check?: true,
on_must_pass_strict_check:
{:error,
%Ash.Error.Forbidden.InitialDataRequired{source: "must pass strict check"}},
filter_with: opts[:authorize_changeset_with] || opts[:authorize_with] || :filter,
alter_source?: true,
run_queries?: false,
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 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()
|> Map.put(:domain, domain)
|> Ash.Changeset.filter(opts[:filter])
|> Ash.Actions.Helpers.add_context(opts)
|> Ash.Changeset.set_context(opts[:context] || %{})
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
|> Ash.Changeset.set_private_arguments_for_action(opts[:private_arguments] || %{})
|> Ash.Changeset.set_arguments(arguments)
|> then(fn changeset ->
changeset =
if select = opts[:select] do
Ash.Changeset.select(changeset, select)
else
changeset
end
if load = opts[:load] do
Ash.Changeset.load(changeset, load)
else
changeset
end
end)
end
defp pre_template_all_changes(action, resource, _type, base, actor, tenant) do
action.changes
|> then(fn changes ->
if action.skip_global_validations? do
changes
else
Enum.concat(changes, Ash.Resource.Info.validations(resource, action.type))
end
end)
|> Enum.concat(Ash.Resource.Info.changes(resource, action.type))
|> Enum.map(fn
%{change: {module, opts}} = change ->
%{change | change: {module, pre_template(opts, base, actor, tenant)}}
%{validation: {module, opts}} = validation ->
%{validation | validation: {module, pre_template(opts, base, actor, tenant)}}
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, tenant)}
end)
end
%{change | where: new_where}
other ->
other
end)
|> Enum.with_index()
end
defp add_changeset_filters(query, changeset) do
{Ash.Query.do_filter(query, changeset.filter), %{changeset | filter: nil}}
end
defp pre_template(opts, changeset, actor, tenant) do
if Ash.Expr.template_references_context?(opts) do
opts
else
{:templated,
Ash.Expr.fill_template(
opts,
actor: actor,
tenant: tenant,
args: changeset.arguments,
context: changeset.context,
changeset: changeset
)}
end
end
defp error_stream(ref) do
the_errors = Process.delete({:bulk_errors, ref})
Stream.resource(
fn -> the_errors end,
fn
{errors, _count} ->
{Stream.map(errors || [], &{:error, &1}), []}
_ ->
{:halt, []}
end,
fn _ -> :ok end
)
end
defp notification_stream(ref) do
the_notifications = Process.delete({:bulk_notifications, ref})
Stream.resource(
fn -> the_notifications 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,
context_key,
metadata_key,
base_changeset,
action_select
) do
%{
must_return_records?: must_return_records_for_changes?,
batch: batch,
re_sort?: re_sort?,
changes: changes
} =
run_action_changes(
batch,
all_changes,
action,
resource,
opts[:actor],
opts[:authorize?],
opts[:tracer],
opts[:tenant],
context_key
)
batch =
if re_sort? do
Enum.sort_by(batch, & &1.context[context_key].index)
else
batch
end
{batch, must_be_simple_results} =
Ash.Actions.Helpers.split_and_run_simple(
batch,
action,
opts,
changes,
all_changes,
context_key,
fn changeset ->
case Ash.Actions.Update.run(
domain,
changeset,
action,
Keyword.merge(opts,
atomic_upgrade?: false,
return_destroyed?: opts[:return_records?]
)
) do
:ok ->
Process.put({:any_success?, ref}, true)
[]
{:ok, result} ->
Process.put({:any_success?, ref}, true)
[
Ash.Resource.set_metadata(result, %{
metadata_key => changeset.context |> Map.get(context_key) |> Map.get(:index)
})
]
{:ok, result, notifications} ->
Process.put({:any_success?, ref}, true)
store_notification(ref, notifications, opts)
[
Ash.Resource.set_metadata(result, %{
metadata_key => changeset.context |> Map.get(context_key) |> Map.get(:index)
})
]
{:error, error} ->
store_error(ref, error, opts)
[]
end
end
)
batch =
Enum.reject(batch, fn
%{valid?: false} = changeset ->
store_error(ref, changeset, opts)
true
_changeset ->
false
end)
if opts[:transaction] == :batch &&
Ash.DataLayer.data_layer_can?(resource, :transact) do
context = batch |> Enum.at(0) |> Kernel.||(%{}) |> Map.get(:context)
notify? = opts[:notify?] && !Process.put(:ash_started_transaction?, true)
try do
Ash.DataLayer.transaction(
List.wrap(resource) ++ action.touches_resources,
fn ->
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
ref,
metadata_key,
context_key,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results,
action_select
)
end,
opts[:timeout],
%{
type: context_key,
metadata: %{
resource: resource,
action: action.name,
actor: opts[:actor]
},
data_layer_context: opts[:data_layer_context] || context
}
)
|> case do
{:ok, result} ->
result
{:error, error} ->
store_error(ref, error, opts)
[]
end
after
if notify? do
Process.delete(:ash_started_transaction?)
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
})
end
end
else
do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
ref,
metadata_key,
context_key,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results,
action_select
)
end
end
defp do_handle_batch(
batch,
domain,
resource,
action,
opts,
all_changes,
ref,
metadata_key,
context_key,
base_changeset,
must_return_records_for_changes?,
changes,
must_be_simple_results,
action_select
) do
must_return_records? =
opts[:notify?] ||
Enum.any?(batch, fn item ->
item.after_action != []
end)
batch =
batch
|> authorize(opts)
|> run_bulk_before_batches(
changes,
all_changes,
opts,
ref,
context_key
)
run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
domain,
ref,
metadata_key,
context_key,
action_select
)
|> run_after_action_hooks(opts, domain, ref, metadata_key)
|> process_results(
changes,
all_changes,
opts,
ref,
batch,
metadata_key,
resource,
domain,
base_changeset
)
|> then(fn stream ->
if opts[:return_stream?] do
stream
|> Stream.concat(must_be_simple_results)
|> Stream.map(&{:ok, &1})
|> Stream.concat(error_stream(ref))
|> Stream.concat(notification_stream(ref))
else
Enum.concat(stream, must_be_simple_results)
end
end)
end
defp setup_changeset(
{record, index},
action,
opts,
input,
argument_names,
domain,
context_key
) do
record
|> Ash.Changeset.new()
|> Map.put(:domain, domain)
|> Ash.Changeset.prepare_changeset_for_action(action, opts)
|> Ash.Changeset.set_private_arguments_for_action(opts[:private_arguments] || %{})
|> Ash.Changeset.put_context(context_key, %{index: index})
|> Ash.Changeset.set_context(opts[:context] || %{})
|> Ash.Changeset.atomic_update(opts[:atomic_update] || [])
|> Ash.Changeset.hydrate_atomic_refs(opts[:actor], opts)
|> handle_params(
Keyword.get(opts, :assume_casted?, false),
action,
opts,
input,
argument_names
)
|> Ash.Changeset.apply_atomic_constraints(opts[:actor])
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
ash_context = Ash.ProcessHelpers.get_context_for_transfer(opts)
Task.async_stream(
stream,
fn batch ->
Ash.ProcessHelpers.transfer_context(ash_context, opts)
Process.put(:ash_started_transaction?, true)
batch_result = callback.(batch)
{errors, _} = Process.get({:bulk_errors, ref}) || {[], 0}
notifications =
if opts[:notify?] do
process_notifications = Process.get(:ash_notifications, [])
bulk_notifications = Process.get({:bulk_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, Process.get({:any_success?, ref})}
end,
timeout: :infinity,
max_concurrency: max_concurrency
)
|> Stream.map(fn
{:ok, {:throw, value}} ->
throw(value)
{:ok,
{%Ash.BulkResult{
records: records,
notifications: notifications,
errors: errors,
error_count: error_count
}, _, _, any_success?}} ->
Process.put({:any_success?, ref}, any_success?)
store_notification(ref, notifications, opts)
store_error(ref, errors, opts, error_count)
records
{:ok, {result, notifications, errors, any_success?}} ->
Process.put({:any_success?, ref}, any_success?)
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, context_key) do
Enum.reduce(batch, %{}, fn changeset, changesets_by_index ->
Map.put(
changesets_by_index,
changeset.context[context_key].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
@doc false
def run_bulk_before_batches(
[],
_changes,
_all_changes,
_opts,
_ref,
_context_key
) do
[]
end
def run_bulk_before_batches(
batch,
changes,
all_changes,
opts,
ref,
context_key
) do
all_changes
|> Enum.reduce(batch, fn
{%{validation: _}, _index}, batch ->
batch
{%{change: {module, change_opts}}, index}, batch ->
# change may return a stream but before_batch/3 expects a list
batch = Enum.to_list(batch)
context =
struct(Ash.Resource.Change.Context, %{
bulk?: true,
source_context: Enum.at(batch, 0).context,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
})
if changes[index] == :all do
case change_opts do
{:templated, change_opts} ->
if module.has_before_batch?() && module.has_batch_change?() &&
module.batch_callbacks?(batch, change_opts, context) do
module.before_batch(
batch,
change_opts,
context
)
else
batch
end
change_opts ->
Enum.flat_map(batch, fn changeset ->
if module.has_before_batch?() && module.has_batch_change?() &&
module.batch_callbacks?(batch, change_opts, context) do
change_opts =
templated_opts(
change_opts,
opts[:actor],
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
module.before_batch(
[changeset],
change_opts,
context
)
else
[changeset]
end
end)
end
else
{matches, non_matches} =
batch
|> Enum.split_with(fn
%{valid?: false} ->
false
changeset ->
changeset.context[context_key].index in List.wrap(changes[index])
end)
before_batch_results =
case change_opts do
{:templated, change_opts} ->
if module.has_before_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(matches, change_opts, context) do
module.before_batch(
matches,
change_opts,
context
)
else
matches
end
change_opts ->
Enum.flat_map(matches, fn changeset ->
change_opts =
templated_opts(
change_opts,
opts[:actor],
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
if module.has_before_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?([changeset], change_opts, context) do
module.before_batch(
[changeset],
change_opts,
context
)
else
[changeset]
end
end)
end
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, 0) when empty in [[], nil], do: :ok
defp store_error(ref, empty, _opts, error_count) when empty in [[], nil] do
{errors, count} = Process.get({:bulk_errors, ref}) || {[], 0}
Process.put(
{:bulk_errors, ref},
{errors, count + (error_count || 1)}
)
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_errors, ref}) || {[], 0}
new_errors =
error
|> List.wrap()
|> Enum.map(&Ash.Error.to_ash_error/1)
Process.put(
{:bulk_errors, ref},
{new_errors ++ errors, count + add}
)
else
{errors, count} = Process.get({:bulk_errors, ref}) || {[], 0}
Process.put({:bulk_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_notifications, ref}) || []
new_notifications =
if is_list(notification) do
notification ++ notifications
else
[notification | notifications]
end
Process.put({:bulk_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,
maybe_is: false,
pre_flight?: 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, metadata_key, opts) do
bulk_result
|> sort(metadata_key, 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, metadata_key, opts) when is_list(records) do
if opts[:sorted?] do
%{result | records: Enum.sort_by(records, & &1.__metadata__[metadata_key])}
else
result
end
end
defp sort(result, _action, _), do: result
defp run_batch(
resource,
batch,
action,
opts,
must_return_records?,
must_return_records_for_changes?,
domain,
ref,
metadata_key,
context_key,
action_select
) do
batch =
Enum.map(batch, fn changeset ->
changeset =
changeset
|> Ash.Changeset.hydrate_atomic_refs(opts[:actor])
|> Ash.Changeset.apply_atomic_constraints(opts[:actor])
if changeset.valid? do
{changeset, %{notifications: new_notifications}} =
Ash.Changeset.run_before_actions(changeset)
if !changeset.valid? && opts[:rollback_on_error?] do
Ash.Actions.Helpers.rollback_if_in_transaction(
{:error, changeset.errors},
resource,
nil
)
end
changeset =
changeset
|> Ash.Changeset.handle_allow_nil_atomics(opts[:actor])
|> Ash.Changeset.require_values(
:update,
true
)
|> Map.put(:action_select, action_select)
changed? =
Ash.Changeset.changing_attributes?(changeset) or
not Enum.empty?(changeset.atomics)
changeset =
Ash.Changeset.put_context(
changeset,
:changed?,
changeset.context[:changed?] || changed?
)
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)
changesets_by_index = index_changesets(batch, context_key)
batch =
batch
|> 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} ->
source_context =
case batch do
[cs | _] -> cs.context
[] -> %{}
end
if function_exported?(mod, :bulk_update, 3) do
mod.bulk_update(batch, manual_opts, %Ash.Resource.ManualUpdate.BulkContext{
actor: opts[:actor],
select: opts[:select],
source_context: source_context,
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?,
return_notifications?: opts[:return_notifications?] || false,
return_errors?: opts[:return_errors?] || false,
tenant: Ash.ToTenant.to_tenant(opts[:tenant], resource)
})
else
ctx = %Ash.Resource.ManualUpdate.Context{
actor: opts[:actor],
source_context: source_context,
select: opts[:select],
authorize?: opts[:authorize?],
tracer: opts[:tracer],
domain: domain,
return_notifications?: opts[:return_notifications?] || false,
tenant: opts[:tenant]
}
[changeset] = batch
[
mod.update(changeset, manual_opts, ctx)
|> Ash.Actions.BulkManualActionHelpers.process_non_bulk_result(
changeset,
:bulk_update
)
]
end
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
|> Ash.Actions.BulkManualActionHelpers.process_bulk_results(
mod,
:bulk_update,
&store_notification/3,
&store_error/3,
ref,
opts
)
_ ->
Enum.reduce_while(
batch,
{:ok, []},
fn changeset, {:ok, results} ->
changed? =
Ash.Changeset.changing_attributes?(changeset) or
not Enum.empty?(changeset.atomics)
if changed? do
changeset =
if Enum.empty?(changeset.attributes) &&
Ash.DataLayer.data_layer_can?(changeset.resource, :atomic_update) do
Ash.Changeset.atomic_defaults(changeset)
else
Ash.Changeset.set_defaults(changeset, :update, true)
end
|> Ash.Changeset.set_action_select()
resource
|> Ash.DataLayer.update(changeset)
|> Ash.Actions.Helpers.rollback_if_in_transaction(resource, nil)
|> case do
{:ok, result} ->
result =
Ash.Resource.put_metadata(
result,
metadata_key,
changeset.context[context_key].index
)
{:cont, {:ok, [result | results]}}
{:error, %Ash.Error.Changes.StaleRecord{}} ->
{:cont, {:ok, results}}
{:error, error} ->
{:halt, {:error, error}}
end
else
result =
Ash.Resource.put_metadata(
changeset.data,
metadata_key,
changeset.context[context_key].index
)
{:cont, {:ok, [result | results]}}
end
end
)
|> case do
{:ok, results} ->
{:ok, Enum.reverse(results)}
{:error, error} ->
{:error, error}
end
end
case result do
{:ok, result} ->
Process.put({:any_success?, ref}, true)
result
{:error, error} ->
store_error(ref, error, opts)
[]
end
end)
end
{batch, changesets_by_index}
end
defp manage_relationships(updated, domain, changeset, engine_opts) do
with {:ok, loaded} <-
Ash.Actions.ManagedRelationships.load(domain, updated, 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, changesets_by_index},
opts,
domain,
ref,
metadata_key
) do
results =
Enum.flat_map(batch_results, fn result ->
changeset = changesets_by_index[result.__metadata__[metadata_key]]
case manage_relationships(result, domain, changeset,
actor: opts[:actor],
tenant: opts[:tenant],
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)
{results, changesets_by_index}
end
defp process_results(
{batch, changesets_by_index},
changes,
all_changes,
opts,
ref,
changesets,
metadata_key,
resource,
domain,
base_changeset
) do
run_bulk_after_changes(
changes,
all_changes,
batch,
changesets_by_index,
changesets,
opts,
ref,
base_changeset.resource,
metadata_key
)
|> Enum.flat_map(fn result ->
changeset = changesets_by_index[result.__metadata__[metadata_key]]
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],
action:
Ash.Resource.Info.primary_action(changeset.resource, :read) || changeset.action,
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],
action: Ash.Resource.Info.primary_action(changeset.resource, :read) || changeset.action,
domain: domain,
actor: opts[:actor],
authorize?: opts[:authorize?],
tracer: opts[:tracer]
)
|> Ash.Actions.Helpers.select(changeset)
other ->
other
end
end
def run_bulk_after_changes(
changes,
all_changes,
results,
changesets_by_index,
changesets,
opts,
ref,
resource,
metadata_key
) do
source_context =
case changesets do
[cs | _] -> cs.context
_ -> %{}
end
context =
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
actor: opts[:actor],
source_context: source_context,
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
all_changes
|> Enum.reduce(results, fn
{%{validation: _}, _index}, results ->
results
{%{change: {module, change_opts}}, index}, results ->
records = results
if changes[index] == :all do
results =
Enum.map(results, fn result ->
{changesets_by_index[result.__metadata__[metadata_key]], result}
end)
case change_opts do
{:templated, change_opts} ->
if module.has_after_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(changesets, change_opts, context) do
module.after_batch(
results,
change_opts,
context
)
else
Enum.map(results, &{:ok, elem(&1, 1)})
end
change_opts ->
Enum.flat_map(results, fn {changeset, record} = result ->
change_opts =
templated_opts(
change_opts,
opts[:actor],
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
if module.has_after_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(changesets, change_opts, context) do
module.after_batch(
[result],
change_opts,
context
)
else
[{:ok, record}]
end
end)
end
|> handle_after_batch_results(records, ref, resource, opts)
else
{matches, non_matches} =
results
|> Enum.split_with(fn
{:ok, result} ->
result.__metadata__[metadata_key] in List.wrap(changes[index])
_ ->
false
end)
match_records = matches
matches =
Enum.map(matches, fn match ->
{changesets_by_index[match.__metadata__[metadata_key]], match}
end)
after_batch_results =
case change_opts do
{:templated, change_opts} ->
if module.has_after_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(changesets, change_opts, context) do
module.after_batch(
matches,
change_opts,
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
source_context: source_context,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
)
else
Enum.map(matches, &{:ok, elem(&1, 1)})
end
change_opts ->
Enum.flat_map(matches, fn {changeset, record} = result ->
if module.has_after_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(changesets, change_opts, context) do
change_opts =
templated_opts(
change_opts,
opts[:actor],
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
module.after_batch(
[result],
change_opts,
struct(
Ash.Resource.Change.Context,
%{
bulk?: true,
source_context: source_context,
actor: opts[:actor],
tenant: opts[:tenant],
tracer: opts[:tracer],
authorize?: opts[:authorize?]
}
)
)
else
[{:ok, record}]
end
end)
end
|> 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, matches, _ref, _resource, _options), do: matches
defp handle_after_batch_results(results, _matches, 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],
for: Ash.Resource.Info.notifiers(changeset.resource) ++ changeset.action.notifiers,
action: changeset.action,
data: result,
changeset: changeset
}
end
def run_action_changes(
batch,
all_changes,
action,
resource,
actor,
authorize?,
tracer,
tenant,
context_key
) do
source_context =
case batch do
[cs | _] ->
cs.context
_ ->
%{}
end
context = %{
actor: actor,
source_context: source_context,
authorize?: authorize? || false,
tracer: tracer,
tenant: tenant
}
Enum.reduce(
all_changes,
%{must_return_records?: false, re_sort?: false, batch: batch, changes: %{}},
fn
{%{validation: {module, _opts}} = validation, _change_index}, %{batch: batch} = state ->
validation_context =
struct(
Ash.Resource.Validation.Context,
Map.put(context, :message, validation.message)
)
batch =
if module.has_validate?() &&
Enum.all?(validation.where, fn {module, _opts} ->
module.has_validate?()
end) do
validate_batch_non_atomically(validation, batch, validation_context, actor)
else
if module.atomic?() do
validate_batch_atomically(validation, batch, validation_context, context, actor)
else
raise """
Cannot use a non-atomic validation with an atomic condition.
Attempting to run action: `#{inspect(resource)}.#{action.name}`
"""
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 ->
{matches, non_matches} =
if Enum.empty?(change.where) && !change.only_when_valid? do
{batch, []}
else
Enum.split_with(batch, fn changeset ->
Enum.all?(change.where || [], fn {module, opts} ->
if module.has_validate?() do
opts =
Ash.Actions.Helpers.templated_opts(
opts,
actor,
changeset.tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
if change.only_when_valid? do
changeset.valid? &&
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
else
module.validate(
changeset,
opts,
struct(Ash.Resource.Validation.Context, context)
) == :ok
end
else
true
end
end)
end)
end
if Enum.empty?(matches) do
%{
state
| must_return_records?: state.must_return_records?,
batch: Enum.concat(matches, non_matches),
changes: state.changes
}
else
matches = batch_change(change, matches, 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) ||
(module.has_after_batch?() &&
module.has_batch_change?() &&
module.batch_callbacks?(batch, change_opts, context))
match_indices =
if Enum.empty?(non_matches) do
:all
else
Enum.map(matches, & &1.context[context_key].index)
end
%{
state
| must_return_records?: must_return_records?,
batch: Enum.concat(matches, non_matches),
re_sort?: true,
changes:
Map.put(
state.changes,
change_index,
match_indices
)
}
end
end
)
end
defp validate_batch_non_atomically(
%{validation: {module, opts}} = validation,
batch,
validation_context,
actor
) do
Enum.map(batch, fn changeset ->
if Enum.all?(validation.where || [], fn {module, opts} ->
opts =
Ash.Actions.Helpers.templated_opts(
opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
module.validate(
changeset,
opts,
validation_context
) == :ok
end) do
opts =
Ash.Actions.Helpers.templated_opts(
opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, opts} = module.init(opts)
case module.validate(
changeset,
opts,
validation_context
) do
:ok ->
changeset
{:error, error} ->
if validation.message do
error = Ash.Changeset.override_validation_message(error, validation.message)
Ash.Changeset.add_error(changeset, error)
else
Ash.Changeset.add_error(changeset, error)
end
end
else
changeset
end
end)
end
defp validate_batch_atomically(validation, batch, validation_context, context, actor) do
Enum.map(batch, fn changeset ->
validation = %{
validation
| opts:
templated_opts(
validation.opts,
actor,
changeset.tenant,
changeset.arguments,
changeset.context,
changeset
)
}
case Ash.Changeset.split_atomic_conditions(
validation,
changeset,
actor,
validation_context
) do
{:ok, validation_with_remaining_conditions} ->
Ash.Changeset.run_atomic_validation(
changeset,
validation_with_remaining_conditions,
context
)
:skip ->
changeset
end
end)
end
def batch_change(%{change: {module, change_opts}, where: where} = change, batch, context, actor) do
must_be_atomic? =
Enum.any?(change.where, fn {mod, _} ->
!mod.has_validate?()
end)
case change_opts do
{:templated, change_opts} ->
cond do
!must_be_atomic? && module.has_batch_change?() &&
module.batch_callbacks?(batch, change_opts, context) ->
{:ok, change_opts} = module.init(change_opts)
module.batch_change(batch, change_opts, context)
!must_be_atomic? && module.has_change?() ->
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)
module.atomic?() ->
Enum.map(batch, fn changeset ->
change = %{
change
| where:
Enum.filter(where, fn {module, _opts} ->
module.atomic?()
end)
|> Enum.map(fn {module, change_opts} ->
{module,
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)}
end),
change:
{module,
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)}
}
case Ash.Changeset.run_atomic_change(changeset, change, context) do
{:not_atomic, reason} ->
Ash.Changeset.add_error(
changeset,
"Could not be made atomic: #{inspect(reason)}"
)
changeset ->
changeset
end
end)
true ->
if must_be_atomic? do
raise "#{inspect(module)} cannot be paired with atomic conditions, as it does not implement `atomic/3`"
else
raise "#{inspect(module)} must define at least one of `atomic/3`, `change/3` or `batch_change/3`"
end
end
change_opts ->
cond do
!must_be_atomic? && module.has_batch_change?() ->
Enum.flat_map(batch, fn changeset ->
if module.batch_callbacks?(batch, change_opts, context) do
change_opts =
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, change_opts} = module.init(change_opts)
module.batch_change(batch, change_opts, context)
else
[
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
]
end
end)
!must_be_atomic? && module.has_change?() ->
Enum.map(batch, fn changeset ->
change_opts =
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)
{:ok, change_opts} = module.init(change_opts)
module.change(
changeset,
change_opts,
struct(struct(Ash.Resource.Change.Context, context), bulk?: true)
)
end)
module.atomic?() ->
Enum.map(batch, fn changeset ->
change = %{
change
| where:
Enum.filter(where, fn {module, _opts} ->
module.atomic?()
end),
change:
{module,
templated_opts(
change_opts,
actor,
changeset.to_tenant,
changeset.arguments,
changeset.context,
changeset
)}
}
case Ash.Changeset.run_atomic_change(changeset, change, context) do
{:not_atomic, reason} ->
Ash.Changeset.add_error(
changeset,
"Could not be made atomic: #{inspect(reason)}"
)
changeset ->
changeset
end
end)
true ->
if must_be_atomic? do
raise "#{inspect(module)} cannot be paired with atomic conditions, as it does not implement `atomic/3`"
else
raise "#{inspect(module)} must define at least one of `atomic/3`, `change/3` or `batch_change/3`"
end
end
end
end
defp templated_opts({:templated, opts}, _actor, _tenant, _arguments, _context, _changeset),
do: opts
defp templated_opts(opts, actor, tenant, arguments, context, changeset) do
Ash.Expr.fill_template(
opts,
actor: actor,
tenant: tenant,
args: arguments,
context: context,
changeset: changeset
)
end
defp get_read_action(resource, action, opts) do
case opts[:read_action] do
nil ->
case action do
%{atomic_upgrade_with: read_action} when not is_nil(read_action) ->
Ash.Resource.Info.action(resource, read_action) ||
raise "No such read action in #{inspect(resource)}.#{action.name}.atomic_upgrade_with: #{read_action}"
action when is_atom(action) and not is_nil(action) ->
action =
Ash.Resource.Info.action(resource, action) ||
raise "No such update action #{inspect(resource)}.#{action}"
if read_action = action.atomic_upgrade_with do
Ash.Resource.Info.action(resource, read_action) ||
raise "No such read action in #{inspect(resource)}.#{action.name}.atomic_upgrade_with: #{read_action}"
else
Ash.Resource.Info.primary_action!(resource, :read)
end
_ ->
Ash.Resource.Info.primary_action!(resource, :read)
end
action ->
Ash.Resource.Info.action(resource, action)
end
end
end