moreminimore-chat/app/models/data_import.rb

# == Schema Information
#
# Table name: data_imports
#
#  id                :bigint           not null, primary key
#  abandoned_at      :datetime
#  access_token      :text
#  completed_at      :datetime
#  cursor            :jsonb            not null
#  data_type         :string           not null
#  import_types      :jsonb            not null
#  last_error_at     :datetime
#  name              :string
#  processed_records :integer
#  processing_errors :text
#  source_metadata   :jsonb            not null
#  source_provider   :string
#  source_type       :string
#  started_at        :datetime
#  stats             :jsonb            not null
#  status            :integer          default("pending"), not null
#  total_records     :integer
#  created_at        :datetime         not null
#  updated_at        :datetime         not null
#  account_id        :bigint           not null
#  initiated_by_id   :integer
#
# Indexes
#
#  index_data_imports_on_account_id       (account_id)
#  index_data_imports_on_initiated_by_id  (initiated_by_id)
#  index_data_imports_on_source_provider  (source_provider)
#
class DataImport < ApplicationRecord
  ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY = 'active_intercom_import_run_id'.freeze
  INTERCOM_STALLED_AFTER = 15.minutes
  LEGACY_DATA_TYPES = ['contacts'].freeze
  INTEGRATION_DATA_TYPES = ['intercom'].freeze
  IMPORT_TYPES = %w[contacts conversations].freeze

  belongs_to :account
  belongs_to :initiated_by, class_name: 'User', optional: true

  encrypts :access_token if Chatwoot.encryption_configured?

  has_many :items, class_name: 'DataImportItem', dependent: :destroy_async
  has_many :mappings, class_name: 'DataImportMapping', dependent: :destroy_async
  has_many :import_errors, class_name: 'DataImportError', dependent: :destroy_async

  validates :data_type, inclusion: { in: LEGACY_DATA_TYPES + INTEGRATION_DATA_TYPES, message: I18n.t('errors.data_import.data_type.invalid') }
  validates :access_token, presence: true, on: :create, if: :intercom_import?
  validate :validate_import_types

  enum status: { pending: 0, processing: 1, completed: 2, failed: 3, completed_with_errors: 6, abandoned: 7 }

  scope :active_intercom, -> { where(data_type: 'intercom', source_provider: 'intercom', status: [:pending, :processing]) }

  has_one_attached :import_file
  has_one_attached :failed_records

  after_create_commit :process_data_import

  def legacy_contacts_csv_import?
    data_type == 'contacts' && source_provider.blank?
  end

  def intercom_import?
    data_type == 'intercom' && source_provider == 'intercom'
  end

  def restartable?
    failed? || abandoned?
  end

  def stalled?
    intercom_import? && (pending? || processing?) && updated_at <= INTERCOM_STALLED_AFTER.ago
  end

  def abandonable?
    intercom_import? && (pending? || processing?)
  end

  def abandon!
    self.class.transaction do
      abandonable_import = self.class.lock.find_by(
        id: id,
        data_type: 'intercom',
        source_provider: 'intercom',
        status: [:pending, :processing]
      )
      abandonable_import&.update!(status: :abandoned, abandoned_at: Time.current)
    end
    reload
  end

  def active_intercom_import_run_id
    source_metadata.to_h[ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY]
  end

  def assign_active_intercom_import_run_id
    self.source_metadata = source_metadata.to_h.merge(ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => SecureRandom.uuid)
    active_intercom_import_run_id
  end

  private

  def process_data_import
    return unless legacy_contacts_csv_import?

    # we wait for the file to be uploaded to the cloud
    DataImportJob.set(wait: 1.minute).perform_later(self)
  end

  def validate_import_types
    return if import_types.blank?

    invalid_types = import_types - IMPORT_TYPES
    return if invalid_types.blank?

    errors.add(:import_types, "contains unsupported values: #{invalid_types.join(', ')}")
  end
end