diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index eefcdfd67..047d5b746 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/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 diff --git a/app/services/data_imports/intercom/message_batch_builder.rb b/app/services/data_imports/intercom/message_batch_builder.rb index 8d9be9de2..5990e74a4 100644 --- a/app/services/data_imports/intercom/message_batch_builder.rb +++ b/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