Create enrichment table and model

app/controllers/enrichments_controller.rb

@@ -0,0 +1,101 @@
+# frozen_string_literal: true
+
+class EnrichmentsController < ApplicationController
+  PAGE_SIZE = 25
+
+  def index
+    doi = params["doi"]&.upcase
+    client_id = params["client_id"]
+    cursor = params.dig("page", "cursor")
+
+    base_enrichments = base_page_enrichments(doi, client_id)
+
+    enrichments = if cursor.present?
+      cursor_updated_at, cursor_id, cursor_page = decode_cursor(cursor)
+      base_enrichments.by_cursor(cursor_updated_at, cursor_id)
+    else
+      base_enrichments
+    end
+
+    enrichments = enrichments.order_by_cursor.limit(PAGE_SIZE).to_a
+
+    cursor_page ||= 1
+
+    options = {
+      meta: build_meta(base_enrichments, cursor_page),
+      links: build_paging_links(enrichments, doi, client_id, cursor_page)
+    }
+
+    render(json: EnrichmentSerializer.new(enrichments, options).serializable_hash, status: :ok)
+  end
+
+  private
+    def base_page_enrichments(doi, client_id)
+      if doi.present?
+        Enrichment.by_doi(doi)
+      elsif client_id.present?
+        Enrichment.by_client(client_id)
+      else
+        Enrichment.all
+      end
+    end
+
+    def encode_cursor(hash)
+      Base64.urlsafe_encode64(hash.to_json, padding: false)
+    rescue
+      raise ActionController::InternalServerError, "Failed to encode cursor"
+    end
+
+    def decode_cursor(token)
+      decoded_cursor = JSON.parse(Base64.urlsafe_decode64(token))
+      cursor_updated_at = Time.iso8601(decoded_cursor.fetch("updated_at"))
+      cursor_id = decoded_cursor.fetch("id").to_i
+      cursor_page = decoded_cursor.fetch("page", nil).to_i || 0
+
+      Rails.logger.info("cursor_page: #{cursor_page}")
+
+      [cursor_updated_at, cursor_id, cursor_page]
+    rescue
+      raise ActionController::BadRequest, "Invalid cursor"
+    end
+
+    def build_meta(enrichments, cursor_page)
+      enrichments_total = enrichments.count
+
+      {
+        total: enrichments_total,
+        totalPages: (enrichments_total / PAGE_SIZE.to_f).ceil,
+        page: cursor_page
+      }
+    end
+
+    def build_next_link(doi, client_id, next_cursor)
+      base_link = request.original_url.split("?").first
+
+      query_string = if doi.present?
+        "doi=#{doi}&cursor=#{next_cursor}"
+      elsif client_id.present?
+        "client-id=#{client_id}&cursor=#{next_cursor}"
+      else
+        "page[cursor]=#{next_cursor}"
+      end
+
+      "#{base_link}?#{query_string}"
+    end
+
+    def build_paging_links(enrichments, doi, client_id, cursor_page)
+      current_link = request.original_url
+
+      next_cursor = if enrichments.any?
+        last = enrichments.last
+        encode_cursor(updated_at: last.updated_at.iso8601(6), id: last.id, page: cursor_page + 1)
+      end
+
+      next_link = build_next_link(doi, client_id, next_cursor)
+
+      {
+        self: current_link,
+        next: enrichments.length == PAGE_SIZE ? next_link : nil
+      }
+    end
+end

app/jobs/enrichment_batch_process_job.rb

@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+class EnrichmentBatchProcessJob < ApplicationJob
+  include ErrorSerializable
+
+  queue_as :enrichment_batch_process_job
+
+  def perform(lines, filename)
+    log_prefix = "EnrichmentBatchProcessJob (#{filename})"
+
+    # We will process the lines in parallel to speed up ingestion.
+    Parallel.each(lines, in_threads: 5) do |line|
+      # with_connection ensures the connection is explicitly checked out and returned to the pool after
+      # each iteration, preventing connection pool exhaustion.
+      ActiveRecord::Base.connection_pool.with_connection do
+        begin
+          parsed_line = JSON.parse(line)
+        rescue JSON::ParserError => e
+          Rails.logger.error("#{log_prefix}: Failed to parse line: #{e.message}")
+          next
+        end
+
+        # We only create enrichments for DOIs that exist and which have an agency of 'datacite'.
+        uid = parsed_line["doi"]&.upcase
+        doi = Doi.find_by(doi: uid, agency: "datacite")
+
+        if doi.blank?
+          Rails.logger.error("#{log_prefix}: Doi #{uid} does not exist")
+          next
+        end
+
+        if doi.enrichment_field(parsed_line["field"]).nil?
+          Rails.logger.error("#{log_prefix}: Unsupported enrichment field #{parsed_line["field"]} for DOI #{uid}")
+          next
+        end
+
+        # We set the only_validate flag on the DOI model to true such that we
+        # ensure that validation functions as expected when not persisting the record.
+        doi.only_validate = true
+
+        enrichment = Enrichment.new(
+          filename: filename,
+          doi: uid,
+          contributors: parsed_line["contributors"],
+          resources: parsed_line["resources"],
+          field: parsed_line["field"],
+          action: parsed_line["action"],
+          original_value: parsed_line["originalValue"],
+          enriched_value: parsed_line["enrichedValue"]
+        )
+
+        doi.apply_enrichment(enrichment)
+
+        unless doi.valid?
+          errors = serialize_errors(doi.errors, uid: enrichment.doi)
+          Rails.logger.error("#{log_prefix}: Enrichment does not generate valid metadata: #{errors}")
+          next
+        end
+
+        unless enrichment.save
+          errors = enrichment.errors.full_messages.join(";")
+          Rails.logger.error("#{log_prefix}: Enrichment failed to save: #{errors}")
+        end
+      end
+    end
+  end
+end

app/models/concerns/enrichable.rb

@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+module Enrichable
+  extend ActiveSupport::Concern
+
+  def apply_enrichment(enrichment)
+    action = enrichment["action"]
+
+    field = enrichment_field(enrichment["field"])
+
+    raise ArgumentError, "Invalid enrichment field #{enrichment["field"]}" if field.nil?
+
+    case action
+    when "insert"
+      self[field] ||= []
+      self[field] << enrichment["enriched_value"]
+    when "update"
+      self[field] = enrichment["enriched_value"]
+    when "update_child"
+      self[field].each_with_index do |item, index|
+        if item == enrichment["original_value"]
+          self[field][index] = enrichment["enriched_value"]
+        end
+      end
+    when "delete_child"
+      self[field] ||= []
+      self[field].each_with_index do |item, index|
+        if item == enrichment["original_value"]
+          self[field].delete_at(index)
+          break
+        end
+      end
+    end
+  end
+
+  def enrichment_field(field)
+    field_mapping = {
+      "alternateIdentifiers" => "alternate_identifiers",
+      "creators" => "creators",
+      "titles" => "titles",
+      "publisher" => "publisher",
+      "publicationYear" => "publication_year",
+      "subjects" => "subjects",
+      "contributors" => "contributors",
+      "dates" => "dates",
+      "language" => "language",
+      "types" => "types",
+      "relatedIdentifiers" => "related_identifiers",
+      "relatedItems" => "related_items",
+      "sizes" => "sizes",
+      "formats" => "formats",
+      "version" => "version",
+      "rightsList" => "rights_list",
+      "descriptions" => "descriptions",
+      "geoLocations" => "geo_locations",
+      "fundingReferences" => "funding_references"
+    }
+
+    field_mapping.fetch(field, nil)
+  end
+end

app/models/doi.rb

@@ -35,6 +35,8 @@ class Doi < ApplicationRecord
 
   include Elasticsearch::Model
 
+  include Enrichable
+
   aasm whiny_transitions: false do
     # draft is initial state for new DOIs.
     state :draft, initial: true

app/models/enrichment.rb

@@ -0,0 +1,54 @@
+class Enrichment < ApplicationRecord
+  validate :validate_json_schema
+
+  belongs_to :doi_record,
+    class_name: "Doi",
+    foreign_key: :doi, # enrichments.doi
+    primary_key: :doi, # dois.doi
+    optional: false
+
+  has_one :client, through: :doi_record
+
+  scope :by_doi, ->(doi) { where(doi: doi) }
+
+  scope :by_client, ->(client_id) { joins(doi_record: :client).where(datacentre: { symbol: client_id }) }
+
+  scope :by_cursor, ->(updated_at, id) {
+    where("(enrichments.updated_at < ?) OR (enrichments.updated_at = ? AND enrichments.id < ?)",
+      updated_at,
+      updated_at,
+      id)
+  }
+
+  scope :order_by_cursor, -> { order(updated_at: :desc, id: :desc) }
+
+  private
+    def validate_json_schema
+      doc = to_enrichment_hash
+      error_list = self.class.enrichment_schemer.validate(doc).to_a
+
+      return if error_list.empty?
+
+      errors.add(:base, "Validation failed: #{error_list.map { |e| e['message'] || e.inspect }.join('; ')}")
+    end
+
+    def to_enrichment_hash
+      {
+        "doi" => doi,
+        "contributors" => contributors,
+        "resources" => resources,
+        "field" => field,
+        "action" => action,
+        "originalValue" => original_value,
+        "enrichedValue" => enriched_value
+      }.compact
+    end
+
+    def self.enrichment_schemer
+      @enrichment_schemer ||= begin
+        schema_path = Rails.root.join("app/models/schemas/enrichment/enrichment.json")
+        schema = JSON.parse(File.read(schema_path))
+        JSONSchemer.schema(schema)
+      end
+    end
+end