fix(imports): reconcile conversation stats on retry

app/services/data_imports/intercom/importer.rb

@@ -158,6 +158,7 @@ class DataImports::Intercom::Importer
     mapped_conversation = mapping&.chatwoot_record
     if mapped_conversation && mapping.data_import_id != @data_import.id
       skip_already_imported_item(item, mapping, already_handled: already_handled)
+      reconcile_item_stats('conversation') if already_handled
       import_source_message(conversation, mapped_conversation, contact)
       return unless import_conversation_parts(conversation, mapped_conversation, contact)
 
@@ -170,7 +171,11 @@ class DataImports::Intercom::Importer
       record_mapping('conversation', source_id, chatwoot_conversation, metadata: conversation_metadata(conversation, inbox, source_type))
     end
     item.update!(status: :imported, chatwoot_record_type: 'Conversation', chatwoot_record_id: chatwoot_conversation.id)
-    increment_stat('conversations', 'imported') unless already_handled
+    if already_handled
+      reconcile_item_stats('conversation')
+    else
+      increment_stat('conversations', 'imported')
+    end
 
     import_source_message(conversation, chatwoot_conversation, contact)
     return unless import_conversation_parts(conversation, chatwoot_conversation, contact)

spec/services/data_imports/intercom/importer_spec.rb

@@ -337,6 +337,29 @@ RSpec.describe DataImports::Intercom::Importer do
       end
     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
+        end
+
+        importer.import_conversations_page
+        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
+
+        retry_importer = described_class.new(data_import: data_import, run_id: next_run_id)
+        retry_importer.import_conversations_page
+        retry_importer.finish!
+
+        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(1)
+        expect(data_import.processed_records).to eq(5)
+      end
+    end
+
     it 'rolls back a newly inserted conversation when mapping persistence fails', :aggregate_failures do
       importer = described_class.new(data_import: data_import)
       allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
@@ -478,6 +501,31 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
     end
 
+    it 'reconciles skipped conversation stats when a superseded run is retried' do
+      described_class.new(data_import: data_import).perform
+      run_id = 'intercom-run-1'
+      next_run_id = 'intercom-run-2'
+      next_data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
+
+      freeze_time do
+        importer = described_class.new(data_import: next_data_import, run_id: run_id)
+        allow(importer).to receive(:skip_existing_message_mapping).and_wrap_original do |method, *args|
+          method.call(*args)
+          next_data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
+          travel 1.minute
+        end
+
+        importer.import_conversations_page
+        expect(next_data_import.reload.stats.dig('conversations', 'skipped')).to eq(0)
+
+        retry_importer = described_class.new(data_import: next_data_import, run_id: next_run_id)
+        retry_importer.import_conversations_page
+        retry_importer.finish!
+
+        expect(next_data_import.reload.stats.dig('conversations', 'skipped')).to eq(1)
+      end
+    end
+
     it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do
       described_class.new(data_import: data_import).perform
       conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')