Merge branch 'codex/cw-7431-document-usage' into codex/cw-7498-document-conversation-usage

Merge remote-tracking branch 'origin/develop' into codex/cw-7431-document-usage
Merge branch 'release/4.16.1' into develop
2026-07-23 14:04:03 +05:30 · 2026-07-23 14:03:56 +05:30 · 2026-07-23 13:58:16 +05:30 · 2026-07-23 13:57:15 +05:30 · 2026-07-23 12:56:21 +05:30 · 2026-07-21 12:07:31 +05:30
41 changed files with 303 additions and 1469 deletions
@@ -1 +1 @@
-4.16.0
+4.16.1
@@ -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, :retry_import, :abandon, :error_logs, :skip_logs]
+  before_action :set_data_import, only: [:show, :start, :abandon, :error_logs, :skip_logs]
  before_action :check_authorization

  def index
@@ -59,24 +59,6 @@ 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
@@ -11,10 +11,6 @@ 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`);
  }
@@ -70,6 +70,10 @@ const props = defineProps({
    type: Number,
    default: 0,
  },
+  usedInConversationsCount: {
+    type: Number,
+    default: 0,
+  },
  isSelected: {
    type: Boolean,
    default: false,
@@ -150,6 +154,11 @@ const createdAtLabel = computed(() => dynamicTime(props.createdAt));
 const responsesCountLabel = computed(() =>
  t('CAPTAIN.DOCUMENTS.FAQ_COUNT', { n: props.responsesCount })
 );
+const usedInConversationsCountLabel = computed(() =>
+  t('CAPTAIN.DOCUMENTS.USED_IN_CONVERSATIONS', {
+    n: props.usedInConversationsCount,
+  })
+);

 const displayLink = computed(() =>
  isPdf.value
@@ -216,7 +225,7 @@ const handleRetry = () => {
        </div>
      </div>
    </div>
-    <div class="flex gap-4 justify-between items-center w-full">
+    <div class="flex flex-wrap gap-4 justify-between items-center w-full">
      <span
        class="flex gap-1 items-center text-sm truncate shrink-0 text-n-slate-11"
      >
@@ -246,6 +255,9 @@ const handleRetry = () => {
      <span class="text-sm shrink-0 text-n-slate-11">
        {{ responsesCountLabel }}
      </span>
+      <span class="text-sm shrink-0 text-n-slate-11">
+        {{ usedInConversationsCountLabel }}
+      </span>
      <DocumentSyncStatus
        v-if="showSyncStatus"
        :status="syncStatus"
@@ -33,9 +33,7 @@ import {
 // constants
 import { BUS_EVENTS } from 'shared/constants/busEvents';
 import { REPLY_POLICY } from 'shared/constants/links';
-import wootConstants, {
-  META_RESTRICTION_STATUS_URL,
-} from 'dashboard/constants/globals';
+import wootConstants from 'dashboard/constants/globals';
 import { LOCAL_STORAGE_KEYS } from 'dashboard/constants/localStorage';
 import { INBOX_TYPES } from 'dashboard/helper/inbox';

@@ -95,7 +93,6 @@ export default {
      currentUserId: 'getCurrentUserID',
      listLoadingStatus: 'getAllMessagesLoaded',
      currentAccountId: 'getCurrentAccountId',
-      isOnChatwootCloud: 'globalConfig/isOnChatwootCloud',
    }),
    isOpen() {
      return this.currentChat?.status === wootConstants.STATUS_TYPE.OPEN;
@@ -173,13 +170,6 @@ export default {
        instagramInbox
      );
    },
-    isInstagramRestrictionBannerVisible() {
-      return this.isOnChatwootCloud && this.isAnInstagramChannel;
-    },
-    instagramRestrictionStatusUrl() {
-      return META_RESTRICTION_STATUS_URL;
-    },
-
    replyWindowBannerMessage() {
      if (this.isAWhatsAppChannel) {
        return this.$t('CONVERSATION.TWILIO_WHATSAPP_CAN_REPLY');
@@ -464,15 +454,7 @@ export default {
  >
    <div ref="topBannerRef">
      <Banner
-        v-if="isInstagramRestrictionBannerVisible"
-        color-scheme="warning"
-        class="mx-2 mt-2 overflow-hidden rounded-lg"
-        :banner-message="$t('CONVERSATION.INSTAGRAM_RESTRICTION_BANNER')"
-        :href-link="instagramRestrictionStatusUrl"
-        :href-link-text="$t('CONVERSATION.INSTAGRAM_RESTRICTION_STATUS_LINK')"
-      />
-      <Banner
-        v-else-if="!currentChat.can_reply"
+        v-if="!currentChat.can_reply"
        color-scheme="alert"
        class="mx-2 mt-2 overflow-hidden rounded-lg"
        :banner-message="replyWindowBannerMessage"
@@ -78,5 +78,3 @@ export default {
  },
 };
 export const DEFAULT_REDIRECT_URL = '/app/';
-export const META_RESTRICTION_STATUS_URL =
-  'https://status.chatwoot.com/incident/948346';
@@ -44,8 +44,6 @@
    "TWILIO_WHATSAPP_CAN_REPLY": "You can only reply to this conversation using a template message due to",
    "TWILIO_WHATSAPP_24_HOURS_WINDOW": "24 hour message window restriction",
    "OLD_INSTAGRAM_INBOX_REPLY_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. All new messages will show up there. You won’t be able to send messages from this conversation anymore.",
-    "INSTAGRAM_RESTRICTION_BANNER": "Instagram is currently restricted. Some messages or actions may be delayed or unavailable while we restore full support.",
-    "INSTAGRAM_RESTRICTION_STATUS_LINK": "View status update",
    "REPLYING_TO": "You are replying to:",
    "REMOVE_SELECTION": "Remove Selection",
    "DOWNLOAD": "Download",
@@ -58,9 +58,7 @@
        "ERROR_MESSAGE": "There was an error connecting to Instagram, please try again",
        "ERROR_AUTH": "There was an error connecting to Instagram, please try again",
        "NEW_INBOX_SUGGESTION": "This Instagram account was previously linked to a different inbox and has now been migrated here. All new messages will appear here. The old inbox will no longer be able to send or receive messages for this account.",
-        "DUPLICATE_INBOX_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. You won’t be able to send/receive Instagram messages from this inbox anymore.",
-        "SETTINGS_RESTRICTED_WARNING": "Instagram is currently restricted. Some messages or actions may be delayed or unavailable while we restore full support.",
-        "STATUS_LINK": "View status update"
+        "DUPLICATE_INBOX_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. You won’t be able to send/receive Instagram messages from this inbox anymore."
      },
      "TIKTOK": {
        "CONTINUE_WITH_TIKTOK": "Continue with TikTok",
@@ -819,6 +819,7 @@
      "HEADER": "Documents",
      "ADD_NEW": "Create a new document",
      "FAQ_COUNT": "{n} FAQ | {n} FAQs",
+      "USED_IN_CONVERSATIONS": "Used in {n} conversation | Used in {n} conversations",
      "SELECTED": "{count} selected",
      "SELECT_ALL": "Select all ({count})",
      "UNSELECT_ALL": "Unselect all ({count})",
@@ -462,7 +462,6 @@
      "STATUS": "Status",
      "IMPORTED": "Imported",
      "CREATED": "Created",
-      "RETRY": "Retry",
      "ABANDON": "Abandon"
    },
    "DETAIL": {
@@ -509,8 +508,6 @@
    },
    "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."
    }
@@ -417,6 +417,7 @@ onUnmounted(() => {
          :sync-in-progress="doc.sync_in_progress"
          :sync-stale-after-hours="syncIntervalHours"
          :responses-count="doc.responses_count"
+          :used-in-conversations-count="doc.used_in_conversations_count"
          :is-selected="canManageDocuments && bulkSelectedIds.has(doc.id)"
          :selectable="canManageDocuments"
          :show-selection-control="shouldShowSelectionControl(doc.id)"
@@ -26,7 +26,6 @@ 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);
@@ -36,7 +35,6 @@ const errorsOpen = ref(true);
 const skipLogsOpen = ref(true);
 let pollTimer;
 let isPageActive = false;
-let importRequestVersion = 0;

 const hasActiveImport = computed(() => isActiveImport(dataImport.value));

@@ -52,7 +50,6 @@ const fetchImport = async ({
  manual = false,
  requestedSkipLogsType = selectedSkipLogsType.value,
 } = {}) => {
-  const requestVersion = importRequestVersion;
  if (showLoader) {
    isLoading.value = true;
  } else if (manual) {
@@ -63,8 +60,6 @@ const fetchImport = async ({
    const response = await DataImportsAPI.show(route.params.dataImportId, {
      skip_logs_type: requestedSkipLogsType || undefined,
    });
-    if (requestVersion !== importRequestVersion) return;
-
    dataImport.value = response.data;
    selectedSkipLogsType.value =
      response.data.skip_logs_filters?.selected_source_object_type ||
@@ -110,13 +105,6 @@ const refreshImportInBackground = async () => {
  }
 };

-const startPolling = () => {
-  stopPolling();
-  if (!isPageActive || !hasActiveImport.value) return;
-
-  pollTimer = window.setInterval(refreshImportInBackground, POLL_INTERVAL_MS);
-};
-
 const abandonImport = async () => {
  isAbandoning.value = true;
  try {
@@ -129,22 +117,6 @@ const abandonImport = async () => {
  }
 };

-const retryImport = async () => {
-  isRetrying.value = true;
-  importRequestVersion += 1;
-  stopPolling();
-  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;
-    if (hasActiveImport.value) startPolling();
-  }
-};
-
 const downloadCsv = (response, filename) => {
  const url = window.URL.createObjectURL(
    new Blob([response.data], { type: 'text/csv' })
@@ -178,6 +150,13 @@ const downloadSkipLogs = async () => {
  }
 };

+const startPolling = () => {
+  stopPolling();
+  if (!isPageActive || !hasActiveImport.value) return;
+
+  pollTimer = window.setInterval(refreshImportInBackground, POLL_INTERVAL_MS);
+};
+
 const handleVisibilityChange = () => {
  if (isPageActive && !document.hidden && hasActiveImport.value) {
    refreshImportInBackground();
@@ -218,11 +197,9 @@ 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>
@@ -21,10 +21,6 @@ const props = defineProps({
    type: Boolean,
    default: false,
  },
-  isRetrying: {
-    type: Boolean,
-    default: false,
-  },
  isAbandoning: {
    type: Boolean,
    default: false,
@@ -35,7 +31,7 @@ const props = defineProps({
  },
 });

-defineEmits(['refresh', 'retry', 'abandon']);
+defineEmits(['refresh', 'abandon']);

 const { t } = useI18n();

@@ -94,23 +90,11 @@ 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')"
          />
@@ -1,85 +0,0 @@
-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);
-  });
-});
@@ -1,161 +0,0 @@
-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';
-import { POLL_INTERVAL_MS } from '../importStatus';
-
-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',
-  props: {
-    dataImport: {
-      type: Object,
-      default: null,
-    },
-  },
-  emits: ['retry'],
-  template: `
-    <button
-      v-if="dataImport?.stalled"
-      data-test="retry"
-      @click="$emit('retry')"
-    />
-  `,
-};
-
-const deferredRequest = () => {
-  let resolve;
-  const promise = new Promise(resolvePromise => {
-    resolve = resolvePromise;
-  });
-  return { promise, resolve };
-};
-
-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();
-  });
-
-  it('ignores an older poll response after retry succeeds', async () => {
-    const pollRequest = deferredRequest();
-    DataImportsAPI.retry.mockResolvedValue({
-      data: {
-        id: 1,
-        status: 'pending',
-        stalled: false,
-        skip_logs_filters: {},
-      },
-    });
-    const wrapper = mountShow();
-    await nextTick();
-    await flushPromises();
-
-    DataImportsAPI.show.mockReturnValueOnce(pollRequest.promise);
-    await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);
-    expect(DataImportsAPI.show).toHaveBeenCalledTimes(2);
-
-    await wrapper.find('[data-test="retry"]').trigger('click');
-    await flushPromises();
-    expect(wrapper.find('[data-test="retry"]').exists()).toBe(false);
-
-    pollRequest.resolve({
-      data: {
-        id: 1,
-        status: 'processing',
-        stalled: true,
-        skip_logs_filters: {},
-      },
-    });
-    await flushPromises();
-
-    expect(wrapper.find('[data-test="retry"]').exists()).toBe(false);
-    wrapper.unmount();
-  });
-});
@@ -4,8 +4,6 @@ import { shouldBeUrl } from 'shared/helpers/Validators';
 import { useAlert } from 'dashboard/composables';
 import { useVuelidate } from '@vuelidate/core';
 import Avatar from 'next/avatar/Avatar.vue';
-import Banner from 'dashboard/components-next/banner/Banner.vue';
-import Icon from 'dashboard/components-next/icon/Icon.vue';
 import SettingIntroBanner from 'dashboard/components/widgets/SettingIntroBanner.vue';
 import SettingsToggleSection from 'dashboard/components-next/Settings/SettingsToggleSection.vue';
 import SettingsFieldSection from 'dashboard/components-next/Settings/SettingsFieldSection.vue';
@@ -46,11 +44,9 @@ import SelectInput from 'dashboard/components-next/select/Select.vue';
 import Widget from 'dashboard/modules/widget-preview/components/Widget.vue';
 import AccessToken from 'dashboard/routes/dashboard/settings/profile/AccessToken.vue';
 import { copyTextToClipboard } from 'shared/helpers/clipboard';
-import { META_RESTRICTION_STATUS_URL } from 'dashboard/constants/globals';

 export default {
  components: {
-    Banner,
    BotConfiguration,
    CollaboratorsPage,
    ConfigurationPage,
@@ -84,7 +80,6 @@ export default {
    WhatsappManualMigrationBanner,
    Widget,
    AccessToken,
-    Icon,
  },
  mixins: [inboxMixin],
  setup() {
@@ -348,12 +343,6 @@ export default {
    instagramUnauthorized() {
      return this.isAnInstagramChannel && this.inbox.reauthorization_required;
    },
-    showInstagramRestrictionSettingsBanner() {
-      return this.isOnChatwootCloud && this.isAnInstagramChannel;
-    },
-    metaRestrictionStatusUrl() {
-      return META_RESTRICTION_STATUS_URL;
-    },
    tiktokUnauthorized() {
      return this.isATiktokChannel && this.inbox.reauthorization_required;
    },
@@ -820,29 +809,6 @@ export default {
          :class="bannerMaxWidth"
          @start="openWhatsAppManualMigrationDialog"
        />
-        <Banner
-          v-if="showInstagramRestrictionSettingsBanner"
-          color="amber"
-          class="mx-6 mb-4 max-w-4xl"
-        >
-          <div class="flex items-start gap-3 text-start">
-            <Icon
-              icon="i-lucide-triangle-alert"
-              class="flex-shrink-0 size-4 mt-0.5"
-            />
-            <span>
-              {{ $t('INBOX_MGMT.ADD.INSTAGRAM.SETTINGS_RESTRICTED_WARNING') }}
-              <a
-                :href="metaRestrictionStatusUrl"
-                class="link underline"
-                rel="noopener noreferrer nofollow"
-                target="_blank"
-              >
-                {{ $t('INBOX_MGMT.ADD.INSTAGRAM.STATUS_LINK') }}
-              </a>
-            </span>
-          </div>
-        </Banner>

        <div
          v-if="selectedTabKey === 'inbox-settings'"
@@ -33,7 +33,6 @@
 #
 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
@@ -72,10 +71,6 @@ 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
@@ -19,10 +19,6 @@ class DataImportPolicy < ApplicationPolicy
    show?
  end

-  def retry_import?
-    show?
-  end
-
  def abandon?
    show?
  end
@@ -7,17 +7,13 @@ class DataImports::Intercom::Importer
  end

  DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze
-  CONTACTS_PER_PAGE = 50
-  CONVERSATIONS_PER_PAGE = 10
-  HEARTBEAT_INTERVAL = 1.minute
-  QUERY_TIMEOUT_RETRY_LIMIT = 1
-  QUERY_TIMEOUT_RETRY_DELAY_RANGE = (0.2..0.5)
  PROVIDER = 'intercom'.freeze
  ALREADY_IMPORTED_ERROR_CODE = 'DataImports::Intercom::AlreadyImported'.freeze
  SKIPPED_MESSAGE_ERROR_CODE = 'DataImports::Intercom::SkippedMessage'.freeze
  TRUNCATED_PARTS_ERROR_CODE = 'DataImports::Intercom::TruncatedConversationParts'.freeze
  E164_REGEX = /\A\+[1-9]\d{1,14}\z/
  INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/
+  REGULAR_MESSAGE_PART_TYPES = %w[comment note source].freeze

  def initialize(data_import:, run_id: nil)
    @data_import = data_import
@@ -26,7 +22,6 @@ class DataImports::Intercom::Importer
    @client = DataImports::Intercom::Client.new(access_token: data_import.access_token)
    @placeholder_inboxes = DataImports::Intercom::PlaceholderInboxBuilder.new(account: @account)
    @stats = default_stats.deep_merge(data_import.stats || {})
-    @dirty_stat_groups = {}
  end

  def perform
@@ -49,9 +44,8 @@ class DataImports::Intercom::Importer
  def finish!
    return if @data_import.reload.abandoned?

-    error_count = @data_import.import_errors.non_skip_logs.count + @data_import.import_errors.failed.count
-    @stats['errors']['count'] = error_count
-    status = error_count.positive? ? :completed_with_errors : :completed
+    has_failures = @data_import.import_errors.non_skip_logs.exists? || @data_import.import_errors.failed.exists?
+    status = has_failures ? :completed_with_errors : :completed
    @data_import.update!(
      status: status,
      completed_at: Time.current,
@@ -62,16 +56,14 @@ class DataImports::Intercom::Importer
  end

  def fail!(error)
-    @data_import.with_lock do
-      next if @data_import.abandoned? || stale_import_run?
+    return if @data_import.reload.abandoned?

-      record_run_error(error)
-      @data_import.update!(status: :failed, last_error_at: Time.current)
-    end
+    record_run_error(error)
+    @data_import.update!(status: :failed, last_error_at: Time.current)
  end

  def import_contacts_page(starting_after: cursor_for('contacts'))
-    response = @client.list_contacts(starting_after: starting_after, per_page: CONTACTS_PER_PAGE)
+    response = @client.list_contacts(starting_after: starting_after)
    update_stat_total('contacts', response['total_count']) if response['total_count'].present?
    Array(response['data'] || response['contacts']).each do |contact|
      break if import_stopped?
@@ -80,14 +72,13 @@ class DataImports::Intercom::Importer
    end
    return PageResult.new(next_cursor: nil) if import_stopped?

-    reconcile_dirty_stats
    next_cursor = response.dig('pages', 'next', 'starting_after')
    update_cursor('contacts', next_cursor)
    PageResult.new(next_cursor: next_cursor)
  end

  def import_conversations_page(starting_after: cursor_for('conversations'))
-    response = @client.list_conversations(starting_after: starting_after, per_page: CONVERSATIONS_PER_PAGE)
+    response = @client.list_conversations(starting_after: starting_after)
    update_stat_total('conversations', response['total_count']) if response['total_count'].present?
    Array(response['data'] || response['conversations']).each do |conversation_summary|
      break if import_stopped?
@@ -96,7 +87,6 @@ class DataImports::Intercom::Importer
    end
    return PageResult.new(next_cursor: nil) if import_stopped?

-    reconcile_dirty_stats
    next_cursor = response.dig('pages', 'next', 'starting_after')
    update_cursor('conversations', next_cursor)
    PageResult.new(next_cursor: next_cursor)
@@ -163,9 +153,8 @@ class DataImports::Intercom::Importer
    mapped_conversation = mapping&.chatwoot_record
    if mapped_conversation && mapping.data_import_id != @data_import.id
      skip_already_imported_item(item, mapping, already_handled: already_handled)
-      reconcile_item_stats('conversation') if already_handled
-      return unless import_conversation_messages(conversation, mapped_conversation, contact)
-
+      import_source_message(conversation, mapped_conversation, contact)
+      import_conversation_parts(conversation, mapped_conversation, contact)
      update_conversation_activity(mapped_conversation)
      return
    end
@@ -175,21 +164,17 @@ class DataImports::Intercom::Importer
      record_mapping('conversation', source_id, chatwoot_conversation, metadata: conversation_metadata(conversation, inbox, source_type))
    end
    item.update!(status: :imported, chatwoot_record_type: 'Conversation', chatwoot_record_id: chatwoot_conversation.id)
-    if already_handled
-      reconcile_item_stats('conversation')
-    else
-      increment_stat('conversations', 'imported')
-    end
-
-    return unless import_conversation_messages(conversation, chatwoot_conversation, contact)
+    increment_stat('conversations', 'imported') unless already_handled

+    import_source_message(conversation, chatwoot_conversation, contact)
+    import_conversation_parts(conversation, chatwoot_conversation, contact)
    update_conversation_activity(chatwoot_conversation)
  rescue StandardError => e
    raise if e.is_a?(DataImports::Intercom::Client::Error)

    fail_item(item, e)
  ensure
-    persist_stats unless @import_stopped
+    persist_stats
  end

  def import_stopped?
@@ -199,49 +184,38 @@ class DataImports::Intercom::Importer
    @import_stopped = @data_import.abandoned? || @data_import.completed? || @data_import.completed_with_errors? || stale_import_run?
  end

-  def continue_import_with_heartbeat?
-    return false if import_stopped?
-    return true if @data_import.updated_at > HEARTBEAT_INTERVAL.ago
-
-    @data_import.touch if @data_import.updated_at <= HEARTBEAT_INTERVAL.ago
-    true
-  end
-
  def stale_import_run?
    active_run_id = @data_import.active_intercom_import_run_id
    @run_id.present? && active_run_id.present? && active_run_id != @run_id
  end

  def import_contact(contact_payload, required_for_conversation: false)
-    item = nil
-    with_query_timeout_retry do
-      source_id = source_id_for(contact_payload)
-      if source_id.present? && (mapping = find_mapping('contact', source_id)) && (mapped_contact = mapping.chatwoot_record)
-        return reuse_mapped_contact(contact_payload, source_id, mapping, mapped_contact)
-      end
-
-      contact_payload = retrieve_contact_payload(contact_payload)
-      source_id = source_id_for(contact_payload)
-      already_handled = item_handled?('contact', source_id)
-      item = import_item('contact', source_id, contact_payload)
-      mapping = find_mapping('contact', source_id)
-
-      mapped_contact = mapping&.chatwoot_record
-      if mapped_contact && mapping.data_import_id != @data_import.id
-        skip_already_imported_item(item, mapping, already_handled: already_handled)
-        return mapped_contact
-      end
-
-      contact = Contact.transaction do
-        imported_contact = mapped_contact || find_existing_contact(contact_payload) || create_contact(contact_payload)
-        update_existing_contact(imported_contact, contact_payload)
-        record_mapping('contact', source_id, imported_contact, metadata: contact_metadata(contact_payload))
-        item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: imported_contact.id)
-        imported_contact
-      end
-      increment_stat('contacts', 'imported') unless already_handled
-      contact
+    source_id = source_id_for(contact_payload)
+    if source_id.present? && (mapping = find_mapping('contact', source_id)) && (mapped_contact = mapping.chatwoot_record)
+      return reuse_mapped_contact(contact_payload, source_id, mapping, mapped_contact)
    end
+
+    contact_payload = retrieve_contact_payload(contact_payload)
+    source_id = source_id_for(contact_payload)
+    already_handled = item_handled?('contact', source_id)
+    item = import_item('contact', source_id, contact_payload)
+    mapping = find_mapping('contact', source_id)
+
+    mapped_contact = mapping&.chatwoot_record
+    if mapped_contact && mapping.data_import_id != @data_import.id
+      skip_already_imported_item(item, mapping, already_handled: already_handled)
+      return mapped_contact
+    end
+
+    contact = Contact.transaction do
+      imported_contact = mapped_contact || find_existing_contact(contact_payload) || create_contact(contact_payload)
+      update_existing_contact(imported_contact, contact_payload)
+      record_mapping('contact', source_id, imported_contact, metadata: contact_metadata(contact_payload))
+      item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: imported_contact.id)
+      imported_contact
+    end
+    increment_stat('contacts', 'imported') unless already_handled
+    contact
  rescue StandardError => e
    raise if e.is_a?(DataImports::Intercom::Client::Error)

@@ -395,85 +369,63 @@ class DataImports::Intercom::Importer
    end
  end

-  def import_conversation_messages(conversation, chatwoot_conversation, contact)
+  def import_source_message(conversation, chatwoot_conversation, contact)
+    source = conversation['source'].to_h
+    return unless source_message_importable?(source)
+
+    message_source_id = "conversation:#{source_id_for(conversation)}:source:#{source['id'].presence || 'initial'}"
+    source_part = source.merge('part_type' => 'source', 'created_at' => conversation['created_at'])
+    if (mapping = find_mapping('message', message_source_id)) && message_mapping_handled?(mapping, source_part)
+      if mapping.data_import_id == @data_import.id
+        reconcile_current_run_message_mapping(chatwoot_conversation, mapping, source_part)
+        return
+      end
+
+      skip_existing_message_mapping(chatwoot_conversation, mapping, source_part)
+      return
+    end
+
+    create_message(chatwoot_conversation, contact, source_part, message_source_id)
+  rescue StandardError => e
+    fail_message(chatwoot_conversation, message_source_id, source_part, e)
+  end
+
+  def import_conversation_parts(conversation, chatwoot_conversation, contact)
    parts_payload = conversation['conversation_parts'].to_h
    parts = Array(parts_payload['conversation_parts'])
-    batch_builder = DataImports::Intercom::MessageBatchBuilder.new(
-      data_import: @data_import,
-      conversation: chatwoot_conversation,
-      source_conversation: conversation
-    )
-    batch = begin
-      with_query_timeout_retry { batch_builder.perform }
-    rescue ActiveRecord::QueryCanceled
-      nil
-    end
-    return import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts.size) if batch.nil?
-
-    batch.source_entries.each { |entry| import_message(chatwoot_conversation, contact, entry) }
    record_truncated_conversation_parts(conversation, parts.size)

-    batch.part_entries.each do |entry|
-      return false unless continue_import_with_heartbeat?
+    parts.each do |part|
+      message_source_id = "conversation:#{source_id_for(conversation)}:part:#{part['id']}"
+      if (mapping = find_mapping('message', message_source_id)) && message_mapping_handled?(mapping, part)
+        if mapping.data_import_id == @data_import.id
+          reconcile_current_run_message_mapping(chatwoot_conversation, mapping, part)
+          next
+        end

-      import_message(chatwoot_conversation, contact, entry)
-    end
-    return false if import_stopped?
-
-    true
-  end
-
-  def import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts_count)
-    source_entries, part_entries = batch_builder.unprepared_entries.partition { |entry| entry[:part]['part_type'] == 'source' }
-    source_entries.each { |entry| import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry) }
-    record_truncated_conversation_parts(conversation, parts_count)
-
-    part_entries.each do |entry|
-      return false unless continue_import_with_heartbeat?
-
-      import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry)
-    end
-    return false if import_stopped?
-
-    true
-  end
-
-  def import_unprepared_message(conversation, contact, batch_builder, source_entry)
-    entry = with_query_timeout_retry { batch_builder.perform([source_entry]).entries.first }
-    import_message(conversation, contact, entry)
-  rescue StandardError => e
-    fail_message(conversation, source_entry[:source_id], source_entry[:part], e)
-  end
-
-  def import_message(conversation, contact, entry)
-    with_query_timeout_retry do
-      case entry.classification
-      when :current_import
-        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
-      when :previous_import
-        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
-      when :repairable_stale_mapping, :existing_message, :new_message
-        create_message(conversation, contact, entry)
-      else
-        raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
+        skip_existing_message_mapping(chatwoot_conversation, mapping, part)
+        next
      end
+
+      create_message(chatwoot_conversation, contact, part, message_source_id)
+    rescue StandardError => e
+      fail_message(chatwoot_conversation, message_source_id, part, e)
    end
-  rescue StandardError => e
-    fail_message(conversation, entry.source_id, entry.part, e)
  end

-  def create_message(conversation, contact, entry)
-    content = content_for(entry.part)
-    return record_skipped_message(conversation, entry) if content.blank?
+  def create_message(conversation, contact, part, message_source_id)
+    content = content_for(part)
+    return record_skipped_message(conversation, message_source_id, part) if content.blank?

-    attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
-    message = entry.message
+    attrs = message_attributes(conversation, contact, part, message_source_id, content)
+    message = nil
    Message.transaction do
+      message = conversation.messages.find_by(source_id: attrs[:source_id])
      unless message
        result = Message.insert_all!([attrs], returning: %w[id])
        message = Message.find(result.rows.first.first)
      end
-      record_message_mapping(entry, message)
+      record_mapping('message', message_source_id, message, metadata: message_metadata(part))
    end
    increment_stat('messages', 'imported')
    reindex_message_for_search(message)
@@ -488,12 +440,13 @@ class DataImports::Intercom::Importer
    Rails.logger.warn("Intercom import message reindex failed for message #{message.id}: #{e.class} - #{e.message}")
  end

-  def record_skipped_message(conversation, entry)
-    if entry.mapping
-      already_recorded = skip_log_recorded?('message', entry.source_id, SKIPPED_MESSAGE_ERROR_CODE)
-      record_skipped_message_log(conversation, entry.source_id, entry.part)
+  def record_skipped_message(conversation, message_source_id, part)
+    mapping = find_mapping('message', message_source_id)
+    if mapping
+      already_recorded = skip_log_recorded?('message', message_source_id, SKIPPED_MESSAGE_ERROR_CODE)
+      record_skipped_message_log(conversation, message_source_id, part)
      increment_stat('messages', 'skipped') unless already_recorded
-      return entry.message
+      return mapping.chatwoot_record
    end

    DataImportMapping.create!(
@@ -501,30 +454,15 @@ class DataImports::Intercom::Importer
      data_import: @data_import,
      source_provider: PROVIDER,
      source_object_type: 'message',
-      source_object_id: entry.source_id,
+      source_object_id: message_source_id,
      chatwoot_record_type: 'Conversation',
      chatwoot_record_id: conversation.id,
-      metadata: message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
+      metadata: message_metadata(part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
    )
-    record_skipped_message_log(conversation, entry.source_id, entry.part)
+    record_skipped_message_log(conversation, message_source_id, part)
    increment_stat('messages', 'skipped')
  end

-  def record_message_mapping(entry, message)
-    (entry.mapping || DataImportMapping.new(
-      account: @account,
-      source_provider: PROVIDER,
-      source_object_type: 'message',
-      source_object_id: entry.source_id
-    )).tap do |mapping|
-      mapping.data_import = @data_import
-      mapping.chatwoot_record_type = 'Message'
-      mapping.chatwoot_record_id = message.id
-      mapping.metadata = message_metadata(entry.part)
-      mapping.save!
-    end
-  end
-
  def message_attributes(conversation, contact, part, message_source_id, content)
    message_type = message_type_for(part)
    created_at = timestamp_for(part['created_at'])
@@ -565,7 +503,8 @@ class DataImports::Intercom::Importer
  end

  def activity_part?(part)
-    DataImports::Intercom::MessageBatchBuilder.activity_part?(part)
+    part_type = part['part_type'].to_s
+    part_type.present? && REGULAR_MESSAGE_PART_TYPES.exclude?(part_type)
  end

  def message_content(part)
@@ -690,7 +629,7 @@ class DataImports::Intercom::Importer
    )
    item = import_item('contact', source_id, contact_payload) unless item&.imported?
    item.update!(status: :imported, chatwoot_record_type: 'Contact', chatwoot_record_id: mapped_contact.id)
-    mark_stat_group_dirty('contacts')
+    reconcile_item_stats('contact')
  end

  def reconcile_item_stats(source_object_type)
@@ -698,37 +637,34 @@ class DataImports::Intercom::Importer
    group = stat_group_for(source_object_type)
    @stats[group]['imported'] = items.imported.count
    @stats[group]['skipped'] = items.skipped.count
+    persist_stats
  end

  def reconcile_current_run_message_mapping(conversation, mapping, part)
    record_skipped_message_log(conversation, mapping.source_object_id, part) if mapping.metadata['skipped']
-    mark_stat_group_dirty('messages')
-  end

-  def reconcile_message_stats
    mappings = @data_import.mappings.where(source_provider: PROVIDER, source_object_type: 'message')
    skipped_mappings = mappings.where("metadata ->> 'skipped' = ?", 'true').count
    message_logs = @data_import.import_errors.where(source_object_type: 'message')
    @stats['messages']['imported'] = mappings.count - skipped_mappings
    @stats['messages']['skipped'] = message_logs.where("details ->> 'kind' = ?", 'skipped').count
+    persist_stats
  end

  def skip_already_imported_item(item, mapping, already_handled:)
-    DataImportItem.transaction do
-      item.update!(
-        status: :skipped,
-        chatwoot_record_type: mapping.chatwoot_record_type,
-        chatwoot_record_id: mapping.chatwoot_record_id,
-        last_error_code: ALREADY_IMPORTED_ERROR_CODE,
-        last_error_message: 'Already imported in a previous import.'
-      )
-      record_already_imported_log(
-        data_import_item: item,
-        source_object_type: item.source_object_type,
-        source_object_id: item.source_object_id,
-        mapping: mapping
-      )
-    end
+    item.update!(
+      status: :skipped,
+      chatwoot_record_type: mapping.chatwoot_record_type,
+      chatwoot_record_id: mapping.chatwoot_record_id,
+      last_error_code: ALREADY_IMPORTED_ERROR_CODE,
+      last_error_message: 'Already imported in a previous import.'
+    )
+    record_already_imported_log(
+      data_import_item: item,
+      source_object_type: item.source_object_type,
+      source_object_id: item.source_object_id,
+      mapping: mapping
+    )
    increment_stat(stat_group_for(item.source_object_type), 'skipped') unless already_handled
  end

@@ -740,12 +676,13 @@ class DataImports::Intercom::Importer
      already_recorded = skip_log_recorded?('message', mapping.source_object_id, ALREADY_IMPORTED_ERROR_CODE)
      record_already_imported_log(source_object_type: 'message', source_object_id: mapping.source_object_id, mapping: mapping)
    end
-    if already_recorded
-      message_logs = @data_import.import_errors.where(source_object_type: 'message')
-      @stats['messages']['skipped'] = message_logs.where("details ->> 'kind' = ?", 'skipped').count
-    else
-      increment_stat('messages', 'skipped')
-    end
+    increment_stat('messages', 'skipped') unless already_recorded
+  end
+
+  def message_mapping_handled?(mapping, part)
+    return false if mapping.metadata['skipped'] && activity_part?(part)
+
+    mapping.metadata['skipped'] || mapping.chatwoot_record.present?
  end

  def fail_item(item, error)
@@ -845,7 +782,7 @@ class DataImports::Intercom::Importer
  end

  def source_message_importable?(source)
-    DataImports::Intercom::MessageBatchBuilder.source_message_importable?(source)
+    source['body'].present? || source['subject'].present? || source['attachments'].present?
  end

  def skipped_message_log_message(part)
@@ -998,42 +935,6 @@ class DataImports::Intercom::Importer
    @stats[group][key] = @stats[group][key].to_i + 1
  end

-  def mark_stat_group_dirty(group)
-    @dirty_stat_groups[group] = true
-  end
-
-  def reconcile_dirty_stats
-    return if @dirty_stat_groups.empty?
-
-    with_query_timeout_retry do
-      @dirty_stat_groups.each_key do |group|
-        case group
-        when 'contacts'
-          reconcile_item_stats('contact')
-        when 'messages'
-          reconcile_message_stats
-        else
-          raise ArgumentError, "Unsupported Intercom import stat group: #{group}"
-        end
-      end
-      persist_stats
-    end
-    @dirty_stat_groups.clear
-  end
-
-  def with_query_timeout_retry
-    retries = 0
-    begin
-      yield
-    rescue ActiveRecord::QueryCanceled
-      raise if retries >= QUERY_TIMEOUT_RETRY_LIMIT
-
-      retries += 1
-      sleep(rand(QUERY_TIMEOUT_RETRY_DELAY_RANGE))
-      retry
-    end
-  end
-
  def update_stat_total(group, total)
    @stats[group] ||= {}
    @stats[group]['total'] = total.to_i
@@ -1,134 +0,0 @@
-class DataImports::Intercom::MessageBatchBuilder
-  PROVIDER = 'intercom'.freeze
-  REGULAR_PART_TYPES = %w[comment note source].freeze
-
-  Entry = Struct.new(:source_id, :part, :position, :mapping, :message, :classification, keyword_init: true) do
-    def source?
-      part['part_type'] == 'source'
-    end
-  end
-
-  Batch = Struct.new(:items, keyword_init: true) do
-    def entries
-      items
-    end
-
-    def source_entries
-      items.select(&:source?)
-    end
-
-    def part_entries
-      items.reject(&:source?)
-    end
-  end
-
-  def self.activity_part?(part)
-    part_type = part['part_type'].to_s
-    part_type.present? && REGULAR_PART_TYPES.exclude?(part_type)
-  end
-
-  def self.source_message_importable?(source)
-    source['body'].present? || source['subject'].present? || source['attachments'].present?
-  end
-
-  def initialize(data_import:, conversation:, source_conversation:)
-    @data_import = data_import
-    @account = data_import.account
-    @conversation = conversation
-    @source_conversation = source_conversation
-  end
-
-  def perform(source_entries = unprepared_entries)
-    return Batch.new(items: []) if source_entries.empty?
-
-    mappings = message_mappings(source_entries)
-    messages = messages_for(source_entries, mappings)
-
-    Batch.new(items: source_entries.map do |source_entry|
-      build_entry(source_entry, source_entry[:position], mappings, messages)
-    end)
-  end
-
-  def unprepared_entries
-    ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) }
-  end
-
-  private
-
-  def ordered_source_entries
-    entries = []
-    source = @source_conversation['source'].to_h
-    if self.class.source_message_importable?(source)
-      entries << {
-        source_id: "conversation:#{source_conversation_id}:source:#{source['id'].presence || 'initial'}",
-        part: source.merge('part_type' => 'source', 'created_at' => @source_conversation['created_at'])
-      }
-    end
-
-    conversation_parts.each do |part|
-      entries << { source_id: "conversation:#{source_conversation_id}:part:#{part['id']}", part: part }
-    end
-    entries
-  end
-
-  def conversation_parts
-    Array(@source_conversation.dig('conversation_parts', 'conversation_parts'))
-  end
-
-  def source_conversation_id
-    @source_conversation['id'].presence || @source_conversation['external_id'].presence || @source_conversation['email'].presence
-  end
-
-  def message_mappings(source_entries)
-    DataImportMapping.where(
-      account: @account,
-      source_provider: PROVIDER,
-      source_object_type: 'message',
-      source_object_id: source_entries.pluck(:source_id)
-    ).index_by(&:source_object_id)
-  end
-
-  def messages_for(source_entries, mappings)
-    mapped_message_ids = mappings.values.filter_map do |mapping|
-      mapping.chatwoot_record_id if mapping.chatwoot_record_type == 'Message'
-    end
-    chatwoot_source_ids = source_entries.map { |entry| "intercom:#{entry[:source_id]}" }
-    messages = Message.where(id: mapped_message_ids).or(
-      Message.where(conversation_id: @conversation.id, source_id: chatwoot_source_ids)
-    ).to_a
-
-    {
-      by_id: messages.index_by(&:id),
-      by_source_id: messages.index_by(&:source_id)
-    }
-  end
-
-  def build_entry(source_entry, position, mappings, messages)
-    source_id = source_entry[:source_id]
-    mapping = mappings[source_id]
-    mapped_message = messages[:by_id][mapping.chatwoot_record_id] if mapping&.chatwoot_record_type == 'Message'
-    existing_message = messages[:by_source_id]["intercom:#{source_id}"]
-
-    Entry.new(
-      source_id: source_id,
-      part: source_entry[:part],
-      position: position,
-      mapping: mapping,
-      message: mapped_message || existing_message,
-      classification: classification_for(mapping, mapped_message, existing_message, source_entry[:part])
-    )
-  end
-
-  def classification_for(mapping, mapped_message, existing_message, part)
-    return existing_message.present? ? :existing_message : :new_message if mapping.blank?
-    return :repairable_stale_mapping unless mapping_handled?(mapping, mapped_message, part)
-
-    mapping.data_import_id == @data_import.id ? :current_import : :previous_import
-  end
-
-  def mapping_handled?(mapping, mapped_message, part)
-    return false if mapping.metadata['skipped'] && self.class.activity_part?(part)
-
-    mapping.metadata['skipped'] || mapped_message.present?
-  end
-end
@@ -1,28 +0,0 @@
-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.with_lock do
-        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
-  end
-
-  private
-
-  def another_active_import?
-    @account.data_imports.active_intercom.where.not(id: @data_import.id).exists?
-  end
-end
@@ -5,7 +5,6 @@ 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
@@ -1,5 +1,5 @@
 shared: &shared
-  version: '4.16.0'
+  version: '4.16.1'

 development:
  <<: *shared
@@ -230,7 +230,6 @@ Rails.application.routes.draw do
            end
            member do
              post :start
-              post :retry, action: :retry_import
              post :abandon
              get :error_logs
              get :skip_logs
@@ -0,0 +1,17 @@
+class CreateCaptainMessageSources < ActiveRecord::Migration[7.1]
+  def change
+    create_table :captain_message_sources do |t|
+      t.references :account, null: false, index: true
+      t.references :assistant, null: false, index: true
+      t.references :conversation, null: false, index: true
+      t.references :message, null: false, index: true
+      t.references :document, null: false, index: true
+      t.bigint :assistant_response_id, null: false
+
+      t.timestamps
+    end
+
+    add_index :captain_message_sources, [:message_id, :assistant_response_id],
+              unique: true, name: 'idx_captain_message_sources_on_message_and_response'
+  end
+end
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.

-ActiveRecord::Schema[7.1].define(version: 2026_07_13_184351) do
+ActiveRecord::Schema[7.1].define(version: 2026_07_21_100000) do
  # These extensions should be enabled to support this database
  enable_extension "pg_stat_statements"
  enable_extension "pg_trgm"
@@ -477,6 +477,23 @@ ActiveRecord::Schema[7.1].define(version: 2026_07_13_184351) do
    t.index ["user_id"], name: "index_captain_message_reports_on_user_id"
  end

+  create_table "captain_message_sources", force: :cascade do |t|
+    t.bigint "account_id", null: false
+    t.bigint "assistant_id", null: false
+    t.bigint "conversation_id", null: false
+    t.bigint "message_id", null: false
+    t.bigint "document_id", null: false
+    t.bigint "assistant_response_id", null: false
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["account_id"], name: "index_captain_message_sources_on_account_id"
+    t.index ["assistant_id"], name: "index_captain_message_sources_on_assistant_id"
+    t.index ["conversation_id"], name: "index_captain_message_sources_on_conversation_id"
+    t.index ["document_id"], name: "index_captain_message_sources_on_document_id"
+    t.index ["message_id", "assistant_response_id"], name: "idx_captain_message_sources_on_message_and_response", unique: true
+    t.index ["message_id"], name: "index_captain_message_sources_on_message_id"
+  end
+
  create_table "captain_scenarios", force: :cascade do |t|
    t.string "title"
    t.text "description"
@@ -11,7 +11,7 @@ class Api::V1::Accounts::Captain::DocumentsController < Api::V1::Accounts::BaseC
    @documents = filtered_documents
    @documents_count = @documents.count
    @sync_interval_hours = current_sync_interval&.in_hours&.to_i
-    @documents = with_responses_count(@documents).page(@current_page).per(RESULTS_PER_PAGE)
+    @documents = with_document_usage(@documents).page(@current_page).per(RESULTS_PER_PAGE)
  end

  def show; end
@@ -61,14 +61,23 @@ class Api::V1::Accounts::Captain::DocumentsController < Api::V1::Accounts::BaseC
    apply_sort(documents, permitted_params[:sort])
  end

-  def with_responses_count(scope)
-    scope.left_joins(:responses)
-         .select('captain_documents.*, COUNT(captain_assistant_responses.id) AS responses_count')
-         .group('captain_documents.id')
+  def with_document_usage(scope)
+    response_counts = Captain::AssistantResponse.where(documentable_type: 'Captain::Document')
+                                                .group(:documentable_id)
+                                                .select('documentable_id AS document_id, COUNT(*) AS responses_count')
+    source_counts = Captain::MessageSource.group(:document_id).select(
+      'document_id, COUNT(DISTINCT message_id) AS used_in_answers_count, COUNT(DISTINCT conversation_id) AS used_in_conversations_count'
+    )
+
+    scope.joins("LEFT JOIN (#{response_counts.to_sql}) response_counts ON response_counts.document_id = captain_documents.id")
+         .joins("LEFT JOIN (#{source_counts.to_sql}) source_counts ON source_counts.document_id = captain_documents.id")
+         .select('captain_documents.*, COALESCE(response_counts.responses_count, 0) AS responses_count, ' \
+                 'COALESCE(source_counts.used_in_answers_count, 0) AS used_in_answers_count, ' \
+                 'COALESCE(source_counts.used_in_conversations_count, 0) AS used_in_conversations_count')
  end

  def set_document
-    @document = @documents.find(permitted_params[:id])
+    @document = with_document_usage(@documents).find(permitted_params[:id])
  end

  def set_assistant
@@ -68,7 +68,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
        # left is the customer-facing follow-up message.
        process_v2_handoff
      end
-      capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0)
+      capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0, capture_message_sources: false)
    elsif v1_handoff_requested?
      # V1 only signals via the response string — no state has been touched yet. If
      # the conversation isn't pending anymore, a human took over mid-run; bail out
@@ -83,7 +83,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
        Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
        account.increment_response_usage
      end
-      capture_assistant_session(result_message: message, credits_consumed: 1.0)
+      capture_assistant_session(result_message: message, credits_consumed: 1.0, capture_message_sources: captain_v2_enabled?)
    end
  end

@@ -142,9 +142,10 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
  # Capture runs outside the delivery transaction and never raises (the service
  # swallows its own failures): a session-logging bug must never roll back the
  # customer reply or trigger the top-level handle_error handoff on top of it.
-  def capture_assistant_session(result_message:, credits_consumed:)
+  def capture_assistant_session(result_message:, credits_consumed:, capture_message_sources:)
    Captain::Assistant::SessionCaptureService.new(assistant: @assistant, conversation: @conversation, run_result: @run_result,
-                                                  result_message: result_message, credits_consumed: credits_consumed).capture
+                                                  result_message: result_message, credits_consumed: credits_consumed,
+                                                  capture_message_sources: capture_message_sources).capture
  end

  def handle_error(error)
@@ -34,6 +34,7 @@ class Captain::Document < ApplicationRecord

  belongs_to :assistant, class_name: 'Captain::Assistant'
  has_many :responses, class_name: 'Captain::AssistantResponse', dependent: :destroy, as: :documentable
+  has_many :message_sources, class_name: 'Captain::MessageSource', dependent: :destroy_async
  belongs_to :account
  has_one_attached :pdf_file
  store_accessor :metadata, :content_fingerprint, :last_sync_error_code, :sync_step, :openai_file_id
@@ -0,0 +1,33 @@
+# == Schema Information
+#
+# Table name: captain_message_sources
+#
+#  id                    :bigint           not null, primary key
+#  created_at            :datetime         not null
+#  updated_at            :datetime         not null
+#  account_id            :bigint           not null
+#  assistant_id          :bigint           not null
+#  assistant_response_id :bigint           not null
+#  conversation_id       :bigint           not null
+#  document_id           :bigint           not null
+#  message_id            :bigint           not null
+#
+# Indexes
+#
+#  idx_captain_message_sources_on_message_and_response  (message_id,assistant_response_id) UNIQUE
+#  index_captain_message_sources_on_account_id          (account_id)
+#  index_captain_message_sources_on_assistant_id        (assistant_id)
+#  index_captain_message_sources_on_conversation_id     (conversation_id)
+#  index_captain_message_sources_on_document_id         (document_id)
+#  index_captain_message_sources_on_message_id          (message_id)
+#
+class Captain::MessageSource < ApplicationRecord
+  self.table_name = 'captain_message_sources'
+
+  belongs_to :account
+  belongs_to :assistant, class_name: 'Captain::Assistant'
+  belongs_to :conversation, class_name: '::Conversation'
+  belongs_to :message
+  belongs_to :document, class_name: 'Captain::Document'
+  belongs_to :assistant_response, class_name: 'Captain::AssistantResponse', optional: true
+end
@@ -16,6 +16,7 @@ module Enterprise::Concerns::Account
    has_many :captain_documents, dependent: :destroy_async, class_name: 'Captain::Document'
    has_many :captain_custom_tools, dependent: :destroy_async, class_name: 'Captain::CustomTool'
    has_many :captain_agent_sessions, dependent: :destroy_async, class_name: 'Captain::AgentSession'
+    has_many :captain_message_sources, dependent: :destroy_async, class_name: 'Captain::MessageSource'

    has_many :copilot_threads, dependent: :destroy_async
    has_many :companies, dependent: :destroy_async
@@ -4,5 +4,6 @@ module Enterprise::Concerns::Message
  included do
    has_one :call, dependent: :nullify
    has_many :message_reports, class_name: 'Captain::MessageReport', dependent: :destroy_async
+    has_many :captain_message_sources, class_name: 'Captain::MessageSource', dependent: :destroy_async
  end
 end
@@ -1,12 +1,13 @@
 class Captain::Assistant::SessionCaptureService
  SCENARIO_AGENT_REGEX = /\A#{Captain::Scenario::HANDOFF_KEY_PREFIX}_(\d+)_/

-  def initialize(assistant:, conversation:, run_result:, result_message:, credits_consumed:)
+  def initialize(assistant:, conversation:, run_result:, result_message:, **options)
    @assistant = assistant
    @conversation = conversation
    @run_result = run_result
    @result_message = result_message
-    @credits_consumed = credits_consumed
+    @credits_consumed = options.fetch(:credits_consumed)
+    @capture_message_sources = options.fetch(:capture_message_sources, false)
  end

  def capture
@@ -23,7 +24,7 @@ class Captain::Assistant::SessionCaptureService
  def capture!
    model = @assistant.agent_model

-    Captain::AgentSession.create!(
+    session = Captain::AgentSession.create!(
      assistant: @assistant,
      session_type: :assistant,
      subject: @conversation,
@@ -35,6 +36,8 @@ class Captain::Assistant::SessionCaptureService
      scenario_ids: scenario_ids,
      run_context: current_turn_history
    )
+    capture_message_sources(metadata) if @capture_message_sources
+    session
  end

  private
@@ -70,6 +73,42 @@ class Captain::Assistant::SessionCaptureService
    ids & @assistant.scenarios.where(id: ids).pluck(:id)
  end

+  def capture_message_sources(metadata)
+    sources = Array(metadata[:message_sources])
+    return if sources.empty?
+
+    document_ids = @assistant.documents.where(id: sources.pluck(:document_id)).pluck(:id)
+    return if document_ids.empty?
+
+    insert_message_sources(message_source_rows(sources, document_ids))
+  end
+
+  def message_source_rows(sources, document_ids)
+    timestamp = Time.current
+    sources.filter_map do |source|
+      next unless document_ids.include?(source[:document_id])
+
+      {
+        account_id: @assistant.account_id,
+        assistant_id: @assistant.id,
+        conversation_id: @conversation.id,
+        message_id: @result_message.id,
+        document_id: source[:document_id],
+        assistant_response_id: source[:assistant_response_id],
+        created_at: timestamp,
+        updated_at: timestamp
+      }
+    end
+  end
+
+  def insert_message_sources(rows)
+    return if rows.empty?
+
+    # rubocop:disable Rails/SkipsModelValidations
+    Captain::MessageSource.insert_all(rows, unique_by: :idx_captain_message_sources_on_message_and_response)
+    # rubocop:enable Rails/SkipsModelValidations
+  end
+
  # Trim to the current turn: the last user message and everything after it
  # (assistant replies, tool calls/results, handoff hops).
  def current_turn_history
@@ -11,6 +11,10 @@ json.file_size resource.file_size
 json.pdf_document resource.pdf_document?
 responses_count = resource.respond_to?(:responses_count) ? resource.responses_count : resource.responses.count
 json.responses_count responses_count.to_i
+used_in_answers_count = resource.respond_to?(:used_in_answers_count) ? resource.used_in_answers_count : resource.message_sources.distinct.count(:message_id)
+json.used_in_answers_count used_in_answers_count.to_i
+used_in_conversations_count = resource.respond_to?(:used_in_conversations_count) ? resource.used_in_conversations_count : resource.message_sources.distinct.count(:conversation_id)
+json.used_in_conversations_count used_in_conversations_count.to_i
 json.id resource.id
 json.name resource.name
 json.status resource.status
@@ -26,8 +26,17 @@ class Captain::Tools::FaqLookupTool < Captain::Tools::BasePublicTool
    metadata = tool_context.state[:cw_metadata] ||= {}
    metadata[:faq_ids] = Array(metadata[:faq_ids]) | responses.map(&:id)

-    document_ids = responses.filter_map { |response| response.documentable_id if response.documentable_type == 'Captain::Document' }
+    document_ids = document_responses(responses).map(&:documentable_id)
    metadata[:document_ids] = Array(metadata[:document_ids]) | document_ids
+    metadata[:message_sources] = Array(metadata[:message_sources]) | message_sources(document_responses(responses))
+  end
+
+  def document_responses(responses)
+    responses.select { |response| response.documentable_type == 'Captain::Document' }
+  end
+
+  def message_sources(responses)
+    responses.map { |response| { assistant_response_id: response.id, document_id: response.documentable_id } }
  end

  def format_responses(responses)
@@ -1,6 +1,6 @@
 {
  "name": "@chatwoot/chatwoot",
-  "version": "4.16.0",
+  "version": "4.16.1",
  "license": "MIT",
  "scripts": {
    "eslint": "eslint app/**/*.{js,vue}",
@@ -64,36 +64,4 @@ 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
@@ -209,63 +209,6 @@ 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) }

@@ -340,7 +283,6 @@ 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
      )
@@ -66,12 +66,12 @@ RSpec.describe DataImports::Intercom::Importer do
  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, per_page: 50).and_return(
+    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, per_page: 10).and_return(
+    allow(client).to receive(:list_conversations).with(starting_after: nil).and_return(
      'conversations' => [{ 'id' => 'conversation_1' }],
      'total_count' => 1,
      'pages' => { 'next' => nil }
@@ -188,49 +188,14 @@ RSpec.describe DataImports::Intercom::Importer do
    expect(item.metadata['message_total_contribution']).to eq(3)
  end

-  it 'uses smaller conversation pages while retaining the contact page size' do
-    importer = described_class.new(data_import: data_import)
-
-    importer.import_contacts_page
-    importer.import_conversations_page
-
-    expect(client).to have_received(:list_contacts).with(starting_after: nil, per_page: 50)
-    expect(client).to have_received(:list_conversations).with(starting_after: nil, per_page: 10)
-  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)
-    importer = described_class.new(data_import: data_import)
-    expect(importer).to receive(:reconcile_message_stats).once.ordered.and_call_original
-    expect(importer).to receive(:update_cursor).with('conversations', nil).once.ordered.and_call_original

-    importer.import_conversations_page
-
-    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-  end
-
-  it 'retries a query timeout during deferred message stats reconciliation', :aggregate_failures 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)
-    importer = described_class.new(data_import: data_import)
-    reconciliation_attempts = 0
-    allow(importer).to receive(:sleep)
-    allow(importer).to receive(:reconcile_message_stats).and_wrap_original do |method|
-      reconciliation_attempts += 1
-      raise ActiveRecord::QueryCanceled, 'statement timeout' if reconciliation_attempts == 1

-      method.call
-    end
-
-    importer.import_conversations_page
-
-    expect(reconciliation_attempts).to eq(2)
-    expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
  end

@@ -302,7 +267,7 @@ RSpec.describe DataImports::Intercom::Importer 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, per_page: 10).and_return(
+      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' } }
      )
@@ -320,90 +285,6 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(data_import.reload.cursor.dig('conversations', 'starting_after')).to be_nil
    end

-    it 'heartbeats at most once per minute while processing conversation parts' do
-      freeze_time do
-        started_at = Time.current
-        long_conversation = conversation_payload.deep_dup
-        template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first
-        long_conversation['conversation_parts']['conversation_parts'] = Array.new(5) do |index|
-          template_part.merge('id' => "part_#{index + 1}")
-        end
-        allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation)
-        heartbeat_times = []
-        allow(data_import).to receive(:touch).and_wrap_original do |method|
-          heartbeat_times << Time.current
-          method.call
-        end
-        importer = described_class.new(data_import: data_import)
-        allow(importer).to receive(:create_message) { travel 30.seconds }
-
-        importer.import_conversations_page
-
-        expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes])
-      end
-    end
-
-    it 'stops parts without heartbeating or persisting stale stats when a newer run takes over', :aggregate_failures do
-      freeze_time do
-        run_id = 'intercom-run-1'
-        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-        allow(data_import).to receive(:touch).and_call_original
-        importer = described_class.new(data_import: data_import, run_id: run_id)
-        allow(importer).to receive(:create_message) do
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-        end
-
-        importer.import_conversations_page
-
-        expect(importer).to have_received(:create_message).once
-        expect(data_import).not_to have_received(:touch)
-        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
-      end
-    end
-
-    it 'does not persist stale stats when a newer run takes over during the final part', :aggregate_failures do
-      run_id = 'intercom-run-1'
-      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-      importer = described_class.new(data_import: data_import, run_id: run_id)
-      allow(importer).to receive(:create_message).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          entry = args[2]
-          next unless entry.part['id'] == 'part_2'
-
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-        end
-      end
-
-      importer.import_conversations_page
-
-      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
-      expect(conversation.messages.count).to eq(3)
-      expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
-    end
-
-    it 'reconciles conversation stats when a superseded run is retried', :aggregate_failures do
-      freeze_time do
-        run_id = 'intercom-run-1'
-        next_run_id = 'intercom-run-2'
-        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-        importer = described_class.new(data_import: data_import, run_id: run_id)
-        allow(importer).to receive(:create_message) do
-          data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
-          travel 1.minute
-        end
-
-        importer.import_conversations_page
-        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
-
-        retry_importer = described_class.new(data_import: data_import, run_id: next_run_id)
-        retry_importer.import_conversations_page
-        retry_importer.finish!
-
-        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(1)
-        expect(data_import.processed_records).to eq(5)
-      end
-    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:|
@@ -441,7 +322,11 @@ RSpec.describe DataImports::Intercom::Importer do

    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_message_mapping).and_raise(StandardError, 'mapping failed')
+      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

@@ -453,79 +338,6 @@ RSpec.describe DataImports::Intercom::Importer do
      )
      expect(error).to have_attributes(error_code: 'StandardError', message: 'mapping failed')
    end
-
-    it 'retries a contact query timeout after rolling back the first transaction', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
-        if object_type == 'contact'
-          mapping_attempts += 1
-          raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
-        end
-
-        method.call(object_type, source_id, record, metadata: metadata)
-      end
-
-      importer.import_contacts_page
-
-      expect(mapping_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(account.contacts.where(email: 'customer@example.com').count).to eq(1)
-      expect(data_import.mappings.where(source_object_type: 'contact', source_object_id: 'contact_1').count).to eq(1)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('contacts', 'imported')).to eq(1)
-    end
-
-    it 'retries a message query timeout after rolling back the first transaction', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      target_source_id = 'conversation:conversation_1:part:part_1'
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
-        if entry.source_id == target_source_id
-          mapping_attempts += 1
-          raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
-        end
-
-        method.call(entry, message)
-      end
-
-      importer.import_conversations_page
-
-      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
-      expect(mapping_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(conversation.messages.where(source_id: "intercom:#{target_source_id}").count).to eq(1)
-      expect(data_import.mappings.where(source_object_type: 'message', source_object_id: target_source_id).count).to eq(1)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-    end
-
-    it 'records one message failure only after the query timeout retry is exhausted', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      target_source_id = 'conversation:conversation_1:part:part_1'
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
-        if entry.source_id == target_source_id
-          mapping_attempts += 1
-          raise ActiveRecord::QueryCanceled, 'statement timeout'
-        end
-
-        method.call(entry, message)
-      end
-
-      importer.import_conversations_page
-
-      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
-      error = data_import.import_errors.find_by!(source_object_type: 'message', source_object_id: target_source_id)
-      expect(mapping_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(conversation.messages.where(source_id: "intercom:#{target_source_id}")).to be_empty
-      expect(error).to have_attributes(error_code: 'ActiveRecord::QueryCanceled', message: 'statement timeout')
-      expect(data_import.reload.stats.dig('errors', 'count')).to eq(1)
-    end
  end

  describe '#finish!' do
@@ -561,26 +373,6 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(data_import.last_error_at).to be_nil
      expect(data_import.import_errors.exists?).to be(false)
    end
-
-    it 'does not fail an import after a newer run takes over', :aggregate_failures do
-      run_id = 'intercom-run-1'
-      data_import.update!(
-        status: :processing,
-        source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }
-      )
-      importer = described_class.new(data_import: data_import, run_id: run_id)
-
-      DataImport.find(data_import.id).update!(
-        status: :pending,
-        source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'intercom-run-2' }
-      )
-
-      importer.fail!(StandardError.new('boom'))
-
-      expect(data_import.reload).to be_pending
-      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
@@ -614,57 +406,6 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
    end

-    it 'reconciles skipped stats when a superseded run is retried', :aggregate_failures do
-      described_class.new(data_import: data_import).perform
-      run_id = 'intercom-run-1'
-      next_run_id = 'intercom-run-2'
-      next_data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-
-      importer = described_class.new(data_import: next_data_import, run_id: run_id)
-      allow(importer).to receive(:skip_existing_message_mapping).and_wrap_original do |method, *args|
-        method.call(*args)
-        part = args[2]
-        next unless part['id'] == 'part_2'
-
-        DataImport.find(next_data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
-      end
-
-      importer.import_conversations_page
-      expect(next_data_import.reload.stats.dig('messages', 'skipped')).to eq(0)
-
-      retry_importer = described_class.new(data_import: next_data_import, run_id: next_run_id)
-      retry_importer.import_conversations_page
-      retry_importer.finish!
-
-      expect(next_data_import.reload.stats.dig('conversations', 'skipped')).to eq(1)
-      expect(next_data_import.stats.dig('messages', 'skipped')).to eq(3)
-      expect(next_data_import.total_records).to eq(5)
-    end
-
-    it 'keeps skipped contact stats when a query timeout is retried after the item update', :aggregate_failures do
-      described_class.new(data_import: data_import).perform
-      importer = described_class.new(data_import: next_data_import)
-      contact_log_attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_already_imported_log).and_wrap_original do |method, **attributes|
-        if attributes[:source_object_type] == 'contact'
-          contact_log_attempts += 1
-          raise ActiveRecord::QueryCanceled, 'statement timeout' if contact_log_attempts == 1
-        end
-
-        method.call(**attributes)
-      end
-
-      importer.import_contacts_page
-
-      contact_item = next_data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
-      expect(contact_log_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(contact_item).to be_skipped
-      expect(next_data_import.reload.stats.dig('contacts', 'skipped')).to eq(1)
-      expect(next_data_import.import_errors.skip_logs.exists?(data_import_item: contact_item)).to be(true)
-    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')
@@ -690,33 +431,6 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3)
    end

-    it 'repairs a missing mapping without recreating the existing message', :aggregate_failures do
-      described_class.new(data_import: data_import).import_conversations_page
-      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
-      message = conversation.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:part_1')
-      DataImportMapping.find_by!(
-        account: account,
-        source_provider: 'intercom',
-        source_object_type: 'message',
-        source_object_id: 'conversation:conversation_1:part:part_1'
-      ).destroy!
-      importer = described_class.new(data_import: data_import)
-      allow(importer).to receive(:find_mapping).and_call_original
-
-      importer.import_conversations_page
-
-      repaired_mapping = DataImportMapping.find_by!(
-        account: account,
-        source_provider: 'intercom',
-        source_object_type: 'message',
-        source_object_id: 'conversation:conversation_1:part:part_1'
-      )
-      expect(conversation.messages.where(source_id: message.source_id).count).to eq(1)
-      expect(repaired_mapping.chatwoot_record).to eq(message)
-      expect(importer).not_to have_received(:find_mapping).with('message', anything)
-      expect(data_import.reload.stats.dig('messages', 'imported')).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',
@@ -779,11 +493,7 @@ RSpec.describe DataImports::Intercom::Importer do
    end

    it 'repairs the item and imported count on retry', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      expect(importer).to receive(:reconcile_item_stats).with('contact').once.ordered.and_call_original
-      expect(importer).to receive(:update_cursor).with('contacts', nil).once.ordered.and_call_original
-
-      importer.import_contacts_page
+      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
@@ -1166,32 +876,6 @@ RSpec.describe DataImports::Intercom::Importer do
      expect(data_import.reload).to be_completed_with_errors
      expect(data_import.stats.dig('errors', 'count')).to eq(1)
    end
-
-    it 'reconciles error stats when a superseded run is retried', :aggregate_failures do
-      run_id = 'intercom-run-1'
-      next_run_id = 'intercom-run-2'
-      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
-      importer = described_class.new(data_import: data_import, run_id: run_id)
-      allow(importer).to receive(:import_message).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          entry = args[2]
-          next unless entry.part['id'] == 'part_2'
-
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
-        end
-      end
-
-      importer.import_conversations_page
-      expect(data_import.reload.stats.dig('errors', 'count')).to eq(0)
-
-      retry_importer = described_class.new(data_import: data_import, run_id: next_run_id)
-      retry_importer.import_conversations_page
-      retry_importer.finish!
-
-      expect(data_import.reload).to be_completed_with_errors
-      expect(data_import.stats.dig('errors', 'count')).to eq(1)
-      expect(data_import.total_records).to eq(6)
-    end
  end

  context 'when the conversation parts total matches the returned parts' do
@@ -1238,8 +922,6 @@ RSpec.describe DataImports::Intercom::Importer do
  end

  context 'when a specific Intercom message part fails to persist' do
-    let(:insert_attempts) { [] }
-
    let(:conversation_payload) do
      super().deep_merge(
        'conversation_parts' => {
@@ -1260,10 +942,7 @@ RSpec.describe DataImports::Intercom::Importer do

    before do
      allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
-        if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part'
-          insert_attempts << records.first[:source_id]
-          raise ActiveRecord::StatementInvalid, 'bad message'
-        end
+        raise ActiveRecord::StatementInvalid, 'bad message' if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part'

        method.call(records, **kwargs)
      end
@@ -1284,7 +963,6 @@ RSpec.describe DataImports::Intercom::Importer do
      )
      expect(data_import.reload).to be_completed_with_errors
      expect(data_import.stats.dig('errors', 'count')).to eq(1)
-      expect(insert_attempts.one?).to be(true)
    end
  end
 end
@@ -1,178 +0,0 @@
-require 'rails_helper'
-
-RSpec.describe DataImports::Intercom::MessageBatchBuilder do
-  let(:account) { create(:account) }
-  let(:data_import) { create(:data_import, :intercom, account: account) }
-  let(:conversation) { create(:conversation, account: account) }
-  let(:source_conversation) do
-    {
-      'id' => 'conversation_1',
-      'created_at' => 1_700_000_000,
-      'source' => {
-        'id' => 'source_1',
-        'part_type' => 'conversation',
-        'body' => '<p>Initial message</p>',
-        'created_at' => 1_700_000_000
-      },
-      'conversation_parts' => {
-        'conversation_parts' => [
-          {
-            'id' => 'part_1',
-            'part_type' => 'comment',
-            'body' => '<p>First reply</p>',
-            'created_at' => 1_700_000_000
-          },
-          {
-            'id' => 'part_2',
-            'part_type' => 'note',
-            'body' => '<p>Internal note</p>',
-            'created_at' => 1_700_000_100
-          }
-        ]
-      }
-    }
-  end
-  let(:builder) do
-    described_class.new(
-      data_import: data_import,
-      conversation: conversation,
-      source_conversation: source_conversation
-    )
-  end
-
-  it 'preserves source order and prefetches mappings and messages once per batch', :aggregate_failures do
-    sql_queries = []
-    subscriber = lambda do |_name, _start, _finish, _id, payload|
-      sql_queries << payload unless payload[:name] == 'SCHEMA'
-    end
-    batch_builder = builder
-
-    batch = ActiveSupport::Notifications.subscribed(subscriber, 'sql.active_record') { batch_builder.perform }
-
-    expect(batch.entries.map(&:source_id)).to eq(
-      %w[
-        conversation:conversation_1:source:source_1
-        conversation:conversation_1:part:part_1
-        conversation:conversation_1:part:part_2
-      ]
-    )
-    expect(batch.entries.map(&:position)).to eq([0, 1, 2])
-    expect(batch.entries.map(&:classification)).to all(eq(:new_message))
-    expect(sql_queries.count { |query| query[:name] == 'DataImportMapping Load' }).to eq(1)
-    message_queries = sql_queries.select { |query| query[:name] == 'Message Load' }
-    expect(message_queries.size).to eq(1), message_queries.pluck(:sql).join("\n")
-  end
-
-  it 'returns an empty batch without prefetch queries when the conversation has no messages' do
-    source_conversation['source'] = nil
-    source_conversation['conversation_parts']['conversation_parts'] = []
-
-    expect(DataImportMapping).not_to receive(:where)
-    expect(Message).not_to receive(:where)
-
-    expect(builder.perform.entries).to be_empty
-  end
-
-  it 'classifies a live mapping from the current import as already handled' do
-    message = create(
-      :message,
-      account: account,
-      conversation: conversation,
-      inbox: conversation.inbox,
-      source_id: 'intercom:conversation:conversation_1:part:part_1'
-    )
-    DataImportMapping.create!(
-      account: account,
-      data_import: data_import,
-      source_provider: 'intercom',
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1',
-      chatwoot_record_type: 'Message',
-      chatwoot_record_id: message.id,
-      metadata: {}
-    )
-
-    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
-
-    expect(entry).to have_attributes(classification: :current_import, message: message)
-  end
-
-  it 'classifies a live mapping from a previous import without changing its owner' do
-    previous_import = create(:data_import, :intercom, account: account)
-    message = create(
-      :message,
-      account: account,
-      conversation: conversation,
-      inbox: conversation.inbox,
-      source_id: 'intercom:conversation:conversation_1:part:part_1'
-    )
-    mapping = DataImportMapping.create!(
-      account: account,
-      data_import: previous_import,
-      source_provider: 'intercom',
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1',
-      chatwoot_record_type: 'Message',
-      chatwoot_record_id: message.id,
-      metadata: {}
-    )
-
-    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
-
-    expect(entry).to have_attributes(classification: :previous_import, mapping: mapping, message: message)
-    expect(mapping.reload.data_import).to eq(previous_import)
-  end
-
-  it 'classifies a mapping whose message was deleted as repairable' do
-    mapping = DataImportMapping.create!(
-      account: account,
-      data_import: data_import,
-      source_provider: 'intercom',
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1',
-      chatwoot_record_type: 'Message',
-      chatwoot_record_id: 0,
-      metadata: {}
-    )
-
-    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
-
-    expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
-  end
-
-  it 'classifies an existing conversation message without a mapping for repair' do
-    message = create(
-      :message,
-      account: account,
-      conversation: conversation,
-      inbox: conversation.inbox,
-      source_id: 'intercom:conversation:conversation_1:part:part_1'
-    )
-
-    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
-
-    expect(entry).to have_attributes(classification: :existing_message, mapping: nil, message: message)
-  end
-
-  it 'repairs a skipped mapping when the source part is now an activity' do
-    source_conversation.dig('conversation_parts', 'conversation_parts').first.merge!(
-      'part_type' => 'assignment',
-      'body' => nil,
-      'assigned_to' => { 'name' => 'Support' }
-    )
-    mapping = DataImportMapping.create!(
-      account: account,
-      data_import: data_import,
-      source_provider: 'intercom',
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1',
-      chatwoot_record_type: 'Conversation',
-      chatwoot_record_id: conversation.id,
-      metadata: { skipped: true }
-    )
-
-    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
-
-    expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
-  end
-end
@@ -1,82 +0,0 @@
-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 'rechecks the import state after acquiring its row lock' do
-    data_import.update!(updated_at: 16.minutes.ago)
-    allow(data_import).to receive(:with_lock).and_wrap_original do |method, *args, &block|
-      data_import.update!(status: :completed, completed_at: Time.current)
-      method.call(*args, &block)
-    end
-
-    result = described_class.new(account: account, data_import: data_import).perform
-
-    expect(result).to eq(:not_stalled)
-    expect(data_import.reload).to be_completed
-  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
Author	SHA1	Message	Date
aakashb95	509865536e	Merge branch 'codex/cw-7431-document-usage' into codex/cw-7498-document-conversation-usage	2026-07-23 14:04:03 +05:30
aakashb95	987000b95f	Merge remote-tracking branch 'origin/develop' into codex/cw-7431-document-usage	2026-07-23 14:03:56 +05:30
Sony Mathew	954e5844a8	Merge branch 'release/4.16.1' into develop	2026-07-23 13:58:16 +05:30
Sony Mathew	0efab5fb43	Bump version to 4.16.1	2026-07-23 13:57:15 +05:30
Muhsin KelothandGitHub	34ad78b122	fix(instagram): remove resolved restriction banners (#15136 )	2026-07-23 12:56:21 +05:30
aakashb95	da88b45f57	feat(captain): show document conversation usage	2026-07-21 12:07:31 +05:30
aakashb95	2ccb466598	feat(captain): track document usage for v2 answers	2026-07-21 12:04:41 +05:30
@@ -1 +1 @@
 .16.0
 .16.1