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/Gemfile.lock b/Gemfile.lock index bd41474a3..68674155e 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1030,7 +1030,7 @@ GEM addressable (>= 2.8.0) crack (>= 0.3.2) hashdiff (>= 0.4.0, < 2.0.0) - websocket-driver (0.7.7) + websocket-driver (0.8.2) base64 websocket-extensions (>= 0.1.0) websocket-extensions (0.1.5) diff --git a/app/controllers/api/v1/accounts/callbacks_controller.rb b/app/controllers/api/v1/accounts/callbacks_controller.rb index 90cdf2418..08c0ffe43 100644 --- a/app/controllers/api/v1/accounts/callbacks_controller.rb +++ b/app/controllers/api/v1/accounts/callbacks_controller.rb @@ -6,6 +6,7 @@ class Api::V1::Accounts::CallbacksController < Api::V1::Accounts::BaseController page_access_token = params[:page_access_token] page_id = params[:page_id] inbox_name = params[:inbox_name] + ActiveRecord::Base.transaction do facebook_channel = Current.account.facebook_pages.create!( page_id: page_id, user_access_token: user_access_token, @@ -15,6 +16,8 @@ class Api::V1::Accounts::CallbacksController < Api::V1::Accounts::BaseController set_instagram_id(page_access_token, facebook_channel) set_avatar(@facebook_inbox, page_id) end + rescue CustomExceptions::Inbox::LimitExceeded => e + render_error_response(e) rescue StandardError => e ChatwootExceptionTracker.new(e).capture_exception Rails.logger.error "Error in register_facebook_page: #{e.message}" diff --git a/app/controllers/api/v1/accounts/channels/twilio_channels_controller.rb b/app/controllers/api/v1/accounts/channels/twilio_channels_controller.rb index f3b14d49f..1691b5489 100644 --- a/app/controllers/api/v1/accounts/channels/twilio_channels_controller.rb +++ b/app/controllers/api/v1/accounts/channels/twilio_channels_controller.rb @@ -6,6 +6,8 @@ class Api::V1::Accounts::Channels::TwilioChannelsController < Api::V1::Accounts: def create process_create + rescue CustomExceptions::Inbox::LimitExceeded => e + render_error_response(e) rescue StandardError => e render_could_not_create_error(e.message) end diff --git a/app/controllers/api/v1/accounts/conversations/participants_controller.rb b/app/controllers/api/v1/accounts/conversations/participants_controller.rb index ebd02380f..7569142b2 100644 --- a/app/controllers/api/v1/accounts/conversations/participants_controller.rb +++ b/app/controllers/api/v1/accounts/conversations/participants_controller.rb @@ -1,27 +1,40 @@ class Api::V1::Accounts::Conversations::ParticipantsController < Api::V1::Accounts::Conversations::BaseController + include Events::Types + def show @participants = @conversation.conversation_participants end def create + participant_ids_to_add = participants_to_be_added_ids + ActiveRecord::Base.transaction do - @participants = participants_to_be_added_ids.map { |user_id| @conversation.conversation_participants.find_or_create_by(user_id: user_id) } + @participants = participant_ids_to_add.map { |user_id| @conversation.conversation_participants.find_or_create_by(user_id: user_id) } end + notify_unread_count_change if participant_ids_to_add.any? end def update + participant_ids_to_add = participants_to_be_added_ids + participant_ids_to_remove = participants_to_be_removed_ids + changed_participant_ids = participant_ids_to_add + participant_ids_to_remove + ActiveRecord::Base.transaction do - participants_to_be_added_ids.each { |user_id| @conversation.conversation_participants.find_or_create_by(user_id: user_id) } - participants_to_be_removed_ids.each { |user_id| @conversation.conversation_participants.find_by(user_id: user_id)&.destroy } + participant_ids_to_add.each { |user_id| @conversation.conversation_participants.find_or_create_by(user_id: user_id) } + participant_ids_to_remove.each { |user_id| @conversation.conversation_participants.find_by(user_id: user_id)&.destroy } end + notify_unread_count_change if changed_participant_ids.any? @participants = @conversation.conversation_participants render action: 'show' end def destroy + participant_ids_to_remove = current_participant_ids & params[:user_ids] + ActiveRecord::Base.transaction do params[:user_ids].map { |user_id| @conversation.conversation_participants.find_by(user_id: user_id)&.destroy } end + notify_unread_count_change if participant_ids_to_remove.any? head :ok end @@ -38,4 +51,11 @@ class Api::V1::Accounts::Conversations::ParticipantsController < Api::V1::Accoun def current_participant_ids @current_participant_ids ||= @conversation.conversation_participants.pluck(:user_id) end + + def notify_unread_count_change + return unless Current.account.feature_enabled?('conversation_unread_counts') + return unless Current.account.feature_enabled?('unread_count_for_filters') + + Rails.configuration.dispatcher.dispatch(CONVERSATION_UNREAD_COUNT_CHANGED, Time.zone.now, conversation: @conversation) + end end diff --git a/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb b/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb index d9f15613b..0b2335475 100644 --- a/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb +++ b/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb @@ -2,12 +2,28 @@ class Api::V1::Accounts::Conversations::UnreadCountsController < Api::V1::Accoun before_action :ensure_unread_counts_enabled def index - counts = ::Conversations::UnreadCounts::Counter.new(account: Current.account, user: Current.user).perform + counts = if filtered_unread_counts_enabled? + instrumentation.summarize_request(account_id: Current.account.id) { unread_counts } + else + unread_counts + end render json: { payload: counts } end private + def unread_counts + ::Conversations::UnreadCounts::Counter.new(account: Current.account, user: Current.user).perform + end + + def filtered_unread_counts_enabled? + Current.account.feature_enabled?(::Conversations::UnreadCounts::FilteredCounter::FEATURE_FLAG) + end + + def instrumentation + ::Conversations::UnreadCounts::FilteredCountInstrumentation + end + def ensure_unread_counts_enabled return if Current.account.feature_enabled?('conversation_unread_counts') diff --git a/app/controllers/api/v1/accounts/conversations_controller.rb b/app/controllers/api/v1/accounts/conversations_controller.rb index 2e53fa7c9..38e8d94aa 100644 --- a/app/controllers/api/v1/accounts/conversations_controller.rb +++ b/app/controllers/api/v1/accounts/conversations_controller.rb @@ -164,6 +164,7 @@ class Api::V1::Accounts::ConversationsController < Api::V1::Accounts::BaseContro # rubocop:enable Rails/SkipsModelValidations ::Conversations::UnreadCounts::Notifier.new(@conversation).perform + ::Conversations::UnreadCounts::FilteredCountInvalidator.new(Current.account).conversation_changed! end def should_update_last_seen? diff --git a/app/controllers/api/v1/accounts/dashboard_apps_controller.rb b/app/controllers/api/v1/accounts/dashboard_apps_controller.rb index a8d7ebcb9..4226db1cc 100644 --- a/app/controllers/api/v1/accounts/dashboard_apps_controller.rb +++ b/app/controllers/api/v1/accounts/dashboard_apps_controller.rb @@ -1,4 +1,5 @@ class Api::V1::Accounts::DashboardAppsController < Api::V1::Accounts::BaseController + before_action :check_authorization before_action :fetch_dashboard_apps, except: [:create] before_action :fetch_dashboard_app, only: [:show, :update, :destroy] 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 757af9b62..bb9ca2a70 100644 --- a/app/controllers/api/v1/accounts/inboxes_controller.rb +++ b/app/controllers/api/v1/accounts/inboxes_controller.rb @@ -2,7 +2,6 @@ class Api::V1::Accounts::InboxesController < Api::V1::Accounts::BaseController include Api::V1::InboxesHelper before_action :fetch_inbox, except: [:index, :create] before_action :fetch_agent_bot, only: [:set_agent_bot] - before_action :validate_limit, only: [:create] # we are already handling the authorization in fetch inbox before_action :check_authorization, except: [:show] @@ -125,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/accounts/microsoft/authorizations_controller.rb b/app/controllers/api/v1/accounts/microsoft/authorizations_controller.rb index c65a3031d..07decfde0 100644 --- a/app/controllers/api/v1/accounts/microsoft/authorizations_controller.rb +++ b/app/controllers/api/v1/accounts/microsoft/authorizations_controller.rb @@ -6,7 +6,10 @@ class Api::V1::Accounts::Microsoft::AuthorizationsController < Api::V1::Accounts { redirect_uri: "#{base_url}/microsoft/callback", scope: scope, - state: state + state: state, + # Force the Microsoft account picker so an already-signed-in account does not + # silently authorize and re-bind to an existing inbox in the new-inbox flow. + prompt: 'select_account' } ) if redirect_url diff --git a/app/controllers/api/v1/accounts/whatsapp/authorizations_controller.rb b/app/controllers/api/v1/accounts/whatsapp/authorizations_controller.rb index d52f396fc..db94113d9 100644 --- a/app/controllers/api/v1/accounts/whatsapp/authorizations_controller.rb +++ b/app/controllers/api/v1/accounts/whatsapp/authorizations_controller.rb @@ -8,8 +8,10 @@ class Api::V1::Accounts::Whatsapp::AuthorizationsController < Api::V1::Accounts: validate_embedded_signup_params! channel = process_embedded_signup render_success_response(channel.inbox) - rescue StandardError => e + rescue CustomExceptions::Inbox::LimitExceeded => e render_error_response(e) + rescue StandardError => e + render_embedded_signup_error(e) end private @@ -55,7 +57,7 @@ class Api::V1::Accounts::Whatsapp::AuthorizationsController < Api::V1::Accounts: render json: response end - def render_error_response(error) + def render_embedded_signup_error(error) Rails.logger.error "[WHATSAPP AUTHORIZATION] Embedded signup error: #{error.message}" Rails.logger.error error.backtrace.join("\n") render json: { 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/controllers/concerns/request_exception_handler.rb b/app/controllers/concerns/request_exception_handler.rb index 7f4e313b1..43d6edf1f 100644 --- a/app/controllers/concerns/request_exception_handler.rb +++ b/app/controllers/concerns/request_exception_handler.rb @@ -9,6 +9,7 @@ module RequestExceptionHandler included do rescue_from ActiveRecord::RecordInvalid, with: :render_record_invalid + rescue_from CustomExceptions::Inbox::LimitExceeded, with: :render_error_response end private @@ -40,8 +41,8 @@ module RequestExceptionHandler render json: { error: message }, status: :not_found end - def render_could_not_create_error(message) - render json: { error: sanitized_error_message(message) }, status: :unprocessable_entity + def render_could_not_create_error(error) + render json: { error: sanitized_error_message(error) }, status: :unprocessable_entity end def render_payment_required(message) diff --git a/app/controllers/instagram/callbacks_controller.rb b/app/controllers/instagram/callbacks_controller.rb index cd317363c..e9065119f 100644 --- a/app/controllers/instagram/callbacks_controller.rb +++ b/app/controllers/instagram/callbacks_controller.rb @@ -11,6 +11,8 @@ class Instagram::CallbacksController < ApplicationController end process_successful_authorization + rescue CustomExceptions::Inbox::LimitExceeded => e + handle_limit_error(e) rescue StandardError => e handle_error(e) end @@ -47,6 +49,14 @@ class Instagram::CallbacksController < ApplicationController redirect_to_error_page(error_info) end + def handle_limit_error(error) + redirect_to_error_page( + 'error_type' => error.class.name, + 'code' => Rack::Utils.status_code(error.http_status), + 'error_message' => error.message + ) + end + # Extract error details from the exception def extract_error_info(error) if error.is_a?(OAuth2::Error) diff --git a/app/controllers/public/api/v1/inboxes/contacts_controller.rb b/app/controllers/public/api/v1/inboxes/contacts_controller.rb index 838b10951..764e02a72 100644 --- a/app/controllers/public/api/v1/inboxes/contacts_controller.rb +++ b/app/controllers/public/api/v1/inboxes/contacts_controller.rb @@ -18,7 +18,8 @@ class Public::Api::V1::Inboxes::ContactsController < Public::Api::V1::InboxesCon contact: @contact_inbox.contact, params: permitted_params.to_h.deep_symbolize_keys.except(:identifier) ) - render json: contact_identify_action.perform + contact_identify_action.perform + @contact_inbox.reload end private diff --git a/app/controllers/tiktok/callbacks_controller.rb b/app/controllers/tiktok/callbacks_controller.rb index 20c0ee9c0..a39fec5ed 100644 --- a/app/controllers/tiktok/callbacks_controller.rb +++ b/app/controllers/tiktok/callbacks_controller.rb @@ -6,6 +6,8 @@ class Tiktok::CallbacksController < ApplicationController return handle_ungranted_scopes_error unless all_scopes_granted? process_successful_authorization + rescue CustomExceptions::Inbox::LimitExceeded => e + handle_limit_error(e) rescue StandardError => e handle_error(e) end @@ -36,6 +38,14 @@ class Tiktok::CallbacksController < ApplicationController redirect_to_error_page(error_type: error.class.name, code: 500, error_message: error.message) end + def handle_limit_error(error) + redirect_to_error_page( + error_type: error.class.name, + code: Rack::Utils.status_code(error.http_status), + error_message: error.message + ) + end + # Handles the case when a user denies permissions or cancels the authorization flow def handle_authorization_error redirect_to_error_page( 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/helpers/api/v1/inboxes_helper.rb b/app/helpers/api/v1/inboxes_helper.rb index 8a10fa99c..6c64dd009 100644 --- a/app/helpers/api/v1/inboxes_helper.rb +++ b/app/helpers/api/v1/inboxes_helper.rb @@ -114,10 +114,4 @@ module Api::V1::InboxesHelper 'sms' => Current.account.sms_channels }[permitted_params[:channel][:type]] end - - def validate_limit - return unless Current.account.inboxes.count >= Current.account.usage_limits[:inboxes] - - render_payment_required('Account limit exceeded. Upgrade to a higher plan') - end end diff --git a/app/helpers/filters/filter_helper.rb b/app/helpers/filters/filter_helper.rb index 4f345676e..d32c9468a 100644 --- a/app/helpers/filters/filter_helper.rb +++ b/app/helpers/filters/filter_helper.rb @@ -68,6 +68,8 @@ module Filters::FilterHelper when 'text_case_insensitive' text_case_insensitive_filter(query_hash, filter_operator_value) else + return text_cast_filter(query_hash, filter_operator_value) if text_search_on_display_id?(query_hash) + default_filter(query_hash, filter_operator_value) end end @@ -82,10 +84,18 @@ module Filters::FilterHelper "#{filter_operator_value} #{query_hash[:query_operator]}" end + def text_cast_filter(query_hash, filter_operator_value) + "(#{filter_config[:table_name]}.#{query_hash[:attribute_key]})::text #{filter_operator_value} #{query_hash[:query_operator]}" + end + def default_filter(query_hash, filter_operator_value) "#{filter_config[:table_name]}.#{query_hash[:attribute_key]} #{filter_operator_value} #{query_hash[:query_operator]}" end + def text_search_on_display_id?(query_hash) + query_hash[:attribute_key] == 'display_id' && %w[contains does_not_contain].include?(query_hash[:filter_operator]) + end + def validate_single_condition(condition) return if condition['query_operator'].nil? return if condition['query_operator'].empty? diff --git a/app/javascript/dashboard/api/captain/assistant.js b/app/javascript/dashboard/api/captain/assistant.js index dcd92f735..1fc17798d 100644 --- a/app/javascript/dashboard/api/captain/assistant.js +++ b/app/javascript/dashboard/api/captain/assistant.js @@ -37,6 +37,20 @@ class CaptainAssistant extends ApiClient { params: { range, timezone_offset: getTimezoneOffset() }, }); } + + getDrilldown({ assistantId, metric, range, page, signal }) { + const requestConfig = { + params: { + metric, + range, + timezone_offset: getTimezoneOffset(), + page, + }, + }; + if (signal) requestConfig.signal = signal; + + return axios.get(`${this.url}/${assistantId}/drilldown`, requestConfig); + } } export default new CaptainAssistant(); 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 = () => {
+ {{ $t('DATA_IMPORTS.EXPORT.DESCRIPTION') }} +
++ {{ $t('DATA_IMPORTS.TABLE.EMPTY_DESCRIPTION') }} +
++ {{ emptyMessage }} +
+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 diff --git a/spec/services/labels/destroy_service_spec.rb b/spec/services/labels/destroy_service_spec.rb index 7d06b72d0..7f14273b3 100644 --- a/spec/services/labels/destroy_service_spec.rb +++ b/spec/services/labels/destroy_service_spec.rb @@ -6,6 +6,7 @@ describe Labels::DestroyService do let(:label) { create(:label, account: account) } let(:contact) { conversation.contact } let(:label_deleted_at) { Time.zone.parse('2026-05-07 10:00:00 UTC') } + let(:store) { Conversations::UnreadCounts::FilteredCountStore } before do conversation.label_list.add(label.title) @@ -74,6 +75,18 @@ describe Labels::DestroyService do ).perform end + it 'invalidates filtered counts when conversation label associations are removed' do + account.enable_features!(:unread_count_for_filters) + + expect do + described_class.new( + label_title: label.title, + account_id: account.id, + label_deleted_at: label_deleted_at + ).perform + end.to change { store.conversation_version(account.id) }.by(1) + end + it 'does not remove label associations created after the label was deleted' do other_conversation = create(:conversation, account: account) other_conversation.label_list.add(label.title) diff --git a/spec/services/whatsapp/channel_creation_service_spec.rb b/spec/services/whatsapp/channel_creation_service_spec.rb index 983af6c78..e7016f6a4 100644 --- a/spec/services/whatsapp/channel_creation_service_spec.rb +++ b/spec/services/whatsapp/channel_creation_service_spec.rb @@ -60,6 +60,17 @@ describe Whatsapp::ChannelCreationService do expect(inbox.name).to eq('Test Business WhatsApp') expect(inbox.account).to eq(account) end + + it 'does not leave an orphan channel when inbox creation fails' do + allow(Inbox).to receive(:create!).and_wrap_original do |method, *args| + method.call(*args) + raise ActiveRecord::RecordInvalid, Inbox.new + end + + expect do + expect { service.perform }.to raise_error(ActiveRecord::RecordInvalid) + end.not_to change(Channel::Whatsapp, :count) + end end context 'when channel already exists for the phone number' do diff --git a/spec/services/whatsapp/webhook_teardown_service_spec.rb b/spec/services/whatsapp/webhook_teardown_service_spec.rb index be94f3c44..a5bdeef0b 100644 --- a/spec/services/whatsapp/webhook_teardown_service_spec.rb +++ b/spec/services/whatsapp/webhook_teardown_service_spec.rb @@ -51,18 +51,41 @@ RSpec.describe Whatsapp::WebhookTeardownService do end end - context 'when channel is whatsapp_cloud but not embedded_signup' do + context 'when channel is whatsapp_cloud with manual setup' do before do + allow(channel).to receive(:setup_webhooks).and_return(true) + channel.update!( provider: 'whatsapp_cloud', - provider_config: { 'source' => 'manual' } + provider_config: { + 'source' => 'manual', + 'phone_number_id' => 'manual_phone_id', + 'business_account_id' => 'manual_waba_id', + 'api_key' => 'manual_api_key' + } ) end - it 'does not attempt to unsubscribe webhook' do - expect(Whatsapp::FacebookApiClient).not_to receive(:new) + it 'clears the phone number callback override' do + api_client = instance_double(Whatsapp::FacebookApiClient) + allow(Whatsapp::FacebookApiClient).to receive(:new).with('manual_api_key').and_return(api_client) + allow(api_client).to receive(:clear_phone_number_callback_override).with('manual_phone_id') service.perform + + expect(api_client).to have_received(:clear_phone_number_callback_override).with('manual_phone_id') + end + + # The manual token belongs to the customer's own Meta app, so its WABA subscription is not ours to remove. + it 'does not unsubscribe the app from the WABA' do + api_client = instance_double(Whatsapp::FacebookApiClient) + allow(Whatsapp::FacebookApiClient).to receive(:new).and_return(api_client) + allow(api_client).to receive(:clear_phone_number_callback_override) + allow(api_client).to receive(:unsubscribe_app_from_waba) + + service.perform + + expect(api_client).not_to have_received(:unsubscribe_app_from_waba) end end