refactor: refactors message dequeueing (#2810)

app/lib/message_dequeuer/base.rb

@@ -0,0 +1,108 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class Base
+
+    class StopProcessing < StandardError
+    end
+
+    attr_reader :queued_message
+    attr_reader :logger
+    attr_reader :state
+
+    def initialize(queued_message, logger:, state: nil)
+      @queued_message = queued_message
+      @logger = logger
+      @state = state || State.new
+    end
+
+    def process
+      raise NotImplemented
+    end
+
+    class << self
+
+      def process(message, **kwargs)
+        new(message, **kwargs).process
+      end
+
+    end
+
+    private
+
+    def stop_processing
+      raise StopProcessing
+    end
+
+    def catch_stops
+      yield if block_given?
+      true
+    rescue StopProcessing
+      false
+    end
+
+    def remove_from_queue
+      @queued_message.destroy
+    end
+
+    def create_delivery(type, **kwargs)
+      @queued_message.message.create_delivery(type, **kwargs)
+    end
+
+    def log(text, **tags)
+      logger.info text, **tags
+    end
+
+    def increment_live_stats
+      queued_message.message.database.live_stats.increment(queued_message.message.scope)
+    end
+
+    def hold_if_server_development_mode
+      return if queued_message.manual?
+      return unless queued_message.server.mode == "Development"
+
+      log "server is in development mode, holding"
+      create_delivery "Held", details: "Server is in development mode."
+      remove_from_queue
+      stop_processing
+    end
+
+    def log_sender_result
+      log_details = @result.details
+
+      if @additional_delivery_details
+        log_details += "." unless log_details =~ /\.\z/
+        log_details += " "
+        log_details += @additional_delivery_details
+      end
+
+      create_delivery @result.type, details: log_details,
+                                    output: @result.output&.strip,
+                                    sent_with_ssl: @result.secure,
+                                    log_id: @result.log_id,
+                                    time: @result.time
+    end
+
+    def handle_exception(exception)
+      log "internal error: #{exception.class}: #{exception.message}"
+      exception.backtrace.each { |line| log(line) }
+
+      queued_message.retry_later unless queued_message.destroyed?
+      log "message requeued for trying later, at #{queued_message.retry_after}"
+
+      if defined?(Sentry)
+        Sentry.capture_exception(exception, extra: {
+          server_id: queued_message.server_id,
+          queued_message_id: queued_message.message_id
+        })
+      end
+
+      queued_message.message&.create_delivery("Error",
+                                              details: "An internal error occurred while sending " \
+                                                       "this message. This message will be retried " \
+                                                       "automatically.",
+                                              output: "#{exception.class}: #{exception.message}")
+    end
+
+  end
+end

app/lib/message_dequeuer/incoming_message_processor.rb

@@ -0,0 +1,215 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class IncomingMessageProcessor < Base
+
+    attr_reader :route
+
+    def process
+      log "message is incoming"
+
+      catch_stops do
+        handle_bounces
+        increment_live_stats
+        inspect_message
+        fail_if_spam
+        hold_if_server_development_mode
+        find_route
+        hold_or_reject_spam
+        accept_mail_without_endpoints
+        hold_messages
+        bounce_messages
+        send_message_to_sender
+        send_bounce_on_hard_fail
+        log_sender_result
+        finish_processing
+      end
+    rescue StandardError => e
+      handle_exception(e)
+    end
+
+    private
+
+    def handle_bounces
+      return unless queued_message.message.bounce
+
+      log "message is a bounce"
+      original_messages = queued_message.message.original_messages
+      unless original_messages.empty?
+        queued_message.message.original_messages.each do |orig_msg|
+          queued_message.message.update(bounce_for_id: orig_msg.id, domain_id: orig_msg.domain_id)
+          create_delivery "Processed", details: "This has been detected as a bounce message for <msg:#{orig_msg.id}>."
+          orig_msg.bounce!(queued_message.message)
+          log "bounce linked with message #{orig_msg.id}"
+        end
+        remove_from_queue
+        stop_processing
+      end
+
+      # This message was sent to the return path but hasn't been matched
+      # to an original message. If we have a route for this, route it
+      # otherwise we'll drop at this point.
+      return unless queued_message.message.route_id.nil?
+
+      log "no source messages found, hard failing"
+      create_delivery "HardFail", details: "This message was a bounce but we couldn't link it with any outgoing message and there was no route for it."
+      remove_from_queue
+      stop_processing
+    end
+
+    def inspect_message
+      return if queued_message.message.inspected
+
+      log "inspecting message"
+      queued_message.message.inspect_message
+      return unless queued_message.message.inspected
+
+      is_spam = queued_message.message.spam_score > queued_message.server.spam_threshold
+      if is_spam
+        queued_message.message.update(spam: true)
+        log "message is spam (scored #{queued_message.message.spam_score}, threshold is #{queued_message.server.spam_threshold})"
+      end
+
+      queued_message.message.append_headers(
+        "X-Postal-Spam: #{queued_message.message.spam ? 'yes' : 'no'}",
+        "X-Postal-Spam-Threshold: #{queued_message.server.spam_threshold}",
+        "X-Postal-Spam-Score: #{queued_message.message.spam_score}",
+        "X-Postal-Threat: #{queued_message.message.threat ? 'yes' : 'no'}"
+      )
+      log "message inspected, headers added", spam: queued_message.message.spam?, spam_score: queued_message.message.spam_score, threat: queued_message.message.threat?
+    end
+
+    def fail_if_spam
+      return if queued_message.message.spam_score < queued_message.server.spam_failure_threshold
+
+      log "message has a spam score higher than the server's maxmimum, hard failing", server_threshold: queued_message.server.spam_failure_threshold
+      create_delivery "HardFail",
+                      details: "Message's spam score is higher than the failure threshold for this server. " \
+                               "Threshold is currently #{queued_message.server.spam_failure_threshold}."
+      remove_from_queue
+      stop_processing
+    end
+
+    def find_route
+      @route = queued_message.message.route
+      return if @route
+
+      log "no route and/or endpoint available for processing, hard failing"
+      create_delivery "HardFail", details: "Message does not have a route and/or endpoint available for delivery."
+      remove_from_queue
+      stop_processing
+    end
+
+    def hold_or_reject_spam
+      return unless queued_message.message.spam
+      return if queued_message.manual?
+
+      case @route.spam_mode
+      when "Quarantine"
+        log "message is spam and route says to quarantine spam message, holding"
+        create_delivery "Held", details: "Message placed into quarantine."
+      when "Fail"
+        log "message is spam and route says to fail spam message, hard failing"
+        create_delivery "HardFail", details: "Message is spam and the route specifies it should be failed."
+      else
+        return
+      end
+
+      remove_from_queue
+      stop_processing
+    end
+
+    def accept_mail_without_endpoints
+      return unless @route.mode == "Accept"
+
+      log "route says to accept without endpoint, marking as processed"
+      create_delivery "Processed", details: "Message has been accepted but not sent to any endpoints."
+      remove_from_queue
+      stop_processing
+    end
+
+    def hold_messages
+      return unless @route.mode == "Hold"
+
+      if queued_message.manual?
+        log "route says to hold and message was queued manually, marking as processed"
+        create_delivery "Processed", details: "Message has been processed."
+      else
+        log "route says to hold, marking as held"
+        create_delivery "Held", details: "Message has been accepted but not sent to any endpoints."
+      end
+
+      remove_from_queue
+      stop_processing
+    end
+
+    def bounce_messages
+      return unless route.mode == "Bounce" || route.mode == "Reject"
+
+      log "route says to bounce, hard failing and sending bounce"
+
+      if id = queued_message.send_bounce
+        log "bounce sent with id #{id}"
+        create_delivery "HardFail", details: "Message has been bounced because the route asks for this. See message <msg:#{id}>"
+      end
+
+      remove_from_queue
+      stop_processing
+    end
+
+    def send_message_to_sender
+      @result = @state.send_result
+      return if @result
+
+      case queued_message.message.endpoint
+      when SMTPEndpoint
+        sender = @state.sender_for(Postal::SMTPSender, queued_message.message.recipient_domain, nil, servers: [queued_message.message.endpoint])
+      when HTTPEndpoint
+        sender = @state.sender_for(Postal::HTTPSender, queued_message.message.endpoint)
+      when AddressEndpoint
+        sender = @state.sender_for(Postal::SMTPSender, queued_message.message.endpoint.domain, nil, force_rcpt_to: queued_message.message.endpoint.address)
+      else
+        log "invalid endpoint for route (#{queued_message.message.endpoint_type})"
+        create_delivery "HardFail", details: "Invalid endpoint for route."
+        remove_from_queue
+        stop_processing
+      end
+
+      @result = sender.send_message(queued_message.message)
+      return unless @result.connect_error
+
+      @state.send_result = @result
+    end
+
+    def send_bounce_on_hard_fail
+      return unless @result.type == "HardFail"
+
+      if @result.suppress_bounce
+        log "suppressing bounce message after hard fail"
+        return
+      end
+
+      return unless queued_message.message.send_bounces?
+
+      log "sending a bounce because message hard failed"
+      return unless bounce_id = queued_message.send_bounce
+
+      @additional_delivery_details = "Sent bounce message to sender (see message <msg:#{bounce_id}>)"
+    end
+
+    def finish_processing
+      if @result.retry
+        queued_message.retry_later(@result.retry.is_a?(Integer) ? @result.retry : nil)
+        log "message requeued for trying later, at #{queued_message.retry_after}"
+        queued_message.allocate_ip_address
+        queued_message.update_column(:ip_address_id, queued_message.ip_address&.id)
+        stop_processing
+      end
+
+      log "message processing completed"
+      queued_message.message.endpoint.mark_as_used
+      remove_from_queue
+    end
+
+  end
+end

app/lib/message_dequeuer/initial_processor.rb

@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class InitialProcessor < Base
+
+    attr_accessor :send_result
+
+    def process
+      logger.tagged(original_queued_message: @queued_message.id) do
+        logger.info "starting message unqueue"
+        begin
+          catch_stops do
+            check_message_exists
+            check_message_is_ready
+            find_other_messages_for_batch
+
+            # Process the original message and then all of those
+            # found for batching.
+            process_message(@queued_message)
+            @other_messages.each { |message| process_message(message) }
+          end
+        ensure
+          @state.finished
+        end
+        logger.info "finished message unqueue"
+      end
+    end
+
+    private
+
+    def check_message_exists
+      @queued_message.message
+    rescue Postal::MessageDB::Message::NotFound
+      log "unqueue because backend message has been removed."
+      remove_from_queue
+      stop_processing
+    end
+
+    def check_message_is_ready
+      return if @queued_message.ready?
+
+      log "skipping because message isn't ready for processing"
+      @queued_message.unlock
+      stop_processing
+    end
+
+    def find_other_messages_for_batch
+      @other_messages = @queued_message.batchable_messages(100)
+      log "found #{@other_messages.size} associated messages to process at the same time", batch_key: @queued_message.batch_key
+    rescue StandardError
+      @queued_message.unlock
+      raise
+    end
+
+    def process_message(queued_message)
+      logger.tagged(queued_message: queued_message.id) do
+        SingleMessageProcessor.process(queued_message, logger: @logger, state: @state)
+      end
+    end
+
+  end
+end

app/lib/message_dequeuer/outgoing_message_processor.rb

@@ -0,0 +1,190 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class OutgoingMessageProcessor < Base
+
+    def process
+      catch_stops do
+        check_domain
+        check_rcpt_to
+        add_tag
+        hold_if_credential_is_set_to_hold
+        hold_if_recipient_on_suppression_list
+        parse_content
+        inspect_message
+        fail_if_spam
+        add_outgoing_headers
+        check_send_limits
+        increment_live_stats
+        hold_if_server_development_mode
+        send_message_to_sender
+        add_recipient_to_suppression_list_on_too_many_hard_fails
+        remove_recipient_from_suppression_list_on_success
+        log_sender_result
+        finish_processing
+      end
+    rescue StandardError => e
+      handle_exception(e)
+    end
+
+    private
+
+    def check_domain
+      return if queued_message.message.domain
+
+      log "message has no domain, hard failing"
+      create_delivery "HardFail", details: "Message's domain no longer exist"
+      remove_from_queue
+      stop_processing
+    end
+
+    def check_rcpt_to
+      return unless queued_message.message.rcpt_to.blank?
+
+      log "message has no 'to' address, hard failing"
+      create_delivery "HardFail", details: "Message doesn't have an RCPT to"
+      remove_from_queue
+      stop_processing
+    end
+
+    def add_tag
+      return if queued_message.message.tag
+      return unless tag = queued_message.message.headers["x-postal-tag"]
+
+      log "added tag: #{tag.last}"
+      queued_message.message.update(tag: tag.last)
+    end
+
+    def hold_if_credential_is_set_to_hold
+      return if queued_message.manual?
+      return if queued_message.message.credential.nil?
+      return unless queued_message.message.credential.hold?
+
+      log "credential wants us to hold messages, holding"
+      create_delivery "Held", details: "Credential is configured to hold all messages authenticated by it."
+      remove_from_queue
+      stop_processing
+    end
+
+    def hold_if_recipient_on_suppression_list
+      return if queued_message.manual?
+      return unless sl = queued_message.server.message_db.suppression_list.get(:recipient, queued_message.message.rcpt_to)
+
+      log "recipient is on the suppression list, holding"
+      create_delivery "Held", details: "Recipient (#{queued_message.message.rcpt_to}) is on the suppression list (reason: #{sl['reason']})"
+      remove_from_queue
+      stop_processing
+    end
+
+    def parse_content
+      return unless queued_message.message.should_parse?
+
+      log "parsing message content as it hasn't been parsed before"
+      queued_message.message.parse_content
+    end
+
+    def inspect_message
+      return if queued_message.message.inspected
+      return unless queued_message.server.outbound_spam_threshold
+
+      log "inspecting message"
+      queued_message.message.inspect_message
+      return unless queued_message.message.inspected
+
+      if queued_message.message.spam_score >= queued_message.server.outbound_spam_threshold
+        queued_message.message.update(spam: true)
+      end
+
+      log "message inspected successfully", spam: queued_message.message.spam?, spam_score: queued_message.message.spam_score
+    end
+
+    def fail_if_spam
+      return unless queued_message.message.spam
+
+      log "message is spam (#{queued_message.message.spam_score}), hard failing", server_threshold: queued_message.server.outbound_spam_threshold
+      create_delivery "HardFail",
+                      details: "Message is likely spam. Threshold is #{queued_message.server.outbound_spam_threshold} and " \
+                               "the message scored #{queued_message.message.spam_score}."
+      remove_from_queue
+      stop_processing
+    end
+
+    def add_outgoing_headers
+      return if queued_message.message.has_outgoing_headers?
+
+      queued_message.message.add_outgoing_headers
+    end
+
+    def check_send_limits
+      if queued_message.server.send_limit_exceeded?
+        # If we're over the limit, we're going to be holding this message
+        log "server send limit has been exceeded, holding", send_limit: queued_message.server.send_limit
+        queued_message.server.update_columns(send_limit_exceeded_at: Time.now, send_limit_approaching_at: nil)
+        create_delivery "Held", details: "Message held because send limit (#{queued_message.server.send_limit}) has been reached."
+        remove_from_queue
+        stop_processing
+      elsif queued_message.server.send_limit_approaching?
+        # If we're approaching the limit, just say we are but continue to process the message
+        queued_message.server.update_columns(send_limit_approaching_at: Time.now, send_limit_exceeded_at: nil)
+      else
+        queued_message.server.update_columns(send_limit_approaching_at: nil, send_limit_exceeded_at: nil)
+      end
+    end
+
+    def send_message_to_sender
+      @result = @state.send_result
+      return if @result
+
+      sender = @state.sender_for(Postal::SMTPSender,
+                                 queued_message.message.recipient_domain,
+                                 queued_message.ip_address)
+
+      @result = sender.send_message(queued_message.message)
+      return unless @result.connect_error
+
+      @state.send_result = @result
+    end
+
+    def add_recipient_to_suppression_list_on_too_many_hard_fails
+      return unless @result.type == "HardFail"
+
+      recent_hard_fails = queued_message.server.message_db.select(:messages,
+                                                                  where: {
+                                                                    rcpt_to: queued_message.message.rcpt_to,
+                                                                    status: "HardFail",
+                                                                    timestamp: { greater_than: 24.hours.ago.to_f }
+                                                                  },
+                                                                  count: true)
+      return if recent_hard_fails < 1
+
+      added = queued_message.server.message_db.suppression_list.add(:recipient, queued_message.message.rcpt_to,
+                                                                    reason: "too many hard fails")
+      return unless added
+
+      log "Added #{queued_message.message.rcpt_to} to suppression list because #{recent_hard_fails} hard fails in 24 hours"
+      @additional_delivery_details = "Recipient added to suppression list (too many hard fails)"
+    end
+
+    def remove_recipient_from_suppression_list_on_success
+      return unless @result.type == "Sent"
+
+      removed = queued_message.server.message_db.suppression_list.remove(:recipient, queued_message.message.rcpt_to)
+      return unless removed
+
+      log "removed #{queued_message.message.rcpt_to} from suppression list"
+      @additional_delivery_details = "Recipient removed from suppression list"
+    end
+
+    def finish_processing
+      if @result.retry
+        queued_message.retry_later(@result.retry.is_a?(Integer) ? @result.retry : nil)
+        log "message requeued for trying later", retry_after: queued_message.retry_after
+        stop_processing
+      end
+
+      log "message processing complete"
+      remove_from_queue
+    end
+
+  end
+end

app/lib/message_dequeuer/single_message_processor.rb

@@ -0,0 +1,83 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class SingleMessageProcessor < Base
+
+    def process
+      catch_stops do
+        check_message_exists
+        check_server_suspension
+        check_delivery_attempts
+        check_raw_message_exists
+
+        processor = nil
+        case queued_message.message.scope
+        when "incoming"
+          processor = IncomingMessageProcessor
+        when "outgoing"
+          processor = OutgoingMessageProcessor
+        else
+          create_delivery "HardFail", details: "Scope #{queued_message.message.scope} is not valid"
+          remove_from_queue
+          stop_processing
+        end
+
+        processor.process(queued_message, logger: @logger, state: @state)
+      end
+    rescue StandardError => e
+      handle_exception(e)
+    end
+
+    private
+
+    def check_message_exists
+      queued_message.message
+    rescue Postal::MessageDB::Message::NotFound
+      log "unqueueing because backend message has been removed"
+      remove_from_queue
+      stop_processing
+    end
+
+    def check_server_suspension
+      return unless queued_message.server.suspended?
+
+      log "server is suspended, holding message"
+      create_delivery "Held", details: "Mail server has been suspended. No e-mails can be processed at present. Contact support for assistance."
+      remove_from_queue
+      stop_processing
+    end
+
+    def check_delivery_attempts
+      return if queued_message.attempts < Postal.config.general.maximum_delivery_attempts
+
+      details = "Maximum number of delivery attempts (#{queued_message.attempts}) has been reached."
+      if queued_message.message.scope == "incoming"
+        # Send bounces to incoming e-mails when they are hard failed
+        if bounce_id = queued_message.send_bounce
+          details += " Bounce sent to sender (see message <msg:#{bounce_id}>)"
+        end
+      elsif queued_message.message.scope == "outgoing"
+        # Add the recipient to the suppression list
+        if queued_message.server.message_db.suppression_list.add(:recipient, queued_message.message.rcpt_to, reason: "too many soft fails")
+          log "added #{queued_message.message.rcpt_to} to suppression list because maximum attempts has been reached"
+          details += " Added #{queued_message.message.rcpt_to} to suppression list because delivery has failed #{queued_message.attempts} times."
+        end
+      end
+
+      log "message has reached maximum number of attempts, hard failing"
+      create_delivery "HardFail", details: details
+      remove_from_queue
+      stop_processing
+    end
+
+    def check_raw_message_exists
+      return if queued_message.message.raw_message?
+
+      log "raw message has been removed, not sending"
+      create_delivery "HardFail", details: "Raw message has been removed. Cannot send message."
+      remove_from_queue
+      stop_processing
+    end
+
+  end
+end

app/lib/message_dequeuer/state.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+  class State
+
+    attr_accessor :send_result
+
+    def sender_for(klass, *args)
+      @cached_senders ||= {}
+      @cached_senders[[klass, args]] ||= begin
+        klass_instance = klass.new(*args)
+        klass_instance.start
+        klass_instance
+      end
+    end
+
+    def finished
+      @cached_senders&.each_value do |sender|
+        sender.finish
+      rescue StandardError
+        false
+      end
+    end
+
+  end
+end

app/services/unqueue_message_service.rb

@@ -1,493 +0,0 @@
-# frozen_string_literal: true
-
-class UnqueueMessageService
-
-  def initialize(queued_message:, logger:)
-    @queued_message = queued_message
-    @logger = logger
-  end
-
-  def call
-    @logger.tagged(original_queued_message: @queued_message.id) do
-      log "starting message unqueue"
-      process_original_message
-      log "finished message unqueue"
-    end
-  end
-
-  private
-
-  def process_original_message
-    begin
-      @queued_message.message
-    rescue Postal::MessageDB::Message::NotFound
-      log "unqueue because backend message has been removed."
-      @queued_message.destroy
-      return
-    end
-
-    unless @queued_message.ready?
-      log "skipping because message isn't ready for processing"
-      return
-    end
-
-    begin
-      other_messages = @queued_message.batchable_messages(100)
-      log "found #{other_messages.size} associated messages to process at the same time", batch_key: @queued_message.batch_key
-    rescue StandardError
-      @queued_message.unlock
-      raise
-    end
-
-    ([@queued_message] + other_messages).each do |queued_message|
-      @logger.tagged(queued_message: queued_message.id) do
-        process_message(queued_message)
-      end
-    end
-  ensure
-    begin
-      @sender&.finish
-    rescue StandardError
-      nil
-    end
-  end
-
-  # rubocop:disable Naming/MemoizedInstanceVariableName
-  def cached_sender(klass, *args)
-    @sender ||= begin
-      sender = klass.new(*args)
-      sender.start
-      sender
-    end
-  end
-  # rubocop:enable Naming/MemoizedInstanceVariableName
-
-  def log(message, **tags)
-    @logger.info(message, **tags)
-  end
-
-  def process_message(queued_message)
-    begin
-      queued_message.message
-    rescue Postal::MessageDB::Message::NotFound
-      log "unqueueing because backend message has been removed"
-      queued_message.destroy
-      return
-    end
-
-    log "processing message"
-
-    #
-    # If the server is suspended, hold all messages
-    #
-    if queued_message.server.suspended?
-      log "server is suspended, holding message"
-      queued_message.message.create_delivery("Held", details: "Mail server has been suspended. No e-mails can be processed at present. Contact support for assistance.")
-      queued_message.destroy
-      return
-    end
-
-    # We might not be able to send this any more, check the attempts
-    if queued_message.attempts >= Postal.config.general.maximum_delivery_attempts
-      details = "Maximum number of delivery attempts (#{queued_message.attempts}) has been reached."
-      if queued_message.message.scope == "incoming"
-        # Send bounces to incoming e-mails when they are hard failed
-        if bounce_id = queued_message.send_bounce
-          details += " Bounce sent to sender (see message <msg:#{bounce_id}>)"
-        end
-      elsif queued_message.message.scope == "outgoing"
-        # Add the recipient to the suppression list
-        if queued_message.server.message_db.suppression_list.add(:recipient, queued_message.message.rcpt_to, reason: "too many soft fails")
-          log "added #{queued_message.message.rcpt_to} to suppression list because maximum attempts has been reached"
-          details += " Added #{queued_message.message.rcpt_to} to suppression list because delivery has failed #{queued_message.attempts} times."
-        end
-      end
-      queued_message.message.create_delivery("HardFail", details: details)
-      queued_message.destroy
-      log "message has reached maximum number of attempts, hard failing"
-      return
-    end
-
-    # If the raw message has been removed (removed by retention)
-    unless queued_message.message.raw_message?
-      log "raw message has been removed, not sending"
-      queued_message.message.create_delivery("HardFail", details: "Raw message has been removed. Cannot send message.")
-      queued_message.destroy
-      return
-    end
-
-    #
-    # Handle Incoming Messages
-    #
-    if queued_message.message.scope == "incoming"
-      log "message is incoming"
-
-      #
-      # If this is a bounce, we need to handle it as such
-      #
-      if queued_message.message.bounce
-        log "message is a bounce"
-        original_messages = queued_message.message.original_messages
-        unless original_messages.empty?
-          queued_message.message.original_messages.each do |orig_msg|
-            queued_message.message.update(bounce_for_id: orig_msg.id, domain_id: orig_msg.domain_id)
-            queued_message.message.create_delivery("Processed", details: "This has been detected as a bounce message for <msg:#{orig_msg.id}>.")
-            orig_msg.bounce!(queued_message.message)
-            log "bounce linked with message #{orig_msg.id}"
-          end
-          queued_message.destroy
-          return
-        end
-
-        # This message was sent to the return path but hasn't been matched
-        # to an original message. If we have a route for this, route it
-        # otherwise we'll drop at this point.
-        if queued_message.message.route_id.nil?
-          log "no source messages found, hard failing"
-          queued_message.message.create_delivery("HardFail", details: "This message was a bounce but we couldn't link it with any outgoing message and there was no route for it.")
-          queued_message.destroy
-          return
-        end
-      end
-
-      #
-      # Update live stats
-      #
-      queued_message.message.database.live_stats.increment(queued_message.message.scope)
-
-      #
-      # Inspect incoming messages
-      #
-      unless queued_message.message.inspected
-        log "inspecting message"
-        queued_message.message.inspect_message
-        if queued_message.message.inspected
-          is_spam = queued_message.message.spam_score > queued_message.server.spam_threshold
-          if is_spam
-            queued_message.message.update(spam: true)
-            log "message is spam (scored #{queued_message.message.spam_score}, threshold is #{queued_message.server.spam_threshold})"
-          end
-          queued_message.message.append_headers(
-            "X-Postal-Spam: #{queued_message.message.spam ? 'yes' : 'no'}",
-            "X-Postal-Spam-Threshold: #{queued_message.server.spam_threshold}",
-            "X-Postal-Spam-Score: #{queued_message.message.spam_score}",
-            "X-Postal-Threat: #{queued_message.message.threat ? 'yes' : 'no'}"
-          )
-          log "message inspected, headers added", spam: queued_message.message.spam?, spam_score: queued_message.message.spam_score, threat: queued_message.message.threat?
-        end
-      end
-
-      #
-      # If this message has a SPAM score higher than is permitted
-      #
-      if queued_message.message.spam_score >= queued_message.server.spam_failure_threshold
-        log "message has a spam score higher than the server's maxmimum, hard failing", server_threshold: queued_message.server.spam_failure_threshold
-        queued_message.message.create_delivery("HardFail",
-                                               details: "Message's spam score is higher than the failure threshold for this server. " \
-                                                        "Threshold is currently #{queued_message.server.spam_failure_threshold}.")
-        queued_message.destroy
-        return
-      end
-
-      # If the server is in development mode, hold it
-      if queued_message.server.mode == "Development" && !queued_message.manual?
-        log "server is in development mode, holding"
-        queued_message.message.create_delivery("Held", details: "Server is in development mode.")
-        queued_message.destroy
-        return
-      end
-
-      #
-      # Find out what sort of message we're supposed to be sending and dispatch this request over to
-      # the sender.
-      #
-      if route = queued_message.message.route
-
-        # If the route says we're holding quananteed mail and this is spam, we'll hold this
-        if route.spam_mode == "Quarantine" && queued_message.message.spam && !queued_message.manual?
-          log "message is spam and route says to quarantine spam message, holding"
-          queued_message.message.create_delivery("Held", details: "Message placed into quarantine.")
-          queued_message.destroy
-          return
-        end
-
-        # If the route says we're holding quananteed mail and this is spam, we'll hold this
-        if route.spam_mode == "Fail" && queued_message.message.spam && !queued_message.manual?
-          log "message is spam and route says to fail spam message, hard failing"
-          queued_message.message.create_delivery("HardFail", details: "Message is spam and the route specifies it should be failed.")
-          queued_message.destroy
-          return
-        end
-
-        #
-        # Messages that should be blindly accepted are blindly accepted
-        #
-        if route.mode == "Accept"
-          log "route says to accept without endpoint, marking as processed"
-          queued_message.message.create_delivery("Processed", details: "Message has been accepted but not sent to any endpoints.")
-          queued_message.destroy
-          return
-        end
-
-        #
-        # Messages that should be accepted and held should be held
-        #
-        if route.mode == "Hold"
-          if queued_message.manual?
-            log "route says to hold and message was queued manually, marking as processed"
-            queued_message.message.create_delivery("Processed", details: "Message has been processed.")
-          else
-            log "route says to hold, marking as held"
-            queued_message.message.create_delivery("Held", details: "Message has been accepted but not sent to any endpoints.")
-          end
-          queued_message.destroy
-          return
-        end
-
-        #
-        # Messages that should be bounced should be bounced (or rejected if they got this far)
-        #
-        if route.mode == "Bounce" || route.mode == "Reject"
-          log "route says to bounce, hard failing and sending bounce"
-          if id = queued_message.send_bounce
-            log "bounce sent with id #{id}"
-            queued_message.message.create_delivery("HardFail", details: "Message has been bounced because the route asks for this. See message <msg:#{id}>")
-          end
-          queued_message.destroy
-          return
-        end
-
-        if @fixed_result
-          result = @fixed_result
-        else
-          case queued_message.message.endpoint
-          when SMTPEndpoint
-            sender = cached_sender(Postal::SMTPSender, queued_message.message.recipient_domain, nil, servers: [queued_message.message.endpoint])
-          when HTTPEndpoint
-            sender = cached_sender(Postal::HTTPSender, queued_message.message.endpoint)
-          when AddressEndpoint
-            sender = cached_sender(Postal::SMTPSender, queued_message.message.endpoint.domain, nil, force_rcpt_to: queued_message.message.endpoint.address)
-          else
-            log "invalid endpoint for route (#{queued_message.message.endpoint_type})"
-            queued_message.message.create_delivery("HardFail", details: "Invalid endpoint for route.")
-            queued_message.destroy
-            return
-          end
-          result = sender.send_message(queued_message.message)
-          if result.connect_error
-            @fixed_result = result
-          end
-        end
-
-        # Log the result
-        log_details = result.details
-        if result.type == "HardFail" && result.suppress_bounce
-          # The delivery hard failed, but requested that no bounce be sent
-          log "suppressing bounce message after hard fail"
-        elsif result.type == "HardFail" && queued_message.message.send_bounces?
-          # If the message is a hard fail, send a bounce message for this message.
-          log "sending a bounce because message hard failed"
-          if bounce_id = queued_message.send_bounce
-            log_details += "." unless log_details =~ /\.\z/
-            log_details += " Sent bounce message to sender (see message <msg:#{bounce_id}>)"
-          end
-        end
-
-        queued_message.message.create_delivery(result.type, details: log_details, output: result.output&.strip, sent_with_ssl: result.secure, log_id: result.log_id, time: result.time)
-
-        if result.retry
-          queued_message.retry_later(result.retry.is_a?(Integer) ? result.retry : nil)
-          log "message requeued for trying later, at #{queued_message.retry_after}"
-          queued_message.allocate_ip_address
-          queued_message.update_column(:ip_address_id, queued_message.ip_address&.id)
-        else
-          log "message processing completed"
-          queued_message.message.endpoint.mark_as_used
-          queued_message.destroy
-        end
-      else
-        log "no route and/or endpoint available for processing, hard failing"
-        queued_message.message.create_delivery("HardFail", details: "Message does not have a route and/or endpoint available for delivery.")
-        queued_message.destroy
-        return
-      end
-    end
-
-    #
-    # Handle Outgoing Messages
-    #
-    return unless queued_message.message.scope == "outgoing"
-
-    log "message is outgoing"
-
-    if queued_message.message.domain.nil?
-      log "message has no domain, hard failing"
-      queued_message.message.create_delivery("HardFail", details: "Message's domain no longer exist")
-      queued_message.destroy
-      return
-    end
-
-    #
-    # If there's no to address, we can't do much. Fail it.
-    #
-    if queued_message.message.rcpt_to.blank?
-      log "message has no 'to' address, hard failing"
-      queued_message.message.create_delivery("HardFail", details: "Message doesn't have an RCPT to")
-      queued_message.destroy
-      return
-    end
-
-    # Extract a tag and add it to the message if one doesn't exist
-    if queued_message.message.tag.nil? && tag = queued_message.message.headers["x-postal-tag"]
-      log "added tag: #{tag.last}"
-      queued_message.message.update(tag: tag.last)
-    end
-
-    #
-    # If the credentials for this message is marked as holding and this isn't manual, hold it
-    #
-    if !queued_message.manual? && queued_message.message.credential && queued_message.message.credential.hold?
-      log "credential wants us to hold messages, holding"
-      queued_message.message.create_delivery("Held", details: "Credential is configured to hold all messages authenticated by it.")
-      queued_message.destroy
-      return
-    end
-
-    #
-    # If the recipient is on the suppression list and this isn't a manual queueing block sending
-    #
-    if !queued_message.manual? && sl = queued_message.server.message_db.suppression_list.get(:recipient, queued_message.message.rcpt_to)
-      log "recipient is on the suppression list, holding"
-      queued_message.message.create_delivery("Held", details: "Recipient (#{queued_message.message.rcpt_to}) is on the suppression list (reason: #{sl['reason']})")
-      queued_message.destroy
-      return
-    end
-
-    # Parse the content of the message as appropriate
-    if queued_message.message.should_parse?
-      log "parsing message content as it hasn't been parsed before"
-      queued_message.message.parse_content
-    end
-
-    # Inspect outgoing messages when there's a threshold set for the server
-    if !queued_message.message.inspected && queued_message.server.outbound_spam_threshold
-      log "inspecting message"
-      queued_message.message.inspect_message
-      if queued_message.message.inspected
-        if queued_message.message.spam_score >= queued_message.server.outbound_spam_threshold
-          queued_message.message.update(spam: true)
-        end
-        log "message inspected successfully", spam: queued_message.message.spam?, spam_score: queued_message.message.spam_score
-      end
-    end
-
-    if queued_message.message.spam
-      log "message is spam (#{queued_message.message.spam_score}), hard failing", server_threshold: queued_message.server.outbound_spam_threshold
-      queued_message.message.create_delivery("HardFail",
-                                             details: "Message is likely spam. Threshold is #{queued_message.server.outbound_spam_threshold} and " \
-                                                      "the message scored #{queued_message.message.spam_score}.")
-      queued_message.destroy
-      return
-    end
-
-    # Add outgoing headers
-    unless queued_message.message.has_outgoing_headers?
-      queued_message.message.add_outgoing_headers
-    end
-
-    # Check send limits
-    if queued_message.server.send_limit_exceeded?
-      # If we're over the limit, we're going to be holding this message
-      log "server send limit has been exceeded, holding", send_limit: queued_message.server.send_limit
-      queued_message.server.update_columns(send_limit_exceeded_at: Time.now, send_limit_approaching_at: nil)
-      queued_message.message.create_delivery("Held", details: "Message held because send limit (#{queued_message.server.send_limit}) has been reached.")
-      queued_message.destroy
-      return
-    elsif queued_message.server.send_limit_approaching?
-      # If we're approaching the limit, just say we are but continue to process the message
-      queued_message.server.update_columns(send_limit_approaching_at: Time.now, send_limit_exceeded_at: nil)
-    else
-      queued_message.server.update_columns(send_limit_approaching_at: nil, send_limit_exceeded_at: nil)
-    end
-
-    # Update the live stats for this message.
-    queued_message.message.database.live_stats.increment(queued_message.message.scope)
-
-    # If the server is in development mode, hold it
-    if queued_message.server.mode == "Development" && !queued_message.manual?
-      log "server is in development mode, holding"
-      queued_message.message.create_delivery("Held", details: "Server is in development mode.")
-      queued_message.destroy
-      return
-    end
-
-    # Send the outgoing message to the SMTP sender
-
-    if @fixed_result
-      result = @fixed_result
-    else
-      sender = cached_sender(Postal::SMTPSender, queued_message.message.recipient_domain, queued_message.ip_address)
-      result = sender.send_message(queued_message.message)
-      if result.connect_error
-        @fixed_result = result
-      end
-    end
-
-    #
-    # If the message has been hard failed, check to see how many other recent hard fails we've had for the address
-    # and if there are more than 2, suppress the address for 30 days.
-    #
-    if result.type == "HardFail"
-      recent_hard_fails = queued_message.server.message_db.select(:messages,
-                                                                  where: {
-                                                                    rcpt_to: queued_message.message.rcpt_to,
-                                                                    status: "HardFail",
-                                                                    timestamp: { greater_than: 24.hours.ago.to_f }
-                                                                  },
-                                                                  count: true)
-      if recent_hard_fails >= 1 && queued_message.server.message_db.suppression_list.add(:recipient, queued_message.message.rcpt_to, reason: "too many hard fails")
-        log "Added #{queued_message.message.rcpt_to} to suppression list because #{recent_hard_fails} hard fails in 24 hours"
-        result.details += "." if result.details =~ /\.\z/
-        result.details += " " if result.details.present?
-        result.details += "Recipient added to suppression list (too many hard fails)."
-      end
-    end
-
-    #
-    # If a message is sent successfully, remove the users from the suppression list
-    #
-    if result.type == "Sent" && queued_message.server.message_db.suppression_list.remove(:recipient, queued_message.message.rcpt_to)
-      log "removed #{queued_message.message.rcpt_to} from suppression list"
-      result.details += "." if result.details =~ /\.\z/
-      result.details += " Recipient removed from suppression list."
-    end
-
-    # Log the result
-    queued_message.message.create_delivery(result.type, details: result.details, output: result.output, sent_with_ssl: result.secure, log_id: result.log_id, time: result.time)
-
-    if result.retry
-      queued_message.retry_later(result.retry.is_a?(Integer) ? result.retry : nil)
-      log "message requeued for trying later", retry_after: queued_message.retry_after
-    else
-      log "message processing complete"
-      queued_message.destroy
-    end
-  rescue StandardError => e
-    log "internal error: #{e.class}: #{e.message}"
-    e.backtrace.each { |line| log(line) }
-
-    queued_message.retry_later
-    log "message requeued for trying later, at #{queued_message.retry_after}"
-
-    if defined?(Sentry)
-      Sentry.capture_exception(e, extra: { server_id: queued_message.server_id, queued_message_id: queued_message.message_id })
-    end
-
-    queued_message.message&.create_delivery("Error",
-                                            details: "An internal error occurred while sending " \
-                                                     "this message. This message will be retried " \
-                                                     "automatically.",
-                                            output: "#{e.class}: #{e.message}", log_id: "J-#{id}")
-  end
-
-end

app/util/message_dequeuer.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+module MessageDequeuer
+
+  class << self
+
+    def process(message, logger:)
+      processor = InitialProcessor.new(message, logger: logger)
+      processor.process
+    end
+
+  end
+
+end

lib/worker/jobs/process_queued_messages_job.rb

@@ -64,7 +64,7 @@ module Worker
       def process_messages
         @messages_to_process.each do |message|
           work_completed!
-          UnqueueMessageService.new(queued_message: message, logger: logger).call
+          MessageDequeuer.process(message, logger: logger)
         end
       end
 

spec/lib/message_dequeuer/base_spec.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+module MessageDequeuer
+
+  RSpec.describe Base do
+    describe ".new" do
+      context "when given state" do
+        it "uses that state" do
+          base = described_class.new(nil, logger: nil, state: 1234)
+          expect(base.state).to eq 1234
+        end
+      end
+
+      context "when not given state" do
+        it "creates a new state" do
+          base = described_class.new(nil, logger: nil)
+          expect(base.state).to be_a State
+        end
+      end
+    end
+
+    describe ".process" do
+      it "creates a new instances of the class and calls process" do
+        message = create(:queued_message)
+        logger = TestLogger.new
+
+        mock = double("Base")
+        expect(mock).to receive(:process).once
+        expect(described_class).to receive(:new).with(message, logger: logger).and_return(mock)
+
+        described_class.process(message, logger: logger)
+      end
+    end
+  end
+
+end

spec/lib/message_dequeuer/incoming_message_processor_spec.rb

@@ -0,0 +1,640 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+module MessageDequeuer
+
+  RSpec.describe IncomingMessageProcessor do
+    let(:server) { create(:server) }
+    let(:state) { State.new }
+    let(:logger) { TestLogger.new }
+    let(:route) { create(:route, server: server) }
+    let(:message) { MessageFactory.incoming(server, route: route) }
+    let(:queued_message) { create(:queued_message, :locked, message: message) }
+
+    subject(:processor) { described_class.new(queued_message, logger: logger, state: state) }
+
+    context "when the message was a bounce but there's no return path for it" do
+      let(:message) do
+        MessageFactory.incoming(server) do |msg|
+          msg.bounce = true
+        end
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/no source messages found, hard failing/)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /was a bounce but we couldn't link it with any outgoing message/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the message is a bounce for an existing message" do
+      let(:existing_message) { MessageFactory.outgoing(server) }
+
+      let(:message) do
+        MessageFactory.incoming(server) do |msg, mail|
+          msg.bounce = true
+          mail["X-Postal-MsgID"] = existing_message.token
+        end
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/message is a bounce/)
+      end
+
+      it "adds the original message as the bounce ID for the received message" do
+        processor.process
+        expect(message.reload.bounce_for_id).to eq existing_message.id
+      end
+
+      it "sets the received message status to Processed" do
+        processor.process
+        expect(message.reload.status).to eq "Processed"
+      end
+
+      it "creates a Processed delivery on the received message" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Processed", details: /This has been detected as a bounce message for <msg:#{existing_message.id}>/i)
+      end
+
+      it "sets the existing message status to Bounced" do
+        processor.process
+        expect(existing_message.reload.status).to eq "Bounced"
+      end
+
+      it "creates a Bounced delivery on the original message" do
+        processor.process
+        delivery = existing_message.deliveries.last
+        expect(delivery).to have_attributes(status: "Bounced", details: /received a bounce message for this e-mail. See <msg:#{message.id}> for/i)
+      end
+
+      it "triggers a MessageBounced webhook event" do
+        expect(WebhookRequest).to receive(:trigger).with(server, "MessageBounced", {
+          original_message: kind_of(Hash),
+          bounce: kind_of(Hash)
+        })
+        processor.process
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the message is not a bounce" do
+      it "increments the stats for the server" do
+        expect { processor.process }.to change { server.message_db.live_stats.total(5) }.by(1)
+      end
+
+      it "inspects the message and adds headers" do
+        expect { processor.process }.to change { message.reload.inspected }.from(false).to(true)
+        new_message = message.reload
+        expect(new_message.headers).to match hash_including(
+          "x-postal-spam" => ["no"],
+          "x-postal-spam-threshold" => ["5.0"],
+          "x-postal-threat" => ["no"]
+        )
+      end
+
+      it "marks the message as spam if the spam score is higher than the server threshold" do
+        inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
+        allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
+        processor.process
+        expect(message.reload.spam).to be true
+      end
+    end
+
+    context "when the message has a spam score greater than the server's spam failure threshold" do
+      before do
+        inspection_result = double("Result", spam_score: 100, threat: false, threat_message: nil, spam_checks: [])
+        allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/message has a spam score higher than the server's maxmimum/)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /spam score is higher than the failure threshold for this server/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the server mode is Development and the message was not manually queued" do
+      before do
+        server.update!(mode: "Development")
+      end
+
+      after do
+        server.update!(mode: "Live")
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/server is in development mode/)
+      end
+
+      it "sets the message status to Held" do
+        processor.process
+        expect(message.reload.status).to eq "Held"
+      end
+
+      it "creates a Held delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Held", details: /server is in development mode/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when there is no route for the incoming message" do
+      let(:route) { nil }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/no route and\/or endpoint available for processing/i)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /does not have a route and\/or endpoint available/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's spam mode is Quarantine, the message is spam and not manually queued" do
+      let(:route) { create(:route, server: server, spam_mode: "Quarantine") }
+
+      before do
+        inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
+        allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/message is spam and route says to quarantine spam message/i)
+      end
+
+      it "sets the message status to Held" do
+        processor.process
+        expect(message.reload.status).to eq "Held"
+      end
+
+      it "creates a Held delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Held", details: /message placed into quarantine/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's spam mode is Fail, the message is spam and not manually queued" do
+      let(:route) { create(:route, server: server, spam_mode: "Fail") }
+
+      before do
+        inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
+        allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/message is spam and route says to fail spam message/i)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /message is spam and the route specifies it should be failed/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's mode is Accept" do
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/route says to accept without endpoint/i)
+      end
+
+      it "sets the message status to Processed" do
+        processor.process
+        expect(message.reload.status).to eq "Processed"
+      end
+
+      it "creates a Processed delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Processed", details: /message has been accepted but not sent to any endpoints/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's mode is Hold" do
+      let(:route) { create(:route, server: server, mode: "Hold") }
+
+      context "when the message was queued manually" do
+        let(:queued_message) { create(:queued_message, :locked, server: server, message: message, manual: true) }
+
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/route says to hold and message was queued manually/i)
+        end
+
+        it "sets the message status to Processed" do
+          processor.process
+          expect(message.reload.status).to eq "Processed"
+        end
+
+        it "creates a Processed delivery" do
+          processor.process
+          delivery = message.deliveries.last
+          expect(delivery).to have_attributes(status: "Processed", details: /message has been processed/i)
+        end
+
+        it "removes the queued message" do
+          processor.process
+          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+        end
+      end
+
+      context "when the message was not queued manually" do
+        let(:queued_message) { create(:queued_message, :locked, server: server, message: message, manual: false) }
+
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/route says to hold, marking as held/i)
+        end
+
+        it "sets the message status to Held" do
+          processor.process
+          expect(message.reload.status).to eq "Held"
+        end
+
+        it "creates a Held delivery" do
+          processor.process
+          delivery = message.deliveries.last
+          expect(delivery).to have_attributes(status: "Held", details: /message has been accepted but not sent to any endpoints/i)
+        end
+
+        it "removes the queued message" do
+          processor.process
+          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+        end
+      end
+    end
+
+    context "when the route's mode is Bounce" do
+      let(:route) { create(:route, server: server, mode: "Bounce") }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/route says to bounce/i)
+      end
+
+      it "sends a bounce" do
+        expect(BounceMessage).to receive(:new).with(server, queued_message.message)
+        processor.process
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /message has been bounced because/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's mode is Reject" do
+      let(:route) { create(:route, server: server, mode: "Reject") }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/route says to bounce/i)
+      end
+
+      it "sends a bounce" do
+        expect(BounceMessage).to receive(:new).with(server, queued_message.message)
+        processor.process
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /message has been bounced because/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the route's endpoint is an HTTP endpoint" do
+      let(:endpoint) { create(:http_endpoint, server: server) }
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
+
+      it "gets a sender from the state and sends the message to it" do
+        http_sender_double = double("HTTPSender")
+        expect(http_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
+        expect(state).to receive(:sender_for).with(Postal::HTTPSender, endpoint).and_return(http_sender_double)
+        processor.process
+      end
+    end
+
+    context "when the route's endpoint is an SMTP endpoint" do
+      let(:endpoint) { create(:smtp_endpoint, server: server) }
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
+
+      it "gets a sender from the state and sends the message to it" do
+        smtp_sender_double = double("SMTPSender")
+        expect(smtp_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
+        expect(state).to receive(:sender_for).with(Postal::SMTPSender, message.recipient_domain, nil, { servers: [endpoint] }).and_return(smtp_sender_double)
+        processor.process
+      end
+    end
+
+    context "when the route's endpoint is an Address endpoint" do
+      let(:endpoint) { create(:address_endpoint, server: server) }
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
+
+      it "gets a sender from the state and sends the message to it" do
+        smtp_sender_double = double("SMTPSender")
+        expect(smtp_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
+        expect(state).to receive(:sender_for).with(Postal::SMTPSender, endpoint.domain, nil, { force_rcpt_to: endpoint.address }).and_return(smtp_sender_double)
+        processor.process
+      end
+    end
+
+    context "when the route's endpoint is an unknown endpoint" do
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: create(:webhook, server: server)) }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/invalid endpoint for route/i)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /invalid endpoint for route/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the message has been sent to a sender" do
+      let(:endpoint) { create(:smtp_endpoint, server: server) }
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
+
+      let(:send_result) do
+        Postal::SendResult.new do |result|
+          result.type = "Sent"
+          result.details = "Sent successfully"
+        end
+      end
+
+      before do
+        smtp_sender_mock = double("SMTPSender")
+        allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
+        allow(smtp_sender_mock).to receive(:start)
+        allow(smtp_sender_mock).to receive(:finish)
+        allow(smtp_sender_mock).to receive(:send_message).and_return(send_result)
+      end
+
+      context "when the sender returns a HardFail and bounces are suppressed" do
+        before do
+          send_result.type = "HardFail"
+          send_result.suppress_bounce = true
+        end
+
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/suppressing bounce message after hard fail/)
+        end
+
+        it "does not send a bounce" do
+          allow(BounceMessage).to receive(:new)
+          processor.process
+          expect(BounceMessage).to_not have_received(:new)
+        end
+      end
+
+      context "when the sender returns a HardFail and bounces should be sent" do
+        before do
+          send_result.type = "HardFail"
+          send_result.details = "Failed to send message"
+        end
+
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/sending a bounce because message hard failed/)
+        end
+
+        it "sends a bounce" do
+          expect(BounceMessage).to receive(:new).with(server, queued_message.message)
+          processor.process
+        end
+
+        it "sets the message status to HardFail" do
+          processor.process
+          expect(message.reload.status).to eq "HardFail"
+        end
+
+        it "creates a delivery with the details and a suffix about the bounce message" do
+          processor.process
+          delivery = message.deliveries.last
+          expect(delivery).to have_attributes(status: "HardFail", details: /Failed to send message. Sent bounce message to sender \(see message <msg:\d+>\)/i)
+        end
+      end
+
+      it "creates a delivery with the result from the sender" do
+        send_result.output = "some output here"
+        send_result.secure = true
+        send_result.log_id = "12345"
+        send_result.time = 2.32
+
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Sent",
+                                            details: "Sent successfully",
+                                            output: "some output here",
+                                            sent_with_ssl: true,
+                                            log_id: "12345",
+                                            time: 2.32)
+      end
+
+      context "when the sender wants to retry" do
+        before do
+          send_result.type = "SoftFail"
+          send_result.retry = true
+        end
+
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/message requeued for trying later, at/i)
+        end
+
+        it "sets the message status to SoftFail" do
+          processor.process
+          expect(message.reload.status).to eq "SoftFail"
+        end
+
+        it "updates the queued message with a new retry time" do
+          Timecop.freeze do
+            retry_time = 5.minutes.from_now.change(usec: 0)
+            processor.process
+            expect(queued_message.reload.retry_after).to eq retry_time
+          end
+        end
+
+        it "allocates a new IP address to send the message from and updates the queued message" do
+          expect(queued_message).to receive(:allocate_ip_address)
+          processor.process
+        end
+
+        it "does not remove the queued message" do
+          processor.process
+          expect(queued_message.reload).to be_present
+        end
+      end
+
+      context "when the sender does not want a retry" do
+        it "logs" do
+          processor.process
+          expect(logger).to have_logged(/message processing completed/i)
+        end
+
+        it "sets the message status to Sent" do
+          processor.process
+          expect(message.reload.status).to eq "Sent"
+        end
+
+        it "marks the endpoint as used" do
+          route.endpoint.update!(last_used_at: nil)
+          Timecop.freeze do
+            expect { processor.process }.to change { route.endpoint.reload.last_used_at.to_i }.from(0).to(Time.now.to_i)
+          end
+        end
+
+        it "removes the queued message" do
+          processor.process
+          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+        end
+      end
+    end
+
+    context "when an exception occurrs during processing" do
+      let(:endpoint) { create(:smtp_endpoint, server: server) }
+      let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
+
+      before do
+        smtp_sender_mock = double("SMTPSender")
+        allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
+        allow(smtp_sender_mock).to receive(:start)
+        allow(smtp_sender_mock).to receive(:finish)
+        allow(smtp_sender_mock).to receive(:send_message) do
+          1 / 0
+        end
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/internal error: ZeroDivisionError/i)
+      end
+
+      it "creates an Error delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Error", details: /internal error/i)
+      end
+
+      it "marks the message for retrying later" do
+        processor.process
+        expect(queued_message.reload.retry_after).to be_present
+      end
+    end
+  end
+
+end

spec/lib/message_dequeuer/initial_message_processor_spec.rb

@@ -0,0 +1,94 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+module MessageDequeuer
+
+  RSpec.describe InitialProcessor do
+    let(:server) { create(:server) }
+    let(:logger) { TestLogger.new }
+    let(:route) { create(:route, server: server) }
+    let(:message) { MessageFactory.incoming(server, route: route) }
+    let(:queued_message) { create(:queued_message, :locked, message: message) }
+
+    subject(:processor) { described_class.new(queued_message, logger: logger) }
+
+    it "has state when not given any" do
+      expect(processor.state).to be_a State
+    end
+
+    context "when associated message does not exist" do
+      let(:queued_message) { create(:queued_message, :locked, message_id: 12_345) }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/unqueue because backend message has been removed/)
+      end
+
+      it "removes from queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the queued message is not ready for processing" do
+      let(:queued_message) { create(:queued_message, :locked, message: message, retry_after: 1.hour.from_now) }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/skipping because message isn't ready for processing/)
+      end
+
+      it "unlocks and keeps the queued message" do
+        processor.process
+        expect(queued_message.reload).to_not be_locked
+      end
+    end
+
+    context "when there are no other batchable messages" do
+      it "calls the single message processor for the initial message" do
+        expect(SingleMessageProcessor).to receive(:process).with(queued_message,
+                                                                 logger: logger,
+                                                                 state: processor.state)
+        processor.process
+      end
+    end
+
+    context "when there are batchable messages" do
+      before do
+        @message2 = MessageFactory.incoming(server, route: route)
+        @queued_message2 = create(:queued_message, message: @message2)
+        @message3 = MessageFactory.incoming(server, route: route)
+        @queued_message3 = create(:queued_message, message: @message3)
+      end
+
+      it "calls the single message process for the initial message and all batchable messages" do
+        [queued_message, @queued_message2, @queued_message3].each do |msg|
+          expect(SingleMessageProcessor).to receive(:process).with(msg,
+                                                                   logger: logger,
+                                                                   state: processor.state)
+        end
+        processor.process
+      end
+    end
+
+    context "when an error occurs while finding batchable messages" do
+      before do
+        allow(queued_message).to receive(:batchable_messages) { 1 / 0 }
+      end
+
+      it "unlocks the queued message and raises the error" do
+        expect { processor.process }.to raise_error(ZeroDivisionError)
+        expect(queued_message.reload).to_not be_locked
+      end
+    end
+
+    context "when finished" do
+      it "notifies the state that processing is complete" do
+        expect(processor.state).to receive(:finished)
+        processor.process
+      end
+    end
+  end
+
+end

spec/lib/message_dequeuer/outgoing_message_processor_spec.rb

@@ -2,145 +2,40 @@
 
 require "rails_helper"
 
-RSpec.describe UnqueueMessageService do
+module MessageDequeuer
+
+  RSpec.describe OutgoingMessageProcessor do
     let(:server) { create(:server) }
+    let(:state) { State.new }
     let(:logger) { TestLogger.new }
-  let(:send_result) do
-    Postal::SendResult.new do |r|
-      r.type = "Sent"
-    end
-  end
-  subject(:service) { described_class.new(queued_message: queued_message, logger: logger) }
-
-  # We're going to, for now, just stop the SMTP sender from doing anything here because
-  # we don't want to leak out of this test in to the real world.
-  before do
-    smtp_sender_mock = double("SMTPSender")
-    allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
-    allow(smtp_sender_mock).to receive(:start)
-    allow(smtp_sender_mock).to receive(:finish)
-    allow(smtp_sender_mock).to receive(:send_message).and_return(send_result)
-  end
-
-  context "for an outgoing message" do
     let(:domain) { create(:domain, server: server) }
     let(:credential) { create(:credential, server: server) }
     let(:message) { MessageFactory.outgoing(server, domain: domain, credential: credential) }
     let(:queued_message) { create(:queued_message, :locked, message: message) }
 
-    context "when the server is suspended" do
-      let(:server) { create(:server, :suspended) }
-
-      it "logs" do
-        service.call
-        expect(logger).to have_logged(/server is suspended/)
-      end
-
-      it "sets the message status to Held" do
-        service.call
-        expect(message.reload.status).to eq "Held"
-      end
-
-      it "creates a Hold delivery" do
-        service.call
-        delivery = message.deliveries.last
-        expect(delivery).to have_attributes(status: "Held", details: /server has been suspended/i)
-      end
-
-      it "removes the queued message" do
-        service.call
-        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-      end
-    end
-
-    context "when the number of attempts is more than the maximum" do
-      let(:queued_message) { create(:queued_message, :locked, message: message, attempts: Postal.config.general.maximum_delivery_attempts + 1) }
-
-      it "logs" do
-        service.call
-        expect(logger).to have_logged(/message has reached maximum number of attempts/)
-      end
-
-      it "adds the recipient to the suppression list and logs this" do
-        Timecop.freeze do
-          service.call
-          entry = server.message_db.suppression_list.get(:recipient, message.rcpt_to)
-          expect(entry).to match hash_including(
-            "address" => message.rcpt_to,
-            "type" => "recipient",
-            "reason" => "too many soft fails"
-          )
-        end
-      end
-
-      it "sets the message status to Held" do
-        service.call
-        expect(message.reload.status).to eq "HardFail"
-      end
-
-      it "creates a HardFail delivery" do
-        service.call
-        delivery = message.deliveries.last
-        expect(delivery).to have_attributes(status: "HardFail", details: /maximum number of delivery attempts.*added [\w.@]+ to suppression list/i)
-      end
-
-      it "removes the queued message" do
-        service.call
-        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-      end
-    end
-
-    context "when the message raw data has been removed" do
-      before do
-        message.raw_table = nil
-        message.save
-      end
-
-      it "logs" do
-        service.call
-        expect(logger).to have_logged(/raw message has been removed/)
-      end
-
-      it "sets the message status to Held" do
-        service.call
-        expect(message.reload.status).to eq "HardFail"
-      end
-
-      it "creates a HardFail delivery" do
-        service.call
-        delivery = message.deliveries.last
-        expect(delivery).to have_attributes(status: "HardFail", details: /Raw message has been removed/i)
-      end
-
-      it "removes the queued message" do
-        service.call
-        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-      end
-    end
+    subject(:processor) { described_class.new(queued_message, logger: logger, state: state) }
 
     context "when the domain belonging to the message no longer exists" do
-      before do
-        domain.destroy
-      end
+      let(:message) { MessageFactory.outgoing(server, domain: nil, credential: credential) }
 
       it "logs" do
-        service.call
+        processor.process
         expect(logger).to have_logged(/message has no domain/)
       end
 
       it "sets the message status to HardFail" do
-        service.call
+        processor.process
         expect(message.reload.status).to eq "HardFail"
       end
 
       it "creates a HardFail delivery" do
-        service.call
+        processor.process
         delivery = message.deliveries.last
         expect(delivery).to have_attributes(status: "HardFail", details: /Message's domain no longer exist/i)
       end
 
       it "removes the queued message" do
-        service.call
+        processor.process
         expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
       end
     end
@@ -151,23 +46,23 @@ RSpec.describe UnqueueMessageService do
       end
 
       it "logs" do
-        service.call
+        processor.process
         expect(logger).to have_logged(/message has no 'to' address/)
       end
 
       it "sets the message status to HardFail" do
-        service.call
+        processor.process
         expect(message.reload.status).to eq "HardFail"
       end
 
       it "creates a HardFail delivery" do
-        service.call
+        processor.process
         delivery = message.deliveries.last
         expect(delivery).to have_attributes(status: "HardFail", details: /Message doesn't have an RCPT to/i)
       end
 
       it "removes the queued message" do
-        service.call
+        processor.process
         expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
       end
     end
@@ -180,12 +75,12 @@ RSpec.describe UnqueueMessageService do
       end
 
       it "logs" do
-        service.call
+        processor.process
         expect(logger).to have_logged(/added tag: example-tag/)
       end
 
       it "adds the tag to the message object" do
-        service.call
+        processor.process
         expect(message.reload.tag).to eq("example-tag")
       end
     end
@@ -197,7 +92,7 @@ RSpec.describe UnqueueMessageService do
         let(:queued_message) { create(:queued_message, :locked, message: message, manual: true) }
 
         it "does not hold the message" do
-          service.call
+          processor.process
           deliveries = message.deliveries.find { |d| d.status == "Held" }
           expect(deliveries).to be_nil
         end
@@ -205,23 +100,23 @@ RSpec.describe UnqueueMessageService do
 
       context "when the message was not queued manually" do
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/credential wants us to hold messages/)
         end
 
         it "sets the message status to Held" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "Held"
         end
 
         it "creates a Held delivery" do
-          service.call
+          processor.process
           delivery = message.deliveries.last
           expect(delivery).to have_attributes(status: "Held", details: /Credential is configured to hold all messages authenticated/i)
         end
 
         it "removes the queued message" do
-          service.call
+          processor.process
           expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
         end
       end
@@ -236,7 +131,7 @@ RSpec.describe UnqueueMessageService do
         let(:queued_message) { create(:queued_message, :locked, message: message, manual: true) }
 
         it "does not hold the message" do
-          service.call
+          processor.process
           deliveries = message.deliveries.find { |d| d.status == "Held" }
           expect(deliveries).to be_nil
         end
@@ -244,23 +139,23 @@ RSpec.describe UnqueueMessageService do
 
       context "when the message was not queued manually" do
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/recipient is on the suppression list/)
         end
 
         it "sets the message status to Held" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "Held"
         end
 
         it "creates a Held delivery" do
-          service.call
+          processor.process
           delivery = message.deliveries.last
           expect(delivery).to have_attributes(status: "Held", details: /Recipient \(#{message.rcpt_to}\) is on the suppression list/i)
         end
 
         it "removes the queued message" do
-          service.call
+          processor.process
           expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
         end
       end
@@ -273,7 +168,7 @@ RSpec.describe UnqueueMessageService do
         allow(mocked_parser).to receive(:tracked_links).and_return(0)
         allow(mocked_parser).to receive(:tracked_images).and_return(0)
         expect(Postal::MessageParser).to receive(:new).with(kind_of(Postal::MessageDB::Message)).and_return(mocked_parser)
-        service.call
+        processor.process
         reloaded_message = message.reload
         expect(reloaded_message.parsed).to eq 1
         expect(reloaded_message.tracked_links).to eq 0
@@ -285,7 +180,7 @@ RSpec.describe UnqueueMessageService do
       let(:server) { create(:server, outbound_spam_threshold: 5.0) }
 
       it "logs" do
-        service.call
+        processor.process
         expect(logger).to have_logged(/inspecting message/)
         expect(logger).to have_logged(/message inspected successfully/)
       end
@@ -293,7 +188,7 @@ RSpec.describe UnqueueMessageService do
       it "inspects the message" do
         inspection_result = double("Result", spam_score: 1.0, threat: false, threat_message: nil, spam_checks: [])
         expect(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
-        service.call
+        processor.process
       end
 
       context "when the message spam score is higher than the threshold" do
@@ -303,28 +198,28 @@ RSpec.describe UnqueueMessageService do
         end
 
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/message is spam/)
         end
 
         it "sets the spam boolean on the message" do
-          service.call
+          processor.process
           expect(message.reload.spam).to be true
         end
 
         it "sets the message status to HardFail" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "HardFail"
         end
 
         it "creates a HardFail delivery" do
-          service.call
+          processor.process
           delivery = message.deliveries.last
           expect(delivery).to have_attributes(status: "HardFail", details: /Message is likely spam. Threshold is 5.0 and the message scored 6.0/i)
         end
 
         it "removes the queued message" do
-          service.call
+          processor.process
           expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
         end
       end
@@ -333,7 +228,7 @@ RSpec.describe UnqueueMessageService do
     context "when the server does not have a outbound spam threshold configured" do
       it "does not inspect the message" do
         expect(Postal::MessageInspection).to_not receive(:scan)
-        service.call
+        processor.process
       end
     end
 
@@ -345,24 +240,24 @@ RSpec.describe UnqueueMessageService do
       end
 
       it "does not another one" do
-        service.call
+        processor.process
         expect(message.reload.headers["x-postal-msgid"]).to eq ["existing-id"]
       end
 
       it "does not add dkim headers" do
-        service.call
+        processor.process
         expect(message.reload.headers["dkim-signature"]).to be_nil
       end
     end
 
     context "when the message does not have a x-postal-msgid header" do
       it "adds it" do
-        service.call
+        processor.process
         expect(message.reload.headers["x-postal-msgid"]).to match [match(/[a-zA-Z0-9]{12}/)]
       end
 
       it "adds a dkim header" do
-        service.call
+        processor.process
         expect(message.reload.headers["dkim-signature"]).to match [match(/\Av=1; a=rsa-sha256/)]
       end
     end
@@ -375,27 +270,27 @@ RSpec.describe UnqueueMessageService do
       end
 
       it "updates the time the limit was exceeded" do
-        expect { service.call }.to change { server.reload.send_limit_exceeded_at }.from(nil).to(kind_of(Time))
+        expect { processor.process }.to change { server.reload.send_limit_exceeded_at }.from(nil).to(kind_of(Time))
       end
 
       it "logs" do
-        service.call
+        processor.process
         expect(logger).to have_logged(/server send limit has been exceeded/)
       end
 
       it "sets the message status to Held" do
-        service.call
+        processor.process
         expect(message.reload.status).to eq "Held"
       end
 
       it "creates a Held delivery" do
-        service.call
+        processor.process
         delivery = message.deliveries.last
         expect(delivery).to have_attributes(status: "Held", details: /Message held because send limit \(5\) has been reached/i)
       end
 
       it "removes the queued message" do
-        service.call
+        processor.process
         expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
       end
     end
@@ -408,11 +303,11 @@ RSpec.describe UnqueueMessageService do
       end
 
       it "updates the time the limit was being approached" do
-        expect { service.call }.to change { server.reload.send_limit_approaching_at }.from(nil).to(kind_of(Time))
+        expect { processor.process }.to change { server.reload.send_limit_approaching_at }.from(nil).to(kind_of(Time))
       end
 
       it "does not set the exceeded time" do
-        expect { service.call }.to_not change { server.reload.send_limit_exceeded_at } # rubocop:disable Lint/AmbiguousBlockAssociation
+        expect { processor.process }.to_not change { server.reload.send_limit_exceeded_at } # rubocop:disable Lint/AmbiguousBlockAssociation
       end
     end
 
@@ -420,7 +315,7 @@ RSpec.describe UnqueueMessageService do
       let(:server) { create(:server, :exceeded_send_limit, send_limit: 10) }
 
       it "clears the approaching and exceeded limits" do
-        service.call
+        processor.process
         server.reload
         expect(server.send_limit_approaching_at).to be_nil
         expect(server.send_limit_exceeded_at).to be_nil
@@ -434,7 +329,7 @@ RSpec.describe UnqueueMessageService do
         let(:queued_message) { create(:queued_message, :locked, message: message, manual: true) }
 
         it "does not hold the message" do
-          service.call
+          processor.process
           deliveries = message.deliveries.find { |d| d.status == "Held" }
           expect(deliveries).to be_nil
         end
@@ -442,47 +337,65 @@ RSpec.describe UnqueueMessageService do
 
       context "when the message was not queued manually" do
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/server is in development mode/)
         end
 
         it "sets the message status to Held" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "Held"
         end
 
         it "creates a Held delivery" do
-          service.call
+          processor.process
           delivery = message.deliveries.last
           expect(delivery).to have_attributes(status: "Held", details: /Server is in development mode/i)
         end
 
         it "removes the queued message" do
-          service.call
+          processor.process
           expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
         end
       end
     end
 
     context "when there are no other impediments" do
+      let(:send_result) do
+        Postal::SendResult.new do |r|
+          r.type = "Sent"
+        end
+      end
+
+      before do
+        mocked_sender = double("SMTPSender")
+        allow(mocked_sender).to receive(:send_message).and_return(send_result)
+        allow(state).to receive(:sender_for).and_return(mocked_sender)
+      end
+
       it "increments the live stats" do
-        expect { service.call }.to change { server.message_db.live_stats.total(60) }.from(0).to(1)
+        expect { processor.process }.to change { server.message_db.live_stats.total(60) }.from(0).to(1)
       end
 
       context "when there is an IP address assigned to the queued message" do
         let(:ip) { create(:ip_address) }
         let(:queued_message) { create(:queued_message, :locked, message: message, ip_address: ip) }
 
-        it "sends the message to the SMTP sender with the IP" do
-          service.call
-          expect(Postal::SMTPSender).to have_received(:new).with(message.recipient_domain, ip)
+        it "gets a sender from the state and sends the message to it" do
+          mocked_sender = double("SMTPSender")
+          expect(mocked_sender).to receive(:send_message).with(queued_message.message).and_return(send_result)
+          expect(state).to receive(:sender_for).with(Postal::SMTPSender, message.recipient_domain, ip).and_return(mocked_sender)
+
+          processor.process
         end
       end
 
       context "when there is no IP address assigned to the queued message" do
-        it "sends the message to the SMTP sender without an IP" do
-          service.call
-          expect(Postal::SMTPSender).to have_received(:new).with(message.recipient_domain, nil)
+        it "gets a sender from the state and sends the message to it" do
+          mocked_sender = double("SMTPSender")
+          expect(mocked_sender).to receive(:send_message).with(queued_message.message).and_return(send_result)
+          expect(state).to receive(:sender_for).with(Postal::SMTPSender, message.recipient_domain, nil).and_return(mocked_sender)
+
+          processor.process
         end
       end
 
@@ -493,7 +406,7 @@ RSpec.describe UnqueueMessageService do
 
         context "when the recipient has got no hard fails in the last 24 hours" do
           it "does not add to the suppression list" do
-            service.call
+            processor.process
             expect(server.message_db.suppression_list.all_with_pagination(1)[:total]).to eq 0
           end
         end
@@ -508,12 +421,12 @@ RSpec.describe UnqueueMessageService do
           end
 
           it "logs" do
-            service.call
+            processor.process
             expect(logger).to have_logged(/added #{message.rcpt_to} to suppression list because 2 hard fails in 24 hours/i)
           end
 
           it "adds the recipient to the suppression list" do
-            service.call
+            processor.process
             entry = server.message_db.suppression_list.get(:recipient, message.rcpt_to)
             expect(entry).to match hash_including(
               "address" => message.rcpt_to,
@@ -532,24 +445,25 @@ RSpec.describe UnqueueMessageService do
         end
 
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/removed #{message.rcpt_to} from suppression list/)
         end
 
         it "removes them from the suppression list" do
-          service.call
+          processor.process
           expect(server.message_db.suppression_list.get(:recipient, message.rcpt_to)).to be_nil
         end
 
-        it "adds the details to the result details" do
-          service.call
-          expect(send_result.details).to include("Recipient removed from suppression list")
+        it "adds the details to the delivery details" do
+          processor.process
+          delivery = message.deliveries.last
+          expect(delivery.details).to include("Recipient removed from suppression list")
         end
       end
 
       it "creates a delivery with the appropriate details" do
         send_result.details = "Sent successfully to mx.example.com"
-        service.call
+        processor.process
         delivery = message.deliveries.last
         expect(delivery).to have_attributes(status: "Sent", details: "Sent successfully to mx.example.com")
       end
@@ -561,19 +475,19 @@ RSpec.describe UnqueueMessageService do
         end
 
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/message requeued for trying later/)
         end
 
         it "sets the message status to SoftFail" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "SoftFail"
         end
 
         it "updates the retry time on the queued message" do
           Timecop.freeze do
             retry_time = 5.minutes.from_now.change(usec: 0)
-            service.call
+            processor.process
             expect(queued_message.reload.retry_after).to eq retry_time
           end
         end
@@ -581,20 +495,48 @@ RSpec.describe UnqueueMessageService do
 
       context "if the message should not be retried" do
         it "logs" do
-          service.call
+          processor.process
           expect(logger).to have_logged(/message processing complete/)
         end
 
         it "sets the message status to Sent" do
-          service.call
+          processor.process
           expect(message.reload.status).to eq "Sent"
         end
 
         it "removes the queued message" do
-          service.call
+          processor.process
           expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
         end
       end
     end
+
+    context "when an exception occurrs during processing" do
+      before do
+        smtp_sender_mock = double("SMTPSender")
+        allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
+        allow(smtp_sender_mock).to receive(:start)
+        allow(smtp_sender_mock).to receive(:send_message) do
+          1 / 0
         end
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/internal error: ZeroDivisionError/i)
+      end
+
+      it "creates an Error delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Error", details: /internal error/i)
+      end
+
+      it "marks the message for retrying later" do
+        processor.process
+        expect(queued_message.reload.retry_after).to be_present
+      end
+    end
+  end
+
 end

spec/lib/message_dequeuer/single_message_processor_spec.rb

@@ -0,0 +1,134 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+module MessageDequeuer
+
+  RSpec.describe SingleMessageProcessor do
+    let(:server) { create(:server) }
+    let(:state) { State.new }
+    let(:logger) { TestLogger.new }
+    let(:route) { create(:route, server: server) }
+    let(:message) { MessageFactory.incoming(server, route: route) }
+    let(:queued_message) { create(:queued_message, :locked, message: message) }
+
+    subject(:processor) { described_class.new(queued_message, logger: logger, state: state) }
+
+    context "when the server is suspended" do
+      before do
+        allow(queued_message.server).to receive(:suspended?).and_return(true)
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/server is suspended/)
+      end
+
+      it "sets the message status to Held" do
+        processor.process
+        expect(message.reload.status).to eq "Held"
+      end
+
+      it "creates a Held delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "Held", details: /server has been suspended/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the number of attempts is more than the maximum" do
+      let(:queued_message) { create(:queued_message, :locked, message: message, attempts: Postal.config.general.maximum_delivery_attempts + 1) }
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/message has reached maximum number of attempts/)
+      end
+
+      it "sends a bounce to the sender" do
+        expect(BounceMessage).to receive(:new).with(server, queued_message.message)
+        processor.process
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /maximum number of delivery attempts.*bounce sent to sender/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the message raw data has been removed" do
+      before do
+        message.raw_table = nil
+        message.save
+      end
+
+      it "logs" do
+        processor.process
+        expect(logger).to have_logged(/raw message has been removed/)
+      end
+
+      it "sets the message status to HardFail" do
+        processor.process
+        expect(message.reload.status).to eq "HardFail"
+      end
+
+      it "creates a HardFail delivery" do
+        processor.process
+        delivery = message.deliveries.last
+        expect(delivery).to have_attributes(status: "HardFail", details: /Raw message has been removed/i)
+      end
+
+      it "removes the queued message" do
+        processor.process
+        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
+      end
+    end
+
+    context "when the message is incoming" do
+      it "calls the incoming message processor" do
+        expect(IncomingMessageProcessor).to receive(:new).with(queued_message,
+                                                               logger: logger,
+                                                               state: processor.state)
+        processor.process
+      end
+
+      it "does not call the outgoing message processor" do
+        expect(OutgoingMessageProcessor).to_not receive(:process)
+        processor.process
+      end
+    end
+
+    context "when the message is outgoing" do
+      let(:message) { MessageFactory.outgoing(server) }
+
+      it "calls the outgoing message processor" do
+        expect(OutgoingMessageProcessor).to receive(:process).with(queued_message,
+                                                                   logger: logger,
+                                                                   state: processor.state)
+
+        processor.process
+      end
+
+      it "does not call the incoming message processor" do
+        expect(IncomingMessageProcessor).to_not receive(:process)
+        processor.process
+      end
+    end
+  end
+
+end

spec/lib/message_dequeuer/state_spec.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+module MessageDequeuer
+
+  RSpec.describe State do
+    subject(:state) { described_class.new }
+
+    describe "#send_result" do
+      it "can be get and set" do
+        result = instance_double(Postal::SendResult)
+        state.send_result = result
+        expect(state.send_result).to be result
+      end
+    end
+
+    describe "#sender_for" do
+      it "returns a instance of the given sender initialized with the args" do
+        sender = state.sender_for(Postal::HTTPSender, "1234")
+        expect(sender).to be_a Postal::HTTPSender
+      end
+
+      it "returns a cached sender on subsequent calls" do
+        sender = state.sender_for(Postal::HTTPSender, "1234")
+        expect(state.sender_for(Postal::HTTPSender, "1234")).to be sender
+      end
+    end
+
+    describe "#finished" do
+      it "calls finish on all cached senders" do
+        sender1 = state.sender_for(Postal::HTTPSender, "1234")
+        sender2 = state.sender_for(Postal::HTTPSender, "4444")
+        expect(sender1).to receive(:finish)
+        expect(sender2).to receive(:finish)
+
+        state.finished
+      end
+    end
+  end
+
+end

spec/lib/worker/jobs/process_queued_messages_job_spec.rb

@@ -7,18 +7,16 @@ module Worker
 
     RSpec.describe ProcessQueuedMessagesJob do
       subject(:job) { described_class.new(logger: Postal.logger) }
-      let(:mocked_service) { instance_double(UnqueueMessageService) }
 
       before do
-        allow(UnqueueMessageService).to receive(:new).and_return(mocked_service)
-        allow(mocked_service).to receive(:call).with(any_args)
+        allow(MessageDequeuer).to receive(:process)
       end
 
       describe "#call" do
         context "when there are no queued messages" do
           it "does nothing" do
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
           end
         end
 
@@ -27,7 +25,7 @@ module Worker
             ip_address = create(:ip_address)
             queued_message = create(:queued_message, ip_address: ip_address)
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be false
           end
         end
@@ -36,10 +34,9 @@ module Worker
           it "locks the message and calls the service" do
             queued_message = create(:queued_message, ip_address: nil, retry_after: nil)
             job.call
-            expect(UnqueueMessageService).to have_received(:new).with(logger: kind_of(Klogger::Logger), queued_message: queued_message)
-            expect(mocked_service).to have_received(:call)
+            expect(MessageDequeuer).to have_received(:process).with(queued_message, logger: kind_of(Klogger::Logger))
             expect(queued_message.reload.locked?).to be true
-            expect(queued_message.locked_by).to eq Postal.locker_name
+            expect(queued_message.locked_by).to match(/\A#{Postal.locker_name} [a-f0-9]{16}\z/)
             expect(queued_message.locked_at).to be_within(1.second).of(Time.current)
           end
         end
@@ -48,10 +45,9 @@ module Worker
           it "locks the message and calls the service" do
             queued_message = create(:queued_message, ip_address: nil, retry_after: 10.minutes.ago)
             job.call
-            expect(UnqueueMessageService).to have_received(:new).with(logger: kind_of(Klogger::Logger), queued_message: queued_message)
-            expect(mocked_service).to have_received(:call)
+            expect(MessageDequeuer).to have_received(:process).with(queued_message, logger: kind_of(Klogger::Logger))
             expect(queued_message.reload.locked?).to be true
-            expect(queued_message.locked_by).to eq Postal.locker_name
+            expect(queued_message.locked_by).to match(/\A#{Postal.locker_name} [a-f0-9]{16}\z/)
             expect(queued_message.locked_at).to be_within(1.second).of(Time.current)
           end
         end
@@ -60,7 +56,7 @@ module Worker
           it "does nothing" do
             queued_message = create(:queued_message, ip_address: nil, retry_after: 10.minutes.from_now)
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be false
           end
         end
@@ -69,7 +65,7 @@ module Worker
           it "does nothing" do
             queued_message = create(:queued_message, :locked, ip_address: nil, retry_after: nil)
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be true
           end
         end
@@ -78,7 +74,7 @@ module Worker
           it "does nothing" do
             queued_message = create(:queued_message, :locked, ip_address: nil, retry_after: 1.month.ago)
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be true
           end
         end
@@ -89,10 +85,9 @@ module Worker
             allow(Socket).to receive(:ip_address_list).and_return([Addrinfo.new(["AF_INET", 1, "localhost.localdomain", "10.20.30.40"])])
             queued_message = create(:queued_message, ip_address: ip_address)
             job.call
-            expect(UnqueueMessageService).to have_received(:new).with(logger: kind_of(Klogger::Logger), queued_message: queued_message)
-            expect(mocked_service).to have_received(:call)
+            expect(MessageDequeuer).to have_received(:process).with(queued_message, logger: kind_of(Klogger::Logger))
             expect(queued_message.reload.locked?).to be true
-            expect(queued_message.locked_by).to eq Postal.locker_name
+            expect(queued_message.locked_by).to match(/\A#{Postal.locker_name} [a-f0-9]{16}\z/)
             expect(queued_message.locked_at).to be_within(1.second).of(Time.current)
           end
         end
@@ -103,7 +98,7 @@ module Worker
             allow(Socket).to receive(:ip_address_list).and_return([Addrinfo.new(["AF_INET", 1, "localhost.localdomain", "10.20.30.40"])])
             queued_message = create(:queued_message, ip_address: ip_address, retry_after: 1.month.from_now)
             job.call
-            expect(UnqueueMessageService).to_not have_received(:new)
+            expect(MessageDequeuer).to_not have_received(:process)
             expect(queued_message.reload.locked?).to be false
           end
         end

spec/lib/worker/jobs/process_webhook_requests_job_spec.rb

@@ -8,8 +8,11 @@ module Worker
     RSpec.describe ProcessWebhookRequestsJob do
       subject(:job) { described_class.new(logger: Postal.logger) }
 
+      let(:mocked_service) { double("Service") }
+
       before do
-        allow_any_instance_of(WebhookRequest).to receive(:deliver)
+        allow(WebhookDeliveryService).to receive(:new).and_return(mocked_service)
+        allow(mocked_service).to receive(:call).with(no_args)
       end
 
       context "when there are no requests to process" do
@@ -21,16 +24,18 @@ module Worker
 
       context "when there is a unlocked request with no retry time" do
         it "delivers the request" do
-          create(:webhook_request)
+          request = create(:webhook_request)
           job.call
+          expect(WebhookDeliveryService).to have_received(:new).with(webhook_request: request)
           expect(job.work_completed?).to be true
         end
       end
 
       context "when there is an unlocked request with a retry time in the past" do
         it "delivers the request" do
-          create(:webhook_request, retry_after: 1.minute.ago)
+          request = create(:webhook_request, retry_after: 1.minute.ago)
           job.call
+          expect(WebhookDeliveryService).to have_received(:new).with(webhook_request: request)
           expect(job.work_completed?).to be true
         end
       end

spec/services/unqueue_message_service/incoming_messages_spec.rb

@@ -1,743 +0,0 @@
-# frozen_string_literal: true
-
-require "rails_helper"
-
-RSpec.describe UnqueueMessageService do
-  let(:server) { create(:server) }
-  let(:logger) { TestLogger.new }
-  let(:queued_message) { create(:queued_message, server: server) }
-  subject(:service) { described_class.new(queued_message: queued_message, logger: logger) }
-
-  # We're going to, for now, just stop the SMTP sender from doing anything here because
-  # we don't want to leak out of this test in to the real world.
-  before do
-    smtp_sender_mock = double("SMTPSender")
-    allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
-    allow(smtp_sender_mock).to receive(:start)
-    allow(smtp_sender_mock).to receive(:finish)
-    allow(smtp_sender_mock).to receive(:send_message) do
-      puts "SMTP SENDING DETECTED!"
-    end
-  end
-
-  describe "#call" do
-    context "for an incoming message" do
-      let(:route) { create(:route, server: server) }
-      let(:message) { MessageFactory.incoming(server, route: route) }
-      let(:queued_message) { create(:queued_message, :locked, message: message) }
-
-      context "when the server is suspended" do
-        before do
-          allow(queued_message.server).to receive(:suspended?).and_return(true)
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/server is suspended/)
-        end
-
-        it "sets the message status to Held" do
-          service.call
-          expect(message.reload.status).to eq "Held"
-        end
-
-        it "creates a Held delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Held", details: /server has been suspended/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the number of attempts is more than the maximum" do
-        let(:queued_message) { create(:queued_message, :locked, message: message, attempts: Postal.config.general.maximum_delivery_attempts + 1) }
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/message has reached maximum number of attempts/)
-        end
-
-        it "sends a bounce to the sender" do
-          expect(BounceMessage).to receive(:new).with(server, queued_message.message)
-          service.call
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /maximum number of delivery attempts.*bounce sent to sender/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the message raw data has been removed" do
-        before do
-          message.raw_table = nil
-          message.save
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/raw message has been removed/)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /Raw message has been removed/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the message is a bounce for an existing message" do
-        let(:existing_message) { MessageFactory.outgoing(server) }
-
-        let(:message) do
-          MessageFactory.incoming(server) do |msg, mail|
-            msg.bounce = true
-            mail["X-Postal-MsgID"] = existing_message.token
-          end
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/message is a bounce/)
-        end
-
-        it "adds the original message as the bounce ID for the received message" do
-          service.call
-          expect(message.reload.bounce_for_id).to eq existing_message.id
-        end
-
-        it "sets the received message status to Processed" do
-          service.call
-          expect(message.reload.status).to eq "Processed"
-        end
-
-        it "creates a Processed delivery on the received message" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Processed", details: /This has been detected as a bounce message for <msg:#{existing_message.id}>/i)
-        end
-
-        it "sets the existing message status to Bounced" do
-          service.call
-          expect(existing_message.reload.status).to eq "Bounced"
-        end
-
-        it "creates a Bounced delivery on the original message" do
-          service.call
-          delivery = existing_message.deliveries.last
-          expect(delivery).to have_attributes(status: "Bounced", details: /received a bounce message for this e-mail. See <msg:#{message.id}> for/i)
-        end
-
-        it "triggers a MessageBounced webhook event" do
-          expect(WebhookRequest).to receive(:trigger).with(server, "MessageBounced", {
-            original_message: kind_of(Hash),
-            bounce: kind_of(Hash)
-          })
-          service.call
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the message was a bounce but there's no return path for it" do
-        let(:message) do
-          MessageFactory.incoming(server) do |msg|
-            msg.bounce = true
-          end
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/no source messages found, hard failing/)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /was a bounce but we couldn't link it with any outgoing message/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the message is not a bounce" do
-        it "increments the stats for the server" do
-          expect { service.call }.to change { server.message_db.live_stats.total(5) }.by(1)
-        end
-
-        it "inspects the message and adds headers" do
-          expect { service.call }.to change { message.reload.inspected }.from(false).to(true)
-          new_message = message.reload
-          expect(new_message.headers).to match hash_including(
-            "x-postal-spam" => ["no"],
-            "x-postal-spam-threshold" => ["5.0"],
-            "x-postal-threat" => ["no"]
-          )
-        end
-
-        it "marks the message as spam if the spam score is higher than the server threshold" do
-          inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
-          allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
-          service.call
-          expect(message.reload.spam).to be true
-        end
-      end
-
-      context "when the message has a spam score greater than the server's spam failure threshold" do
-        before do
-          inspection_result = double("Result", spam_score: 100, threat: false, threat_message: nil, spam_checks: [])
-          allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/message has a spam score higher than the server's maxmimum/)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /spam score is higher than the failure threshold for this server/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the server mode is Development and the message was not manually queued" do
-        before do
-          server.update!(mode: "Development")
-        end
-
-        after do
-          server.update!(mode: "Live")
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/server is in development mode/)
-        end
-
-        it "sets the message status to Held" do
-          service.call
-          expect(message.reload.status).to eq "Held"
-        end
-
-        it "creates a Held delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Held", details: /server is in development mode/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when there is no route for the incoming message" do
-        let(:route) { nil }
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/no route and\/or endpoint available for processing/i)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /does not have a route and\/or endpoint available/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's spam mode is Quarantine, the message is spam and not manually queued" do
-        let(:route) { create(:route, server: server, spam_mode: "Quarantine") }
-
-        before do
-          inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
-          allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/message is spam and route says to quarantine spam message/i)
-        end
-
-        it "sets the message status to Held" do
-          service.call
-          expect(message.reload.status).to eq "Held"
-        end
-
-        it "creates a Held delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Held", details: /message placed into quarantine/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's spam mode is Fail, the message is spam and not manually queued" do
-        let(:route) { create(:route, server: server, spam_mode: "Fail") }
-
-        before do
-          inspection_result = double("Result", spam_score: server.spam_threshold + 1, threat: false, threat_message: nil, spam_checks: [])
-          allow(Postal::MessageInspection).to receive(:scan).and_return(inspection_result)
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/message is spam and route says to fail spam message/i)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /message is spam and the route specifies it should be failed/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's mode is Accept" do
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/route says to accept without endpoint/i)
-        end
-
-        it "sets the message status to Processed" do
-          service.call
-          expect(message.reload.status).to eq "Processed"
-        end
-
-        it "creates a Processed delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Processed", details: /message has been accepted but not sent to any endpoints/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's mode is Hold" do
-        let(:route) { create(:route, server: server, mode: "Hold") }
-
-        context "when the message was queued manually" do
-          let(:queued_message) { create(:queued_message, :locked, server: server, message: message, manual: true) }
-
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/route says to hold and message was queued manually/i)
-          end
-
-          it "sets the message status to Processed" do
-            service.call
-            expect(message.reload.status).to eq "Processed"
-          end
-
-          it "creates a Processed delivery" do
-            service.call
-            delivery = message.deliveries.last
-            expect(delivery).to have_attributes(status: "Processed", details: /message has been processed/i)
-          end
-
-          it "removes the queued message" do
-            service.call
-            expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-          end
-        end
-
-        context "when the message was not queued manually" do
-          let(:queued_message) { create(:queued_message, :locked, server: server, message: message, manual: false) }
-
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/route says to hold, marking as held/i)
-          end
-
-          it "sets the message status to Held" do
-            service.call
-            expect(message.reload.status).to eq "Held"
-          end
-
-          it "creates a Held delivery" do
-            service.call
-            delivery = message.deliveries.last
-            expect(delivery).to have_attributes(status: "Held", details: /message has been accepted but not sent to any endpoints/i)
-          end
-
-          it "removes the queued message" do
-            service.call
-            expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-          end
-        end
-      end
-
-      context "when the route's mode is Bounce" do
-        let(:route) { create(:route, server: server, mode: "Bounce") }
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/route says to bounce/i)
-        end
-
-        it "sends a bounce" do
-          expect(BounceMessage).to receive(:new).with(server, queued_message.message)
-          service.call
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /message has been bounced because/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's mode is Reject" do
-        let(:route) { create(:route, server: server, mode: "Reject") }
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/route says to bounce/i)
-        end
-
-        it "sends a bounce" do
-          expect(BounceMessage).to receive(:new).with(server, queued_message.message)
-          service.call
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /message has been bounced because/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the route's endpoint is an HTTP endpoint" do
-        let(:endpoint) { create(:http_endpoint, server: server) }
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
-
-        it "sends the message to the HTTPSender" do
-          http_sender_double = double("HTTPSender")
-          expect(Postal::HTTPSender).to receive(:new).with(endpoint).and_return(http_sender_double)
-          expect(http_sender_double).to receive(:start).with(no_args)
-          expect(http_sender_double).to receive(:finish).with(no_args)
-          expect(http_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
-          service.call
-        end
-      end
-
-      context "when the route's endpoint is an SMTP endpoint" do
-        let(:endpoint) { create(:smtp_endpoint, server: server) }
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
-
-        it "sends the message to the SMTPSender" do
-          smtp_sender_double = double("SMTPSender")
-          expect(smtp_sender_double).to receive(:start).with(no_args)
-          expect(smtp_sender_double).to receive(:finish).with(no_args)
-          expect(smtp_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
-          expect(Postal::SMTPSender).to receive(:new).with(message.recipient_domain, nil, { servers: [endpoint] }).and_return(smtp_sender_double)
-          service.call
-        end
-      end
-
-      context "when the route's endpoint is an Address endpoint" do
-        let(:endpoint) { create(:address_endpoint, server: server) }
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
-
-        it "sends the message to the SMTPSender" do
-          smtp_sender_double = double("SMTPSender")
-          expect(smtp_sender_double).to receive(:start).with(no_args)
-          expect(smtp_sender_double).to receive(:finish).with(no_args)
-          expect(smtp_sender_double).to receive(:send_message).with(queued_message.message).and_return(Postal::SendResult.new)
-          expect(Postal::SMTPSender).to receive(:new).with(endpoint.domain, nil, { force_rcpt_to: endpoint.address }).and_return(smtp_sender_double)
-          service.call
-        end
-      end
-
-      context "when the route's endpoint is an unknown endpoint" do
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: create(:webhook, server: server)) }
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/invalid endpoint for route/i)
-        end
-
-        it "sets the message status to HardFail" do
-          service.call
-          expect(message.reload.status).to eq "HardFail"
-        end
-
-        it "creates a HardFail delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "HardFail", details: /invalid endpoint for route/i)
-        end
-
-        it "removes the queued message" do
-          service.call
-          expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-
-      context "when the message has been sent to a sender" do
-        let(:endpoint) { create(:smtp_endpoint, server: server) }
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
-
-        let(:send_result) do
-          Postal::SendResult.new do |result|
-            result.type = "Sent"
-            result.details = "Sent successfully"
-          end
-        end
-
-        before do
-          smtp_sender_mock = double("SMTPSender")
-          allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
-          allow(smtp_sender_mock).to receive(:start)
-          allow(smtp_sender_mock).to receive(:finish)
-          allow(smtp_sender_mock).to receive(:send_message).and_return(send_result)
-        end
-
-        context "when the sender returns a HardFail and bounces are suppressed" do
-          before do
-            send_result.type = "HardFail"
-            send_result.suppress_bounce = true
-          end
-
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/suppressing bounce message after hard fail/)
-          end
-
-          it "does not send a bounce" do
-            allow(BounceMessage).to receive(:new)
-            service.call
-            expect(BounceMessage).to_not have_received(:new)
-          end
-        end
-
-        context "when the sender returns a HardFail and bounces should be sent" do
-          before do
-            send_result.type = "HardFail"
-            send_result.details = "Failed to send message"
-          end
-
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/sending a bounce because message hard failed/)
-          end
-
-          it "sends a bounce" do
-            expect(BounceMessage).to receive(:new).with(server, queued_message.message)
-            service.call
-          end
-
-          it "sets the message status to HardFail" do
-            service.call
-            expect(message.reload.status).to eq "HardFail"
-          end
-
-          it "creates a delivery with the details and a suffix about the bounce message" do
-            service.call
-            delivery = message.deliveries.last
-            expect(delivery).to have_attributes(status: "HardFail", details: /Failed to send message. Sent bounce message to sender \(see message <msg:\d+>\)/i)
-          end
-        end
-
-        it "creates a delivery with the result from the sender" do
-          send_result.output = "some output here"
-          send_result.secure = true
-          send_result.log_id = "12345"
-          send_result.time = 2.32
-
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Sent",
-                                              details: "Sent successfully",
-                                              output: "some output here",
-                                              sent_with_ssl: true,
-                                              log_id: "12345",
-                                              time: 2.32)
-        end
-
-        context "when the sender wants to retry" do
-          before do
-            send_result.type = "SoftFail"
-            send_result.retry = true
-          end
-
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/message requeued for trying later, at/i)
-          end
-
-          it "sets the message status to SoftFail" do
-            service.call
-            expect(message.reload.status).to eq "SoftFail"
-          end
-
-          it "updates the queued message with a new retry time" do
-            Timecop.freeze do
-              retry_time = 5.minutes.from_now.change(usec: 0)
-              service.call
-              expect(queued_message.reload.retry_after).to eq retry_time
-            end
-          end
-
-          it "allocates a new IP address to send the message from and updates the queued message" do
-            expect(queued_message).to receive(:allocate_ip_address)
-            service.call
-          end
-
-          it "does not remove the queued message" do
-            service.call
-            expect(queued_message.reload).to be_present
-          end
-        end
-
-        context "when the sender does not want a retry" do
-          it "logs" do
-            service.call
-            expect(logger).to have_logged(/message processing completed/i)
-          end
-
-          it "sets the message status to Sent" do
-            service.call
-            expect(message.reload.status).to eq "Sent"
-          end
-
-          it "marks the endpoint as used" do
-            route.endpoint.update!(last_used_at: nil)
-            Timecop.freeze do
-              expect { service.call }.to change { route.endpoint.reload.last_used_at.to_i }.from(0).to(Time.now.to_i)
-            end
-          end
-
-          it "removes the queued message" do
-            service.call
-            expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-          end
-        end
-      end
-
-      context "when an exception occurrs during processing" do
-        let(:endpoint) { create(:smtp_endpoint, server: server) }
-        let(:route) { create(:route, server: server, mode: "Endpoint", endpoint: endpoint) }
-
-        before do
-          smtp_sender_mock = double("SMTPSender")
-          allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
-          allow(smtp_sender_mock).to receive(:start)
-          allow(smtp_sender_mock).to receive(:finish)
-          allow(smtp_sender_mock).to receive(:send_message) do
-            1 / 0
-          end
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/internal error: ZeroDivisionError/i)
-        end
-
-        it "creates an Error delivery" do
-          service.call
-          delivery = message.deliveries.last
-          expect(delivery).to have_attributes(status: "Error", details: /internal error/i)
-        end
-
-        it "marks the message for retrying later" do
-          service.call
-          expect(queued_message.reload.retry_after).to be_present
-        end
-      end
-    end
-  end
-end

spec/services/unqueue_message_service_spec.rb

@@ -1,99 +0,0 @@
-# frozen_string_literal: true
-
-require "rails_helper"
-
-RSpec.describe UnqueueMessageService do
-  let(:server) { create(:server) }
-  let(:logger) { TestLogger.new }
-  let(:queued_message) { create(:queued_message, server: server) }
-  subject(:service) { described_class.new(queued_message: queued_message, logger: logger) }
-
-  describe "#call" do
-    context "when the backend message does not exist" do
-      it "deletes the queued message" do
-        service.call
-        expect(logger).to have_logged(/unqueue because backend message has been removed/)
-        expect { queued_message.reload }.to raise_error(ActiveRecord::RecordNotFound)
-      end
-    end
-
-    context "when the message is not ready for processing" do
-      let(:message) { MessageFactory.outgoing(server) }
-      let(:queued_message) { create(:queued_message, :retry_in_future, message: message) }
-
-      it "does not do anything" do
-        service.call
-        expect(logger).to have_logged(/skipping because message isn't ready for processing/)
-      end
-    end
-
-    context "when there are other messages to batch with this one" do
-      let(:domain) { create(:domain, server: server) }
-      let(:message) { MessageFactory.outgoing(server, domain: domain) }
-      let(:queued_message) { create(:queued_message, :locked, message: message) }
-      let(:send_result) { Postal::SendResult.new }
-
-      before do
-        smtp_sender_mock = double("SMTPSender")
-        allow(Postal::SMTPSender).to receive(:new).and_return(smtp_sender_mock)
-        allow(smtp_sender_mock).to receive(:start)
-        allow(smtp_sender_mock).to receive(:finish)
-        allow(smtp_sender_mock).to receive(:send_message).and_return(send_result)
-      end
-
-      before do
-        # Create 2 extra messages which are similar to the original
-        @message2 = MessageFactory.outgoing(server, domain: domain)
-        @queued_message2 = create(:queued_message, message: @message2)
-        @message3 = MessageFactory.outgoing(server, domain: domain)
-        @queued_message3 = create(:queued_message, message: @message3)
-      end
-
-      it "logs" do
-        service.call
-        expect(logger).to have_logged(/found 2 associated messages/)
-      end
-
-      it "sends processes each message" do
-        allow(service).to receive(:process_message).and_call_original
-        service.call
-        expect(service).to have_received(:process_message).with(queued_message)
-        expect(service).to have_received(:process_message).with(@queued_message2)
-        expect(service).to have_received(:process_message).with(@queued_message3)
-      end
-
-      context "when there is a connect error" do
-        before do
-          send_result.type = "SoftFail"
-          send_result.connect_error = true
-          send_result.details = "Connection Error"
-          send_result.retry = true
-        end
-
-        it "uses the same result for subsequent messages" do
-          service.call
-          expect(Postal::SMTPSender).to have_received(:new).once
-          expect(message.reload.status).to eq "SoftFail"
-          expect(@message2.reload.status).to eq "SoftFail"
-          expect(@message3.reload.status).to eq "SoftFail"
-        end
-      end
-
-      context "when the backend message of a sub-message has been removed" do
-        before do
-          @message2.delete
-        end
-
-        it "logs" do
-          service.call
-          expect(logger).to have_logged(/unqueueing because backend message has been removed/)
-        end
-
-        it "removes the queued message for that message" do
-          service.call
-          expect { @queued_message2.reload }.to raise_error(ActiveRecord::RecordNotFound)
-        end
-      end
-    end
-  end
-end

spec/util/message_dequeuer_spec.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+RSpec.describe MessageDequeuer do
+  describe ".process" do
+    it "calls the initial process with the given message and logger" do
+      message = create(:queued_message)
+      logger = TestLogger.new
+
+      mock = double("InitialProcessor")
+      expect(mock).to receive(:process).with(no_args)
+      expect(MessageDequeuer::InitialProcessor).to receive(:new).with(message, logger: logger).and_return(mock)
+
+      described_class.process(message, logger: logger)
+    end
+  end
+end