+
+
+
+
+
+
+ + {{ $t('DATA_IMPORTS.EXPORT.TITLE') }} +
++ {{ $t('DATA_IMPORTS.EXPORT.DESCRIPTION') }} +
+
+
+
+
+
+
+
+ + {{ $t('DATA_IMPORTS.TABLE.EMPTY') }} +
++ {{ $t('DATA_IMPORTS.TABLE.EMPTY_DESCRIPTION') }} +
+
+
+
+
+
+
+ ![]()
+
+
+
+
+
+
+
+ {{ dataImport.name || $t('DATA_IMPORTS.TABLE.UNNAMED') }}
+
+
+
+
+ {{ formatStatus(dataImport.status) }}
+
+
+
+
+ {{ importTypeLabel(dataImport) }}
+
+
+ {{
+ $t('DATA_IMPORTS.TABLE.IMPORTED_COUNT', {
+ count: importedCount(dataImport),
+ })
+ }}
+
+
+ {{ formatDate(dataImport.created_at) }}
+
+
+
+
+
+
+
+
+
+
+ {{ monitorTitle }}
+
+
+
+
+ {{
+ isPolling
+ ? $t('DATA_IMPORTS.MONITOR.REFRESHING')
+ : $t('DATA_IMPORTS.MONITOR.LIVE', {
+ seconds: pollIntervalSeconds,
+ })
+ }}
+
+
+
+ + {{ title }} +
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ + {{ emptyMessage }} +
+
+
+
+
+
+ + {{ title }} +
+
+
+
+ {{ item.label }}
+
+
+
+ {{ item.importedLabel }}
+
+
+ {{ `${item.percent}%` }}
+
+
+
+
+
+ {{ item.caption }}
+
+
+
+
+
+ -
+
-
+
+ {{ item.label }} +
+ - + {{ item.value }} + +
+
+ Hello there
', + 'author' => { 'type' => 'user', 'id' => 'contact_1', 'email' => 'CUSTOMER@example.com' } + }, + 'conversation_parts' => { + 'conversation_parts' => [ + { + 'id' => 'part_1', + 'part_type' => 'comment', + 'body' => 'Admin reply
', + 'created_at' => 1_700_000_100, + 'updated_at' => 1_700_000_100, + 'author' => { 'type' => 'admin', 'id' => 'admin_1' }, + 'attachments' => [] + }, + { + 'id' => 'part_2', + 'part_type' => 'note', + 'body' => 'Internal note', + 'created_at' => 1_700_000_150, + 'updated_at' => 1_700_000_150, + 'author' => { 'type' => 'admin', 'id' => 'admin_1' }, + 'attachments' => [] + } + ] + } + } + end + + 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( + 'data' => [contact_payload], + 'total_count' => 1, + 'pages' => { 'next' => nil } + ) + allow(client).to receive(:list_conversations).with(starting_after: nil).and_return( + 'conversations' => [{ 'id' => 'conversation_1' }], + 'total_count' => 1, + 'pages' => { 'next' => nil } + ) + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(conversation_payload) + allow(client).to receive(:retrieve_contact).with('contact_1').and_return(contact_payload) + end + + it 'imports contacts, conversations, messages, and source-bucket inboxes without normal message creation callbacks', :aggregate_failures do + described_class.new(data_import: data_import).perform + + contact = account.contacts.find_by!(email: 'customer@example.com') + expect(contact.name).to eq('Customer One') + expect(contact.phone_number).to eq('+15551234567') + expect(contact).to be_lead + expect(contact.custom_attributes).to include('intercom_contact_id' => 'contact_1') + + inbox = account.inboxes.find_by!(name: 'Intercom Import - Email') + expect(inbox.channel.additional_attributes).to include('source_bucket' => 'email', 'import_placeholder' => true) + + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + expect(conversation).to have_attributes( + status: 'resolved', + inbox_id: inbox.id, + contact_id: contact.id + ) + expect(conversation.additional_attributes.dig('source', 'routing_method')).to eq('source_bucket_api_inbox') + + expect(conversation.messages.order(:created_at).pluck(:content)).to eq(["Need help\n\nHello there", 'Admin reply', 'Internal note']) + expect(conversation.messages.order(:created_at).map(&:message_type)).to eq(%w[incoming outgoing outgoing]) + expect(conversation.messages.order(:created_at).last.private).to be(true) + + expect(data_import.reload).to be_completed + expect(data_import.stats).to include( + 'contacts' => include('imported' => 1, 'skipped' => 0, 'total' => 1), + 'conversations' => include('imported' => 1, 'skipped' => 0, 'total' => 1), + 'messages' => include('imported' => 3, 'skipped' => 0, 'total' => 3), + 'errors' => { 'count' => 0 } + ) + expect(data_import.processed_records).to eq(5) + expect(data_import.items.imported.count).to eq(2) + expect(DataImportMapping.where(data_import: data_import).count).to eq(5) + 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| + dispatched_events << event_name + end + clear_enqueued_jobs + + described_class.new(data_import: data_import).perform + + record_events = [ + Events::Types::CONTACT_CREATED, + Events::Types::CONTACT_UPDATED, + Events::Types::CONVERSATION_CREATED, + Events::Types::CONVERSATION_UPDATED, + Events::Types::CONVERSATION_STATUS_CHANGED, + Events::Types::ASSIGNEE_CHANGED, + Events::Types::TEAM_CHANGED, + Events::Types::MESSAGE_CREATED, + Events::Types::FIRST_REPLY_CREATED, + Events::Types::REPLY_CREATED + ] + side_effect_jobs = [SendReplyJob, EventDispatcherJob, ActionCableBroadcastJob, WebhookJob, HookJob] + + expect(dispatched_events & record_events).to be_empty + expect(enqueued_jobs.pluck(:job) & side_effect_jobs).to be_empty + expect(Notification.where(account: account)).to be_empty + end + + context 'when Intercom contact activity timestamps are available' do + let(:contact_payload) do + super().merge('last_seen_at' => 1_700_000_050, 'last_replied_at' => 1_700_000_090) + end + + it 'prefers last_seen_at for contact activity' do + described_class.new(data_import: data_import).import_contacts_page + + contact = account.contacts.find_by!(email: 'customer@example.com') + expect(contact.last_activity_at).to eq(Time.zone.at(1_700_000_050)) + end + end + + context 'when Intercom contact last_seen_at is unavailable' do + let(:contact_payload) do + super().merge('last_seen_at' => nil, 'last_replied_at' => 1_700_000_090) + end + + it 'falls back to last_replied_at for contact activity' do + described_class.new(data_import: data_import).import_contacts_page + + contact = account.contacts.find_by!(email: 'customer@example.com') + expect(contact.last_activity_at).to eq(Time.zone.at(1_700_000_090)) + end + end + + it 'leaves contact activity blank when Intercom activity timestamps are unavailable' do + described_class.new(data_import: data_import).import_contacts_page + + contact = account.contacts.find_by!(email: 'customer@example.com') + expect(contact.last_activity_at).to be_nil + end + + it 'updates message totals by delta when a conversation page is retried' do + importer = described_class.new(data_import: data_import) + + importer.import_conversations_page + importer.import_conversations_page + + expect(data_import.reload.stats.dig('messages', 'total')).to eq(3) + item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1') + expect(item.metadata['message_total_contribution']).to eq(3) + 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 + stats['messages']['imported'] = 0 + data_import.update!(stats: stats) + + described_class.new(data_import: data_import).import_conversations_page + + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + + it 'indexes imported messages for advanced search' do + allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true) + allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false) + reindexed_message_ids = [] + original_reindex_for_search = Message.instance_method(:reindex_for_search) + Message.define_method(:reindex_for_search) { reindexed_message_ids << id } + 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)) + ensure + Message.define_method(:reindex_for_search, original_reindex_for_search) + Message.__send__(:private, :reindex_for_search) + end + + it 'keeps imported messages successful when search reindexing fails', :aggregate_failures do + allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true) + allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false) + # rubocop:disable RSpec/AnyInstance + allow_any_instance_of(Message).to receive(:reindex_for_search).and_raise(StandardError, 'search unavailable') + # rubocop:enable RSpec/AnyInstance + + described_class.new(data_import: data_import).perform + + message = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:source:source_1') + mapping = data_import.mappings.find_by!(source_object_type: 'message', source_object_id: 'conversation:conversation_1:source:source_1') + expect(mapping.chatwoot_record).to eq(message) + expect(data_import.reload).to be_completed + expect(data_import.import_errors.exists?).to be(false) + expect(data_import.stats.dig('messages', 'imported')).to eq(3) + end + + describe '#start!' do + it 'does not overwrite an import abandoned by another process', :aggregate_failures do + importer = described_class.new(data_import: data_import) + + DataImport.find(data_import.id).update!( + status: :abandoned, + abandoned_at: Time.current + ) + + expect(importer.start!).to be_nil + expect(data_import.reload).to be_abandoned + expect(data_import.started_at).to be_nil + end + end + + describe '#perform' do + it 'stops when the import was abandoned before processing starts' do + importer = described_class.new(data_import: data_import) + DataImport.find(data_import.id).update!( + status: :abandoned, + abandoned_at: Time.current + ) + + expect(client).not_to receive(:list_contacts) + + importer.perform + + expect(data_import.reload).to be_abandoned + end + end + + describe '#import_conversations_page' 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( + 'conversations' => [{ 'id' => 'conversation_1' }, { 'id' => 'conversation_2' }], + 'pages' => { 'next' => { 'starting_after' => 'next-conversation-cursor' } } + ) + allow(client).to receive(:retrieve_conversation).with('conversation_1') do + data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) + conversation_payload + end + + result = described_class.new(data_import: data_import, run_id: run_id).import_conversations_page + + expect(result).to be_done + expect(client).not_to have_received(:retrieve_conversation).with('conversation_2') + expect(account.conversations.where(identifier: 'intercom:conversation_1')).to be_empty + expect(account.contacts.where(email: 'customer@example.com')).to be_empty + expect(data_import.reload.cursor.dig('conversations', 'starting_after')).to be_nil + 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:| + raise StandardError, 'mapping failed' if object_type == 'conversation' + + method.call(object_type, source_id, record, metadata: metadata) + end + + importer.import_conversations_page + + expect(account.conversations.where(identifier: 'intercom:conversation_1')).to be_empty + item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1') + expect(item).to be_failed + expect(item.last_error_message).to eq('mapping failed') + end + + it 'rolls back a newly inserted contact when mapping persistence fails', :aggregate_failures do + sparse_contact = contact_payload.slice('id', 'name', 'created_at', 'updated_at') + allow(client).to receive(:retrieve_contact).with('contact_1').and_return(sparse_contact) + 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 == 'contact' + + method.call(object_type, source_id, record, metadata: metadata) + end + + importer.import_conversations_page + + expect(account.contacts.where(name: 'Customer One')).to be_empty + expect(data_import.mappings.where(source_object_type: 'contact', source_object_id: 'contact_1')).to be_empty + contact_item = data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(contact_item).to be_failed + expect(contact_item.last_error_message).to eq('mapping failed') + end + + 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 + + importer.import_conversations_page + + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + expect(conversation.messages.where(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_empty + error = data_import.import_errors.find_by!( + source_object_type: 'message', + source_object_id: 'conversation:conversation_1:source:source_1' + ) + expect(error).to have_attributes(error_code: 'StandardError', message: 'mapping failed') + end + end + + describe '#finish!' do + it 'does not overwrite an import abandoned by another process' do + data_import.update!(status: :processing) + importer = described_class.new(data_import: data_import) + + DataImport.find(data_import.id).update!( + status: :abandoned, + abandoned_at: Time.current + ) + + importer.finish! + + expect(data_import.reload).to be_abandoned + expect(data_import.completed_at).to be_nil + end + end + + describe '#fail!' do + it 'does not overwrite an import abandoned by another process', :aggregate_failures do + data_import.update!(status: :processing) + importer = described_class.new(data_import: data_import) + + DataImport.find(data_import.id).update!( + status: :abandoned, + abandoned_at: Time.current + ) + + importer.fail!(StandardError.new('boom')) + + expect(data_import.reload).to be_abandoned + expect(data_import.last_error_at).to be_nil + expect(data_import.import_errors.exists?).to be(false) + end + end + + context 'when the Intercom records were imported by an earlier run' do + let(:next_data_import) do + create( + :data_import, :intercom, + account: account + ) + end + + it 'records the already mapped records as skipped for the current import run', :aggregate_failures do + described_class.new(data_import: data_import).perform + + described_class.new(data_import: next_data_import).perform + + expect(next_data_import.reload.stats).to include( + 'contacts' => include('imported' => 0, 'skipped' => 1, 'total' => 1), + 'conversations' => include('imported' => 0, 'skipped' => 1, 'total' => 1), + 'messages' => include('imported' => 0, 'skipped' => 3, 'total' => 3), + 'errors' => { 'count' => 0 } + ) + expect(next_data_import).to be_completed + expect(next_data_import.total_records).to eq(5) + expect(next_data_import.processed_records).to eq(0) + expect(next_data_import.items.skipped.count).to eq(2) + expect(next_data_import.import_errors.skip_logs.group(:source_object_type).count).to eq( + 'contact' => 1, + 'conversation' => 1, + 'message' => 3 + ) + expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported']) + end + + it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do + described_class.new(data_import: data_import).perform + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + Message.where(conversation_id: conversation.id).delete_all + + described_class.new(data_import: next_data_import).perform + + expect(conversation.reload.messages.pluck(:source_id)).to match_array( + %w[ + intercom:conversation:conversation_1:source:source_1 + intercom:conversation:conversation_1:part:part_1 + intercom:conversation:conversation_1:part:part_2 + ] + ) + expect(next_data_import.reload.stats).to include( + 'contacts' => include('imported' => 0, 'skipped' => 1, 'total' => 1), + 'conversations' => include('imported' => 0, 'skipped' => 1, 'total' => 1), + 'messages' => include('imported' => 3, 'skipped' => 0, 'total' => 3), + 'errors' => { 'count' => 0 } + ) + 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) + end + + it 'updates conversation activity when a later import adds new messages to the mapped conversation', :aggregate_failures do + new_part = { + 'id' => 'part_3', + 'part_type' => 'comment', + 'body' => 'Follow-up reply
', + 'created_at' => 1_700_000_300, + 'updated_at' => 1_700_000_300, + 'author' => { 'type' => 'admin', 'id' => 'admin_1' }, + 'attachments' => [] + } + updated_conversation_payload = conversation_payload.deep_dup + updated_conversation_payload['updated_at'] = 1_700_000_300 + updated_conversation_payload['conversation_parts']['conversation_parts'] << new_part + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return( + conversation_payload, + updated_conversation_payload + ) + + described_class.new(data_import: data_import).perform + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + + described_class.new(data_import: next_data_import).perform + + expect(conversation.reload.last_activity_at).to eq(Time.zone.at(1_700_000_300)) + expect(conversation.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:part_3').content).to eq('Follow-up reply') + end + end + + context 'when a conversation references an already mapped contact' do + it 'reuses the mapped contact without hydrating the sparse reference' do + described_class.new(data_import: data_import).import_contacts_page + + expect(client).not_to receive(:retrieve_contact) + + described_class.new(data_import: data_import).import_conversations_page + end + end + + context 'when a same-run contact mapping outlives its item progress' do + let!(:mapped_contact) { create(:contact, account: account) } + + before do + DataImportMapping.create!( + account: account, + data_import: data_import, + source_provider: 'intercom', + source_object_type: 'contact', + source_object_id: 'contact_1', + chatwoot_record_type: 'Contact', + chatwoot_record_id: mapped_contact.id, + metadata: {} + ) + data_import.items.create!( + source_provider: 'intercom', + source_object_type: 'contact', + source_object_id: 'contact_1', + status: :processing, + metadata: contact_payload + ) + end + + it 'repairs the item and imported count on retry', :aggregate_failures do + described_class.new(data_import: data_import).import_contacts_page + + item = data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(item).to be_imported + expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: mapped_contact.id) + expect(data_import.reload.stats.dig('contacts', 'imported')).to eq(1) + end + end + + context 'when an existing contact has the same email but a different external id' do + let(:contact_payload) do + super().merge('last_replied_at' => 1_700_000_090) + end + let!(:existing_contact) { create(:contact, account: account, email: 'customer@example.com', identifier: nil) } + + it 'updates the existing contact instead of creating a duplicate', :aggregate_failures do + described_class.new(data_import: data_import).import_contacts_page + + expect(existing_contact.reload.identifier).to eq('external_1') + expect(existing_contact.last_activity_at).to eq(Time.zone.at(1_700_000_090)) + expect(account.contacts.where(email: 'customer@example.com').count).to eq(1) + item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id) + end + end + + context 'when an existing contact has the same phone but a different external id' do + let(:contact_payload) do + super().merge('email' => nil) + end + let!(:existing_contact) { create(:contact, account: account, phone_number: '+15551234567', identifier: nil) } + + it 'updates the existing contact instead of creating a duplicate', :aggregate_failures do + described_class.new(data_import: data_import).import_contacts_page + + expect(existing_contact.reload.identifier).to eq('external_1') + expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1) + item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id) + end + end + + context 'when an existing contact has the same phone but Intercom sends a new email' do + let!(:existing_contact) { create(:contact, account: account, phone_number: '+15551234567', identifier: nil) } + + it 'falls through to the phone match after the email lookup misses', :aggregate_failures do + described_class.new(data_import: data_import).import_contacts_page + + expect(existing_contact.reload.email).to eq('customer@example.com') + expect(existing_contact.identifier).to eq('external_1') + expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1) + item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id) + end + end + + context 'when an existing visitor contact matches the Intercom external id' do + let!(:existing_contact) { create(:contact, account: account, identifier: 'external_1') } + + it 'promotes the contact to a lead when adding email or phone', :aggregate_failures do + expect(existing_contact).to be_visitor + + described_class.new(data_import: data_import).import_contacts_page + + expect(existing_contact.reload).to be_lead + expect(existing_contact.email).to eq('customer@example.com') + expect(existing_contact.phone_number).to eq('+15551234567') + end + end + + context 'when an identifier match has contact details owned by another contact' do + let!(:existing_contact) { create(:contact, account: account, identifier: 'external_1') } + let!(:email_owner) { create(:contact, account: account, email: 'customer@example.com') } + let!(:phone_owner) { create(:contact, account: account, phone_number: '+15551234567') } + + it 'does not copy the conflicting email or phone number', :aggregate_failures do + described_class.new(data_import: data_import).import_contacts_page + + expect(existing_contact.reload.email).to be_nil + expect(existing_contact.phone_number).to be_nil + expect(existing_contact).to be_visitor + expect(email_owner.reload.email).to eq('customer@example.com') + expect(phone_owner.reload.phone_number).to eq('+15551234567') + expect(account.contacts.where(email: 'customer@example.com').count).to eq(1) + expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1) + + item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id) + end + end + + context 'when Intercom rate limits a conversation detail request' do + before do + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_raise( + DataImports::Intercom::Client::RateLimitError.new('rate limited', status: 429) + ) + end + + it 're-raises the provider error so the page job can retry', :aggregate_failures do + expect { described_class.new(data_import: data_import).import_conversations_page } + .to raise_error(DataImports::Intercom::Client::RateLimitError) + + item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1') + expect(item).to be_processing + expect(data_import.import_errors.exists?).to be(false) + end + end + + context 'when Intercom rate limits a contact hydration request' do + before do + allow(client).to receive(:retrieve_contact).with('contact_1').and_raise( + DataImports::Intercom::Client::RateLimitError.new('rate limited', status: 429) + ) + end + + it 're-raises the provider error instead of importing a sparse contact', :aggregate_failures do + expect { described_class.new(data_import: data_import).import_conversations_page } + .to raise_error(DataImports::Intercom::Client::RateLimitError) + + expect(data_import.items.exists?(source_object_type: 'contact')).to be(false) + expect(data_import.import_errors.exists?).to be(false) + end + end + + context 'when Intercom no longer has a sparse contact referenced by a conversation' do + before do + allow(client).to receive(:retrieve_contact).with('contact_1').and_raise( + DataImports::Intercom::Client::Error.new('not found', status: 404) + ) + end + + it 'falls back to the conversation contact reference', :aggregate_failures do + expect { described_class.new(data_import: data_import).import_conversations_page }.not_to raise_error + + expect(data_import.items.imported.exists?(source_object_type: 'contact', source_object_id: 'contact_1')).to be(true) + expect(data_import.import_errors.exists?).to be(false) + end + end + + context 'when the Intercom source message only has attachments' do + let(:conversation_payload) do + super().deep_merge( + 'source' => { + 'subject' => nil, + 'body' => nil, + 'attachments' => [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }] + }, + 'conversation_parts' => { + 'conversation_parts' => [] + } + ) + end + + it 'imports the source message attachment placeholder', :aggregate_failures do + described_class.new(data_import: data_import).perform + + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + expect(conversation.messages.pluck(:content)).to eq(['[Intercom attachment skipped: 1]']) + expect(conversation.messages.first.additional_attributes.dig('source', 'attachments')).to eq( + [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }] + ) + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(1) + end + end + + context 'when the Intercom source message has text and attachments' do + let(:conversation_payload) do + super().deep_merge( + 'source' => { + 'attachments' => [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }] + } + ) + end + + it 'adds an attachment omission marker to the imported message', :aggregate_failures do + described_class.new(data_import: data_import).perform + + message = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:source:source_1') + expect(message.content).to eq("Need help\n\nHello there\n\n[Intercom attachment skipped: 1]") + expect(message.additional_attributes.dig('source', 'attachments')).to eq( + [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }] + ) + expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(0) + end + end + + context 'when Intercom omits the conversation source' do + let(:conversation_payload) do + super().merge( + 'source' => nil, + 'first_contact_reply' => { + 'type' => 'whatsapp', + 'created_at' => 1_700_000_000, + 'url' => nil + } + ) + end + + it 'routes the conversation from the first contact reply type', :aggregate_failures do + described_class.new(data_import: data_import).perform + + inbox = account.inboxes.find_by!(name: 'Intercom Import - WhatsApp') + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + + expect(conversation.inbox).to eq(inbox) + expect(conversation.additional_attributes.dig('source', 'source_type')).to eq('whatsapp') + end + end + + context 'when an Intercom chat message part cannot be imported' do + let(:conversation_payload) do + super().deep_merge( + 'conversation_parts' => { + 'conversation_parts' => [ + { + 'id' => 'blank_part', + 'part_type' => 'comment', + 'body' => nil, + 'created_at' => 1_700_000_175, + 'updated_at' => 1_700_000_175, + 'author' => { 'type' => 'admin', 'id' => 'admin_1' }, + 'attachments' => [] + } + ] + } + ) + end + + it 'records a skip log with the Intercom message source id', :aggregate_failures do + described_class.new(data_import: data_import).perform + + skip_log = data_import.import_errors.skip_logs.find_by!(source_object_type: 'message') + expect(skip_log).to have_attributes( + source_object_id: 'conversation:conversation_1:part:blank_part', + error_code: 'DataImports::Intercom::SkippedMessage', + message: 'Skipped Intercom comment event blank_part: no message body or attachments to import.' + ) + expect(skip_log.details).to include( + 'kind' => 'skipped', + 'reason' => 'blank_or_unsupported_intercom_part', + 'reason_details' => 'no message body or attachments to import', + 'event_name' => 'comment', + 'event_type' => 'comment', + 'author_type' => 'admin' + ) + expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1) + end + + it 'records the skip log again for a later import run', :aggregate_failures do + described_class.new(data_import: data_import).perform + next_data_import = create( + :data_import, :intercom, + account: account + ) + + described_class.new(data_import: next_data_import).perform + + skip_log = next_data_import.import_errors.skip_logs.find_by!( + source_object_type: 'message', + source_object_id: 'conversation:conversation_1:part:blank_part', + error_code: 'DataImports::Intercom::SkippedMessage' + ) + expect(skip_log).to have_attributes( + source_object_id: 'conversation:conversation_1:part:blank_part', + error_code: 'DataImports::Intercom::SkippedMessage' + ) + expect(next_data_import.reload.stats.dig('messages', 'skipped')).to eq(2) + end + + it 'reconciles a same-run skipped mapping and missing skip log on retry', :aggregate_failures do + described_class.new(data_import: data_import).import_conversations_page + data_import.import_errors.where(source_object_type: 'message').delete_all + stats = data_import.reload.stats.deep_dup + stats['messages']['skipped'] = 0 + data_import.update!(stats: stats) + + described_class.new(data_import: data_import).import_conversations_page + + expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1) + expect(data_import.import_errors.skip_logs.exists?(source_object_id: 'conversation:conversation_1:part:blank_part')).to be(true) + end + + it 'repairs a previously skipped mapping when the part is now an activity', :aggregate_failures do + described_class.new(data_import: data_import).perform + previous_skip_log = data_import.import_errors.skip_logs.find_by!(source_object_id: 'conversation:conversation_1:part:blank_part') + conversation_payload.dig('conversation_parts', 'conversation_parts').first.merge!( + 'part_type' => 'assignment', + 'assigned_to' => { 'name' => 'Support' } + ) + next_data_import = create(:data_import, :intercom, account: account) + + described_class.new(data_import: next_data_import).perform + + activity = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:blank_part') + mapping = DataImportMapping.find_by!( + account: account, + source_provider: 'intercom', + source_object_type: 'message', + source_object_id: 'conversation:conversation_1:part:blank_part' + ) + expect(activity).to be_activity + expect(activity.content).to eq('Intercom teammate assigned the conversation to Support') + expect(mapping.chatwoot_record).to eq(activity) + expect(data_import.import_errors.skip_logs).to include(previous_skip_log) + expect(next_data_import.import_errors.skip_logs.where(source_object_id: mapping.source_object_id)).to be_empty + end + end + + context 'when Intercom returns bodyless lifecycle events' do + let(:conversation_payload) do + super().deep_merge( + 'conversation_parts' => { + 'total_count' => 1, + 'conversation_parts' => [ + { + 'id' => 'assignment_part', + 'part_type' => 'assignment', + 'body' => nil, + 'created_at' => 1_700_000_175, + 'author' => { 'type' => 'admin', 'name' => 'Avery' }, + 'assigned_to' => { 'type' => 'team', 'name' => 'Support' }, + 'state' => 'open', + 'tags' => { 'tags' => [{ 'name' => 'priority' }] }, + 'event_details' => { 'source' => 'workflow' }, + 'app_package_code' => 'workflow' + } + ] + } + ) + end + + it 'imports events as public activity messages with source metadata', :aggregate_failures do + described_class.new(data_import: data_import).perform + + activity = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:assignment_part') + expect(activity).to have_attributes( + message_type: 'activity', + content: 'Avery assigned the conversation to Support', + private: false, + sender: nil, + created_at: Time.zone.at(1_700_000_175) + ) + expect(activity.additional_attributes['source']).to include( + 'part_type' => 'assignment', + 'assigned_to' => include('name' => 'Support'), + 'state' => 'open', + 'event_details' => include('source' => 'workflow'), + 'app_package_code' => 'workflow' + ) + expect(data_import.reload.stats['messages']).to include('imported' => 2, 'skipped' => 0, 'total' => 2) + expect(data_import.import_errors.skip_logs).to be_empty + end + end + + context 'when Intercom omits older conversation parts from the retrieved conversation' do + let(:conversation_payload) do + super().deep_merge( + 'conversation_parts' => { + 'total_count' => 503 + }, + 'statistics' => { + 'count_conversation_parts' => 503 + } + ) + end + + it 'records an incomplete import error and completes with errors', :aggregate_failures do + described_class.new(data_import: data_import).perform + + error = data_import.import_errors.non_skip_logs.find_by!( + source_object_type: 'conversation', + source_object_id: 'conversation_1', + error_code: 'DataImports::Intercom::TruncatedConversationParts' + ) + expect(error.message).to eq('Intercom returned 2 of 503 conversation parts.') + expect(error.details).to include( + 'kind' => 'incomplete', + 'imported_parts_count' => 2, + 'total_parts_count' => 503 + ) + expect(data_import.reload).to be_completed_with_errors + expect(data_import.stats.dig('errors', 'count')).to eq(1) + end + end + + context 'when the conversation parts total matches the returned parts' do + let(:conversation_payload) do + super().deep_merge( + 'conversation_parts' => { + 'total_count' => 2 + }, + 'statistics' => { + 'count_conversation_parts' => 2 + } + ) + end + + it 'does not record a truncated parts error', :aggregate_failures do + described_class.new(data_import: data_import).perform + + expect(data_import.import_errors.non_skip_logs).to be_empty + expect(data_import.reload).to be_completed + expect(data_import.stats.dig('errors', 'count')).to eq(0) + end + end + + context 'when Intercom statistics count is higher than the conversation parts total' do + let(:conversation_payload) do + super().deep_merge( + 'source' => {}, + 'conversation_parts' => { + 'total_count' => 2 + }, + 'statistics' => { + 'count_conversation_parts' => 3 + } + ) + end + + it 'trusts the returned conversation parts total over the statistics counter', :aggregate_failures do + described_class.new(data_import: data_import).perform + + expect(data_import.import_errors.non_skip_logs).to be_empty + expect(data_import.reload).to be_completed + expect(data_import.stats.dig('errors', 'count')).to eq(0) + end + end + + context 'when a specific Intercom message part fails to persist' do + let(:conversation_payload) do + super().deep_merge( + 'conversation_parts' => { + 'conversation_parts' => [ + { + 'id' => 'bad_part', + 'part_type' => 'comment', + 'body' => 'Message that cannot be stored
', + 'created_at' => 1_700_000_175, + 'updated_at' => 1_700_000_175, + 'author' => { 'type' => 'admin', 'id' => 'admin_1' }, + 'attachments' => [] + } + ] + } + ) + end + + before do + allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs| + raise ActiveRecord::StatementInvalid, 'bad message' if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part' + + method.call(records, **kwargs) + end + end + + it 'records a skip log with the Intercom message part id', :aggregate_failures do + described_class.new(data_import: data_import).perform + + skip_log = data_import.import_errors.skip_logs.find_by!(source_object_type: 'message') + expect(skip_log).to have_attributes( + source_object_id: 'conversation:conversation_1:part:bad_part', + error_code: 'ActiveRecord::StatementInvalid', + message: 'bad message' + ) + expect(skip_log.details).to include( + 'kind' => 'failed', + 'conversation_id' => 'intercom:conversation_1' + ) + expect(data_import.reload).to be_completed_with_errors + expect(data_import.stats.dig('errors', 'count')).to eq(1) + end + end +end diff --git a/spec/services/data_imports/intercom/placeholder_inbox_builder_spec.rb b/spec/services/data_imports/intercom/placeholder_inbox_builder_spec.rb new file mode 100644 index 000000000..c1a7baab5 --- /dev/null +++ b/spec/services/data_imports/intercom/placeholder_inbox_builder_spec.rb @@ -0,0 +1,33 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::PlaceholderInboxBuilder do + let(:account) { create(:account) } + + describe '#inbox_for' do + it 'creates a source-bucket API inbox for an Intercom conversation source' do + inbox = described_class.new(account: account).inbox_for('email') + + expect(inbox.name).to eq('Intercom Import - Email') + expect(inbox.channel).to be_a(Channel::Api) + expect(inbox.enable_auto_assignment).to be(false) + expect(inbox.allow_messages_after_resolved).to be(false) + expect(inbox.channel.additional_attributes).to include( + 'source_provider' => 'intercom', + 'source_bucket' => 'email', + 'import_placeholder' => true, + 'agent_reply_time_window' => 1 + ) + end + + it 'reuses an existing placeholder inbox for the same source bucket' do + builder = described_class.new(account: account) + + first_inbox = builder.inbox_for('phone_call') + expect(account).not_to receive(:inboxes) + second_inbox = builder.inbox_for('phone_switch') + + expect(second_inbox).to eq(first_inbox) + expect(Inbox.where(account: account, channel_type: 'Channel::Api').count).to eq(1) + end + end +end diff --git a/spec/services/data_imports/intercom/restart_service_spec.rb b/spec/services/data_imports/intercom/restart_service_spec.rb new file mode 100644 index 000000000..5dbc8f5ab --- /dev/null +++ b/spec/services/data_imports/intercom/restart_service_spec.rb @@ -0,0 +1,64 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::RestartService do + let(:account) { create(:account) } + let(:data_import) { create(:data_import, :intercom, account: account, status: :abandoned, abandoned_at: 1.hour.ago) } + + it 'prepares a failed or abandoned import for another run', :aggregate_failures do + data_import.update!( + stats: { + 'contacts' => { 'imported' => 1, 'skipped' => 9, 'total' => 10 }, + 'conversations' => { 'imported' => 2, 'skipped' => 8, 'total' => 10 }, + 'messages' => { 'imported' => 3, 'skipped' => 7, 'total' => 10 }, + 'errors' => { 'count' => 6 } + } + ) + data_import.import_errors.create!(error_code: 'StandardError', message: 'old run error') + data_import.import_errors.create!( + error_code: 'ContactFailed', + message: 'old contact error', + source_object_type: 'contact', + details: { kind: 'failed' } + ) + retained_skip_log = data_import.import_errors.create!( + error_code: DataImports::Intercom::Importer::ALREADY_IMPORTED_ERROR_CODE, + message: 'old skip log', + source_object_type: 'contact', + details: { kind: 'skipped' } + ) + previous_run_id = data_import.assign_active_intercom_import_run_id + data_import.save! + service = described_class.new(account: account, data_import: data_import) + + expect(service.perform).to eq(:enqueue) + expect(service.data_import).to be_pending + expect(service.data_import.abandoned_at).to be_nil + expect(service.data_import.started_at).to be_nil + expect(service.data_import.active_intercom_import_run_id).not_to eq(previous_run_id) + expect(service.data_import.import_errors).to contain_exactly(retained_skip_log) + expect(service.data_import.stats).to eq( + 'contacts' => { 'imported' => 1, 'skipped' => 1, 'total' => 10 }, + 'conversations' => { 'imported' => 2, 'skipped' => 0, 'total' => 10 }, + 'messages' => { 'imported' => 3, 'skipped' => 0, 'total' => 10 }, + 'errors' => { 'count' => 0 } + ) + end + + it 'returns the active import instead of restarting another import', :aggregate_failures do + active_import = create(:data_import, :intercom, account: account, status: :processing) + service = described_class.new(account: account, data_import: data_import) + + expect(service.perform).to eq(:render_show) + expect(service.data_import).to eq(active_import) + expect(data_import.reload).to be_abandoned + end + + it 'does not restart when the stored access token is missing' do + data_import.update!(access_token: nil) + + result = described_class.new(account: account, data_import: data_import).perform + + expect(result).to eq(:access_token_missing) + expect(data_import.reload).to be_abandoned + end +end diff --git a/spec/services/data_imports/intercom/source_bucket_spec.rb b/spec/services/data_imports/intercom/source_bucket_spec.rb new file mode 100644 index 000000000..b3db4294c --- /dev/null +++ b/spec/services/data_imports/intercom/source_bucket_spec.rb @@ -0,0 +1,17 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::SourceBucket do + describe '.for' do + it 'maps Intercom source types to Chatwoot inbox buckets' do + expect(described_class.for('email')).to eq({ key: 'email', name: 'Email' }) + expect(described_class.for('phone_switch')).to eq({ key: 'phone', name: 'Phone' }) + expect(described_class.for('inapp')).to eq({ key: 'messenger', name: 'Messenger' }) + expect(described_class.for('messenger')).to eq({ key: 'messenger', name: 'Messenger' }) + expect(described_class.for('push')).to eq({ key: 'messenger', name: 'Messenger' }) + end + + it 'uses an unknown bucket for unsupported source types' do + expect(described_class.for('unsupported_source')).to eq({ key: 'unknown', name: 'Unknown' }) + end + end +end