<commit_before>require 'stash/sword'

require 'concurrent/async'

module StashEngine
  module Sword
    # Background job for asynchronous SWORD submission
    class SubmitJob
      include Concurrent::Async

      # Creates a {SubmitJob} and submits it on a background thread, logging the result.
      #
      # @param sword_package [Package] the package to submit
      # @return [Concurrent::Ivar<Integer>] a future containing the ID of the submitted resource, or an error
      def self.submit_async(sword_package)
        title = sword_package.title
        resource_id = sword_package.resource_id
        doi = sword_package.doi
        Rails.logger.debug("Creating SubmitJob for resource #{resource_id}: '#{title}' (#{doi})")

        future = SubmitJob.new(
          title: title,
          doi: doi,
          zipfile: sword_package.zipfile,
          resource_id: resource_id,
          sword_params: sword_package.sword_params,
          request_host: sword_package.request_host,
          request_port: sword_package.request_port
        ).async.submit

        future.add_observer(FileCleanupObserver.new(resource_id: resource_id))
        future.add_observer(ResultLoggingObserver.new(title: title, doi: doi))
        future
      end

      # Creates a new {SubmitJob}.
      #
      # @param title [String] The title of the dataset being submitted
      # @param doi [String] The DOI of the dataset being submitted
      # @param zipfile [String] The local path, on the server, of the zipfile to be submitted
      # @param resource_id [Integer] The ID of the resource being submitted
      # @param sword_params [Hash] Initialization parameters for `Stash::Sword::Client`. See the
      #   [stash-sword documentation](http://www.rubydoc.info/gems/stash-sword/Stash/Sword/Client#initialize-instance_method)
      #   for details.
      # @param request_host [String] The public hostname of the application UI. Used to generate links in the
      #   notification email.
      # @param request_port [Integer] The public-facing port of the application UI. Used to generate links in the
      #   notification email.
      def initialize(title:, doi:, zipfile:, resource_id:, sword_params:, request_host:, request_port:) # rubocop:disable Metrics/ParameterLists, Metrics/LineLength
        super()
        @title = title
        @doi = doi
        @zipfile = zipfile
        @resource_id = resource_id
        @sword_params = sword_params
        @request_host = request_host
        @request_port = request_port
      end

      # Submits this job
      # @return [Integer] the ID of the submitted resource
      def submit
        sword_params_str = (sword_params.map { |k, v| "#{k}: #{v}" }).join(', ')
        request_msg = "#{self.class}: Submitting #{zipfile} for '#{title}' (#{doi}) (id: #{resource_id}) at #{Time.now} with sword_params: #{sword_params_str}"
        log.debug(request_msg)

        # Ensure we don't leak connections when running in a background thread
        ActiveRecord::Base.connection_pool.with_connection do
          begin
            do_submit(request_msg)
          rescue => e
            report_error(e, request_msg)
            raise
          end
        end
      end

      private

      attr_reader :title
      attr_reader :doi
      attr_reader :zipfile
      attr_reader :resource_id
      attr_reader :sword_params
      attr_reader :request_host
      attr_reader :request_port

      def log
        Rails.logger
      end

      def client
        @client ||= Stash::Sword::Client.new(logger: log, **sword_params)
      end

      def do_submit(request_msg)
        resource = Resource.find(resource_id)
        # TODO: collapse this into single method on resource
        resource.update_uri ? update(resource) : create(resource)
        resource.current_state = 'published'
        resource.version_zipfile = zipfile
        update_submission_log(resource_id: resource_id, request_msg: request_msg, response_msg: 'Success')
        resource_id
      end

      def report_error(e, request_msg)
        log.error("#{e}\n#{e.backtrace.join("\n")}")

        update_submission_log(resource_id: resource_id, request_msg: request_msg, response_msg: "Failed: #{e}")
        resource = Resource.find(resource_id)
        resource.current_state = 'error' if resource
<commit_msg>increasing the timeout for sword client to an hour.<commit_after>require 'stash/sword'

require 'concurrent/async'

module StashEngine
  module Sword
    # Background job for asynchronous SWORD submission
    class SubmitJob
      include Concurrent::Async

      # Creates a {SubmitJob} and submits it on a background thread, logging the result.
      #
      # @param sword_package [Package] the package to submit
      # @return [Concurrent::Ivar<Integer>] a future containing the ID of the submitted resource, or an error
      def self.submit_async(sword_package)
        title = sword_package.title
        resource_id = sword_package.resource_id
        doi = sword_package.doi
        Rails.logger.debug("Creating SubmitJob for resource #{resource_id}: '#{title}' (#{doi})")

        future = SubmitJob.new(
          title: title,
          doi: doi,
          zipfile: sword_package.zipfile,
          resource_id: resource_id,
          sword_params: sword_package.sword_params,
          request_host: sword_package.request_host,
          request_port: sword_package.request_port
        ).async.submit

        future.add_observer(FileCleanupObserver.new(resource_id: resource_id))
        future.add_observer(ResultLoggingObserver.new(title: title, doi: doi))
        future
      end

      # Creates a new {SubmitJob}.
      #
      # @param title [String] The title of the dataset being submitted
      # @param doi [String] The DOI of the dataset being submitted
      # @param zipfile [String] The local path, on the server, of the zipfile to be submitted
      # @param resource_id [Integer] The ID of the resource being submitted
      # @param sword_params [Hash] Initialization parameters for `Stash::Sword::Client`. See the
      #   [stash-sword documentation](http://www.rubydoc.info/gems/stash-sword/Stash/Sword/Client#initialize-instance_method)
      #   for details.
      # @param request_host [String] The public hostname of the application UI. Used to generate links in the
      #   notification email.
      # @param request_port [Integer] The public-facing port of the application UI. Used to generate links in the
      #   notification email.
      def initialize(title:, doi:, zipfile:, resource_id:, sword_params:, request_host:, request_port:) # rubocop:disable Metrics/ParameterLists, Metrics/LineLength
        super()
        @title = title
        @doi = doi
        @zipfile = zipfile
        @resource_id = resource_id
        @sword_params = sword_params
        @request_host = request_host
        @request_port = request_port
      end

      # Submits this job
      # @return [Integer] the ID of the submitted resource
      def submit
        sword_params_str = (sword_params.map { |k, v| "#{k}: #{v}" }).join(', ')
        request_msg = "#{self.class}: Submitting #{zipfile} for '#{title}' (#{doi}) (id: #{resource_id}) at #{Time.now} with sword_params: #{sword_params_str}"
        log.debug(request_msg)

        # Ensure we don't leak connections when running in a background thread
        ActiveRecord::Base.connection_pool.with_connection do
          begin
            do_submit(request_msg)
          rescue => e
            report_error(e, request_msg)
            raise
          end
        end
      end

      private

      attr_reader :title
      attr_reader :doi
      attr_reader :zipfile
      attr_reader :resource_id
      attr_reader :sword_params
      attr_reader :request_host
      attr_reader :request_port

      def log
        Rails.logger
      end

      def client
        if @client
          @client
        else
          @client = Stash::Sword::Client.new(logger: log, **sword_params)
          @client.helper.timeout = 3600
          @client
        end
      end

      def do_submit(request_msg)
        resource = Resource.find(resource_id)
        # TODO: collapse this into single method on resource
        resource.update_uri ? update(resource) : create(resource)
        resource.current_state = 'published'
        resource.version_zipfile = zipfile
        update_submission_log(resource_id: resource_id, request_msg: request_msg, response_msg: 'Success')
        resource_id
      end

      def report_error(e, request_msg)
        log.error("#{e}\n#{e.backtrace.join("\n")}")

        update_submission_log(resource_id: resource_id, request_msg: request_msg, response_msg: "Failed: #{e}")
        resource = Resource.find(resource_id)
        resource.current_state = 'error' if resource

{'fraction_non_alphanumeric': 0.08695652173913043, 'fraction_numerical': 0.0004528985507246377, 'mean_word_length': 2.694270179841071, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2221814', 'n_tokens_mistral': 2533, 'n_tokens_neox': 2457, 'n_words': 706}