Reworking the event import worker

Author: wendelfabianchinsamy

Gemfile

@@ -24,6 +24,7 @@ gem "git", "~> 1.5"
 gem "faraday", "~> 2.9"
 gem "faraday_middleware-aws-sigv4", "~> 0.3.0"
 gem "faraday-excon"
+gem "uuid", "~> 2.3", ">= 2.3.9"
 
 # This gem will allow us to write tests without the need for a database
 gem "activerecord-nulldb-adapter", "~> 1.1", ">= 1.1.1"

Gemfile.lock

@@ -50,6 +50,11 @@ GEM
       erubi (~> 1.11)
       rails-dom-testing (~> 2.2)
       rails-html-sanitizer (~> 1.6)
+    active_model_serializers (0.10.15)
+      actionpack (>= 4.1)
+      activemodel (>= 4.1)
+      case_transform (>= 0.2)
+      jsonapi-renderer (>= 0.1.1.beta1, < 0.3)
     activejob (7.1.5.1)
       activesupport (= 7.1.5.1)
       globalid (>= 0.3.6)
@@ -107,6 +112,8 @@ GEM
     bundler-audit (0.9.2)
       bundler (>= 1.2.0, < 3)
       thor (~> 1.0)
+    case_transform (0.2)
+      activesupport
     concurrent-ruby (1.3.5)
     connection_pool (2.5.0)
     crass (1.0.6)
@@ -166,6 +173,7 @@ GEM
       reline (>= 0.4.2)
     jmespath (1.6.2)
     json (2.10.2)
+    jsonapi-renderer (0.2.2)
     language_server-protocol (3.17.0.4)
     lint_roller (1.1.0)
     logger (1.7.0)
@@ -180,6 +188,8 @@ GEM
     loofah (2.24.0)
       crass (~> 1.0.2)
       nokogiri (>= 1.12.0)
+    macaddr (1.7.2)
+      systemu (~> 2.6.5)
     mail (2.8.1)
       mini_mime (>= 0.1.1)
       net-imap
@@ -335,6 +345,7 @@ GEM
       concurrent-ruby
       thor
     stringio (3.1.5)
+    systemu (2.6.5)
     thor (1.3.2)
     timeout (0.4.3)
     tzinfo (2.0.6)
@@ -343,6 +354,8 @@ GEM
       unicode-emoji (~> 4.0, >= 4.0.4)
     unicode-emoji (4.0.4)
     uri (1.0.3)
+    uuid (2.3.9)
+      macaddr (~> 1.0)
     websocket-driver (0.7.7)
       base64
       websocket-extensions (>= 0.1.0)
@@ -355,6 +368,7 @@ PLATFORMS
   x86_64-linux
 
 DEPENDENCIES
+  active_model_serializers (~> 0.10.0)
   activerecord-nulldb-adapter (~> 1.1, >= 1.1.1)
   aws-sdk-sqs (~> 1.3)
   bootsnap
@@ -388,6 +402,7 @@ DEPENDENCIES
   sentry-rails
   sentry-ruby
   shoryuken (~> 4.0)
+  uuid (~> 2.3, >= 2.3.9)
 
 RUBY VERSION
    ruby 3.1.6p260

app/factories/event_factory.rb

@@ -0,0 +1,24 @@
+module EventFactory
+  # Creates and returns an event from the dequeued message from the events SQS queue.
+  def self.create_from_sqs(message)
+    now = Time.now.utc
+
+    Event.new(
+      uuid: message["uuid"] || SecureRandom.uuid,
+      subj_id: DoiUtilities.normalize_doi(message["subjId"]) || message["subjId"],
+      obj_id: DoiUtilities.normalize_doi(message["objId"]) || message["objId"],
+      source_id: message["sourceId"],
+      aasm_state: "waiting",
+      source_token: message["sourceToken"],
+      created_at: now,
+      updated_at: now,
+      total: message["total"] || 1,
+      occurred_at: message["occurred_at"] || now,
+      message_action: "create",
+      relation_type_id: message["relation_type_id"] || "references",
+      subj: message["subj"].to_json,
+      obj: message["obj"].to_json,
+      license: message["license"] || "https://creativecommons.org/publicdomain/zero/1.0/",
+    )
+  end
+end

app/models/concerns/relation_type_handler.rb

@@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+# Handles setting the source and target doi data of an event when an event is created or updated.
+# Method is only invoked via a before_validation callback in the events model.
+
+module RelationTypeHandler
+  extend ActiveSupport::Concern
+
+  def set_source_and_target_doi!
+    return if subj_id.blank? || obj_id.blank?
+
+    case relation_type_id
+    when *REFERENCE_RELATION_TYPES
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.source_relation_type_id = "references"
+      self.target_relation_type_id = "citations"
+    when *CITATION_RELATION_TYPES
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.source_relation_type_id = "references"
+      self.target_relation_type_id = "citations"
+    when "unique-dataset-investigations-regular"
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.target_relation_type_id = "views"
+    when "unique-dataset-requests-regular"
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.target_relation_type_id = "downloads"
+    when "has-version"
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.source_relation_type_id = "versions"
+      self.target_relation_type_id = "version_of"
+    when "is-version-of"
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.source_relation_type_id = "versions"
+      self.target_relation_type_id = "version_of"
+    when "has-part"
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.source_relation_type_id = "parts"
+      self.target_relation_type_id = "part_of"
+    when "is-part-of"
+      self.source_doi = DoiUtilities.uppercase_doi_from_url(obj_id)
+      self.target_doi = DoiUtilities.uppercase_doi_from_url(subj_id)
+      self.source_relation_type_id = "parts"
+      self.target_relation_type_id = "part_of"
+    end
+  end
+
+  REFERENCE_RELATION_TYPES = [
+    "cites",
+    "is-supplemented-by",
+    "references",
+  ].freeze
+
+  CITATION_RELATION_TYPES = [
+    "is-cited-by",
+    "is-supplement-to",
+    "is-referenced-by",
+  ].freeze
+end

app/models/event.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 class Event < ApplicationRecord
+  include RelationTypeHandler
   # include Modelable
   # include Identifiable
   # include Elasticsearch::Model
@@ -9,17 +10,16 @@ class Event < ApplicationRecord
   attribute :uuid, :text
   attribute :subj_id, :text
   attribute :obj_id, :text
-  attribute :source_id, :string
-  attribute :aasm_state, :string # could we remove this
-  attribute :state_event, :string # could we remove this
+  attribute :aasm_state, :string
+  attribute :state_event, :string
   attribute :callback, :text
   attribute :error_messages, :text
   attribute :source_token, :text
   attribute :created_at, :datetime
   attribute :updated_at, :datetime
   attribute :indexed_at, :datetime, default: -> { Time.zone.at(0) }
   attribute :occurred_at, :datetime
-  attribute :message_action, :string, default: "create" # how is this set?
+  attribute :message_action, :string, default: "create"
   attribute :subj, :text
   attribute :obj, :text
   attribute :total, :integer, default: 1
@@ -28,15 +28,19 @@ class Event < ApplicationRecord
   attribute :target_doi, :text
   attribute :source_relation_type_id, :string
   attribute :target_relation_type_id, :string
+  attribute :relation_type_id, :string
 
   # Validations
-  validates :uuid, presence: true, uniqueness: {
-    case_sensitive: false,
-    length: { maximum: 36 },
-  }
+  validates :uuid, presence: true, uuid_format: true, uniqueness: { case_sensitive: false, length: { maximum: 36 } }
   validates :subj_id, presence: true
+  validates :obj_id, presence: true
+  validates :source_id, presence: true
+  validates :source_token, presence: true
   validates :message_action, presence: true, length: { maximum: 191 }
   validates :created_at, presence: true
   validates :updated_at, presence: true
   validates :indexed_at, presence: true
+
+  # Callbacks
+  before_validation :set_source_and_target_doi!
 end

app/utilities/doi_utilities.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+module DoiUtilities
+  def self.normalize_doi(url)
+    doi =
+      Array(%r{\A(?:(http|https):/(/)?(dx\.)?(doi.org|handle.test.datacite.org)/)?(doi:)?(10\.\d{4,5}/.+)\z}
+      .match(doi)).last
+
+    doi = doi.delete("\u200B").downcase if doi.present?
+
+    "https://doi.org/#{doi}" if doi.present?
+  end
+
+  def self.uppercase_doi_from_url(url)
+    if %r{\A(?:(http|https)://(dx\.)?(doi.org|handle.test.datacite.org)/)?(doi:)?(10\.\d{4,5}/.+)\z}
+        .match?(url)
+
+      uri = Addressable::URI.parse(url)
+
+      uri.path.gsub(%r{^/}, "").upcase
+    end
+  end
+end

app/validators/uuid_format_validator.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class UuidFormatValidator < ActiveModel::EachValidator
+  def validate_each(record, attribute, value)
+    unless UUID.validate(value)
+      record.errors.add(attribute, "#{value} is not a valid UUID")
+    end
+  end
+end

app/workers/event_import_worker.rb

@@ -8,32 +8,77 @@ class EventImportWorker
   def perform(sqs_message = nil, data = nil)
     log_prefix = "[Events:EventImportWorker]"
 
-    if data.blank?
+    event_data = event_data(data)
+
+    if event_data.nil?
       Rails.logger.info("#{log_prefix} Message data was blank")
       return
     end
 
-    log_identifier = "subj_id: #{data["subjId"]}, " \
-      "obj_id: #{data["objId"]}, " \
-      "source_id: #{data["sourceId"]}, " \
-      "relation_type_id: #{data["relationTypeId"]}"
+    log_identifier = log_identifier(event_data)
 
     Rails.logger.info("#{log_prefix} Start of event message processing for #{log_identifier}")
-    Rails.logger.info("#{log_prefix} Searching for event with #{log_identifier}")
+    Rails.logger.info("#{log_prefix} Searching for event for #{log_identifier}")
 
-    event = Event.find_by(
-      subj_id: data["subjId"],
-      obj_id: data["objId"],
-      source_id: data["sourceId"],
-      relation_type_id: data["relationTypeId"],
-    )
+    event = find_event(event_data)
 
-    if event.blank?
-      Rails.logger.info("#{log_prefix} Creating a new event with #{log_identifier}")
+    if event.nil?
+      create_event(event_data, log_prefix, log_identifier)
     else
-      Rails.logger.info("#{log_prefix} Update an existing event with #{log_identifier}")
+      update_event(event, event_data, log_prefix, log_identifier)
     end
 
-    Rails.logger.info("#{log_prefix} End of event message processing for #{log_identifier}")
+    Rails.logger.info("#{log_prefix} Completed processing event for #{log_identifier}")
+  end
+
+  private
+
+  # Returns the SQS event data as a hash.
+  # Will return nil if either a data or attributes field is missing.
+  def event_data(data)
+    data_hash = JSON.parse(data)
+
+    data_hash.dig("data", "attributes")
+  end
+
+  # Returns a string which serves as an identifier for this event import worker run.
+  # Consists of the event subj_id, obj_id, source_id and relation_type_id.
+  def log_identifier(event_data)
+    "subj_id: #{event_data["subjId"]}, " \
+      "obj_id: #{event_data["objId"]}, " \
+      "source_id: #{event_data["sourceId"]}, " \
+      "relation_type_id: #{event_data["relationTypeId"]}"
+  end
+
+  # Searchs for an existing event using subj_id, obj_id, source_id and relation_type_id
+  def find_event(event_data)
+    Event.find_by(
+      subj_id: event_data["subjId"],
+      obj_id: event_data["objId"],
+      source_id: event_data["sourceId"],
+      relation_type_id: event_data["relationTypeId"],
+    )
+  end
+
+  def create_event(event_data, log_prefix, log_identifier)
+    Rails.logger.info("#{log_prefix} Creating a new event for #{log_identifier}")
+
+    event = EventFactory.create_from_sqs(event_data)
+
+    if event.save
+      Rails.logger.info("#{log_prefix} Event successfully created for #{log_identifier}")
+    elsif event.errors.any?
+      Rails.logger.error("#{log_prefix} Creating event failed for #{log_identifier}: #{event.errors.inspect}")
+    end
+  end
+
+  def update_event(event, event_data, log_prefix, log_identifier)
+    Rails.logger.info("#{log_prefix} Update an existing event for #{log_identifier}")
+
+    if event.update(data)
+      Rails.logger.info("#{log_prefix} Event successfully updated for #{log_identifier}")
+    else
+      Rails.logger.error("#{log_prefix} Updating event failed for #{log_identifier}: #{event.errors.inspect}")
+    end
   end
 end

config/initializers/sentry.rb

@@ -25,10 +25,3 @@
     event
   end
 end
-# Raven.configure do |config|
-#   config.environments = ["stage", "production"]
-#   config.dsn = ENV["SENTRY_DSN"]
-#   config.release = "events:" + Events::Application::VERSION
-#   config.sanitize_fields = Rails.application.config.filter_parameters.map(&:to_s)
-#   config.logger = Rails.application.config.lograge.logger
-# end