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

perf: shorten Intercom import jobs (#15051)

Browse Source 
## Description

Reduces sustained load during large Intercom imports by limiting
conversation list pages to 10 while retaining 50-contact pages. Cursor
and provider total-count behavior remain unchanged.

The one-minute progress heartbeat now lives in parent PR #15050 because
stalled-import detection must be safe when that PR is deployed
independently. This child PR therefore contains only the smaller-page
delta.

This is Phase 1, Task 2 of the [Intercom import optimization plan
(CW-7615)](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion).

This PR is stacked on #15050 and should be reviewed as the two-file
delta from `codex/cw-7519-intercom-stalled-retry-15m`.

## Closes

-
[CW-7615](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion)

## Type of change

- [x] Bug fix (non-breaking change which fixes an issue)

## How to test

1. Start an Intercom import containing contacts and conversations across
multiple source pages.
2. Confirm contact list requests retain a page size of 50.
3. Confirm conversation list requests use a page size of 10.
4. Confirm page cursors and displayed provider totals continue advancing
normally.
5. Confirm the parent PR heartbeat keeps long conversations fresh while
these shorter pages are processed.

## Checklist

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [x] I have added tests that prove the change is effective
- [x] New and existing focused tests pass locally with my changes
This commit is contained in:
Sony Mathew
2026-07-23 23:38:01 +05:30
committed by GitHub
parent ea783a89ea
commit b6efdae243
4 changed files with 777 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/data_imports/intercom/importer.rb
+194 -113
View File
@@ -7,14 +7,17 @@ class DataImports::Intercom::Importer
end
DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze
CONTACTS_PER_PAGE = 50
CONVERSATIONS_PER_PAGE = 10
HEARTBEAT_INTERVAL = 1.minute
QUERY_TIMEOUT_RETRY_LIMIT = 1
QUERY_TIMEOUT_RETRY_DELAY_RANGE = (0.2..0.5)
PROVIDER = 'intercom'.freeze
ALREADY_IMPORTED_ERROR_CODE = 'DataImports::Intercom::AlreadyImported'.freeze
SKIPPED_MESSAGE_ERROR_CODE = 'DataImports::Intercom::SkippedMessage'.freeze
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
@@ -23,6 +26,7 @@ class DataImports::Intercom::Importer
@client = DataImports::Intercom::Client.new(access_token: data_import.access_token)
@placeholder_inboxes = DataImports::Intercom::PlaceholderInboxBuilder.new(account: @account)
@stats = default_stats.deep_merge(data_import.stats || {})
@dirty_stat_groups = {}
end
def perform
@@ -45,8 +49,9 @@ class DataImports::Intercom::Importer
def finish!
return if @data_import.reload.abandoned?
has_failures = @data_import.import_errors.non_skip_logs.exists? || @data_import.import_errors.failed.exists?
status = has_failures ? :completed_with_errors : :completed
error_count = @data_import.import_errors.non_skip_logs.count + @data_import.import_errors.failed.count
@stats['errors']['count'] = error_count
status = error_count.positive? ? :completed_with_errors : :completed
@data_import.update!(
status: status,
completed_at: Time.current,
@@ -66,7 +71,7 @@ class DataImports::Intercom::Importer
end
def import_contacts_page(starting_after: cursor_for('contacts'))
response = @client.list_contacts(starting_after: starting_after)
response = @client.list_contacts(starting_after: starting_after, per_page: CONTACTS_PER_PAGE)
update_stat_total('contacts', response['total_count']) if response['total_count'].present?
Array(response['data'] || response['contacts']).each do |contact|
break if import_stopped?
@@ -75,13 +80,14 @@ class DataImports::Intercom::Importer
end
return PageResult.new(next_cursor: nil) if import_stopped?
reconcile_dirty_stats
next_cursor = response.dig('pages', 'next', 'starting_after')
update_cursor('contacts', next_cursor)
PageResult.new(next_cursor: next_cursor)
end
def import_conversations_page(starting_after: cursor_for('conversations'))
response = @client.list_conversations(starting_after: starting_after)
response = @client.list_conversations(starting_after: starting_after, per_page: CONVERSATIONS_PER_PAGE)
update_stat_total('conversations', response['total_count']) if response['total_count'].present?
Array(response['data'] || response['conversations']).each do |conversation_summary|
break if import_stopped?
@@ -90,6 +96,7 @@ class DataImports::Intercom::Importer
end
return PageResult.new(next_cursor: nil) if import_stopped?
reconcile_dirty_stats
next_cursor = response.dig('pages', 'next', 'starting_after')
update_cursor('conversations', next_cursor)
PageResult.new(next_cursor: next_cursor)
@@ -156,8 +163,8 @@ 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)
reconcile_item_stats('conversation') if already_handled
return unless import_conversation_messages(conversation, mapped_conversation, contact)
update_conversation_activity(mapped_conversation)
return
@@ -168,10 +175,13 @@ class DataImports::Intercom::Importer
record_mapping('conversation', source_id, chatwoot_conversation, metadata: conversation_metadata(conversation, inbox, source_type))
end
item.update!(status: :imported, chatwoot_record_type: 'Conversation', chatwoot_record_id: chatwoot_conversation.id)
increment_stat('conversations', 'imported') unless already_handled
if already_handled
reconcile_item_stats('conversation')
else
increment_stat('conversations', 'imported')
end
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
@@ -203,32 +213,35 @@ class DataImports::Intercom::Importer
end
def import_contact(contact_payload, required_for_conversation: false)
source_id = source_id_for(contact_payload)
if source_id.present? && (mapping = find_mapping('contact', source_id)) && (mapped_contact = mapping.chatwoot_record)
return reuse_mapped_contact(contact_payload, source_id, mapping, mapped_contact)
end
item = nil
with_query_timeout_retry do
source_id = source_id_for(contact_payload)
if source_id.present? && (mapping = find_mapping('contact', source_id)) && (mapped_contact = mapping.chatwoot_record)
return reuse_mapped_contact(contact_payload, source_id, mapping, mapped_contact)
end
contact_payload = retrieve_contact_payload(contact_payload)
source_id = source_id_for(contact_payload)
already_handled = item_handled?('contact', source_id)
item = import_item('contact', source_id, contact_payload)
mapping = find_mapping('contact', source_id)
contact_payload = retrieve_contact_payload(contact_payload)
source_id = source_id_for(contact_payload)
already_handled = item_handled?('contact', source_id)
item = import_item('contact', source_id, contact_payload)
mapping = find_mapping('contact', source_id)
mapped_contact = mapping&.chatwoot_record
if mapped_contact && mapping.data_import_id != @data_import.id
skip_already_imported_item(item, mapping, already_handled: already_handled)
return mapped_contact
end
mapped_contact = mapping&.chatwoot_record
if mapped_contact && mapping.data_import_id != @data_import.id
skip_already_imported_item(item, mapping, already_handled: already_handled)
return mapped_contact
end
contact = Contact.transaction do
imported_contact = mapped_contact || find_existing_contact(contact_payload) || create_contact(contact_payload)
update_existing_contact(imported_contact, contact_payload)
record_mapping('contact', source_id, imported_contact, metadata: contact_metadata(contact_payload))
item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: imported_contact.id)
imported_contact
contact = Contact.transaction do
imported_contact = mapped_contact || find_existing_contact(contact_payload) || create_contact(contact_payload)
update_existing_contact(imported_contact, contact_payload)
record_mapping('contact', source_id, imported_contact, metadata: contact_metadata(contact_payload))
item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: imported_contact.id)
imported_contact
end
increment_stat('contacts', 'imported') unless already_handled
contact
end
increment_stat('contacts', 'imported') unless already_handled
contact
rescue StandardError => e
raise if e.is_a?(DataImports::Intercom::Client::Error)
@@ -382,68 +395,85 @@ 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'])
if (mapping = find_mapping('message', message_source_id)) && message_mapping_handled?(mapping, source_part)
if mapping.data_import_id == @data_import.id
reconcile_current_run_message_mapping(chatwoot_conversation, mapping, source_part)
return
end
skip_existing_message_mapping(chatwoot_conversation, mapping, source_part)
return
end
create_message(chatwoot_conversation, contact, source_part, message_source_id)
rescue StandardError => e
fail_message(chatwoot_conversation, message_source_id, source_part, e)
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_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)
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']}"
begin
if (mapping = find_mapping('message', message_source_id)) && message_mapping_handled?(mapping, part)
if mapping.data_import_id == @data_import.id
reconcile_current_run_message_mapping(chatwoot_conversation, mapping, part)
next
end
skip_existing_message_mapping(chatwoot_conversation, mapping, part)
next
end
create_message(chatwoot_conversation, contact, part, message_source_id)
rescue StandardError => e
fail_message(chatwoot_conversation, message_source_id, part, e)
end
import_message(chatwoot_conversation, contact, entry)
end
return false if import_stopped?
true
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 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)
attrs = message_attributes(conversation, contact, part, message_source_id, content)
message = nil
part_entries.each do |entry|
return false unless continue_import_with_heartbeat?
import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry)
end
return false if import_stopped?
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
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
raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
end
end
rescue StandardError => e
fail_message(conversation, entry.source_id, entry.part, e)
end
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, 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)
@@ -458,13 +488,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!(
@@ -472,15 +501,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'])
@@ -521,8 +565,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)
@@ -647,7 +690,7 @@ class DataImports::Intercom::Importer
)
item = import_item('contact', source_id, contact_payload) unless item&.imported?
item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: mapped_contact.id)
reconcile_item_stats('contact')
mark_stat_group_dirty('contacts')
end
def reconcile_item_stats(source_object_type)
@@ -655,34 +698,37 @@ class DataImports::Intercom::Importer
group = stat_group_for(source_object_type)
@stats[group]['imported'] = items.imported.count
@stats[group]['skipped'] = items.skipped.count
persist_stats
end
def reconcile_current_run_message_mapping(conversation, mapping, part)
record_skipped_message_log(conversation, mapping.source_object_id, part) if mapping.metadata['skipped']
mark_stat_group_dirty('messages')
end
def reconcile_message_stats
mappings = @data_import.mappings.where(source_provider: PROVIDER, source_object_type: 'message')
skipped_mappings = mappings.where("metadata ->> 'skipped' = ?", 'true').count
message_logs = @data_import.import_errors.where(source_object_type: 'message')
@stats['messages']['imported'] = mappings.count - skipped_mappings
@stats['messages']['skipped'] = message_logs.where("details ->> 'kind' = ?", 'skipped').count
persist_stats
end
def skip_already_imported_item(item, mapping, already_handled:)
item.update!(
status: :skipped,
chatwoot_record_type: mapping.chatwoot_record_type,
chatwoot_record_id: mapping.chatwoot_record_id,
last_error_code: ALREADY_IMPORTED_ERROR_CODE,
last_error_message: 'Already imported in a previous import.'
)
record_already_imported_log(
data_import_item: item,
source_object_type: item.source_object_type,
source_object_id: item.source_object_id,
mapping: mapping
)
DataImportItem.transaction do
item.update!(
status: :skipped,
chatwoot_record_type: mapping.chatwoot_record_type,
chatwoot_record_id: mapping.chatwoot_record_id,
last_error_code: ALREADY_IMPORTED_ERROR_CODE,
last_error_message: 'Already imported in a previous import.'
)
record_already_imported_log(
data_import_item: item,
source_object_type: item.source_object_type,
source_object_id: item.source_object_id,
mapping: mapping
)
end
increment_stat(stat_group_for(item.source_object_type), 'skipped') unless already_handled
end
@@ -694,13 +740,12 @@ class DataImports::Intercom::Importer
already_recorded = skip_log_recorded?('message', mapping.source_object_id, ALREADY_IMPORTED_ERROR_CODE)
record_already_imported_log(source_object_type: 'message', source_object_id: mapping.source_object_id, mapping: mapping)
end
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?
if already_recorded
message_logs = @data_import.import_errors.where(source_object_type: 'message')
@stats['messages']['skipped'] = message_logs.where("details ->> 'kind' = ?", 'skipped').count
else
increment_stat('messages', 'skipped')
end
end
def fail_item(item, error)
@@ -800,7 +845,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)
@@ -953,6 +998,42 @@ class DataImports::Intercom::Importer
@stats[group][key] = @stats[group][key].to_i + 1
end
def mark_stat_group_dirty(group)
@dirty_stat_groups[group] = true
end
def reconcile_dirty_stats
return if @dirty_stat_groups.empty?
with_query_timeout_retry do
@dirty_stat_groups.each_key do |group|
case group
when 'contacts'
reconcile_item_stats('contact')
when 'messages'
reconcile_message_stats
else
raise ArgumentError, "Unsupported Intercom import stat group: #{group}"
end
end
persist_stats
end
@dirty_stat_groups.clear
end
def with_query_timeout_retry
retries = 0
begin
yield
rescue ActiveRecord::QueryCanceled
raise if retries >= QUERY_TIMEOUT_RETRY_LIMIT
retries += 1
sleep(rand(QUERY_TIMEOUT_RETRY_DELAY_RANGE))
retry
end
end
def update_stat_total(group, total)
@stats[group] ||= {}
@stats[group]['total'] = total.to_i
app/services/data_imports/intercom/message_batch_builder.rb
+134
View File
@@ -0,0 +1,134 @@
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 = 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 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
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] if mapping&.chatwoot_record_type == 'Message'
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