handle filter

lib/graphql/resolver.ex

@@ -507,15 +507,28 @@ defmodule AshGraphql.Graphql.Resolver do
   end
 
   def resolve(
-        %{arguments: _arguments, context: context, root_value: data} = resolution,
+        %{arguments: arguments, context: context, root_value: notification} = resolution,
         {domain, resource, %AshGraphql.Resource.Subscription{read_action: read_action}, _input?}
       ) do
+    dbg(NOTIFICATION: notification)
+    data = notification.data
+
     read_action =
       read_action || Ash.Resource.Info.primary_action!(resource, :read).name
 
+    query =
+      Ash.Resource.Info.primary_key(resource)
+      |> Enum.reduce(resource, fn key, query ->
+        value = Map.get(data, key)
+        Ash.Query.filter(query, ^ref(key) == ^value)
+      end)
+
+    query =
+      Ash.Query.do_filter(query, massage_filter(query.resource, Map.get(arguments, :filter)))
+
     query =
       AshGraphql.Subscription.query_for_subscription(
-        resource
+        query
         |> Ash.Query.for_read(read_action, %{},
           actor: Map.get(context, :actor),
           tenant: Map.get(context, :tenant)
@@ -524,13 +537,6 @@ defmodule AshGraphql.Graphql.Resolver do
         resolution
       )
 
-    query =
-      Ash.Resource.Info.primary_key(resource)
-      |> Enum.reduce(query, fn key, query ->
-        value = Map.get(data, key)
-        Ash.Query.filter(query, ^ref(key) == ^value)
-      end)
-
     case query |> Ash.read_one() do
       # should only happen if a resource is created/updated and the subscribed user is not allowed to see it
       {:ok, nil} ->
@@ -1641,9 +1647,9 @@ defmodule AshGraphql.Graphql.Resolver do
      end)}
   end
 
-  defp massage_filter(_resource, nil), do: nil
+  def massage_filter(_resource, nil), do: nil
 
-  defp massage_filter(resource, filter) when is_map(filter) do
+  def massage_filter(resource, filter) when is_map(filter) do
     Map.new(filter, fn {key, value} ->
       cond do
         rel = Ash.Resource.Info.relationship(resource, key) ->
@@ -1658,7 +1664,7 @@ defmodule AshGraphql.Graphql.Resolver do
     end)
   end
 
-  defp massage_filter(_resource, other), do: other
+  def massage_filter(_resource, other), do: other
 
   defp calc_input(key, value) do
     case Map.fetch(value, :input) do

lib/resource/resource.ex

@@ -1163,9 +1163,7 @@ defmodule AshGraphql.Resource do
     |> subscriptions()
     |> Enum.map(fn %Subscription{name: name} = subscription ->
       %Absinthe.Blueprint.Schema.FieldDefinition{
-        arguments:
-          args(:subscription, resource, nil, schema, nil)
-          |> IO.inspect(label: "args"),
+        arguments: args(:subscription, resource, nil, schema, nil),
         identifier: name,
         name: to_string(name),
         config:

lib/resource/subscription/actor.ex

@@ -1,3 +1,5 @@
 defmodule AshGraphql.Resource.Subscription.Actor do
+  # I'd like to have the typespsay that actor can be anything
+  # but that the input and output must be the same
   @callback author(actor :: any) :: actor :: any
 end

lib/subscription/config.ex

@@ -15,6 +15,8 @@ defmodule AshGraphql.Subscription.Config do
 
         actor =
           if is_function(@subscription.actor) do
+            # might be nice to also pass in the subscription, that way you could potentially
+            # deduplicate on an action basis as well if you wanted to
             @subscription.actor.(context[:actor])
           else
             context[:actor]
@@ -26,6 +28,10 @@ defmodule AshGraphql.Subscription.Config do
         case Ash.can(
                @resource
                |> Ash.Query.new()
+               # not sure if we need this here
+               |> Ash.Query.do_filter(
+                 AshGraphql.Graphql.Resolver.massage_filter(@resource, Map.get(args, :filter))
+               )
                |> Ash.Query.set_tenant(context[:tenant])
                |> Ash.Query.for_read(read_action),
                actor,

lib/subscription/endpoint.ex

@@ -7,13 +7,11 @@ defmodule AshGraphql.Subscription.Endpoint do
 
       require Logger
 
-      def run_docset(pubsub, docs_and_topics, mutation_result) do
-        dbg(mutation_result, structs: false)
-
+      def run_docset(pubsub, docs_and_topics, notification) do
         for {topic, key_strategy, doc} <- docs_and_topics do
           try do
             pipeline =
-              Absinthe.Subscription.Local.pipeline(doc, mutation_result.data)
+              Absinthe.Subscription.Local.pipeline(doc, notification)
 
             {:ok, %{result: data}, _} = Absinthe.Pipeline.run(doc.source, pipeline)
 
@@ -43,12 +41,12 @@ defmodule AshGraphql.Subscription.Endpoint do
         # return any data we do not send the error to the client
         # because it would just expose unnecessary information
         # and the user can not really do anything usefull with it
-        errors
-        |> List.wrap()
-        |> Enum.any?(fn error -> Map.get(error, :code) in ["forbidden", "not_found"] end)
+        not (errors
+             |> List.wrap()
+             |> Enum.any?(fn error -> Map.get(error, :code) in ["forbidden", "not_found"] end))
       end
 
-      defp is_forbidden(_), do: false
+      defp should_send?(_), do: true
 
       defp get_filter(topic) do
         [_, rest] = String.split(topic, "__absinthe__:doc:")

lib/subscription/notifier.ex

@@ -6,8 +6,6 @@ defmodule AshGraphql.Subscription.Notifier do
   def notify(notification) do
     pub_sub = Info.subscription_pubsub(notification.resource)
 
-    dbg(notification, structs: false)
-
     for subscription <- AshGraphql.Resource.Info.subscriptions(notification.resource) do
       if is_nil(subscription.actions) or
            notification.action.name in List.wrap(subscription.actions) do