fix(imports): reload run state before heartbeat checks

app/services/data_imports/intercom/importer.rb

@@ -197,8 +197,8 @@ class DataImports::Intercom::Importer
   end
 
   def continue_import_with_heartbeat?
-    return true if @data_import.updated_at > HEARTBEAT_INTERVAL.ago
     return false if import_stopped?
+    return true if @data_import.updated_at > HEARTBEAT_INTERVAL.ago
 
     @data_import.touch if @data_import.updated_at <= HEARTBEAT_INTERVAL.ago
     true
@@ -435,6 +435,8 @@ class DataImports::Intercom::Importer
         fail_message(chatwoot_conversation, message_source_id, part, e)
       end
     end
+    return false if import_stopped?
+
     true
   end
 

spec/services/data_imports/intercom/importer_spec.rb

@@ -325,8 +325,7 @@ RSpec.describe DataImports::Intercom::Importer do
         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
-          data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-          travel 1.minute
+          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
         end
 
         importer.import_conversations_page
@@ -337,6 +336,26 @@ RSpec.describe DataImports::Intercom::Importer do
       end
     end
 
+    it 'does not persist stale stats when a newer run takes over during the final part', :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|
+        method.call(*args).tap do
+          part = args[2]
+          next unless part['id'] == 'part_2'
+
+          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        end
+      end
+
+      importer.import_conversations_page
+
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+      expect(conversation.messages.count).to eq(3)
+      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'