fix(imports): isolate Intercom prefetch failures

app/services/data_imports/intercom/importer.rb

@@ -392,13 +392,17 @@ class DataImports::Intercom::Importer
   def import_conversation_messages(conversation, chatwoot_conversation, contact)
     parts_payload = conversation['conversation_parts'].to_h
     parts = Array(parts_payload['conversation_parts'])
-    batch = with_query_timeout_retry do
-      DataImports::Intercom::MessageBatchBuilder.new(
-        data_import: @data_import,
-        conversation: chatwoot_conversation,
-        source_conversation: conversation
-      ).perform
+    batch_builder = DataImports::Intercom::MessageBatchBuilder.new(
+      data_import: @data_import,
+      conversation: chatwoot_conversation,
+      source_conversation: conversation
+    )
+    batch = begin
+      with_query_timeout_retry { batch_builder.perform }
+    rescue ActiveRecord::QueryCanceled
+      nil
     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)
@@ -411,6 +415,26 @@ class DataImports::Intercom::Importer
     true
   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) }
+    record_truncated_conversation_parts(conversation, parts_count)
+
+    part_entries.each do |entry|
+      return false unless continue_import_with_heartbeat?
+
+      import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry)
+    end
+    true
+  end
+
+  def import_unprepared_message(conversation, contact, batch_builder, source_entry)
+    entry = with_query_timeout_retry { batch_builder.perform([source_entry]).entries.first }
+    import_message(conversation, contact, entry)
+  rescue StandardError => e
+    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

app/services/data_imports/intercom/message_batch_builder.rb

@@ -38,18 +38,21 @@ class DataImports::Intercom::MessageBatchBuilder
     @source_conversation = source_conversation
   end
 
-  def perform
-    source_entries = ordered_source_entries
+  def perform(source_entries = unprepared_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, position, mappings, messages)
+    Batch.new(items: source_entries.map do |source_entry|
+      build_entry(source_entry, source_entry[:position], mappings, messages)
     end)
   end
 
+  def unprepared_entries
+    ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) }
+  end
+
   private
 
   def ordered_source_entries