nested bulk_create failing when using transaction: :batch

**Describe the bug**

A nested `bulk_create` fails when a parent `bulk_create` is called with `transaction: :batch`.

**Explanation:**

I have two resources that are joined using a through resource:

1. Subscriber
2. List
3. SubscriberList (through resource)

I then use `bulk_create` to create a bunch of `Subscriber`s and then use the `after_batch` hook to create `SubscriberList` records to relate the `Subscriber` with an existing `List`. I use a `bulk_create` within the `after_batch` hook to create these through records.

This fails when I call the original `bulk_create` on the `Subscriber` and pass in `transaction: :batch`. However, it passes when I pass `transaction: false`. The failure I get is this:

```
* Invalid value provided for subscriber_id: does not exist.
```

This is what the `after_batch` hook looks like:

```elixir
@impl true
def after_batch(results, _opts, context) do
Stream.map(results, fn {changeset, subscriber} ->
%{
subscriber_id: subscriber.id,
list_id: changeset.arguments.list_id,
status: changeset.arguments.list_subscription_status
}
end)
|> Enum.map(&dbg/1)
|> Ash.bulk_create!(
Daakia.Newsletters.SubscriberList,
:create,
return_records?: false,
upsert?: true,
batch_size: 1000,
upsert_fields: [:status, :meta],
transaction: false,
return_errors?: true,
max_concurrency: 10,
tenant: context.tenant,
domain: Daakia.Newsletters
)

Stream.map(results, fn {_changeset, result} ->
{:ok, result}
end)
end
```

I added a `dbg` in the `after_batch` hook (visible in code above) to make sure the subscribers are getting created before the nested `bulk_create` is executed and it does print proper maps:

```elixir
%{
status: :confirmed,
list_id: "01914426-b560-7f9b-8102-c56e969fc41e",
subscriber_id: "01914426-c052-717b-b160-ebb3c2935e84"
}
```

So even though the `subscriber_id` is populated, presumably after creating a successful record in the db (within the transaction), the `create` action of `SubscriberList` complains that `subscriber_id` does not exist.

**To Reproduce**

Here is a Subscriber resource:

```elixir
defmodule Daakia.Newsletters.Subscriber do
alias Daakia.Newsletters.Subscriber.Types

use Ash.Resource,
data_layer: AshPostgres.DataLayer,
authorizers: [Ash.Policy.Authorizer],
domain: Daakia.Newsletters

postgres do
table "subscribers"
repo Daakia.Repo

references do
reference :organization, on_delete: :delete
end
end

actions do
defaults [:create, :update, :destroy]

create :new_subscriber do
primary? true
upsert? true
upsert_identity :unique_email
accept [:email, :name, :attributes, :status]
argument :list_id, :uuid, allow_nil?: false
argument :list_subscription_status, Types.ListSubscriptionStatus, allow_nil?: false

change Daakia.Newsletters.Changes.CreateSubscriberlistEntry
end

read :read do
primary? true

pagination do
offset? true
countable :by_default
default_limit 10
end
end
end

policies do
policy action_type(:read) do
authorize_if always()
end

policy action_type(:create) do
authorize_if always()
end

policy action_type(:update) do
authorize_if relates_to_actor_via(:user)
end

policy action_type(:destroy) do
authorize_if relates_to_actor_via(:user)
end
end

multitenancy do
strategy :attribute
attribute :organization_id
end

attributes do
uuid_v7_primary_key :id

attribute :email, :ci_string do
allow_nil? false
end

attribute :name, :string do
allow_nil? true
end

attribute :attributes, :map do
allow_nil? false
default %{}
end

attribute :status, Types.OverallSubscriptionStatus do
allow_nil? false
default :enabled
end

create_timestamp :created_at
update_timestamp :updated_at
end

relationships do
many_to_many :lists, Daakia.Newsletters.List do
through Daakia.Newsletters.SubscriberList
source_attribute_on_join_resource :subscriber_id
destination_attribute_on_join_resource :list_id
join_relationship :list_subscribers
end

belongs_to :organization, Daakia.Accounts.Organization,
domain: Daakia.Accounts,
allow_nil?: false
end

identities do
identity :unique_email, [:email]
end
end
```

Here is a List resource

```elixir
defmodule Daakia.Newsletters.List do
use Ash.Resource,
otp_app: :daakia,
data_layer: AshPostgres.DataLayer,
authorizers: [Ash.Policy.Authorizer],
domain: Daakia.Newsletters

postgres do
table "lists"
repo Daakia.Repo

references do
reference :organization, on_delete: :delete
end
end

actions do
defaults [:read, :update, :destroy]

create :create_new_list do
primary? true
accept [:name, :description, :type, :optin]
argument :subscribers, {:array, :map}, allow_nil?: true

change relate_actor(:owner)

change manage_relationship(:subscribers,
type: :append_and_remove,
on_no_match: :create
)
end
end

policies do
policy action_type(:read) do
authorize_if always()
end

policy action_type(:create) do
authorize_if actor_present()
end

policy action_type(:update) do
authorize_if relates_to_actor_via(:owner)
end

policy action_type(:destroy) do
authorize_if relates_to_actor_via(:owner)
end
end

multitenancy do
strategy :attribute
attribute :organization_id
end

attributes do
uuid_v7_primary_key :id

attribute :name, :string do
allow_nil? false
public? true
end

attribute :description, :string do
allow_nil? true
end

attribute :type, Daakia.Newsletters.List.Types.Type do
allow_nil? false
end

attribute :optin, Daakia.Newsletters.List.Types.OptinType do
allow_nil? false
end

create_timestamp :created_at
update_timestamp :updated_at
end

relationships do
belongs_to :owner, Daakia.Accounts.User do
domain Daakia.Accounts
allow_nil? false
public? true
end

many_to_many :subscribers, Daakia.Newsletters.Subscriber do
through Daakia.Newsletters.SubscriberList
source_attribute_on_join_resource :list_id
destination_attribute_on_join_resource :subscriber_id
end

belongs_to :organization, Daakia.Accounts.Organization do
domain Daakia.Accounts
allow_nil? false
end
end

identities do
identity :unique_name, [:name]
end
end
```

Here is the associated SubscriberList resource

```elixir
defmodule Daakia.Newsletters.SubscriberList do
alias Daakia.Newsletters.Subscriber.Types

use Ash.Resource,
data_layer: AshPostgres.DataLayer,
domain: Daakia.Newsletters

postgres do
table "subscriber_lists"
repo Daakia.Repo

references do
reference :organization, on_delete: :delete
reference :list, on_delete: :delete
reference :subscriber, on_delete: :delete
end
end

code_interface do
define :create, action: :create
end

actions do
defaults [:read, :destroy, update: :*]

create :create do
primary? true
accept [:status, :subscriber_id, :list_id]
upsert? true
end

update :unsubscribe_from_list do
accept []

validate attribute_does_not_equal(:status, :unsubscribed) do
message "User is already unsubscribed"
end

change set_attribute(:status, :unsubscribed)
end
end

multitenancy do
strategy :attribute
attribute :organization_id
end

attributes do
attribute :status, Types.ListSubscriptionStatus do
allow_nil? false
end

attribute :meta, :map do
allow_nil? false
default %{}
end

create_timestamp :created_at
update_timestamp :updated_at
end

relationships do
belongs_to :subscriber, Daakia.Newsletters.Subscriber,
primary_key?: true,
allow_nil?: false

belongs_to :list, Daakia.Newsletters.List,
primary_key?: true,
allow_nil?: false

belongs_to :organization, Daakia.Accounts.Organization,
domain: Daakia.Accounts,
allow_nil?: false
end
end
```

Here is the CreateSubscriberListEntry change

```elixir
defmodule Daakia.Newsletters.Changes.CreateSubscriberlistEntry do
use Ash.Resource.Change

@doc """
This will create the associated through record in the subscribers_list table
"""
@impl true
def change(changeset, _, _) do
status = Ash.Changeset.get_argument(changeset, :list_subscription_status)

Ash.Changeset.after_action(changeset, fn changeset, subscriber ->
Daakia.Newsletters.SubscriberList.create!(
%{
subscriber_id: subscriber.id,
list_id: changeset.arguments.list_id,
status: status
},
upsert?: true,
upsert_fields: [:status],
tenant: changeset.tenant
)

{:ok, subscriber}
end)
end

@doc """
We do not run an after_action hook for batches but rather the after_batch hook
"""
@impl true
def batch_change(changesets, _opts, _context) do
changesets
end

@doc """
This will create the associated through records for the created subscribers in batch
"""
@impl true
def after_batch(results, _opts, context) do
Stream.map(results, fn {changeset, subscriber} ->
%{
subscriber_id: subscriber.id,
list_id: changeset.arguments.list_id,
status: changeset.arguments.list_subscription_status
}
end)
|> Enum.map(&dbg/1)
|> Ash.bulk_create!(
Daakia.Newsletters.SubscriberList,
:create,
return_records?: false,
upsert?: true,
batch_size: 1000,
upsert_fields: [:status, :meta],
transaction: false,
return_errors?: true,
max_concurrency: 10,
tenant: context.tenant,
domain: Daakia.Newsletters
)

Stream.map(results, fn {_changeset, result} ->
{:ok, result}
end)
end
end
```

This code is exercised like this:

```elixir
random_subscribers =
Task.async_stream(
0..100,
fn _i ->
%{
email: Faker.Internet.email(),
name: Faker.Person.name(),
status: :enabled,
list_subscription_status: :confirmed,
list_id: list.id
}
end,
ordered: false
)
|> Stream.chunk_every(1000)
|> Stream.map(fn list ->
Enum.reduce(list, [], fn {:ok, result}, acc -> [result | acc] end)
|> Ash.bulk_create!(
Subscriber,
:new_subscriber,
return_records?: true,
upsert?: true,
batch_size: 1000,
upsert_identity: :unique_email,
upsert_fields: [:name, :attributes, :status],
transaction: false,
return_errors?: true,
max_concurrency: 10,
tenant: organization.id,
domain: Newsletters
)
end)
|> Enum.to_list()
```
**Expected behavior**

I expected the nested `bulk_create` to succeed with `transaction: :batch`.

**Runtime**
- Elixir version: 1.17.2
- Erlang version: Erlang/OTP 26
- OS: MacOS
- Ash version: latest from main
- any related extension versions: All latest from hex

**Additional context**
N/A

Start by reproducing the nested Ash.bulk_create! calls shown in the issue, comparing transaction: :batch with transaction: false. Trace the after_batch hook and AshPostgres.DataLayer handling of the parent transaction and foreign-key validation. Done means the SubscriberList records created in after_batch succeed with transaction: :batch without the invalid subscriber_id error.

Written by the indexing model from the issue text.