Publish to S3 via a debounced queue (#1022)

- Debounces sync events initiated by publishing
- Locks the API while syncing is in progress
- Uploads manifest with all uploaded S3 files
- Triggers Argo workflow for copying items and indexing
- Add rake task to mark baseline sync start point

Full information about each item is available
[here](https://github.com/CredentialEngine/CredentialRegistry/blob/090c7e0ba58284d7c79d48b09b1c5e971d85aca0/docs/11_registry_changeset_sync.md).

---------

Co-authored-by: Ariel Rolfo <arielr-lt+username@users.noreply.github.com>

Author: rsaksida

app/api/entities/version.rb

@@ -11,13 +11,16 @@ class Version < Grape::Entity
       expose :created_at,
              documentation: { type: 'datetime',
                               desc: 'When the version was created' }
-      expose :whodunnit,
-             as: :actor,
+      expose :actor,
              documentation: { type: 'string',
                               desc: 'Who performed the changes' }
       expose :url,
              documentation: { type: 'string', desc: 'Version URL' }
 
+      def actor
+        object.author.presence || object.whodunnit
+      end
+
       def created_at
         Time.zone.parse(object.created_at) if object.created_at?
       end

app/api/helpers/community_helpers.rb

@@ -1,3 +1,5 @@
+require 'registry_changeset_sync'
+
 # EnvelopeCommunity specific helpers
 module CommunityHelpers
   extend Grape::API::Helpers
@@ -21,6 +23,12 @@ def community_error(msg)
     json_error! [msg], nil, :unprocessable_entity
   end
 
+  def assert_publish_unlocked!
+    return unless RegistryChangesetSync.syncing?(current_community)
+
+    json_error!([RegistryChangesetSync::PUBLISH_LOCKED], nil, 503)
+  end
+
   def normalized_community_names
     [
       params[:community_name].try(:underscore),

app/api/v1/publish.rb

@@ -1,6 +1,5 @@
 require 'policies/envelope_policy'
 require 'services/publish_interactor'
-require 'services/sync_envelope_graph_with_s3'
 
 module API
   module V1
@@ -37,6 +36,7 @@ module Publish
             end
             post do
               authorize Envelope, :create?
+              assert_publish_unlocked!
 
               secondary_token_header = request.headers['Secondary-Token']
               secondary_token = if secondary_token_header.present?

app/api/v2/publish.rb

@@ -41,6 +41,8 @@ class Publish < MountableAPI # rubocop:todo Style/Documentation
             use :update_if_exists
           end
           post do
+            assert_publish_unlocked!
+
             secondary_token_header = request.headers['Secondary-Token']
             secondary_token = (secondary_token_header.split.last if secondary_token_header.present?)
 

app/jobs/extract_envelope_resources_job.rb

@@ -1,12 +1,38 @@
 require 'extract_envelope_resources'
+require 'registry_changeset_sync'
 require 'index_envelope_job'
 require 'precalculate_description_sets_job'
 
 # Runs the ExtractEnvelopeResources service in background
 class ExtractEnvelopeResourcesJob < ActiveJob::Base
+  self.enqueue_after_transaction_commit = true
+
   def perform(envelope_id)
     envelope = Envelope.find(envelope_id)
     ExtractEnvelopeResources.call(envelope:)
+    record_s3_sync_activity(envelope)
     IndexEnvelopeJob.perform_later(envelope_id)
   end
+
+  private
+
+  def record_s3_sync_activity(envelope)
+    RegistryChangesetSync.record_activity!(
+      envelope.envelope_community,
+      version_id: version_id_for_s3_sync(envelope),
+      resource_event_id: latest_resource_event_id(envelope)
+    )
+  end
+
+  def version_id_for_s3_sync(envelope)
+    return if envelope.envelope_ceterms_ctid.blank?
+
+    EnvelopeVersion.where(item_type: 'Envelope', item_id: envelope.id).maximum(:id)
+  end
+
+  def latest_resource_event_id(envelope)
+    EnvelopeResourceSyncEvent.where(
+      envelope_community: envelope.envelope_community
+    ).maximum(:id)
+  end
 end

app/jobs/publish_envelope_job.rb

@@ -26,18 +26,18 @@ def build_args(request_params) # rubocop:todo Metrics/AbcSize, Metrics/MethodLen
       current_user: User.find(request_params[:user_id])
     }
 
-    if request_params.key?(:envelope_id)
+    if request_params[:envelope_id].present?
       interactor_args[:envelope] = Envelope.find(request_params[:envelope_id])
     else
       interactor_args[:raw_resource] = request_params[:raw_resource]
     end
 
-    if request_params.key?(:publishing_organization_id)
+    if request_params[:publishing_organization_id].present?
       interactor_args[:publishing_organization] =
         Organization.find(request_params[:publishing_organization_id])
     end
 
-    if request_params.key?(:secondary_token)
+    if request_params[:secondary_token].present?
       interactor_args[:secondary_token] =
         request_params[:secondary_token]
     end

app/jobs/sync_registry_changesets_job.rb

@@ -0,0 +1,60 @@
+require 'registry_changeset_sync'
+require 'sync_pending_registry_changesets'
+
+class SyncRegistryChangesetsJob < ActiveJob::Base # rubocop:todo Style/Documentation
+  def perform(sync_id)
+    sync = RegistryChangesetSync.find(sync_id)
+    last_activity_at = nil
+    last_activity_version_id = nil
+    last_activity_resource_event_id = nil
+    pending_sync = false
+    wait_until = nil
+
+    sync.with_lock do
+      last_activity_at = sync.last_activity_at
+      last_activity_version_id = sync.last_activity_version_id
+      last_activity_resource_event_id = sync.last_activity_resource_event_id
+      pending_graph_sync = last_activity_version_id.present? &&
+                           (sync.last_synced_version_id.blank? ||
+                            sync.last_synced_version_id < last_activity_version_id)
+      pending_resource_sync = last_activity_resource_event_id.present? &&
+                              (sync.last_synced_resource_event_id.blank? ||
+                               sync.last_synced_resource_event_id < last_activity_resource_event_id)
+      pending_sync = pending_graph_sync || pending_resource_sync
+
+      return if sync.scheduled_for_at.blank? && !pending_sync
+
+      if sync.scheduled_for_at.present?
+        wait_until = last_activity_at + RegistryChangesetSync.debounce_window
+
+        if wait_until > Time.current
+          sync.update!(scheduled_for_at: wait_until)
+        else
+          wait_until = nil
+          sync.update!(scheduled_for_at: nil)
+        end
+      end
+    end
+
+    if wait_until
+      self.class.set(wait_until: wait_until).perform_later(sync.id)
+      return
+    end
+
+    return unless last_activity_at && pending_sync && sync.mark_syncing!
+
+    begin
+      SyncPendingRegistryChangesets.new(
+        envelope_community: sync.envelope_community,
+        cutoff_version_id: last_activity_version_id,
+        cutoff_resource_event_id: last_activity_resource_event_id,
+        sync: sync
+      ).call
+    rescue StandardError => e
+      sync.mark_sync_error!(e)
+      raise
+    ensure
+      sync.clear_syncing! if sync.reload.argo_workflows.blank?
+    end
+  end
+end

app/models/envelope.rb

@@ -3,6 +3,8 @@
 require 'authorized_key'
 require 'export_to_ocn_job'
 require 'delete_from_ocn_job'
+require 'registry_changeset_sync'
+require 'envelope_resource_sync_event'
 require 'envelope_version'
 require_relative 'extensions/transactionable_envelope'
 require_relative 'extensions/learning_registry_resources'
@@ -46,10 +48,10 @@ class Envelope < ActiveRecord::Base
   before_validation :process_resource, :process_headers
   before_save :assign_last_verified_on
   after_save :update_headers
-  after_save :upload_to_s3
   before_destroy :delete_description_sets, prepend: true
+  before_destroy :record_resource_sync_delete_events, prepend: true
   after_destroy :delete_from_ocn
-  after_destroy :delete_from_s3
+  after_commit :schedule_s3_delete, on: :destroy
   after_commit :export_to_ocn
 
   validates :envelope_community, :envelope_type, :envelope_version,
@@ -271,11 +273,48 @@ def export_to_ocn
     ExportToOCNJob.perform_later(id)
   end
 
-  def upload_to_s3
-    SyncEnvelopeGraphWithS3.upload(self)
+  def schedule_s3_delete
+    return if envelope_ceterms_ctid.blank? || envelope_community.blank?
+
+    RegistryChangesetSync.record_activity!(
+      envelope_community,
+      version_id: latest_s3_sync_version_id,
+      resource_event_id: @last_resource_sync_event_id_for_destroy
+    )
+  end
+
+  def latest_s3_sync_version_id
+    EnvelopeVersion
+      .where(item_type: self.class.name, item_id: id)
+      .maximum(:id)
+  end
+
+  def latest_resource_sync_event_id
+    EnvelopeResourceSyncEvent
+      .where(envelope_community: envelope_community)
+      .maximum(:id)
   end
 
-  def delete_from_s3
-    SyncEnvelopeGraphWithS3.remove(self)
+  def record_resource_sync_delete_events
+    return if envelope_ceterms_ctid.blank? || envelope_community.blank?
+
+    resource_ids = envelope_resources.pluck(:resource_id)
+    return if resource_ids.empty?
+
+    now = Time.current
+
+    EnvelopeResourceSyncEvent.insert_all!(
+      resource_ids.map do |resource_id|
+        {
+          envelope_community_id: envelope_community.id,
+          resource_id: resource_id,
+          action: EnvelopeResourceSyncEvent::ACTIONS[:delete],
+          created_at: now,
+          updated_at: now
+        }
+      end
+    )
+
+    @last_resource_sync_event_id_for_destroy = latest_resource_sync_event_id
   end
 end

app/models/envelope_metadata.rb

@@ -0,0 +1,65 @@
+class EnvelopeMetadata
+  attr_reader :envelope
+
+  def self.from_envelope(envelope)
+    new(envelope)
+  end
+
+  def initialize(envelope)
+    @envelope = envelope
+  end
+
+  def as_json(*)
+    {
+      envelope_community: envelope.envelope_community_name,
+      envelope_id: envelope.envelope_id,
+      envelope_ceterms_ctid: envelope.envelope_ceterms_ctid,
+      envelope_ctdl_type: envelope.envelope_ctdl_type,
+      envelope_type: envelope.envelope_type,
+      envelope_version: envelope.envelope_version,
+      publisher_id: envelope.publisher_id,
+      secondary_publisher_id: envelope.secondary_publisher_id,
+      resource_publish_type: envelope.resource_publish_type,
+      node_headers: node_headers,
+      owned_by: envelope.organization&._ctid,
+      published_by: envelope.publishing_organization&._ctid,
+      changed: false,
+      updated_at: envelope.updated_at,
+      last_verified_on: envelope.last_verified_on
+    }
+  end
+
+  private
+
+  def node_headers
+    headers = envelope.decoded_node_headers
+
+    {
+      resource_digest: headers.resource_digest,
+      revision_history: Array(headers.versions).map { |version| revision_history_entry(version) },
+      created_at: parse_time(headers.created_at),
+      updated_at: parse_time(headers.updated_at),
+      deleted_at: parse_time(headers.deleted_at)
+    }
+  end
+
+  def revision_history_entry(version)
+    {
+      head: version.head,
+      event: version.event,
+      created_at: parse_time(version.created_at),
+      actor: revision_actor(version),
+      url: version.url
+    }
+  end
+
+  def revision_actor(version)
+    version.author.presence || version.whodunnit
+  end
+
+  def parse_time(value)
+    return if value.blank?
+
+    Time.zone.parse(value)
+  end
+end

app/models/envelope_resource_sync_event.rb

@@ -0,0 +1,23 @@
+class EnvelopeResourceSyncEvent < ActiveRecord::Base
+  ACTIONS = {
+    upsert: 0,
+    delete: 1
+  }.freeze
+
+  belongs_to :envelope_community
+
+  validates :envelope_community_id, :resource_id, :action, presence: true
+  validates :action, inclusion: { in: ACTIONS.values }
+
+  def delete?
+    action == ACTIONS[:delete]
+  end
+
+  def upsert?
+    action == ACTIONS[:upsert]
+  end
+
+  def action_name
+    ACTIONS.key(action).to_s
+  end
+end