Async proofing calls (LG-3330)

app/jobs/application_job.rb

@@ -0,0 +1,7 @@
+class ApplicationJob < ActiveJob::Base
+  # Automatically retry jobs that encountered a deadlock
+  # retry_on ActiveRecord::Deadlocked
+
+  # Most jobs are safe to ignore if the underlying records are no longer available
+  # discard_on ActiveJob::DeserializationError
+end

app/jobs/vendor_proof_job.rb

@@ -0,0 +1,12 @@
+class VendorProofJob < ApplicationJob
+  queue_as :default
+
+  def perform(document_capture_session_id, stages)
+    # binding.pry
+    dcs = DocumentCaptureSession.find_by(uuid: document_capture_session_id)
+    result = dcs.load_proofing_result
+    stages = stages.map(&:to_sym)
+    idv_result = Idv::Agent.new(result.pii).proof(*stages)
+    dcs.store_proofing_result(result.pii, idv_result)
+  end
+end

app/models/document_capture_session.rb

@@ -7,6 +7,10 @@ def load_result
     DocumentCaptureSessionResult.load(result_id)
   end
 
+  def load_proofing_result
+    ProofingDocumentCaptureSessionResult.load(result_id)
+  end
+
   def store_result_from_response(doc_auth_response)
     DocumentCaptureSessionResult.store(
       id: generate_result_id,
@@ -16,6 +20,23 @@ def store_result_from_response(doc_auth_response)
     save!
   end
 
+  def store_proofing_pii_from_doc(pii_from_doc)
+    ProofingDocumentCaptureSessionResult.store(
+      id: generate_result_id,
+      pii: pii_from_doc,
+      result: nil,
+    )
+    save!
+  end
+
+  def store_proofing_result(pii_from_doc, result)
+    ProofingDocumentCaptureSessionResult.store(
+      id: result_id,
+      pii: pii_from_doc,
+      result: result,
+    )
+  end
+
   def expired?
     return true unless requested_at
     requested_at + Figaro.env.doc_capture_request_valid_for_minutes.to_i.minutes < Time.zone.now

app/services/flow/base_flow.rb

@@ -26,15 +26,18 @@ def redirect_to(url)
     def handle(step)
       @flow_session[:error_message] = nil
       @flow_session[:notice] = nil
-      handler = steps[step] || actions[step]
-      return failure("Unhandled step #{step}") unless handler
-      wrap_send(handler)
+      return failure("Unhandled step #{step}") unless handler(step)
+      wrap_send(step)
+    end
+
+    def handler(step)
+      steps[step] || actions[step]
     end
 
     private
 
-    def wrap_send(handler)
-      obj = handler.new(self)
+    def wrap_send(step)
+      obj = handler(step).new(self)
       value = obj.base_call
       form_response(obj, value)
     end

app/services/flow/base_step.rb

@@ -26,6 +26,10 @@ def mark_step_incomplete(step = nil)
       flow_session.delete(klass.to_s)
     end
 
+    def async?
+      false
+    end
+
     def self.acceptable_response_object?(obj)
       obj.is_a?(FormResponse) || obj.is_a?(DocAuth::Response)
     end

app/services/flow/flow_state_machine.rb

@@ -14,24 +14,82 @@ def index
     end
 
     def show
-      step = current_step
-      analytics.track_event(analytics_visited, step: step) if @analytics_id
-      Funnel::DocAuth::RegisterStep.new(user_id, issuer).call(step, :view, true)
-      register_campaign
-      render_step(step, flow.flow_session)
+      flow_handler = flow.handler(current_step).new(flow)
+
+      if flow_handler.async?
+        # binding.pry
+        async_show(flow_handler)
+      else
+        begin_step
+      end
     end
 
     def update
-      step = current_step
-      result = flow.handle(step)
-      analytics.track_event(analytics_submitted, result.to_h.merge(step: step)) if @analytics_id
-      register_update_step(step, result)
-      flow_finish and return unless next_step
-      render_update(step, result)
+      flow_handler = flow.handler(current_step).new(flow)
+
+      if flow_handler.async?
+        async_update(flow_handler)
+      else
+        result = flow.handle(current_step)
+        end_step(result)
+      end
     end
 
     private
 
+    def async_show(flow_handler)
+      async_state = flow_handler.async_state
+      case async_state.status
+      when :none
+        begin_step
+      when :in_progress
+        redirect_to send(@step_url, step: current_step)
+      when :timed_out
+        begin_step
+      when :done
+        result = flow_handler.after_call(async_state.pii, async_state.result)
+        flow_handler.mark_step_complete(current_step) if result.success?
+        flow_handler.delete_async unless result.success?
+        end_step(result)
+      end
+    end
+
+    def async_update(flow_handler)
+      case flow_handler.async_state.status
+      when :none
+        flow.handle(current_step)
+        # binding.pry
+        flow_handler.mark_step_incomplete(current_step)
+        redirect_to send(@step_url, step: current_step)
+      when :in_progress
+        redirect_to send(@step_url, step: current_step)
+      when :timed_out
+        flow.handle(current_step)
+        flow_handler.mark_step_incomplete(current_step)
+        redirect_to send(@step_url, step: current_step)
+      when :done
+        redirect_to send(@step_url, step: current_step)
+      end
+    end
+
+    def begin_step
+      analytics.track_event(analytics_visited, step: current_step) if @analytics_id
+      Funnel::DocAuth::RegisterStep.new(user_id, issuer).call(current_step, :view, true)
+      register_campaign
+      render_step(current_step, flow.flow_session)
+    end
+
+    def end_step(result)
+      # binding.pry
+      if @analytics_id
+        analytics.track_event(analytics_submitted, result.to_h.merge(step: current_step))
+      end
+      register_update_step(current_step, result)
+      flow_finish and return unless next_step
+
+      render_update(current_step, result)
+    end
+
     def current_step
       params[:step]&.underscore
     end
@@ -58,10 +116,15 @@ def issuer
 
     def fsm_initialize
       klass = self.class
+      flow = klass::FSM_SETTINGS[:flow]
+      @step_url = klass::FSM_SETTINGS[:step_url]
+      @final_url = klass::FSM_SETTINGS[:final_url]
+      @analytics_id = klass::FSM_SETTINGS[:analytics_id]
+      @view = klass::FSM_SETTINGS[:view]
       @name = klass.name.underscore.gsub('_controller', '')
-      klass::FSM_SETTINGS.each { |key, value| instance_variable_set("@#{key}", value) }
+
       current_session[@name] ||= {}
-      @flow = @flow.new(self, current_session, @name)
+      @flow = flow.new(self, current_session, @name)
     end
 
     def render_update(step, result)

app/services/idv/session.rb

@@ -1,8 +1,6 @@
 module Idv
   class Session
     VALID_SESSION_ATTRIBUTES = %i[
-      async_result_id
-      async_result_started_at
       address_verification_mechanism
       applicant
       vendor_phone_confirmation
@@ -15,7 +13,6 @@ class Session
       profile_step_params
       personal_key
       resolution_successful
-      selected_jurisdiction
       step_attempts
     ].freeze
 

app/services/idv/steps/verify_base_step.rb

@@ -11,9 +11,11 @@ class VerifyBaseStep < DocAuthBaseStep
       def perform_resolution_and_check_ssn
         pii_from_doc = flow_session[:pii_from_doc]
         # do resolution first to prevent ssn time/discovery. resolution time order > than db call
-        result = perform_resolution(pii_from_doc)
-        result = check_ssn(pii_from_doc) if result.success?
-        summarize_result_and_throttle_failures(result)
+        idv_result = perform_resolution(pii_from_doc)
+        add_proofing_costs(idv_result)
+        response = idv_result_to_form_response(idv_result)
+        response = check_ssn(pii_from_doc) if response.success?
+        summarize_result_and_throttle_failures(response)
       end
 
       def summarize_result_and_throttle_failures(summary_result)
@@ -53,8 +55,10 @@ def skip_legacy_steps
 
       def perform_resolution(pii_from_doc)
         stages = should_use_aamva?(pii_from_doc) ? %i[resolution state_id] : [:resolution]
-        idv_result = Idv::Agent.new(pii_from_doc).proof(*stages)
-        add_proofing_costs(idv_result)
+        Idv::Agent.new(pii_from_doc).proof(*stages)
+      end
+
+      def idv_result_to_form_response(idv_result)
         FormResponse.new(
           success: idv_success(idv_result),
           errors: idv_errors(idv_result),
@@ -63,6 +67,7 @@ def perform_resolution(pii_from_doc)
       end
 
       def add_proofing_costs(results)
+        # binding.pry
         vendors = results[:context][:stages]
         vendors.each do |hash|
           add_cost(:aamva) if hash[:state_id]

app/services/idv/steps/verify_step.rb

@@ -1,8 +1,88 @@
 module Idv
   module Steps
     class VerifyStep < VerifyBaseStep
+      State = Struct.new(:status, :pii, :result, keyword_init: true) do
+        def self.none
+          new(status: :none)
+        end
+
+        def self.timed_out
+          new(status: :timed_out)
+        end
+
+        def self.in_progress
+          new(status: :in_progress)
+        end
+
+        def self.done(pii:, result:)
+          new(status: :done, pii: pii, result: result)
+        end
+
+        private_class_method :new
+      end
+
       def call
-        perform_resolution_and_check_ssn
+        case async_state.status
+        when :none
+          enqueue_job
+        when :in_progress
+          nil
+        when :timed_out
+          enqueue_job
+        when :done
+          nil
+        end
+      end
+
+      def after_call(pii, idv_result)
+        # binding.pry
+
+        add_proofing_costs(idv_result)
+        response = idv_result_to_form_response(idv_result)
+        response = check_ssn(pii) if response.success?
+        summarize_result_and_throttle_failures(response)
+      end
+
+      def async?
+        true
+      end
+
+      # @return [State]
+      def async_state
+        dcs_uuid = flow_session[:idv_verify_step_document_capture_session_uuid]
+        dcs = DocumentCaptureSession.find_by(uuid: dcs_uuid)
+        return State.none if dcs_uuid.nil?
+        return State.timed_out if dcs.nil?
+
+        proofing_job_result = dcs.load_proofing_result
+        return State.timed_out if proofing_job_result.nil?
+
+        if proofing_job_result.result
+          proofing_job_result.result.deep_symbolize_keys!
+          proofing_job_result.pii.deep_symbolize_keys!
+          State.done(pii: proofing_job_result.pii, result: proofing_job_result.result)
+        elsif dcs.pii
+          State.in_progress
+        end
+      end
+
+      def delete_async
+        flow_session.delete(:idv_verify_step_document_capture_session_uuid)
+      end
+
+      private
+
+      def enqueue_job
+        pii_from_doc = flow_session[:pii_from_doc]
+
+        document_capture_session = DocumentCaptureSession.create(user_id: user_id,
+                                                                 requested_at: Time.zone.now)
+        document_capture_session.store_proofing_pii_from_doc(pii_from_doc)
+
+        flow_session[:idv_verify_step_document_capture_session_uuid] = document_capture_session.uuid
+
+        stages = should_use_aamva?(pii_from_doc) ? %w[resolution state_id] : ['resolution']
+        VendorProofJob.perform_later(document_capture_session.uuid, stages)
       end
     end
   end