feat: allow limit to be applied to clean task

lib/pact_broker/api/resources/clean.rb

@@ -0,0 +1,36 @@
+require 'pact_broker/api/resources/base_resource'
+require 'pact_broker/db/clean'
+require 'pact_broker/matrix/unresolved_selector'
+
+module PactBroker
+  module Api
+    module Resources
+      class Clean < BaseResource
+        def content_types_accepted
+          [["application/json"]]
+        end
+
+        def content_types_provided
+          [["application/hal+json"]]
+        end
+
+        def allowed_methods
+          ["POST", "OPTIONS"]
+        end
+
+        def process_post
+          keep_selectors = (params[:keep] || []).collect do | hash |
+            PactBroker::Matrix::UnresolvedSelector.new(hash)
+          end
+
+          result = PactBroker::DB::Clean.call(Sequel::Model.db, { keep: keep_selectors })
+          response.body = result.to_json
+        end
+
+        def policy_name
+          :'integrations::clean'
+        end
+      end
+    end
+  end
+end

lib/pact_broker/db/clean.rb

@@ -1,10 +1,13 @@
 require 'sequel'
 require 'pact_broker/project_root'
 require 'pact_broker/pacts/latest_tagged_pact_publications'
+require 'pact_broker/logging'
 
 module PactBroker
   module DB
     class Clean
+      include PactBroker::Logging
+
 
       class Unionable < Array
         alias_method :union, :+
@@ -46,7 +49,7 @@ def pact_publication_ids_to_keep_for_version_ids_to_keep
       def latest_tagged_pact_publications_ids_to_keep
         @latest_tagged_pact_publications_ids_to_keep ||= resolve_ids(keep.select(&:tag).select(&:latest).collect do | selector |
           PactBroker::Pacts::LatestTaggedPactPublications.select(:id).for_selector(selector)
-        end.reduce(&:union))
+        end.reduce(&:union) || [])
       end
 
 

lib/pact_broker/db/clean_incremental.rb

@@ -0,0 +1,78 @@
+require 'pact_broker/logging'
+require 'pact_broker/matrix/unresolved_selector'
+
+module PactBroker
+  module DB
+    class CleanIncremental
+      include PactBroker::Logging
+
+      DEFAULT_KEEP_SELECTORS = [
+        PactBroker::Matrix::UnresolvedSelector.new(tag: true, latest: true),
+        PactBroker::Matrix::UnresolvedSelector.new(latest: true),
+        PactBroker::Matrix::UnresolvedSelector.new(max_age: 90)
+      ]
+      TABLES = [:versions, :pact_publications, :pact_versions, :verifications, :triggered_webhooks, :webhook_executions]
+
+      def self.call database_connection, options = {}
+        new(database_connection, options).call
+      end
+
+      def initialize database_connection, options = {}
+        @db = database_connection
+        @options = options
+      end
+
+      def keep
+        options[:keep] || DEFAULT_KEEP_SELECTORS
+      end
+
+      def limit
+        options[:limit] || 1000
+      end
+
+      def resolve_ids(query, column_name = :id)
+        query.collect { |h| h[column_name] }
+      end
+
+      def version_ids_to_delete
+        db[:versions].where(id: version_ids_to_keep).invert.limit(limit).order(Sequel.asc(:id))
+      end
+
+      def version_ids_to_keep
+        @version_ids_to_keep ||= keep.collect do | selector |
+          PactBroker::Domain::Version.select(:id).for_selector(selector)
+        end.reduce(&:union)
+      end
+
+      def call
+        require 'pact_broker/domain/version'
+        before_counts = current_counts
+
+        result = PactBroker::Domain::Version.where(id: resolve_ids(version_ids_to_delete)).delete
+        delete_orphan_pact_versions
+
+        after_counts = current_counts
+
+        TABLES.each_with_object({}) do | table_name, comparison_counts |
+          comparison_counts[table_name.to_s] = { "deleted" => before_counts[table_name] - after_counts[table_name], "kept" => after_counts[table_name] }
+        end
+      end
+
+      private
+
+      attr_reader :db, :options
+
+      def current_counts
+        TABLES.each_with_object({}) do | table_name, counts |
+          counts[table_name] = db[table_name].count
+        end
+      end
+
+      def delete_orphan_pact_versions
+        logger.info("Deleting orphan pact versions")
+        referenced_pact_version_ids = db[:pact_publications].select(:pact_version_id).union(db[:verifications].select(:pact_version_id))
+        db[:pact_versions].where(id: referenced_pact_version_ids).invert.delete
+      end
+    end
+  end
+end

lib/pact_broker/domain/verification.rb

@@ -29,6 +29,12 @@ def before_create
         # Beware that when columns with the same name exist in both datasets
         # you may get the wrong column back in your model.
 
+        def delete
+          require 'pact_broker/webhooks/triggered_webhook'
+          PactBroker::Webhooks::TriggeredWebhook.where(verification: self).delete
+          super
+        end
+
         def consumer consumer_name
           where(name_like(:consumer_name, consumer_name))
         end

lib/pact_broker/domain/version.rb

@@ -79,6 +79,11 @@ def where_age_less_than(days)
         end
 
         def delete
+          require 'pact_broker/pacts/pact_publication'
+          require 'pact_broker/domain/verification'
+          require 'pact_broker/domain/tag'
+
+          PactBroker::Domain::Verification.where(provider_version: self).delete
           PactBroker::Pacts::PactPublication.where(consumer_version: self).delete
           PactBroker::Domain::Tag.where(version: self).delete
           super

lib/pact_broker/matrix/unresolved_selector.rb

@@ -1,10 +1,18 @@
+require 'pact_broker/hash_refinements'
+
 module PactBroker
   module Matrix
     class UnresolvedSelector < Hash
+      using PactBroker::HashRefinements
+
       def initialize(params = {})
         merge!(params)
       end
 
+      def self.from_hash(hash)
+        new(hash.symbolize_keys.snakecase_keys.slice(:pacticipant_name, :pacticipant_version_number, :latest, :tag, :max_age))
+      end
+
       def pacticipant_name
         self[:pacticipant_name]
       end

lib/pact_broker/pacts/pact_publication.rb

@@ -101,6 +101,12 @@ def where_consumer_if_set(consumer)
             self
           end
         end
+
+        def delete
+          require 'pact_broker/webhooks/triggered_webhook'
+          PactBroker::Webhooks::TriggeredWebhook.where(pact_publication: self).delete
+          super
+        end
       end
 
       def before_create

lib/pact_broker/tasks/clean_task.rb

@@ -3,12 +3,23 @@ module DB
     class CleanTask < ::Rake::TaskLib
 
       attr_accessor :database_connection
-      attr_accessor :keep
+      attr_reader :keep
+      attr_accessor :limit
 
       def initialize &block
+        require 'pact_broker/db/clean_incremental'
+        @limit = 1000
+        @keep = PactBroker::DB::CleanIncremental::DEFAULT_KEEP_SELECTORS
         rake_task &block
       end
 
+      def keep=(keep)
+        require 'pact_broker/matrix/unresolved_selector'
+        @keep = [*keep].collect do | hash |
+          PactBroker::Matrix::UnresolvedSelector.from_hash(hash)
+        end
+      end
+
       def rake_task &block
         namespace :pact_broker do
           namespace :db do
@@ -17,23 +28,28 @@ def rake_task &block
 
               instance_eval(&block)
 
-              require 'pact_broker/db/clean'
-              require 'pact_broker/matrix/unresolved_selector'
+              require 'pact_broker/db/clean_incremental'
+              require 'pact_broker/error'
               require 'yaml'
+              require 'benchmark'
 
-              keep_selectors = nil
-              if keep
-                keep_selectors = [*keep].collect do | hash |
-                  PactBroker::Matrix::UnresolvedSelector.new(hash)
-                end
+              raise PactBroker::Error.new("You must specify a limit for the number of versions to delete") unless limit
+
+              if keep.nil? || keep.empty?
+                raise PactBroker::Error.new("You must specify which versions to keep")
+              else
+                puts "Deleting oldest #{limit} versions, keeping versions that match the following selectors: #{keep}..."
               end
-              # TODO time it
-              results = PactBroker::DB::Clean.call(database_connection, keep: keep_selectors)
-              puts results.to_yaml
+
+              start_time = Time.now
+              results = PactBroker::DB::CleanIncremental.call(database_connection, keep: keep, limit: limit)
+              end_time = Time.now
+              elapsed_seconds = (end_time - start_time).to_i
+              puts results.to_yaml.gsub("---", "\nResults (#{elapsed_seconds} seconds)\n-------")
             end
           end
         end
       end
     end
   end
-end
+end

lib/pact_broker/webhooks/triggered_webhook.rb

@@ -21,6 +21,12 @@ class TriggeredWebhook < Sequel::Model(:triggered_webhooks)
       dataset_module do
         include PactBroker::Repositories::Helpers
 
+        def delete
+          require 'pact_broker/webhooks/execution'
+          PactBroker::Webhooks::Execution.where(triggered_webhook: self).delete
+          super
+        end
+
         def retrying
           where(status: STATUS_RETRYING)
         end

script/docker/db-start.sh

@@ -4,4 +4,4 @@ docker run --name pact-broker-postgres \
   -e POSTGRES_PASSWORD=postgres \
   -p 5432:5432 \
   -v $PWD:/data \
-  -d postgres:10
+  -d postgres:12

spec/fixtures/approvals/modifiable_resources.approved.json

@@ -3,6 +3,9 @@
     {
       "resource_class_name": "PactBroker::Api::Resources::AllWebhooks"
     },
+    {
+      "resource_class_name": "PactBroker::Api::Resources::Clean"
+    },
     {
       "resource_class_name": "PactBroker::Api::Resources::ErrorTest"
     },

spec/lib/pact_broker/db/clean_incremental_spec.rb

@@ -0,0 +1,93 @@
+require 'pact_broker/db/clean_incremental'
+require 'pact_broker/matrix/unresolved_selector'
+
+IS_MYSQL = !!DB.mysql?
+
+module PactBroker
+  module DB
+    # Inner queries don't work on MySQL. Seriously, MySQL???
+    describe CleanIncremental, pending: IS_MYSQL  do
+
+      def pact_publication_count_for(consumer_name, version_number)
+        PactBroker::Pacts::PactPublication.where(consumer_version: PactBroker::Domain::Version.where_pacticipant_name(consumer_name).where(number: version_number)).count
+      end
+
+      let(:options) { {} }
+      let(:db) { PactBroker::DB.connection }
+
+      subject { CleanIncremental.call(PactBroker::DB.connection, options) }
+      let(:latest_dev_selector) { PactBroker::Matrix::UnresolvedSelector.new(tag: "dev", latest: true) }
+      let(:all_prod_selector) { PactBroker::Matrix::UnresolvedSelector.new(tag: "prod") }
+      let(:limit) { 3 }
+
+      describe ".call"do
+        context "when there are specified versions to keep" do
+          before do
+            td.create_pact_with_hierarchy("Foo", "1", "Bar")
+              .create_consumer_version_tag("prod").comment("keep as one of prod")
+              .create_consumer_version_tag("dev")
+              .add_day
+              .create_consumer_version("2").comment("DELETE")
+              .add_day
+              .create_consumer_version("3", tag_names: %w{prod}).comment("keep as one of prod")
+              .create_pact
+              .add_day
+              .create_consumer_version("4", tag_names: %w{dev}).comment("DELETE as not latest")
+              .create_pact
+              .add_day
+              .create_consumer_version("5", tag_names: %w{dev}).comment("keep as latest dev")
+              .create_pact
+              .add_day
+              .create_consumer_version("6", tag_names: %w{foo}).comment("DELETE as not specified")
+              .create_pact
+              .add_day
+              .create_consumer_version("7").comment("keep as deletion limit is 3")
+              .create_pact
+          end
+
+          let(:options) { { keep: [all_prod_selector, latest_dev_selector], limit: limit } }
+
+          it "does not delete the consumer versions specified" do
+            expect(PactBroker::Domain::Version.where(number: "1").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "2").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "3").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "4").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "5").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "6").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "7").count).to be 1
+            subject
+            expect(PactBroker::Domain::Version.where(number: "1").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "2").count).to be 0
+            expect(PactBroker::Domain::Version.where(number: "3").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "4").count).to be 0
+            expect(PactBroker::Domain::Version.where(number: "5").count).to be 1
+            expect(PactBroker::Domain::Version.where(number: "6").count).to be 0
+            expect(PactBroker::Domain::Version.where(number: "7").count).to be 1
+          end
+        end
+
+        context "with orphan pact versions" do
+          before do
+            # Create a pact that will not be deleted
+            td.create_pact_with_hierarchy("Foo", "0", "Bar", json_content_1)
+              .create_consumer_version_tag("dev")
+            # Create an orphan pact version
+            pact_version_params = PactBroker::Pacts::PactVersion.first.to_hash
+            pact_version_params.delete(:id)
+            pact_version_params[:sha] = "1234"
+            PactBroker::Pacts::PactVersion.create(pact_version_params)
+          end
+
+          let(:json_content_1) { { interactions: ['a', 'b']}.to_json }
+          let(:json_content_2) { { interactions: ['a', 'c']}.to_json }
+
+          let(:options) { { keep: [latest_dev_selector] } }
+
+          it "deletes them" do
+            expect { subject }.to change { PactBroker::Pacts::PactVersion.count }.by(-1)
+          end
+        end
+      end
+    end
+  end
+end