support action_types and arguments in read actions

lib/graphql/resolver.ex

@@ -540,6 +540,7 @@ defmodule AshGraphql.Graphql.Resolver do
           cond do
             notification.action.type in [:create, :update] ->
               data = notification.data
+              {filter, args} = Map.pop(args, :filter)
 
               read_action =
                 read_action || Ash.Resource.Info.primary_action!(resource, :read).name
@@ -554,13 +555,13 @@ defmodule AshGraphql.Graphql.Resolver do
               query =
                 Ash.Query.do_filter(
                   query,
-                  massage_filter(query.resource, Map.get(args, :filter))
+                  massage_filter(query.resource, filter)
                 )
 
               query =
                 AshGraphql.Subscription.query_for_subscription(
                   query
-                  |> Ash.Query.for_read(read_action, %{}, opts),
+                  |> Ash.Query.for_read(read_action, args, opts),
                   domain,
                   resolution,
                   subscription_result_type(name),

lib/resource/resource.ex

@@ -1217,8 +1217,12 @@ defmodule AshGraphql.Resource do
     |> Enum.map(fn %Subscription{name: name} = subscription ->
       result_type = name |> to_string() |> then(&(&1 <> "_result")) |> String.to_atom()
 
+      action =
+        Ash.Resource.Info.action(resource, subscription.read_action) ||
+          Ash.Resource.Info.primary_action(resource, :read)
+
       %Absinthe.Blueprint.Schema.FieldDefinition{
-        arguments: args(:subscription, resource, nil, schema, nil),
+        arguments: args(:subscription, resource, action, schema, nil),
         identifier: name,
         name: to_string(name),
         config:
@@ -1760,26 +1764,29 @@ defmodule AshGraphql.Resource do
     read_args(resource, action, schema, hide_inputs)
   end
 
-  defp args(:subscription, resource, _action, schema, _identity, _hide_inputs, _query) do
-    if AshGraphql.Resource.Info.derive_filter?(resource) do
-      case resource_filter_fields(resource, schema) do
-        [] ->
-          []
+  defp args(:subscription, resource, action, schema, _identity, hide_inputs, _query) do
+    args =
+      if AshGraphql.Resource.Info.derive_filter?(resource) do
+        case resource_filter_fields(resource, schema) do
+          [] ->
+            []
 
-        _ ->
-          [
-            %Absinthe.Blueprint.Schema.InputValueDefinition{
-              name: "filter",
-              identifier: :filter,
-              type: resource_filter_type(resource),
-              description: "A filter to limit the results",
-              __reference__: ref(__ENV__)
-            }
-          ]
+          _ ->
+            [
+              %Absinthe.Blueprint.Schema.InputValueDefinition{
+                name: "filter",
+                identifier: :filter,
+                type: resource_filter_type(resource),
+                description: "A filter to limit the results",
+                __reference__: ref(__ENV__)
+              }
+            ]
+        end
+      else
+        []
       end
-    else
-      []
-    end
+
+    args ++ read_args(resource, action, schema, hide_inputs)
   end
 
   defp related_list_args(resource, related_resource, relationship_name, action, schema) do

lib/resource/subscription.ex

@@ -3,6 +3,7 @@ defmodule AshGraphql.Resource.Subscription do
   defstruct [
     :name,
     :actions,
+    :action_types,
     :read_action,
     :actor
   ]
@@ -20,7 +21,11 @@ defmodule AshGraphql.Resource.Subscription do
     ],
     actions: [
       type: {:or, [{:list, :atom}, :atom]},
-      doc: "The create/update/destroy actions the subsciption should listen to. Defaults to all."
+      doc: "The create/update/destroy actions the subsciption should listen to."
+    ],
+    action_types: [
+      type: {:or, [{:list, :atom}, :atom]},
+      doc: "The type of actions the subsciption should listen to."
     ],
     read_action: [
       type: :atom,

lib/subscription/notifier.ex

@@ -7,8 +7,8 @@ defmodule AshGraphql.Subscription.Notifier do
     pub_sub = Info.subscription_pubsub(notification.resource)
 
     for subscription <- AshGraphql.Resource.Info.subscriptions(notification.resource) do
-      if is_nil(subscription.actions) or
-           notification.action.name in List.wrap(subscription.actions) do
+      if notification.action.name in List.wrap(subscription.actions) or
+           notification.action.type in List.wrap(subscription.action_types) do
         Absinthe.Subscription.publish(pub_sub, notification, [{subscription.name, "*"}])
       end
     end

test/subscription_test.exs

@@ -257,4 +257,47 @@ defmodule AshGraphql.SubscriptionTest do
     assert subscribable.id ==
              subscription_data["dedupedSubscribableEvents"]["created"]["id"]
   end
+
+  test "can subscribe to read actions that take arguments" do
+    actor1 = %{
+      id: 1,
+      role: :user
+    }
+
+    subscription = """
+    subscription WithArguments($topic: String!) {
+      subscribableEventsWithArguments(topic: $topic) {
+        created {
+          id
+          text
+        }
+        updated {
+          id
+          text
+        }
+        destroyed
+      }
+    }
+    """
+
+    assert {:ok, %{"subscribed" => topic1}} =
+             Absinthe.run(
+               subscription,
+               Schema,
+               variables: %{"topic" => "news"},
+               context: %{actor: actor1, pubsub: PubSub}
+             )
+
+    subscribable =
+      Subscribable
+      |> Ash.Changeset.for_create(:create, %{text: "foo", topic: "news", actor_id: 1},
+        actor: @admin
+      )
+      |> Ash.create!()
+
+    assert_receive {^topic1, %{data: subscription_data}}
+
+    assert subscribable.id ==
+             subscription_data["subscribableEventsWithArguments"]["created"]["id"]
+  end
 end

test/support/resources/subscribable.ex

@@ -25,7 +25,7 @@ defmodule AshGraphql.Test.Subscribable do
       pubsub(AshGraphql.Test.PubSub)
 
       subscribe(:subscribable_events) do
-        actions([:create, :update, :destroy])
+        action_types([:create, :update, :destroy])
       end
 
       subscribe(:deduped_subscribable_events) do
@@ -36,6 +36,11 @@ defmodule AshGraphql.Test.Subscribable do
           %{id: -1, role: :deduped_actor}
         end)
       end
+
+      subscribe(:subscribable_events_with_arguments) do
+        read_action(:read_with_arg)
+        actions([:create])
+      end
     end
   end
 
@@ -48,7 +53,7 @@ defmodule AshGraphql.Test.Subscribable do
       authorize_if(expr(actor_id == ^actor(:id)))
     end
 
-    policy action(:open_read) do
+    policy action([:open_read, :read_with_arg]) do
       authorize_if(always())
     end
   end
@@ -58,12 +63,21 @@ defmodule AshGraphql.Test.Subscribable do
     defaults([:create, :read, :update, :destroy])
 
     read(:open_read)
+
+    read :read_with_arg do
+      argument(:topic, :string) do
+        allow_nil? false
+      end
+
+      filter(expr(topic == ^arg(:topic)))
+    end
   end
 
   attributes do
     uuid_primary_key(:id)
 
     attribute(:text, :string, public?: true)
+    attribute(:topic, :string, public?: true)
     attribute(:actor_id, :integer, public?: true)
     create_timestamp(:created_at)
     update_timestamp(:updated_at)