diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index db769e331..8e22f58dd 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -163,6 +163,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) @@ -175,7 +176,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) @@ -197,8 +202,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 @@ -412,6 +417,8 @@ class DataImports::Intercom::Importer message_source_id = "conversation:#{source_id_for(conversation)}:part:#{part['id']}" import_message(chatwoot_conversation, contact, part, message_source_id) end + return false if import_stopped? + true end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index e886c75af..d578b5e60 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -328,8 +328,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 @@ -340,6 +339,49 @@ 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' + 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:| @@ -554,6 +596,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')