Extend PipelineProcessWorker to accept a list of builds

This changes used worker from `BuildProcessWorker`
to `PipelineProcessWorker` to make pipeline
processing much simpler. We process `pipeline_id`,
based on some triggers.

app/models/ci/pipeline.rb

@@ -611,8 +611,8 @@ module Ci
     end
 
     # rubocop: disable CodeReuse/ServiceClass
-    def process!(trigger_build_name = nil)
-      Ci::ProcessPipelineService.new(project, user).execute(self, trigger_build_name)
+    def process!(trigger_build_ids = nil)
+      Ci::ProcessPipelineService.new(project, user).execute(self, trigger_build_ids)
     end
     # rubocop: enable CodeReuse/ServiceClass
 

app/models/commit_status.rb

@@ -122,7 +122,7 @@ class CommitStatus < ApplicationRecord
       commit_status.run_after_commit do
         if pipeline_id
           if complete? || manual?
-            BuildProcessWorker.perform_async(id)
+            PipelineProcessWorker.perform_async(pipeline_id, [id])
           else
             PipelineUpdateWorker.perform_async(pipeline_id)
           end

app/services/ci/process_pipeline_service.rb

@@ -4,7 +4,7 @@ module Ci
   class ProcessPipelineService < BaseService
     attr_reader :pipeline
 
-    def execute(pipeline, trigger_build_name = nil)
+    def execute(pipeline, trigger_build_ids = nil)
       @pipeline = pipeline
 
       update_retried
@@ -16,7 +16,7 @@ module Ci
 
       # we evaluate dependent needs,
       # only when the another job has finished
-      success = process_builds_with_needs(trigger_build_name) || success
+      success = process_builds_with_needs(trigger_build_ids) || success
 
       @pipeline.update_status
 
@@ -40,12 +40,18 @@ module Ci
       end
     end
 
-    def process_builds_with_needs(trigger_build_name)
-      return false unless trigger_build_name
+    def process_builds_with_needs(trigger_build_ids)
+      return false unless trigger_build_ids.present?
       return false unless Feature.enabled?(:ci_dag_support, project)
 
+      # rubocop: disable CodeReuse/ActiveRecord
+      trigger_build_names = pipeline.statuses
+        .where(id: trigger_build_ids)
+        .select(:name)
+      # rubocop: enable CodeReuse/ActiveRecord
+
       created_processables
-        .with_needs(trigger_build_name)
+        .with_needs(trigger_build_names)
         .find_each
         .map(&method(:process_build_with_needs))
         .any?

app/workers/build_process_worker.rb

@@ -9,7 +9,7 @@ class BuildProcessWorker
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(build_id)
     CommitStatus.find_by(id: build_id).try do |build|
-      build.pipeline.process!(build.name)
+      build.pipeline.process!([build_id])
     end
   end
   # rubocop: enable CodeReuse/ActiveRecord

app/workers/pipeline_process_worker.rb

@@ -7,9 +7,10 @@ class PipelineProcessWorker
   queue_namespace :pipeline_processing
 
   # rubocop: disable CodeReuse/ActiveRecord
-  def perform(pipeline_id)
-    Ci::Pipeline.find_by(id: pipeline_id)
-      .try(:process!)
+  def perform(pipeline_id, build_ids = nil)
+    Ci::Pipeline.find_by(id: pipeline_id).try do |pipeline|
+      pipeline.process!(build_ids)
+    end
   end
   # rubocop: enable CodeReuse/ActiveRecord
 end

spec/workers/build_process_worker_spec.rb

@@ -10,7 +10,7 @@ describe BuildProcessWorker do
 
       it 'processes build' do
         expect_any_instance_of(Ci::Pipeline).to receive(:process!)
-          .with(build.name)
+          .with([build.id])
 
         described_class.new.perform(build.id)
       end

spec/workers/pipeline_process_worker_spec.rb

@@ -12,6 +12,17 @@ describe PipelineProcessWorker do
 
         described_class.new.perform(pipeline.id)
       end
+
+      context 'when build_ids are passed' do
+        let(:build) { create(:ci_build, pipeline: pipeline, name: 'my-build') }
+
+        it 'processes pipeline with a list of builds' do
+          expect_any_instance_of(Ci::Pipeline).to receive(:process!)
+            .with([build.id])
+
+          described_class.new.perform(pipeline.id, [build.id])
+        end
+      end
     end
 
     context 'when pipeline does not exist' do