diff --git a/lib/graphql/resolver.ex b/lib/graphql/resolver.ex
index c4278041..c86a9db7 100644
--- a/lib/graphql/resolver.ex
+++ b/lib/graphql/resolver.ex
@@ -528,7 +528,9 @@ defmodule AshGraphql.Graphql.Resolver do
         Ash.Query.filter(query, ^ref(key) == ^value)
       end)
 
-    case query |> domain.read_one() do
+    dbg()
+
+    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} ->
         resolution
diff --git a/lib/resource/resource.ex b/lib/resource/resource.ex
index 7017f80b..ddedbd34 100644
--- a/lib/resource/resource.ex
+++ b/lib/resource/resource.ex
@@ -1159,7 +1159,7 @@ defmodule AshGraphql.Resource do
         identifier: name,
         name: to_string(name),
         config:
-          AshGraphql.Resource.Subscription.DefaultConfig.create_config(
+          AshGraphql.Subscription.Config.create_config(
             subscription,
             api,
             resource
diff --git a/lib/resource/subscription.ex b/lib/resource/subscription.ex
index 8404dca6..ac90a9d8 100644
--- a/lib/resource/subscription.ex
+++ b/lib/resource/subscription.ex
@@ -3,7 +3,8 @@ defmodule AshGraphql.Resource.Subscription do
   defstruct [
     :name,
     :actions,
-    :read_action
+    :read_action,
+    :actor
   ]
 
   @subscription_schema [
@@ -11,6 +12,12 @@ defmodule AshGraphql.Resource.Subscription do
       type: :atom,
       doc: "The name to use for the subscription."
     ],
+    actor: [
+      type:
+        {:spark_function_behaviour, AshGraphql.Resource.Subscription.Actor,
+         {AshGraphql.Resource.Subscription.Actor, 1}},
+      doc: "The actor to use for authorization."
+    ],
     actions: [
       type: {:or, [{:list, :atom}, :atom]},
       doc: "The create/update/destroy actions the subsciption should listen to. Defaults to all."
diff --git a/lib/resource/subscription/actor.ex b/lib/resource/subscription/actor.ex
new file mode 100644
index 00000000..c6f04770
--- /dev/null
+++ b/lib/resource/subscription/actor.ex
@@ -0,0 +1,3 @@
+defmodule AshGraphql.Resource.Subscription.Actor do
+  @callback author(actor :: any) :: actor :: any
+end
diff --git a/lib/resource/transformers/subscription.ex b/lib/resource/transformers/subscription.ex
index d02757cb..74a28140 100644
--- a/lib/resource/transformers/subscription.ex
+++ b/lib/resource/transformers/subscription.ex
@@ -18,7 +18,7 @@ defmodule AshGraphql.Resource.Transformers.Subscription do
          |> Transformer.persist(
            :simple_notifiers,
            [
-             AshGraphql.Resource.Subscription.Notifier
+             AshGraphql.Subscription.Notifier
            ] ++
              Transformer.get_persisted(dsl, :simple_notifiers, [])
          )}
diff --git a/lib/resource/subscription/default_config.ex b/lib/subscription/config.ex
similarity index 85%
rename from lib/resource/subscription/default_config.ex
rename to lib/subscription/config.ex
index 733d44df..0f0c025f 100644
--- a/lib/resource/subscription/default_config.ex
+++ b/lib/subscription/config.ex
@@ -1,4 +1,4 @@
-defmodule AshGraphql.Resource.Subscription.DefaultConfig do
+defmodule AshGraphql.Subscription.Config do
   alias AshGraphql.Resource.Subscription
 
   def create_config(%Subscription{} = subscription, _domain, resource) do
@@ -26,8 +26,10 @@ defmodule AshGraphql.Resource.Subscription.DefaultConfig do
             # context_id is exposed to the client so we might need to encrypt it
             # or save it in ets or something and send generate a hash or something
             # as the context_id
-            dbg(filter)
-            {:ok, topic: "*", context_id: dbg(Base.encode64(:erlang.term_to_binary(filter)))}
+            dbg(filter, structs: false)
+
+            {:ok,
+             topic: "*", context_id: dbg(Base.encode64(:erlang.term_to_binary(filter.filter)))}
 
           _ ->
             {:error, "unauthorized"}
diff --git a/lib/endpoint.ex b/lib/subscription/endpoint.ex
similarity index 94%
rename from lib/endpoint.ex
rename to lib/subscription/endpoint.ex
index 396ba984..834565a6 100644
--- a/lib/endpoint.ex
+++ b/lib/subscription/endpoint.ex
@@ -1,4 +1,4 @@
-defmodule AshGraphql.Endpoint do
+defmodule AshGraphql.Subscription.Endpoint do
   defmacro __using__(_opts) do
     quote do
       use Absinthe.Phoenix.Endpoint
@@ -8,10 +8,12 @@ defmodule AshGraphql.Endpoint do
       require Logger
 
       def run_docset(pubsub, docs_and_topics, mutation_result) do
+        dbg(mutation_result, structs: false)
+
         for {topic, key_strategy, doc} <- docs_and_topics do
           try do
             pipeline =
-              Absinthe.Subscription.Local.pipeline(doc, mutation_result)
+              Absinthe.Subscription.Local.pipeline(doc, mutation_result.data)
               # why though?
               |> List.flatten()
               |> Absinthe.Pipeline.insert_before(
diff --git a/lib/resource/subscription/notifier.ex b/lib/subscription/notifier.ex
similarity index 69%
rename from lib/resource/subscription/notifier.ex
rename to lib/subscription/notifier.ex
index 5e73a7e4..ed039988 100644
--- a/lib/resource/subscription/notifier.ex
+++ b/lib/subscription/notifier.ex
@@ -1,4 +1,4 @@
-defmodule AshGraphql.Resource.Subscription.Notifier do
+defmodule AshGraphql.Subscription.Notifier do
   alias AshGraphql.Resource.Info
   use Ash.Notifier
 
@@ -6,10 +6,12 @@ defmodule AshGraphql.Resource.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
-        Absinthe.Subscription.publish(pub_sub, notification.data, [{subscription.name, "*"}])
+        Absinthe.Subscription.publish(pub_sub, notification, [{subscription.name, "*"}])
       end
     end
   end