fix: preserve Intercom bulk failure isolation (#15116)

This preserves per-message failure isolation when an Intercom bulk message chunk cannot be committed. Query timeouts are retried once after the failed transaction rolls back, while other database failures move directly to an entry-by-entry fallback. The fallback reacquires the import lock and refreshes each entry before writing, so a competing worker cannot create duplicate messages. Each individual write uses a savepoint, and search indexing happens only after the import lock transaction commits. This is Phase 2, Task 6 of CW-7615 and is stacked on #15111. The review delta is one commit across the importer and its focused spec. ## Closes - [CW-7615](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion) ## How to test 1. Run an Intercom import where a bulk mapping write times out once and confirm the chunk succeeds on retry. 2. Force a persistent timeout or database write failure and confirm valid messages are imported through the fallback. 3. Force one fallback record to fail and confirm only that message is recorded as an import error. 4. Simulate another worker repairing a message before fallback and confirm no duplicate is created.
2026-07-23 22:52:06 +05:30
parent d38ef6c9c8
commit 580cbf91c5
2 changed files with 267 additions and 32 deletions
@@ -1,5 +1,7 @@
 # rubocop:disable Metrics/ClassLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength, Rails/SkipsModelValidations
 class DataImports::Intercom::Importer
+  class InvalidMessagePayloadError < StandardError; end
+
  PageResult = Struct.new(:next_cursor, keyword_init: true) do
    def done?
      next_cursor.blank?
@@ -27,6 +29,7 @@ class DataImports::Intercom::Importer
    :current_entries,
    :previous_entries,
    :messages,
+    :failed_entries,
    keyword_init: true
  )

@@ -450,19 +453,43 @@ class DataImports::Intercom::Importer
    result.skipped_entries.each do |entry|
      reconcile_bulk_message_entry(conversation, entry) { record_bulk_skipped_message(conversation, entry) }
    end
+    Array(result.failed_entries).each { |entry, error| fail_message(conversation, entry.source_id, entry.part, error) }
    increment_stat('messages', 'imported', result.imported_entries.size)
    result.messages.each { |message| reindex_message_for_search(message) }
  end

  def bulk_message_batch_result(conversation, contact, batch_builder, entries)
-    bulk_write_message_entries(conversation, contact, batch_builder, entries)
-  rescue StandardError
+    with_query_timeout_retry do
+      bulk_write_message_entries(conversation, contact, batch_builder, entries)
+    end
+  rescue ActiveRecord::ActiveRecordError
+    fallback_message_entries(conversation, contact, batch_builder, entries) unless @import_stopped
+    nil
+  end
+
+  def fallback_message_entries(conversation, contact, batch_builder, entries)
    entries.each do |entry|
      break unless continue_import_with_heartbeat?

-      import_message(conversation, contact, entry)
+      message = fallback_message_entry(conversation, contact, batch_builder, entry)
+      reindex_message_for_search(message) if message.is_a?(Message)
    end
-    nil
+  end
+
+  def fallback_message_entry(conversation, contact, batch_builder, entry)
+    with_query_timeout_retry do
+      @data_import.with_lock do
+        if inactive_import_run?
+          @import_stopped = true
+          next
+        end
+
+        refreshed_entry = batch_builder.refresh([entry]).entries.first
+        import_message(conversation, contact, refreshed_entry, reindex: false)
+      end
+    end
+  rescue ActiveRecord::ActiveRecordError => e
+    fail_message(conversation, entry.source_id, entry.part, e)
  end

  def bulk_write_message_entries(conversation, contact, batch_builder, entries)
@@ -487,9 +514,23 @@ class DataImports::Intercom::Importer
    writable_entries = entries.select do |entry|
      %i[repairable_stale_mapping existing_message new_message].include?(entry.classification)
    end
-    content_by_source_id = writable_entries.to_h { |entry| [entry.source_id, content_for(entry.part)] }
+    content_by_source_id = {}
+    attributes_by_source_id = {}
+    failed_entries = []
+    writable_entries.select! do |entry|
+      validate_message_payload!(entry.part)
+      content = content_for(entry.part)
+      content_by_source_id[entry.source_id] = content
+      if content.present? && entry.message.blank?
+        attributes_by_source_id[entry.source_id] = message_attributes(conversation, contact, entry.part, entry.source_id, content)
+      end
+      true
+    rescue InvalidMessagePayloadError => e
+      failed_entries << [entry, e]
+      false
+    end
    skipped_entries, imported_entries = writable_entries.partition { |entry| content_by_source_id[entry.source_id].blank? }
-    messages = insert_messages(conversation, contact, imported_entries, content_by_source_id)
+    messages = insert_messages(imported_entries, attributes_by_source_id)
    upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)

    MessageBatchResult.new(
@@ -497,16 +538,15 @@ class DataImports::Intercom::Importer
      skipped_entries: skipped_entries,
      current_entries: grouped_entries.fetch(:current_import, []),
      previous_entries: grouped_entries.fetch(:previous_import, []),
-      messages: messages
+      messages: messages,
+      failed_entries: failed_entries
    )
  end

-  def insert_messages(conversation, contact, entries, content_by_source_id)
+  def insert_messages(entries, attributes_by_source_id)
    new_entries = entries.reject(&:message)
    if new_entries.present?
-      attributes = new_entries.map do |entry|
-        message_attributes(conversation, contact, entry.part, entry.source_id, content_by_source_id.fetch(entry.source_id))
-      end
+      attributes = new_entries.map { |entry| attributes_by_source_id.fetch(entry.source_id) }
      result = Message.insert_all!(attributes, returning: %w[id source_id])
      inserted_messages = Message.where(id: result.pluck('id')).index_by(&:source_id)
    end
@@ -516,6 +556,36 @@ class DataImports::Intercom::Importer
    end
  end

+  def validate_message_payload!(part)
+    raise InvalidMessagePayloadError, 'Intercom message payload must be an object' unless part.is_a?(Hash)
+
+    %w[author assigned_to event_details].each do |field|
+      value = part[field]
+      next if value.nil? || value.is_a?(Hash)
+
+      raise InvalidMessagePayloadError, "Intercom message #{field} must be an object"
+    end
+
+    participant = part.dig('event_details', 'participant')
+    unless participant.nil? || participant.is_a?(Hash)
+      raise InvalidMessagePayloadError, 'Intercom message event_details.participant must be an object'
+    end
+
+    %w[created_at updated_at].each do |field|
+      value = part[field]
+      valid_timestamp = value.nil? || value.is_a?(Integer) || value.is_a?(Float) ||
+                        (value.is_a?(String) && (value.blank? || value.match?(/\A-?\d+(?:\.\d+)?\z/)))
+      raise InvalidMessagePayloadError, "Intercom message #{field} must be a Unix timestamp" unless valid_timestamp
+    end
+
+    %w[body subject].each do |field|
+      value = part[field]
+      next unless value.is_a?(String) && !value.valid_encoding?
+
+      raise InvalidMessagePayloadError, "Intercom message #{field} must use valid encoding"
+    end
+  end
+
  def upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)
    now = Time.current
    mapping_attributes = imported_entries.zip(messages).map do |entry, message|
@@ -584,19 +654,23 @@ class DataImports::Intercom::Importer
    fail_message(conversation, source_entry[:source_id], source_entry[:part], e)
  end

-  def import_message(conversation, contact, entry)
-    with_query_timeout_retry do
-      case entry.classification
-      when :current_import
-        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
-      when :previous_import
-        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
-      when :repairable_stale_mapping, :existing_message, :new_message
-        create_message(conversation, contact, entry)
-      else
-        raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
+  def import_message(conversation, contact, entry, reindex: true)
+    message = with_query_timeout_retry do
+      Message.transaction(requires_new: true) do
+        case entry.classification
+        when :current_import
+          reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
+        when :previous_import
+          skip_existing_message_mapping(conversation, entry.mapping, entry.part)
+        when :repairable_stale_mapping, :existing_message, :new_message
+          create_message(conversation, contact, entry)
+        else
+          raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
+        end
      end
    end
+    reindex_message_for_search(message) if reindex && message.is_a?(Message)
+    message
  rescue StandardError => e
    fail_message(conversation, entry.source_id, entry.part, e)
  end
@@ -607,15 +681,12 @@ class DataImports::Intercom::Importer

    attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
    message = entry.message
-    Message.transaction do
-      unless message
-        result = Message.insert_all!([attrs], returning: %w[id])
-        message = Message.find(result.rows.first.first)
-      end
-      record_message_mapping(entry, message)
+    unless message
+      result = Message.insert_all!([attrs], returning: %w[id])
+      message = Message.find(result.rows.first.first)
    end
+    record_message_mapping(entry, message)
    increment_stat('messages', 'imported')
-    reindex_message_for_search(message)
    message
  end

@@ -196,6 +196,169 @@ RSpec.describe DataImports::Intercom::Importer do
    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(206)
  end

+  context 'when a bulk message chunk fails' do
+    it 'retries a query timeout once and keeps the successful bulk result', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      mapping_attempts = 0
+      allow(importer).to receive(:sleep)
+      allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
+        mapping_attempts += 1
+        raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
+
+        method.call(records, **kwargs)
+      end
+      expect(importer).not_to receive(:fallback_message_entries)
+
+      importer.perform
+
+      expect(mapping_attempts).to eq(2)
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(account.messages.count).to eq(3)
+      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
+      expect(data_import.import_errors).to be_empty
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+    end
+
+    it 'falls back individually after the query timeout retry is exhausted', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      mapping_attempts = 0
+      reindex_transaction_depths = []
+      transaction_depth_before_import = Message.connection.open_transactions
+      allow(importer).to receive(:sleep)
+      allow(importer).to receive(:fallback_message_entries).and_call_original
+      allow(importer).to receive(:create_message).and_call_original
+      allow(importer).to receive(:reindex_message_for_search).and_wrap_original do |method, message|
+        reindex_transaction_depths << Message.connection.open_transactions
+        method.call(message)
+      end
+      allow(DataImportMapping).to receive(:upsert_all) do
+        mapping_attempts += 1
+        raise ActiveRecord::QueryCanceled, 'statement timeout'
+      end
+
+      importer.perform
+
+      expect(mapping_attempts).to eq(2)
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(importer).to have_received(:fallback_message_entries).once
+      expect(importer).to have_received(:create_message).exactly(3).times
+      expect(account.messages.count).to eq(3)
+      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
+      expect(data_import.import_errors).to be_empty
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+      expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import))
+    end
+
+    it 'falls back immediately for a non-timeout database error', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      mapping_attempts = 0
+      allow(importer).to receive(:sleep)
+      allow(importer).to receive(:fallback_message_entries).and_call_original
+      allow(DataImportMapping).to receive(:upsert_all) do
+        mapping_attempts += 1
+        raise ActiveRecord::StatementInvalid, 'bulk mapping failed'
+      end
+
+      importer.perform
+
+      expect(mapping_attempts).to eq(1)
+      expect(importer).not_to have_received(:sleep)
+      expect(importer).to have_received(:fallback_message_entries).once
+      expect(account.messages.count).to eq(3)
+      expect(data_import.import_errors).to be_empty
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+    end
+
+    it 'refreshes fallback entries when another worker repairs a message', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
+      allow(importer).to receive(:fallback_message_entries).and_wrap_original do |method, conversation, contact, batch_builder, entries|
+        source_entry = entries.first
+        message = create(
+          :message,
+          account: account,
+          inbox: conversation.inbox,
+          conversation: conversation,
+          source_id: "intercom:#{source_entry.source_id}",
+          created_at: Time.zone.at(source_entry.part['created_at']),
+          updated_at: Time.zone.at(source_entry.part['created_at'])
+        )
+        DataImportMapping.create!(
+          account: account,
+          data_import: data_import,
+          source_provider: 'intercom',
+          source_object_type: 'message',
+          source_object_id: source_entry.source_id,
+          chatwoot_record_type: 'Message',
+          chatwoot_record_id: message.id,
+          metadata: {}
+        )
+        method.call(conversation, contact, batch_builder, entries)
+      end
+
+      importer.perform
+
+      source_id = 'intercom:conversation:conversation_1:source:source_1'
+      expect(account.messages.where(source_id: source_id).count).to eq(1)
+      expect(account.messages.count).to eq(3)
+      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
+      expect(data_import.import_errors).to be_empty
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+    end
+
+    it 'retries a fallback refresh timeout after the lock transaction rolls back', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      refresh_attempts = 0
+      allow(importer).to receive(:sleep)
+      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
+      allow(DataImports::Intercom::MessageBatchBuilder).to receive(:new).and_wrap_original do |method, **kwargs|
+        method.call(**kwargs).tap do |batch_builder|
+          allow(batch_builder).to receive(:refresh).and_wrap_original do |refresh, entries|
+            refresh_attempts += 1
+            raise ActiveRecord::QueryCanceled, 'statement timeout' if refresh_attempts == 1
+
+            refresh.call(entries)
+          end
+        end
+      end
+
+      importer.perform
+
+      expect(refresh_attempts).to eq(4)
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(account.messages.count).to eq(3)
+      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
+      expect(data_import.import_errors).to be_empty
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+    end
+  end
+
+  it 'isolates a malformed message payload while importing valid messages', :aggregate_failures do
+    malformed_conversation = conversation_payload.deep_dup
+    malformed_conversation['source']['subject'] = nil
+    malformed_conversation['source']['body'] = nil
+    malformed_conversation.dig('conversation_parts', 'conversation_parts').first['created_at'] = { 'unexpected' => true }
+    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(malformed_conversation)
+    importer = described_class.new(data_import: data_import)
+    expect(importer).not_to receive(:fallback_message_entries)
+
+    importer.perform
+
+    expect(account.messages.pluck(:source_id)).to eq(['intercom:conversation:conversation_1:part:part_2'])
+    error = data_import.import_errors.find_by!(
+      source_object_type: 'message',
+      source_object_id: 'conversation:conversation_1:part:part_1'
+    )
+    expect(error).to have_attributes(
+      error_code: described_class::InvalidMessagePayloadError.name,
+      message: 'Intercom message created_at must be a Unix timestamp'
+    )
+    expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1)
+    expect(data_import.reload).to be_completed_with_errors
+    expect(data_import.stats.dig('messages', 'imported')).to eq(1)
+    expect(data_import.stats.dig('errors', 'count')).to eq(1)
+  end
+
  it 'imports historical records without dispatching record events or outbound side effects', :aggregate_failures do
    dispatched_events = []
    allow(Rails.configuration.dispatcher).to receive(:dispatch) do |event_name, *_args|
@@ -493,7 +656,7 @@ RSpec.describe DataImports::Intercom::Importer do
      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
      importer = described_class.new(data_import: data_import, run_id: run_id)
      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
-      allow(importer).to receive(:import_message).and_wrap_original do |method, *args|
+      allow(importer).to receive(:fallback_message_entry).and_wrap_original do |method, *args|
        method.call(*args).tap do
          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
        end
@@ -501,7 +664,7 @@ RSpec.describe DataImports::Intercom::Importer do

      importer.import_conversations_page

-      expect(importer).to have_received(:import_message).once
+      expect(importer).to have_received(:fallback_message_entry).once
      expect(account.messages.count).to eq(1)
      expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
    end
@@ -567,7 +730,7 @@ RSpec.describe DataImports::Intercom::Importer do

    it 'rolls back a newly inserted message when mapping persistence fails', :aggregate_failures do
      importer = described_class.new(data_import: data_import)
-      allow(DataImportMapping).to receive(:upsert_all).and_raise(StandardError, 'bulk mapping failed')
+      allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk mapping failed')
      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
        raise StandardError, 'mapping failed' if entry.source_id == 'conversation:conversation_1:source:source_1'

@@ -1430,6 +1593,7 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(data_import.reload).to be_completed_with_errors
      expect(data_import.stats.dig('errors', 'count')).to eq(1)
      expect(insert_attempts.size).to eq(2)
+      expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1)
      expect(account.messages.find_by(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_present
    end
  end