Refactor processing of ref changes in PostReceive

To support creating push events in bulk and aggregate them by
type (branch or tag) and action (created, pushed, deleted), we
need to be able to segregate the changes into groups.

Having that inside the `PostReceive` worker class will increase
its reponsibility further. To help with that, a new service
called `Git::ProcessRefChangesService` has been added.

It's now responsible for processing branch/tag changes and
delegate it to the appropriate push service class.

This will be later on used to determime whether we need to create
individual or bulk push events.

app/services/git/process_ref_changes_service.rb

+# frozen_string_literal: true
+
+module Git
+  class ProcessRefChangesService < BaseService
+    PIPELINE_PROCESS_LIMIT = 4
+
+    def execute
+      changes = params[:changes]
+
+      process_changes_by_action(:branch, changes.branch_changes)
+      process_changes_by_action(:tag, changes.tag_changes)
+    end
+
+    private
+
+    def process_changes_by_action(ref_type, changes)
+      changes_by_action = group_changes_by_action(changes)
+
+      changes_by_action.each do |_, changes|
+        process_changes(ref_type, changes) if changes.any?
+      end
+    end
+
+    def group_changes_by_action(changes)
+      changes.group_by do |change|
+        change_action(change)
+      end
+    end
+
+    def change_action(change)
+      return :created if Gitlab::Git.blank_ref?(change[:oldrev])
+      return :removed if Gitlab::Git.blank_ref?(change[:newrev])
+
+      :pushed
+    end
+
+    def process_changes(ref_type, changes)
+      push_service_class = push_service_class_for(ref_type)
+
+      changes.each do |change|
+        push_service_class.new(
+          project,
+          current_user,
+          change: change,
+          push_options: params[:push_options],
+          create_pipelines: change[:index] < PIPELINE_PROCESS_LIMIT || Feature.enabled?(:git_push_create_all_pipelines, project)
+        ).execute
+      end
+    end
+
+    def push_service_class_for(ref_type)
+      return Git::TagPushService if ref_type == :tag
+
+      Git::BranchPushService
+    end
+  end
+end

app/workers/post_receive.rb

@@ -3,8 +3,6 @@
 class PostReceive
   include ApplicationWorker
 
-  PIPELINE_PROCESS_LIMIT = 4
-
   def perform(gl_repository, identifier, changes, push_options = {})
     project, repo_type = Gitlab::GlRepository.parse(gl_repository)
 
@@ -49,8 +47,7 @@ class PostReceive
     expire_caches(post_received, post_received.project.repository)
     enqueue_repository_cache_update(post_received)
 
-    process_changes(Git::BranchPushService, project, user, push_options, changes.branch_changes)
-    process_changes(Git::TagPushService, project, user, push_options, changes.tag_changes)
+    process_ref_changes(project, user, push_options: push_options, changes: changes)
     update_remote_mirrors(post_received)
     after_project_changes_hooks(project, user, changes.refs, changes.repository_data)
   end
@@ -75,18 +72,10 @@ class PostReceive
     )
   end
 
-  def process_changes(service_class, project, user, push_options, changes)
-    return if changes.empty?
-
-    changes.each do |change|
-      service_class.new(
-        project,
-        user,
-        change: change,
-        push_options: push_options,
-        create_pipelines: change[:index] < PIPELINE_PROCESS_LIMIT || Feature.enabled?(:git_push_create_all_pipelines, project)
-      ).execute
-    end
+  def process_ref_changes(project, user, params = {})
+    return unless params[:changes].any?
+
+    Git::ProcessRefChangesService.new(project, user, params).execute
   end
 
   def update_remote_mirrors(post_received)

spec/services/git/process_ref_changes_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Git::ProcessRefChangesService do
+  let(:project) { create(:project, :repository) }
+  let(:user) { project.owner }
+  let(:params) { { changes: git_changes } }
+
+  subject { described_class.new(project, user, params) }
+
+  shared_examples_for 'service for processing ref changes' do |push_service_class|
+    let(:service) { double(execute: true) }
+    let(:git_changes) { double(branch_changes: [], tag_changes: []) }
+
+    let(:changes) do
+      [
+        { index: 0, oldrev: Gitlab::Git::BLANK_SHA, newrev: '789012', ref: "#{ref_prefix}/create" },
+        { index: 1, oldrev: '123456', newrev: '789012', ref: "#{ref_prefix}/update" },
+        { index: 2, oldrev: '123456', newrev: Gitlab::Git::BLANK_SHA, ref: "#{ref_prefix}/delete" }
+      ]
+    end
+
+    before do
+      allow(git_changes).to receive(changes_method).and_return(changes)
+    end
+
+    it "calls #{push_service_class}" do
+      expect(push_service_class)
+        .to receive(:new)
+        .exactly(changes.count).times
+        .and_return(service)
+
+      subject.execute
+    end
+
+    context 'pipeline creation' do
+      context 'with valid .gitlab-ci.yml' do
+        before do
+          stub_ci_pipeline_to_return_yaml_file
+
+          allow_any_instance_of(Project)
+            .to receive(:commit)
+            .and_return(project.commit)
+
+          allow_any_instance_of(Repository)
+            .to receive(:branch_exists?)
+            .and_return(true)
+        end
+
+        context 'when git_push_create_all_pipelines is disabled' do
+          before do
+            stub_feature_flags(git_push_create_all_pipelines: false)
+          end
+
+          it 'creates pipeline for branches and tags' do
+            subject.execute
+
+            expect(Ci::Pipeline.pluck(:ref)).to contain_exactly('create', 'update', 'delete')
+          end
+
+          it "creates exactly #{described_class::PIPELINE_PROCESS_LIMIT} pipelines" do
+            stub_const("#{described_class}::PIPELINE_PROCESS_LIMIT", changes.count - 1)
+
+            expect { subject.execute }.to change { Ci::Pipeline.count }.by(described_class::PIPELINE_PROCESS_LIMIT)
+          end
+        end
+
+        context 'when git_push_create_all_pipelines is enabled' do
+          before do
+            stub_feature_flags(git_push_create_all_pipelines: true)
+          end
+
+          it 'creates all pipelines' do
+            expect { subject.execute }.to change { Ci::Pipeline.count }.by(changes.count)
+          end
+        end
+      end
+
+      context 'with invalid .gitlab-ci.yml' do
+        before do
+          stub_ci_pipeline_yaml_file(nil)
+        end
+
+        it 'does not create a pipeline' do
+          expect { subject.execute }.not_to change { Ci::Pipeline.count }
+        end
+      end
+    end
+  end
+
+  context 'branch changes' do
+    let(:changes_method) { :branch_changes }
+    let(:ref_prefix) { 'refs/heads' }
+
+    it_behaves_like 'service for processing ref changes', Git::BranchPushService
+  end
+
+  context 'tag changes' do
+    let(:changes_method) { :tag_changes }
+    let(:ref_prefix) { 'refs/tags' }
+
+    it_behaves_like 'service for processing ref changes', Git::TagPushService
+  end
+end

spec/workers/post_receive_spec.rb

@@ -73,8 +73,7 @@ describe PostReceive do
 
     context 'empty changes' do
       it "does not call any PushService but runs after project hooks" do
-        expect(Git::BranchPushService).not_to receive(:new)
-        expect(Git::TagPushService).not_to receive(:new)
+        expect(Git::ProcessRefChangesService).not_to receive(:new)
         expect_next_instance_of(SystemHooksService) { |service| expect(service).to receive(:execute_hooks) }
 
         perform(changes: "")
@@ -87,8 +86,7 @@ describe PostReceive do
       let!(:key_id) { "" }
 
       it 'returns false' do
-        expect(Git::BranchPushService).not_to receive(:new)
-        expect(Git::TagPushService).not_to receive(:new)
+        expect(Git::ProcessRefChangesService).not_to receive(:new)
 
         expect(perform).to be false
       end
@@ -131,13 +129,11 @@ describe PostReceive do
           perform
         end
 
-        it 'calls Git::BranchPushService' do
-          expect_any_instance_of(Git::BranchPushService) do |service|
+        it 'calls Git::ProcessRefChangesService' do
+          expect_next_instance_of(Git::ProcessRefChangesService) do |service|
             expect(service).to receive(:execute).and_return(true)
           end
 
-          expect(Git::TagPushService).not_to receive(:new)
-
           perform
         end
 
@@ -174,8 +170,6 @@ describe PostReceive do
             654321 210987 refs/tags/tag1
             654322 210986 refs/tags/tag2
             654323 210985 refs/tags/tag3
-            654324 210984 refs/tags/tag4
-            654325 210983 refs/tags/tag5
           EOF
         end
 
@@ -189,23 +183,19 @@ describe PostReceive do
           perform
         end
 
-        it "only invalidates tags once" do
-          expect(project.repository).to receive(:repository_event).exactly(5).times.with(:push_tag).and_call_original
+        it 'only invalidates tags once' do
+          expect(project.repository).to receive(:repository_event).exactly(3).times.with(:push_tag).and_call_original
           expect(project.repository).to receive(:expire_caches_for_tags).once.and_call_original
           expect(project.repository).to receive(:expire_tags_cache).once.and_call_original
 
           perform
         end
 
-        it "calls Git::TagPushService" do
-          expect(Git::BranchPushService).not_to receive(:new)
-
-          expect_any_instance_of(Git::TagPushService) do |service|
+        it 'calls Git::ProcessRefChangesService' do
+          expect_next_instance_of(Git::ProcessRefChangesService) do |service|
             expect(service).to receive(:execute).and_return(true)
           end
 
-          expect(Git::BranchPushService).not_to receive(:new)
-
           perform
         end
 
@@ -223,8 +213,7 @@ describe PostReceive do
         let(:changes) { "123456 789012 refs/merge-requests/123" }
 
         it "does not call any of the services" do
-          expect(Git::BranchPushService).not_to receive(:new)
-          expect(Git::TagPushService).not_to receive(:new)
+          expect(Git::ProcessRefChangesService).not_to receive(:new)
 
           perform
         end
@@ -232,72 +221,6 @@ describe PostReceive do
         it_behaves_like 'not updating remote mirrors'
       end
 
-      context "gitlab-ci.yml" do
-        let(:changes) do
-          <<-EOF.strip_heredoc
-            123456 789012 refs/heads/feature
-            654321 210987 refs/tags/tag
-            123456 789012 refs/heads/feature2
-            123458 789013 refs/heads/feature3
-            123459 789015 refs/heads/feature4
-          EOF
-        end
-
-        let(:changes_count) { changes.lines.count }
-
-        subject { perform }
-
-        context "with valid .gitlab-ci.yml" do
-          before do
-            stub_ci_pipeline_to_return_yaml_file
-
-            allow_any_instance_of(Project)
-              .to receive(:commit)
-              .and_return(project.commit)
-
-            allow_any_instance_of(Repository)
-              .to receive(:branch_exists?)
-              .and_return(true)
-          end
-
-          context 'when git_push_create_all_pipelines is disabled' do
-            before do
-              stub_feature_flags(git_push_create_all_pipelines: false)
-            end
-
-            it "creates pipeline for branches and tags" do
-              subject
-
-              expect(Ci::Pipeline.pluck(:ref)).to contain_exactly("feature", "tag", "feature2", "feature3")
-            end
-
-            it "creates exactly #{described_class::PIPELINE_PROCESS_LIMIT} pipelines" do
-              expect(changes_count).to be > described_class::PIPELINE_PROCESS_LIMIT
-
-              expect { subject }.to change { Ci::Pipeline.count }.by(described_class::PIPELINE_PROCESS_LIMIT)
-            end
-          end
-
-          context 'when git_push_create_all_pipelines is enabled' do
-            before do
-              stub_feature_flags(git_push_create_all_pipelines: true)
-            end
-
-            it "creates all pipelines" do
-              expect { subject }.to change { Ci::Pipeline.count }.by(changes_count)
-            end
-          end
-        end
-
-        context "does not create a Ci::Pipeline" do
-          before do
-            stub_ci_pipeline_yaml_file(nil)
-          end
-
-          it { expect { subject }.not_to change { Ci::Pipeline.count } }
-        end
-      end
-
       context 'after project changes hooks' do
         let(:changes) { '123456 789012 refs/heads/tést' }
         let(:fake_hook_data) { Hash.new(event_name: 'repository_update') }
@@ -306,7 +229,7 @@ describe PostReceive do
           allow_any_instance_of(Gitlab::DataBuilder::Repository).to receive(:update).and_return(fake_hook_data)
           # silence hooks so we can isolate
           allow_any_instance_of(Key).to receive(:post_create_hook).and_return(true)
-          expect_next_instance_of(Git::BranchPushService) do |service|
+          expect_next_instance_of(Git::ProcessRefChangesService) do |service|
             expect(service).to receive(:execute).and_return(true)
           end
         end