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

perf(imports): bulk write Intercom messages

Browse Source
This commit is contained in:
Sony Mathew
2026-07-21 18:02:07 +05:30
parent 901b234b4d
commit 92d3536601
4 changed files with 321 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/data_imports/intercom/importer.rb
+137 -13
View File
@@ -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)
app/services/data_imports/intercom/message_batch_builder.rb
+13 -4
View File
@@ -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
spec/services/data_imports/intercom/importer_spec.rb
+144 -25
View File
@@ -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
spec/services/data_imports/intercom/message_batch_builder_spec.rb
+27
View File
@@ -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,