Add upvotes field to issues index

EE: true
Changelog: changed

app/models/award_emoji.rb

@@ -64,3 +64,5 @@ class AwardEmoji < ApplicationRecord
     awardable.try(:expire_etag_cache)
   end
 end
+
+AwardEmoji.prepend_mod_with('AwardEmoji')

ee/app/models/ee/award_emoji.rb

+# frozen_string_literal: true
+
+module EE
+  module AwardEmoji
+    extend ActiveSupport::Concern
+
+    prepended do
+      UPDATE_ELASTIC_ASSOCIATIONS_FOR = [::Issue].freeze
+
+      after_commit :update_elastic_associations, on: [:create, :destroy]
+
+      def update_elastic_associations
+        return unless UPDATE_ELASTIC_ASSOCIATIONS_FOR.any? { |model| awardable.is_a?(model) }
+        return unless awardable.maintaining_elasticsearch?
+
+        awardable.maintain_elasticsearch_update
+      end
+    end
+  end
+end

ee/elastic/migrate/20210623081800_add_upvotes_to_issues.rb

+# frozen_string_literal: true
+
+class AddUpvotesToIssues < Elastic::Migration
+  batched!
+  throttle_delay 3.minutes
+
+  DOCUMENT_TYPE = Issue
+  QUERY_BATCH_SIZE = 5000
+  UPDATE_BATCH_SIZE = 100
+
+  def migrate
+    update_mappings!
+
+    if completed?
+      log "Skipping adding upvotes field to issues documents migration since it is already applied"
+      return
+    end
+
+    log "Adding upvotes field to issues documents for batch of #{QUERY_BATCH_SIZE} documents"
+
+    document_references = process_batch!
+
+    log "Adding upvotes field to issues documents is completed for batch of #{document_references.size} documents"
+  end
+
+  def completed?
+    helper.refresh_index(index_name: index_name)
+
+    query = {
+      size: 0,
+      aggs: {
+        issues: {
+          filter: issues_missing_upvotes_field
+        }
+      }
+    }
+
+    results = client.search(index: index_name, body: query)
+    doc_count = results.dig('aggregations', 'issues', 'doc_count')
+
+    log "Checking if there are documents without upvotes field: #{doc_count} documents left"
+
+    doc_count == 0
+  end
+
+  private
+
+  def update_mappings!
+    client.indices.put_mapping index: index_name, body: {
+      properties: {
+        upvotes: { type: 'integer' }
+      }
+    }
+  end
+
+  def process_batch!
+    query = {
+      size: QUERY_BATCH_SIZE,
+      query: {
+        bool: {
+          filter: issues_missing_upvotes_field
+        }
+      }
+    }
+
+    results = client.search(index: index_name, body: query)
+    hits = results.dig('hits', 'hits') || []
+
+    document_references = hits.map! do |hit|
+      id = hit.dig('_source', 'id')
+      es_id = hit['_id']
+      es_parent = "project_#{hit.dig('_source', 'project_id')}"
+
+      Gitlab::Elastic::DocumentReference.new(DOCUMENT_TYPE, id, es_id, es_parent)
+    end
+
+    document_references.each_slice(UPDATE_BATCH_SIZE) do |refs|
+      Elastic::ProcessInitialBookkeepingService.track!(*refs)
+    end
+
+    document_references
+  end
+
+  def issues_missing_upvotes_field
+    {
+      bool: {
+        must_not: {
+          exists: {
+            field: 'upvotes'
+          }
+        }
+      }
+    }
+  end
+
+  def index_name
+    DOCUMENT_TYPE.__elasticsearch__.index_name
+  end
+end

ee/lib/elastic/latest/issue_class_proxy.rb

@@ -30,7 +30,7 @@ module Elastic
 
       # rubocop: disable CodeReuse/ActiveRecord
       def preload_indexing_data(relation)
-        relation.includes(:issue_assignees, project: [:project_feature])
+        relation.includes(:issue_assignees, :award_emoji, project: [:project_feature])
       end
       # rubocop: enable CodeReuse/ActiveRecord
 

ee/lib/elastic/latest/issue_config.rb

@@ -35,6 +35,7 @@ module Elastic
 
         indexes :visibility_level, type: :integer
         indexes :issues_access_level, type: :integer
+        indexes :upvotes, type: :integer
       end
     end
   end

ee/lib/elastic/latest/issue_instance_proxy.rb

@@ -20,6 +20,8 @@ module Elastic
         data['visibility_level'] = target.project.visibility_level
         data['issues_access_level'] = safely_read_project_feature_for_elasticsearch(:issues)
 
+        data['upvotes'] = count_emoji
+
         data.merge(generic_attributes)
       end
 
@@ -28,6 +30,12 @@ module Elastic
       def generic_attributes
         super.except('join_field')
       end
+
+      def count_emoji
+        target.award_emoji.count do |c|
+          c.name == AwardEmoji::UPVOTE_NAME
+        end
+      end
     end
   end
 end

ee/spec/elastic/migrate/20210623081800_add_upvotes_to_issues_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require File.expand_path('ee/elastic/migrate/20210623081800_add_upvotes_to_issues.rb')
+
+RSpec.describe AddUpvotesToIssues, :elastic, :sidekiq_inline do
+  let(:version) { 20210623081800 }
+  let(:migration) { described_class.new(version) }
+  let(:issues) { create_list(:issue, 3) }
+
+  before do
+    stub_ee_application_setting(elasticsearch_search: true, elasticsearch_indexing: true)
+
+    # ensure issues are indexed
+    issues
+
+    ensure_elasticsearch_index!
+  end
+
+  describe 'migration_options' do
+    it 'has migration options set', :aggregate_failures do
+      expect(migration).to be_batched
+      expect(migration.throttle_delay).to eq(3.minutes)
+    end
+  end
+
+  describe '.migrate' do
+    subject { migration.migrate }
+
+    context 'when migration is already completed' do
+      it 'does not modify data' do
+        expect(::Elastic::ProcessInitialBookkeepingService).not_to receive(:track!)
+
+        subject
+      end
+    end
+
+    context 'migration process' do
+      before do
+        remove_upvotes_from_issues(issues)
+      end
+
+      it 'updates all issue documents' do
+        # track calls are batched in groups of 100
+        expect(::Elastic::ProcessInitialBookkeepingService).to receive(:track!).once do |*tracked_refs|
+          expect(tracked_refs.count).to eq(3)
+        end
+
+        subject
+      end
+
+      it 'only updates issue documents missing upvotes', :aggregate_failures do
+        issue = issues.first
+        add_upvotes_for_issues(issues[1..-1])
+
+        expected = [Gitlab::Elastic::DocumentReference.new(Issue, issue.id, issue.es_id, issue.es_parent)]
+        expect(::Elastic::ProcessInitialBookkeepingService).to receive(:track!).with(*expected).once
+
+        subject
+      end
+
+      it 'processes in batches', :aggregate_failures do
+        stub_const("#{described_class}::QUERY_BATCH_SIZE", 2)
+        stub_const("#{described_class}::UPDATE_BATCH_SIZE", 1)
+
+        expect(::Elastic::ProcessInitialBookkeepingService).to receive(:track!).exactly(3).times.and_call_original
+
+        # cannot use subject in spec because it is memoized
+        migration.migrate
+
+        ensure_elasticsearch_index!
+
+        migration.migrate
+      end
+    end
+  end
+
+  describe '.completed?' do
+    context 'when documents are missing upvotes' do
+      before do
+        remove_upvotes_from_issues(issues)
+      end
+
+      specify { expect(migration).not_to be_completed }
+    end
+
+    context 'when no documents are missing upvotes' do
+      specify { expect(migration).to be_completed }
+    end
+  end
+
+  private
+
+  def add_upvotes_for_issues(issues)
+    script =  {
+      source: "ctx._source['upvotes'] = params.upvotes;",
+      lang: "painless",
+      params: {
+        upvotes: 0
+      }
+    }
+
+    update_by_query(issues, script)
+  end
+
+  def remove_upvotes_from_issues(issues)
+    script =  {
+      source: "ctx._source.remove('upvotes')"
+    }
+
+    update_by_query(issues, script)
+  end
+
+  def update_by_query(issues, script)
+    issue_ids = issues.map { |i| i.id }
+
+    client = Issue.__elasticsearch__.client
+    client.update_by_query({
+      index: Issue.__elasticsearch__.index_name,
+      wait_for_completion: true, # run synchronously
+      refresh: true, # make operation visible to search
+      body: {
+        script: script,
+        query: {
+          bool: {
+            must: [
+              {
+                terms: {
+                  id: issue_ids
+                }
+              }
+            ]
+          }
+        }
+      }
+    })
+  end
+end

ee/spec/models/award_emoji_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AwardEmoji do
+  describe '#update_elastic_associations' do
+    let_it_be(:issue) { create(:issue) }
+    let_it_be(:merge_request) { create(:merge_request) }
+
+    context 'maintaining_elasticsearch is true' do
+      before do
+        allow(issue).to receive(:maintaining_elasticsearch?).and_return(true)
+        allow(merge_request).to receive(:maintaining_elasticsearch?).and_return(true)
+      end
+
+      it 'calls maintain_elasticsearch_update on create' do
+        expect(issue).to receive(:maintain_elasticsearch_update)
+
+        create(:award_emoji, :upvote, awardable: issue)
+      end
+
+      it 'calls maintain_elasticsearch_update on destroy' do
+        award_emoji = create(:award_emoji, :upvote, awardable: issue)
+
+        expect(issue).to receive(:maintain_elasticsearch_update)
+
+        award_emoji.destroy!
+      end
+
+      it 'does nothing for other awardable_type' do
+        expect(merge_request).not_to receive(:maintain_elasticsearch_update)
+
+        create(:award_emoji, :upvote, awardable: merge_request)
+      end
+    end
+
+    context 'maintaining_elasticsearch is false' do
+      it 'does not call maintain_elasticsearch_update' do
+        expect(issue).not_to receive(:maintain_elasticsearch_update)
+
+        award_emoji = create(:award_emoji, :upvote, awardable: issue)
+
+        expect(issue).not_to receive(:maintain_elasticsearch_update)
+
+        award_emoji.destroy!
+      end
+    end
+  end
+end

ee/spec/models/concerns/elastic/issue_spec.rb

@@ -109,6 +109,7 @@ RSpec.describe Issue, :elastic do
     assignee = create(:user)
     project = create(:project, :internal)
     issue = create :issue, project: project, assignees: [assignee]
+    create(:award_emoji, :upvote, awardable: issue)
 
     expected_hash = issue.attributes.extract!(
       'id',
@@ -122,7 +123,8 @@ RSpec.describe Issue, :elastic do
       'confidential'
     ).merge({
       'type' => issue.es_type,
-      'state' => issue.state
+      'state' => issue.state,
+      'upvotes' => 1
     })
 
     expected_hash['assignee_id'] = [assignee.id]