a step in the right direction?

lib/resource/notifier.ex

@@ -0,0 +1,12 @@
+defmodule AshGraphql.Resource.Notifier do
+  use Ash.Notifier
+
+  @impl Ash.Notifier
+  def notify(notification) do
+    IO.inspect(notification, label: :Notifier)
+
+    Absinthe.Subscription.publish(AshGraphql.Test.PubSub, notification.data,
+      subscrible_created: "*"
+    )
+  end
+end

test/subscription_test.exs

@@ -4,8 +4,8 @@ defmodule AshGraphql.SubscriptionTest do
   alias AshGraphql.Test.PubSub
   alias AshGraphql.Test.Schema
 
-  setup_all do
-    Application.put_env(PubSub, :notifier_test_pid, self())
+  setup do
+    Application.put_env(PubSub, :notifier_test_pid, self() |> IO.inspect(label: :test_process))
     {:ok, _} = PubSub.start_link()
     {:ok, _} = Absinthe.Subscription.start_link(PubSub)
     :ok
@@ -28,8 +28,6 @@ defmodule AshGraphql.SubscriptionTest do
                context: %{pubsub: PubSub, actor: %{id: id}}
              )
 
-    PubSub.subscribe("subscribable:created")
-
     mutation = """
     mutation CreateSubscribable($input: CreateSubscribableInput) {
         createSubscribable(input: $input) {
@@ -43,28 +41,22 @@ defmodule AshGraphql.SubscriptionTest do
       }
     """
 
+    IO.inspect(self())
+
     assert {:ok, %{data: data}} =
              run_subscription(mutation, Schema,
                variables: %{"input" => %{"text" => "foo"}},
                context: %{pubsub: PubSub}
              )
 
-    assert_receive({:broadcast, msg})
-
-    Absinthe.Subscription.publish(PubSub, data, subscribable_created: "subscribable:created")
-    |> IO.inspect(label: :publish)
-
-    assert %{
-             event: "subscription:data",
-             result: %{data: %{"user" => %{"id" => "1", "name" => "foo"}}},
-             topic: topic
-           } == msg
+    assert_receive({:broadcast, absinthe_proxy, data, fields})
   end
 
   defp run_subscription(query, schema, opts) do
     opts = Keyword.update(opts, :context, %{pubsub: PubSub}, &Map.put(&1, :pubsub, PubSub))
 
-    case Absinthe.run(query, schema, opts) |> IO.inspect(label: :absinthe_run) do
+    case Absinthe.run(query, schema, opts) do
+      #  |> IO.inspect(label: :absinthe_run) do
       {:ok, %{"subscribed" => topic}} = val ->
         PubSub.subscribe(topic)
         val

test/support/pub_sub.ex

@@ -10,7 +10,7 @@ defmodule AshGraphql.Test.PubSub do
   end
 
   def subscribe(topic) do
-    IO.inspect([topic: topic], label: "subscribe")
+    # IO.inspect([topic: topic], label: "subscribe")
     Registry.register(__MODULE__, topic, [self()])
     :ok
   end
@@ -22,7 +22,8 @@ defmodule AshGraphql.Test.PubSub do
         event: "subscription:data",
         result: data
       }
-      |> IO.inspect(label: :publish_subscription)
+
+    # |> IO.inspect(label: :publish_subscription)
 
     Registry.dispatch(__MODULE__, topic, fn entries ->
       for {pid, _} <- entries, do: send(pid, {:broadcast, message})
@@ -30,7 +31,7 @@ defmodule AshGraphql.Test.PubSub do
   end
 
   def broadcast(topic, event, notification) do
-    IO.inspect([topic: topic, event: event, notification: notification], label: "broadcast")
+    # IO.inspect([topic: topic, event: event, notification: notification], label: "broadcast")
 
     message =
       %{
@@ -44,9 +45,16 @@ defmodule AshGraphql.Test.PubSub do
     end)
   end
 
-  def publish_mutation(_proxy_topic, _mutation_result, _subscribed_fields) do
+  def publish_mutation(proxy_topic, mutation_result, subscribed_fields) do
     # this pubsub is local and doesn't support clusters
     IO.inspect("publish mutation")
+
+    send(
+      Application.get_env(__MODULE__, :notifier_test_pid) |> IO.inspect(label: :send_to),
+      {:broadcast, proxy_topic, mutation_result, subscribed_fields}
+    )
+    |> IO.inspect(label: :send)
+
     :ok
   end
 end

test/support/resources/post.ex

@@ -140,6 +140,10 @@ defmodule AshGraphql.Test.Post do
 
   require Ash.Query
 
+  resource do
+    simple_notifiers [AshGraphql.Resource.Notifier]
+  end
+
   policies do
     policy always() do
       authorize_if(always())

test/support/resources/subscribable.ex

@@ -4,11 +4,14 @@ defmodule AshGraphql.Test.Subscribable do
 
   use Ash.Resource,
     data_layer: Ash.DataLayer.Ets,
-    notifiers: [Ash.Notifier.PubSub],
     extensions: [AshGraphql.Resource]
 
   require Ash.Query
 
+  resource do
+    simple_notifiers([AshGraphql.Resource.Notifier])
+  end
+
   graphql do
     type :subscribable
 
@@ -21,14 +24,6 @@ defmodule AshGraphql.Test.Subscribable do
     end
   end
 
-  pub_sub do
-    module(PubSub)
-    prefix("subscribable")
-    broadcast_type(:notification)
-
-    publish_all(:create, "created")
-  end
-
   actions do
     defaults([:create, :read, :update, :destroy])
   end

test/support/schema.ex

@@ -35,11 +35,8 @@ defmodule AshGraphql.Test.Schema do
   subscription do
     field :subscribable_created, :subscribable do
       config(fn
-        _args, %{context: %{actor: %{id: user_id}}} ->
-          {:ok, topic: "subscribable:created", context_id: "user/#{user_id}"}
-
-        _args, _context ->
-          {:error, :unauthorized}
+        _args, _info ->
+          {:ok, topic: "*"}
       end)
 
       resolve(fn args, _, resolution ->