Merge branch 'codex/cw-7519-intercom-stalled-retry-15m' into codex/cw-7615-intercom-bulk-message-writes

## Description

Prepares each Intercom conversation's source message and parts as one
deterministic batch before persistence. The builder preserves provider
order, prefetches mappings and live messages once, and classifies every
entry as current-run, previous-run, stale-mapping repair,
existing-message repair, or new.

The importer consumes those prepared entries through the existing
individual transaction and retry path, so activity conversion, private
notes, sender attribution, metadata, timestamps, skip logs, idempotency,
indexing, and heartbeat behavior remain unchanged. This PR intentionally
does not bulk insert messages; batching writes is Phase 2 Task 2.

This is Phase 2, Task 1 (Task 4 overall) of
[CW-7615](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion).

This PR is stacked on #15052 and should be reviewed as the four-file
delta from `codex/cw-7615-intercom-query-timeout-retries`.

## Closes

- Tracking plan:
[CW-7615](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion)

## Type of change

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality not to work as expected)
- [ ] This change requires a documentation update

## How to test

1. Import a conversation containing a source message, comments, notes,
activities, and entries with equal timestamps.
2. Retry the same run and verify messages and counters remain
idempotent.
3. Retry after a previous run and verify existing mappings are reported
as skips without changing their owner.
4. Delete a mapped Message, rerun, and verify the message and mapping
are repaired.
5. Delete only a mapping, rerun, and verify the existing Message is
reused without duplication.
6. Rerun a previously skipped part that now classifies as an activity
and verify it is repaired.
7. Import an empty conversation and verify no message or mapping
prefetch is performed.

## Checklist

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have commented on my code, particularly in hard-to-understand
areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream
modules

app/services/data_imports/intercom/importer.rb

@@ -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?
@@ -9,6 +11,7 @@ class DataImports::Intercom::Importer
   DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze
   CONTACTS_PER_PAGE = 50
   CONVERSATIONS_PER_PAGE = 10
+  MESSAGES_PER_BATCH = 100
   HEARTBEAT_INTERVAL = 1.minute
   QUERY_TIMEOUT_RETRY_LIMIT = 1
   QUERY_TIMEOUT_RETRY_DELAY_RANGE = (0.2..0.5)
@@ -16,9 +19,20 @@ class DataImports::Intercom::Importer
   ALREADY_IMPORTED_ERROR_CODE = 'DataImports::Intercom::AlreadyImported'.freeze
   SKIPPED_MESSAGE_ERROR_CODE = 'DataImports::Intercom::SkippedMessage'.freeze
   TRUNCATED_PARTS_ERROR_CODE = 'DataImports::Intercom::TruncatedConversationParts'.freeze
+  MESSAGE_MAPPING_UNIQUE_INDEX = :idx_data_import_mappings_on_account_and_source
   E164_REGEX = /\A\+[1-9]\d{1,14}\z/
   INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/
 
+  MessageBatchResult = Struct.new(
+    :imported_entries,
+    :skipped_entries,
+    :current_entries,
+    :previous_entries,
+    :messages,
+    :failed_entries,
+    keyword_init: true
+  )
+
   def initialize(data_import:, run_id: nil)
     @data_import = data_import
     @run_id = run_id
@@ -410,19 +424,221 @@ class DataImports::Intercom::Importer
     end
     return import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts.size) if batch.nil?
 
-    batch.source_entries.each { |entry| import_message(chatwoot_conversation, contact, entry) }
     record_truncated_conversation_parts(conversation, parts.size)
 
-    batch.part_entries.each do |entry|
+    batch.entries.each_slice(MESSAGES_PER_BATCH) do |entries|
       return false unless continue_import_with_heartbeat?
 
-      import_message(chatwoot_conversation, contact, entry)
+      import_message_batch(chatwoot_conversation, contact, batch_builder, entries)
+      return false if @import_stopped
     end
     return false if import_stopped?
 
     true
   end
 
+  def import_message_batch(conversation, contact, batch_builder, entries)
+    result = bulk_message_batch_result(conversation, contact, batch_builder, entries)
+    return if result.blank?
+
+    result.current_entries.each do |entry|
+      reconcile_bulk_message_entry(conversation, entry) do
+        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
+      end
+    end
+    result.previous_entries.each do |entry|
+      reconcile_bulk_message_entry(conversation, entry) do
+        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
+      end
+    end
+    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)
+    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?
+
+      message = fallback_message_entry(conversation, contact, batch_builder, entry)
+      reindex_message_for_search(message) if message.is_a?(Message)
+    end
+  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)
+    @data_import.with_lock do
+      if inactive_import_run?
+        @import_stopped = true
+        next
+      end
+
+      refreshed_entries = batch_builder.refresh(entries).entries
+      persist_message_entries(conversation, contact, refreshed_entries)
+    end
+  end
+
+  def inactive_import_run?
+    @data_import.abandoned? || @data_import.failed? || @data_import.completed? ||
+      @data_import.completed_with_errors? || stale_import_run?
+  end
+
+  def persist_message_entries(conversation, contact, entries)
+    grouped_entries = entries.group_by(&:classification)
+    writable_entries = entries.select do |entry|
+      %i[repairable_stale_mapping existing_message new_message].include?(entry.classification)
+    end
+    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(imported_entries, attributes_by_source_id)
+    upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)
+
+    MessageBatchResult.new(
+      imported_entries: imported_entries,
+      skipped_entries: skipped_entries,
+      current_entries: grouped_entries.fetch(:current_import, []),
+      previous_entries: grouped_entries.fetch(:previous_import, []),
+      messages: messages,
+      failed_entries: failed_entries
+    )
+  end
+
+  def insert_messages(entries, attributes_by_source_id)
+    new_entries = entries.reject(&:message)
+    if new_entries.present?
+      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
+
+    entries.map do |entry|
+      entry.message || inserted_messages.fetch("intercom:#{entry.source_id}")
+    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
+
+      timestamp_for(value) if value.present?
+    rescue RangeError
+      raise InvalidMessagePayloadError, "Intercom message #{field} must be a Unix 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|
+      message_mapping_attributes(entry, 'Message', message.id, message_metadata(entry.part), now)
+    end
+    mapping_attributes.concat(skipped_entries.filter_map do |entry|
+      next if entry.mapping
+
+      metadata = message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
+      message_mapping_attributes(entry, 'Conversation', conversation.id, metadata, now)
+    end)
+    return if mapping_attributes.empty?
+
+    DataImportMapping.upsert_all(
+      mapping_attributes,
+      unique_by: MESSAGE_MAPPING_UNIQUE_INDEX,
+      update_only: %i[data_import_id chatwoot_record_type chatwoot_record_id metadata updated_at],
+      record_timestamps: false
+    )
+  end
+
+  def message_mapping_attributes(entry, record_type, record_id, metadata, now)
+    {
+      account_id: @account.id,
+      data_import_id: @data_import.id,
+      source_provider: PROVIDER,
+      source_object_type: 'message',
+      source_object_id: entry.source_id,
+      chatwoot_record_type: record_type,
+      chatwoot_record_id: record_id,
+      metadata: metadata,
+      created_at: entry.mapping&.created_at || now,
+      updated_at: now
+    }
+  end
+
+  def record_bulk_skipped_message(conversation, entry)
+    already_recorded = skip_log_recorded?('message', entry.source_id, SKIPPED_MESSAGE_ERROR_CODE)
+    record_skipped_message_log(conversation, entry.source_id, entry.part)
+    increment_stat('messages', 'skipped') unless already_recorded
+  end
+
+  def reconcile_bulk_message_entry(conversation, entry, &)
+    with_query_timeout_retry(&)
+  rescue StandardError => e
+    fail_message(conversation, entry.source_id, entry.part, e)
+  end
+
   def import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts_count)
     source_entries, part_entries = batch_builder.unprepared_entries.partition { |entry| entry[:part]['part_type'] == 'source' }
     source_entries.each { |entry| import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry) }
@@ -445,19 +661,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
@@ -468,15 +688,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
 
@@ -993,9 +1210,9 @@ class DataImports::Intercom::Importer
     @import_types ||= (@data_import.import_types.presence || DEFAULT_IMPORT_TYPES)
   end
 
-  def increment_stat(group, key)
+  def increment_stat(group, key, amount = 1)
     @stats[group] ||= {}
-    @stats[group][key] = @stats[group][key].to_i + 1
+    @stats[group][key] = @stats[group][key].to_i + amount
   end
 
   def mark_stat_group_dirty(group)

app/services/data_imports/intercom/message_batch_builder.rb

@@ -39,13 +39,12 @@ class DataImports::Intercom::MessageBatchBuilder
   end
 
   def perform(source_entries = unprepared_entries)
-    return Batch.new(items: []) if source_entries.empty?
+    classify(source_entries)
+  end
 
-    mappings = message_mappings(source_entries)
-    messages = messages_for(source_entries, mappings)
-
-    Batch.new(items: source_entries.map do |source_entry|
-      build_entry(source_entry, source_entry[:position], mappings, messages)
+  def refresh(entries)
+    classify(entries.map do |entry|
+      { source_id: entry.source_id, part: entry.part, position: entry.position }
     end)
   end
 
@@ -55,6 +54,17 @@ class DataImports::Intercom::MessageBatchBuilder
 
   private
 
+  def classify(source_entries)
+    return Batch.new(items: []) if source_entries.empty?
+
+    mappings = message_mappings(source_entries)
+    messages = messages_for(source_entries, mappings)
+
+    Batch.new(items: source_entries.map.with_index do |source_entry, position|
+      build_entry(source_entry, source_entry.fetch(:position, position), mappings, messages)
+    end)
+  end
+
   def ordered_source_entries
     entries = []
     source = @source_conversation['source'].to_h

spec/services/data_imports/intercom/importer_spec.rb

@@ -116,6 +116,275 @@ RSpec.describe DataImports::Intercom::Importer do
     expect(DataImportMapping.where(data_import: data_import).count).to eq(5)
   end
 
+  it 'writes a normal conversation with one message insert and one mapping upsert', :aggregate_failures do
+    importer = described_class.new(data_import: data_import)
+    message_batch_sizes = []
+    mapping_batch_sizes = []
+    mapping_upsert_options = []
+    allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
+      message_batch_sizes << records.size
+      method.call(records, **kwargs)
+    end
+    allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
+      mapping_batch_sizes << records.size
+      mapping_upsert_options << kwargs
+      method.call(records, **kwargs)
+    end
+    expect(importer).not_to receive(:create_message)
+
+    importer.perform
+
+    expect(message_batch_sizes).to eq([3])
+    expect(mapping_batch_sizes).to eq([3])
+    expect(mapping_upsert_options).to contain_exactly(
+      include(unique_by: described_class::MESSAGE_MAPPING_UNIQUE_INDEX, record_timestamps: false)
+    )
+    expect(account.messages.count).to eq(3)
+    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+  end
+
+  it 'preserves provider order when imported messages share a timestamp' do
+    equal_timestamp_conversation = conversation_payload.deep_dup
+    equal_timestamp_conversation['conversation_parts']['conversation_parts'].each do |part|
+      part['created_at'] = conversation_payload['created_at']
+      part['updated_at'] = conversation_payload['created_at']
+    end
+    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(equal_timestamp_conversation)
+
+    described_class.new(data_import: data_import).perform
+
+    conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+    expect(conversation.messages.order(:created_at, :id).pluck(:source_id)).to eq(
+      %w[
+        intercom:conversation:conversation_1:source:source_1
+        intercom:conversation:conversation_1:part:part_1
+        intercom:conversation:conversation_1:part:part_2
+      ]
+    )
+  end
+
+  it 'writes conversations in batches of at most 100 messages', :aggregate_failures do
+    bulk_conversation = conversation_payload.deep_dup
+    template_part = bulk_conversation.dig('conversation_parts', 'conversation_parts').first
+    bulk_conversation['conversation_parts']['conversation_parts'] = Array.new(205) do |index|
+      template_part.merge(
+        'id' => "part_#{index + 1}",
+        'created_at' => 1_700_000_100 + index,
+        'updated_at' => 1_700_000_100 + index
+      )
+    end
+    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(bulk_conversation)
+    message_batch_sizes = []
+    mapping_batch_sizes = []
+    allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
+      message_batch_sizes << records.size
+      method.call(records, **kwargs)
+    end
+    allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
+      mapping_batch_sizes << records.size
+      method.call(records, **kwargs)
+    end
+    importer = described_class.new(data_import: data_import)
+    expect(importer).to receive(:update_conversation_activity).once.and_call_original
+
+    importer.import_conversations_page
+
+    expect(message_batch_sizes).to eq([100, 100, 6])
+    expect(mapping_batch_sizes).to eq([100, 100, 6])
+    expect(account.messages.order(:created_at).pick(:source_id)).to eq('intercom:conversation:conversation_1:source:source_1')
+    expect(account.messages.count).to eq(206)
+    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 'isolates an out-of-range message timestamp 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'] = Float::INFINITY
+    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|
@@ -238,13 +507,19 @@ RSpec.describe DataImports::Intercom::Importer do
     allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true)
     allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false)
     reindexed_message_ids = []
+    reindex_transaction_depths = []
+    transaction_depth_before_import = Message.connection.open_transactions
     original_reindex_for_search = Message.instance_method(:reindex_for_search)
-    Message.define_method(:reindex_for_search) { reindexed_message_ids << id }
+    Message.define_method(:reindex_for_search) do
+      reindexed_message_ids << id
+      reindex_transaction_depths << self.class.connection.open_transactions
+    end
     Message.__send__(:private, :reindex_for_search)
 
     described_class.new(data_import: data_import).perform
 
     expect(reindexed_message_ids).to match_array(Message.where(account_id: account.id).pluck(:id))
+    expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import))
   ensure
     Message.define_method(:reindex_for_search, original_reindex_for_search)
     Message.__send__(:private, :reindex_for_search)
@@ -320,12 +595,12 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.reload.cursor.dig('conversations', 'starting_after')).to be_nil
     end
 
-    it 'heartbeats at most once per minute while processing conversation parts' do
+    it 'heartbeats at most once per minute while processing conversation message batches' do
       freeze_time do
         started_at = Time.current
         long_conversation = conversation_payload.deep_dup
         template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first
-        long_conversation['conversation_parts']['conversation_parts'] = Array.new(5) do |index|
+        long_conversation['conversation_parts']['conversation_parts'] = Array.new(400) do |index|
           template_part.merge('id' => "part_#{index + 1}")
         end
         allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation)
@@ -335,41 +610,62 @@ RSpec.describe DataImports::Intercom::Importer do
           method.call
         end
         importer = described_class.new(data_import: data_import)
-        allow(importer).to receive(:create_message) { travel 30.seconds }
-
-        importer.import_conversations_page
-
-        expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes])
-      end
-    end
-
-    it 'stops parts without heartbeating or persisting stale stats when a newer run takes over', :aggregate_failures do
-      freeze_time do
-        run_id = 'intercom-run-1'
-        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-        allow(data_import).to receive(:touch).and_call_original
-        importer = described_class.new(data_import: data_import, run_id: run_id)
-        allow(importer).to receive(:create_message) do
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        empty_result = described_class::MessageBatchResult.new(
+          imported_entries: [],
+          skipped_entries: [],
+          current_entries: [],
+          previous_entries: [],
+          messages: []
+        )
+        allow(importer).to receive(:bulk_write_message_entries) do
+          travel 30.seconds
+          empty_result
         end
 
         importer.import_conversations_page
 
-        expect(importer).to have_received(:create_message).once
-        expect(data_import).not_to have_received(:touch)
-        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
+        expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes])
+        expect(importer).to have_received(:bulk_write_message_entries).exactly(5).times
       end
     end
 
-    it 'does not persist stale stats when a newer run takes over during the final part', :aggregate_failures do
+    it 'stops batches without heartbeating or persisting stale stats when a newer run takes over', :aggregate_failures do
+      freeze_time do
+        run_id = 'intercom-run-1'
+        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
+        long_conversation = conversation_payload.deep_dup
+        template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first
+        long_conversation['conversation_parts']['conversation_parts'] = Array.new(100) do |index|
+          template_part.merge('id' => "part_#{index + 1}")
+        end
+        allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation)
+        allow(data_import).to receive(:touch).and_call_original
+        importer = described_class.new(data_import: data_import, run_id: run_id)
+        batch_write_count = 0
+        allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args|
+          result = method.call(*args)
+          batch_write_count += 1
+          if batch_write_count == 1
+            DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+          end
+          result
+        end
+
+        importer.import_conversations_page
+
+        expect(importer).to have_received(:bulk_write_message_entries).once
+        expect(data_import).not_to have_received(:touch)
+        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
+        expect(account.messages.count).to eq(100)
+      end
+    end
+
+    it 'does not persist stale stats when a newer run takes over during the final batch', :aggregate_failures do
       run_id = 'intercom-run-1'
       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(:create_message).and_wrap_original do |method, *args|
+      allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args|
         method.call(*args).tap do
-          entry = args[2]
-          next unless entry.part['id'] == 'part_2'
-
           DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
         end
       end
@@ -381,15 +677,69 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
     end
 
+    it 'does not persist stale stats when a newer run takes over during the final prefetch fallback entry', :aggregate_failures do
+      run_id = 'intercom-run-1'
+      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(:sleep)
+      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(:perform).and_wrap_original do |perform, *args|
+            raise ActiveRecord::QueryCanceled, 'statement timeout' if args.empty?
+
+            perform.call(*args)
+          end
+        end
+      end
+      allow(importer).to receive(:import_unprepared_message).and_wrap_original do |method, *args|
+        method.call(*args).tap do
+          source_entry = args.last
+          next unless source_entry.dig(:part, 'id') == 'part_2'
+
+          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        end
+      end
+      expect(importer).not_to receive(:update_conversation_activity)
+
+      importer.import_conversations_page
+
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(importer).to have_received(:import_unprepared_message).exactly(3).times
+      expect(account.messages.count).to eq(3)
+      expect(data_import.reload.stats).to include(
+        'conversations' => include('imported' => 0),
+        'messages' => include('imported' => 0)
+      )
+    end
+
+    it 'stops individual fallback entries when a newer run takes over', :aggregate_failures do
+      run_id = 'intercom-run-1'
+      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(: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
+      end
+
+      importer.import_conversations_page
+
+      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
+
     it 'reconciles conversation stats when a superseded run is retried', :aggregate_failures do
       freeze_time do
         run_id = 'intercom-run-1'
         next_run_id = 'intercom-run-2'
         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(:create_message) do
-          data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
-          travel 1.minute
+        allow(importer).to receive(:bulk_write_message_entries).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 => next_run_id })
+          end
         end
 
         importer.import_conversations_page
@@ -441,7 +791,12 @@ 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(importer).to receive(:record_message_mapping).and_raise(StandardError, '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'
+
+        method.call(entry, message)
+      end
 
       importer.import_conversations_page
 
@@ -481,6 +836,7 @@ RSpec.describe DataImports::Intercom::Importer do
       importer = described_class.new(data_import: data_import)
       mapping_attempts = 0
       target_source_id = 'conversation:conversation_1:part:part_1'
+      allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable')
       allow(importer).to receive(:sleep)
       allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
         if entry.source_id == target_source_id
@@ -506,6 +862,7 @@ RSpec.describe DataImports::Intercom::Importer do
       importer = described_class.new(data_import: data_import)
       mapping_attempts = 0
       target_source_id = 'conversation:conversation_1:part:part_1'
+      allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable')
       allow(importer).to receive(:sleep)
       allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
         if entry.source_id == target_source_id
@@ -612,6 +969,9 @@ RSpec.describe DataImports::Intercom::Importer do
         'message' => 3
       )
       expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
+      expect(
+        DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message').distinct.pluck(:data_import_id)
+      ).to eq([data_import.id])
     end
 
     it 'reconciles skipped stats when a superseded run is retried', :aggregate_failures do
@@ -688,6 +1048,7 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(next_data_import.import_errors.skip_logs.where(source_object_type: 'message')).to be_empty
       message_mappings = DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message')
       expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3)
+      expect(message_mappings.distinct.pluck(:data_import_id)).to eq([next_data_import.id])
     end
 
     it 'repairs a missing mapping without recreating the existing message', :aggregate_failures do
@@ -1031,6 +1392,42 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1)
     end
 
+    it 'retries skip-log reconciliation after a query timeout', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      attempts = 0
+      allow(importer).to receive(:sleep)
+      allow(importer).to receive(:record_skipped_message_log).and_wrap_original do |method, *args|
+        attempts += 1
+        raise ActiveRecord::QueryCanceled, 'statement timeout' if attempts == 1
+
+        method.call(*args)
+      end
+
+      importer.perform
+
+      expect(attempts).to eq(2)
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(data_import.reload).to be_completed
+      expect(data_import.stats.dig('messages', 'skipped')).to eq(1)
+    end
+
+    it 'isolates a persistent skip-log reconciliation failure to the message', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+      allow(importer).to receive(:record_skipped_message_log).and_raise(ActiveRecord::StatementInvalid, 'skip log failed')
+
+      importer.perform
+
+      item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1')
+      error = data_import.import_errors.find_by!(
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:blank_part'
+      )
+      expect(item).to be_imported
+      expect(error).to have_attributes(error_code: 'ActiveRecord::StatementInvalid', message: 'skip log failed')
+      expect(data_import.reload).to be_completed_with_errors
+      expect(data_import.stats.dig('errors', 'count')).to eq(1)
+    end
+
     it 'records the skip log again for a later import run', :aggregate_failures do
       described_class.new(data_import: data_import).perform
       next_data_import = create(
@@ -1172,11 +1569,8 @@ RSpec.describe DataImports::Intercom::Importer do
       next_run_id = 'intercom-run-2'
       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(:import_message).and_wrap_original do |method, *args|
+      allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args|
         method.call(*args).tap do
-          entry = args[2]
-          next unless entry.part['id'] == 'part_2'
-
           DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
         end
       end
@@ -1260,8 +1654,8 @@ RSpec.describe DataImports::Intercom::Importer do
 
     before do
       allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
-        if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part'
-          insert_attempts << records.first[:source_id]
+        if records.any? { |record| record[:source_id] == 'intercom:conversation:conversation_1:part:bad_part' }
+          insert_attempts << 'intercom:conversation:conversation_1:part:bad_part'
           raise ActiveRecord::StatementInvalid, 'bad message'
         end
 
@@ -1284,7 +1678,9 @@ 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.one?).to be(true)
+      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
 end

spec/services/data_imports/intercom/message_batch_builder_spec.rb

@@ -73,6 +73,33 @@ RSpec.describe DataImports::Intercom::MessageBatchBuilder do
     expect(builder.perform.entries).to be_empty
   end
 
+  it 'refreshes classifications while preserving source positions' do
+    batch = builder.perform
+    target_entry = batch.entries.second
+    message = create(
+      :message,
+      account: account,
+      conversation: conversation,
+      inbox: conversation.inbox,
+      source_id: "intercom:#{target_entry.source_id}"
+    )
+    DataImportMapping.create!(
+      account: account,
+      data_import: data_import,
+      source_provider: 'intercom',
+      source_object_type: 'message',
+      source_object_id: target_entry.source_id,
+      chatwoot_record_type: 'Message',
+      chatwoot_record_id: message.id,
+      metadata: {}
+    )
+
+    refreshed_batch = builder.refresh(batch.entries)
+
+    expect(refreshed_batch.entries.map(&:position)).to eq([0, 1, 2])
+    expect(refreshed_batch.entries.second).to have_attributes(classification: :current_import, message: message)
+  end
+
   it 'classifies a live mapping from the current import as already handled' do
     message = create(
       :message,