fix: set notify?: true when return_notifications?: true is set

closes #1100

lib/ash/actions/create/bulk.ex

@@ -8,6 +8,13 @@ defmodule Ash.Actions.Create.Bulk do
   def run(domain, resource, action, inputs, opts) do
     action = Ash.Resource.Info.action(resource, action)
 
+    opts =
+      if opts[:return_notifications?] do
+        Keyword.put(opts, :notify?, true)
+      else
+        opts
+      end
+
     if !action do
       raise Ash.Error.Invalid.NoSuchAction, resource: resource, action: action, type: :create
     end
@@ -179,9 +186,9 @@ defmodule Ash.Actions.Create.Bulk do
           else
             if opts[:notify?] do
               Ash.Notifier.notify(Process.delete({:bulk_create_notifications, ref}))
+            else
+              []
             end
-
-            []
           end
 
         {errors, error_count} = Process.get({:bulk_create_errors, ref}) || {[], 0}
@@ -612,11 +619,11 @@ defmodule Ash.Actions.Create.Bulk do
                   process_notifications ++ bulk_notifications
                 else
                   if opts[:transaction] && opts[:transaction] != :all do
-                    Ash.Notifier.notify(bulk_notifications)
-                    Ash.Notifier.notify(process_notifications)
+                    Ash.Notifier.notify(bulk_notifications) ++
+                      Ash.Notifier.notify(process_notifications)
+                  else
+                    []
                   end
-
-                  []
                 end
               end
 

lib/ash/actions/destroy/bulk.ex

@@ -11,6 +11,13 @@ defmodule Ash.Actions.Destroy.Bulk do
       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
+
     run(
       domain,
       stream,
@@ -47,6 +54,13 @@ defmodule Ash.Actions.Destroy.Bulk do
 
     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
@@ -278,6 +292,13 @@ defmodule Ash.Actions.Destroy.Bulk do
 
     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 =

lib/ash/actions/update/bulk.ex

@@ -15,6 +15,13 @@ defmodule Ash.Actions.Update.Bulk do
   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 =
       if query.__validated_for_action__ do
         query
@@ -255,6 +262,13 @@ defmodule Ash.Actions.Update.Bulk do
 
     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],

test/actions/bulk/bulk_create_test.exs

@@ -4,6 +4,14 @@ defmodule Ash.Test.Actions.BulkCreateTest do
 
   alias Ash.Test.Domain, as: Domain
 
+  defmodule Notifier do
+    use Ash.Notifier
+
+    def notify(notification) do
+      send(self(), {:notification, notification})
+    end
+  end
+
   defmodule AddAfterToTitle do
     use Ash.Resource.Change
 
@@ -63,6 +71,7 @@ defmodule Ash.Test.Actions.BulkCreateTest do
     use Ash.Resource,
       domain: Domain,
       data_layer: Ash.DataLayer.Ets,
+      notifiers: [Notifier],
       authorizers: [Ash.Policy.Authorizer]
 
     alias Ash.Test.Actions.BulkCreateTest.Org
@@ -698,12 +707,41 @@ defmodule Ash.Test.Actions.BulkCreateTest do
                  tenant: org.id,
                  authorize?: true,
                  return_stream?: true,
-                 notify?: true,
                  return_notifications?: true
                )
                |> Enum.to_list()
     end
 
+    test "notifications are sent with notify?: true" do
+      org =
+        Org
+        |> Ash.Changeset.for_create(:create, %{})
+        |> Ash.create!()
+
+      assert [{:ok, %{title: "title1"}}, {:ok, %{title: "title2"}}] =
+               [%{title: "title1", authorize?: true}, %{title: "title2", authorize?: true}]
+               |> Ash.bulk_create!(
+                 Post,
+                 :create_with_policy,
+                 authorize?: true,
+                 tenant: org.id,
+                 notify?: true,
+                 return_stream?: true,
+                 return_records?: true
+               )
+               |> Enum.to_list()
+               |> Enum.sort_by(fn
+                 {:ok, v} ->
+                   v.title
+
+                 _ ->
+                   nil
+               end)
+
+      assert_received {:notification, %{data: %{title: "title1"}}}
+      assert_received {:notification, %{data: %{title: "title2"}}}
+    end
+
     test "by returning records, you get the records in the stream" do
       org =
         Org
@@ -748,7 +786,6 @@ defmodule Ash.Test.Actions.BulkCreateTest do
                  :create_with_policy,
                  authorize?: true,
                  tenant: org.id,
-                 notify?: true,
                  return_stream?: true,
                  return_notifications?: true,
                  return_records?: true
@@ -785,7 +822,6 @@ defmodule Ash.Test.Actions.BulkCreateTest do
                  Post,
                  :create_with_policy,
                  authorize?: true,
-                 notify?: true,
                  tenant: org.id,
                  return_stream?: true,
                  return_notifications?: true,

test/actions/bulk/bulk_destroy_test.exs

@@ -5,6 +5,14 @@ defmodule Ash.Test.Actions.BulkDestroyTest do
   require Ash.Query
   alias Ash.Test.Domain, as: Domain
 
+  defmodule Notifier do
+    use Ash.Notifier
+
+    def notify(notification) do
+      send(self(), {:notification, notification})
+    end
+  end
+
   defmodule AddAfterToTitle do
     use Ash.Resource.Change
 
@@ -29,6 +37,7 @@ defmodule Ash.Test.Actions.BulkDestroyTest do
     @moduledoc false
     use Ash.Resource,
       domain: Domain,
+      notifiers: [Notifier],
       data_layer: Ash.DataLayer.Ets,
       authorizers: [Ash.Policy.Authorizer]
 
@@ -152,6 +161,45 @@ defmodule Ash.Test.Actions.BulkDestroyTest do
     assert [] = Ash.read!(Post)
   end
 
+  test "sends notifications" do
+    assert %Ash.BulkResult{records: [%{}, %{}]} =
+             Ash.bulk_create!([%{title: "title1"}, %{title: "title2"}], Post, :create,
+               return_stream?: true,
+               return_records?: true
+             )
+             |> Stream.map(fn {:ok, result} ->
+               result
+             end)
+             |> Ash.bulk_destroy!(:destroy, %{},
+               resource: Post,
+               strategy: :stream,
+               notify?: true,
+               return_records?: true,
+               return_errors?: true
+             )
+
+    assert_received {:notification, %{data: %{title: "title1"}}}
+    assert_received {:notification, %{data: %{title: "title2"}}}
+  end
+
+  test "notifications can be returned" do
+    assert %Ash.BulkResult{records: [%{}, %{}], notifications: [%{}, %{}]} =
+             Ash.bulk_create!([%{title: "title1"}, %{title: "title2"}], Post, :create,
+               return_stream?: true,
+               return_records?: true
+             )
+             |> Stream.map(fn {:ok, result} ->
+               result
+             end)
+             |> Ash.bulk_destroy!(:destroy, %{},
+               resource: Post,
+               strategy: :stream,
+               return_notifications?: true,
+               return_records?: true,
+               return_errors?: true
+             )
+  end
+
   test "runs changes" do
     assert %Ash.BulkResult{
              records: [

test/actions/bulk/bulk_update_test.exs

@@ -4,6 +4,14 @@ defmodule Ash.Test.Actions.BulkUpdateTest do
 
   alias Ash.Test.Domain, as: Domain
 
+  defmodule Notifier do
+    use Ash.Notifier
+
+    def notify(notification) do
+      send(self(), {:notification, notification})
+    end
+  end
+
   defmodule AddAfterToTitle do
     use Ash.Resource.Change
 
@@ -78,6 +86,7 @@ defmodule Ash.Test.Actions.BulkUpdateTest do
     use Ash.Resource,
       domain: Domain,
       data_layer: Ash.DataLayer.Ets,
+      notifiers: [Notifier],
       authorizers: [Ash.Policy.Authorizer]
 
     ets do
@@ -227,6 +236,55 @@ defmodule Ash.Test.Actions.BulkUpdateTest do
              )
   end
 
+  test "sends notifications" do
+    assert %Ash.BulkResult{records: [%{title2: "updated value"}, %{title2: "updated value"}]} =
+             Ash.bulk_create!([%{title: "title1"}, %{title: "title2"}], Post, :create,
+               return_stream?: true,
+               return_records?: true,
+               authorize?: false
+             )
+             |> Stream.map(fn {:ok, result} ->
+               result
+             end)
+             |> Ash.bulk_update!(:update, %{title2: "updated value"},
+               resource: Post,
+               strategy: :atomic_batches,
+               return_records?: true,
+               notify?: true,
+               return_errors?: true,
+               authorize?: false
+             )
+
+    assert_received {:notification, %{data: %{title: "title2"}}}
+    assert_received {:notification, %{data: %{title: "title2"}}}
+  end
+
+  test "notifications can be returned" do
+    assert %Ash.BulkResult{
+             records: [%{title2: "updated value"}, %{title2: "updated value"}],
+             notifications: [
+               %{data: %{title2: "updated value"}},
+               %{data: %{title2: "updated value"}}
+             ]
+           } =
+             Ash.bulk_create!([%{title: "title1"}, %{title: "title2"}], Post, :create,
+               return_stream?: true,
+               return_records?: true,
+               authorize?: false
+             )
+             |> Stream.map(fn {:ok, result} ->
+               result
+             end)
+             |> Ash.bulk_update!(:update, %{title2: "updated value"},
+               resource: Post,
+               strategy: :atomic_batches,
+               return_records?: true,
+               return_notifications?: true,
+               return_errors?: true,
+               authorize?: false
+             )
+  end
+
   test "runs changes" do
     assert %Ash.BulkResult{
              records: [