<commit_before>require 'barbeque/docker_image'
require 'barbeque/execution_log'
require 'barbeque/runner'
require 'barbeque/slack_client'

module Barbeque
  module MessageHandler
    class JobExecution
      # @param [Barbeque::Message::JobExecution] message
      # @param [Barbeque::JobQueue] job_queue
      def initialize(message:, job_queue:)
        @message = message
        @job_queue = job_queue
      end

      def run
        job_execution = Barbeque::JobExecution.find_or_initialize_by(message_id: @message.id)
        raise DuplicatedExecution if job_execution.persisted?
        job_execution.update!(job_definition: job_definition, job_queue_id: @job_queue.id)

        stdout, stderr, status = run_command
        job_execution.update!(status: status.success? ? :success : :failed, finished_at: Time.now)
        notify_slack(job_execution, status)

        log_result(job_execution, stdout, stderr)
      end

      private

      def log_result(execution, stdout, stderr)
        log = { message: @message.body.to_json, stdout: stdout, stderr: stderr }
        Barbeque::ExecutionLog.save(execution: execution, log: log)
      end

      def notify_slack(job_execution, status)
        return if job_execution.slack_notification.nil?

        client = Barbeque::SlackClient.new(job_execution.slack_notification.channel)
        if status.success?
          if job_execution.slack_notification.notify_success
            client.notify_success("*[SUCCESS]* Succeeded to execute #{job_execution_link(job_execution)}")<commit_msg>Make message deduplication operation atomic<commit_after>require 'barbeque/docker_image'
require 'barbeque/execution_log'
require 'barbeque/runner'
require 'barbeque/slack_client'

module Barbeque
  module MessageHandler
    class JobExecution
      # @param [Barbeque::Message::JobExecution] message
      # @param [Barbeque::JobQueue] job_queue
      def initialize(message:, job_queue:)
        @message = message
        @job_queue = job_queue
      end

      def run
        begin
          job_execution = Barbeque::JobExecution.create(message_id: @message.id, job_definition: job_definition, job_queue: @job_queue)
        rescue ActiveRecord::RecordNotUnique
          raise DuplicatedExecution
        end

        stdout, stderr, status = run_command
        job_execution.update!(status: status.success? ? :success : :failed, finished_at: Time.now)
        notify_slack(job_execution, status)

        log_result(job_execution, stdout, stderr)
      end

      private

      def log_result(execution, stdout, stderr)
        log = { message: @message.body.to_json, stdout: stdout, stderr: stderr }
        Barbeque::ExecutionLog.save(execution: execution, log: log)
      end

      def notify_slack(job_execution, status)
        return if job_execution.slack_notification.nil?

        client = Barbeque::SlackClient.new(job_execution.slack_notification.channel)
        if status.success?
          if job_execution.slack_notification.notify_success
            client.notify_success("*[SUCCESS]* Succeeded to execute #{job_execution_link(job_execution)}")
{'fraction_non_alphanumeric': 0.08659397049390635, 'fraction_numerical': 0.0, 'mean_word_length': 3.5072254335260116, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1075718', 'n_tokens_mistral': 962, 'n_tokens_neox': 899, 'n_words': 187}