Setup ActionCable for GraphQL subscriptions

app/channels/application_cable/channel.rb

@@ -2,5 +2,45 @@
 
 module ApplicationCable
   class Channel < ActionCable::Channel::Base
+    def subscribe_to_channel
+      with_context { super }
+    end
+
+    def perform_action(data)
+      with_context { super }
+    end
+
+    protected
+
+    def find_authentication(authorization)
+      return Sagittarius::Authentication.new(:none, nil) if authorization.blank?
+
+      token_type, token = authorization.split(' ', 2)
+
+      create_authentication(token_type, token)
+    end
+
+    def create_authentication(token_type, token)
+      case token_type
+      when 'Session'
+        Sagittarius::Authentication.new(:session, UserSession.find_by(token: token, active: true))
+      else
+        Sagittarius::Authentication.new(:invalid, nil)
+      end
+    end
+
+    def with_context(&block)
+      Code0::ZeroTrack::Context.with_context(
+        application: 'cable',
+        ip_address: request_ip,
+        &block
+      )
+    end
+
+    def request_ip
+      return unless connection.respond_to?(:env)
+
+      ::Rack::Request.new(connection.env).ip
+    end
   end
 end

app/channels/graphql_channel.rb

@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+
+class GraphqlChannel < ApplicationCable::Channel
+  periodically :verify_authentication, every: 30.seconds
+
+  def subscribed
+    @token = params[:token]
+    @subscription_ids = []
+
+    verify_authentication
+  end
+
+  def execute(data)
+    result = SagittariusSchema.execute(
+      query: data['query'],
+      context: {
+        current_authentication: find_authentication(@token),
+        visibility_profile: :execution,
+        channel: self,
+      },
+      variables: data['variables'],
+      operation_name: data['operationName']
+    )
+
+    @subscription_ids << result.context[:subscription_id] if result.context[:subscription_id]
+
+    transmit({ result: result.to_h, more: result.subscription? })
+  end
+
+  def unsubscribed
+    @subscription_ids.each do |sid|
+      SagittariusSchema.subscriptions.delete_subscription(sid)
+    end
+  end
+
+  private
+
+  def verify_authentication
+    with_context do
+      authentication = find_authentication(@token)
+      return unless authentication.invalid? || authentication.none?
+
+      @subscription_ids.each { |sid| SagittariusSchema.subscriptions.delete_subscription(sid) }
+      reject
+    end
+  end
+end

app/graphql/sagittarius_schema.rb

@@ -4,15 +4,18 @@
 class SagittariusSchema < GraphQL::Schema
   mutation(Types::MutationType)
   query(Types::QueryType)
+  subscription(Types::SubscriptionType)
 
   default_max_page_size 50
   max_depth 20
   connections.add(ActiveRecord::Relation, Sagittarius::Graphql::StableConnection)
 
+  use GraphQL::Subscriptions::ActionCableSubscriptions
+
   # For batch-loading (see https://graphql-ruby.org/dataloader/overview.html)
   use GraphQL::Dataloader
 
-  use GraphQL::Schema::Visibility, profiles: {
+  use GraphQL::Schema::Visibility, preload: true, profiles: {
     execution: {},
     types: {},
     docs: {},

app/graphql/subscriptions/base_subscription.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+module Subscriptions
+  class BaseSubscription < GraphQL::Schema::Subscription
+    argument_class Types::BaseArgument
+    field_class Types::BaseField
+    object_class Types::BaseObject
+
+    def current_authentication
+      context[:current_authentication]
+    end
+
+    def self.generate_payload_type
+      result = super
+      result.graphql_name("#{graphql_name}SubscriptionPayload")
+      result
+    end
+  end
+end

app/graphql/subscriptions/echo.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+module Subscriptions
+  class Echo < BaseSubscription
+    description <<~DOC
+      A subscription that does not perform any real updates.
+
+      This is expected to be used for testing of endpoints, to verify
+      that a user has subscription access.
+    DOC
+
+    argument :message,
+             type: GraphQL::Types::String,
+             required: false,
+             description: 'Message to return to the user.'
+
+    field :message,
+          type: GraphQL::Types::String,
+          null: true,
+          description: 'Message returned to the user.'
+
+    def subscribe(message: nil)
+      { message: message }
+    end
+
+    def update(*)
+      { message: object[:message] }
+    end
+  end
+end

app/graphql/types/subscription_type.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module Types
+  class SubscriptionType < Types::BaseObject
+    description 'Root subscription type'
+
+    include Sagittarius::Graphql::MountSubscription
+
+    mount_subscription Subscriptions::Echo
+  end
+end

config.cable.ru

@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+
+require_relative 'config/environment'
+Rails.application.eager_load!
+
+run ActionCable.server

config/cable.yml

@@ -1,10 +1,8 @@
 development:
-  adapter: async
+  adapter: postgresql
 
 test:
   adapter: test
 
 production:
-  adapter: redis
-  url: <%= ENV.fetch("REDIS_URL") { "redis://localhost:6379/1" } %>
-  channel_prefix: sagittarius_production
+  adapter: postgresql

config/initializers/action_cable.rb

@@ -0,0 +1,5 @@
+# frozen_string_literal: true
+
+Rails.application.configure do
+  config.action_cable.disable_request_forgery_protection = true
+end

config/puma.rb

@@ -4,6 +4,8 @@
 # are invoked here are part of Puma's configuration DSL. For more information
 # about methods provided by the DSL, see https://puma.io/puma/Puma/DSL.html.
 
+require_relative 'environment'
+
 # Puma can serve each request in a thread from an internal thread pool.
 # The `threads` method setting takes two numbers: a minimum and maximum.
 # Any libraries that use thread pools should be configured to match

config/routes.rb

@@ -11,4 +11,5 @@
   get '/health/liveness' => 'rails/health#show', as: :rails_health_check
 
   mount GoodJob::Engine => 'good_job'
+  mount ActionCable.server => '/cable'
 end

docs/graphql/subscription/echo.md

@@ -0,0 +1,20 @@
+---
+title: echo
+---
+
+A subscription that does not perform any real updates.
+
+This is expected to be used for testing of endpoints, to verify
+that a user has subscription access.
+
+## Arguments
+
+| Name | Type | Description |
+|------|------|-------------|
+| `message` | [`String`](../scalar/string.md) | Message to return to the user. |
+
+## Fields
+
+| Name | Type | Description |
+|------|------|-------------|
+| `message` | [`String`](../scalar/string.md) | Message returned to the user. |

lib/sagittarius/graphql/mount_subscription.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module Sagittarius
+  module Graphql
+    module MountSubscription
+      extend ActiveSupport::Concern
+
+      class_methods do
+        def mount_subscription(subscription_class, **custom_kwargs)
+          # Using an underscored field name symbol will make `graphql-ruby`
+          # standardize the field name
+          field subscription_class.graphql_name.underscore.to_sym,
+                subscription: subscription_class,
+                **custom_kwargs
+        end
+
+        def mount_aliased_subscription(alias_name, subscription_class, **custom_kwargs)
+          aliased_subscription_class = Class.new(subscription_class) do
+            graphql_name alias_name
+          end
+
+          mount_subscription(aliased_subscription_class, **custom_kwargs)
+        end
+      end
+    end
+  end
+end

spec/channels/graphql_channel_spec.rb

@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe GraphqlChannel do
+  include AuthenticationHelpers
+
+  include_context 'with graphql subscription support'
+
+  let(:user) { create(:user) }
+  let(:token) { "Session #{authorization_token(user)}" }
+
+  describe '#subscribed' do
+    context 'with valid token' do
+      it 'confirms subscription' do
+        subscribe(token: token)
+        expect(subscription).to be_confirmed
+      end
+    end
+
+    context 'with invalid token' do
+      it 'rejects subscription' do
+        subscribe(token: 'Session invalid')
+        expect(subscription).to be_rejected
+      end
+    end
+
+    context 'without token' do
+      it 'rejects subscription' do
+        subscribe(token: nil)
+        expect(subscription).to be_rejected
+      end
+    end
+  end
+
+  describe '#execute' do
+    before { subscribe(token: token) }
+
+    it 'returns the initial subscription result' do
+      perform :execute,
+              query: 'subscription($message: String) { echo(message: $message) { message } }',
+              variables: { message: 'hello' }
+
+      expect(transmissions.last).to include(
+        'result' => { 'data' => { 'echo' => { 'message' => 'hello' } } },
+        'more' => true
+      )
+    end
+
+    it 'receives updates when triggered' do
+      variables = { message: 'hello' }
+      perform :execute,
+              query: 'subscription($message: String) { echo(message: $message) { message } }',
+              variables: variables
+
+      SagittariusSchema.subscriptions.trigger(:echo, variables, { message: 'updated' },
+                                              context: { visibility_profile: :execution })
+
+      expect(transmissions.last).to include(
+        'result' => { 'data' => { 'echo' => { 'message' => 'updated' } } },
+        'more' => true
+      )
+    end
+  end
+end

spec/support/shared_examples/graphql_subscription_support.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+RSpec.shared_context 'with graphql subscription support' do
+  let(:stream_callbacks) { {} }
+  let(:subscription_streams) { {} }
+
+  before do
+    callbacks = stream_callbacks
+    sub_streams = subscription_streams
+    test_spec = self
+
+    ActionCable::Channel::ChannelStub.define_method(:stream_from) do |broadcasting, coder: nil, &block|
+      streams << broadcasting
+      if block
+        callbacks[broadcasting] = { block: block, coder: coder }
+      else
+        sub_streams[broadcasting] = true
+      end
+    end
+
+    pubsub = ActionCable.server.pubsub
+    allow(pubsub).to receive(:broadcast).and_wrap_original do |method, stream, message|
+      method.call(stream, message)
+      if (cb = callbacks[stream])
+        decoded = cb[:coder] ? cb[:coder].decode(message) : message
+        cb[:block].call(decoded)
+      elsif sub_streams[stream]
+        decoded = ActiveSupport::JSON.decode(message)
+        test_spec.subscription.send(:transmit, decoded)
+      end
+    end
+  end
+
+  after do
+    ActionCable::Channel::ChannelStub.define_method(:stream_from) do |broadcasting, *|
+      streams << broadcasting
+    end
+  end
+end

tooling/graphql/docs/helper.rb

@@ -18,6 +18,7 @@ def rendering_objects
             enum: parser.elements[:enum],
             object: parser.elements[:object],
             mutation: parser.elements[:mutation],
+            subscription: parser.elements[:subscription],
             input_object: parser.elements[:input_object],
           }
         end