fix: ensure notifications are dispatched from bulk actions

This commit is contained in:
Zach Daniel 2024-06-28 17:10:29 -04:00
parent 66ebb1422e
commit 659a061b16
6 changed files with 116 additions and 137 deletions

View file

@ -196,16 +196,12 @@ defmodule Ash.Actions.Destroy.Bulk do
end
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
context =
@ -250,8 +246,8 @@ defmodule Ash.Actions.Destroy.Bulk do
:bulk_destroy
end
bulk_result =
if has_after_batch_hooks? && opts[:transaction] do
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 ->
@ -268,38 +264,20 @@ defmodule Ash.Actions.Destroy.Bulk do
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} ->
bulk_result
{:error, error} ->
%Ash.BulkResult{
status: :error,
error_count: 1,
errors: [Ash.Error.to_error_class(error)]
}
end
else
do_atomic_destroy(query, atomic_changeset, has_after_batch_hooks?, input, opts)
end
notifications =
if notify? do
List.wrap(bulk_result.notifications) ++
List.wrap(Process.delete(:ash_notifications))
else
List.wrap(bulk_result.notifications)
end
if opts[:return_notifications?] do
%{bulk_result | notifications: notifications}
bulk_result
else
if notify? do
notifications =
(bulk_result.notifications || []) ++ Process.get(:ash_notifications, [])
List.wrap(Process.delete(:ash_notifications)) ++ bulk_result.notifications
if opts[:notify?] do
remaining_notifications = Ash.Notifier.notify(notifications)
Process.delete(:ash_notifications) || []
Ash.Actions.Helpers.warn_missed!(atomic_changeset.resource, action, %{
resource_notifications: remaining_notifications
@ -307,15 +285,27 @@ defmodule Ash.Actions.Destroy.Bulk do
%{bulk_result | notifications: notifications}
else
if opts[:notify?] do
Ash.Actions.Helpers.warn_missed!(atomic_changeset.resource, action, %{
resource_notifications: notifications
})
%{bulk_result | notifications: []}
end
else
process_notifications = List.wrap(Process.get(:ash_notifications, []))
Process.put(
:ash_notifications,
process_notifications ++ 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)
@ -452,7 +442,7 @@ defmodule Ash.Actions.Destroy.Bulk do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
end
end
else
@ -1295,8 +1285,6 @@ defmodule Ash.Actions.Destroy.Bulk do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
notifications = Process.get(:ash_notifications, [])
remaining_notifications = Ash.Notifier.notify(notifications)
Process.delete(:ash_notifications) || []
@ -1304,6 +1292,8 @@ defmodule Ash.Actions.Destroy.Bulk do
Ash.Actions.Helpers.warn_missed!(resource, action, %{
resource_notifications: remaining_notifications
})
Process.delete(:ash_started_transaction?)
end
end
else
@ -1462,7 +1452,7 @@ defmodule Ash.Actions.Destroy.Bulk do
{batch_result, notifications, errors}
after
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
end
end,
timeout: :infinity,

View file

@ -804,7 +804,7 @@ defmodule Ash.Actions.Read do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
end
end
end

View file

@ -152,16 +152,12 @@ defmodule Ash.Actions.Update.Bulk do
atomic_changeset = %{atomic_changeset | domain: domain}
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
context =
@ -233,21 +229,14 @@ defmodule Ash.Actions.Update.Bulk do
end
|> case do
{:ok, bulk_result} ->
notifications =
if notify? do
List.wrap(bulk_result.notifications) ++
List.wrap(Process.delete(:ash_notifications))
else
List.wrap(bulk_result.notifications)
end
if opts[:return_notifications?] do
%{bulk_result | notifications: notifications}
else
if opts[:return_notifications?] do
bulk_result
else
if notify? do
notifications =
List.wrap(Process.delete(:ash_notifications)) ++ bulk_result.notifications
if opts[:notify?] do
remaining_notifications = Ash.Notifier.notify(notifications)
Ash.Actions.Helpers.warn_missed!(atomic_changeset.resource, action, %{
@ -256,8 +245,17 @@ defmodule Ash.Actions.Update.Bulk do
%{bulk_result | notifications: notifications}
else
bulk_result
%{bulk_result | notifications: []}
end
else
process_notifications = List.wrap(Process.get(:ash_notifications, []))
Process.put(
:ash_notifications,
process_notifications ++ bulk_result.notifications
)
%{bulk_result | notifications: []}
end
end
@ -270,7 +268,7 @@ defmodule Ash.Actions.Update.Bulk do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
end
end
end
@ -409,7 +407,7 @@ defmodule Ash.Actions.Update.Bulk do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
end
end
else
@ -1517,7 +1515,7 @@ defmodule Ash.Actions.Update.Bulk do
end
after
if notify? do
Process.put(:ash_started_transaction?, false)
Process.delete(:ash_started_transaction?)
notifications = Process.get(:ash_notifications, [])
remaining_notifications = Ash.Notifier.notify(notifications)
Process.delete(:ash_notifications) || []
@ -1666,7 +1664,6 @@ defmodule Ash.Actions.Update.Bulk do
Task.async_stream(
stream,
fn batch ->
try do
Process.put(:ash_started_transaction?, true)
batch_result = callback.(batch)
{errors, _} = Process.get({:bulk_update_errors, ref}) || {[], 0}
@ -1689,9 +1686,6 @@ defmodule Ash.Actions.Update.Bulk do
end
{batch_result, notifications, errors}
after
Process.put(:ash_started_transaction?, false)
end
end,
timeout: :infinity,
max_concurrency: max_concurrency

View file

@ -75,7 +75,7 @@ defmodule Ash.Actions.Update do
nil}
opts[:atomic_upgrade?] == false ->
{{:not_atomic, "atomic upgrade was disabled"}, nil}
{{:not_atomic, "atomic upgrade was disabled with opts"}, nil}
true ->
params =

View file

@ -2981,17 +2981,7 @@ defmodule Ash.Changeset do
)
|> case do
{:ok, {:ok, value, changeset, instructions}} ->
notifications =
if notify? && !opts[:return_notifications?] do
Enum.concat(
instructions[:notifications] || [],
Process.delete(:ash_notifications) || []
)
else
instructions[:notifications] || []
end
{:ok, value, changeset, Map.put(instructions, :notifications, notifications)}
{:ok, value, changeset, instructions}
{:ok, {:error, error}} ->
{:error, error}
@ -3039,11 +3029,15 @@ defmodule Ash.Changeset do
{:ok, value, changeset, Map.put(instructions, :notifications, [])}
else
notifications =
List.wrap(Process.delete(:ash_notifications)) ++
(instructions[:notifications] || [])
notifications =
if opts[:return_notifications?] do
instructions[:notifications] || []
notifications
else
Ash.Notifier.notify(instructions[:notifications] || [])
Ash.Notifier.notify(notifications)
end
{:ok, value, changeset, Map.put(instructions, :notifications, notifications)}

View file

@ -2905,7 +2905,8 @@ defmodule Ash.Query do
def data_layer_query(%{resource: resource, domain: domain} = ash_query, opts) do
query = opts[:initial_query] || Ash.DataLayer.resource_to_query(resource, domain)
context = ash_query.context
context =
ash_query.context
|> Map.put(:action, ash_query.action)
|> Map.put_new(:private, %{})
|> put_in([:private, :tenant], ash_query.tenant)