🐛 Attach imported file sets in one pass

app/factories/bulkrax/object_factory.rb

@@ -12,9 +12,8 @@ class ObjectFactory < ObjectFactoryInterface
     # @note This does not save either object.  We need to do that in another
     #       loop.  Why?  Because we might be adding many items to the parent.
     def self.add_child_to_parent_work(parent:, child:)
-      return true if parent.ordered_members.to_a.include?(child)
-
-      parent.ordered_members << child
+      parent.ordered_members << child unless parent.ordered_members.to_a.include?(child)
+      parent
     end
 
     def self.add_resource_to_collection(collection:, resource:, user:)

app/factories/bulkrax/object_factory_interface.rb

@@ -59,6 +59,17 @@ def self.add_resource_to_collection(collection:, resource:, user:)
       raise NotImplementedError, "#{self}.#{__method__}"
     end
 
+    ##
+    # Uploads that wait for CreateRelationshipsJob to add their file set to its work.
+    # The job queues each with .ingest_later once it has saved the work.
+    def self.uploads_awaiting_ingest(resource:) # rubocop:disable Lint/UnusedMethodArgument
+      []
+    end
+
+    def self.ingest_later(upload:)
+      raise NotImplementedError, "#{self}.#{__method__}"
+    end
+
     ##
     # Add the user to the collection; assuming the given collection is a
     # Collection.  This is also only something we use in Hyrax.

app/factories/bulkrax/valkyrie_object_factory.rb

@@ -115,11 +115,30 @@ def self.transactions
     # To do this we are bypassing the save! method defined below
     def self.add_child_to_parent_work(parent:, child:)
       parent = self.find(parent.id)
-      return true if parent.member_ids.include?(child.id)
+      return parent if parent.member_ids.include?(child.id)
       parent.member_ids += [child.id]
+      if child.is_a?(Bulkrax.file_model_class)
+        parent.representative_id = child.id if parent.respond_to?(:representative_id) && parent.representative_id.blank?
+        parent.thumbnail_id = child.id if parent.respond_to?(:thumbnail_id) && parent.thumbnail_id.blank?
+      end
       Hyrax.persister.save(resource: parent)
     end
 
+    ##
+    # File sets are created unattached (see #create_file_set), so their files are ingested once
+    # CreateRelationshipsJob has attached them. A file set already a member with no files is one
+    # a killed pass attached but never queued.
+    def self.uploads_awaiting_ingest(resource:)
+      return [] unless resource.is_a?(Bulkrax.file_model_class) && resource.file_ids.blank?
+
+      Hyrax::UploadedFile.where(file_set_uri: resource.id.to_s).to_a
+    end
+
+    def self.ingest_later(upload:)
+      # perform_later returns false, without raising, when a callback aborts the enqueue
+      ::ValkyrieIngestJob.perform_later(upload) || raise("Could not queue the ingest of upload #{upload.id}")
+    end
+
     ##
     # The resource added to a collection can be either a work or another collection.
     def self.add_resource_to_collection(collection:, resource:, user:)
@@ -393,18 +412,81 @@ def conditionally_set_reindex_extent
       nil
     end
 
-    # @note We perform the transaction against the *parent* here, because the FileSets are generated and updated in relationship with their parent, not in isolation
+    # Saving the parent here would save the member_ids it was read with, dropping file sets
+    # other rows added meanwhile, so the file set is left for CreateRelationshipsJob to attach.
+    # It takes the parent's permissions, then the row's visibility, or else the parent's
+    # visibility, embargo and lease.
     def create_file_set(attrs)
       attrs = HashWithIndifferentAccess.new(attrs)
-      parent_object = find_record(attributes[related_parents_parsed_mapping].first, importer_run_id).last
-      perform_transaction_for(object: parent_object, attrs: {}) do
-        fs_attrs = attrs.merge(attributes).symbolize_keys
-        uploaded_files, = prep_fileset_content(attrs)
-        transactions['change_set.update_work']
-          .with_step_args(
-            'work_resource.add_file_sets' => { uploaded_files: uploaded_files, file_set_params: [fs_attrs] },
-            'work_resource.save_acl' => { permissions_params: [attrs.try('visibility') || 'open'].compact }
-          )
+      parent = row_parent || raise(StandardError, "#{source_identifier_value} names no parent")
+      uploaded_files, = prep_fileset_content(attrs)
+      raise StandardError, "#{source_identifier_value} lists #{uploaded_files.size} files; a file set row takes one" if uploaded_files.size > 1
+
+      upload = uploaded_files.first
+      raise StandardError, "#{source_identifier_value} has no file to attach" if upload.nil? || upload.file_set_uri.present?
+
+      params = attrs.merge(attributes).symbolize_keys
+      # A visibility only a work can carry, such as embargo or lease, counts as none
+      params.delete(:visibility) unless Hyrax.config.visibility_map.visibilities.include?(params[:visibility])
+      file_set = create_unattached_file_set(parent, upload, params)
+      params[:visibility].present? ? file_set : inherit_visibility(from: parent, to: file_set)
+    end
+
+    # The row's first parent, or nil when it names none
+    def row_parent
+      identifier = Array(attributes[related_parents_parsed_mapping]).first
+      return if identifier.blank?
+
+      parent = find_record(identifier, importer_run_id).last
+      raise StandardError, "#{source_identifier_value} names parent #{identifier}, which was not found" unless parent
+      raise StandardError, "#{source_identifier_value} names parent #{identifier}, which is a #{parent.class}, not a work" unless parent.work?
+
+      parent
+    end
+
+    # Hyrax::WorkUploadsHandler#make_file_set_and_ingest as of Hyrax 5.3, minus adding the file
+    # set to the work and, as Hyrax 4 also does there, queueing its ingest
+    def create_unattached_file_set(parent, upload, params)
+      file_set = Hyrax.persister.save(resource: Bulkrax.file_model_class.new(file_set_attributes(upload).merge(params)))
+      Hyrax.publisher.publish('object.deposited', object: file_set, user: upload.user)
+      upload.add_file_set!(file_set)
+      Hyrax::AccessControlList.copy_permissions(source: Hyrax::AccessControlList.new(resource: parent), target: file_set)
+      file_set.visibility = params[:visibility] if params[:visibility].present?
+      file_set.permission_manager.acl.save if file_set.permission_manager.acl.pending_changes?
+      file_set
+    end
+
+    def file_set_attributes(upload)
+      { depositor: upload.user.user_key,
+        creator: [upload.user.user_key],
+        date_uploaded: upload.created_at,
+        date_modified: Hyrax::TimeService.time_in_utc,
+        label: upload.uploader.filename,
+        title: [upload.uploader.filename] }
+    end
+
+    def inherit_visibility(from:, to:)
+      clear_ended_embargo_and_lease(to)
+      to.visibility = from.visibility
+      Hyrax::EmbargoManager.new(resource: from).copy_embargo_to(target: to)
+      Hyrax::LeaseManager.new(resource: from).copy_lease_to(target: to)
+      to.permission_manager.acl.save
+      Hyrax.persister.save(resource: to).tap { |saved| Hyrax.index_adapter.save(resource: saved) }
+    end
+
+    POLICY_FIELDS = { embargo: %i[embargo_release_date visibility_during_embargo visibility_after_embargo],
+                      lease: %i[lease_expiration_date visibility_during_lease visibility_after_lease] }.freeze
+
+    # An ended embargo or lease left on the file set would let Hyrax's expiry jobs later
+    # release it to the visibility it was to have afterwards
+    def clear_ended_embargo_and_lease(file_set)
+      POLICY_FIELDS.each do |policy_name, fields|
+        policy = file_set.public_send(policy_name)
+        next if policy.nil? || policy.active?
+
+        fields.each { |field| policy.public_send("#{field}=", nil) }
+        # Hyrax 4 embeds it with no id, and saving the file set saves it
+        Hyrax.persister.save(resource: policy) if policy.id.present?
       end
     end
 
@@ -571,13 +653,22 @@ def update_file_set(attrs)
       remote_files = Array(attrs[:remote_files])
       raise StandardError, "#{source_identifier_value} lists #{remote_files.size} remote files; a file set takes one" if remote_files.size > 1
 
+      parent = row_parent
       # Download before saving, so a failed download leaves the file set as it was
       path = download_file(remote_files.first[:url]) if remote_files.any?
       fs_attrs = attrs.merge(attributes).symbolize_keys
       file_set = perform_transaction_for(object: object, attrs: fs_attrs) do
         transactions['change_set.update_file_set']
       end
-      path ? replace_original_file(file_set, path, remote_files.first[:file_name]) : file_set
+      file_set = replace_original_file(file_set, path, remote_files.first[:file_name]) if path
+      return file_set if fs_attrs[:visibility].present? || parent.nil? || under_embargo_or_lease?(file_set)
+
+      inherit_visibility(from: parent, to: file_set)
+    end
+
+    # Its own embargo or lease keeps a file set as it is until it ends
+    def under_embargo_or_lease?(file_set)
+      Hyrax::EmbargoManager.new(resource: file_set).under_embargo? || Hyrax::LeaseManager.new(resource: file_set).under_lease?
     end
 
     # Replace the file set's original file with the downloaded one, and drop the old original and the

app/jobs/bulkrax/create_relationships_job.rb

@@ -177,8 +177,9 @@ def process_parent_as_collection(parent_record:, parent_identifier:)
     def process_parent_as_work(parent_record:, parent_identifier:)
       conditionally_acquire_lock_for(parent_record.id.to_s) do
         ActiveRecord::Base.uncached do
+          # find_each would ignore the order; id breaks ties among children named in one cell
           Bulkrax::PendingRelationship.where(parent_id: parent_identifier, importer_run_id: @importer_run_id)
-                                      .ordered.find_each do |rel|
+                                      .ordered.order(:id).each do |rel|
             raise "#{rel} needs a child to create relationship" if rel.child_id.nil?
             raise "#{rel} needs a parent to create relationship" if rel.parent_id.nil?
             # depending on the object factory, we may not save the actual parent record, so we update it here
@@ -202,7 +203,13 @@ def process_parent_as_work(parent_record:, parent_identifier:)
           Bulkrax.object_factory.update_index(resources: [reloaded_parent])
           Bulkrax.object_factory.publish(event: 'object.membership.updated', object: reloaded_parent, user: @user)
         end
+        # Only once the work is saved: if that fails, the relationships stay for the retry.
+        # Under the lock, so another pass for the work cannot finish the same ones.
+        finish_linked_relationships
       end
+    rescue => e
+      # Recorded rather than raised, so #perform reschedules the pass whatever the queue adapter's retry policy
+      record_pass_failure(parent_identifier, e)
     end
     # rubocop:enable Layout/RescueEnsureAlignment
 
@@ -247,11 +254,35 @@ def add_to_work(relationship:, parent_record:, ability:)
       )
       # default is false for this... do not typically need to index file sets of child records
       Bulkrax.object_factory.update_index_for_file_sets_of(resource: child_record) if update_child_records_works_file_sets?
-      relationship.destroy
+      linked_relationships << [relationship, Bulkrax.object_factory.uploads_awaiting_ingest(resource: child_record)]
 
       updated_parent
     end
 
+    def linked_relationships
+      @linked_relationships ||= []
+    end
+
+    def record_pass_failure(parent_identifier, error)
+      @number_of_failures += 1
+      @errors << error
+      # No longer Pending, so a stalled ScheduleRelationshipsJob leaves them to this job's
+      # reschedule rather than retrying them hourly, which looks like progress and never gives up
+      relationships = Bulkrax::PendingRelationship.where(parent_id: parent_identifier, importer_run_id: @importer_run_id)
+      relationships.pending.update_all(status_message: error.message.truncate(STATUS_MESSAGE_LIMIT)) # rubocop:disable Rails/SkipsModelValidations
+      # Counted, so the pass that links them recounts and sets the importer's status back
+      record_relationship_failures
+    end
+
+    # Ingest runs with the file set in its saved work, never mid-pass
+    def finish_linked_relationships
+      queued = Set.new
+      linked_relationships.each do |relationship, uploads|
+        uploads.each { |upload| Bulkrax.object_factory.ingest_later(upload: upload) if queued.add?(upload.id) }
+        relationship.destroy
+      end
+    end
+
     def child_not_found(relationship)
       "Child record #{relationship.child_id} could not be found for parent #{relationship.parent_id}"
     end

app/models/concerns/bulkrax/file_set_entry_behavior.rb

@@ -51,12 +51,33 @@ def validate_presence_of_parent!
       raise OrphanFileSetError, 'File set must be related to at least one work'
     end
 
+    # ObjectFactory#create_file_set attaches the file set to its parent. ValkyrieObjectFactory
+    # leaves that to CreateRelationshipsJob, and only for the first parent, the one it reads.
     def parent_jobs
-      false # FileSet relationships are handled in ObjectFactory#create_file_set
+      return false unless valkyrie_object_factory?
+
+      parent_identifier = Array(parsed_metadata[related_parents_parsed_mapping]).first
+      # try: ImportBehavior#deleted_identifier? arrives with samvera/bulkrax#1279
+      return if parent_identifier.blank? || try(:deleted_identifier?, parent_identifier)
+
+      PendingRelationship.create!(child_id: identifier, parent_id: parent_identifier, importer_run_id: importerexporter.last_run.id, order: id)
+    end
+
+    # ValkyrieObjectFactory gives a file set without a visibility its parent's
+    def add_visibility
+      return super unless valkyrie_object_factory?
+
+      parsed_metadata.delete('visibility') if parsed_metadata['visibility'].blank?
     end
 
     def child_jobs
       raise ::StandardError, "A #{Bulkrax.file_model_class} cannot be a parent of a #{Bulkrax.collection_model_class}, Work, or other #{Bulkrax.file_model_class}"
     end
+
+    private
+
+    def valkyrie_object_factory?
+      Bulkrax.object_factory <= ValkyrieObjectFactory
+    end
   end
 end