Update integrations using batching and queue

This updates all inherited integrations using the same approach of
batching and Sidekiq queue that we use when bulk create integrations for
project and group integrations.

app/models/service.rb

@@ -63,6 +63,7 @@ class Service < ApplicationRecord
   scope :active, -> { where(active: true) }
   scope :by_type, -> (type) { where(type: type) }
   scope :by_active_flag, -> (flag) { where(active: flag) }
+  scope :inherit_from_id, -> (id) { where(inherit_from_id: id) }
   scope :for_group, -> (group) { where(group_id: group, type: available_services_types) }
   scope :for_template, -> { where(template: true, type: available_services_types) }
   scope :for_instance, -> { where(instance: true, type: available_services_types) }

app/services/admin/propagate_integration_service.rb

@@ -5,39 +5,22 @@ module Admin
     include PropagateService
 
     def propagate
+      update_inherited_integrations
+
       create_integration_for_groups_without_integration if Feature.enabled?(:group_level_integrations)
       create_integration_for_projects_without_integration
-      update_inherited_integrations
     end
 
     private
 
     # rubocop: disable Cop/InBatches
-    # rubocop: disable CodeReuse/ActiveRecord
     def update_inherited_integrations
-      Service.where(type: integration.type, inherit_from_id: integration.id).in_batches(of: BATCH_SIZE) do |batch|
-        bulk_update_from_integration(batch)
+      Service.by_type(integration.type).inherit_from_id(integration.id).in_batches(of: BATCH_SIZE) do |services|
+        min_id, max_id = services.pick("MIN(services.id), MAX(services.id)")
+        PropagateIntegrationInheritWorker.perform_async(integration.id, min_id, max_id)
       end
     end
     # rubocop: enable Cop/InBatches
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def bulk_update_from_integration(batch)
-      # Retrieving the IDs instantiates the ActiveRecord relation (batch)
-      # into concrete models, otherwise update_all will clear the relation.
-      # https://stackoverflow.com/q/34811646/462015
-      batch_ids = batch.pluck(:id)
-
-      Service.transaction do
-        batch.update_all(service_hash)
-
-        if integration.data_fields_present?
-          integration.data_fields.class.where(service_id: batch_ids).update_all(data_fields_hash)
-        end
-      end
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     def create_integration_for_groups_without_integration
       Group.without_integration(integration).each_batch(of: BATCH_SIZE) do |groups|
@@ -45,13 +28,5 @@ module Admin
         PropagateIntegrationGroupWorker.perform_async(integration.id, min_id, max_id)
       end
     end
-
-    def service_hash
-      integration.to_service_hash.tap { |json| json['inherit_from_id'] = integration.id }
-    end
-
-    def data_fields_hash
-      integration.to_data_fields_hash
-    end
   end
 end

app/services/bulk_update_integration_service.rb

+# frozen_string_literal: true
+
+class BulkUpdateIntegrationService
+  def initialize(integration, batch)
+    @integration = integration
+    @batch = batch
+  end
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def execute
+    # Retrieving the IDs instantiates the ActiveRecord relation (batch)
+    # into concrete models, otherwise update_all will clear the relation.
+    # https://stackoverflow.com/q/34811646/462015
+    batch_ids = batch.pluck(:id)
+
+    Service.transaction do
+      batch.update_all(service_hash)
+
+      if integration.data_fields_present?
+        integration.data_fields.class.where(service_id: batch_ids).update_all(data_fields_hash)
+      end
+    end
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+
+  private
+
+  attr_reader :integration, :batch
+
+  def service_hash
+    integration.to_service_hash.tap { |json| json['inherit_from_id'] = integration.id }
+  end
+
+  def data_fields_hash
+    integration.to_data_fields_hash
+  end
+end

app/workers/all_queues.yml

@@ -1724,6 +1724,14 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: propagate_integration_inherit
+  :feature_category: :integrations
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: propagate_integration_project
   :feature_category: :integrations
   :has_external_dependencies:

app/workers/propagate_integration_inherit_worker.rb

+# frozen_string_literal: true
+
+class PropagateIntegrationInheritWorker
+  include ApplicationWorker
+
+  feature_category :integrations
+  idempotent!
+  loggable_arguments 1
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform(integration_id, min_id, max_id)
+    integration = Service.find(integration_id)
+    services = Service.where(id: min_id..max_id).by_type(integration.type).inherit_from_id(integration.id)
+
+    BulkUpdateIntegrationService.new(integration, services).execute
+  end
+  # rubocop: enable CodeReuse/ActiveRecord
+end

config/sidekiq_queues.yml

@@ -234,6 +234,8 @@
   - 1
 - - propagate_integration_group
   - 1
+- - propagate_integration_inherit
+  - 1
 - - propagate_integration_project
   - 1
 - - propagate_service_template

spec/services/admin/propagate_integration_service_spec.rb

@@ -11,7 +11,6 @@ RSpec.describe Admin::PropagateIntegrationService do
     end
 
     let_it_be(:project) { create(:project) }
-    let(:excluded_attributes) { %w[id project_id group_id inherit_from_id instance template created_at updated_at] }
     let!(:instance_integration) do
       JiraService.create!(
         instance: true,
@@ -66,41 +65,11 @@ RSpec.describe Admin::PropagateIntegrationService do
     context 'with inherited integration' do
       let(:integration) { inherited_integration }
 
-      it 'updates the integration' do
-        described_class.propagate(instance_integration)
-
-        expect(integration.reload.inherit_from_id).to eq(instance_integration.id)
-        expect(integration.attributes.except(*excluded_attributes))
-          .to eq(instance_integration.attributes.except(*excluded_attributes))
-      end
-
-      context 'with integration with data fields' do
-        let(:excluded_attributes) { %w[id service_id created_at updated_at] }
+      it 'calls to PropagateIntegrationProjectWorker' do
+        expect(PropagateIntegrationInheritWorker).to receive(:perform_async)
+          .with(instance_integration.id, inherited_integration.id, inherited_integration.id)
 
-        it 'updates the data fields from the integration' do
         described_class.propagate(instance_integration)
-
-          expect(integration.reload.data_fields.attributes.except(*excluded_attributes))
-            .to eq(instance_integration.data_fields.attributes.except(*excluded_attributes))
-        end
-      end
-    end
-
-    context 'with not inherited integration' do
-      let(:integration) { not_inherited_integration }
-
-      it 'does not update the integration' do
-        expect { described_class.propagate(instance_integration) }
-          .not_to change { instance_integration.attributes.except(*excluded_attributes) }
-      end
-    end
-
-    context 'with different type inherited integration' do
-      let(:integration) { different_type_inherited_integration }
-
-      it 'does not update the integration' do
-        expect { described_class.propagate(instance_integration) }
-          .not_to change { instance_integration.attributes.except(*excluded_attributes) }
       end
     end
 

spec/services/bulk_update_integration_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkUpdateIntegrationService do
+  include JiraServiceHelper
+
+  before do
+    stub_jira_service_test
+  end
+
+  let(:excluded_attributes) { %w[id project_id group_id inherit_from_id instance template created_at updated_at] }
+  let!(:instance_integration) do
+    JiraService.create!(
+      instance: true,
+      active: true,
+      push_events: true,
+      url: 'http://update-jira.instance.com',
+      username: 'user',
+      password: 'secret'
+    )
+  end
+
+  let!(:integration) do
+    JiraService.create!(
+      project: create(:project),
+      inherit_from_id: instance_integration.id,
+      instance: false,
+      active: true,
+      push_events: false,
+      url: 'http://jira.instance.com',
+      username: 'user',
+      password: 'secret'
+    )
+  end
+
+  context 'with inherited integration' do
+    it 'updates the integration' do
+      described_class.new(instance_integration, Service.inherit_from_id(instance_integration.id)).execute
+
+      expect(integration.reload.inherit_from_id).to eq(instance_integration.id)
+      expect(integration.attributes.except(*excluded_attributes))
+        .to eq(instance_integration.attributes.except(*excluded_attributes))
+    end
+
+    context 'with integration with data fields' do
+      let(:excluded_attributes) { %w[id service_id created_at updated_at] }
+
+      it 'updates the data fields from the integration' do
+        described_class.new(instance_integration, Service.inherit_from_id(instance_integration.id)).execute
+
+        expect(integration.reload.data_fields.attributes.except(*excluded_attributes))
+          .to eq(instance_integration.data_fields.attributes.except(*excluded_attributes))
+      end
+    end
+  end
+end

spec/workers/propagate_integration_inherit_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe PropagateIntegrationInheritWorker do
+  describe '#perform' do
+    let(:integration) { create(:redmine_service, :instance) }
+    let!(:integration1) { create(:redmine_service, inherit_from_id: integration.id) }
+    let!(:integration2) { create(:bugzilla_service, inherit_from_id: integration.id) }
+    let!(:integration3) { create(:redmine_service) }
+
+    it 'calls to BulkCreateIntegrationService' do
+      expect(BulkUpdateIntegrationService).to receive(:new)
+        .with(integration, match_array(integration1))
+        .and_return(double(execute: nil))
+
+      subject.perform(integration.id, integration1.id, integration3.id)
+    end
+  end
+end