From 718adb48915a3ddc082599ede2152f7a0c4a0319 Mon Sep 17 00:00:00 2001
From: Frank Dugan III <frank3@duganator.com>
Date: Wed, 1 May 2024 15:26:33 -0500
Subject: [PATCH] feat: add read_action option to bulk actions (#1088)

---
 lib/ash.ex                     |  4 ++++
 lib/ash/actions/update/bulk.ex | 19 +++++++++++++++----
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/lib/ash.ex b/lib/ash.ex
index e5a933b5..1ef83e88 100644
--- a/lib/ash.ex
+++ b/lib/ash.ex
@@ -324,6 +324,10 @@ defmodule Ash do
   def create_opts, do: @create_opts_schema
 
   @shared_bulk_opts_schema [
+    read_action: [
+      type: :atom,
+      doc: "The action to use when building the read query."
+    ],
     assume_casted?: [
       type: :boolean,
       default: false,
diff --git a/lib/ash/actions/update/bulk.ex b/lib/ash/actions/update/bulk.ex
index 051d8222..24eb0c7f 100644
--- a/lib/ash/actions/update/bulk.ex
+++ b/lib/ash/actions/update/bulk.ex
@@ -22,7 +22,7 @@ defmodule Ash.Actions.Update.Bulk do
         query =
           Ash.Query.for_read(
             query,
-            Ash.Resource.Info.primary_action!(query.resource, :read).name,
+            get_read_action(query.resource, opts).name,
             %{},
             actor: opts[:actor],
             tenant: opts[:tenant]
@@ -785,6 +785,7 @@ defmodule Ash.Actions.Update.Bulk do
   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, opts)
 
     {_, opts} =
       Ash.Actions.Helpers.set_context_and_get_opts(domain, Ash.Changeset.new(resource), opts)
@@ -797,7 +798,7 @@ defmodule Ash.Actions.Update.Bulk do
         Enum.empty?(Ash.Resource.Info.primary_key(resource)) ->
           {:not_atomic, "cannot atomically update a stream without a primary key"}
 
-        !Ash.Resource.Info.primary_action(resource, :read) ->
+        !read_action ->
           {:not_atomic, "cannot atomically update a stream without a primary read action"}
 
         Ash.DataLayer.data_layer_can?(resource, :update_query) ->
@@ -812,7 +813,7 @@ defmodule Ash.Actions.Update.Bulk do
         query =
           resource
           |> Ash.Query.new()
-          |> Map.put(:action, Ash.Resource.Info.primary_action!(resource, :read))
+          |> Map.put(:action, read_action)
 
         case Ash.Actions.Read.Stream.stream_strategy(
                query,
@@ -899,7 +900,7 @@ defmodule Ash.Actions.Update.Bulk do
         pkeys = [or: Enum.map(batch, &Map.take(&1, pkey))]
 
         resource
-        |> Ash.Query.for_read(Ash.Resource.Info.primary_action!(resource, :read).name, %{},
+        |> Ash.Query.for_read(get_read_action(resource, opts).name, %{},
           actor: opts[:actor],
           authorize?: false,
           context: atomic_changeset.context,
@@ -2454,4 +2455,14 @@ defmodule Ash.Actions.Update.Bulk do
       context
     )
   end
+
+  defp get_read_action(resource, opts) do
+    case opts[:read_action] do
+      nil ->
+        Ash.Resource.Info.primary_action!(resource, :read)
+
+      action ->
+        Ash.Resource.Info.action(resource, action)
+    end
+  end
 end