diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index 2e3345738..72b264131 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -173,6 +173,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 return unless import_conversation_messages(conversation, mapped_conversation, contact) update_conversation_activity(mapped_conversation) @@ -184,7 +185,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 return unless import_conversation_messages(conversation, chatwoot_conversation, contact) @@ -205,8 +210,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 @@ -408,7 +413,12 @@ class DataImports::Intercom::Importer conversation: chatwoot_conversation, source_conversation: conversation ) - batch = with_query_timeout_retry { batch_builder.perform } + 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? record_truncated_conversation_parts(conversation, parts.size) batch.entries.each_slice(MESSAGES_PER_BATCH) do |entries| @@ -417,6 +427,8 @@ class DataImports::Intercom::Importer import_message_batch(chatwoot_conversation, contact, batch_builder, entries) return false if @import_stopped end + return false if import_stopped? + true end @@ -535,6 +547,26 @@ class DataImports::Intercom::Importer increment_stat('messages', 'skipped') unless already_recorded 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 @@ -804,19 +836,21 @@ class DataImports::Intercom::Importer end def skip_already_imported_item(item, mapping, already_handled:) - item.update!( - status: :skipped, - chatwoot_record_type: mapping.chatwoot_record_type, - chatwoot_record_id: mapping.chatwoot_record_id, - last_error_code: ALREADY_IMPORTED_ERROR_CODE, - last_error_message: 'Already imported in a previous import.' - ) - record_already_imported_log( - data_import_item: item, - source_object_type: item.source_object_type, - source_object_id: item.source_object_id, - mapping: mapping - ) + DataImportItem.transaction do + item.update!( + status: :skipped, + chatwoot_record_type: mapping.chatwoot_record_type, + chatwoot_record_id: mapping.chatwoot_record_id, + last_error_code: ALREADY_IMPORTED_ERROR_CODE, + last_error_message: 'Already imported in a previous import.' + ) + record_already_imported_log( + data_import_item: item, + source_object_type: item.source_object_type, + source_object_id: item.source_object_id, + mapping: mapping + ) + end increment_stat(stat_group_for(item.source_object_type), 'skipped') unless already_handled end @@ -1088,17 +1122,19 @@ class DataImports::Intercom::Importer def reconcile_dirty_stats return if @dirty_stat_groups.empty? - @dirty_stat_groups.each_key do |group| - case group - when 'contacts' - reconcile_item_stats('contact') - when 'messages' - reconcile_message_stats - else - raise ArgumentError, "Unsupported Intercom import stat group: #{group}" + with_query_timeout_retry do + @dirty_stat_groups.each_key do |group| + case group + when 'contacts' + reconcile_item_stats('contact') + when 'messages' + reconcile_message_stats + else + raise ArgumentError, "Unsupported Intercom import stat group: #{group}" + end end + persist_stats end - persist_stats @dirty_stat_groups.clear end diff --git a/app/services/data_imports/intercom/message_batch_builder.rb b/app/services/data_imports/intercom/message_batch_builder.rb index 07a7002aa..b4cb728cc 100644 --- a/app/services/data_imports/intercom/message_batch_builder.rb +++ b/app/services/data_imports/intercom/message_batch_builder.rb @@ -38,8 +38,8 @@ class DataImports::Intercom::MessageBatchBuilder @source_conversation = source_conversation end - def perform - classify(ordered_source_entries) + def perform(source_entries = unprepared_entries) + classify(source_entries) end def refresh(entries) @@ -48,6 +48,10 @@ class DataImports::Intercom::MessageBatchBuilder end) end + def unprepared_entries + ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) } + end + private def classify(source_entries) @@ -112,7 +116,7 @@ class DataImports::Intercom::MessageBatchBuilder def build_entry(source_entry, position, mappings, messages) source_id = source_entry[:source_id] mapping = mappings[source_id] - mapped_message = messages[:by_id][mapping&.chatwoot_record_id] + mapped_message = messages[:by_id][mapping.chatwoot_record_id] if mapping&.chatwoot_record_type == 'Message' existing_message = messages[:by_source_id]["intercom:#{source_id}"] Entry.new( diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index 746145e84..3a6a61f6f 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -292,6 +292,28 @@ RSpec.describe DataImports::Intercom::Importer do expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) end + it 'retries a query timeout during deferred message stats reconciliation', :aggregate_failures do + described_class.new(data_import: data_import).import_conversations_page + stats = data_import.reload.stats.deep_dup + stats['messages']['imported'] = 0 + data_import.update!(stats: stats) + importer = described_class.new(data_import: data_import) + reconciliation_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:reconcile_message_stats).and_wrap_original do |method| + reconciliation_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if reconciliation_attempts == 1 + + method.call + end + + importer.import_conversations_page + + expect(reconciliation_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + it 'indexes imported messages for advanced search' do allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true) allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false) @@ -434,19 +456,62 @@ RSpec.describe DataImports::Intercom::Importer do allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args| result = method.call(*args) batch_write_count += 1 - data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) if batch_write_count == 1 + if batch_write_count == 1 + DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) + end result end importer.import_conversations_page - expect(importer).to have_received(:bulk_write_message_entries).twice + expect(importer).to have_received(:bulk_write_message_entries).once expect(data_import).not_to have_received(:touch) expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0) expect(account.messages.count).to eq(100) end end + it 'does not persist stale stats when a newer run takes over during the final batch', :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_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 + + 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(:bulk_write_message_entries).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 => next_run_id }) + end + 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:| @@ -667,6 +732,55 @@ RSpec.describe DataImports::Intercom::Importer do ).to eq([data_import.id]) 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 'keeps skipped contact stats when a query timeout is retried after the item update', :aggregate_failures do + described_class.new(data_import: data_import).perform + importer = described_class.new(data_import: next_data_import) + contact_log_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:record_already_imported_log).and_wrap_original do |method, **attributes| + if attributes[:source_object_type] == 'contact' + contact_log_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if contact_log_attempts == 1 + end + + method.call(**attributes) + end + + importer.import_contacts_page + + contact_item = next_data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(contact_log_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(contact_item).to be_skipped + expect(next_data_import.reload.stats.dig('contacts', 'skipped')).to eq(1) + expect(next_data_import.import_errors.skip_logs.exists?(data_import_item: contact_item)).to be(true) + 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')