diff --git a/.circleci/config.yml b/.circleci/config.yml index 59702c139..c533e1402 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -77,7 +77,8 @@ jobs: - node/install: node-version: '24.13' - - node/install-pnpm + - node/install-pnpm: + version: '10.2.0' - node/install-packages: pkg-manager: pnpm override-ci-command: pnpm i @@ -118,7 +119,8 @@ jobs: - checkout - node/install: node-version: '24.13' - - node/install-pnpm + - node/install-pnpm: + version: '10.2.0' - node/install-packages: pkg-manager: pnpm override-ci-command: pnpm i @@ -149,7 +151,8 @@ jobs: - checkout - node/install: node-version: '24.13' - - node/install-pnpm + - node/install-pnpm: + version: '10.2.0' - node/install-packages: pkg-manager: pnpm override-ci-command: pnpm i diff --git a/app/controllers/api/v1/accounts/data_imports_controller.rb b/app/controllers/api/v1/accounts/data_imports_controller.rb new file mode 100644 index 000000000..7f0d28e81 --- /dev/null +++ b/app/controllers/api/v1/accounts/data_imports_controller.rb @@ -0,0 +1,159 @@ +require 'csv' + +class Api::V1::Accounts::DataImportsController < Api::V1::Accounts::BaseController + DATA_IMPORT_FEATURE = 'data_import'.freeze + + before_action :ensure_data_import_feature_enabled + before_action :set_data_import, only: [:show, :start, :abandon, :error_logs, :skip_logs] + before_action :check_authorization + + def index + @data_imports = policy_scope(Current.account.data_imports).includes(:initiated_by).order(created_at: :desc) + data_import_ids = @data_imports.map(&:id) + @import_errors_counts = DataImportError.non_skip_logs.where(data_import_id: data_import_ids).group(:data_import_id).count + @skip_logs_counts = DataImportError.skip_logs.where(data_import_id: data_import_ids).group(:data_import_id).count + end + + def show + render_show + end + + def validate_source + totals = validate_intercom_source + render json: { valid: true, totals: totals } + rescue DataImports::Intercom::Client::AuthenticationError + render_source_validation_error('We could not validate this Intercom access key. Check the key and its permissions.') + rescue DataImports::Intercom::Client::Error + render_source_validation_error('Intercom could not be reached. Please try again.') + rescue ArgumentError => e + render_source_validation_error(e.message) + end + + def create + @data_import = creation_service.perform + unless @data_import + render json: { message: 'Another data import is already in progress.' }, status: :unprocessable_entity + return + end + + DataImports::Intercom::ImportJob.perform_later(@data_import, @data_import.active_intercom_import_run_id) + render_show + rescue DataImports::Intercom::Client::AuthenticationError + render_source_validation_error('We could not validate this Intercom access key. Check the key and its permissions.') + rescue DataImports::Intercom::Client::Error + render_source_validation_error('Intercom could not be reached. Please try again.') + rescue ArgumentError => e + render_source_validation_error(e.message) + end + + def start + restart_service = DataImports::Intercom::RestartService.new(account: Current.account, data_import: @data_import) + restart_result = restart_service.perform + @data_import = restart_service.data_import + if restart_result == :access_token_missing + render json: { message: 'The Intercom access key for this import is unavailable.' }, status: :unprocessable_entity + return + end + + DataImports::Intercom::ImportJob.perform_later(@data_import, @data_import.active_intercom_import_run_id) if restart_result == :enqueue + render_show + end + + def abandon + @data_import.abandon! + render_show + end + + def skip_logs + send_data( + skip_logs_csv, + filename: "data-import-#{@data_import.id}-skip-logs.csv", + type: 'text/csv' + ) + end + + def error_logs + send_data( + error_logs_csv, + filename: "data-import-#{@data_import.id}-error-logs.csv", + type: 'text/csv' + ) + end + + private + + def ensure_data_import_feature_enabled + raise Pundit::NotAuthorizedError unless Current.account.feature_enabled?(DATA_IMPORT_FEATURE) + end + + def set_data_import + @data_import = Current.account.data_imports.find(params[:id]) + end + + def check_authorization + authorize(@data_import || DataImport) + end + + def permitted_params + params.permit(:name, :source_provider, :access_token, import_types: []) + end + + def creation_service + DataImports::Intercom::CreationService.new( + account: Current.account, + initiated_by: Current.user, + source_params: permitted_params.to_h + ) + end + + def import_types + return DataImports::Intercom::Importer::DEFAULT_IMPORT_TYPES unless permitted_params.key?(:import_types) + + Array(permitted_params[:import_types]).compact_blank + end + + def validate_intercom_source + raise ArgumentError, 'Unsupported import source.' unless permitted_params[:source_provider] == 'intercom' + + DataImports::Intercom::CredentialsValidator.new( + access_token: permitted_params[:access_token], + import_types: import_types + ).perform + end + + def render_source_validation_error(message) + render json: { valid: false, message: message }, status: :unprocessable_entity + end + + def render_show + @import_errors_finder = DataImportErrorFinder.new(@data_import) + @skip_logs_finder = DataImportSkipLogFinder.new(@data_import, params) + render :show + end + + def skip_logs_csv + logs_csv(@data_import.import_errors.skip_logs) + end + + def error_logs_csv + logs_csv(@data_import.import_errors.non_skip_logs) + end + + def logs_csv(logs) + CSV.generate(headers: true) do |csv| + csv << %w[created_at kind source_object_type source_object_id error_code message details] + + logs.order(:created_at).find_each do |log| + csv << [ + log.created_at.iso8601, + log.details['kind'], + log.source_object_type, + log.source_object_id, + log.error_code, + log.message, + log.details.to_json + ] + end + end + end +end diff --git a/app/controllers/api/v1/accounts/inboxes_controller.rb b/app/controllers/api/v1/accounts/inboxes_controller.rb index 9f56c3817..bb9ca2a70 100644 --- a/app/controllers/api/v1/accounts/inboxes_controller.rb +++ b/app/controllers/api/v1/accounts/inboxes_controller.rb @@ -124,8 +124,8 @@ class Api::V1::Accounts::InboxesController < Api::V1::Accounts::BaseController end def reauthorize_and_update_channel(channel_attributes) - @inbox.channel.reauthorized! if @inbox.channel.respond_to?(:reauthorized!) @inbox.channel.update!(permitted_params(channel_attributes)[:channel]) + @inbox.channel.reauthorized! if @inbox.channel.respond_to?(:reauthorized!) end def update_channel_feature_flags diff --git a/app/controllers/api/v1/widget/base_controller.rb b/app/controllers/api/v1/widget/base_controller.rb index 5b87e2d1a..3912e5b6e 100644 --- a/app/controllers/api/v1/widget/base_controller.rb +++ b/app/controllers/api/v1/widget/base_controller.rb @@ -59,6 +59,10 @@ class Api::V1::Widget::BaseController < ApplicationController permitted_params.dig(:contact, :phone_number) end + def contact_custom_attributes + permitted_params.dig(:contact, :custom_attributes)&.to_h + end + def browser_params { browser_name: browser.name, diff --git a/app/controllers/api/v1/widget/conversations_controller.rb b/app/controllers/api/v1/widget/conversations_controller.rb index 00e718614..8f5977d54 100644 --- a/app/controllers/api/v1/widget/conversations_controller.rb +++ b/app/controllers/api/v1/widget/conversations_controller.rb @@ -19,7 +19,7 @@ class Api::V1::Widget::ConversationsController < Api::V1::Widget::BaseController def process_update_contact @contact = ContactIdentifyAction.new( contact: @contact, - params: { email: contact_email, phone_number: contact_phone_number, name: contact_name }, + params: { email: contact_email, phone_number: contact_phone_number, name: contact_name, custom_attributes: contact_custom_attributes }, retain_original_contact_name: true, discard_invalid_attrs: true ).perform @@ -95,7 +95,7 @@ class Api::V1::Widget::ConversationsController < Api::V1::Widget::BaseController end def permitted_params - params.permit(:id, :typing_status, :website_token, :email, contact: [:name, :email, :phone_number], + params.permit(:id, :typing_status, :website_token, :email, contact: [:name, :email, :phone_number, { custom_attributes: {} }], message: [:content, :referer_url, :timestamp, :echo_id], custom_attributes: {}) end diff --git a/app/finders/data_import_error_finder.rb b/app/finders/data_import_error_finder.rb new file mode 100644 index 000000000..69d85b23c --- /dev/null +++ b/app/finders/data_import_error_finder.rb @@ -0,0 +1,11 @@ +class DataImportErrorFinder + RESULTS_LIMIT = 5 + + def initialize(data_import) + @data_import = data_import + end + + def import_errors + @data_import.import_errors.non_skip_logs.order(created_at: :desc).limit(RESULTS_LIMIT) + end +end diff --git a/app/finders/data_import_skip_log_finder.rb b/app/finders/data_import_skip_log_finder.rb new file mode 100644 index 000000000..ee932f1da --- /dev/null +++ b/app/finders/data_import_skip_log_finder.rb @@ -0,0 +1,35 @@ +class DataImportSkipLogFinder + RESULTS_LIMIT = 5 + SOURCE_OBJECT_TYPES = %w[contact conversation message].freeze + + attr_reader :selected_source_object_type + + def initialize(data_import, params = {}) + @data_import = data_import + @selected_source_object_type = valid_source_object_type(params[:skip_logs_type]) + end + + def skip_logs + filtered_scope.order(created_at: :desc).limit(RESULTS_LIMIT) + end + + def counts_by_type + base_scope.group(:source_object_type).count + end + + private + + def base_scope + @base_scope ||= @data_import.import_errors.skip_logs + end + + def filtered_scope + return base_scope if selected_source_object_type.blank? + + base_scope.where(source_object_type: selected_source_object_type) + end + + def valid_source_object_type(source_object_type) + source_object_type if SOURCE_OBJECT_TYPES.include?(source_object_type) + end +end diff --git a/app/javascript/dashboard/api/dataImports.js b/app/javascript/dashboard/api/dataImports.js new file mode 100644 index 000000000..b4c15b98a --- /dev/null +++ b/app/javascript/dashboard/api/dataImports.js @@ -0,0 +1,39 @@ +/* global axios */ + +import ApiClient from './ApiClient'; + +class DataImportsAPI extends ApiClient { + constructor() { + super('data_imports', { accountScoped: true }); + } + + start(id) { + return axios.post(`${this.url}/${id}/start`); + } + + abandon(id) { + return axios.post(`${this.url}/${id}/abandon`); + } + + show(id, params = {}) { + return axios.get(`${this.url}/${id}`, { params }); + } + + validateSource(payload) { + return axios.post(`${this.url}/validate_source`, payload); + } + + downloadSkipLogs(id) { + return axios.get(`${this.url}/${id}/skip_logs.csv`, { + responseType: 'blob', + }); + } + + downloadErrorLogs(id) { + return axios.get(`${this.url}/${id}/error_logs.csv`, { + responseType: 'blob', + }); + } +} + +export default new DataImportsAPI(); diff --git a/app/javascript/dashboard/components-next/banner/Banner.vue b/app/javascript/dashboard/components-next/banner/Banner.vue index c9b86d42b..f44a6ae5e 100644 --- a/app/javascript/dashboard/components-next/banner/Banner.vue +++ b/app/javascript/dashboard/components-next/banner/Banner.vue @@ -61,10 +61,10 @@ const triggerAction = () => {
Customer confirmed resolution
' + } + + expect(described_class.new(part).perform).to eq( + 'Intercom teammate closed the conversation: Customer confirmed resolution' + ) + end +end diff --git a/spec/services/data_imports/intercom/client_spec.rb b/spec/services/data_imports/intercom/client_spec.rb new file mode 100644 index 000000000..78498121a --- /dev/null +++ b/spec/services/data_imports/intercom/client_spec.rb @@ -0,0 +1,16 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::Client do + let(:client) { described_class.new(access_token: 'intercom-token') } + + describe '#list_contacts' do + it 'wraps transport failures in a retryable client error', :aggregate_failures do + allow(HTTParty).to receive(:get).and_raise(SocketError, 'getaddrinfo failed') + + expect { client.list_contacts }.to raise_error(DataImports::Intercom::Client::Error) do |error| + expect(error.message).to eq('Intercom API request failed before receiving a response: getaddrinfo failed') + expect(error.body).to include(transport_error_class: 'SocketError') + end + end + end +end diff --git a/spec/services/data_imports/intercom/creation_service_spec.rb b/spec/services/data_imports/intercom/creation_service_spec.rb new file mode 100644 index 000000000..4345b3715 --- /dev/null +++ b/spec/services/data_imports/intercom/creation_service_spec.rb @@ -0,0 +1,52 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::CreationService do + let(:account) { create(:account) } + let(:user) { create(:user, account: account) } + let(:validator) { instance_double(DataImports::Intercom::CredentialsValidator, perform: { 'contacts' => 12 }) } + + before do + allow(DataImports::Intercom::CredentialsValidator).to receive(:new).and_return(validator) + end + + it 'validates and creates an import with its credentials and totals', :aggregate_failures do + data_import = described_class.new( + account: account, + initiated_by: user, + source_params: { + name: 'Migration run', + source_provider: 'intercom', + access_token: ' intercom-token ', + import_types: %w[contacts] + } + ).perform + + expect(data_import).to have_attributes( + name: 'Migration run', + source_type: 'api', + source_provider: 'intercom', + import_types: %w[contacts], + access_token: 'intercom-token', + initiated_by_id: user.id + ) + expect(data_import.stats.dig('contacts', 'total')).to eq(12) + expect(data_import.active_intercom_import_run_id).to be_present + end + + it 'returns no import without validating when another import is active' do + create(:data_import, :intercom, account: account, status: :processing) + + data_import = described_class.new( + account: account, + initiated_by: user, + source_params: { + name: 'Second run', + source_provider: 'intercom', + access_token: 'intercom-token' + } + ).perform + + expect(data_import).to be_nil + expect(validator).not_to have_received(:perform) + end +end diff --git a/spec/services/data_imports/intercom/credentials_validator_spec.rb b/spec/services/data_imports/intercom/credentials_validator_spec.rb new file mode 100644 index 000000000..3d3f9698b --- /dev/null +++ b/spec/services/data_imports/intercom/credentials_validator_spec.rb @@ -0,0 +1,54 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::CredentialsValidator do + let(:client) { instance_double(DataImports::Intercom::Client) } + + before do + allow(DataImports::Intercom::Client).to receive(:new).with(access_token: 'intercom-token').and_return(client) + allow(client).to receive(:list_contacts) + allow(client).to receive(:list_conversations) + end + + it 'validates and counts only contacts when conversations are not selected' do + allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 42) + + totals = described_class.new(access_token: ' intercom-token ', import_types: %w[contacts]).perform + + expect(totals).to eq('contacts' => 42) + expect(client).not_to have_received(:list_conversations) + end + + it 'validates contact access and counts only conversations when contacts are not selected' do + allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 42) + allow(client).to receive(:list_conversations).with(per_page: 1).and_return('total_count' => 17) + + totals = described_class.new(access_token: 'intercom-token', import_types: %w[conversations]).perform + + expect(totals).to eq('conversations' => 17) + expect(client).to have_received(:list_contacts).with(per_page: 1) + end + + it 'keeps an undiscovered total absent' do + allow(client).to receive(:list_contacts).with(per_page: 1).and_return('data' => []) + + totals = described_class.new(access_token: 'intercom-token', import_types: %w[contacts]).perform + + expect(totals).to be_empty + end + + it 'preserves a known zero total' do + allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 0) + + totals = described_class.new(access_token: 'intercom-token', import_types: %w[contacts]).perform + + expect(totals).to eq('contacts' => 0) + end + + it 'rejects an empty access key before calling Intercom' do + expect do + described_class.new(access_token: '', import_types: %w[contacts]).perform + end.to raise_error(ArgumentError, 'Intercom access key is required.') + + expect(DataImports::Intercom::Client).not_to have_received(:new) + end +end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb new file mode 100644 index 000000000..67c2ec576 --- /dev/null +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -0,0 +1,968 @@ +require 'rails_helper' + +RSpec.describe DataImports::Intercom::Importer do + let(:account) { create(:account) } + let(:data_import) do + create( + :data_import, :intercom, + account: account + ) + end + let(:client) { instance_double(DataImports::Intercom::Client) } + let(:contact_payload) do + { + 'id' => 'contact_1', + 'external_id' => 'external_1', + 'email' => 'CUSTOMER@Example.com', + 'phone' => '15551234567', + 'name' => 'Customer One', + 'created_at' => 1_700_000_000, + 'updated_at' => 1_700_000_100 + } + end + let(:conversation_payload) do + { + 'id' => 'conversation_1', + 'created_at' => 1_700_000_000, + 'updated_at' => 1_700_000_200, + 'state' => 'closed', + 'open' => false, + 'admin_assignee_id' => 123, + 'team_assignee_id' => 456, + 'contacts' => { 'contacts' => [{ 'id' => 'contact_1' }] }, + 'source' => { + 'id' => 'source_1', + 'type' => 'email', + 'delivered_as' => 'customer_initiated', + 'subject' => 'Need help', + 'body' => '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