fix: fix performance issues due to contention in the integrations tab…

…le when publishing a large number of contracts (> 20) per request, in parallel (#654)

PACT-1352

lib/pact_broker/api.rb

@@ -5,7 +5,6 @@
 require "pact_broker/api/contracts"
 require "pact_broker/application_context"
 require "pact_broker/feature_toggle"
-require "pact_broker/initializers/subscriptions"
 
 module Webmachine
   class Request

lib/pact_broker/api/resources/event_methods.rb

@@ -0,0 +1,15 @@
+require "pact_broker/events/subscriber"
+
+module PactBroker
+  module Api
+    module Resources
+      module EventMethods
+        def subscribe(listener)
+          PactBroker::Events.subscribe(listener) do
+            yield
+          end
+        end
+      end
+    end
+  end
+end

lib/pact_broker/api/resources/pact.rb

@@ -9,14 +9,17 @@
 require "pact_broker/webhooks/execution_configuration"
 require "pact_broker/api/resources/webhook_execution_methods"
 require "pact_broker/api/resources/pact_resource_methods"
+require "pact_broker/api/resources/event_methods"
+require "pact_broker/integrations/event_listener"
 
 module PactBroker
   module Api
     module Resources
       class Pact < BaseResource
+        include EventMethods
         include PacticipantResourceMethods
-        include WebhookExecutionMethods
         include PactResourceMethods
+        include WebhookExecutionMethods
         include PactBroker::Messages
 
         def content_types_provided
@@ -65,11 +68,13 @@ def resource_exists?
         def from_json
           response_code = pact ? 200 : 201
 
-          handle_webhook_events do
-            if request.patch? && resource_exists?
-              @pact = pact_service.merge_pact(pact_params)
-            else
-              @pact = pact_service.create_or_update_pact(pact_params)
+          subscribe(PactBroker::Integrations::EventListener.new) do
+            handle_webhook_events do
+              if request.patch? && resource_exists?
+                @pact = pact_service.merge_pact(pact_params)
+              else
+                @pact = pact_service.create_or_update_pact(pact_params)
+              end
             end
           end
           response.body = to_json

lib/pact_broker/api/resources/verifications.rb

@@ -5,13 +5,16 @@
 require "pact_broker/api/decorators/verification_decorator"
 require "pact_broker/api/resources/webhook_execution_methods"
 require "pact_broker/api/resources/metadata_resource_methods"
+require "pact_broker/api/resources/event_methods"
+require "pact_broker/integrations/event_listener"
 
 module PactBroker
   module Api
     module Resources
       class Verifications < BaseResource
         include WebhookExecutionMethods
         include MetadataResourceMethods
+        include EventMethods
 
         def content_types_accepted
           [["application/json", :from_json]]
@@ -42,10 +45,12 @@ def create_path
         end
 
         def from_json
-          handle_webhook_events(build_url: verification_params["buildUrl"]) do
-            verified_pacts = pact_service.find_for_verification_publication(pact_params, event_context[:consumer_version_selectors])
-            verification = verification_service.create(next_verification_number, verification_params, verified_pacts, event_context)
-            response.body = decorator_for(verification).to_json(**decorator_options)
+          subscribe(PactBroker::Integrations::EventListener.new) do
+            handle_webhook_events(build_url: verification_params["buildUrl"]) do
+              verified_pacts = pact_service.find_for_verification_publication(pact_params, event_context[:consumer_version_selectors])
+              verification = verification_service.create(next_verification_number, verification_params, verified_pacts, event_context)
+              response.body = decorator_for(verification).to_json(**decorator_options)
+            end
           end
           true
         end

lib/pact_broker/contracts/service.rb

@@ -35,6 +35,7 @@ def publish(parsed_contracts, base_url: )
         version, version_notices = create_version(parsed_contracts)
         tags = create_tags(parsed_contracts, version)
         pacts, pact_notices = create_pacts(parsed_contracts, base_url)
+        update_integrations(pacts)
         notices = version_notices + pact_notices
         ContractsPublicationResults.from_hash(
           pacticipant: version.pacticipant,
@@ -303,6 +304,10 @@ def url_for_triggered_webhook(triggered_webhook, base_url)
         PactBroker::Api::PactBrokerUrls.triggered_webhook_logs_url(triggered_webhook, base_url)
       end
 
+      def update_integrations(pacts)
+        integration_service.handle_bulk_contract_data_published(pacts)
+      end
+
       private :url_for_triggered_webhook
     end
   end

lib/pact_broker/integrations/repository.rb

@@ -21,7 +21,8 @@ def create_for_pact(consumer_id, provider_id)
           Integration.new(
             consumer_id: consumer_id,
             provider_id: provider_id,
-            created_at: Sequel.datetime_class.now
+            created_at: Sequel.datetime_class.now,
+            contract_data_updated_at: Sequel.datetime_class.now
           ).insert_ignore
         end
         nil
@@ -39,6 +40,16 @@ def set_contract_data_updated_at(consumer, provider)
           .where({ consumer_id: consumer&.id, provider_id: provider.id }.compact )
           .update(contract_data_updated_at: Sequel.datetime_class.now)
       end
+
+
+      # Sets the contract_data_updated_at for the integrations as specified by an array of objects which each have a consumer and provider
+      # @param [Array<Object>] where each object has a consumer and a provider
+      def set_contract_data_updated_at_for_multiple_integrations(objects_with_consumer_and_provider)
+        consumer_and_provider_ids = objects_with_consumer_and_provider.collect{ | object | [object.consumer.id, object.provider.id] }.uniq
+        Integration
+          .where([:consumer_id, :provider_id] => consumer_and_provider_ids)
+          .update(contract_data_updated_at: Sequel.datetime_class.now)
+      end
     end
   end
 end

lib/pact_broker/integrations/service.rb

@@ -24,6 +24,13 @@ def self.handle_contract_data_published(consumer, provider)
         integration_repository.set_contract_data_updated_at(consumer, provider)
       end
 
+
+      # Callback to invoke when a batch of contract data is published (eg. the publish contracts endpoint)
+      # @param [Array<Object>] where each object has a consumer and a provider
+      def self.handle_bulk_contract_data_published(objects_with_consumer_and_provider)
+        integration_repository.set_contract_data_updated_at_for_multiple_integrations(objects_with_consumer_and_provider)
+      end
+
       def self.delete(consumer_name, provider_name)
         consumer = pacticipant_service.find_pacticipant_by_name!(consumer_name)
         provider = pacticipant_service.find_pacticipant_by_name!(provider_name)

spec/lib/pact_broker/integrations/repository_spec.rb

@@ -22,10 +22,11 @@ module Integrations
             td.create_verification(provider_version: "2")
           end
 
-          # No contract data date
+          # Nil contract data date
           td.create_consumer("Dog")
             .create_provider("Cat")
             .create_integration
+          Integration.order(:id).last.update(contract_data_updated_at: nil)
         end
 
         subject { Repository.new.find }
@@ -93,6 +94,46 @@ module Integrations
           end
         end
       end
+
+      describe "#set_contract_data_updated_at_for_multiple_integrations" do
+        before do
+          Timecop.freeze(date_1) do
+            td.create_consumer("Foo1")
+              .create_provider("Bar1")
+              .create_integration
+              .create_consumer("Foo2")
+              .create_provider("Bar2")
+              .create_integration
+              .create_consumer("Foo3")
+              .create_provider("Bar3")
+              .create_integration
+          end
+        end
+
+        let(:date_1) { Time.new(2023, 1, 1).utc.to_datetime }
+        let(:date_2) { Time.new(2023, 1, 2).utc.to_datetime }
+
+        let(:objects_with_consumer_and_provider) do
+          [
+            OpenStruct.new(consumer: td.find_pacticipant("Foo1"), provider: td.find_pacticipant("Bar1")),
+            OpenStruct.new(consumer: td.find_pacticipant("Foo2"), provider: td.find_pacticipant("Bar2"))
+          ]
+        end
+
+        subject do
+          Timecop.freeze(date_2) do
+            Repository.new.set_contract_data_updated_at_for_multiple_integrations(objects_with_consumer_and_provider)
+          end
+        end
+
+        it "sets the contract_data_updated_at of the specified integrations" do
+          subject
+          integrations = Integration.order(:id).all
+          expect(integrations[0].contract_data_updated_at).to be_date_time(date_2)
+          expect(integrations[1].contract_data_updated_at).to be_date_time(date_2)
+          expect(integrations[2].contract_data_updated_at).to be_date_time(date_1)
+        end
+      end
     end
   end
 end