mirror of
https://github.com/ash-project/ash.git
synced 2024-09-20 05:23:03 +12:00
fix: move to simpler transaction logic
This commit is contained in:
parent
05f2d9ad64
commit
807b16e268
8 changed files with 291 additions and 295 deletions
|
@ -27,12 +27,14 @@ locals_without_parens = [
|
||||||
many_to_many: 3,
|
many_to_many: 3,
|
||||||
primary?: 1,
|
primary?: 1,
|
||||||
primary_key?: 1,
|
primary_key?: 1,
|
||||||
|
private?: 1,
|
||||||
read: 1,
|
read: 1,
|
||||||
read: 2,
|
read: 2,
|
||||||
resource: 1,
|
resource: 1,
|
||||||
resource: 2,
|
resource: 2,
|
||||||
source_field: 1,
|
source_field: 1,
|
||||||
source_field_on_join_table: 1,
|
source_field_on_join_table: 1,
|
||||||
|
table: 1,
|
||||||
through: 1,
|
through: 1,
|
||||||
type: 1,
|
type: 1,
|
||||||
update: 1,
|
update: 1,
|
||||||
|
|
|
@ -11,7 +11,10 @@ defmodule Ash.Actions.Create do
|
||||||
side_load = opts[:side_load] || []
|
side_load = opts[:side_load] || []
|
||||||
upsert? = opts[:upsert?] || false
|
upsert? = opts[:upsert?] || false
|
||||||
|
|
||||||
engine_opts = Keyword.take(opts, [:verbose?, :actor, :authorize?])
|
engine_opts =
|
||||||
|
opts
|
||||||
|
|> Keyword.take([:verbose?, :actor, :authorize?])
|
||||||
|
|> Keyword.put(:transaction?, true)
|
||||||
|
|
||||||
action =
|
action =
|
||||||
if is_atom(action) and not is_nil(action) do
|
if is_atom(action) and not is_nil(action) do
|
||||||
|
@ -143,7 +146,7 @@ defmodule Ash.Actions.Create do
|
||||||
[authorization_request | [commit_request | relationship_read_requests]] ++
|
[authorization_request | [commit_request | relationship_read_requests]] ++
|
||||||
side_load_requests,
|
side_load_requests,
|
||||||
api,
|
api,
|
||||||
Keyword.put(engine_opts, :transaction?, true)
|
engine_opts
|
||||||
)
|
)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
|
@ -6,7 +6,10 @@ defmodule Ash.Actions.Destroy do
|
||||||
@spec run(Ash.api(), Ash.record(), Ash.action(), Keyword.t()) ::
|
@spec run(Ash.api(), Ash.record(), Ash.action(), Keyword.t()) ::
|
||||||
:ok | {:error, Ecto.Changeset.t()} | {:error, Ash.error()}
|
:ok | {:error, Ecto.Changeset.t()} | {:error, Ash.error()}
|
||||||
def run(api, %resource{} = record, action, opts) do
|
def run(api, %resource{} = record, action, opts) do
|
||||||
engine_opts = Keyword.take(opts, [:verbose?, :actor, :authorize?])
|
engine_opts =
|
||||||
|
opts
|
||||||
|
|> Keyword.take([:verbose?, :actor, :authorize?])
|
||||||
|
|> Keyword.put(:transaction?, true)
|
||||||
|
|
||||||
action =
|
action =
|
||||||
if is_atom(action) and not is_nil(action) do
|
if is_atom(action) and not is_nil(action) do
|
||||||
|
|
|
@ -11,7 +11,11 @@ defmodule Ash.Actions.Update do
|
||||||
attributes = Keyword.get(opts, :attributes, %{})
|
attributes = Keyword.get(opts, :attributes, %{})
|
||||||
relationships = Keyword.get(opts, :relationships, %{})
|
relationships = Keyword.get(opts, :relationships, %{})
|
||||||
side_load = opts[:side_load] || []
|
side_load = opts[:side_load] || []
|
||||||
engine_opts = Keyword.take(opts, [:verbose?, :actor, :authorize?])
|
|
||||||
|
engine_opts =
|
||||||
|
opts
|
||||||
|
|> Keyword.take([:verbose?, :actor, :authorize?])
|
||||||
|
|> Keyword.put(:transaction?, true)
|
||||||
|
|
||||||
action =
|
action =
|
||||||
if is_atom(action) and not is_nil(action) do
|
if is_atom(action) and not is_nil(action) do
|
||||||
|
|
|
@ -45,7 +45,7 @@ defmodule Ash.Engine do
|
||||||
end)
|
end)
|
||||||
|
|
||||||
transaction_result =
|
transaction_result =
|
||||||
maybe_transact(opts, requests, fn innermost_resource ->
|
maybe_transact(opts, api, fn innermost_resource ->
|
||||||
{local_requests, async_requests} = split_local_async_requests(requests)
|
{local_requests, async_requests} = split_local_async_requests(requests)
|
||||||
|
|
||||||
opts =
|
opts =
|
||||||
|
@ -55,38 +55,7 @@ defmodule Ash.Engine do
|
||||||
|> Keyword.put(:runner_pid, self())
|
|> Keyword.put(:runner_pid, self())
|
||||||
|> Keyword.put(:api, api)
|
|> Keyword.put(:api, api)
|
||||||
|
|
||||||
if async_requests == [] do
|
run_requests(async_requests, local_requests, opts, innermost_resource)
|
||||||
case Runner.run(local_requests, opts[:verbose?]) do
|
|
||||||
%{errors: errors} = runner when errors == [] ->
|
|
||||||
runner
|
|
||||||
|
|
||||||
%{errors: errors} ->
|
|
||||||
if innermost_resource do
|
|
||||||
Ash.rollback(innermost_resource, errors)
|
|
||||||
else
|
|
||||||
{:error, errors}
|
|
||||||
end
|
|
||||||
end
|
|
||||||
else
|
|
||||||
Process.flag(:trap_exit, true)
|
|
||||||
{:ok, pid} = GenServer.start(__MODULE__, opts)
|
|
||||||
_ = Process.monitor(pid)
|
|
||||||
|
|
||||||
receive do
|
|
||||||
{:pid_info, pid_info} ->
|
|
||||||
case Runner.run(local_requests, opts[:verbose?], pid, pid_info) do
|
|
||||||
%{errors: errors} = runner when errors == [] ->
|
|
||||||
runner
|
|
||||||
|
|
||||||
%{errors: errors} ->
|
|
||||||
if innermost_resource do
|
|
||||||
Ash.rollback(innermost_resource, errors)
|
|
||||||
else
|
|
||||||
{:error, errors}
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end)
|
end)
|
||||||
|
|
||||||
case transaction_result do
|
case transaction_result do
|
||||||
|
@ -99,13 +68,57 @@ defmodule Ash.Engine do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp maybe_transact(opts, requests, func) do
|
defp run_requests(async_requests, local_requests, opts, innermost_resource) do
|
||||||
|
if async_requests == [] do
|
||||||
|
run_and_return_or_rollback(local_requests, opts, innermost_resource)
|
||||||
|
else
|
||||||
|
Process.flag(:trap_exit, true)
|
||||||
|
{:ok, pid} = GenServer.start(__MODULE__, opts)
|
||||||
|
_ = Process.monitor(pid)
|
||||||
|
|
||||||
|
receive do
|
||||||
|
{:pid_info, pid_info} ->
|
||||||
|
run_and_return_or_rollback(
|
||||||
|
local_requests,
|
||||||
|
opts,
|
||||||
|
innermost_resource,
|
||||||
|
pid,
|
||||||
|
pid_info
|
||||||
|
)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp run_and_return_or_rollback(
|
||||||
|
local_requests,
|
||||||
|
opts,
|
||||||
|
innermost_resource,
|
||||||
|
pid \\ nil,
|
||||||
|
pid_info \\ %{}
|
||||||
|
) do
|
||||||
|
case Runner.run(local_requests, opts[:verbose?], pid, pid_info) do
|
||||||
|
%{errors: errors} = runner when errors == [] ->
|
||||||
|
runner
|
||||||
|
|
||||||
|
%{errors: errors} ->
|
||||||
|
rollback_or_return(innermost_resource, errors)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp rollback_or_return(innermost_resource, errors) do
|
||||||
|
if innermost_resource do
|
||||||
|
Ash.rollback(innermost_resource, errors)
|
||||||
|
else
|
||||||
|
{:error, errors}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp maybe_transact(opts, api, func) do
|
||||||
if opts[:transaction?] do
|
if opts[:transaction?] do
|
||||||
resources =
|
resources =
|
||||||
requests
|
api
|
||||||
|> Enum.map(& &1.resource)
|
|> Ash.Api.resources()
|
||||||
|> Enum.filter(&Ash.data_layer_can?(&1, :transact))
|
|> Enum.filter(&Ash.data_layer_can?(&1, :transact))
|
||||||
|> Enum.uniq()
|
|
||||||
|
|
||||||
do_in_transaction(resources, func)
|
do_in_transaction(resources, func)
|
||||||
else
|
else
|
||||||
|
|
|
@ -48,227 +48,126 @@ defmodule Ash.Filter do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def attach_other_data_layer_filters(filter, context) do
|
def run_other_data_layer_filters(resource, api, filter) do
|
||||||
context
|
reduce(filter, {:ok, filter}, fn
|
||||||
|> Map.get(:datalayer_filter)
|
%Expression{op: :or}, {:ok, filter} ->
|
||||||
|> Kernel.||(%{})
|
{:halt, {:ok, filter}}
|
||||||
|> Map.values()
|
|
||||||
|> Enum.flat_map(&Map.values/1)
|
|
||||||
|> Enum.reduce_while({:ok, filter}, fn %{data: new_filter}, {:ok, filter} ->
|
|
||||||
case add_to_filter(filter, new_filter) do
|
|
||||||
{:ok, new_filter} -> {:cont, {:ok, new_filter}}
|
|
||||||
{:error, error} -> {:halt, {:error, error}}
|
|
||||||
end
|
|
||||||
end)
|
|
||||||
end
|
|
||||||
|
|
||||||
def cross_datalayer_filter_requests(resource, api, filter, filter_requests, authorize?) do
|
%Predicate{} = expression, {:ok, filter} ->
|
||||||
map_reduce(filter, {[], true}, fn
|
|
||||||
%Expression{op: :or}, {requests, _} ->
|
|
||||||
{filter, requests, true}
|
|
||||||
|
|
||||||
%Predicate{} = expression, {requests, true} ->
|
|
||||||
expression
|
expression
|
||||||
|> relationship_paths(:ands_only)
|
|> relationship_paths(:ands_only)
|
||||||
|> filter_paths_that_change_data_layers(resource)
|
|> filter_paths_that_change_data_layers(resource)
|
||||||
|> Enum.uniq()
|
|> Enum.reduce_while({:halt, {:ok, filter}}, fn path, {:halt, {:ok, filter}} ->
|
||||||
|> Enum.reduce({expression, requests}, fn path, {expression, requests} ->
|
{for_path, without_path} = split_expression_by_relationship_path(filter, path)
|
||||||
{for_path, _without_path} = split_expression_by_relationship_path(expression, path)
|
|
||||||
|
|
||||||
relationship = Ash.relationship(resource, path)
|
relationship = Ash.relationship(resource, path)
|
||||||
|
|
||||||
query =
|
query =
|
||||||
relationship.destination
|
relationship.destination
|
||||||
|> api.query()
|
|> api.query()
|
||||||
|> Map.put(:filter, %__MODULE__{
|
|> Map.put(:filter, for_path)
|
||||||
expression: for_path,
|
|
||||||
resource: Ash.related(resource, path),
|
|
||||||
api: api
|
|
||||||
})
|
|
||||||
|
|
||||||
{new_requests, replace_with_path} =
|
add_other_datalayer_read_results(query, relationship, path, without_path)
|
||||||
other_datalayer_requests(query, relationship, path, filter_requests, authorize?)
|
|
||||||
|
|
||||||
{Map.put(expression, :__replace__, replace_with_path), {requests ++ new_requests, true}}
|
|
||||||
end)
|
end)
|
||||||
|
|
||||||
# TODO: this is an optimization. We can take all predicates that appear in a single nested `and` statement
|
%Expression{op: :and} = expression, {:ok, filter} ->
|
||||||
# and run those together
|
expression
|
||||||
# %Expression{op: :and} = expression, {requests, false} ->
|
|> relationship_paths(:ands_only)
|
||||||
# expression
|
|> filter_paths_that_change_data_layers(resource)
|
||||||
# |> relationship_paths(:ands_only)
|
|> Enum.reduce_while({:halt, {:ok, filter}}, fn path, {:halt, {:ok, filter}} ->
|
||||||
# |> filter_paths_that_change_data_layers(resource)
|
{for_path, without_path} = split_expression_by_relationship_path(filter, path)
|
||||||
# |> Enum.reduce({filter, requests}, fn path, {filter, requests} ->
|
|
||||||
# {for_path, without_path} = split_expression_by_relationship_path(filter, path)
|
|
||||||
|
|
||||||
# relationship = Ash.relationship(resource, path)
|
relationship = Ash.relationship(resource, path)
|
||||||
|
|
||||||
# query =
|
query =
|
||||||
# relationship.destination
|
relationship.destination
|
||||||
# |> api.query()
|
|> api.query()
|
||||||
# |> Map.put(:filter, %__MODULE__{
|
|> Map.put(:filter, for_path)
|
||||||
# expression: for_path,
|
|
||||||
# resource: Ash.related(resource, path),
|
|
||||||
# api: api
|
|
||||||
# })
|
|
||||||
|
|
||||||
# {new_requests, replace_with_path} =
|
add_other_datalayer_read_results(query, relationship, path, without_path)
|
||||||
# other_datalayer_requests(query, relationship, path, filter_requests, authorize?)
|
end)
|
||||||
|
|
||||||
# case without_path do
|
_, {:ok, filter} ->
|
||||||
|
{:ok, filter}
|
||||||
# end
|
|
||||||
# if is_map(without_path) do
|
|
||||||
# {Map.put(without_path, :__replace__, replace_with_path), {requests ++ new_requests, false}}
|
|
||||||
# end
|
|
||||||
# end)
|
|
||||||
|
|
||||||
expression, {requests, add_predicates?} ->
|
|
||||||
{expression, {requests, add_predicates?}}
|
|
||||||
end)
|
end)
|
||||||
end
|
end
|
||||||
|
|
||||||
defp other_datalayer_requests(query, relationship, path, filter_requests, authorize?) do
|
defp add_other_datalayer_read_results(query, relationship, path, filter_without_path) do
|
||||||
# Using System.unique_integer is essentially just a hack here
|
|
||||||
# because there are multiple requests possible per path, adding
|
|
||||||
# a unique integer here ensures that they all get a unique path
|
|
||||||
request_path =
|
|
||||||
if relationship.type == :many_to_many do
|
|
||||||
[:datalayer_filter_join, System.unique_integer(), path]
|
|
||||||
else
|
|
||||||
[:datalayer_filter, System.unique_integer(), path]
|
|
||||||
end
|
|
||||||
|
|
||||||
{requests, replace_with_path} =
|
|
||||||
if relationship.type == :many_to_many do
|
|
||||||
id = System.unique_integer()
|
|
||||||
|
|
||||||
{[many_to_many_datalayer_request(query, relationship, path, request_path, id)],
|
|
||||||
[:datalayer_filter, id, path]}
|
|
||||||
else
|
|
||||||
{[], request_path}
|
|
||||||
end
|
|
||||||
|
|
||||||
action = Ash.primary_action!(query.resource, :read)
|
|
||||||
query_path = request_path ++ [:query]
|
|
||||||
|
|
||||||
request =
|
|
||||||
Request.new(
|
|
||||||
resource: query.resource,
|
|
||||||
api: query.api,
|
|
||||||
query:
|
|
||||||
attach_other_datalayer_authorization_filter(query, path, filter_requests, authorize?),
|
|
||||||
path: request_path,
|
|
||||||
authorize?: false,
|
|
||||||
action: action,
|
|
||||||
name: "cross data layer filter: #{Enum.join(path, ",")}",
|
|
||||||
data:
|
|
||||||
Request.resolve([query_path], fn context ->
|
|
||||||
query = get_in(context, query_path)
|
|
||||||
|
|
||||||
if relationship.type == :many_to_many do
|
|
||||||
query.api.read(query)
|
|
||||||
else
|
|
||||||
case query.api.read(query) do
|
case query.api.read(query) do
|
||||||
{:ok, results} ->
|
{:ok, results} ->
|
||||||
values =
|
new_filter =
|
||||||
|
case relationship.type do
|
||||||
|
:many_to_many ->
|
||||||
|
many_to_many_read_results(results, relationship, query, path)
|
||||||
|
|
||||||
|
_ ->
|
||||||
|
results
|
||||||
|
|> Enum.map(&Map.get(&1, relationship.destination_field))
|
||||||
|
|> Enum.reject(&is_nil/1)
|
||||||
|
|> record_filters_or_false(relationship)
|
||||||
|
|> put_at_path(:lists.droplast(path))
|
||||||
|
end
|
||||||
|
|
||||||
|
case add_to_filter(filter_without_path, new_filter) do
|
||||||
|
{:ok, filter} -> {:cont, {:halt, {:ok, filter}}}
|
||||||
|
{:error, error} -> {:halt, {:return, {:error, error}}}
|
||||||
|
end
|
||||||
|
|
||||||
|
{:error, error} ->
|
||||||
|
{:halt, {:return, {:error, error}}}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp record_filters_or_false(records, relationship) do
|
||||||
|
case records do
|
||||||
|
[] ->
|
||||||
|
false
|
||||||
|
|
||||||
|
[value] ->
|
||||||
|
[{relationship.source_field, value}]
|
||||||
|
|
||||||
|
values ->
|
||||||
|
[{relationship.source_field, [in: values]}]
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp many_to_many_read_results(results, relationship, query, path) do
|
||||||
|
destination_values =
|
||||||
results
|
results
|
||||||
|> Enum.map(&Map.get(&1, relationship.destination_field))
|
|> Enum.map(&Map.get(&1, relationship.destination_field))
|
||||||
|> Enum.reject(&is_nil/1)
|
|> Enum.reject(&is_nil/1)
|
||||||
|
|
||||||
Predicate.new(
|
join_query =
|
||||||
query.resource,
|
|
||||||
Ash.attribute(query.resource, relationship.destination_field),
|
|
||||||
In,
|
|
||||||
values,
|
|
||||||
:lists.droplast(path)
|
|
||||||
)
|
|
||||||
|
|
||||||
{:error, error} ->
|
|
||||||
{:error, error}
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end)
|
|
||||||
)
|
|
||||||
|
|
||||||
{[request | requests], replace_with_path}
|
|
||||||
end
|
|
||||||
|
|
||||||
defp attach_other_datalayer_authorization_filter(query, _path, _filter_requests, false),
|
|
||||||
do: query
|
|
||||||
|
|
||||||
defp attach_other_datalayer_authorization_filter(query, _path, [], _), do: query
|
|
||||||
|
|
||||||
defp attach_other_datalayer_authorization_filter(query, path, filter_requests, true) do
|
|
||||||
if [:filter, path] in Enum.map(filter_requests, & &1.path) do
|
|
||||||
filter_dependency = [:filter, path, :authorization_filter]
|
|
||||||
|
|
||||||
Request.resolve([filter_dependency], fn context ->
|
|
||||||
authorization_filter = get_in(context, filter_dependency)
|
|
||||||
{:ok, Ash.Query.filter(query, authorization_filter)}
|
|
||||||
end)
|
|
||||||
else
|
|
||||||
query
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
defp many_to_many_datalayer_request(query, relationship, path, many_to_many_request_path, id) do
|
|
||||||
action = Ash.primary_action!(relationship.through, :read)
|
|
||||||
dependency = many_to_many_request_path ++ [:data]
|
|
||||||
request_path = [:datalayer_filter, id, path]
|
|
||||||
query_dependency = request_path ++ [:query]
|
|
||||||
|
|
||||||
Request.new(
|
|
||||||
resource: relationship.through,
|
|
||||||
api: query.api,
|
|
||||||
path: request_path,
|
|
||||||
authorize?: false,
|
|
||||||
action: action,
|
|
||||||
name: "cross data layer `through` filter: #{Enum.join(path, ",")}",
|
|
||||||
query:
|
|
||||||
Request.resolve([dependency], fn context ->
|
|
||||||
data = get_in(context, dependency) || []
|
|
||||||
|
|
||||||
destination_values =
|
|
||||||
data
|
|
||||||
|> Enum.map(&Map.get(&1, relationship.destination_field))
|
|
||||||
|> Enum.reject(&is_nil/1)
|
|
||||||
|
|
||||||
{:ok,
|
|
||||||
relationship.through
|
relationship.through
|
||||||
|> query.api.query()
|
|> query.api.query()
|
||||||
|> Ash.Query.filter([
|
|> Ash.Query.filter([
|
||||||
{relationship.destination_field_on_join_table, [in: destination_values]}
|
{relationship.destination_field_on_join_table, [in: destination_values]}
|
||||||
])}
|
])
|
||||||
end),
|
|
||||||
data:
|
|
||||||
Request.resolve([query_dependency], fn context ->
|
|
||||||
query = get_in(context, query_dependency)
|
|
||||||
|
|
||||||
case query.api.read(query) do
|
case query.api.read(join_query) do
|
||||||
{:ok, results} ->
|
{:ok, results} ->
|
||||||
values =
|
|
||||||
results
|
results
|
||||||
|> Enum.map(&Map.get(&1, relationship.source_field_on_join_table))
|
|> Enum.map(&Map.get(&1, relationship.source_field_on_join_table))
|
||||||
|> Enum.reject(&is_nil/1)
|
|> Enum.reject(&is_nil/1)
|
||||||
|
|> case do
|
||||||
|
[] ->
|
||||||
|
false
|
||||||
|
|
||||||
Predicate.new(
|
[value] ->
|
||||||
query.resource,
|
[{relationship.source_field, value}]
|
||||||
Ash.attribute(query.resource, relationship.destination_field),
|
|
||||||
In,
|
values ->
|
||||||
values,
|
[{relationship.source_field, [in: values]}]
|
||||||
:lists.droplast(path)
|
end
|
||||||
)
|
|> put_at_path(:lists.droplast(path))
|
||||||
|
|
||||||
{:error, error} ->
|
{:error, error} ->
|
||||||
{:error, error}
|
{:error, error}
|
||||||
end
|
end
|
||||||
end)
|
|
||||||
)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
defp filter_paths_that_change_data_layers(paths, resource, acc \\ [])
|
defp filter_paths_that_change_data_layers(paths, resource, acc \\ [])
|
||||||
defp filter_paths_that_change_data_layers([], _resource, acc), do: Enum.uniq(acc)
|
defp filter_paths_that_change_data_layers([], _resource, acc), do: acc
|
||||||
|
|
||||||
defp filter_paths_that_change_data_layers([path | rest], resource, acc) do
|
defp filter_paths_that_change_data_layers([path | rest], resource, acc) do
|
||||||
case shortest_path_to_changed_data_layer(resource, path) do
|
case shortest_path_to_changed_data_layer(resource, path) do
|
||||||
|
@ -281,10 +180,6 @@ defmodule Ash.Filter do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def path_crosses_datalayer?(resource, path) do
|
|
||||||
shortest_path_to_changed_data_layer(resource, path) != :error
|
|
||||||
end
|
|
||||||
|
|
||||||
defp shortest_path_to_changed_data_layer(resource, path, acc \\ [])
|
defp shortest_path_to_changed_data_layer(resource, path, acc \\ [])
|
||||||
defp shortest_path_to_changed_data_layer(_resource, [], _acc), do: :error
|
defp shortest_path_to_changed_data_layer(_resource, [], _acc), do: :error
|
||||||
|
|
||||||
|
@ -292,15 +187,25 @@ defmodule Ash.Filter do
|
||||||
relationship = Ash.relationship(resource, relationship)
|
relationship = Ash.relationship(resource, relationship)
|
||||||
data_layer = Ash.data_layer(relationship.destination)
|
data_layer = Ash.data_layer(relationship.destination)
|
||||||
|
|
||||||
if Ash.data_layer_can?(resource, :join) && data_layer == Ash.data_layer(resource) &&
|
if relationship.type == :many_to_many do
|
||||||
(relationship.type != :many_to_many ||
|
if data_layer == Ash.data_layer(resource) &&
|
||||||
data_layer == Ash.data_layer(relationship.through)) do
|
data_layer == Ash.data_layer(relationship.through) &&
|
||||||
|
Ash.data_layer_can?(resource, :join) do
|
||||||
shortest_path_to_changed_data_layer(relationship.destination, rest, [
|
shortest_path_to_changed_data_layer(relationship.destination, rest, [
|
||||||
relationship.name | acc
|
relationship.name | acc
|
||||||
])
|
])
|
||||||
else
|
else
|
||||||
{:ok, Enum.reverse([relationship.name | acc])}
|
{:ok, Enum.reverse([relationship.name | acc])}
|
||||||
end
|
end
|
||||||
|
else
|
||||||
|
if data_layer == Ash.data_layer(resource) && Ash.data_layer_can?(resource, :join) do
|
||||||
|
shortest_path_to_changed_data_layer(relationship.destination, rest, [
|
||||||
|
relationship.name | acc
|
||||||
|
])
|
||||||
|
else
|
||||||
|
{:ok, Enum.reverse([relationship.name | acc])}
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp put_at_path(value, []), do: value
|
defp put_at_path(value, []), do: value
|
||||||
|
@ -453,72 +358,134 @@ defmodule Ash.Filter do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def reduce(filter, acc \\ nil, func) do
|
def reduce(filter, acc \\ nil, func)
|
||||||
map_reduce(filter, acc, fn expression, acc ->
|
def reduce(%__MODULE__{expression: nil}, acc, _), do: acc
|
||||||
|
|
||||||
|
def reduce(%__MODULE__{expression: expression}, acc, func) do
|
||||||
case func.(expression, acc) do
|
case func.(expression, acc) do
|
||||||
{:halt, acc} ->
|
{:halt, acc} ->
|
||||||
{:halt, {expression, acc}}
|
acc
|
||||||
|
|
||||||
|
{:return, value} ->
|
||||||
|
value
|
||||||
|
|
||||||
acc ->
|
acc ->
|
||||||
{expression, acc}
|
case do_reduce(expression, acc, func) do
|
||||||
|
{:halt, acc} -> acc
|
||||||
|
{:return, value} -> value
|
||||||
|
acc -> acc
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
def map_reduce(filter, acc \\ nil, func)
|
def reduce(expression, acc, func) do
|
||||||
def map_reduce(%__MODULE__{expression: nil} = filter, acc, _), do: {filter, acc}
|
case func.(expression, acc) do
|
||||||
|
{:halt, acc} ->
|
||||||
|
acc
|
||||||
|
|
||||||
def map_reduce(%__MODULE__{expression: expression} = filter, acc, func) do
|
{:return, value} ->
|
||||||
{expression, acc} = map_reduce(expression, acc, func)
|
value
|
||||||
|
|
||||||
{%{filter | expression: expression}, acc}
|
acc ->
|
||||||
|
case do_reduce(expression, acc, func) do
|
||||||
|
{:halt, acc} -> acc
|
||||||
|
{:return, value} -> value
|
||||||
|
acc -> acc
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def map_reduce(expression, acc, func) do
|
def do_reduce(expression, acc, func) do
|
||||||
{expression, acc} = func.(expression, acc)
|
|
||||||
do_map_reduce(expression, acc, func)
|
|
||||||
end
|
|
||||||
|
|
||||||
def do_map_reduce(expression, acc, func) do
|
|
||||||
case expression do
|
case expression do
|
||||||
%Expression{left: left, right: right} = expression ->
|
%Expression{} = expression ->
|
||||||
{new_left, acc} = func.(left, acc)
|
do_reduce_expression(expression, acc, func)
|
||||||
{new_right, acc} = func.(right, acc)
|
|
||||||
|
|
||||||
{new_left, acc} = do_map_reduce(new_left, acc, func)
|
%Not{expression: not_expr} ->
|
||||||
{new_right, acc} = do_map_reduce(new_right, acc, func)
|
case func.(not_expr, acc) do
|
||||||
|
{:halt, acc} ->
|
||||||
|
acc
|
||||||
|
|
||||||
{%{expression | left: new_left, right: new_right}, acc}
|
{:return, value} ->
|
||||||
|
{:return, value}
|
||||||
|
|
||||||
%Not{expression: not_expr} = not_struct ->
|
acc ->
|
||||||
{expression, acc} = func.(not_expr, acc)
|
do_reduce(not_expr, acc, func)
|
||||||
{expression, acc} = do_map_reduce(expression, acc, func)
|
end
|
||||||
{%{not_struct | expression: expression}, acc}
|
|
||||||
|
{:return, value} ->
|
||||||
|
{:return, value}
|
||||||
|
|
||||||
|
{:halt, value} ->
|
||||||
|
{:halt, value}
|
||||||
|
|
||||||
other ->
|
other ->
|
||||||
func.(other, acc)
|
func.(other, acc)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
# defp split_expression_by_relationship_path(%{expression: expression} = filter, _path)
|
defp do_reduce_expression(%Expression{left: left, right: right}, acc, func) do
|
||||||
# when expression in [nil, true, false] do
|
case func.(right, acc) do
|
||||||
# {filter, filter}
|
{:halt, acc} ->
|
||||||
# end
|
case func.(left, acc) do
|
||||||
|
{:return, value} ->
|
||||||
|
{:return, value}
|
||||||
|
|
||||||
# defp split_expression_by_relationship_path(filter, path) do
|
{:halt, acc} ->
|
||||||
# {for_path, without_path} = do_split_expression_by_relationship_path(filter.expression, path)
|
acc
|
||||||
|
|
||||||
# {%__MODULE__{
|
acc ->
|
||||||
# api: filter.api,
|
do_reduce(left, acc, func)
|
||||||
# resource: Ash.related(filter.resource, path),
|
end
|
||||||
# expression: for_path
|
|
||||||
# },
|
{:return, value} ->
|
||||||
# %__MODULE__{
|
{:return, value}
|
||||||
# api: filter.api,
|
|
||||||
# resource: filter.resource,
|
acc ->
|
||||||
# expression: without_path
|
continue_reduce(left, right, acc, func)
|
||||||
# }}
|
end
|
||||||
# end
|
end
|
||||||
|
|
||||||
|
defp continue_reduce(left, right, acc, func) do
|
||||||
|
case func.(left, acc) do
|
||||||
|
{:halt, acc} ->
|
||||||
|
do_reduce(right, acc, func)
|
||||||
|
|
||||||
|
{:return, value} ->
|
||||||
|
{:return, value}
|
||||||
|
|
||||||
|
acc ->
|
||||||
|
case do_reduce(left, acc, func) do
|
||||||
|
{:halt, acc} ->
|
||||||
|
{:halt, acc}
|
||||||
|
|
||||||
|
{:return, acc} ->
|
||||||
|
{:return, acc}
|
||||||
|
|
||||||
|
acc ->
|
||||||
|
do_reduce(right, acc, func)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp split_expression_by_relationship_path(%{expression: expression} = filter, _path)
|
||||||
|
when expression in [nil, true, false] do
|
||||||
|
{filter, filter}
|
||||||
|
end
|
||||||
|
|
||||||
|
defp split_expression_by_relationship_path(filter, path) do
|
||||||
|
{for_path, without_path} = do_split_expression_by_relationship_path(filter.expression, path)
|
||||||
|
|
||||||
|
{%__MODULE__{
|
||||||
|
api: filter.api,
|
||||||
|
resource: Ash.related(filter.resource, path),
|
||||||
|
expression: for_path
|
||||||
|
},
|
||||||
|
%__MODULE__{
|
||||||
|
api: filter.api,
|
||||||
|
resource: filter.resource,
|
||||||
|
expression: without_path
|
||||||
|
}}
|
||||||
|
end
|
||||||
|
|
||||||
defp filter_expression_by_relationship_path(filter, path) do
|
defp filter_expression_by_relationship_path(filter, path) do
|
||||||
%__MODULE__{
|
%__MODULE__{
|
||||||
|
@ -528,25 +495,26 @@ defmodule Ash.Filter do
|
||||||
}
|
}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp split_expression_by_relationship_path(
|
defp do_split_expression_by_relationship_path(
|
||||||
%Expression{op: op, left: left, right: right},
|
%Expression{op: op, left: left, right: right},
|
||||||
path
|
path
|
||||||
) do
|
) do
|
||||||
{new_for_path_left, new_without_path_left} = split_expression_by_relationship_path(left, path)
|
{new_for_path_left, new_without_path_left} =
|
||||||
|
do_split_expression_by_relationship_path(left, path)
|
||||||
|
|
||||||
{new_for_path_right, new_without_path_right} =
|
{new_for_path_right, new_without_path_right} =
|
||||||
split_expression_by_relationship_path(right, path)
|
do_split_expression_by_relationship_path(right, path)
|
||||||
|
|
||||||
{Expression.new(op, new_for_path_left, new_for_path_right),
|
{Expression.new(op, new_for_path_left, new_for_path_right),
|
||||||
Expression.new(op, new_without_path_left, new_without_path_right)}
|
Expression.new(op, new_without_path_left, new_without_path_right)}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp split_expression_by_relationship_path(%Not{expression: expression}, path) do
|
defp do_split_expression_by_relationship_path(%Not{expression: expression}, path) do
|
||||||
{new_for_path, new_without_path} = split_expression_by_relationship_path(expression, path)
|
{new_for_path, new_without_path} = do_split_expression_by_relationship_path(expression, path)
|
||||||
{Not.new(new_for_path), Not.new(new_without_path)}
|
{Not.new(new_for_path), Not.new(new_without_path)}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp split_expression_by_relationship_path(
|
defp do_split_expression_by_relationship_path(
|
||||||
%Predicate{relationship_path: predicate_path} = predicate,
|
%Predicate{relationship_path: predicate_path} = predicate,
|
||||||
path
|
path
|
||||||
) do
|
) do
|
||||||
|
|
2
mix.exs
2
mix.exs
|
@ -18,7 +18,7 @@ defmodule Ash.MixProject do
|
||||||
elixirc_paths: elixirc_paths(Mix.env()),
|
elixirc_paths: elixirc_paths(Mix.env()),
|
||||||
package: package(),
|
package: package(),
|
||||||
deps: deps(),
|
deps: deps(),
|
||||||
dialyzer: [plt_add_apps: [:mix]],
|
dialyzer: [plt_add_apps: [:mix, :mnesia]],
|
||||||
test_coverage: [tool: ExCoveralls],
|
test_coverage: [tool: ExCoveralls],
|
||||||
preferred_cli_env: [
|
preferred_cli_env: [
|
||||||
coveralls: :test,
|
coveralls: :test,
|
||||||
|
|
|
@ -1,6 +1,8 @@
|
||||||
defmodule Ash.Test.Filter.FilterInteractionTest do
|
defmodule Ash.Test.Filter.FilterInteractionTest do
|
||||||
use ExUnit.Case, async: false
|
use ExUnit.Case, async: false
|
||||||
|
|
||||||
|
alias Ash.DataLayer.Mnesia
|
||||||
|
|
||||||
defmodule Profile do
|
defmodule Profile do
|
||||||
@moduledoc false
|
@moduledoc false
|
||||||
use Ash.Resource, data_layer: Ash.DataLayer.Ets
|
use Ash.Resource, data_layer: Ash.DataLayer.Ets
|
||||||
|
@ -118,7 +120,7 @@ defmodule Ash.Test.Filter.FilterInteractionTest do
|
||||||
end
|
end
|
||||||
|
|
||||||
setup do
|
setup do
|
||||||
Ash.DataLayer.Mnesia.start(Api)
|
Mnesia.start(Api)
|
||||||
|
|
||||||
on_exit(fn ->
|
on_exit(fn ->
|
||||||
:mnesia.stop()
|
:mnesia.stop()
|
||||||
|
@ -171,8 +173,7 @@ defmodule Ash.Test.Filter.FilterInteractionTest do
|
||||||
post1 =
|
post1 =
|
||||||
Api.create!(Post,
|
Api.create!(Post,
|
||||||
attributes: %{title: "one"},
|
attributes: %{title: "one"},
|
||||||
relationships: %{related_posts: [post2]},
|
relationships: %{related_posts: [post2]}
|
||||||
verbose?: true
|
|
||||||
)
|
)
|
||||||
|
|
||||||
query =
|
query =
|
||||||
|
@ -180,6 +181,8 @@ defmodule Ash.Test.Filter.FilterInteractionTest do
|
||||||
|> Api.query()
|
|> Api.query()
|
||||||
|> Ash.Query.filter(related_posts: [title: "two"])
|
|> Ash.Query.filter(related_posts: [title: "two"])
|
||||||
|
|
||||||
|
post1 = Api.reload!(post1)
|
||||||
|
|
||||||
assert [^post1] = Api.read!(query)
|
assert [^post1] = Api.read!(query)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
Loading…
Reference in a new issue