feat(imports): retry stalled Intercom imports

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

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`);
   }

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."
     }

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(() => {
       <ImportDetailHeader
         :data-import="dataImport"
         :is-refreshing="isRefreshing"
+        :is-retrying="isRetrying"
         :is-abandoning="isAbandoning"
         :is-polling="isPolling"
         @refresh="fetchImport({ manual: true })"
+        @retry="retryImport"
         @abandon="abandonImport"
       />
     </template>

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')"
           />
+          <Button
+            v-if="dataImport?.stalled"
+            outline
+            slate
+            size="sm"
+            icon="i-lucide-rotate-ccw"
+            :is-loading="isRetrying"
+            :disabled="isAbandoning"
+            :label="$t('DATA_IMPORTS.TABLE.RETRY')"
+            @click="$emit('retry')"
+          />
           <Button
             v-if="canAbandonImport"
             ruby
             size="sm"
             :is-loading="isAbandoning"
+            :disabled="isRetrying"
             :label="$t('DATA_IMPORTS.TABLE.ABANDON')"
             @click="$emit('abandon')"
           />

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: `
+    <button
+      :data-label="label"
+      :data-icon="icon"
+      :data-loading="isLoading"
+      @click="$emit('click')"
+    >
+      {{ label }}
+    </button>
+  `,
+};
+
+const BaseSettingsHeaderStub = {
+  template: `
+    <section>
+      <slot name="title" />
+      <slot name="description" />
+    </section>
+  `,
+};
+
+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);
+  });
+});

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: `
+    <main>
+      <slot name="header" />
+      <slot name="body" />
+    </main>
+  `,
+};
+
+const ImportDetailHeaderStub = {
+  name: 'ImportDetailHeader',
+  emits: ['retry'],
+  template: '<button data-test="retry" @click="$emit(\'retry\')" />',
+};
+
+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();
+  });
+});

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 = 10.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

app/policies/data_import_policy.rb

@@ -19,6 +19,10 @@ class DataImportPolicy < ApplicationPolicy
     show?
   end
 
+  def retry_import?
+    show?
+  end
+
   def abandon?
     show?
   end

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

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

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

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 ten 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: 10.minutes.ago)
+        pending_import.update!(updated_at: 10.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

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: 11.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: 11.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: 11.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
       )

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: 11.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: 11.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: 11.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: 11.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