Action Cable Subscriptions

development:
  adapter: redis
  url: redis://localhost:6379/1
  channel_prefix: myapp_development

production:
  adapter: redis
  url: <%= ENV.fetch("REDIS_URL") { "redis://localhost:6379/1" } %>
  channel_prefix: myapp_production

Rails.application.routes.draw do
  mount ActionCable.server => '/cable'
end

module GraphQL
  class AppSchema < Rails::GraphQL::Schema
    # Configure subscription provider
    use_subscription_provider :action_cable,
      cable: ::ActionCable,
      prefix: 'rails-graphql',
      store: Rails::GraphQL::Subscription::Store::Memory.new,
      logger: Rails.logger
  end
end

class Message < ApplicationRecord
  after_create_commit :broadcast_creation
  
  private
  
  def broadcast_creation
    field = GraphQL::AppSchema[:subscription][:message_added]
    field.trigger(scope: channel)
  end
end

class Message < ApplicationRecord
  belongs_to :channel
  
  after_create_commit :broadcast_creation
  
  private
  
  def broadcast_creation
    field = GraphQL::AppSchema[:subscription][:message_added]
    
    # Only notify subscriptions for this channel
    field.trigger(scope: channel)
  end
end

module GraphQL
  module Subscription
    module Provider
      class Webhook < Rails::GraphQL::Subscription::Provider::Base
        self.abstract = false
        
        attr_reader :http_client
        
        def initialize(**options)
          @http_client = options.fetch(:http_client, HTTP)
          super
        end
        
        def accepts?(operation)
          operation.request.context[:webhook_url].present?
        end
        
        def add(*subscriptions)
          subscriptions.each do |item|
            log(:added, item)
            store.add(item)
          end
        end
        
        def async_remove(item)
          return if (item = store.fetch(item)).nil?
          store.remove(item)
          log(:removed, item)
        end
        
        def async_update(item, data = nil, **xargs)
          return if (item = store.fetch(item)).nil?
          
          log(:updated, item) do
            data = execute(item, **xargs) if data.nil?
            store.update!(item)
            
            unless unsubscribing?(data)
              webhook_url = item.context[:webhook_url]
              http_client.post(webhook_url, json: { result: data })
            else
              async_remove(item)
            end
          end
        end
      end
    end
  end
end

module GraphQL
  class AppSchema < Rails::GraphQL::Schema
    use_subscription_provider :webhook,
      store: Rails::GraphQL::Subscription::Store::Memory.new,
      http_client: HTTP.timeout(5)
  end
end

module GraphQL
  module Subscription
    module Store
      class Redis < Rails::GraphQL::Subscription::Store::Base
        self.abstract = false
        
        attr_reader :redis
        
        def initialize(redis_client = ::Redis.new)
          @redis = redis_client
        end
        
        def all
          redis.smembers('subscriptions:sids')
        end
        
        def add(subscription)
          sid = subscription.sid
          
          # Store subscription data
          redis.hset("subscription:#{sid}", subscription.to_h)
          redis.sadd('subscriptions:sids', sid)
          
          # Index for searching
          field_key = "index:field:#{subscription.field.hash}"
          redis.sadd(field_key, sid)
          
          sid
        end
        
        def fetch(*sids)
          return nil if sids.empty?
          
          items = sids.map do |sid|
            data = redis.hgetall("subscription:#{sid}")
            data.empty? ? nil : deserialize_subscription(data)
          end
          
          items.one? ? items.first : items
        end
        
        def remove(item)
          sid = instance?(item) ? item.sid : item
          
          # Remove from index
          data = redis.hgetall("subscription:#{sid}")
          if data.present?
            field_key = "index:field:#{data['field_hash']}"
            redis.srem(field_key, sid)
          end
          
          # Remove subscription data
          redis.del("subscription:#{sid}")
          redis.srem('subscriptions:sids', sid)
        end
        
        def has?(item)
          sid = instance?(item) ? item.sid : item
          redis.sismember('subscriptions:sids', sid)
        end
        
        def search(**options, &block)
          # Implement search logic based on indexed fields
          # Return matching SIDs
        end
      end
    end
  end
end

ActiveSupport::Notifications.subscribe('subscription.graphql') do |name, start, finish, id, payload|
  duration = finish - start
  
  Rails.logger.info({
    event: 'subscription',
    type: payload[:type],
    sid: payload[:item]&.sid,
    duration: duration
  }.to_json)
end

require 'rails_helper'

RSpec.describe 'MessageAdded subscription' do
  let(:channel) { create(:channel) }
  let(:subscription_query) do
    <<~GQL
      subscription MessageAdded($channelId: ID!) {
        messageAdded(channelId: $channelId) {
          id
          body
        }
      }
    GQL
  end
  
  it 'triggers when message is created' do
    # Subscribe
    result = GraphQL::AppSchema.execute(
      subscription_query,
      variables: { channelId: channel.id },
      context: { channel: mock_channel }
    )
    
    expect(result.dig('data', 'messageAdded')).to be_nil
    
    # Trigger update
    message = create(:message, channel: channel)
    
    # Verify broadcast was called
    expect(ActionCable.server).to have_received(:broadcast)
      .with(matching(/rails-graphql:/), hash_including('result'))
  end
end