diff --git a/app/controllers/api/v1/accounts/data_imports_controller.rb b/app/controllers/api/v1/accounts/data_imports_controller.rb
index 7f0d28e81..70432bc33 100644
--- a/app/controllers/api/v1/accounts/data_imports_controller.rb
+++ b/app/controllers/api/v1/accounts/data_imports_controller.rb
@@ -4,7 +4,7 @@ class Api::V1::Accounts::DataImportsController < Api::V1::Accounts::BaseControll
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 :set_data_import, only: [:show, :start, :retry_import, :abandon, :error_logs, :skip_logs]
before_action :check_authorization
def index
@@ -59,6 +59,24 @@ class Api::V1::Accounts::DataImportsController < Api::V1::Accounts::BaseControll
render_show
end
+ def retry_import
+ retry_service = DataImports::Intercom::RetryService.new(account: Current.account, data_import: @data_import)
+ retry_result = retry_service.perform
+ @data_import = retry_service.data_import
+
+ case retry_result
+ when :enqueue
+ DataImports::Intercom::ImportJob.perform_later(@data_import, @data_import.active_intercom_import_run_id)
+ render_show
+ when :not_stalled
+ render json: { message: 'This Intercom import is no longer stalled.' }, status: :unprocessable_entity
+ when :active_import_exists
+ render json: { message: 'Another Intercom import is already in progress.' }, status: :unprocessable_entity
+ when :access_token_missing
+ render json: { message: 'The Intercom access key for this import is unavailable.' }, status: :unprocessable_entity
+ end
+ end
+
def abandon
@data_import.abandon!
render_show
diff --git a/app/javascript/dashboard/api/dataImports.js b/app/javascript/dashboard/api/dataImports.js
index b4c15b98a..e6929d420 100644
--- a/app/javascript/dashboard/api/dataImports.js
+++ b/app/javascript/dashboard/api/dataImports.js
@@ -11,6 +11,10 @@ class DataImportsAPI extends ApiClient {
return axios.post(`${this.url}/${id}/start`);
}
+ retry(id) {
+ return axios.post(`${this.url}/${id}/retry`);
+ }
+
abandon(id) {
return axios.post(`${this.url}/${id}/abandon`);
}
diff --git a/app/javascript/dashboard/i18n/locale/en/settings.json b/app/javascript/dashboard/i18n/locale/en/settings.json
index ceb0438b1..6b3b63ea9 100644
--- a/app/javascript/dashboard/i18n/locale/en/settings.json
+++ b/app/javascript/dashboard/i18n/locale/en/settings.json
@@ -461,6 +461,7 @@
"STATUS": "Status",
"IMPORTED": "Imported",
"CREATED": "Created",
+ "RETRY": "Retry",
"ABANDON": "Abandon"
},
"DETAIL": {
@@ -507,6 +508,8 @@
},
"ALERTS": {
"IMPORT_STARTED": "Intercom import has started.",
+ "IMPORT_RETRIED": "Intercom import has been queued to resume.",
+ "IMPORT_RETRY_FAILED": "Could not retry the Intercom import.",
"IMPORT_ABANDONED": "Intercom import has been abandoned.",
"IMPORT_FAILED": "Could not start the Intercom import."
}
diff --git a/app/javascript/dashboard/routes/dashboard/settings/data/Show.vue b/app/javascript/dashboard/routes/dashboard/settings/data/Show.vue
index ed89c66a9..2bab87f75 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/data/Show.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/data/Show.vue
@@ -26,6 +26,7 @@ const dataImport = ref(null);
const isLoading = ref(true);
const isRefreshing = ref(false);
const isPolling = ref(false);
+const isRetrying = ref(false);
const isAbandoning = ref(false);
const isDownloadingErrorLogs = ref(false);
const isDownloadingSkipLogs = ref(false);
@@ -117,6 +118,19 @@ const abandonImport = async () => {
}
};
+const retryImport = async () => {
+ isRetrying.value = true;
+ try {
+ const response = await DataImportsAPI.retry(dataImport.value.id);
+ dataImport.value = response.data;
+ useAlert(t('DATA_IMPORTS.ALERTS.IMPORT_RETRIED'));
+ } catch {
+ useAlert(t('DATA_IMPORTS.ALERTS.IMPORT_RETRY_FAILED'));
+ } finally {
+ isRetrying.value = false;
+ }
+};
+
const downloadCsv = (response, filename) => {
const url = window.URL.createObjectURL(
new Blob([response.data], { type: 'text/csv' })
@@ -197,9 +211,11 @@ onBeforeUnmount(() => {
diff --git a/app/javascript/dashboard/routes/dashboard/settings/data/components/ImportDetailHeader.vue b/app/javascript/dashboard/routes/dashboard/settings/data/components/ImportDetailHeader.vue
index 935d63b87..66324a384 100644
--- a/app/javascript/dashboard/routes/dashboard/settings/data/components/ImportDetailHeader.vue
+++ b/app/javascript/dashboard/routes/dashboard/settings/data/components/ImportDetailHeader.vue
@@ -21,6 +21,10 @@ const props = defineProps({
type: Boolean,
default: false,
},
+ isRetrying: {
+ type: Boolean,
+ default: false,
+ },
isAbandoning: {
type: Boolean,
default: false,
@@ -31,7 +35,7 @@ const props = defineProps({
},
});
-defineEmits(['refresh', 'abandon']);
+defineEmits(['refresh', 'retry', 'abandon']);
const { t } = useI18n();
@@ -90,11 +94,23 @@ const canAbandonImport = computed(() => isAbandonableImport(props.dataImport));
:title="$t('DATA_IMPORTS.MONITOR.REFRESH')"
@click="$emit('refresh')"
/>
+
diff --git a/app/javascript/dashboard/routes/dashboard/settings/data/specs/ImportDetailHeader.spec.js b/app/javascript/dashboard/routes/dashboard/settings/data/specs/ImportDetailHeader.spec.js
new file mode 100644
index 000000000..671f58fae
--- /dev/null
+++ b/app/javascript/dashboard/routes/dashboard/settings/data/specs/ImportDetailHeader.spec.js
@@ -0,0 +1,85 @@
+import { mount } from '@vue/test-utils';
+import ImportDetailHeader from '../components/ImportDetailHeader.vue';
+
+vi.mock('vue-i18n', () => ({
+ useI18n: () => ({ t: key => key }),
+}));
+
+const ButtonStub = {
+ name: 'Button',
+ props: {
+ label: { type: String, default: '' },
+ icon: { type: String, default: '' },
+ isLoading: { type: Boolean, default: false },
+ },
+ emits: ['click'],
+ template: `
+
+ `,
+};
+
+const BaseSettingsHeaderStub = {
+ template: `
+
+ `,
+};
+
+const mountHeader = props =>
+ mount(ImportDetailHeader, {
+ props,
+ global: {
+ stubs: {
+ Button: ButtonStub,
+ BaseSettingsHeader: BaseSettingsHeaderStub,
+ },
+ mocks: {
+ $t: key => key,
+ },
+ },
+ });
+
+describe('ImportDetailHeader', () => {
+ const activeImport = {
+ id: 1,
+ name: 'Intercom import',
+ data_type: 'intercom',
+ source_provider: 'intercom',
+ status: 'processing',
+ stalled: true,
+ };
+
+ it('shows Retry between Refresh and Abandon for stalled imports', async () => {
+ const wrapper = mountHeader({ dataImport: activeImport });
+ const buttons = wrapper.findAll('button');
+
+ expect(buttons.map(button => button.attributes('data-label'))).toEqual([
+ '',
+ 'DATA_IMPORTS.TABLE.RETRY',
+ 'DATA_IMPORTS.TABLE.ABANDON',
+ ]);
+
+ await buttons[1].trigger('click');
+
+ expect(wrapper.emitted('retry')).toHaveLength(1);
+ });
+
+ it('hides Retry when the server does not report the import as stalled', () => {
+ const wrapper = mountHeader({
+ dataImport: { ...activeImport, stalled: false },
+ });
+
+ expect(
+ wrapper.find('[data-label="DATA_IMPORTS.TABLE.RETRY"]').exists()
+ ).toBe(false);
+ });
+});
diff --git a/app/javascript/dashboard/routes/dashboard/settings/data/specs/showActions.spec.js b/app/javascript/dashboard/routes/dashboard/settings/data/specs/showActions.spec.js
new file mode 100644
index 000000000..eb7b83393
--- /dev/null
+++ b/app/javascript/dashboard/routes/dashboard/settings/data/specs/showActions.spec.js
@@ -0,0 +1,104 @@
+import { flushPromises, mount } from '@vue/test-utils';
+import { KeepAlive, defineComponent, h, nextTick } from 'vue';
+import { useAlert } from 'dashboard/composables';
+import DataImportsAPI from 'dashboard/api/dataImports';
+import Show from '../Show.vue';
+
+vi.mock('dashboard/api/dataImports', () => ({
+ default: {
+ show: vi.fn(),
+ retry: vi.fn(),
+ },
+}));
+
+vi.mock('dashboard/composables', () => ({
+ useAlert: vi.fn(),
+}));
+
+vi.mock('vue-i18n', () => ({
+ useI18n: () => ({ t: key => key }),
+}));
+
+vi.mock('vue-router', async importOriginal => ({
+ ...(await importOriginal()),
+ useRoute: () => ({ params: { dataImportId: 1 } }),
+}));
+
+const SettingsLayoutStub = {
+ template: `
+
+
+
+
+ `,
+};
+
+const ImportDetailHeaderStub = {
+ name: 'ImportDetailHeader',
+ emits: ['retry'],
+ template: '',
+};
+
+const mountShow = () => {
+ const Host = defineComponent({
+ render() {
+ return h(KeepAlive, null, { default: () => h(Show) });
+ },
+ });
+
+ return mount(Host, {
+ global: {
+ stubs: {
+ SettingsLayout: SettingsLayoutStub,
+ ImportDetailHeader: ImportDetailHeaderStub,
+ ImportSummaryTiles: true,
+ ImportProgress: true,
+ ImportErrorsSection: true,
+ ImportSkipLogsSection: true,
+ },
+ mocks: {
+ $t: key => key,
+ },
+ },
+ });
+};
+
+describe('data import detail actions', () => {
+ beforeEach(() => {
+ vi.useFakeTimers();
+ DataImportsAPI.show.mockResolvedValue({
+ data: {
+ id: 1,
+ status: 'processing',
+ stalled: true,
+ skip_logs_filters: {},
+ },
+ });
+ });
+
+ afterEach(() => {
+ vi.useRealTimers();
+ vi.clearAllMocks();
+ });
+
+ it('retries a stalled import and replaces the page state', async () => {
+ DataImportsAPI.retry.mockResolvedValue({
+ data: {
+ id: 1,
+ status: 'pending',
+ stalled: false,
+ skip_logs_filters: {},
+ },
+ });
+ const wrapper = mountShow();
+ await nextTick();
+ await flushPromises();
+
+ await wrapper.find('[data-test="retry"]').trigger('click');
+ await flushPromises();
+
+ expect(DataImportsAPI.retry).toHaveBeenCalledWith(1);
+ expect(useAlert).toHaveBeenCalledWith('DATA_IMPORTS.ALERTS.IMPORT_RETRIED');
+ wrapper.unmount();
+ });
+});
diff --git a/app/models/data_import.rb b/app/models/data_import.rb
index 5f0ad4737..c78c3ce0b 100644
--- a/app/models/data_import.rb
+++ b/app/models/data_import.rb
@@ -33,6 +33,7 @@
#
class DataImport < ApplicationRecord
ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY = 'active_intercom_import_run_id'.freeze
+ INTERCOM_STALLED_AFTER = 15.minutes
LEGACY_DATA_TYPES = ['contacts'].freeze
INTEGRATION_DATA_TYPES = ['intercom'].freeze
IMPORT_TYPES = %w[contacts conversations].freeze
@@ -71,6 +72,10 @@ class DataImport < ApplicationRecord
failed? || abandoned?
end
+ def stalled?
+ intercom_import? && (pending? || processing?) && updated_at <= INTERCOM_STALLED_AFTER.ago
+ end
+
def abandonable?
intercom_import? && (pending? || processing?)
end
diff --git a/app/policies/data_import_policy.rb b/app/policies/data_import_policy.rb
index 3f908ca5d..58bd57722 100644
--- a/app/policies/data_import_policy.rb
+++ b/app/policies/data_import_policy.rb
@@ -19,6 +19,10 @@ class DataImportPolicy < ApplicationPolicy
show?
end
+ def retry_import?
+ show?
+ end
+
def abandon?
show?
end
diff --git a/app/services/data_imports/intercom/retry_service.rb b/app/services/data_imports/intercom/retry_service.rb
new file mode 100644
index 000000000..6d0cb5c9d
--- /dev/null
+++ b/app/services/data_imports/intercom/retry_service.rb
@@ -0,0 +1,27 @@
+class DataImports::Intercom::RetryService
+ attr_reader :data_import
+
+ def initialize(account:, data_import:)
+ @account = account
+ @data_import = data_import
+ end
+
+ def perform
+ @account.with_lock do
+ @data_import.reload
+ next :not_stalled unless @data_import.stalled?
+ next :active_import_exists if another_active_import?
+ next :access_token_missing if @data_import.access_token.blank?
+
+ @data_import.assign_active_intercom_import_run_id
+ @data_import.update!(status: :pending)
+ :enqueue
+ end
+ end
+
+ private
+
+ def another_active_import?
+ @account.data_imports.active_intercom.where.not(id: @data_import.id).exists?
+ end
+end
diff --git a/app/views/api/v1/accounts/data_imports/_data_import.json.jbuilder b/app/views/api/v1/accounts/data_imports/_data_import.json.jbuilder
index ba56ac4d0..46c224fb1 100644
--- a/app/views/api/v1/accounts/data_imports/_data_import.json.jbuilder
+++ b/app/views/api/v1/accounts/data_imports/_data_import.json.jbuilder
@@ -5,6 +5,7 @@ json.source_type data_import.source_type
json.source_provider data_import.source_provider
json.import_types data_import.import_types
json.status data_import.status
+json.stalled data_import.stalled?
json.total_records data_import.total_records
json.processed_records data_import.processed_records
json.stats data_import.stats
diff --git a/config/routes.rb b/config/routes.rb
index 1481c14c7..51349f42b 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -228,6 +228,7 @@ Rails.application.routes.draw do
end
member do
post :start
+ post :retry, action: :retry_import
post :abandon
get :error_logs
get :skip_logs
diff --git a/spec/models/data_import_spec.rb b/spec/models/data_import_spec.rb
index fc31ebb1d..244d9f8d4 100644
--- a/spec/models/data_import_spec.rb
+++ b/spec/models/data_import_spec.rb
@@ -64,4 +64,36 @@ RSpec.describe DataImport do
expect(data_import.abandoned_at).to be_nil
end
end
+
+ describe '#stalled?' do
+ let(:account) { create(:account) }
+
+ it 'identifies active Intercom imports without updates for fifteen minutes', :aggregate_failures do
+ freeze_time do
+ processing_import = create(:data_import, :intercom, account: account, status: :processing)
+ pending_import = create(:data_import, :intercom, account: account, status: :pending)
+ processing_import.update!(updated_at: 15.minutes.ago)
+ pending_import.update!(updated_at: 15.minutes.ago)
+
+ expect(processing_import.reload).to be_stalled
+ expect(pending_import.reload).to be_stalled
+ end
+ end
+
+ it 'does not identify recent or terminal Intercom imports as stalled', :aggregate_failures do
+ recent_import = create(:data_import, :intercom, account: account, status: :processing)
+ completed_import = create(:data_import, :intercom, account: account, status: :completed)
+ completed_import.update!(updated_at: 1.hour.ago)
+
+ expect(recent_import).not_to be_stalled
+ expect(completed_import.reload).not_to be_stalled
+ end
+
+ it 'does not identify legacy imports as stalled' do
+ legacy_import = create(:data_import, account: account, status: :processing)
+ legacy_import.update!(updated_at: 1.hour.ago)
+
+ expect(legacy_import.reload).not_to be_stalled
+ end
+ end
end
diff --git a/spec/requests/api/v1/accounts/data_imports_spec.rb b/spec/requests/api/v1/accounts/data_imports_spec.rb
index 82a298ef7..4ded7ff4a 100644
--- a/spec/requests/api/v1/accounts/data_imports_spec.rb
+++ b/spec/requests/api/v1/accounts/data_imports_spec.rb
@@ -209,6 +209,63 @@ RSpec.describe 'Data Imports API', type: :request do
end
end
+ describe 'POST /api/v1/accounts/:account_id/data_imports/:id/retry' do
+ let(:data_import) { create(:data_import, :intercom, account: account, status: :processing, started_at: 2.hours.ago) }
+
+ it 'resumes a stalled import with a new run identifier while preserving progress', :aggregate_failures do
+ started_at = data_import.started_at
+ data_import.update!(
+ cursor: { 'conversations' => { 'starting_after' => 'cursor-1' } },
+ stats: { 'conversations' => { 'imported' => 20 } },
+ source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'previous-run' },
+ updated_at: 16.minutes.ago
+ )
+
+ expect do
+ post retry_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+ headers: admin.create_new_auth_token,
+ as: :json
+ end.to have_enqueued_job(DataImports::Intercom::ImportJob).with(data_import, a_kind_of(String))
+
+ expect(response).to have_http_status(:ok)
+ expect(response.parsed_body).to include('status' => 'pending', 'stalled' => false)
+ expect(data_import.reload.started_at).to eq(started_at)
+ expect(data_import.cursor.dig('conversations', 'starting_after')).to eq('cursor-1')
+ expect(data_import.stats.dig('conversations', 'imported')).to eq(20)
+ expect(data_import.active_intercom_import_run_id).not_to eq('previous-run')
+ end
+
+ it 'rejects duplicate retries after the import becomes active again' do
+ data_import.update!(updated_at: 16.minutes.ago)
+ post retry_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+ headers: admin.create_new_auth_token,
+ as: :json
+
+ expect do
+ post retry_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+ headers: admin.create_new_auth_token,
+ as: :json
+ end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+ expect(response).to have_http_status(:unprocessable_entity)
+ expect(response.parsed_body['message']).to eq('This Intercom import is no longer stalled.')
+ end
+
+ it 'rejects retry while another Intercom import is active' do
+ data_import.update!(updated_at: 16.minutes.ago)
+ create(:data_import, :intercom, account: account, status: :processing)
+
+ expect do
+ post retry_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+ headers: admin.create_new_auth_token,
+ as: :json
+ end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+ expect(response).to have_http_status(:unprocessable_entity)
+ expect(response.parsed_body['message']).to eq('Another Intercom import is already in progress.')
+ end
+ end
+
describe 'POST /api/v1/accounts/:account_id/data_imports/:id/abandon' do
let(:data_import) { create(:data_import, :intercom, account: account) }
@@ -283,6 +340,7 @@ RSpec.describe 'Data Imports API', type: :request do
'id' => data_import.id,
'name' => 'July Intercom migration',
'source_provider' => 'intercom',
+ 'stalled' => false,
'import_errors_count' => 1,
'skip_logs_count' => 1
)
diff --git a/spec/services/data_imports/intercom/retry_service_spec.rb b/spec/services/data_imports/intercom/retry_service_spec.rb
new file mode 100644
index 000000000..f94283993
--- /dev/null
+++ b/spec/services/data_imports/intercom/retry_service_spec.rb
@@ -0,0 +1,69 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::RetryService do
+ let(:account) { create(:account) }
+ let(:data_import) { create(:data_import, :intercom, account: account, status: :processing, started_at: 2.hours.ago) }
+
+ before do
+ account.enable_features!('data_import')
+ end
+
+ it 'prepares a stalled import for another run without clearing progress', :aggregate_failures do
+ original_cursor = { 'contacts' => { 'completed' => true }, 'conversations' => { 'starting_after' => 'cursor-1' } }
+ original_stats = { 'contacts' => { 'imported' => 10 }, 'conversations' => { 'imported' => 5 } }
+ started_at = data_import.started_at
+ error = data_import.import_errors.create!(error_code: 'MessageFailed', message: 'Timed out')
+ data_import.update!(
+ cursor: original_cursor,
+ stats: original_stats,
+ source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'previous-run' },
+ updated_at: 16.minutes.ago
+ )
+
+ result = described_class.new(account: account, data_import: data_import).perform
+
+ expect(result).to eq(:enqueue)
+ expect(data_import.reload).to be_pending
+ expect(data_import.started_at).to eq(started_at)
+ expect(data_import.cursor).to eq(original_cursor)
+ expect(data_import.stats).to eq(original_stats)
+ expect(data_import.import_errors).to contain_exactly(error)
+ expect(data_import.active_intercom_import_run_id).not_to eq('previous-run')
+ end
+
+ it 'does not retry an import that is still receiving updates' do
+ result = described_class.new(account: account, data_import: data_import).perform
+
+ expect(result).to eq(:not_stalled)
+ expect(data_import.reload).to be_processing
+ end
+
+ it 'does not retry while another Intercom import is active' do
+ data_import.update!(updated_at: 16.minutes.ago)
+ create(:data_import, :intercom, account: account, status: :processing)
+
+ result = described_class.new(account: account, data_import: data_import).perform
+
+ expect(result).to eq(:active_import_exists)
+ expect(data_import.reload).to be_processing
+ end
+
+ it 'allows an active legacy import to continue alongside the retry' do
+ data_import.update!(updated_at: 16.minutes.ago)
+ create(:data_import, account: account, status: :processing)
+
+ result = described_class.new(account: account, data_import: data_import).perform
+
+ expect(result).to eq(:enqueue)
+ expect(data_import.reload).to be_pending
+ end
+
+ it 'does not retry when the stored access key is unavailable' do
+ data_import.update!(access_token: nil, updated_at: 16.minutes.ago)
+
+ result = described_class.new(account: account, data_import: data_import).perform
+
+ expect(result).to eq(:access_token_missing)
+ expect(data_import.reload).to be_processing
+ end
+end