Impliment queueing for /search and /graphql queries

Gemfile

@@ -50,8 +50,11 @@ gem 'grape', '1.3.0'
 # gem 'puma_worker_killer'
 
 # Bundle edge Rails instead: gem 'rails', github: 'rails/rails'
+gem 'backburner'
 gem 'rack', '>= 2.0.6'
 gem 'rails', '~> 5.2.0'
+
+gem 'beanstalkd_view', group: :development
 # Use sqlite3 as the database for Active Record
 gem 'sqlite3'
 # Use SCSS for stylesheets

Gemfile.lock

@@ -103,8 +103,19 @@ GEM
       aws-sigv4 (~> 1.1)
     aws-sigv4 (1.1.0)
       aws-eventstream (~> 1.0, >= 1.0.2)
+    backburner (1.5.0)
+      beaneater (~> 1.0)
+      concurrent-ruby (~> 1.0, >= 1.0.1)
+      dante (> 0.1.5)
     backports (3.15.0)
     bcrypt (3.1.13)
+    beaneater (1.0.0)
+    beanstalkd_view (2.0.0)
+      beaneater (~> 1.0.0)
+      json
+      sinatra (>= 1.3.0)
+      sinatra-contrib (>= 1.3.0)
+      vegas (~> 0.1.2)
     bindex (0.8.1)
     builder (3.2.3)
     byebug (11.0.1)
@@ -148,6 +159,7 @@ GEM
     crack (0.4.3)
       safe_yaml (~> 1.0.0)
     crass (1.0.5)
+    dante (0.2.0)
     devise (4.7.2)
       bcrypt (~> 3.0)
       orm_adapter (~> 0.1)
@@ -291,6 +303,8 @@ GEM
     rack-cors (0.4.1)
     rack-mini-profiler (1.1.0)
       rack (>= 1.2.0)
+    rack-protection (2.0.8.1)
+      rack
     rack-test (1.1.0)
       rack (>= 1.0, < 3)
     rails (5.2.3)
@@ -368,6 +382,18 @@ GEM
       json (>= 1.8, < 3)
       simplecov-html (~> 0.10.0)
     simplecov-html (0.10.2)
+    sinatra (2.0.8.1)
+      mustermann (~> 1.0)
+      rack (~> 2.0)
+      rack-protection (= 2.0.8.1)
+      tilt (~> 2.0)
+    sinatra-contrib (2.0.8.1)
+      backports (>= 2.8.2)
+      multi_json
+      mustermann (~> 1.0)
+      rack-protection (= 2.0.8.1)
+      sinatra (= 2.0.8.1)
+      tilt (~> 2.0)
     spring (2.1.0)
     sprockets (4.0.0)
       concurrent-ruby (~> 1.0)
@@ -405,6 +431,8 @@ GEM
     uglifier (4.2.0)
       execjs (>= 0.3.0, < 3)
     unicode-display_width (1.6.0)
+    vegas (0.1.11)
+      rack (>= 1.0.0)
     warden (1.2.8)
       rack (>= 2.0.6)
     web-console (3.7.0)
@@ -436,6 +464,8 @@ DEPENDENCIES
   audited (~> 4.4)
   awesome_print
   aws-sdk-s3
+  backburner
+  beanstalkd_view
   blazer (= 2.2.1.charcoal)!
   byebug
   capistrano

Procfile

@@ -1,3 +1,5 @@
-web: bundle exec rails s
+web: bundle exec bin/rails s
 webpacker: ./bin/webpack-dev-server
 redis: ./redis-git/src/redis-server --port 6378 --loadmodule ./zhregex/module.so --dbfilename dump.rdb
+beanstalkd: beanstalkd -l 127.0.0.1 -V
+backburner: bundle exec rake backburner:work QUEUE=default,graphql_queries

Rakefile

@@ -4,5 +4,7 @@
 # for example lib/tasks/capistrano.rake, and they will automatically be available to Rake.
 
 require File.expand_path('../config/application', __FILE__)
+require 'backburner/tasks'
+require_relative 'config/initializers/backburner.rb'
 
 Rails.application.load_tasks

app/channels/topbar_channel.rb

@@ -4,9 +4,7 @@
 class TopbarChannel < ApplicationCable::Channel
   def subscribed
     stream_from 'topbar'
-    Thread.new do
-      ActionCable.server.broadcast 'topbar', commit: CurrentCommit
-    end
+    ActionCable.server.broadcast 'topbar', commit: CurrentCommit
   end
 
   def unsubscribed

app/controllers/authentication_controller.rb

@@ -27,11 +27,8 @@ def redirect_target
     ActiveRecord::Base.logger = old_logger
 
     if current_user.write_authenticated
-      u = current_user
-      Thread.new do
-        # Do this in the background to keep the page load fast.
-        u.update_moderator_sites
-      end
+      # Do this in the background to keep the page load fast.
+      UpdateModeratorSitesJob.perform_later(current_user.id)
     end
 
     flash[:success] = 'Successfully registered token'
@@ -74,11 +71,8 @@ def login_redirect_target
 
       user.save!
 
-      Thread.new do
-        # Do this in the background to keep the page load fast.
-        user.update_chat_ids
-        user.save!
-      end
+      # Do this in the background to keep the page load fast.
+      UpdateChatIdsJob.perform_later(user.id)
 
       flash[:success] = "New account created for #{user.username}. Have fun!"
     end

app/controllers/flag_conditions_controller.rb

@@ -131,9 +131,7 @@ def user_overview
 
   def validate_user
     @user = User.find params[:user]
-    Thread.new do
-      FlagCondition.validate_for_user @user, current_user
-    end
+    ValidateFlagConditionForUserJob.perform_later(@user.id, current_user.id)
     flash[:info] = 'Validation launched in background.'
     redirect_back fallback_location: root_path
   end

app/controllers/flag_settings_controller.rb

@@ -55,9 +55,7 @@ def update
           # we want to re-validate all existing FlagConditions
           # and disable them if they aren't in compliance with the
           # new settings
-          Thread.new do
-            FlagCondition.revalidate_all
-          end
+          RevalidateFlagConditionsJob.perform_later
         end
 
         format.html { redirect_to flag_settings_path, notice: 'Flag setting was successfully updated.' }

app/controllers/graphql_controller.rb

@@ -2,34 +2,77 @@
 
 class GraphqlController < ApplicationController
   skip_before_action :verify_authenticity_token, only: [:execute]
-  before_action :verify_authorization
 
   def execute
     variables = ensure_hash(params[:variables])
     query = params[:query]
     operation_name = params[:operationName]
     context = {
       # Query context goes here, for example:
-      current_user: current_user
+      # current_user: current_user
     }
     query_params = { variables: variables, context: context, operation_name: operation_name }
     if user_signed_in? && current_user.has_role?(:core)
       # query_params.merge!({max_depth: 8, max_complexity:20})
     end
-    @results = MetasmokeSchema.execute(query, **query_params)
-    respond_to do |format|
-      format.json { render json: @results }
-      format.html { @results = JSON.pretty_generate(@results.to_hash) }
+
+    api_key = APIKey.find_by(key: params[:key])
+    if (user_signed_in? && current_user.has_role?(:core)) || (!api_key.nil? && api_key.api_tokens.exists?(token: params[:token]))
+      @results = MetasmokeSchema.execute(query, **query_params)
+      respond_to do |format|
+        format.json { render json: @results }
+        format.html { @results = JSON.pretty_generate(@results.to_hash) }
+      end
+    elsif user_signed_in? || !api_key.nil?
+      @job_id = QueueGraphqlQueryJob.perform_later(query, **query_params).job_id
+      respond_to do |format|
+        format.json { render json: { job_id: @job_id } }
+        format.html { redirect_to view_graphql_job_path(job_id: @job_id) }
+      end
+    else
+      return head :forbidden
+    end
+  end
+
+  def view_job
+    k = "graphql_queries/#{params[:job_id]}"
+    if redis.sismember 'pending_graphql_queries', params[:job_id]
+      respond_to do |format|
+        format.json { render json: { complete: false, pending: true } }
+        format.html { render action: :pending_job }
+      end
+    elsif redis.exists k
+      v = redis.get k
+      time_remaining = redis.ttl(k).to_i
+      @time_elapsed = 600 - time_remaining
+      if v == ''
+        respond_to do |format|
+          format.json { render json: { complete: false, time_elapsed: @time_elapsed } }
+          format.html { render action: :pending_job }
+        end
+      else
+        @results = JSON.parse(v)
+        respond_to do |format|
+          format.json { render json: @results }
+          format.html do
+            @results = JSON.pretty_generate(@results.to_hash)
+            render action: :execute
+          end
+        end
+      end
+    else
+      return head 404
     end
   end
 
   def query; end
 
   private
 
-  def verify_authorization
-    return head 403 unless !APIKey.find_by(key: params[:key]).nil? || (user_signed_in? && current_user.has_role?(:core))
-  end
+  # def valid_api_token
+  #   key = APIKey.find_by(key: params[:key])
+  #   !key.nil? && key.api_tokens.exists?(token: params[:token])
+  # end
 
   # Handle form data, JSON body, or a blank value
   def ensure_hash(ambiguous_param)

app/controllers/review_queues_controller.rb

@@ -83,11 +83,7 @@ def reviews
   end
 
   def recheck_items
-    Thread.new do
-      @queue.items.includes(:reviewable).each do |i|
-        i.update(completed: true) if i.reviewable.should_dq?(@queue)
-      end
-    end
+    ReviewQueueRecheckItemsJob.perform_later(@queue.id)
     flash[:info] = 'Checking started in background.'
     redirect_back fallback_location: review_queues_path
   end

app/controllers/search_controller.rb

@@ -1,6 +1,37 @@
 # frozen_string_literal: true
 
 class SearchController < ApplicationController
+  def new_search; end
+
+  def create_search
+    ops = %i[title body why username].map do |s|
+      SearchHelper.parse_search_params(params, s, current_user)
+    end.flatten
+    job_id = SearchJob.perform_later(ops, params.permit(VALID_SEARCH_PARAMS)).job_id
+    redirect_to search_pending_path(job_id: job_id)
+  end
+
+  def search_pending
+    return head :not_found unless redis.exists "search_jobs/#{params[:job_id]}/sid"
+    @sid = redis.get "search_jobs/#{params[:job_id]}/sid"
+    @sid_ttl = redis.ttl("search_jobs/#{params[:job_id]}/sid").to_i
+    @be_page = redis.get("search_jobs/#{params[:job_id]}/be_page").to_i
+    return unless redis.exists "searches/#{@sid}/results/#{@be_page}"
+    per_page = 100
+    redirect_to search_results_path(sid: @sid, page: (@be_page * SearchJob.search_page_length) / per_page, per_page: per_page)
+  end
+
+  def search_results
+    @search_params = JSON.parse(redis.get("searches/#{params[:sid]}/params")).symbolize_keys
+    @result_count = redis.get("searches/#{params[:sid].to_i}/result_count").to_i
+    @results = search_get_page(params[:page].to_i, params[:per_page].to_i, sid: params[:sid].to_i)
+    redirect_to search_pending_path(job_id: @results) if @results.is_a? String
+    total_pages = (@result_count / params[:per_page].to_i)
+    current_page = params[:page].to_i
+    @results.define_singleton_method(:total_pages) { total_pages }
+    @results.define_singleton_method(:current_page) { current_page }
+  end
+
   def index
     # This might be ugly, but it's better than the alternative.
     #
@@ -319,4 +350,29 @@ def index_fast
     end
     redis.del final_key
   end
+
+  private
+
+  def search_get_page(page_num, per_page = 100, **hsh)
+    total_offset = per_page * page_num
+    per_page_real = [per_page, SearchJob.search_page_length].min
+    be_page_offset = total_offset % SearchJob.search_page_length
+    be_page = (total_offset / SearchJob.search_page_length).floor
+    if redis.exists "searches/#{hsh[:sid]}/results/#{be_page}"
+      @counts_by_accuracy_group = JSON.parse(redis.get("searches/#{hsh[:sid]}/counts_by_accuracy_group")).symbolize_keys
+      @counts_by_feedback = JSON.parse(redis.get("searches/#{hsh[:sid]}/counts_by_feedback")).symbolize_keys
+      return Post.where(id: redis.get("searches/#{hsh[:sid]}/results/#{be_page}").unpack('I!*'))
+                 .order(created_at: :desc)
+                 .offset(be_page_offset)
+                 .limit(per_page_real)
+                 .includes_for_post_row
+    elsif hsh.key?(:sid)
+      unless redis.set("searches/#{hsh[:sid]}/results/#{be_page}/job_id", '', nx: true)
+        return redis.get("searches/#{hsh[:sid]}/results/#{be_page}/job_id")
+      end
+      job_id = SearchExtendJob.perform_later(hsh[:sid], be_page).job_id
+      redis.set "searches/#{hsh[:sid]}/results/#{be_page}/job_id", job_id
+      return job_id
+    end
+  end
 end

app/controllers/status_controller.rb

@@ -25,12 +25,10 @@ def status_update
 
     @smoke_detector.save!
 
-    Thread.new do
-      ActionCable.server.broadcast 'status', status_channel_data
-      ActionCable.server.broadcast 'status_blacklist_manager', status_channel_data.merge(failover_link: failover_link)
-      ActionCable.server.broadcast 'topbar', last_ping: @smoke_detector.last_ping.to_f
-      ActionCable.server.broadcast 'smokey_pings', smokey: @smoke_detector.as_json
-    end
+    ActionCable.server.broadcast 'status', status_channel_data
+    ActionCable.server.broadcast 'status_blacklist_manager', status_channel_data.merge(failover_link: failover_link)
+    ActionCable.server.broadcast 'topbar', last_ping: @smoke_detector.last_ping.to_f
+    ActionCable.server.broadcast 'smokey_pings', smokey: @smoke_detector.as_json
 
     respond_to do |format|
       format.json do

app/jobs/queue_graphql_query_job.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class QueueGraphqlQueryJob < ApplicationJob
+  queue_as :graphql_queries
+
+  def perform(query, **query_params)
+    puts 'PERF'
+    @results = MetasmokeSchema.execute(query, **query_params)
+  end
+
+  before_enqueue do
+    redis.sadd 'pending_graphql_queries', job_id
+  end
+
+  before_perform do
+    redis.set "graphql_queries/#{job_id}", '', ex: 600
+    redis.srem 'pending_graphql_queries', job_id
+  end
+
+  after_perform do
+    redis.set "graphql_queries/#{job_id}", JSON.generate(@results), ex: 300
+  end
+end

app/jobs/revalidate_flag_conditions_job.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+class RevalidateFlagConditionsJob < ApplicationJob
+  queue_as :default
+
+  def perform
+    FlagCondition.where(flags_enabled: true).find_each do |fc|
+      unless fc.validate
+        failures = fc.errors.full_messages
+        fc.flags_enabled = false
+        fc.save(validate: false)
+        ActionCable.server.broadcast 'smokedetector_messages',
+                                     message: "@#{fc.user&.username&.tr(' ', '')} " \
+                                                "Your flag condition was disabled: #{failures.join(',')}"
+      end
+    end
+  end
+end