akadmin/chatwoot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(imports): prepare Intercom message batches

Browse Source
This commit is contained in:
Sony Mathew
2026-07-21 17:32:07 +05:30
parent 1c16c427f4
commit 901b234b4d
4 changed files with 398 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/data_imports/intercom/importer.rb
+55 -53
View File
@@ -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)
app/services/data_imports/intercom/message_batch_builder.rb
+131
View File
@@ -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
spec/services/data_imports/intercom/importer_spec.rb
+34 -11
View File
@@ -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',
spec/services/data_imports/intercom/message_batch_builder_spec.rb
+178
View File
@@ -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' => '<p>Initial message</p>',
'created_at' => 1_700_000_000
},
'conversation_parts' => {
'conversation_parts' => [
{
'id' => 'part_1',
'part_type' => 'comment',
'body' => '<p>First reply</p>',
'created_at' => 1_700_000_000
},
{
'id' => 'part_2',
'part_type' => 'note',
'body' => '<p>Internal note</p>',
'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