fix: various fixes from json:api integration

lib/ash/engine/engine.ex

@@ -161,6 +161,10 @@ defmodule Ash.Engine do
     {:ok, state, {:continue, :spawn_requests}}
   end
 
+  def send_wont_receive(pid, caller_path, request_path, field) do
+    GenServer.cast(pid, {:wont_receive, caller_path, request_path, field})
+  end
+
   def handle_continue(:spawn_requests, state) do
     log(state, "Spawning request processes", :debug)
 
@@ -241,11 +245,7 @@ defmodule Ash.Engine do
     |> maybe_shutdown()
   end
 
-  def send_wont_receive(pid, caller_path, request_path, field) do
-    GenServer.cast(pid, {:wont_receive, caller_path, request_path, field})
-  end
-
-  def handle_info({:error, error, request_handler_state}, state) do
+  def handle_cast({:error, error, request_handler_state}, state) do
     state
     |> log("Error received from request_handler #{inspect(error)}")
     |> move_to_error(request_handler_state.request.path)

lib/ash/engine/request_handler.ex

@@ -5,6 +5,7 @@ defmodule Ash.Engine.RequestHandler do
   use GenServer
   require Logger
 
+  alias Ash.Engine
   alias Ash.Engine.Request
 
   def init(opts) do
@@ -36,7 +37,11 @@ defmodule Ash.Engine.RequestHandler do
         {:noreply, new_state, {:continue, :next}}
 
       {:error, error, new_request} ->
-        {:stop, {:error, error, %{new_request | state: :error}}, state}
+        new_request = %{new_request | state: :error}
+        new_state = %{state | request: new_request}
+        notify_error(new_state, error)
+
+        {:noreply, new_state}
 
       {:already_complete, new_request, notifications, dependencies} ->
         new_state = %{state | request: new_request}
@@ -66,10 +71,21 @@ defmodule Ash.Engine.RequestHandler do
   def handle_cast({:wont_receive, _receiver_path, path, field}, state) do
     case Request.wont_receive(state.request, path, field) do
       {:stop, :dependency_failed, request} ->
-        {:stop, :shutdown, request}
+        new_state = %{state | request: %{request | state: :error}}
+        notify_error(new_state, :dependency_failed)
+        {:noreply, new_state}
     end
   end
 
+  def handle_cast(
+        {:send_field, receiver_path, pid, dep},
+        %{request: %{path: path, state: :error}} = state
+      ) do
+    Engine.send_wont_receive(pid, receiver_path, path, List.last(dep))
+
+    {:noreply, state}
+  end
+
   def handle_cast({:send_field, receiver_path, _pid, dep}, state) do
     field = List.last(dep)
 
@@ -92,7 +108,10 @@ defmodule Ash.Engine.RequestHandler do
         {:noreply, new_state}
 
       {:error, error, new_request} ->
-        {:stop, {:error, error}, %{new_request | state: :error}}
+        new_request = %{new_request | state: :error}
+        new_state = %{state | request: new_request}
+        notify_error(new_state, error)
+        {:noreply, new_state}
     end
   end
 
@@ -103,6 +122,11 @@ defmodule Ash.Engine.RequestHandler do
     end
   end
 
+  defp notify_error(state, error) do
+    log(state, "Request error, notifying engine")
+    GenServer.cast(state.engine_pid, {:error, error, state})
+  end
+
   defp notify(state, notifications) do
     Enum.each(notifications, fn {receiver_path, request_path, field, value} ->
       receiver_path =

lib/ash/engine/runner.ex

@@ -162,7 +162,7 @@ defmodule Ash.Engine.Runner do
   end
 
   defp fake_handle_cast({:wont_receive, receiver_path, path, field}, state) do
-    request = Map.get(state.requests, receiver_path)
+    request = Enum.find(state.requests, &(&1.path == receiver_path))
 
     case Request.wont_receive(request, path, field) do
       {:stop, :dependency_failed, new_request} ->

lib/ash/filter/filter.ex

@@ -6,6 +6,7 @@ defmodule Ash.Filter do
   terminating in a `%Ash.Filter.Predicate{}` struct. An expression is simply a boolean operator
   and the left and right hand side of that operator.
   """
+  alias Ash.Actions.SideLoad
   alias Ash.Engine.Request
 
   alias Ash.Error.Query.{
@@ -30,6 +31,10 @@ defmodule Ash.Filter do
     greater_than: GreaterThan
   ]
 
+  @string_builtin_predicates Enum.into(@built_in_predicates, %{}, fn {key, value} ->
+                               {to_string(key), value}
+                             end)
+
   defstruct [:resource, :expression]
 
   @type t :: %__MODULE__{}
@@ -318,7 +323,7 @@ defmodule Ash.Filter do
     filter
     |> Ash.Filter.relationship_paths()
     |> Enum.map(fn path ->
-      {path, filter_expression_by_relationship_path(filter, path)}
+      {path, filter_expression_by_relationship_path(filter, path, true)}
     end)
     |> Enum.reduce_while({:ok, []}, fn {path, scoped_filter}, {:ok, requests} ->
       %{resource: resource} = scoped_filter
@@ -331,7 +336,37 @@ defmodule Ash.Filter do
           Request.new(
             resource: resource,
             api: api,
-            query: query,
+            query:
+              Request.resolve(
+                [[:data, :authorization_filter]],
+                fn %{
+                     data: %{
+                       authorization_filter: authorization_filter
+                     }
+                   } ->
+                  if authorization_filter do
+                    relationship =
+                      Ash.Resource.relationship(
+                        resource,
+                        List.first(path)
+                      )
+
+                    case SideLoad.reverse_relationship_path(
+                           relationship,
+                           tl(path)
+                         ) do
+                      :error ->
+                        {:ok, query}
+
+                      {:ok, reverse_relationship} ->
+                        filter = put_at_path(authorization_filter, reverse_relationship)
+                        {:ok, Ash.Query.filter(query, filter)}
+                    end
+                  else
+                    {:ok, query}
+                  end
+                end
+              ),
             async?: false,
             path: [:filter, path],
             strict_check_only?: true,
@@ -880,7 +915,7 @@ defmodule Ash.Filter do
     end)
   end
 
-  defp parse_predicates(value, field, context) when not is_list(value) do
+  defp parse_predicates(value, field, context) when not is_list(value) and not is_map(value) do
     parse_predicates([eq: value], field, context)
   end
 
@@ -892,9 +927,9 @@ defmodule Ash.Filter do
         []
       )
 
-    if Keyword.keyword?(values) do
+    if is_map(values) || Keyword.keyword?(values) do
       Enum.reduce_while(values, {:ok, nil}, fn {key, value}, {:ok, expression} ->
-        case @built_in_predicates[key] || data_layer_predicates[key] do
+        case get_predicate(key, data_layer_predicates) do
           value when value in [nil, []] ->
             error = NoSuchFilterPredicate.exception(key: key, resource: context.resource)
             {:halt, {:error, error}}
@@ -921,6 +956,23 @@ defmodule Ash.Filter do
     end
   end
 
+  defp get_predicate(key, data_layer_predicates) when is_atom(key) do
+    @built_in_predicates[key] || data_layer_predicates[key]
+  end
+
+  defp get_predicate(key, data_layer_predicates) when is_binary(key) do
+    Map.get(@string_builtin_predicates, key) ||
+      Enum.find_value(data_layer_predicates, fn {pred, value} ->
+        if to_string(pred) == key do
+          value
+        else
+          false
+        end
+      end)
+  end
+
+  defp get_predicate(_, _), do: nil
+
   defimpl Inspect do
     import Inspect.Algebra
 

lib/ash/query/aggregate.ex

@@ -39,6 +39,8 @@ defmodule Ash.Query.Aggregate do
 
   defp default_value(:count), do: 0
 
+  defp validate_query(nil), do: {:ok, nil}
+
   defp validate_query(query) do
     cond do
       query.side_load != [] ->
@@ -65,63 +67,6 @@ defmodule Ash.Query.Aggregate do
   def kind_to_type(:count), do: {:ok, Ash.Type.Integer}
   def kind_to_type(kind), do: {:error, "Invalid aggregate kind: #{kind}"}
 
-  # def requests_with_initial_data(query, authorizing?) do
-  #   query.aggregates
-  #   |> Map.values()
-  #   |> Enum.group_by(& &1.relationship_path)
-  #   |> Enum.reduce({[], []}, fn {relationship_path, aggregates},
-  #                               {auth_requests, value_requests} ->
-  #     related = Ash.Resource.related(initial_query.resource, relationship_path)
-
-  #     relationship =
-  #       Ash.Resource.relationship(
-  #         initial_query.resource,
-  #         List.first(relationship_path)
-  #       )
-
-  #     remaining_path = List.delete_at(relationship_path, 0)
-
-  #     reverse_relationship =
-  #       case SideLoad.reverse_relationship_path(relationship, remaining_path) do
-  #         :error ->
-  #           nil
-
-  #         {:ok, reverse_relationship} ->
-  #           reverse_relationship
-  #       end
-
-  #     auth_request =
-  #       if authorizing? do
-  #         auth_request(related, initial_query, reverse_relationship, relationship_path)
-  #       else
-  #         nil
-  #       end
-
-  #     new_auth_requests =
-  #       if auth_request do
-  #         [auth_request | auth_requests]
-  #       else
-  #         auth_requests
-  #       end
-
-  #     if reverse_relationship do
-  #       request =
-  #         value_request(
-  #           initial_query,
-  #           related,
-  #           reverse_relationship,
-  #           relationship_path,
-  #           aggregates,
-  #           auth_request
-  #         )
-
-  #       {new_auth_requests, [request | value_requests]}
-  #     else
-  #       raise "Unimplemented"
-  #     end
-  #   end)
-  # end
-
   def requests(initial_query, can_be_in_query?, authorizing?) do
     initial_query.aggregates
     |> Map.values()
@@ -136,10 +81,8 @@ defmodule Ash.Query.Aggregate do
           List.first(relationship_path)
         )
 
-      remaining_path = List.delete_at(relationship_path, 0)
-
       {in_query?, reverse_relationship} =
-        case SideLoad.reverse_relationship_path(relationship, remaining_path) do
+        case SideLoad.reverse_relationship_path(relationship, tl(relationship_path)) do
           :error ->
             {can_be_in_query?, nil}
 

lib/ash/query/query.ex

@@ -117,6 +117,9 @@ defmodule Ash.Query do
           {:error, error} ->
             add_error(query, :aggregates, Ash.Error.to_ash_error(error))
         end
+
+      true ->
+        add_error(query, :load, "Could not load #{inspect(field)}")
     end
   end
 

lib/ash/resource.ex

@@ -142,6 +142,12 @@ defmodule Ash.Resource do
     Extension.get_entities(resource, [:aggregates])
   end
 
+  def aggregate(resource, name) when is_bitstring(name) do
+    resource
+    |> aggregates()
+    |> Enum.find(&(to_string(&1.name) == name))
+  end
+
   def aggregate(resource, name) do
     resource
     |> aggregates()