add endpoint with custom run_docset function

lib/endpoint.ex

@@ -0,0 +1,68 @@
+defmodule AshGraphql.Endpoint do
+  defmacro __using__(_opts) do
+    quote do
+      use Absinthe.Phoenix.Endpoint
+
+      require Logger
+
+      def run_docset(pubsub, docs_and_topics, mutation_result) do
+        for {topic, key_strategy, doc} <- docs_and_topics do
+          try do
+            pipeline =
+              Absinthe.Subscription.Local.pipeline(doc, mutation_result)
+              # why though?
+              |> List.flatten()
+              |> Absinthe.Pipeline.insert_before(
+                Absinthe.Phase.Document.OverrideRoot,
+                {Absinthe.Phase.Document.Context, context: %{ash_filter: get_filter(topic)}}
+              )
+
+            {:ok, %{result: data}, _} = Absinthe.Pipeline.run(doc.source, pipeline)
+
+            Logger.debug("""
+            Absinthe Subscription Publication
+            Field Topic: #{inspect(key_strategy)}
+            Subscription id: #{inspect(topic)}
+            Data: #{inspect(data)}
+            """)
+
+            case is_forbidden(data) do
+              true ->
+                # do not send anything to the client if he is not allowed to see it
+                :ok
+
+              false ->
+                :ok = pubsub.publish_subscription(topic, data)
+            end
+          rescue
+            e ->
+              BatchResolver.pipeline_error(e, __STACKTRACE__)
+          end
+        end
+      end
+
+      defp is_forbidden(%{errors: errors}) do
+        errors
+        |> List.wrap()
+        |> Enum.any?(fn error -> Map.get(error, :code) == "forbidden" end)
+      end
+
+      defp is_forbidden(_), do: false
+
+      defp get_filter(topic) do
+        [_, rest] = String.split(topic, "__absinthe__:doc:")
+        [filter, _] = String.split(rest, ":")
+
+        case Base.decode64(filter) do
+          {:ok, filter} ->
+            :erlang.binary_to_term(filter)
+
+          _ ->
+            nil
+        end
+      rescue
+        _ -> nil
+      end
+    end
+  end
+end

lib/graphql/resolver.ex

@@ -507,14 +507,14 @@ defmodule AshGraphql.Graphql.Resolver do
   end
 
   def resolve(
-        %{arguments: arguments, context: context, root_value: data} = resolution,
-        {api, resource, %AshGraphql.Resource.Subscription{}, _input?}
+        %{arguments: _arguments, context: context, root_value: data} = resolution,
+        {domain, resource, %AshGraphql.Resource.Subscription{}, _input?}
       ) do
     query =
       AshGraphql.Subscription.query_for_subscription(
         resource
         |> Ash.Query.new(),
-        api,
+        domain,
         resolution
       )
 
@@ -525,12 +525,15 @@ defmodule AshGraphql.Graphql.Resolver do
         Ash.Query.filter(query, ^ref(key) == ^value)
       end)
 
-    result =
-      query
-      |> api.read_one(actor: resolution.context[:current_user])
+    case query |> domain.read_one(actor: resolution.context[:current_user]) do
+      {:ok, result} ->
+        resolution
+        |> Absinthe.Resolution.put_result({:ok, result})
 
-    resolution
-    |> Absinthe.Resolution.put_result(result)
+      {:error, error} ->
+        resolution
+        |> Absinthe.Resolution.put_result({:error, to_errors([error], context, domain)})
+    end
   end
 
   defp read_one_query(resource, args) do

lib/resource/subscription/default_config.ex

@@ -3,7 +3,6 @@ defmodule AshGraphql.Resource.Subscription.DefaultConfig do
 
   def create_config(%Subscription{} = subscription, _domain, resource) do
     config_module = String.to_atom(Macro.camelize(Atom.to_string(subscription.name)) <> ".Config")
-    dbg()
 
     defmodule config_module do
       require Ash.Query
@@ -15,16 +14,18 @@ defmodule AshGraphql.Resource.Subscription.DefaultConfig do
           @subscription.read_action || Ash.Resource.Info.primary_action!(@resource, :read).name
 
         case Ash.can(
-               Ash.Query.for_read(@resource, read_action)
-               |> Ash.Query.filter(id == "test"),
+               Ash.Query.for_read(@resource, read_action),
                context[:actor],
                run_queries?: false,
                alter_source?: true
              ) do
           {:ok, true} ->
-            dbg({:ok, topic: "*", context_id: "global"})
+            {:ok, topic: "*", context_id: "global"}
 
           {:ok, true, filter} ->
+            # 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
             {:ok, topic: "*", context_id: Base.encode64(:erlang.term_to_binary(filter))}
 
           _ ->

lib/resource/subscription/notifier.ex

@@ -6,7 +6,7 @@ defmodule AshGraphql.Resource.Subscription.Notifier do
   def notify(notification) do
     pub_sub = Info.subscription_pubsub(notification.resource)
 
-    for subscription <- dbg(AshGraphql.Resource.Info.subscriptions(notification.resource)) do
+    for subscription <- AshGraphql.Resource.Info.subscriptions(notification.resource) do
       Absinthe.Subscription.publish(pub_sub, notification.data, [{subscription.name, "*"}])
     end
   end