fix(imports): harden bulk message recovery

app/services/data_imports/intercom/importer.rb

@@ -438,12 +438,18 @@ class DataImports::Intercom::Importer
     return if result.blank?
 
     result.current_entries.each do |entry|
-      reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
+      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|
-      skip_existing_message_mapping(conversation, entry.mapping, entry.part)
+      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
-    result.skipped_entries.each { |entry| record_bulk_skipped_message(conversation, entry) }
     increment_stat('messages', 'imported', result.imported_entries.size)
     result.messages.each { |message| reindex_message_for_search(message) }
   end
@@ -451,7 +457,11 @@ class DataImports::Intercom::Importer
   def bulk_message_batch_result(conversation, contact, batch_builder, entries)
     bulk_write_message_entries(conversation, contact, batch_builder, entries)
   rescue StandardError
-    entries.each { |entry| import_message(conversation, contact, entry) } unless @import_stopped
+    entries.each do |entry|
+      break unless continue_import_with_heartbeat?
+
+      import_message(conversation, contact, entry)
+    end
     nil
   end
 
@@ -548,6 +558,12 @@ class DataImports::Intercom::Importer
     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) }

spec/services/data_imports/intercom/importer_spec.rb

@@ -488,6 +488,24 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(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(:import_message).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(:import_message).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'
@@ -1148,6 +1166,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(