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

perf(imports): shorten Intercom jobs and add heartbeats

Browse Source
This commit is contained in:
Sony Mathew
2026-07-17 14:49:58 +05:30
parent 2decd8c544
commit c58ba2f9c2
2 changed files with 87 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/data_imports/intercom/importer.rb
+32 -14
View File
@@ -7,6 +7,9 @@ class DataImports::Intercom::Importer
end
DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze
CONTACTS_PER_PAGE = 50
CONVERSATIONS_PER_PAGE = 10
HEARTBEAT_INTERVAL = 1.minute
PROVIDER = 'intercom'.freeze
ALREADY_IMPORTED_ERROR_CODE = 'DataImports::Intercom::AlreadyImported'.freeze
SKIPPED_MESSAGE_ERROR_CODE = 'DataImports::Intercom::SkippedMessage'.freeze
@@ -63,7 +66,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?
@@ -78,7 +81,7 @@ class DataImports::Intercom::Importer
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?
@@ -154,7 +157,8 @@ class DataImports::Intercom::Importer
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)
import_conversation_parts(conversation, mapped_conversation, contact)
return unless import_conversation_parts(conversation, mapped_conversation, contact)
update_conversation_activity(mapped_conversation)
return
end
@@ -167,14 +171,15 @@ class DataImports::Intercom::Importer
increment_stat('conversations', 'imported') unless already_handled
import_source_message(conversation, chatwoot_conversation, contact)
import_conversation_parts(conversation, chatwoot_conversation, contact)
return unless import_conversation_parts(conversation, chatwoot_conversation, contact)
update_conversation_activity(chatwoot_conversation)
rescue StandardError => e
raise if e.is_a?(DataImports::Intercom::Client::Error)
fail_item(item, e)
ensure
persist_stats
persist_stats unless @import_stopped
end
def import_stopped?
@@ -184,6 +189,14 @@ class DataImports::Intercom::Importer
@import_stopped = @data_import.abandoned? || @data_import.completed? || @data_import.completed_with_errors? || stale_import_run?
end
def continue_import_with_heartbeat?
return true if @data_import.updated_at > HEARTBEAT_INTERVAL.ago
return false if import_stopped?
@data_import.touch if @data_import.updated_at <= HEARTBEAT_INTERVAL.ago
true
end
def stale_import_run?
active_run_id = @data_import.active_intercom_import_run_id
@run_id.present? && active_run_id.present? && active_run_id != @run_id
@@ -396,21 +409,26 @@ class DataImports::Intercom::Importer
record_truncated_conversation_parts(conversation, parts.size)
parts.each do |part|
return false unless continue_import_with_heartbeat?
message_source_id = "conversation:#{source_id_for(conversation)}:part:#{part['id']}"
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)
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
skip_existing_message_mapping(chatwoot_conversation, mapping, part)
next
create_message(chatwoot_conversation, contact, part, message_source_id)
rescue StandardError => e
fail_message(chatwoot_conversation, message_source_id, part, e)
end
create_message(chatwoot_conversation, contact, part, message_source_id)
rescue StandardError => e
fail_message(chatwoot_conversation, message_source_id, part, e)
end
true
end
def create_message(conversation, contact, part, message_source_id)
spec/services/data_imports/intercom/importer_spec.rb
+55 -3
View File
@@ -66,12 +66,12 @@ RSpec.describe DataImports::Intercom::Importer do
before do
account.enable_features!('data_import')
allow(DataImports::Intercom::Client).to receive(:new).with(access_token: 'intercom-token').and_return(client)
allow(client).to receive(:list_contacts).with(starting_after: nil).and_return(
allow(client).to receive(:list_contacts).with(starting_after: nil, per_page: 50).and_return(
'data' => [contact_payload],
'total_count' => 1,
'pages' => { 'next' => nil }
)
allow(client).to receive(:list_conversations).with(starting_after: nil).and_return(
allow(client).to receive(:list_conversations).with(starting_after: nil, per_page: 10).and_return(
'conversations' => [{ 'id' => 'conversation_1' }],
'total_count' => 1,
'pages' => { 'next' => nil }
@@ -188,6 +188,16 @@ RSpec.describe DataImports::Intercom::Importer do
expect(item.metadata['message_total_contribution']).to eq(3)
end
it 'uses smaller conversation pages while retaining the contact page size' do
importer = described_class.new(data_import: data_import)
importer.import_contacts_page
importer.import_conversations_page
expect(client).to have_received(:list_contacts).with(starting_after: nil, per_page: 50)
expect(client).to have_received(:list_conversations).with(starting_after: nil, per_page: 10)
end
it 'reconciles imported message stats from same-run mappings on retry' do
described_class.new(data_import: data_import).import_conversations_page
stats = data_import.reload.stats.deep_dup
@@ -267,7 +277,7 @@ RSpec.describe DataImports::Intercom::Importer do
it 'stops an in-flight page when a newer import run takes over', :aggregate_failures do
run_id = 'intercom-run-1'
data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
allow(client).to receive(:list_conversations).with(starting_after: nil).and_return(
allow(client).to receive(:list_conversations).with(starting_after: nil, per_page: 10).and_return(
'conversations' => [{ 'id' => 'conversation_1' }, { 'id' => 'conversation_2' }],
'pages' => { 'next' => { 'starting_after' => 'next-conversation-cursor' } }
)
@@ -285,6 +295,48 @@ 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
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|
template_part.merge('id' => "part_#{index + 1}")
end
allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation)
heartbeat_times = []
allow(data_import).to receive(:touch).and_wrap_original do |method|
heartbeat_times << Time.current
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
end
importer.import_conversations_page
expect(importer).to have_received(:create_message).once
expect(data_import).not_to have_received(:touch)
expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
end
end
it 'rolls back a newly inserted conversation 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:|