Merge remote-tracking branch 'origin/develop' into feature/cw-7513

# Conflicts:
#	app/javascript/dashboard/featureFlags.js
#	config/features.yml

app/models/data_import.rb

@@ -3,33 +3,115 @@
 # 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_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
+  LEGACY_DATA_TYPES = ['contacts'].freeze
+  INTEGRATION_DATA_TYPES = ['intercom'].freeze
+  IMPORT_TYPES = %w[contacts conversations].freeze
+
   belongs_to :account
-  validates :data_type, inclusion: { in: ['contacts'], message: I18n.t('errors.data_import.data_type.invalid') }
-  enum status: { pending: 0, processing: 1, completed: 2, failed: 3 }
+  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 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

app/models/data_import_error.rb

@@ -0,0 +1,33 @@
+# == Schema Information
+#
+# Table name: data_import_errors
+#
+#  id                  :bigint           not null, primary key
+#  details             :jsonb            not null
+#  error_code          :string           not null
+#  message             :text
+#  source_object_type  :string
+#  created_at          :datetime         not null
+#  updated_at          :datetime         not null
+#  data_import_id      :bigint           not null
+#  data_import_item_id :bigint
+#  source_object_id    :string
+#
+# Indexes
+#
+#  idx_data_import_errors_on_source                 (source_object_type,source_object_id)
+#  index_data_import_errors_on_data_import_id       (data_import_id)
+#  index_data_import_errors_on_data_import_item_id  (data_import_item_id)
+#
+class DataImportError < ApplicationRecord
+  SKIP_LOG_KINDS = %w[failed skipped].freeze
+
+  belongs_to :data_import
+  belongs_to :data_import_item, optional: true
+
+  validates :error_code, presence: true
+
+  scope :skip_logs, -> { where("details ->> 'kind' IN (:kinds)", kinds: SKIP_LOG_KINDS) }
+  scope :failed, -> { where("details ->> 'kind' = ?", 'failed') }
+  scope :non_skip_logs, -> { where("details ->> 'kind' IS NULL OR details ->> 'kind' NOT IN (:kinds)", kinds: SKIP_LOG_KINDS) }
+end

app/models/data_import_item.rb

@@ -0,0 +1,35 @@
+# == Schema Information
+#
+# Table name: data_import_items
+#
+#  id                   :bigint           not null, primary key
+#  attempt_count        :integer          default(0), not null
+#  chatwoot_record_type :string
+#  last_error_code      :string
+#  last_error_message   :text
+#  metadata             :jsonb            not null
+#  source_object_type   :string           not null
+#  source_provider      :string           not null
+#  status               :integer          default("pending"), not null
+#  created_at           :datetime         not null
+#  updated_at           :datetime         not null
+#  chatwoot_record_id   :bigint
+#  data_import_id       :bigint           not null
+#  source_object_id     :string           not null
+#
+# Indexes
+#
+#  idx_data_import_items_on_import_and_source  (data_import_id,source_object_type,source_object_id) UNIQUE
+#  idx_data_import_items_on_record             (chatwoot_record_type,chatwoot_record_id)
+#  idx_data_import_items_on_source             (source_provider,source_object_type,source_object_id)
+#  index_data_import_items_on_data_import_id   (data_import_id)
+#
+class DataImportItem < ApplicationRecord
+  belongs_to :data_import
+  has_many :import_errors, class_name: 'DataImportError', dependent: :destroy_async
+
+  validates :source_provider, :source_object_type, :source_object_id, presence: true
+  validates :source_object_id, uniqueness: { scope: [:data_import_id, :source_object_type] }
+
+  enum status: { pending: 0, processing: 1, imported: 2, skipped: 3, failed: 4 }
+end

app/models/data_import_mapping.rb

@@ -0,0 +1,33 @@
+# == Schema Information
+#
+# Table name: data_import_mappings
+#
+#  id                   :bigint           not null, primary key
+#  chatwoot_record_type :string           not null
+#  metadata             :jsonb            not null
+#  source_object_type   :string           not null
+#  source_provider      :string           not null
+#  created_at           :datetime         not null
+#  updated_at           :datetime         not null
+#  account_id           :integer          not null
+#  chatwoot_record_id   :bigint           not null
+#  data_import_id       :bigint           not null
+#  source_object_id     :string           not null
+#
+# Indexes
+#
+#  idx_data_import_mappings_on_account_and_source  (account_id,source_provider,source_object_type,source_object_id) UNIQUE
+#  idx_data_import_mappings_on_record              (chatwoot_record_type,chatwoot_record_id)
+#  index_data_import_mappings_on_data_import_id    (data_import_id)
+#
+class DataImportMapping < ApplicationRecord
+  belongs_to :data_import
+  belongs_to :account
+
+  validates :source_provider, :source_object_type, :source_object_id, :chatwoot_record_type, :chatwoot_record_id, presence: true
+  validates :source_object_id, uniqueness: { scope: [:account_id, :source_provider, :source_object_type] }
+
+  def chatwoot_record
+    chatwoot_record_type.constantize.find_by(id: chatwoot_record_id)
+  end
+end