improvement: add backwards compatibility for data layers returning only index metadata

Author: DGollings

lib/ash/actions/create/bulk.ex

@@ -1456,6 +1456,7 @@ defmodule Ash.Actions.Create.Bulk do
       all_changes,
       results,
       changesets_by_index,
+      nil,
       changesets,
       opts,
       ref,

lib/ash/actions/destroy/bulk.ex

@@ -2171,6 +2171,7 @@ defmodule Ash.Actions.Destroy.Bulk do
       all_changes,
       batch,
       changesets_by_index,
+      nil,
       changesets,
       opts,
       ref,

lib/ash/actions/update/bulk.ex

@@ -2147,13 +2147,15 @@ defmodule Ash.Actions.Update.Bulk do
     end
   end
 
-  defp index_changesets_by_ref(batch, context_key) do
-    Enum.reduce(batch, %{}, fn changeset, changesets_by_ref ->
-      Map.put(
-        changesets_by_ref,
-        changeset.context[context_key].ref,
-        changeset
-      )
+  defp index_changesets(batch, context_key) do
+    Enum.reduce(batch, {%{}, %{}}, fn changeset, {by_ref, by_index} ->
+      ref = changeset.context[context_key].ref
+      index = changeset.context[context_key].index
+
+      {
+        Map.put(by_ref, ref, changeset),
+        Map.put(by_index, index, ref)
+      }
     end)
   end
 
@@ -2529,7 +2531,7 @@ defmodule Ash.Actions.Update.Bulk do
         end
       end)
 
-    changesets_by_ref = index_changesets_by_ref(batch, context_key)
+    {changesets_by_ref, changesets_by_index} = index_changesets(batch, context_key)
 
     batch =
       batch
@@ -2686,7 +2688,7 @@ defmodule Ash.Actions.Update.Bulk do
           end)
       end
 
-    {batch, changesets_by_ref}
+    {batch, changesets_by_ref, changesets_by_index}
   end
 
   defp manage_relationships(updated, domain, changeset, engine_opts) do
@@ -2704,7 +2706,7 @@ defmodule Ash.Actions.Update.Bulk do
   end
 
   defp run_after_action_hooks(
-         {batch_results, changesets_by_ref},
+         {batch_results, changesets_by_ref, changesets_by_index},
          opts,
          domain,
          ref,
@@ -2759,11 +2761,11 @@ defmodule Ash.Actions.Update.Bulk do
         end
       end)
 
-    {results, changesets_by_ref}
+    {results, changesets_by_ref, changesets_by_index}
   end
 
   defp process_results(
-         {batch, changesets_by_ref},
+         {batch, changesets_by_ref, changesets_by_index},
          changes,
          all_changes,
          opts,
@@ -2780,6 +2782,7 @@ defmodule Ash.Actions.Update.Bulk do
       all_changes,
       batch,
       changesets_by_ref,
+      changesets_by_index,
       changesets,
       opts,
       ref,
@@ -2872,6 +2875,7 @@ defmodule Ash.Actions.Update.Bulk do
         all_changes,
         results,
         changesets_by_ref,
+        changesets_by_index,
         changesets,
         opts,
         ref,
@@ -2964,6 +2968,7 @@ defmodule Ash.Actions.Update.Bulk do
                   result,
                   changes[index],
                   changesets_by_ref,
+                  changesets_by_index,
                   metadata_key,
                   ref_metadata_key
                 )
@@ -3573,10 +3578,26 @@ defmodule Ash.Actions.Update.Bulk do
     changeset
   end
 
-  defp result_matches_changes?(result, changes, changesets_by_ref, metadata_key, ref_metadata_key) do
+  defp result_matches_changes?(
+         result,
+         changes,
+         changesets_by_ref,
+         changesets_by_index,
+         metadata_key,
+         ref_metadata_key
+       ) do
     if ref_metadata_key do
       ref_key = result.__metadata__[ref_metadata_key]
 
+      ref_key =
+        if ref_key do
+          ref_key
+        else
+          # Fallback: data layer didn't return ref, look it up by index
+          index = result.__metadata__[metadata_key]
+          changesets_by_index[index]
+        end
+
       changesets_by_ref
       |> Map.get(ref_key)
       |> ensure_changeset!(result, metadata_key, ref_metadata_key)

lib/ash/data_layer/ets/ets.ex

@@ -1880,15 +1880,23 @@ defmodule Ash.DataLayer.Ets do
             {:ok, result} ->
               result =
                 if result_changeset.context[:bulk_update] do
-                  result
-                  |> Ash.Resource.put_metadata(
-                    :bulk_update_index,
-                    result_changeset.context.bulk_update.index
-                  )
-                  |> Ash.Resource.put_metadata(
-                    :bulk_action_ref,
-                    result_changeset.context.bulk_update.ref
-                  )
+                  result =
+                    result
+                    |> Ash.Resource.put_metadata(
+                      :bulk_update_index,
+                      result_changeset.context.bulk_update.index
+                    )
+
+                  # For testing backwards compatibility, allow skipping ref metadata
+                  if Application.get_env(:ash, :test_bulk_index_only, false) do
+                    result
+                  else
+                    Ash.Resource.put_metadata(
+                      result,
+                      :bulk_action_ref,
+                      result_changeset.context.bulk_update.ref
+                    )
+                  end
                 else
                   result
                 end

test/actions/bulk/bulk_update_test.exs

@@ -964,6 +964,35 @@ defmodule Ash.Test.Actions.BulkUpdateTest do
              end)
   end
 
+  test "runs after batch hooks with legacy data layers (no refs)" do
+    assert %Ash.BulkResult{
+             records: [
+               %{title: "before_title1_after", title2: "updated value"},
+               %{title: "before_title2_after", 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)
+             |> Enum.to_list()
+             |> Enum.map(&Ash.Test.Helpers.strip_bulk_action_ref/1)
+             |> Ash.bulk_update!(:update_with_after_batch, %{title2: "updated value"},
+               resource: Post,
+               strategy: :stream,
+               return_records?: true,
+               return_errors?: true,
+               authorize?: false
+             )
+             |> Map.update!(:records, fn records ->
+               Enum.sort_by(records, & &1.title)
+             end)
+  end
+
   test "runs changes in batches" do
     create_records = fn count ->
       Stream.iterate(1, &(&1 + 1))

test/support/helpers.ex

@@ -36,4 +36,23 @@ defmodule Ash.Test.Helpers do
       })
     end
   end
+
+  @doc """
+  Strips bulk_action_ref metadata from records to simulate legacy data layer behavior.
+
+  Use this in tests that verify backwards compatibility with data layers that don't
+  support returning the bulk action ref.
+  """
+  def strip_bulk_action_refs(%Ash.BulkResult{records: records} = result)
+      when is_list(records) do
+    %{result | records: Enum.map(records, &strip_bulk_action_ref/1)}
+  end
+
+  def strip_bulk_action_refs(result), do: result
+
+  def strip_bulk_action_ref(record) when is_struct(record) do
+    Ash.Resource.set_metadata(record, Map.delete(record.__metadata__, :bulk_action_ref))
+  end
+
+  def strip_bulk_action_ref(other), do: other
 end