Rework gbif cleanup

app/jobs/delete_gbif_events_job.rb

@@ -3,11 +3,18 @@
 class DeleteGbifEventsJob < ApplicationJob
   queue_as :lupo_background
 
-  def perform(id, options = {})
-    event = Event.find_by(uuid: id)
+  def perform(ids, options = {})
+    label = options[:label]
+    index = options[:index]
 
-    event.destroy! if event.present?
+    # delete event records from mysql
+    sql = ActiveRecord::Base.sanitize_sql_array(["DELETE FROM events WHERE id IN (?)", ids])
+    ActiveRecord::Base.connection.execute(sql)
+
+    # delete event documents from elasticsearch
+    bulk_payload = ids.map { |id| { delete: { _index: index, _id: id } } }
+    Event.__elasticsearch__.client.bulk(body: bulk_payload)
   rescue => err
-    Rails.logger.info("#{options[:label]}: event delete error: #{err.message}")
+    Rails.logger.error("#{label}: #{err.message}")
   end
 end

app/models/event.rb

@@ -880,6 +880,41 @@ def self.loop_through_events(options)
     Rails.logger.info("#{label}: task completed")
   end
 
+  def self.loop_through_gbif_events(options)
+    size = (options[:size] || 1_000).to_i
+    cursor = options[:cursor] || []
+    filter = options[:filter] || {}
+    label = options[:label] || ""
+    job_name = options[:job_name] || ""
+    query = options[:query].presence
+    delete_count = 0
+    max_delete_count = options[:max_delete_count]
+
+    response = Event.query(query, filter.merge(page: { size: 1, cursor: [] }))
+
+    if response.size.positive?
+      while response.size.positive? && delete_count < max_delete_count
+        response = Event.query(query, filter.merge(page: { size: size, cursor: cursor }))
+
+        break unless response.size.positive?
+
+        Rails.logger.info("#{label}: #{response.size} events starting with _id #{response.results.to_a.first[:_id]}")
+
+        cursor = response.results.to_a.last[:sort]
+
+        Rails.logger.info "#{label}: cursor: #{cursor}"
+
+        ids = response.results.map(&:_id).uniq
+
+        Object.const_get(job_name).perform_later(ids, options)
+
+        delete_count += response.size
+      end
+    end
+
+    Rails.logger.info("#{label}: task completed")
+  end
+
   def metric_type
     if /(requests|investigations)/.match?(relation_type_id.to_s)
       arr = relation_type_id.split("-", 4)

lib/tasks/event.rake

@@ -87,17 +87,33 @@ end
 namespace :gbif_events do
   desc "delete gbif events"
   task delete_gbif_events: :environment do
+    index = ENV["INDEX"]
+
+    if index.blank?
+      Rails.logger.error("You must provide an INDEX environment variable")
+      exit
+    end
+
+    max_delete_count = ENV["MAX_DELETE_COUNT"]
+
+    if max_delete_count.blank?
+      Rails.logger.error("You must provide an MAX_DELETE_COUNT environment variable")
+      exit
+    end
+
     options = {
-      size: 100,
+      size: 1000,
       from_id: (ENV["FROM_ID"] || Event.minimum(:id)).to_i,
       until_id: (ENV["UNTIL_ID"] || Event.maximum(:id)).to_i,
       filter: {},
       query: "+subj.registrantId:datacite.gbif.gbif +relation_type_id:references -source_doi:(\"10.15468/QJGWBA\" OR \"10.35035/GDWQ-3V93\" OR \"10.15469/3XSWXB\" OR \"10.15469/UBP6QO\" OR \"10.35000/TEDB-QD70\" OR \"10.15469/2YMQOZ\")",
+      job_name: "DeleteGbifEventsJob",
       label: "gbif_event_cleanup_#{Time.now.utc.strftime("%d%m%Y%H%M%S")}",
-      job_name: "DeleteGbifEventsJob"
+      max_delete_count: max_delete_count.to_i,
+      index: index
     }
 
-    Event.loop_through_events(options)
+    Event.loop_through_gbif_events(options)
   end
 
   desc "delete orphaned gbif_events"