diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb
index 882bdcd93..8c5a29108 100644
--- a/app/services/data_imports/intercom/importer.rb
+++ b/app/services/data_imports/intercom/importer.rb
@@ -18,7 +18,6 @@ class DataImports::Intercom::Importer
TRUNCATED_PARTS_ERROR_CODE = 'DataImports::Intercom::TruncatedConversationParts'.freeze
E164_REGEX = /\A\+[1-9]\d{1,14}\z/
INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/
- REGULAR_MESSAGE_PART_TYPES = %w[comment note source].freeze
def initialize(data_import:, run_id: nil)
@data_import = data_import
@@ -161,8 +160,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)
- import_source_message(conversation, mapped_conversation, contact)
- return unless import_conversation_parts(conversation, mapped_conversation, contact)
+ return unless import_conversation_messages(conversation, mapped_conversation, contact)
update_conversation_activity(mapped_conversation)
return
@@ -175,8 +173,7 @@ class DataImports::Intercom::Importer
item.update!(status: :imported, chatwoot_record_type: 'Conversation', chatwoot_record_id: chatwoot_conversation.id)
increment_stat('conversations', 'imported') unless already_handled
- import_source_message(conversation, chatwoot_conversation, contact)
- return unless import_conversation_parts(conversation, chatwoot_conversation, contact)
+ return unless import_conversation_messages(conversation, chatwoot_conversation, contact)
update_conversation_activity(chatwoot_conversation)
rescue StandardError => e
@@ -390,59 +387,57 @@ class DataImports::Intercom::Importer
end
end
- def import_source_message(conversation, chatwoot_conversation, contact)
- source = conversation['source'].to_h
- return unless source_message_importable?(source)
-
- message_source_id = "conversation:#{source_id_for(conversation)}:source:#{source['id'].presence || 'initial'}"
- source_part = source.merge('part_type' => 'source', 'created_at' => conversation['created_at'])
- import_message(chatwoot_conversation, contact, source_part, message_source_id)
- end
-
- def import_conversation_parts(conversation, chatwoot_conversation, contact)
+ 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) }
record_truncated_conversation_parts(conversation, parts.size)
- parts.each do |part|
+ batch.part_entries.each do |entry|
return false unless continue_import_with_heartbeat?
- message_source_id = "conversation:#{source_id_for(conversation)}:part:#{part['id']}"
- import_message(chatwoot_conversation, contact, part, message_source_id)
+ import_message(chatwoot_conversation, contact, entry)
end
true
end
- def import_message(conversation, contact, part, message_source_id)
+ def import_message(conversation, contact, entry)
with_query_timeout_retry do
- mapping = find_mapping('message', message_source_id)
- if mapping && message_mapping_handled?(mapping, part)
- if mapping.data_import_id == @data_import.id
- reconcile_current_run_message_mapping(conversation, mapping, part)
- else
- skip_existing_message_mapping(conversation, mapping, part)
- end
+ case entry.classification
+ when :current_import
+ reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
+ when :previous_import
+ skip_existing_message_mapping(conversation, entry.mapping, entry.part)
+ when :repairable_stale_mapping, :existing_message, :new_message
+ create_message(conversation, contact, entry)
else
- create_message(conversation, contact, part, message_source_id)
+ raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
end
end
rescue StandardError => e
- fail_message(conversation, message_source_id, part, e)
+ fail_message(conversation, entry.source_id, entry.part, e)
end
- def create_message(conversation, contact, part, message_source_id)
- content = content_for(part)
- return record_skipped_message(conversation, message_source_id, part) if content.blank?
+ def create_message(conversation, contact, entry)
+ content = content_for(entry.part)
+ return record_skipped_message(conversation, entry) if content.blank?
- attrs = message_attributes(conversation, contact, part, message_source_id, content)
- message = nil
+ attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
+ message = entry.message
Message.transaction do
- message = conversation.messages.find_by(source_id: attrs[:source_id])
unless message
result = Message.insert_all!([attrs], returning: %w[id])
message = Message.find(result.rows.first.first)
end
- record_mapping('message', message_source_id, message, metadata: message_metadata(part))
+ record_message_mapping(entry, message)
end
increment_stat('messages', 'imported')
reindex_message_for_search(message)
@@ -457,13 +452,12 @@ class DataImports::Intercom::Importer
Rails.logger.warn("Intercom import message reindex failed for message #{message.id}: #{e.class} - #{e.message}")
end
- def record_skipped_message(conversation, message_source_id, part)
- mapping = find_mapping('message', message_source_id)
- if mapping
- already_recorded = skip_log_recorded?('message', message_source_id, SKIPPED_MESSAGE_ERROR_CODE)
- record_skipped_message_log(conversation, message_source_id, part)
+ def record_skipped_message(conversation, entry)
+ if entry.mapping
+ 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
- return mapping.chatwoot_record
+ return entry.message
end
DataImportMapping.create!(
@@ -471,15 +465,30 @@ class DataImports::Intercom::Importer
data_import: @data_import,
source_provider: PROVIDER,
source_object_type: 'message',
- source_object_id: message_source_id,
+ source_object_id: entry.source_id,
chatwoot_record_type: 'Conversation',
chatwoot_record_id: conversation.id,
- metadata: message_metadata(part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
+ metadata: message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
)
- record_skipped_message_log(conversation, message_source_id, part)
+ record_skipped_message_log(conversation, entry.source_id, entry.part)
increment_stat('messages', 'skipped')
end
+ def record_message_mapping(entry, message)
+ (entry.mapping || DataImportMapping.new(
+ account: @account,
+ source_provider: PROVIDER,
+ source_object_type: 'message',
+ source_object_id: entry.source_id
+ )).tap do |mapping|
+ mapping.data_import = @data_import
+ mapping.chatwoot_record_type = 'Message'
+ mapping.chatwoot_record_id = message.id
+ mapping.metadata = message_metadata(entry.part)
+ mapping.save!
+ end
+ end
+
def message_attributes(conversation, contact, part, message_source_id, content)
message_type = message_type_for(part)
created_at = timestamp_for(part['created_at'])
@@ -520,8 +529,7 @@ class DataImports::Intercom::Importer
end
def activity_part?(part)
- part_type = part['part_type'].to_s
- part_type.present? && REGULAR_MESSAGE_PART_TYPES.exclude?(part_type)
+ DataImports::Intercom::MessageBatchBuilder.activity_part?(part)
end
def message_content(part)
@@ -697,12 +705,6 @@ class DataImports::Intercom::Importer
increment_stat('messages', 'skipped') unless already_recorded
end
- def message_mapping_handled?(mapping, part)
- return false if mapping.metadata['skipped'] && activity_part?(part)
-
- mapping.metadata['skipped'] || mapping.chatwoot_record.present?
- end
-
def fail_item(item, error)
increment_stat('errors', 'count')
item&.update!(status: :failed, last_error_code: error.class.name, last_error_message: error.message)
@@ -800,7 +802,7 @@ class DataImports::Intercom::Importer
end
def source_message_importable?(source)
- source['body'].present? || source['subject'].present? || source['attachments'].present?
+ DataImports::Intercom::MessageBatchBuilder.source_message_importable?(source)
end
def skipped_message_log_message(part)
diff --git a/app/services/data_imports/intercom/message_batch_builder.rb b/app/services/data_imports/intercom/message_batch_builder.rb
new file mode 100644
index 000000000..8d9be9de2
--- /dev/null
+++ b/app/services/data_imports/intercom/message_batch_builder.rb
@@ -0,0 +1,131 @@
+class DataImports::Intercom::MessageBatchBuilder
+ PROVIDER = 'intercom'.freeze
+ REGULAR_PART_TYPES = %w[comment note source].freeze
+
+ Entry = Struct.new(:source_id, :part, :position, :mapping, :message, :classification, keyword_init: true) do
+ def source?
+ part['part_type'] == 'source'
+ end
+ end
+
+ Batch = Struct.new(:items, keyword_init: true) do
+ def entries
+ items
+ end
+
+ def source_entries
+ items.select(&:source?)
+ end
+
+ def part_entries
+ items.reject(&:source?)
+ end
+ end
+
+ def self.activity_part?(part)
+ part_type = part['part_type'].to_s
+ part_type.present? && REGULAR_PART_TYPES.exclude?(part_type)
+ end
+
+ def self.source_message_importable?(source)
+ source['body'].present? || source['subject'].present? || source['attachments'].present?
+ end
+
+ def initialize(data_import:, conversation:, source_conversation:)
+ @data_import = data_import
+ @account = data_import.account
+ @conversation = conversation
+ @source_conversation = source_conversation
+ end
+
+ def perform
+ source_entries = ordered_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)
+ end)
+ end
+
+ private
+
+ def ordered_source_entries
+ entries = []
+ source = @source_conversation['source'].to_h
+ if self.class.source_message_importable?(source)
+ entries << {
+ source_id: "conversation:#{source_conversation_id}:source:#{source['id'].presence || 'initial'}",
+ part: source.merge('part_type' => 'source', 'created_at' => @source_conversation['created_at'])
+ }
+ end
+
+ conversation_parts.each do |part|
+ entries << { source_id: "conversation:#{source_conversation_id}:part:#{part['id']}", part: part }
+ end
+ entries
+ end
+
+ def conversation_parts
+ Array(@source_conversation.dig('conversation_parts', 'conversation_parts'))
+ end
+
+ def source_conversation_id
+ @source_conversation['id'].presence || @source_conversation['external_id'].presence || @source_conversation['email'].presence
+ end
+
+ def message_mappings(source_entries)
+ DataImportMapping.where(
+ account: @account,
+ source_provider: PROVIDER,
+ source_object_type: 'message',
+ source_object_id: source_entries.pluck(:source_id)
+ ).index_by(&:source_object_id)
+ end
+
+ def messages_for(source_entries, mappings)
+ mapped_message_ids = mappings.values.filter_map do |mapping|
+ mapping.chatwoot_record_id if mapping.chatwoot_record_type == 'Message'
+ end
+ chatwoot_source_ids = source_entries.map { |entry| "intercom:#{entry[:source_id]}" }
+ messages = Message.where(id: mapped_message_ids).or(
+ Message.where(conversation_id: @conversation.id, source_id: chatwoot_source_ids)
+ ).to_a
+
+ {
+ by_id: messages.index_by(&:id),
+ by_source_id: messages.index_by(&:source_id)
+ }
+ end
+
+ 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]
+ existing_message = messages[:by_source_id]["intercom:#{source_id}"]
+
+ Entry.new(
+ source_id: source_id,
+ part: source_entry[:part],
+ position: position,
+ mapping: mapping,
+ message: mapped_message || existing_message,
+ classification: classification_for(mapping, mapped_message, existing_message, source_entry[:part])
+ )
+ end
+
+ def classification_for(mapping, mapped_message, existing_message, part)
+ return existing_message.present? ? :existing_message : :new_message if mapping.blank?
+ return :repairable_stale_mapping unless mapping_handled?(mapping, mapped_message, part)
+
+ mapping.data_import_id == @data_import.id ? :current_import : :previous_import
+ end
+
+ def mapping_handled?(mapping, mapped_message, part)
+ return false if mapping.metadata['skipped'] && self.class.activity_part?(part)
+
+ mapping.metadata['skipped'] || mapped_message.present?
+ end
+end
diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb
index 433d542cd..4799d9df9 100644
--- a/spec/services/data_imports/intercom/importer_spec.rb
+++ b/spec/services/data_imports/intercom/importer_spec.rb
@@ -377,11 +377,7 @@ 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_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
- raise StandardError, 'mapping failed' if object_type == 'message'
-
- method.call(object_type, source_id, record, metadata: metadata)
- end
+ allow(importer).to receive(:record_message_mapping).and_raise(StandardError, 'mapping failed')
importer.import_conversations_page
@@ -422,13 +418,13 @@ RSpec.describe DataImports::Intercom::Importer do
mapping_attempts = 0
target_source_id = 'conversation:conversation_1:part:part_1'
allow(importer).to receive(:sleep)
- allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
- if object_type == 'message' && source_id == target_source_id
+ allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
+ if entry.source_id == target_source_id
mapping_attempts += 1
raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
end
- method.call(object_type, source_id, record, metadata: metadata)
+ method.call(entry, message)
end
importer.import_conversations_page
@@ -447,13 +443,13 @@ RSpec.describe DataImports::Intercom::Importer do
mapping_attempts = 0
target_source_id = 'conversation:conversation_1:part:part_1'
allow(importer).to receive(:sleep)
- allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
- if object_type == 'message' && source_id == target_source_id
+ allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
+ if entry.source_id == target_source_id
mapping_attempts += 1
raise ActiveRecord::QueryCanceled, 'statement timeout'
end
- method.call(object_type, source_id, record, metadata: metadata)
+ method.call(entry, message)
end
importer.import_conversations_page
@@ -559,6 +555,33 @@ RSpec.describe DataImports::Intercom::Importer do
expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3)
end
+ it 'repairs a missing mapping without recreating the existing message', :aggregate_failures do
+ described_class.new(data_import: data_import).import_conversations_page
+ conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+ message = conversation.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:part_1')
+ DataImportMapping.find_by!(
+ account: account,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1'
+ ).destroy!
+ importer = described_class.new(data_import: data_import)
+ allow(importer).to receive(:find_mapping).and_call_original
+
+ importer.import_conversations_page
+
+ repaired_mapping = DataImportMapping.find_by!(
+ account: account,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1'
+ )
+ expect(conversation.messages.where(source_id: message.source_id).count).to eq(1)
+ expect(repaired_mapping.chatwoot_record).to eq(message)
+ expect(importer).not_to have_received(:find_mapping).with('message', anything)
+ expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+ end
+
it 'updates conversation activity when a later import adds new messages to the mapped conversation', :aggregate_failures do
new_part = {
'id' => 'part_3',
diff --git a/spec/services/data_imports/intercom/message_batch_builder_spec.rb b/spec/services/data_imports/intercom/message_batch_builder_spec.rb
new file mode 100644
index 000000000..dc61cbabb
--- /dev/null
+++ b/spec/services/data_imports/intercom/message_batch_builder_spec.rb
@@ -0,0 +1,178 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::MessageBatchBuilder do
+ let(:account) { create(:account) }
+ let(:data_import) { create(:data_import, :intercom, account: account) }
+ let(:conversation) { create(:conversation, account: account) }
+ let(:source_conversation) do
+ {
+ 'id' => 'conversation_1',
+ 'created_at' => 1_700_000_000,
+ 'source' => {
+ 'id' => 'source_1',
+ 'part_type' => 'conversation',
+ 'body' => 'Initial message
',
+ 'created_at' => 1_700_000_000
+ },
+ 'conversation_parts' => {
+ 'conversation_parts' => [
+ {
+ 'id' => 'part_1',
+ 'part_type' => 'comment',
+ 'body' => 'First reply
',
+ 'created_at' => 1_700_000_000
+ },
+ {
+ 'id' => 'part_2',
+ 'part_type' => 'note',
+ 'body' => 'Internal note
',
+ 'created_at' => 1_700_000_100
+ }
+ ]
+ }
+ }
+ end
+ let(:builder) do
+ described_class.new(
+ data_import: data_import,
+ conversation: conversation,
+ source_conversation: source_conversation
+ )
+ end
+
+ it 'preserves source order and prefetches mappings and messages once per batch', :aggregate_failures do
+ sql_queries = []
+ subscriber = lambda do |_name, _start, _finish, _id, payload|
+ sql_queries << payload unless payload[:name] == 'SCHEMA'
+ end
+ batch_builder = builder
+
+ batch = ActiveSupport::Notifications.subscribed(subscriber, 'sql.active_record') { batch_builder.perform }
+
+ expect(batch.entries.map(&:source_id)).to eq(
+ %w[
+ conversation:conversation_1:source:source_1
+ conversation:conversation_1:part:part_1
+ conversation:conversation_1:part:part_2
+ ]
+ )
+ expect(batch.entries.map(&:position)).to eq([0, 1, 2])
+ expect(batch.entries.map(&:classification)).to all(eq(:new_message))
+ expect(sql_queries.count { |query| query[:name] == 'DataImportMapping Load' }).to eq(1)
+ message_queries = sql_queries.select { |query| query[:name] == 'Message Load' }
+ expect(message_queries.size).to eq(1), message_queries.pluck(:sql).join("\n")
+ end
+
+ it 'returns an empty batch without prefetch queries when the conversation has no messages' do
+ source_conversation['source'] = nil
+ source_conversation['conversation_parts']['conversation_parts'] = []
+
+ expect(DataImportMapping).not_to receive(:where)
+ expect(Message).not_to receive(:where)
+
+ expect(builder.perform.entries).to be_empty
+ end
+
+ it 'classifies a live mapping from the current import as already handled' do
+ message = create(
+ :message,
+ account: account,
+ conversation: conversation,
+ inbox: conversation.inbox,
+ source_id: 'intercom:conversation:conversation_1:part:part_1'
+ )
+ DataImportMapping.create!(
+ account: account,
+ data_import: data_import,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1',
+ chatwoot_record_type: 'Message',
+ chatwoot_record_id: message.id,
+ metadata: {}
+ )
+
+ entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+ expect(entry).to have_attributes(classification: :current_import, message: message)
+ end
+
+ it 'classifies a live mapping from a previous import without changing its owner' do
+ previous_import = create(:data_import, :intercom, account: account)
+ message = create(
+ :message,
+ account: account,
+ conversation: conversation,
+ inbox: conversation.inbox,
+ source_id: 'intercom:conversation:conversation_1:part:part_1'
+ )
+ mapping = DataImportMapping.create!(
+ account: account,
+ data_import: previous_import,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1',
+ chatwoot_record_type: 'Message',
+ chatwoot_record_id: message.id,
+ metadata: {}
+ )
+
+ entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+ expect(entry).to have_attributes(classification: :previous_import, mapping: mapping, message: message)
+ expect(mapping.reload.data_import).to eq(previous_import)
+ end
+
+ it 'classifies a mapping whose message was deleted as repairable' do
+ mapping = DataImportMapping.create!(
+ account: account,
+ data_import: data_import,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1',
+ chatwoot_record_type: 'Message',
+ chatwoot_record_id: 0,
+ metadata: {}
+ )
+
+ entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+ expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
+ end
+
+ it 'classifies an existing conversation message without a mapping for repair' do
+ message = create(
+ :message,
+ account: account,
+ conversation: conversation,
+ inbox: conversation.inbox,
+ source_id: 'intercom:conversation:conversation_1:part:part_1'
+ )
+
+ entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+ expect(entry).to have_attributes(classification: :existing_message, mapping: nil, message: message)
+ end
+
+ it 'repairs a skipped mapping when the source part is now an activity' do
+ source_conversation.dig('conversation_parts', 'conversation_parts').first.merge!(
+ 'part_type' => 'assignment',
+ 'body' => nil,
+ 'assigned_to' => { 'name' => 'Support' }
+ )
+ mapping = DataImportMapping.create!(
+ account: account,
+ data_import: data_import,
+ source_provider: 'intercom',
+ source_object_type: 'message',
+ source_object_id: 'conversation:conversation_1:part:part_1',
+ chatwoot_record_type: 'Conversation',
+ chatwoot_record_id: conversation.id,
+ metadata: { skipped: true }
+ )
+
+ entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+ expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
+ end
+end