[deploy] Create a Bulk Index Elasticsearch Worker (#7265)

app/models/article.rb

@@ -172,6 +172,8 @@ class Article < ApplicationRecord
 
   scope :with_video, -> { published.where.not(video: [nil, ""], video_thumbnail_url: [nil, ""]).where("score > ?", -4) }
 
+  scope :eager_load_serialized_data, -> { includes(:user, :organization, :tags) }
+
   algoliasearch per_environment: true, auto_remove: false, enqueue: :trigger_index do
     attribute :title
     add_index "searchables", id: :index_id, per_environment: true, enqueue: :trigger_index do

app/models/chat_channel_membership.rb

@@ -19,6 +19,8 @@ class ChatChannelMembership < ApplicationRecord
 
   delegate :channel_type, to: :chat_channel
 
+  scope :eager_load_serialized_data, -> { includes(:user, :channel) }
+
   def channel_last_message_at
     chat_channel.last_message_at
   end

app/models/comment.rb

@@ -48,6 +48,8 @@ class Comment < ApplicationRecord
   before_validation :evaluate_markdown, if: -> { body_markdown }
   validate :permissions, if: :commentable
 
+  scope :eager_load_serialized_data, -> { includes(:user, :commentable) }
+
   alias touch_by_reaction save
 
   def self.tree_for(commentable, limit = 0)

app/models/podcast.rb

@@ -22,6 +22,7 @@ class Podcast < ApplicationRecord
   scope :reachable, -> { where(id: PodcastEpisode.reachable.select(:podcast_id)) }
   scope :published, -> { where(published: true) }
   scope :available, -> { reachable.published }
+  scope :eager_load_serialized_data, -> { includes(:user, :podcast, :tags) }
 
   alias_attribute :path, :slug
   alias_attribute :profile_image_url, :image_url

app/models/reaction.rb

@@ -20,6 +20,7 @@ class Reaction < ApplicationRecord
 
   scope :positive, -> { where("points > ?", 0) }
   scope :readinglist, -> { where(category: "readinglist") }
+  scope :eager_load_serialized_data, -> { includes(:reactable, :user) }
 
   validates :category, inclusion: { in: CATEGORIES }
   validates :reactable_type, inclusion: { in: REACTABLE_TYPES }

app/models/tag.rb

@@ -36,6 +36,8 @@ class Tag < ActsAsTaggableOn::Tag
   after_commit :sync_related_elasticsearch_docs, on: [:update]
   after_commit :remove_from_elasticsearch, on: [:destroy]
 
+  scope :eager_load_serialized_data, -> {}
+
   include Searchable
   SEARCH_SERIALIZER = Search::TagSerializer
   SEARCH_CLASS = Search::Tag

app/models/user.rb

@@ -159,6 +159,7 @@ class User < ApplicationRecord
   scope :top_commenters, lambda { |number = 10|
     includes(:counters).order(Arel.sql("user_counters.data -> 'comments_these_7_days' DESC")).limit(number)
   }
+  scope :eager_load_serialized_data, -> { includes(:roles) }
 
   after_save :bust_cache
   after_save :subscribe_to_mailchimp_newsletter

app/services/search/base.rb

@@ -9,6 +9,22 @@ module Search
         )
       end
 
+      def bulk_index(data_hashes)
+        indexing_hashes = data_hashes.map do |data_hash|
+          model_hash = data_hash.with_indifferent_access
+          model_hash[:last_indexed_at] = Time.current
+          {
+            index: {
+              _index: self::INDEX_ALIAS,
+              _id: model_hash[:id],
+              data: model_hash
+            }
+          }
+        end
+
+        process_hashes(indexing_hashes)
+      end
+
       def find_document(doc_id)
         Search::Client.get(id: doc_id, index: self::INDEX_ALIAS)
       end
@@ -80,6 +96,30 @@ module Search
       def index_settings
         raise "Search classes must implement their own index settings"
       end
+
+      def process_hashes(indexing_hashes)
+        indexing_hashes.in_groups_of(1000, false).flat_map do |hashes|
+          indexing_chunks(hashes).select(&:any?).map { |chunk| Search::Client.bulk(body: chunk) }
+        end
+      end
+
+      def indexing_chunks(hashes)
+        return [] unless hashes.any?
+        return to_enum(__method__, hashes) unless block_given?
+
+        size = 0
+        chunk = []
+        hashes.each do |hash|
+          chunk << hash
+          size += hash.to_s.bytesize
+          next unless size > 5.megabytes
+
+          yield chunk
+          size = 0
+          chunk = []
+        end
+        yield chunk
+      end
     end
   end
 end

app/workers/search/bulk_index_to_elasticsearch_worker.rb

@@ -0,0 +1,16 @@
+module Search
+  class BulkIndexToElasticsearchWorker
+    include Sidekiq::Worker
+
+    sidekiq_options queue: :high_priority, lock: :until_executing
+
+    def perform(object_class, ids)
+      data_hashes = object_class.constantize.
+        eager_load_serialized_data.
+        where(id: ids).
+        find_each.map(&:serialized_search_hash)
+
+      "Search::#{object_class}".constantize.bulk_index(data_hashes)
+    end
+  end
+end

spec/services/search/base_spec.rb

@@ -27,6 +27,30 @@ RSpec.describe Search::Base, type: :service, elasticsearch: true do
     end
   end
 
+  describe "::bulk_index" do
+    it "indexes a set of data hashes to Elasticsearch" do
+      id_list = [123, 456, 789]
+      id_list.each do |document_id|
+        expect { described_class.find_document(document_id) }.to raise_error(Search::Errors::Transport::NotFound)
+      end
+      data_hashes = id_list.map { |id| { id: id, name: "i_am_a_tag" } }
+      described_class.bulk_index(data_hashes)
+
+      id_list.each do |document_id|
+        doc = described_class.find_document(document_id)
+        expect(doc.dig("_source", "id")).to eql(document_id)
+      end
+    end
+
+    it "sets last_indexed_at field" do
+      Timecop.freeze(Time.current) do
+        described_class.bulk_index([{ id: document_id, name: "i_am_a_tag" }])
+        last_indexed_at = described_class.find_document(document_id).dig("_source", "last_indexed_at")
+        expect(Time.zone.parse(last_indexed_at).to_i).to eq(Time.current.to_i)
+      end
+    end
+  end
+
   describe "::find_document" do
     it "fetches a document for a given ID from elasticsearch" do
       described_class.index(document_id, id: document_id)

spec/workers/search/bulk_index_to_elasticsearch_worker_spec.rb

@@ -0,0 +1,22 @@
+require "rails_helper"
+
+RSpec.describe Search::BulkIndexToElasticsearchWorker, type: :worker, elasticsearch: true do
+  let(:worker) { subject }
+  let(:article) { create(:article) }
+
+  include_examples "#enqueues_on_correct_queue", "high_priority", ["Reaction", 1]
+
+  it "indexes documents for a set of given ids and object class" do
+    reactions = [create(:reaction, reactable: article), create(:reaction), create(:reaction)]
+    Sidekiq::Worker.clear_all
+
+    reactions.each do |reaction|
+      expect { reaction.elasticsearch_doc }.to raise_error(Search::Errors::Transport::NotFound)
+    end
+    worker.perform("Reaction", reactions.map(&:id))
+
+    reactions.each do |reaction|
+      expect(reaction.elasticsearch_doc.dig("_source", "id")).to eql(reaction.id)
+    end
+  end
+end