refactor: use pub/sub to trigger webhook events

lib/pact_broker/api/resources/pact.rb

@@ -58,12 +58,13 @@ def resource_exists?
         def from_json
           response_code = pact ? 200 : 201
 
-          if request.patch? && resource_exists?
-            @pact = pact_service.merge_pact(pact_params, webhook_options)
-          else
-            @pact = pact_service.create_or_update_pact(pact_params, webhook_options)
+          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
-
           response.body = to_json
           response_code
         end
@@ -111,13 +112,6 @@ def pact
         def pact_params
           @pact_params ||= PactBroker::Pacts::PactParams.from_request request, path_info
         end
-
-        def webhook_options
-          {
-            database_connector: database_connector,
-            webhook_execution_configuration: webhook_execution_configuration
-          }
-        end
       end
     end
   end

lib/pact_broker/api/resources/verifications.rb

@@ -5,6 +5,7 @@
 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/webhooks/event_listener'
 
 module PactBroker
   module Api
@@ -50,8 +51,10 @@ def create_path
         end
 
         def from_json
-          verification = verification_service.create(next_verification_number, verification_params, pact, event_context, webhook_options)
-          response.body = decorator_for(verification).to_json(decorator_options)
+          handle_webhook_events do
+            verification = verification_service.create(next_verification_number, verification_params, pact, event_context)
+            response.body = decorator_for(verification).to_json(decorator_options)
+          end
           true
         end
 
@@ -85,13 +88,6 @@ def event_context
           metadata
         end
 
-        def webhook_options
-          {
-            database_connector: database_connector,
-            webhook_execution_configuration: webhook_execution_configuration
-          }
-        end
-
         def verification_params
           params(symbolize_names: false).merge('wip' => wip?)
         end

lib/pact_broker/api/resources/webhook_execution.rb

@@ -26,7 +26,7 @@ def allowed_methods
         end
 
         def process_post
-          webhook_execution_result = webhook_service.test_execution(webhook, webhook_execution_configuration.webhook_context, webhook_execution_configuration)
+          webhook_execution_result = webhook_trigger_service.test_execution(webhook, webhook_execution_configuration.webhook_context, webhook_execution_configuration)
           response.headers['Content-Type'] = 'application/hal+json;charset=utf-8'
           response.body = post_response_body(webhook_execution_result)
           true

lib/pact_broker/api/resources/webhook_execution_methods.rb

@@ -1,10 +1,40 @@
+require 'pact_broker/webhooks/event_listener'
+
 module PactBroker
   module Api
     module Resources
       module WebhookExecutionMethods
         def webhook_execution_configuration
           application_context.webhook_execution_configuration_creator.call(self)
         end
+
+        def webhook_options
+          {
+            database_connector: database_connector,
+            webhook_execution_configuration: webhook_execution_configuration
+          }
+        end
+
+        def webhook_event_listener
+          @webhook_event_listener ||= PactBroker::Webhooks::EventListener.new(webhook_options)
+        end
+
+        def handle_webhook_events
+          Wisper.subscribe(webhook_event_listener) do
+            yield
+          end
+        end
+
+        def schedule_triggered_webhooks
+          webhook_event_listener.schedule_triggered_webhooks
+        end
+
+        def finish_request
+          if response.code < 400
+            schedule_triggered_webhooks
+          end
+          super
+        end
       end
     end
   end

lib/pact_broker/events/event.rb

@@ -0,0 +1,5 @@
+module PactBroker
+  module Events
+    Event = Struct.new(:name, :comment, :triggered_webhooks)
+  end
+end

lib/pact_broker/pacts/service.rb

@@ -4,12 +4,14 @@
 require 'pact_broker/pacts/merger'
 require 'pact_broker/pacts/verifiable_pact'
 require 'pact_broker/pacts/squash_pacts_for_verification'
+require 'wisper'
 
 module PactBroker
   module Pacts
     module Service
 
       extend self
+      extend Wisper::Publisher
 
       extend PactBroker::Repositories
       extend PactBroker::Services
@@ -43,20 +45,20 @@ def delete params
         pact_repository.delete(params)
       end
 
-      def create_or_update_pact params, webhook_options
+      def create_or_update_pact params
         provider = pacticipant_repository.find_by_name_or_create params[:provider_name]
         consumer = pacticipant_repository.find_by_name_or_create params[:consumer_name]
         consumer_version = version_repository.find_by_pacticipant_id_and_number_or_create consumer.id, params[:consumer_version_number]
         existing_pact = pact_repository.find_by_version_and_provider(consumer_version.id, provider.id)
 
         if existing_pact
-          update_pact params, existing_pact, webhook_options
+          update_pact params, existing_pact
         else
-          create_pact params, consumer_version, provider, webhook_options
+          create_pact params, consumer_version, provider
         end
       end
 
-      def merge_pact params, webhook_options
+      def merge_pact params
         provider = pacticipant_repository.find_by_name_or_create params[:provider_name]
         consumer = pacticipant_repository.find_by_name_or_create params[:consumer_name]
         consumer_version = version_repository.find_by_pacticipant_id_and_number_or_create consumer.id, params[:consumer_version_number]
@@ -66,7 +68,7 @@ def merge_pact params, webhook_options
           existing_pact.json_content, params[:json_content]
         )
 
-        update_pact params, existing_pact, webhook_options
+        update_pact params, existing_pact
       end
 
       def find_all_pact_versions_between consumer, options
@@ -129,8 +131,6 @@ def find_for_verification(provider_name, provider_version_branch, provider_versi
         verifiable_pacts_specified_in_request + verifiable_wip_pacts
       end
 
-      private
-
       def exclude_specified_pacts(wip_pacts, specified_pacts)
         wip_pacts.reject do | wip_pact |
           specified_pacts.any? do | specified_pacts |
@@ -139,8 +139,10 @@ def exclude_specified_pacts(wip_pacts, specified_pacts)
         end
       end
 
+      private :exclude_specified_pacts
+
       # Overwriting an existing pact with the same consumer/provider/consumer version number
-      def update_pact params, existing_pact, webhook_options
+      def update_pact params, existing_pact
         logger.info "Updating existing pact publication with params #{params.reject{ |k, v| k == :json_content}}"
         logger.debug "Content #{params[:json_content]}"
         pact_version_sha = generate_sha(params[:json_content])
@@ -149,13 +151,23 @@ def update_pact params, existing_pact, webhook_options
         updated_pact = pact_repository.update(existing_pact.id, update_params)
 
         event_context = { consumer_version_tags: updated_pact.consumer_version_tag_names }
-        webhook_trigger_service.trigger_webhooks_for_updated_pact(existing_pact, updated_pact, event_context, merge_consumer_version_info(webhook_options, updated_pact))
+        event_params = { event_context: event_context, pact: updated_pact }
+
+        broadcast(:contract_published, event_params)
+
+        if existing_pact.pact_version_sha != updated_pact.pact_version_sha
+          broadcast(:contract_content_changed, event_params.merge(event_comment: "Pact content modified since previous revision"))
+        else
+          broadcast(:contract_content_unchanged, event_params.merge(event_comment: "Pact content was unchanged"))
+        end
 
         updated_pact
       end
 
+      private :update_pact
+
       # When no publication for the given consumer/provider/consumer version number exists
-      def create_pact params, version, provider, webhook_options
+      def create_pact params, version, provider
         logger.info "Creating new pact publication with params #{params.reject{ |k, v| k == :json_content}}"
         logger.debug "Content #{params[:json_content]}"
         pact_version_sha = generate_sha(params[:json_content])
@@ -167,24 +179,72 @@ def create_pact params, version, provider, webhook_options
           pact_version_sha: pact_version_sha,
           json_content: json_content
         )
-        event_context = { consumer_version_tags: pact.consumer_version_tag_names }
-        webhook_trigger_service.trigger_webhooks_for_new_pact(pact, event_context, merge_consumer_version_info(webhook_options, pact))
+
+        event_params = { event_context: { consumer_version_tags: pact.consumer_version_tag_names }, pact: pact }
+        broadcast(:contract_published, event_params)
+
+        content_changed, explanation = pact_is_new_or_newly_tagged_or_pact_has_changed_since_previous_version?(pact)
+        if content_changed
+          broadcast(:contract_content_changed, event_params.merge(event_comment: explanation))
+        else
+          broadcast(:contract_content_unchanged, event_params.merge(event_comment: "Pact content the same as previous version and no new tags were applied"))
+        end
+
         pact
       end
 
+      private :create_pact
+
       def generate_sha(json_content)
         PactBroker.configuration.sha_generator.call(json_content)
       end
 
+      private :generate_sha
+
       def add_interaction_ids(json_content)
         Content.from_json(json_content).with_ids.to_json
       end
 
-      def merge_consumer_version_info(webhook_options, pact)
-        execution_configuration = webhook_options[:webhook_execution_configuration]
-                                    .with_webhook_context(consumer_version_tags: pact.consumer_version_tag_names)
-        webhook_options.merge(webhook_execution_configuration: execution_configuration)
+      private :add_interaction_ids
+
+      def pact_is_new_or_newly_tagged_or_pact_has_changed_since_previous_version? pact
+        changed_pacts = pact_repository
+          .find_previous_pacts(pact)
+          .reject { |_, previous_pact| !sha_changed_or_no_previous_version?(previous_pact, pact) }
+        explanation = explanation_for_content_changed(changed_pacts)
+        return changed_pacts.any?, explanation
+      end
+
+      private :pact_is_new_or_newly_tagged_or_pact_has_changed_since_previous_version?
+
+      def sha_changed_or_no_previous_version?(previous_pact, new_pact)
+        previous_pact.nil? || new_pact.pact_version_sha != previous_pact.pact_version_sha
+      end
+
+      private :sha_changed_or_no_previous_version?
+
+      def explanation_for_content_changed(changed_pacts)
+        if changed_pacts.any?
+          messages = changed_pacts.collect do |tag, previous_pact|
+            if tag == :untagged
+              if previous_pact
+                "Pact content has changed since previous untagged version"
+              else
+                "First time untagged pact published"
+              end
+            else
+              if previous_pact
+                "Pact content has changed since the last consumer version tagged with #{tag}"
+              else
+                "First time pact published with consumer version tagged #{tag}"
+              end
+            end
+          end
+          messages.join(',')
+        end
       end
+
+      private :explanation_for_content_changed
     end
   end
 end

lib/pact_broker/test/test_data_builder.rb

@@ -4,6 +4,7 @@
 require 'pact_broker/services'
 require 'pact_broker/webhooks/repository'
 require 'pact_broker/webhooks/service'
+require 'pact_broker/webhooks/trigger_service'
 require 'pact_broker/webhooks/webhook_execution_result'
 require 'pact_broker/pacts/repository'
 require 'pact_broker/pacts/service'
@@ -339,7 +340,7 @@ def create_triggered_webhook params = {}
         event_name = params.key?(:event_name) ? params[:event_name] : @webhook.events.first.name # could be nil, for backwards compatibility
         verification = @webhook.trigger_on_provider_verification_published? ? @verification : nil
         event_context = params[:event_context]
-        @triggered_webhook = webhook_repository.create_triggered_webhook(trigger_uuid, @webhook, @pact, verification, PactBroker::Webhooks::Service::RESOURCE_CREATION, event_name, event_context)
+        @triggered_webhook = webhook_repository.create_triggered_webhook(trigger_uuid, @webhook, @pact, verification, PactBroker::Webhooks::TriggerService::RESOURCE_CREATION, event_name, event_context)
         @triggered_webhook.update(status: params[:status]) if params[:status]
         set_created_at_if_set params[:created_at], :triggered_webhooks, { id: @triggered_webhook.id }
         self

lib/pact_broker/verifications/service.rb

@@ -3,6 +3,7 @@
 require 'pact_broker/verifications/summary_for_consumer_version'
 require 'pact_broker/logging'
 require 'pact_broker/hash_refinements'
+require 'wisper'
 
 module PactBroker
 
@@ -15,12 +16,13 @@ module Service
       extend PactBroker::Services
       include PactBroker::Logging
       using PactBroker::HashRefinements
+      extend Wisper::Publisher
 
       def next_number
         verification_repository.next_number
       end
 
-      def create next_verification_number, params, pact, event_context, webhook_options
+      def create next_verification_number, params, pact, event_context
         logger.info "Creating verification #{next_verification_number} for pact_id=#{pact.id}", payload: params.reject{ |k,_| k == "testResults"}
         verification = PactBroker::Domain::Verification.new
         provider_version_number = params.fetch('providerApplicationVersion')
@@ -29,15 +31,8 @@ def create next_verification_number, params, pact, event_context, webhook_option
         verification.number = next_verification_number
         verification = verification_repository.create(verification, provider_version_number, pact)
 
-        execution_configuration = webhook_options[:webhook_execution_configuration]
-                                    .with_webhook_context(provider_version_tags: verification.provider_version_tag_names)
+        broadcast_events(verification, pact, event_context)
 
-        webhook_trigger_service.trigger_webhooks_for_verification_results_publication(
-          pact,
-          verification,
-          event_context.merge(provider_version_tags: verification.provider_version_tag_names),
-          webhook_options.deep_merge(webhook_execution_configuration: execution_configuration)
-        )
         verification
       end
 
@@ -89,6 +84,23 @@ def verification_summary_for_consumer_version params
       def delete_all_verifications_between(consumer_name, options)
         verification_repository.delete_all_verifications_between(consumer_name, options)
       end
+
+      private
+
+      def broadcast_events(verification, pact, event_context)
+        event_params = {
+          pact: pact,
+          verification: verification,
+          event_context: event_context.merge(provider_version_tags: verification.provider_version_tag_names)
+        }
+
+        broadcast(:provider_verification_published, event_params)
+        if verification.success
+          broadcast(:provider_verification_succeeded, event_params)
+        else
+          broadcast(:provider_verification_failed, event_params)
+        end
+      end
     end
   end
 end