LG-8440: Ingest in-person enrollment status updates from SQS

Gemfile

@@ -11,6 +11,7 @@ gem 'aws-sdk-pinpoint'
 gem 'aws-sdk-pinpointsmsvoice'
 gem 'aws-sdk-ses', '~> 1.6'
 gem 'aws-sdk-sns'
+gem 'aws-sdk-sqs'
 gem 'barby', '~> 0.6.8'
 gem 'base32-crockford'
 gem 'bootsnap', '~> 1.0', require: false
@@ -35,6 +36,7 @@ gem 'jwt'
 gem 'lograge', '>= 0.11.2'
 gem 'lookbook', '~> 1.5.3', require: false
 gem 'lru_redux'
+gem 'mail'
 gem 'msgpack', '~> 1.6'
 gem 'maxminddb'
 gem 'multiset'

Gemfile.lock

@@ -164,6 +164,9 @@ GEM
     aws-sdk-sns (1.49.0)
       aws-sdk-core (~> 3, >= 3.122.0)
       aws-sigv4 (~> 1.1)
+    aws-sdk-sqs (1.53.0)
+      aws-sdk-core (~> 3, >= 3.165.0)
+      aws-sigv4 (~> 1.1)
     aws-sigv4 (1.5.2)
       aws-eventstream (~> 1, >= 1.0.2)
     axe-core-api (4.3.2)
@@ -729,6 +732,7 @@ DEPENDENCIES
   aws-sdk-pinpointsmsvoice
   aws-sdk-ses (~> 1.6)
   aws-sdk-sns
+  aws-sdk-sqs
   axe-core-rspec (~> 4.2)
   barby (~> 0.6.8)
   base32-crockford
@@ -769,6 +773,7 @@ DEPENDENCIES
   lograge (>= 0.11.2)
   lookbook (~> 1.5.3)
   lru_redux
+  mail
   maxminddb
   msgpack (~> 1.6)
   multiset

app/jobs/in_person/enrollments_ready_for_status_check/batch_processor.rb

@@ -0,0 +1,75 @@
+module InPerson::EnrollmentsReadyForStatusCheck
+  class BatchProcessor
+    # @param [InPerson::EnrollmentsReadyForStatusCheck::ErrorReporter] error_reporter
+    # @param [InPerson::EnrollmentsReadyForStatusCheck::SqsBatchWrapper] sqs_batch_wrapper
+    # @param [InPerson::EnrollmentsReadyForStatusCheck::EnrollmentPipeline] enrollment_pipeline
+    def initialize(error_reporter:, sqs_batch_wrapper:, enrollment_pipeline:)
+      @error_reporter = error_reporter
+      @sqs_batch_wrapper = sqs_batch_wrapper
+      @enrollment_pipeline = enrollment_pipeline
+    end
+
+    # Process a batch of incoming messages corresponding to in-person
+    # enrollments that are ready to have their status checked.
+    #
+    # Note: Stats are accepted as param to increment and facilitate logging even
+    # if this method raises an error.
+    #
+    # @param [Array<Aws::SQS::Types::Message>] messages In-person enrollment SQS messages
+    # @param [Hash] analytics_stats Counters for aggregating info about how items were processed
+    # @option analytics_stats [Integer] :fetched_items Items received from SQS
+    # @option analytics_stats [Integer] :valid_items Items matching the expected format/data
+    # @option analytics_stats [Integer] :invalid_items Items not matching the expected format/data
+    # @option analytics_stats [Integer] :processed_items Items processed without errors
+    # @option analytics_stats [Integer] :deleted_items Items successfully deleted from queue
+    def process_batch(messages, analytics_stats)
+      analytics_stats[:fetched_items] += messages.size
+      # Keep messages to delete in an array for a batch call
+      deletion_batch = []
+      messages.each do |sqs_message|
+        if process_message(sqs_message)
+          analytics_stats[:valid_items] += 1
+        else
+          analytics_stats[:invalid_items] += 1
+        end
+
+        # Append messages to batch so we can dequeue any that we've processed.
+        #
+        # If we fail to process the message now but could process it later, then
+        # we should exclude that message from the deletion batch.
+        deletion_batch.append(sqs_message)
+        analytics_stats[:processed_items] += 1
+      end
+    ensure
+      # The messages were processed, so remove them from the queue
+      analytics_stats[:deleted_items] += process_deletions(deletion_batch)
+    end
+
+    private
+
+    attr_reader :error_reporter, :sqs_batch_wrapper, :enrollment_pipeline
+
+    delegate :report_error, to: :error_reporter
+    delegate :delete_message_batch, to: :sqs_batch_wrapper
+    delegate :process_message, to: :enrollment_pipeline
+
+    # Delete messages from the queue and report deletion errors
+    # @param [Array<Aws::SQS::Types::Message>] deletion_batch SQS messages to delete
+    # @return [Integer] Number of items deleted
+    def process_deletions(deletion_batch)
+      return 0 if deletion_batch.empty?
+
+      delete_result = delete_message_batch(deletion_batch)
+      delete_result.failed.each do |error_entry|
+        report_error(
+          'Failed to delete item from queue',
+          sqs_delete_error: error_entry.to_h,
+        )
+      end
+      delete_result.successful.size
+    rescue StandardError => err
+      report_error(err)
+      0
+    end
+  end
+end

app/jobs/in_person/enrollments_ready_for_status_check/enrollment_pipeline.rb

@@ -0,0 +1,145 @@
+module InPerson::EnrollmentsReadyForStatusCheck
+  class EnrollmentPipeline
+    # @param [InPerson::EnrollmentsReadyForStatusCheck::ErrorReporter] error_reporter
+    # @param [Regexp] email_body_pattern Pattern matching the expected email body
+    # Note: email_body_pattern must include a capture group named "enrollment_code"
+    def initialize(error_reporter:, email_body_pattern:)
+      @error_reporter = error_reporter
+      @email_body_pattern = email_body_pattern
+    end
+
+    # Process a message from USPS indicating that an in-person
+    # enrollment is ready to have its status checked.
+    #
+    # When a message can't be processed, then this function will return
+    # false to indicate that a problem occurred with the message. Otherwise it
+    # will return true. If an error occurs that can't be clearly identified
+    # as a problem with the message, then an error will be raised instead.
+    #
+    # When a message is successfully processed, then the corresponding
+    # InPersonEnrollment record will be marked as ready for a status check.
+    #
+    # @param [Aws::SQS::Types::Message] sqs_message
+    # @return [Boolean] Returns false for messages that can't be processed
+    # @raise [StandardError] Raised when an unhandled error occurs
+    def process_message(sqs_message)
+      error_extra = {
+        sqs_message_id: sqs_message.message_id,
+      }
+
+      # Unwrap SQS message to get SNS message
+      begin
+        sns_message = JSON.parse(sqs_message.body, { symbolize_names: true })
+      rescue JSON::JSONError => err
+        report_error(err, **error_extra)
+        return false
+      end
+
+      unless sns_message.is_a?(Hash) && sns_message.key?(:MessageId) && sns_message.key?(:Message)
+        report_error('SQS message body is not valid SNS payload', **error_extra)
+        return false
+      end
+
+      error_extra[:sns_message_id] = sns_message[:MessageId]
+
+      # Unwrap SNS message to get SES message
+      begin
+        ses_message = JSON.parse(sns_message[:Message], { symbolize_names: true })
+      rescue JSON::JSONError => err
+        report_error(err, **error_extra)
+        return false
+      end
+
+      unless ses_message.is_a?(Hash) && ses_message.key?(:content) && ses_message.key?(:mail)
+        report_error('SNS "Message" field is not a valid SES payload', **error_extra)
+        return false
+      end
+
+      # Add information to help with debugging
+      error_extra[:ses_aws_message_id] = ses_message.dig(:mail, :messageId)
+      error_extra[:ses_mail_timestamp] = ses_message.dig(:mail, :timestamp)
+      error_extra[:ses_mail_source] = ses_message.dig(:mail, :source)
+
+      # https://datatracker.ietf.org/doc/html/rfc5322#section-3.6.1
+      error_extra[:ses_rfc_origination_date] = ses_message.
+        dig(:mail, :commonHeaders, :date)&.then do |date|
+          DateTime.parse(date).to_s
+        end
+      # https://datatracker.ietf.org/doc/html/rfc5322#section-3.6.4
+      error_extra[:ses_rfc_message_id] = ses_message.dig(:mail, :commonHeaders, :messageId)
+
+      # Parse email from content of SES message
+      begin
+        mail = Mail.read_from_string(ses_message[:content])
+        # Depending on how the email is created, we may need to read different
+        # parts of the message
+        if mail.multipart?
+          text_body = mail.text_part&.decoded
+        else
+          text_body = mail.body.decoded
+        end
+      rescue StandardError
+        report_error(err, **error_extra)
+        return false
+      end
+
+      unless text_body.respond_to?(:to_s) && text_body.to_s.present?
+        report_error('Failure occurred when attempting to get email body', **error_extra)
+        return false
+      end
+
+      # Extract enrollment code from email body
+      enrollment_code = email_body_pattern.match(text_body.to_s)&.[](:enrollment_code)
+
+      unless enrollment_code.is_a?(String)
+        report_error(
+          'Failed to extract enrollment code using regex, check email body format and regex',
+          **error_extra,
+        )
+        return false
+      end
+
+      error_extra[:enrollment_code] = enrollment_code
+
+      # Look up existing enrollment
+      id, user_id, ready_for_status_check = InPersonEnrollment.
+        where(enrollment_code:, status: :pending).
+        order(created_at: :desc).
+        limit(1).
+        pick(
+          :id, :user_id, :ready_for_status_check
+        )
+
+      if id.nil?
+        report_error(
+          'Received code for enrollment that does not exist in the database',
+          **error_extra,
+        )
+        return false
+      end
+
+      error_extra[:enrollment_id] = id
+      error_extra[:user_id] = user_id
+
+      # SQS can deliver the message multiple times, so it's expected that
+      # sometimes ready_for_status_check will already be set to true.
+      unless ready_for_status_check
+        # Mark enrollment as ready for the USPS status check.
+        #
+        # Note: There is still a chance of duplicate updates at this point,
+        # but the conditional above should catch most cases.
+        InPersonEnrollment.update(id, ready_for_status_check: true)
+      end
+      return true
+    rescue StandardError => err
+      # Report and re-throw unhandled errors
+      report_error(err, **error_extra)
+      raise err
+    end
+
+    private
+
+    attr_reader :error_reporter, :email_body_pattern
+    delegate :report_error, to: :error_reporter
+  end
+end

app/jobs/in_person/enrollments_ready_for_status_check/error_reporter.rb

@@ -0,0 +1,29 @@
+module InPerson
+  module EnrollmentsReadyForStatusCheck
+    class ErrorReporter
+      # @param [String] class_name Class for which to report errors
+      # @param [Analytics] analytics
+      def initialize(class_name, analytics)
+        @class_name = class_name
+        @analytics = analytics
+      end
+
+      # Reports an error. A non-StandardError will be converted to
+      # a RuntimeError before being reported.
+      # @param [#to_s,StandardError] error
+      def report_error(error, **extra)
+        error = RuntimeError.new("#{@class_name}: #{error}") unless error.is_a?(StandardError)
+        analytics.idv_in_person_proofing_enrollments_ready_for_status_check_job_ingestion_error(
+          exception_class: error.class,
+          exception_message: error.message,
+          **extra,
+        )
+        NewRelic::Agent.notice_error(error)
+      end
+
+      private
+
+      attr_reader :analytics
+    end
+  end
+end

app/jobs/in_person/enrollments_ready_for_status_check/sqs_batch_wrapper.rb

@@ -0,0 +1,37 @@
+module InPerson::EnrollmentsReadyForStatusCheck
+  class SqsBatchWrapper
+    # @param [Aws::SQS::Client] sqs_client AWS SQS Client
+    # @param [String] queue_url The URL identifying the SQS queue
+    # @param [Hash] receive_params Parameters passed to #receive_message
+    def initialize(sqs_client:, queue_url:, receive_params:)
+      @sqs_client = sqs_client
+      @queue_url = queue_url
+      @receive_params = receive_params
+    end
+
+    # Fetch a batch of messages from the SQS queue
+    # @return [Array<Aws::SQS::Types::Message>]
+    def poll
+      sqs_client.receive_message(receive_params).messages
+    end
+
+    # Delete the provided messages from the SQS queue
+    # @param [Array<Aws::SQS::Types::Message>] batch
+    # @return [Aws::SQS::Types::DeleteMessageBatchResult]
+    def delete_message_batch(batch)
+      sqs_client.delete_message_batch(
+        queue_url:,
+        entries: batch.map do |message|
+          {
+            id: message.message_id,
+            receipt_handle: message.receipt_handle,
+          }
+        end,
+      )
+    end
+
+    private
+
+    attr_reader :sqs_client, :queue_url, :receive_params
+  end
+end