[deploy] Resync Elasticsearch Docs that had Previous Bugs (#7230)

lib/data_update_scripts/20200410152018_resync_elasticsearch_documents.rb

@@ -0,0 +1,75 @@
+module DataUpdateScripts
+  class ResyncElasticsearchDocuments
+    def run
+      # Previous bug: Not getting removed properly
+      sync_docs(Article.pluck(:id), "Article")
+      sync_docs(Comment.pluck(:id), "Comment")
+      sync_docs(User.pluck(:id), "User")
+
+      # Previous bug: Not getting indexed on creation properly
+      index_docs(PodcastEpisode.pluck(:id), "PodcastEpisode")
+      index_docs(Tag.pluck(:id), "Tag")
+    end
+
+    private
+
+    def index_docs(ids, doc_type)
+      ids.each do |id|
+        Search::IndexToElasticsearchWorker.set(queue: :low_priority).perform_async(
+          doc_type, id
+        )
+      end
+    end
+
+    # This is essentially doing a "find_each" in Elasticsearch by scrolling through
+    # all of the documents and collecting their IDs. Then we remove the mismatched Postgres ones
+    def sync_docs(db_ids, doc_type)
+      es_ids = []
+      response = Search::Client.search(
+        index: search_class(doc_type)::INDEX_ALIAS, scroll: "2m", body: search_body(doc_type), size: 3000,
+      )
+      loop do
+        hits = response.dig("hits", "hits")
+        break if hits.empty?
+
+        hits.each do |hit|
+          es_ids << hit["_id"].split("_").last.to_i
+        end
+
+        response = Search::Client.scroll(body: { scroll_id: response["_scroll_id"] }, scroll: "2m")
+      end
+
+      remove_ids(db_ids, es_ids, doc_type)
+    end
+
+    def ids_to_remove(db_ids, es_ids, doc_type)
+      dif = es_ids - db_ids
+      doc_type == "User" ? dif : dif.map { |id| "#{doc_type.downcase}_#{id}" }
+    end
+
+    def remove_ids(db_ids, es_ids, doc_type)
+      dif_ids = ids_to_remove(db_ids, es_ids, doc_type)
+      index_name = search_class(doc_type)::INDEX_ALIAS
+      dif_ids.map do |id|
+        Search::Client.delete(id: id, index: index_name)
+      end
+    end
+
+    def search_body(doc_type)
+      if doc_type == "User"
+        { query: { match_all: {} }, _source: [:id] }
+      else
+        {
+          query: {
+            bool: { filter: { term: { class_name: doc_type } } }
+          },
+          _source: [:id]
+        }
+      end
+    end
+
+    def search_class(doc_type)
+      doc_type == "User" ? Search::User : Search::FeedContent
+    end
+  end
+end

spec/lib/data_update_scripts/resync_elasticsearch_documents_spec.rb

@@ -0,0 +1,51 @@
+require "rails_helper"
+require Rails.root.join("lib/data_update_scripts/20200410152018_resync_elasticsearch_documents.rb")
+
+describe DataUpdateScripts::ResyncElasticsearchDocuments, elasticsearch: true do
+  it "indexes podcast episodes and tags to Elasticsearch" do
+    tag = create(:tag)
+    podcast_episode = create(:podcast_episode)
+    Sidekiq::Worker.clear_all
+
+    expect { tag.elasticsearch_doc }.to raise_error(Search::Errors::Transport::NotFound)
+    expect { podcast_episode.elasticsearch_doc }.to raise_error(Search::Errors::Transport::NotFound)
+
+    sidekiq_perform_enqueued_jobs { described_class.new.run }
+    expect(tag.elasticsearch_doc).not_to be_nil
+    expect(podcast_episode.elasticsearch_doc).not_to be_nil
+  end
+
+  it "syncs articles, comments, and users to Elasticsearch", :aggregate_failures do
+    article = create(:article)
+    comment = create(:comment)
+    user = create(:user)
+    index_real_and_mock_documents(article, comment, user)
+
+    expect(Article::SEARCH_CLASS.articles_document_count).not_to eq(Article.count)
+    expect(Comment::SEARCH_CLASS.comments_document_count).not_to eq(Comment.count)
+    expect(User::SEARCH_CLASS.document_count).not_to eq(User.count)
+
+    sidekiq_perform_enqueued_jobs { described_class.new.run }
+    refresh_indexes
+
+    expect(article.elasticsearch_doc).not_to be_nil
+    expect(user.elasticsearch_doc).not_to be_nil
+    expect(comment.elasticsearch_doc).not_to be_nil
+    expect(Article::SEARCH_CLASS.articles_document_count).to eq(Article.count)
+    expect(Comment::SEARCH_CLASS.comments_document_count).to eq(Comment.count)
+    expect(User::SEARCH_CLASS.document_count).to eq(User.count)
+  end
+
+  def index_real_and_mock_documents(article, comment, user)
+    sidekiq_perform_enqueued_jobs
+    Article::SEARCH_CLASS.index("article_#{article.id + 100}", class_name: "Article")
+    Comment::SEARCH_CLASS.index("comment_#{comment.id + 100}", class_name: "Comment")
+    User::SEARCH_CLASS.index(user.id + 100, name: "User")
+    refresh_indexes
+  end
+
+  def refresh_indexes
+    User::SEARCH_CLASS.refresh_index
+    Article::SEARCH_CLASS.refresh_index
+  end
+end