diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index 8c5a29108..ad45f16e2 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -9,6 +9,7 @@ class DataImports::Intercom::Importer DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze CONTACTS_PER_PAGE = 50 CONVERSATIONS_PER_PAGE = 10 + MESSAGES_PER_BATCH = 100 HEARTBEAT_INTERVAL = 1.minute QUERY_TIMEOUT_RETRY_LIMIT = 1 QUERY_TIMEOUT_RETRY_DELAY_RANGE = (0.2..0.5) @@ -16,9 +17,19 @@ class DataImports::Intercom::Importer ALREADY_IMPORTED_ERROR_CODE = 'DataImports::Intercom::AlreadyImported'.freeze SKIPPED_MESSAGE_ERROR_CODE = 'DataImports::Intercom::SkippedMessage'.freeze TRUNCATED_PARTS_ERROR_CODE = 'DataImports::Intercom::TruncatedConversationParts'.freeze + MESSAGE_MAPPING_UNIQUE_INDEX = :idx_data_import_mappings_on_account_and_source E164_REGEX = /\A\+[1-9]\d{1,14}\z/ INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/ + MessageBatchResult = Struct.new( + :imported_entries, + :skipped_entries, + :current_entries, + :previous_entries, + :messages, + keyword_init: true + ) + def initialize(data_import:, run_id: nil) @data_import = data_import @run_id = run_id @@ -390,25 +401,138 @@ 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 - end - - batch.source_entries.each { |entry| import_message(chatwoot_conversation, contact, entry) } + batch_builder = DataImports::Intercom::MessageBatchBuilder.new( + data_import: @data_import, + conversation: chatwoot_conversation, + source_conversation: conversation + ) + batch = with_query_timeout_retry { batch_builder.perform } record_truncated_conversation_parts(conversation, parts.size) - batch.part_entries.each do |entry| + batch.entries.each_slice(MESSAGES_PER_BATCH) do |entries| return false unless continue_import_with_heartbeat? - import_message(chatwoot_conversation, contact, entry) + import_message_batch(chatwoot_conversation, contact, batch_builder, entries) + return false if @import_stopped end true end + def import_message_batch(conversation, contact, batch_builder, entries) + result = bulk_message_batch_result(conversation, contact, batch_builder, entries) + return if result.blank? + + result.current_entries.each do |entry| + reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part) + end + result.previous_entries.each do |entry| + skip_existing_message_mapping(conversation, entry.mapping, entry.part) + end + result.skipped_entries.each { |entry| record_bulk_skipped_message(conversation, entry) } + increment_stat('messages', 'imported', result.imported_entries.size) + result.messages.each { |message| reindex_message_for_search(message) } + end + + def bulk_message_batch_result(conversation, contact, batch_builder, entries) + bulk_write_message_entries(conversation, contact, batch_builder, entries) + rescue StandardError + entries.each { |entry| import_message(conversation, contact, entry) } unless @import_stopped + nil + end + + def bulk_write_message_entries(conversation, contact, batch_builder, entries) + @data_import.with_lock do + if inactive_import_run? + @import_stopped = true + next + end + + refreshed_entries = batch_builder.refresh(entries).entries + persist_message_entries(conversation, contact, refreshed_entries) + end + end + + def inactive_import_run? + @data_import.abandoned? || @data_import.failed? || @data_import.completed? || + @data_import.completed_with_errors? || stale_import_run? + end + + def persist_message_entries(conversation, contact, entries) + grouped_entries = entries.group_by(&:classification) + writable_entries = entries.select do |entry| + %i[repairable_stale_mapping existing_message new_message].include?(entry.classification) + end + content_by_source_id = writable_entries.to_h { |entry| [entry.source_id, content_for(entry.part)] } + skipped_entries, imported_entries = writable_entries.partition { |entry| content_by_source_id[entry.source_id].blank? } + messages = insert_messages(conversation, contact, imported_entries, content_by_source_id) + upsert_message_mappings(conversation, imported_entries, skipped_entries, messages) + + MessageBatchResult.new( + imported_entries: imported_entries, + skipped_entries: skipped_entries, + current_entries: grouped_entries.fetch(:current_import, []), + previous_entries: grouped_entries.fetch(:previous_import, []), + messages: messages + ) + end + + def insert_messages(conversation, contact, entries, content_by_source_id) + new_entries = entries.reject(&:message) + if new_entries.present? + attributes = new_entries.map do |entry| + message_attributes(conversation, contact, entry.part, entry.source_id, content_by_source_id.fetch(entry.source_id)) + end + result = Message.insert_all!(attributes, returning: %w[id source_id]) + inserted_messages = Message.where(id: result.pluck('id')).index_by(&:source_id) + end + + entries.map do |entry| + entry.message || inserted_messages.fetch("intercom:#{entry.source_id}") + end + end + + def upsert_message_mappings(conversation, imported_entries, skipped_entries, messages) + now = Time.current + mapping_attributes = imported_entries.zip(messages).map do |entry, message| + message_mapping_attributes(entry, 'Message', message.id, message_metadata(entry.part), now) + end + mapping_attributes.concat(skipped_entries.filter_map do |entry| + next if entry.mapping + + metadata = message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part') + message_mapping_attributes(entry, 'Conversation', conversation.id, metadata, now) + end) + return if mapping_attributes.empty? + + DataImportMapping.upsert_all( + mapping_attributes, + unique_by: MESSAGE_MAPPING_UNIQUE_INDEX, + update_only: %i[data_import_id chatwoot_record_type chatwoot_record_id metadata updated_at], + record_timestamps: false + ) + end + + def message_mapping_attributes(entry, record_type, record_id, metadata, now) + { + account_id: @account.id, + data_import_id: @data_import.id, + source_provider: PROVIDER, + source_object_type: 'message', + source_object_id: entry.source_id, + chatwoot_record_type: record_type, + chatwoot_record_id: record_id, + metadata: metadata, + created_at: entry.mapping&.created_at || now, + updated_at: now + } + end + + def record_bulk_skipped_message(conversation, entry) + already_recorded = skip_log_recorded?('message', entry.source_id, SKIPPED_MESSAGE_ERROR_CODE) + record_skipped_message_log(conversation, entry.source_id, entry.part) + increment_stat('messages', 'skipped') unless already_recorded + end + def import_message(conversation, contact, entry) with_query_timeout_retry do case entry.classification @@ -950,9 +1074,9 @@ class DataImports::Intercom::Importer @import_types ||= (@data_import.import_types.presence || DEFAULT_IMPORT_TYPES) end - def increment_stat(group, key) + def increment_stat(group, key, amount = 1) @stats[group] ||= {} - @stats[group][key] = @stats[group][key].to_i + 1 + @stats[group][key] = @stats[group][key].to_i + amount end def mark_stat_group_dirty(group) diff --git a/app/services/data_imports/intercom/message_batch_builder.rb b/app/services/data_imports/intercom/message_batch_builder.rb index 8d9be9de2..07a7002aa 100644 --- a/app/services/data_imports/intercom/message_batch_builder.rb +++ b/app/services/data_imports/intercom/message_batch_builder.rb @@ -39,19 +39,28 @@ class DataImports::Intercom::MessageBatchBuilder end def perform - source_entries = ordered_source_entries + classify(ordered_source_entries) + end + + def refresh(entries) + classify(entries.map do |entry| + { source_id: entry.source_id, part: entry.part, position: entry.position } + end) + end + + private + + def classify(source_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) + build_entry(source_entry, source_entry.fetch(:position, position), mappings, messages) end) end - private - def ordered_source_entries entries = [] source = @source_conversation['source'].to_h diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index 4799d9df9..b98339eea 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -116,6 +116,86 @@ RSpec.describe DataImports::Intercom::Importer do expect(DataImportMapping.where(data_import: data_import).count).to eq(5) end + it 'writes a normal conversation with one message insert and one mapping upsert', :aggregate_failures do + importer = described_class.new(data_import: data_import) + message_batch_sizes = [] + mapping_batch_sizes = [] + mapping_upsert_options = [] + allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs| + message_batch_sizes << records.size + method.call(records, **kwargs) + end + allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs| + mapping_batch_sizes << records.size + mapping_upsert_options << kwargs + method.call(records, **kwargs) + end + expect(importer).not_to receive(:create_message) + + importer.perform + + expect(message_batch_sizes).to eq([3]) + expect(mapping_batch_sizes).to eq([3]) + expect(mapping_upsert_options).to contain_exactly( + include(unique_by: described_class::MESSAGE_MAPPING_UNIQUE_INDEX, record_timestamps: false) + ) + expect(account.messages.count).to eq(3) + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + + it 'preserves provider order when imported messages share a timestamp' do + equal_timestamp_conversation = conversation_payload.deep_dup + equal_timestamp_conversation['conversation_parts']['conversation_parts'].each do |part| + part['created_at'] = conversation_payload['created_at'] + part['updated_at'] = conversation_payload['created_at'] + end + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(equal_timestamp_conversation) + + described_class.new(data_import: data_import).perform + + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + expect(conversation.messages.order(:created_at, :id).pluck(:source_id)).to eq( + %w[ + intercom:conversation:conversation_1:source:source_1 + intercom:conversation:conversation_1:part:part_1 + intercom:conversation:conversation_1:part:part_2 + ] + ) + end + + it 'writes conversations in batches of at most 100 messages', :aggregate_failures do + bulk_conversation = conversation_payload.deep_dup + template_part = bulk_conversation.dig('conversation_parts', 'conversation_parts').first + bulk_conversation['conversation_parts']['conversation_parts'] = Array.new(205) do |index| + template_part.merge( + 'id' => "part_#{index + 1}", + 'created_at' => 1_700_000_100 + index, + 'updated_at' => 1_700_000_100 + index + ) + end + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(bulk_conversation) + message_batch_sizes = [] + mapping_batch_sizes = [] + allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs| + message_batch_sizes << records.size + method.call(records, **kwargs) + end + allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs| + mapping_batch_sizes << records.size + method.call(records, **kwargs) + end + importer = described_class.new(data_import: data_import) + expect(importer).to receive(:update_conversation_activity).once.and_call_original + + importer.import_conversations_page + + expect(message_batch_sizes).to eq([100, 100, 6]) + expect(mapping_batch_sizes).to eq([100, 100, 6]) + expect(account.messages.order(:created_at).pick(:source_id)).to eq('intercom:conversation:conversation_1:source:source_1') + expect(account.messages.count).to eq(206) + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(206) + end + it 'imports historical records without dispatching record events or outbound side effects', :aggregate_failures do dispatched_events = [] allow(Rails.configuration.dispatcher).to receive(:dispatch) do |event_name, *_args| @@ -216,13 +296,19 @@ RSpec.describe DataImports::Intercom::Importer do allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true) allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false) reindexed_message_ids = [] + reindex_transaction_depths = [] + transaction_depth_before_import = Message.connection.open_transactions original_reindex_for_search = Message.instance_method(:reindex_for_search) - Message.define_method(:reindex_for_search) { reindexed_message_ids << id } + Message.define_method(:reindex_for_search) do + reindexed_message_ids << id + reindex_transaction_depths << self.class.connection.open_transactions + end Message.__send__(:private, :reindex_for_search) described_class.new(data_import: data_import).perform expect(reindexed_message_ids).to match_array(Message.where(account_id: account.id).pluck(:id)) + expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import)) ensure Message.define_method(:reindex_for_search, original_reindex_for_search) Message.__send__(:private, :reindex_for_search) @@ -298,12 +384,12 @@ RSpec.describe DataImports::Intercom::Importer do expect(data_import.reload.cursor.dig('conversations', 'starting_after')).to be_nil end - it 'heartbeats at most once per minute while processing conversation parts' do + it 'heartbeats at most once per minute while processing conversation message batches' do freeze_time do started_at = Time.current long_conversation = conversation_payload.deep_dup template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first - long_conversation['conversation_parts']['conversation_parts'] = Array.new(5) do |index| + long_conversation['conversation_parts']['conversation_parts'] = Array.new(400) do |index| template_part.merge('id' => "part_#{index + 1}") end allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation) @@ -313,30 +399,51 @@ RSpec.describe DataImports::Intercom::Importer do method.call end importer = described_class.new(data_import: data_import) - allow(importer).to receive(:create_message) { travel 30.seconds } - - importer.import_conversations_page - - expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes]) - end - end - - it 'stops parts without heartbeating or persisting stale stats when a newer run takes over', :aggregate_failures do - freeze_time do - run_id = 'intercom-run-1' - data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }) - 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 + empty_result = described_class::MessageBatchResult.new( + imported_entries: [], + skipped_entries: [], + current_entries: [], + previous_entries: [], + messages: [] + ) + allow(importer).to receive(:bulk_write_message_entries) do + travel 30.seconds + empty_result end importer.import_conversations_page - expect(importer).to have_received(:create_message).once + expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes]) + expect(importer).to have_received(:bulk_write_message_entries).exactly(5).times + end + end + + it 'stops batches without heartbeating or persisting stale stats when a newer run takes over', :aggregate_failures do + freeze_time do + run_id = 'intercom-run-1' + data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }) + long_conversation = conversation_payload.deep_dup + template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first + long_conversation['conversation_parts']['conversation_parts'] = Array.new(100) do |index| + template_part.merge('id' => "part_#{index + 1}") + end + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation) + allow(data_import).to receive(:touch).and_call_original + importer = described_class.new(data_import: data_import, run_id: run_id) + batch_write_count = 0 + 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 + result + end + + importer.import_conversations_page + + expect(importer).to have_received(:bulk_write_message_entries).twice 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 @@ -377,7 +484,12 @@ RSpec.describe DataImports::Intercom::Importer do it 'rolls back a newly inserted message when mapping persistence fails', :aggregate_failures do importer = described_class.new(data_import: data_import) - allow(importer).to receive(:record_message_mapping).and_raise(StandardError, 'mapping failed') + allow(DataImportMapping).to receive(:upsert_all).and_raise(StandardError, 'bulk mapping failed') + allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message| + raise StandardError, 'mapping failed' if entry.source_id == 'conversation:conversation_1:source:source_1' + + method.call(entry, message) + end importer.import_conversations_page @@ -417,6 +529,7 @@ RSpec.describe DataImports::Intercom::Importer do importer = described_class.new(data_import: data_import) mapping_attempts = 0 target_source_id = 'conversation:conversation_1:part:part_1' + allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable') allow(importer).to receive(:sleep) allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message| if entry.source_id == target_source_id @@ -442,6 +555,7 @@ RSpec.describe DataImports::Intercom::Importer do importer = described_class.new(data_import: data_import) mapping_attempts = 0 target_source_id = 'conversation:conversation_1:part:part_1' + allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable') allow(importer).to receive(:sleep) allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message| if entry.source_id == target_source_id @@ -528,6 +642,9 @@ RSpec.describe DataImports::Intercom::Importer do 'message' => 3 ) expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported']) + expect( + DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message').distinct.pluck(:data_import_id) + ).to eq([data_import.id]) end it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do @@ -553,6 +670,7 @@ RSpec.describe DataImports::Intercom::Importer do expect(next_data_import.import_errors.skip_logs.where(source_object_type: 'message')).to be_empty message_mappings = DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message') expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3) + expect(message_mappings.distinct.pluck(:data_import_id)).to eq([next_data_import.id]) end it 'repairs a missing mapping without recreating the existing message', :aggregate_failures do @@ -1099,8 +1217,8 @@ RSpec.describe DataImports::Intercom::Importer do before do allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs| - if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part' - insert_attempts << records.first[:source_id] + if records.any? { |record| record[:source_id] == 'intercom:conversation:conversation_1:part:bad_part' } + insert_attempts << 'intercom:conversation:conversation_1:part:bad_part' raise ActiveRecord::StatementInvalid, 'bad message' end @@ -1123,7 +1241,8 @@ RSpec.describe DataImports::Intercom::Importer do ) expect(data_import.reload).to be_completed_with_errors expect(data_import.stats.dig('errors', 'count')).to eq(1) - expect(insert_attempts.one?).to be(true) + expect(insert_attempts.size).to eq(2) + expect(account.messages.find_by(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_present end end end diff --git a/spec/services/data_imports/intercom/message_batch_builder_spec.rb b/spec/services/data_imports/intercom/message_batch_builder_spec.rb index dc61cbabb..9a2420320 100644 --- a/spec/services/data_imports/intercom/message_batch_builder_spec.rb +++ b/spec/services/data_imports/intercom/message_batch_builder_spec.rb @@ -73,6 +73,33 @@ RSpec.describe DataImports::Intercom::MessageBatchBuilder do expect(builder.perform.entries).to be_empty end + it 'refreshes classifications while preserving source positions' do + batch = builder.perform + target_entry = batch.entries.second + message = create( + :message, + account: account, + conversation: conversation, + inbox: conversation.inbox, + source_id: "intercom:#{target_entry.source_id}" + ) + DataImportMapping.create!( + account: account, + data_import: data_import, + source_provider: 'intercom', + source_object_type: 'message', + source_object_id: target_entry.source_id, + chatwoot_record_type: 'Message', + chatwoot_record_id: message.id, + metadata: {} + ) + + refreshed_batch = builder.refresh(batch.entries) + + expect(refreshed_batch.entries.map(&:position)).to eq([0, 1, 2]) + expect(refreshed_batch.entries.second).to have_attributes(classification: :current_import, message: message) + end + it 'classifies a live mapping from the current import as already handled' do message = create( :message,