Merge branch 'develop' into feat/captain-credit-usage

app/builders/agent_builder.rb

@@ -2,14 +2,6 @@
 # It initializes with necessary attributes and provides a perform method
 # to create a user and account user in a transaction.
 class AgentBuilder
-  LIMIT_EXCEEDED_MESSAGE = 'Account limit exceeded. Please purchase more licenses'.freeze
-
-  class LimitExceededError < StandardError
-    def initialize
-      super(AgentBuilder::LIMIT_EXCEEDED_MESSAGE)
-    end
-  end
-
   # Initializes an AgentBuilder with necessary attributes.
   # @param email [String] the email of the user.
   # @param name [String] the name of the user.
@@ -22,23 +14,15 @@ class AgentBuilder
   # Creates a user and account user in a transaction.
   # @return [User] the created user.
   def perform
-    account.with_lock do
-      raise LimitExceededError unless can_add_agent?
-
-      ActiveRecord::Base.transaction do
-        @user = find_or_create_user
-        create_account_user
-      end
+    ActiveRecord::Base.transaction do
+      @user = find_or_create_user
+      create_account_user
     end
     @user
   end
 
   private
 
-  def can_add_agent?
-    account.usage_limits[:agents] > account.account_users.count
-  end
-
   # Finds a user by email or creates a new one with a temporary password.
   # @return [User] the found or created user.
   def find_or_create_user

app/controllers/api/v1/accounts/agents_controller.rb

@@ -1,6 +1,8 @@
 class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
   before_action :fetch_agent, except: [:create, :index, :bulk_create]
   before_action :check_authorization
+  before_action :validate_limit, only: [:create]
+  before_action :validate_limit_for_bulk_create, only: [:bulk_create]
 
   def index
     @agents = agents
@@ -18,8 +20,6 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
     )
 
     @agent = builder.perform
-  rescue AgentBuilder::LimitExceededError => e
-    render_payment_required(e.message)
   end
 
   def update
@@ -36,13 +36,25 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
   def bulk_create
     emails = params[:emails]
 
-    bulk_create_agents(emails)
+    emails.each do |email|
+      builder = AgentBuilder.new(
+        email: email,
+        name: email.split('@').first,
+        inviter: current_user,
+        account: Current.account
+      )
+      begin
+        builder.perform
+      rescue ActiveRecord::RecordInvalid => e
+        Rails.logger.info "[Agent#bulk_create] ignoring email #{email}, errors: #{e.record.errors}"
+      end
+    end
+
     # This endpoint is used to bulk create agents during onboarding
     # onboarding_step key in present in Current account custom attributes, since this is a one time operation
-    clear_onboarding_step
+    Current.account.custom_attributes.delete('onboarding_step')
+    Current.account.save!
     head :ok
-  rescue AgentBuilder::LimitExceededError => e
-    render_payment_required(e.message)
   end
 
   private
@@ -75,33 +87,22 @@ class Api::V1::Accounts::AgentsController < Api::V1::Accounts::BaseController
     @agents ||= Current.account.users.order_by_full_name.includes(:account_users, { avatar_attachment: [:blob] })
   end
 
-  def bulk_create_agents(emails)
-    Current.account.with_lock do
-      raise AgentBuilder::LimitExceededError if emails.count > available_agent_count
+  def validate_limit_for_bulk_create
+    limit_available = params[:emails].count <= available_agent_count
 
-      emails.each { |email| create_agent_from_email(email) }
-    end
+    render_payment_required('Account limit exceeded. Please purchase more licenses') unless limit_available
   end
 
-  def create_agent_from_email(email)
-    builder = AgentBuilder.new(
-      email: email,
-      name: email.split('@').first,
-      inviter: current_user,
-      account: Current.account
-    )
-    builder.perform
-  rescue ActiveRecord::RecordInvalid => e
-    Rails.logger.info "[Agent#bulk_create] ignoring email #{email}, errors: #{e.record.errors}"
-  end
-
-  def clear_onboarding_step
-    Current.account.custom_attributes.delete('onboarding_step')
-    Current.account.save!
+  def validate_limit
+    render_payment_required('Account limit exceeded. Please purchase more licenses') unless can_add_agent?
   end
 
   def available_agent_count
-    Current.account.usage_limits[:agents] - Current.account.account_users.count
+    Current.account.usage_limits[:agents] - agents.count
+  end
+
+  def can_add_agent?
+    available_agent_count.positive?
   end
 
   def delete_user_record(agent)

app/controllers/api/v1/accounts/data_imports_controller.rb

@@ -4,7 +4,7 @@ class Api::V1::Accounts::DataImportsController < Api::V1::Accounts::BaseControll
   DATA_IMPORT_FEATURE = 'data_import'.freeze
 
   before_action :ensure_data_import_feature_enabled
-  before_action :set_data_import, only: [:show, :start, :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

app/controllers/api/v1/accounts/integrations/base_controller.rb

@@ -1,9 +0,0 @@
-class Api::V1::Accounts::Integrations::BaseController < Api::V1::Accounts::BaseController
-  private
-
-  # Managing an integration hook (create/update/destroy) is admin-only, enforced via HookPolicy.
-  # Subclasses opt in per action with `before_action :check_authorization, only: [...]`.
-  def check_authorization
-    authorize(:hook)
-  end
-end

app/controllers/api/v1/accounts/integrations/hooks_controller.rb

@@ -1,4 +1,4 @@
-class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::Integrations::BaseController
+class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::BaseController
   before_action :fetch_hook, except: [:create]
   before_action :check_authorization
 
@@ -35,6 +35,10 @@ class Api::V1::Accounts::Integrations::HooksController < Api::V1::Accounts::Inte
     @hook = Current.account.hooks.find(params[:id])
   end
 
+  def check_authorization
+    authorize(:hook)
+  end
+
   def permitted_params
     params.require(:hook).permit(:app_id, :inbox_id, :status, settings: {})
   end

app/controllers/api/v1/accounts/integrations/linear_controller.rb

@@ -1,7 +1,6 @@
-class Api::V1::Accounts::Integrations::LinearController < Api::V1::Accounts::Integrations::BaseController
+class Api::V1::Accounts::Integrations::LinearController < Api::V1::Accounts::BaseController
   before_action :fetch_conversation, only: [:create_issue, :link_issue, :unlink_issue, :linked_issues]
   before_action :fetch_hook, only: [:destroy]
-  before_action :check_authorization, only: [:destroy]
 
   def destroy
     revoke_linear_token

app/controllers/api/v1/accounts/integrations/notion_controller.rb

@@ -1,6 +1,5 @@
-class Api::V1::Accounts::Integrations::NotionController < Api::V1::Accounts::Integrations::BaseController
+class Api::V1::Accounts::Integrations::NotionController < Api::V1::Accounts::BaseController
   before_action :fetch_hook, only: [:destroy]
-  before_action :check_authorization, only: [:destroy]
 
   def destroy
     @hook.destroy!

app/controllers/api/v1/accounts/integrations/shopify_controller.rb

@@ -1,8 +1,7 @@
-class Api::V1::Accounts::Integrations::ShopifyController < Api::V1::Accounts::Integrations::BaseController
+class Api::V1::Accounts::Integrations::ShopifyController < Api::V1::Accounts::BaseController
   include Shopify::IntegrationHelper
   before_action :setup_shopify_context, only: [:orders]
   before_action :fetch_hook, except: [:auth]
-  before_action :check_authorization, only: [:destroy]
   before_action :validate_contact, only: [:orders]
 
   def auth

app/controllers/webhooks/whatsapp_controller.rb

@@ -47,10 +47,18 @@ class Webhooks::WhatsappController < ActionController::API
     metadata = params.dig(:entry, 0, :changes, 0, :value, :metadata)
     return if metadata.blank?
 
-    Whatsapp::WebhookChannelFinderService.new(
-      display_phone_number: metadata[:display_phone_number],
-      phone_number_id: metadata[:phone_number_id]
-    ).perform
+    phone_number = normalized_phone_number(metadata[:display_phone_number])
+    phone_number_id = metadata[:phone_number_id]
+    channel = Channel::Whatsapp.find_by(phone_number: phone_number)
+
+    return channel if channel && channel.provider_config['phone_number_id'] == phone_number_id
+  end
+
+  def normalized_phone_number(phone_number)
+    return if phone_number.blank?
+
+    phone_number = phone_number.to_s
+    phone_number.start_with?('+') ? phone_number : "+#{phone_number}"
   end
 
   def inactive_whatsapp_number?

app/javascript/dashboard/api/captain/assistant.js

@@ -26,20 +26,13 @@ class CaptainAssistant extends ApiClient {
     });
   }
 
-  getMetrics({ assistantId, range, signal }) {
+  getStats({ assistantId, range, signal }) {
     const requestConfig = {
       params: { range, timezone_offset: getTimezoneOffset() },
     };
     if (signal) requestConfig.signal = signal;
 
-    return axios.get(`${this.url}/${assistantId}/metrics`, requestConfig);
-  }
-
-  getFaqStats({ assistantId, signal }) {
-    const requestConfig = {};
-    if (signal) requestConfig.signal = signal;
-
-    return axios.get(`${this.url}/${assistantId}/faq_stats`, requestConfig);
+    return axios.get(`${this.url}/${assistantId}/stats`, requestConfig);
   }
 
   getSummary({ assistantId, range, stats }) {

app/javascript/dashboard/api/dataImports.js

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

app/javascript/dashboard/components-next/captain/pageComponents/overview/CreditUsageCard.vue

@@ -0,0 +1,167 @@
+<script setup>
+import { computed } from 'vue';
+import { useI18n } from 'vue-i18n';
+
+const props = defineProps({
+  // The credit_usage stat from the overview stats endpoint:
+  // { current, previous, trend, daily: [{ date, value }] }
+  usage: { type: Object, default: null },
+});
+
+const { t } = useI18n();
+
+const daily = computed(() => props.usage?.daily ?? []);
+
+// Long ranges (90 days) get too many daily bars to read, so past this many
+// days the series is bucketed into weeks.
+const WEEKLY_THRESHOLD = 35;
+
+const isWeekly = computed(() => daily.value.length > WEEKLY_THRESHOLD);
+
+// One bucket per day, or per 7 consecutive days on long ranges (the last week
+// may be partial and always includes today). A bucket is unrecorded (null)
+// only when every day in it predates credit tracking.
+const buckets = computed(() => {
+  if (!isWeekly.value) {
+    return daily.value.map(day => ({
+      start: day.date,
+      end: day.date,
+      value: day.value,
+    }));
+  }
+
+  const weeks = [];
+  for (let i = 0; i < daily.value.length; i += 7) {
+    const chunk = daily.value.slice(i, i + 7);
+    const recorded = chunk.filter(day => day.value !== null);
+    weeks.push({
+      start: chunk[0].date,
+      end: chunk[chunk.length - 1].date,
+      value: recorded.length
+        ? +recorded.reduce((sum, day) => sum + day.value, 0).toFixed(2)
+        : null,
+    });
+  }
+  return weeks;
+});
+
+// Bar heights as a percentage of the peak bucket, with a small floor so even
+// the lowest one stays visible. Unrecorded buckets render as muted
+// placeholder bars.
+const bars = computed(() => {
+  const max = Math.max(...buckets.value.map(bucket => bucket.value ?? 0), 0);
+  return buckets.value.map(bucket => ({
+    ...bucket,
+    key: bucket.start,
+    recorded: bucket.value !== null,
+    height:
+      max === 0 || bucket.value === null
+        ? 6
+        : Math.max(6, Math.round((bucket.value / max) * 100)),
+  }));
+});
+
+// Bucket dates arrive as date-only strings ('2026-07-17'), which new Date()
+// would parse as midnight UTC and shift a day back for viewers west of UTC.
+// Parse the parts as a local date so the label matches the server-side bucket.
+const formatDate = date => {
+  const [year, month, day] = date.split('-').map(Number);
+  return new Date(year, month - 1, day).toLocaleDateString(undefined, {
+    month: 'short',
+    day: 'numeric',
+  });
+};
+
+const barTooltip = bar => {
+  const period =
+    bar.start === bar.end
+      ? formatDate(bar.start)
+      : `${formatDate(bar.start)} – ${formatDate(bar.end)}`;
+  const usage = bar.recorded
+    ? `${bar.value} ${t('CAPTAIN.OVERVIEW.CREDITS.UNIT')}`
+    : t('CAPTAIN.OVERVIEW.CREDITS.NO_DATA');
+  return `${period} · ${usage}`;
+};
+
+const total = computed(() =>
+  props.usage ? props.usage.current.toLocaleString() : '—'
+);
+
+const trend = computed(() => {
+  if (!props.usage?.trend) return '';
+  const sign = props.usage.trend > 0 ? '+' : '';
+  return `${sign}${props.usage.trend}%`;
+});
+
+const axisStart = computed(() =>
+  daily.value.length ? formatDate(daily.value[0].date) : ''
+);
+const axisEnd = computed(() =>
+  daily.value.length ? formatDate(daily.value[daily.value.length - 1].date) : ''
+);
+</script>
+
+<template>
+  <section
+    class="flex flex-col gap-4 p-5 border rounded-xl bg-n-solid-1 border-n-weak"
+  >
+    <div class="flex items-start justify-between gap-4">
+      <div class="flex flex-col gap-1">
+        <div class="flex items-center gap-1.5">
+          <h3 class="text-sm font-medium text-n-slate-12">
+            {{ $t('CAPTAIN.OVERVIEW.CREDITS.TITLE') }}
+          </h3>
+          <span
+            v-tooltip="$t('CAPTAIN.OVERVIEW.CREDITS.DISCLAIMER')"
+            class="cursor-help i-lucide-info size-3.5 text-n-slate-9"
+          />
+        </div>
+        <div class="flex items-baseline gap-2">
+          <span class="text-2xl font-semibold tabular-nums text-n-slate-12">
+            {{ total }}
+          </span>
+          <span class="text-sm text-n-slate-11">
+            {{ $t('CAPTAIN.OVERVIEW.CREDITS.UNIT') }}
+          </span>
+          <span
+            v-if="trend"
+            class="text-sm font-medium tabular-nums text-n-slate-11"
+          >
+            {{ trend }}
+          </span>
+        </div>
+      </div>
+      <div class="flex items-center gap-2 mt-1">
+        <span class="rounded-full size-2.5 bg-n-brand" />
+        <span class="text-xs text-n-slate-11">
+          {{
+            isWeekly
+              ? $t('CAPTAIN.OVERVIEW.CREDITS.LEGEND_WEEKLY')
+              : $t('CAPTAIN.OVERVIEW.CREDITS.LEGEND')
+          }}
+        </span>
+      </div>
+    </div>
+
+    <div class="flex flex-col justify-end flex-1">
+      <div class="flex items-end flex-1 gap-1 min-h-24">
+        <div
+          v-for="bar in bars"
+          :key="bar.key"
+          v-tooltip="barTooltip(bar)"
+          class="flex-1 rounded-t transition-colors"
+          :class="
+            bar.recorded
+              ? 'bg-n-brand hover:bg-n-brand/70'
+              : 'bg-n-alpha-2 hover:bg-n-alpha-3'
+          "
+          :style="{ height: `${bar.height}%` }"
+        />
+      </div>
+      <div class="flex items-center justify-between mt-3">
+        <span class="text-xs text-n-slate-10">{{ axisStart }}</span>
+        <span class="text-xs text-n-slate-10">{{ axisEnd }}</span>
+      </div>
+    </div>
+  </section>
+</template>

app/javascript/dashboard/components-next/captain/pageComponents/overview/KnowledgeCard.vue

@@ -57,13 +57,13 @@ const stats = computed(() => [
         {{ $t('CAPTAIN.OVERVIEW.KNOWLEDGE.COVERAGE', { pct: approvedPct }) }}
       </span>
     </div>
-    <div class="w-full h-2 overflow-hidden rounded-full bg-n-alpha-2">
+    <div class="w-full h-2 mt-4 overflow-hidden rounded-full bg-n-alpha-2">
       <div
         class="h-full rounded-full bg-n-brand"
         :style="{ width: `${approvedPct}%` }"
       />
     </div>
-    <div class="grid grid-cols-3 gap-3">
+    <div class="grid grid-cols-3 gap-3 mt-4">
       <RouterLink
         v-for="stat in stats"
         :key="stat.key"

app/javascript/dashboard/components-next/captain/pageComponents/overview/MetricCard.vue

@@ -26,7 +26,7 @@ const onActivate = () => {
 
 <template>
   <div
-    class="flex flex-col gap-3 p-5 group bg-n-solid-1"
+    class="flex flex-col gap-3 p-5 bg-n-solid-1"
     :class="
       clickable
         ? 'cursor-pointer transition-colors hover:bg-n-slate-2/50 focus-visible:outline focus-visible:outline-2 focus-visible:outline-n-brand'
@@ -43,7 +43,7 @@ const onActivate = () => {
       <span
         v-if="hint"
         v-tooltip="hint"
-        class="transition-opacity opacity-0 cursor-help i-lucide-info size-3.5 text-n-slate-10 group-hover:opacity-100"
+        class="cursor-help i-lucide-info size-3.5 text-n-slate-9"
       />
     </div>
     <div v-if="loading" class="flex items-end justify-between gap-2">

app/javascript/dashboard/components/widgets/WootWriter/FullEditor.vue

@@ -8,8 +8,6 @@ import {
   EditorState,
   Selection,
   imageResizeView,
-  toggleMark,
-  wrapInList,
 } from '@chatwoot/prosemirror-schema';
 import {
   suggestionsPlugin,
@@ -19,6 +17,8 @@ import imagePastePlugin from '@chatwoot/prosemirror-schema/src/plugins/image';
 import embedPreviewPlugin from '@chatwoot/prosemirror-schema/src/plugins/embedPreview';
 import trailingParagraphPlugin from '@chatwoot/prosemirror-schema/src/plugins/trailingParagraph';
 import { embeds as markdownEmbeds } from 'dashboard/helper/markdownEmbeds';
+import { toggleMark } from 'prosemirror-commands';
+import { wrapInList } from 'prosemirror-schema-list';
 import { toggleBlockType } from '@chatwoot/prosemirror-schema/src/menu/common';
 import { checkFileSizeLimit } from 'shared/helpers/FileHelper';
 import { isEscape } from 'shared/helpers/KeyboardHelpers';

app/javascript/dashboard/composables/spec/useAgentsList.spec.js

@@ -1,9 +1,9 @@
-import { useMapGetter } from 'dashboard/composables/store';
-import * as agentHelper from 'dashboard/helper/agentHelper';
-import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { ref } from 'vue';
+import { describe, it, expect, vi, beforeEach } from 'vitest';
 import { useAgentsList } from '../useAgentsList';
+import { useMapGetter } from 'dashboard/composables/store';
 import { allAgentsData, formattedAgentsData } from './fixtures/agentFixtures';
+import * as agentHelper from 'dashboard/helper/agentHelper';
 
 // Mock vue-i18n
 vi.mock('vue-i18n', () => ({
@@ -94,32 +94,6 @@ describe('useAgentsList', () => {
     expect(agentsList.value.length).toBe(formattedAgentsData.slice(1).length);
   });
 
-  it('keeps nameless agent bots and applies a fallback label', () => {
-    const namelessBot = {
-      id: 91,
-      name: null,
-      assignee_type: 'AgentBot',
-      availability_status: 'offline',
-    };
-    mockUseMapGetter({
-      'inboxAssignableAgents/getAssignableAgents': ref(() => [
-        ...allAgentsData,
-        namelessBot,
-      ]),
-    });
-
-    const { agentsList } = useAgentsList();
-    // access the computed to trigger evaluation
-    expect(agentsList.value).toBeDefined();
-
-    const passedAgents =
-      agentHelper.getAgentsByUpdatedPresence.mock.calls[0][0];
-    expect(passedAgents).toContainEqual({
-      ...namelessBot,
-      name: '-',
-    });
-  });
-
   it('handles empty assignable agents', () => {
     mockUseMapGetter({
       'inboxAssignableAgents/getAssignableAgents': ref(() => []),

app/javascript/dashboard/composables/useAgentsList.js

@@ -1,10 +1,10 @@
+import { computed } from 'vue';
 import { useMapGetter } from 'dashboard/composables/store';
+import { useI18n } from 'vue-i18n';
 import {
   getAgentsByUpdatedPresence,
   getSortedAgentsByAvailability,
 } from 'dashboard/helper/agentHelper';
-import { computed } from 'vue';
-import { useI18n } from 'vue-i18n';
 
 /**
  * A composable function that provides a list of agents for assignment.
@@ -53,11 +53,7 @@ export function useAgentsList(
    * @type {import('vue').ComputedRef<Array>}
    */
   const agentsList = computed(() => {
-    const agents = (assignableAgents.value || []).map(agent =>
-      !agent.name && agent.assignee_type === 'AgentBot'
-        ? { ...agent, name: '-' }
-        : agent
-    );
+    const agents = assignableAgents.value || [];
     const agentsByUpdatedPresence = getAgentsByUpdatedPresence(
       agents,
       currentUser.value,

app/javascript/dashboard/helper/agentHelper.js

@@ -7,7 +7,7 @@
 export const getAgentsByAvailability = (agents, availability) => {
   return agents
     .filter(agent => agent.availability_status === availability)
-    .sort((a, b) => (a.name || '').localeCompare(b.name || ''));
+    .sort((a, b) => a.name.localeCompare(b.name));
 };
 
 /**

app/javascript/dashboard/helper/editorHelper.js

@@ -1,6 +1,4 @@
 import {
-  InputRule,
-  inputRules,
   MessageMarkdownSerializer,
   MessageMarkdownTransformer,
   messageSchema,
@@ -11,6 +9,7 @@ import * as Sentry from '@sentry/vue';
 import camelcaseKeys from 'camelcase-keys';
 import { FORMATTING, MARKDOWN_PATTERNS } from 'dashboard/constants/editor';
 import { INBOX_TYPES, TWILIO_CHANNEL_MEDIUM } from 'dashboard/helper/inbox';
+import { InputRule, inputRules } from 'prosemirror-inputrules';
 
 /**
  * Extract text from markdown, and remove all images, code blocks, links, headers, bold, italic, lists etc.

app/javascript/dashboard/helper/specs/agentHelper.spec.js

@@ -26,18 +26,6 @@ describe('agentHelper', () => {
         offlineAgentsData
       );
     });
-
-    it('does not throw when an agent has a null name', () => {
-      const agents = [
-        { id: 1, name: null, availability_status: 'offline' },
-        { id: 2, name: 'Zoe', availability_status: 'offline' },
-      ];
-
-      expect(() => getAgentsByAvailability(agents, 'offline')).not.toThrow();
-      expect(
-        getAgentsByAvailability(agents, 'offline').map(agent => agent.id)
-      ).toEqual([1, 2]);
-    });
   });
 
   describe('getSortedAgentsByAvailability', () => {

app/javascript/dashboard/i18n/locale/en/integrations.json

@@ -460,6 +460,14 @@
         "PENDING": "Pending FAQs",
         "DOCUMENTS": "Documents"
       },
+      "CREDITS": {
+        "TITLE": "Credit usage",
+        "UNIT": "credits",
+        "LEGEND": "Daily credits used",
+        "LEGEND_WEEKLY": "Weekly credits used",
+        "NO_DATA": "No data recorded",
+        "DISCLAIMER": "Shows credits used by this assistant only. Your account's total credit usage also includes other assistants and features."
+      },
       "LINKS": {
         "DOCS": {
           "TITLE": "Captain docs",

app/javascript/dashboard/i18n/locale/en/settings.json

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

app/javascript/dashboard/routes/dashboard/captain/assistants/overview/Index.vue

@@ -13,6 +13,7 @@ import WelcomeCard from 'dashboard/components-next/captain/pageComponents/overvi
 import MetricCard from 'dashboard/components-next/captain/pageComponents/overview/MetricCard.vue';
 import AssistantDrilldownDrawer from 'dashboard/components-next/captain/pageComponents/overview/AssistantDrilldownDrawer.vue';
 import KnowledgeCard from 'dashboard/components-next/captain/pageComponents/overview/KnowledgeCard.vue';
+import CreditUsageCard from 'dashboard/components-next/captain/pageComponents/overview/CreditUsageCard.vue';
 import QuickLinks from 'dashboard/components-next/captain/pageComponents/overview/QuickLinks.vue';
 import InboxBanner from 'dashboard/components-next/captain/pageComponents/overview/InboxBanner.vue';
 import CoverageBanner from 'dashboard/components-next/captain/pageComponents/overview/CoverageBanner.vue';
@@ -26,28 +27,25 @@ const canDrilldown = computed(() => checkPermissions(['administrator']));
 const selectedRange = ref('this_month');
 
 const assistantId = computed(() => route.params.assistantId);
-const metricStats = ref(null);
-const faqStats = ref(null);
-const isFetchingMetrics = ref(false);
+const stats = ref(null);
+const isFetching = ref(false);
 
 // Increments on every fetch so a response (or retry) from a superseded
 // range/assistant can't clobber the latest request's state.
-let metricsFetchToken = 0;
-let faqStatsFetchToken = 0;
-let metricsAbortController = null;
-let faqStatsAbortController = null;
+let fetchToken = 0;
+let abortController = null;
 
-const fetchMetrics = async () => {
-  metricsFetchToken += 1;
-  const token = metricsFetchToken;
-  metricsAbortController?.abort();
-  metricsAbortController = new AbortController();
-  const { signal } = metricsAbortController;
-  metricStats.value = null;
-  isFetchingMetrics.value = true;
+const fetchStats = async () => {
+  fetchToken += 1;
+  const token = fetchToken;
+  abortController?.abort();
+  abortController = new AbortController();
+  const { signal } = abortController;
+  stats.value = null;
+  isFetching.value = true;
 
-  const requestMetrics = () =>
-    CaptainAssistant.getMetrics({
+  const requestStats = () =>
+    CaptainAssistant.getStats({
       assistantId: assistantId.value,
       range: selectedRange.value,
       signal,
@@ -55,54 +53,25 @@ const fetchMetrics = async () => {
 
   let data = null;
   try {
-    ({ data } = await requestMetrics());
+    ({ data } = await requestStats());
   } catch {
     // One silent retry before giving up, unless the request was aborted.
     try {
-      if (token === metricsFetchToken && !signal.aborted)
-        ({ data } = await requestMetrics());
+      if (token === fetchToken && !signal.aborted)
+        ({ data } = await requestStats());
     } catch {
       data = null;
     }
   }
 
-  if (token !== metricsFetchToken || signal.aborted) return;
-  metricStats.value = data;
-  isFetchingMetrics.value = false;
+  if (token !== fetchToken || signal.aborted) return;
+  stats.value = data;
+  isFetching.value = false;
 };
 
-const fetchFaqStats = async () => {
-  faqStatsFetchToken += 1;
-  const token = faqStatsFetchToken;
-  faqStatsAbortController?.abort();
-  faqStatsAbortController = new AbortController();
-  const { signal } = faqStatsAbortController;
-  faqStats.value = null;
+onUnmounted(() => abortController?.abort());
 
-  try {
-    const { data } = await CaptainAssistant.getFaqStats({
-      assistantId: assistantId.value,
-      signal,
-    });
-    if (token === faqStatsFetchToken && !signal.aborted) faqStats.value = data;
-  } catch {
-    if (token === faqStatsFetchToken && !signal.aborted) faqStats.value = null;
-  }
-};
-
-const summaryStats = computed(() => {
-  if (!metricStats.value || !faqStats.value) return null;
-
-  return { ...metricStats.value, knowledge: faqStats.value };
-});
-
-onUnmounted(() => {
-  metricsAbortController?.abort();
-  faqStatsAbortController?.abort();
-});
-
-watch([selectedRange, assistantId], fetchMetrics, { immediate: true });
-watch(assistantId, fetchFaqStats, { immediate: true });
+watch([selectedRange, assistantId], fetchStats, { immediate: true });
 
 // `direction` says whether a rising trend is good ('up'), bad ('down'), or
 // neutral, so we can colour the delta independently of its sign.
@@ -122,7 +91,7 @@ const formatDuration = hours =>
   hours >= 100 ? `${Math.round(hours / 24)}d` : `${hours}h`;
 
 const metricFor = (statKey, formatValue, direction, trendKind = 'percent') => {
-  const data = metricStats.value?.[statKey];
+  const data = stats.value?.[statKey];
   if (!data) return { value: '—', trend: '', trendGood: null };
 
   const sign = data.trend > 0 ? '+' : '';
@@ -216,9 +185,9 @@ const closeDrilldown = () => {
       <div class="flex flex-col gap-6 pb-8">
         <InboxBanner />
 
-        <CoverageBanner :knowledge="faqStats ?? undefined" />
+        <CoverageBanner :knowledge="stats?.knowledge" />
 
-        <WelcomeCard :range="selectedRange" :stats="summaryStats" />
+        <WelcomeCard :range="selectedRange" :stats="stats" />
 
         <div
           class="grid grid-cols-1 gap-px overflow-hidden border rounded-xl sm:grid-cols-2 lg:grid-cols-3 bg-n-weak border-n-weak"
@@ -231,15 +200,16 @@ const closeDrilldown = () => {
             :trend="metric.trend"
             :hint="metric.hint"
             :trend-good="metric.trendGood"
-            :loading="isFetchingMetrics"
-            :clickable="
-              canDrilldown && Boolean(metric.metric) && !isFetchingMetrics
-            "
+            :loading="isFetching"
+            :clickable="canDrilldown && Boolean(metric.metric) && !isFetching"
             @click="openDrilldown(metric)"
           />
         </div>
 
-        <KnowledgeCard :knowledge="faqStats ?? undefined" />
+        <div class="grid grid-cols-1 gap-4 lg:grid-cols-2">
+          <KnowledgeCard :knowledge="stats?.knowledge" />
+          <CreditUsageCard :usage="stats?.credit_usage" />
+        </div>
 
         <QuickLinks />
       </div>

app/javascript/dashboard/routes/dashboard/settings/agentBots/Index.vue

@@ -135,7 +135,7 @@ onMounted(() => {
               <BaseTableCell class="max-w-0">
                 <div class="flex items-center gap-4 min-w-0">
                   <Avatar
-                    :name="bot.name || ''"
+                    :name="bot.name"
                     :src="bot.thumbnail"
                     :size="40"
                     class="flex-shrink-0"

app/javascript/dashboard/routes/dashboard/settings/data/Show.vue

@@ -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>

app/javascript/dashboard/routes/dashboard/settings/data/components/ImportDetailHeader.vue

@@ -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')"
           />

app/javascript/dashboard/routes/dashboard/settings/data/specs/ImportDetailHeader.spec.js

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

app/javascript/dashboard/routes/dashboard/settings/data/specs/showActions.spec.js

@@ -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();
-  });
-});

app/javascript/shared/components/ui/MultiselectDropdown.vue

@@ -68,10 +68,6 @@ const isAgentBot = computed(
   () => props.selectedItem?.assignee_type === 'AgentBot'
 );
 
-const selectedItemName = computed(() =>
-  !props.selectedItem?.name && isAgentBot.value ? '-' : props.selectedItem?.name
-);
-
 const selectedThumbnail = computed(
   () => props.selectedItem?.thumbnail || props.selectedItem?.avatar_url
 );
@@ -99,16 +95,16 @@ const selectedThumbnail = computed(
           <h4
             v-else
             class="items-center overflow-hidden text-sm leading-tight whitespace-nowrap text-ellipsis text-n-slate-12"
-            :title="selectedItemName"
+            :title="selectedItem.name"
           >
-            {{ selectedItemName }}
+            {{ selectedItem.name }}
           </h4>
         </div>
         <Avatar
           v-if="hasValue && hasThumbnail && (isAgentBot || !hasIcon)"
           :src="selectedThumbnail"
           :status="selectedItem.availability_status"
-          :name="selectedItemName"
+          :name="selectedItem.name"
           :icon-name="isAgentBot ? 'i-lucide-bot' : undefined"
           :size="24"
           hide-offline-status

app/javascript/shared/components/ui/MultiselectDropdownItems.vue

@@ -53,9 +53,7 @@ export default {
   computed: {
     filteredOptions() {
       return this.options.filter(option => {
-        return (option.name || '')
-          .toLowerCase()
-          .includes(this.search.toLowerCase());
+        return option.name.toLowerCase().includes(this.search.toLowerCase());
       });
     },
     noResult() {

app/javascript/widget/components/ChatFooter.vue

@@ -11,8 +11,6 @@ import { IFrameHelper } from '../helpers/utils';
 import { CHATWOOT_ON_START_CONVERSATION } from '../constants/sdkEvents';
 import { emitter } from 'shared/helpers/mitt';
 
-const TRANSCRIPT_COOLDOWN_MS = 15000;
-
 export default {
   components: {
     ChatInputWrap,
@@ -26,9 +24,6 @@ export default {
   data() {
     return {
       inReplyTo: null,
-      isSendingTranscript: false,
-      transcriptCooldown: false,
-      transcriptCooldownTimer: null,
     };
   },
   computed: {
@@ -62,9 +57,6 @@ export default {
   mounted() {
     emitter.on(BUS_EVENTS.TOGGLE_REPLY_TO_MESSAGE, this.toggleReplyTo);
   },
-  beforeUnmount() {
-    clearTimeout(this.transcriptCooldownTimer);
-  },
   methods: {
     ...mapActions('conversation', ['sendMessage', 'sendAttachment']),
     ...mapActions('conversationAttributes', ['getAttributes']),
@@ -98,35 +90,19 @@ export default {
     toggleReplyTo(message) {
       this.inReplyTo = message;
     },
-    startTranscriptCooldown() {
-      this.transcriptCooldown = true;
-      clearTimeout(this.transcriptCooldownTimer);
-      this.transcriptCooldownTimer = setTimeout(() => {
-        this.transcriptCooldown = false;
-      }, TRANSCRIPT_COOLDOWN_MS);
-    },
     async sendTranscript() {
-      if (
-        !this.hasEmail ||
-        this.isSendingTranscript ||
-        this.transcriptCooldown
-      ) {
-        return;
-      }
-      this.isSendingTranscript = true;
-      try {
-        await sendEmailTranscript();
-        this.startTranscriptCooldown();
-        emitter.emit(BUS_EVENTS.SHOW_ALERT, {
-          message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_SUCCESS'),
-          type: 'success',
-        });
-      } catch (error) {
-        emitter.emit(BUS_EVENTS.SHOW_ALERT, {
-          message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_ERROR'),
-        });
-      } finally {
-        this.isSendingTranscript = false;
+      if (this.hasEmail) {
+        try {
+          await sendEmailTranscript();
+          emitter.emit(BUS_EVENTS.SHOW_ALERT, {
+            message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_SUCCESS'),
+            type: 'success',
+          });
+        } catch (error) {
+          emitter.$emit(BUS_EVENTS.SHOW_ALERT, {
+            message: this.$t('EMAIL_TRANSCRIPT.SEND_EMAIL_ERROR'),
+          });
+        }
       }
     },
   },
@@ -168,7 +144,6 @@ export default {
       v-if="showEmailTranscriptButton"
       type="clear"
       class="font-normal"
-      :disabled="isSendingTranscript || transcriptCooldown"
       @click="sendTranscript"
     >
       {{ $t('EMAIL_TRANSCRIPT.BUTTON_TEXT') }}

app/jobs/webhooks/whatsapp_events_job.rb

@@ -153,11 +153,11 @@ class Webhooks::WhatsappEventsJob < MutexApplicationJob
   end
 
   def get_channel_from_wb_payload(wb_params)
-    metadata = wb_params[:entry].first[:changes].first.dig(:value, :metadata) || {}
-    Whatsapp::WebhookChannelFinderService.new(
-      display_phone_number: metadata[:display_phone_number],
-      phone_number_id: metadata[:phone_number_id]
-    ).perform
+    phone_number = "+#{wb_params[:entry].first[:changes].first.dig(:value, :metadata, :display_phone_number)}"
+    phone_number_id = wb_params[:entry].first[:changes].first.dig(:value, :metadata, :phone_number_id)
+    channel = Channel::Whatsapp.find_by(phone_number: phone_number)
+    # validate to ensure the phone number id matches the whatsapp channel
+    return channel if channel && channel.provider_config['phone_number_id'] == phone_number_id
   end
 end
 

app/models/data_import.rb

@@ -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

app/policies/data_import_policy.rb

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

app/services/data_imports/intercom/importer.rb

@@ -1,7 +1,5 @@
 # rubocop:disable Metrics/ClassLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength, Rails/SkipsModelValidations
 class DataImports::Intercom::Importer
-  class InvalidMessagePayloadError < StandardError; end
-
   PageResult = Struct.new(:next_cursor, keyword_init: true) do
     def done?
       next_cursor.blank?
@@ -9,29 +7,13 @@ class DataImports::Intercom::Importer
   end
 
   DEFAULT_IMPORT_TYPES = %w[contacts conversations].freeze
-  CONTACTS_PER_PAGE = 50
-  CONVERSATIONS_PER_PAGE = 10
-  MESSAGES_PER_BATCH = 100
-  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
-  MESSAGE_MAPPING_UNIQUE_INDEX = :idx_data_import_mappings_on_account_and_source
   E164_REGEX = /\A\+[1-9]\d{1,14}\z/
   INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/
-
-  MessageBatchResult = Struct.new(
-    :imported_entries,
-    :skipped_entries,
-    :current_entries,
-    :previous_entries,
-    :messages,
-    :failed_entries,
-    keyword_init: true
-  )
+  REGULAR_MESSAGE_PART_TYPES = %w[comment note source].freeze
 
   def initialize(data_import:, run_id: nil)
     @data_import = data_import
@@ -40,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
@@ -63,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,
@@ -76,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?
@@ -94,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?
@@ -110,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)
@@ -177,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
@@ -189,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?
@@ -213,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)
 
@@ -409,291 +369,66 @@ 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?
-
     record_truncated_conversation_parts(conversation, parts.size)
 
-    batch.entries.each_slice(MESSAGES_PER_BATCH) do |entries|
-      return false unless continue_import_with_heartbeat?
-
-      import_message_batch(chatwoot_conversation, contact, batch_builder, entries)
-      return false if @import_stopped
-    end
-    return false if import_stopped?
-
-    true
-  end
-
-  def import_message_batch(conversation, contact, batch_builder, entries)
-    result = bulk_message_batch_result(conversation, contact, batch_builder, entries)
-    return if result.blank?
-
-    result.current_entries.each do |entry|
-      reconcile_bulk_message_entry(conversation, entry) do
-        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
-      end
-    end
-    result.previous_entries.each do |entry|
-      reconcile_bulk_message_entry(conversation, entry) do
-        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
-      end
-    end
-    result.skipped_entries.each do |entry|
-      reconcile_bulk_message_entry(conversation, entry) { record_bulk_skipped_message(conversation, entry) }
-    end
-    Array(result.failed_entries).each { |entry, error| fail_message(conversation, entry.source_id, entry.part, error) }
-    increment_stat('messages', 'imported', result.imported_entries.size)
-    result.messages.each { |message| reindex_message_for_search(message) }
-  end
-
-  def bulk_message_batch_result(conversation, contact, batch_builder, entries)
-    with_query_timeout_retry do
-      bulk_write_message_entries(conversation, contact, batch_builder, entries)
-    end
-  rescue ActiveRecord::ActiveRecordError
-    fallback_message_entries(conversation, contact, batch_builder, entries) unless @import_stopped
-    nil
-  end
-
-  def fallback_message_entries(conversation, contact, batch_builder, entries)
-    entries.each do |entry|
-      break unless continue_import_with_heartbeat?
-
-      message = fallback_message_entry(conversation, contact, batch_builder, entry)
-      reindex_message_for_search(message) if message.is_a?(Message)
-    end
-  end
-
-  def fallback_message_entry(conversation, contact, batch_builder, entry)
-    with_query_timeout_retry do
-      @data_import.with_lock do
-        if inactive_import_run?
-          @import_stopped = true
+    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
 
-        refreshed_entry = batch_builder.refresh([entry]).entries.first
-        import_message(conversation, contact, refreshed_entry, reindex: false)
-      end
-    end
-  rescue ActiveRecord::ActiveRecordError => e
-    fail_message(conversation, entry.source_id, entry.part, e)
-  end
-
-  def bulk_write_message_entries(conversation, contact, batch_builder, entries)
-    @data_import.with_lock do
-      if inactive_import_run?
-        @import_stopped = true
+        skip_existing_message_mapping(chatwoot_conversation, mapping, part)
         next
       end
 
-      refreshed_entries = batch_builder.refresh(entries).entries
-      persist_message_entries(conversation, contact, refreshed_entries)
+      create_message(chatwoot_conversation, contact, part, message_source_id)
+    rescue StandardError => e
+      fail_message(chatwoot_conversation, message_source_id, part, e)
     end
   end
 
-  def inactive_import_run?
-    @data_import.abandoned? || @data_import.failed? || @data_import.completed? ||
-      @data_import.completed_with_errors? || stale_import_run?
-  end
+  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?
 
-  def persist_message_entries(conversation, contact, entries)
-    grouped_entries = entries.group_by(&:classification)
-    writable_entries = entries.select do |entry|
-      %i[repairable_stale_mapping existing_message new_message].include?(entry.classification)
-    end
-    content_by_source_id = {}
-    attributes_by_source_id = {}
-    failed_entries = []
-    writable_entries.select! do |entry|
-      validate_message_payload!(entry.part)
-      content = content_for(entry.part)
-      content_by_source_id[entry.source_id] = content
-      if content.present? && entry.message.blank?
-        attributes_by_source_id[entry.source_id] = message_attributes(conversation, contact, entry.part, entry.source_id, content)
+    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
-      true
-    rescue InvalidMessagePayloadError => e
-      failed_entries << [entry, e]
-      false
+      record_mapping('message', message_source_id, message, metadata: message_metadata(part))
     end
-    skipped_entries, imported_entries = writable_entries.partition { |entry| content_by_source_id[entry.source_id].blank? }
-    messages = insert_messages(imported_entries, attributes_by_source_id)
-    upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)
-
-    MessageBatchResult.new(
-      imported_entries: imported_entries,
-      skipped_entries: skipped_entries,
-      current_entries: grouped_entries.fetch(:current_import, []),
-      previous_entries: grouped_entries.fetch(:previous_import, []),
-      messages: messages,
-      failed_entries: failed_entries
-    )
-  end
-
-  def insert_messages(entries, attributes_by_source_id)
-    new_entries = entries.reject(&:message)
-    if new_entries.present?
-      attributes = new_entries.map { |entry| attributes_by_source_id.fetch(entry.source_id) }
-      result = Message.insert_all!(attributes, returning: %w[id source_id])
-      inserted_messages = Message.where(id: result.pluck('id')).index_by(&:source_id)
-    end
-
-    entries.map do |entry|
-      entry.message || inserted_messages.fetch("intercom:#{entry.source_id}")
-    end
-  end
-
-  def validate_message_payload!(part)
-    raise InvalidMessagePayloadError, 'Intercom message payload must be an object' unless part.is_a?(Hash)
-
-    %w[author assigned_to event_details].each do |field|
-      value = part[field]
-      next if value.nil? || value.is_a?(Hash)
-
-      raise InvalidMessagePayloadError, "Intercom message #{field} must be an object"
-    end
-
-    participant = part.dig('event_details', 'participant')
-    unless participant.nil? || participant.is_a?(Hash)
-      raise InvalidMessagePayloadError, 'Intercom message event_details.participant must be an object'
-    end
-
-    %w[created_at updated_at].each do |field|
-      value = part[field]
-      valid_timestamp = value.nil? || value.is_a?(Integer) || value.is_a?(Float) ||
-                        (value.is_a?(String) && (value.blank? || value.match?(/\A-?\d+(?:\.\d+)?\z/)))
-      raise InvalidMessagePayloadError, "Intercom message #{field} must be a Unix timestamp" unless valid_timestamp
-
-      timestamp_for(value) if value.present?
-    rescue RangeError
-      raise InvalidMessagePayloadError, "Intercom message #{field} must be a Unix timestamp"
-    end
-
-    %w[body subject].each do |field|
-      value = part[field]
-      next unless value.is_a?(String) && !value.valid_encoding?
-
-      raise InvalidMessagePayloadError, "Intercom message #{field} must use valid encoding"
-    end
-  end
-
-  def upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)
-    now = Time.current
-    mapping_attributes = imported_entries.zip(messages).map do |entry, message|
-      message_mapping_attributes(entry, 'Message', message.id, message_metadata(entry.part), now)
-    end
-    mapping_attributes.concat(skipped_entries.filter_map do |entry|
-      next if entry.mapping
-
-      metadata = message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
-      message_mapping_attributes(entry, 'Conversation', conversation.id, metadata, now)
-    end)
-    return if mapping_attributes.empty?
-
-    DataImportMapping.upsert_all(
-      mapping_attributes,
-      unique_by: MESSAGE_MAPPING_UNIQUE_INDEX,
-      update_only: %i[data_import_id chatwoot_record_type chatwoot_record_id metadata updated_at],
-      record_timestamps: false
-    )
-  end
-
-  def message_mapping_attributes(entry, record_type, record_id, metadata, now)
-    {
-      account_id: @account.id,
-      data_import_id: @data_import.id,
-      source_provider: PROVIDER,
-      source_object_type: 'message',
-      source_object_id: entry.source_id,
-      chatwoot_record_type: record_type,
-      chatwoot_record_id: record_id,
-      metadata: metadata,
-      created_at: entry.mapping&.created_at || now,
-      updated_at: now
-    }
-  end
-
-  def record_bulk_skipped_message(conversation, entry)
-    already_recorded = skip_log_recorded?('message', entry.source_id, SKIPPED_MESSAGE_ERROR_CODE)
-    record_skipped_message_log(conversation, entry.source_id, entry.part)
-    increment_stat('messages', 'skipped') unless already_recorded
-  end
-
-  def reconcile_bulk_message_entry(conversation, entry, &)
-    with_query_timeout_retry(&)
-  rescue StandardError => e
-    fail_message(conversation, entry.source_id, entry.part, e)
-  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, reindex: true)
-    message = with_query_timeout_retry do
-      Message.transaction(requires_new: true) 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}"
-        end
-      end
-    end
-    reindex_message_for_search(message) if reindex && message.is_a?(Message)
-    message
-  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?
-
-    attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
-    message = entry.message
-    unless message
-      result = Message.insert_all!([attrs], returning: %w[id])
-      message = Message.find(result.rows.first.first)
-    end
-    record_message_mapping(entry, message)
     increment_stat('messages', 'imported')
+    reindex_message_for_search(message)
     message
   end
 
@@ -705,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!(
@@ -718,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'])
@@ -782,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)
@@ -907,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)
@@ -915,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
 
@@ -957,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)
@@ -1062,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)
@@ -1210,45 +930,9 @@ class DataImports::Intercom::Importer
     @import_types ||= (@data_import.import_types.presence || DEFAULT_IMPORT_TYPES)
   end
 
-  def increment_stat(group, key, amount = 1)
+  def increment_stat(group, key)
     @stats[group] ||= {}
-    @stats[group][key] = @stats[group][key].to_i + amount
-  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
+    @stats[group][key] = @stats[group][key].to_i + 1
   end
 
   def update_stat_total(group, total)

app/services/data_imports/intercom/message_batch_builder.rb

@@ -1,144 +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)
-    classify(source_entries)
-  end
-
-  def refresh(entries)
-    classify(entries.map do |entry|
-      { source_id: entry.source_id, part: entry.part, position: entry.position }
-    end)
-  end
-
-  def unprepared_entries
-    ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) }
-  end
-
-  private
-
-  def classify(source_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.with_index do |source_entry, position|
-      build_entry(source_entry, source_entry.fetch(:position, position), mappings, messages)
-    end)
-  end
-
-  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

app/services/data_imports/intercom/retry_service.rb

@@ -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

app/services/whatsapp/send_on_whatsapp_service.rb

@@ -6,10 +6,12 @@ class Whatsapp::SendOnWhatsappService < Base::SendOnChannelService
   end
 
   def perform_reply
-    return send_template_message if template_params.present?
-    return send_session_message if message.conversation.can_reply?
-
-    message.update!(status: :failed, external_error: I18n.t('errors.whatsapp.message_outside_messaging_window'))
+    should_send_template_message = template_params.present? || !message.conversation.can_reply?
+    if should_send_template_message
+      send_template_message
+    else
+      send_session_message
+    end
   end
 
   def send_template_message

app/services/whatsapp/webhook_channel_finder_service.rb

@@ -1,35 +0,0 @@
-# Resolves the WhatsApp channel for an inbound WhatsApp Cloud webhook. Meta's
-# display_phone_number can arrive formatted or in a country-specific variant (e.g. Brazil
-# omits the mobile 9, Argentina adds a digit after the country code), so we try the
-# raw digits first and then a normalized fallback, accepting only a candidate whose
-# phone_number_id matches.
-class Whatsapp::WebhookChannelFinderService
-  def initialize(display_phone_number:, phone_number_id:)
-    @display_phone_number = display_phone_number
-    @phone_number_id = phone_number_id
-  end
-
-  def perform
-    return if digits.blank?
-
-    candidates = [
-      Channel::Whatsapp.find_by(phone_number: "+#{digits}"),
-      channel_by_normalized_number
-    ]
-    candidates.compact.find { |channel| channel.provider_config['phone_number_id'] == @phone_number_id }
-  end
-
-  private
-
-  def digits
-    @digits ||= @display_phone_number.to_s.gsub(/[^0-9]/, '')
-  end
-
-  def channel_by_normalized_number
-    normalizer = Whatsapp::PhoneNumberNormalizationService::NORMALIZERS
-                 .lazy.map(&:new).find { |n| n.handles_country?(digits) }
-    return unless normalizer
-
-    Channel::Whatsapp.find_by(phone_number: "+#{normalizer.normalize(digits)}")
-  end
-end

app/views/api/v1/accounts/data_imports/_data_import.json.jbuilder

@@ -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

config/locales/en.yml

@@ -154,7 +154,6 @@ en:
       invalid_token_permissions: 'The access token does not have the required permissions for WhatsApp.'
       phone_info_fetch_failed: 'Failed to fetch phone number information. Please try again.'
       phone_number_already_exists: 'Channel already exists for this phone number: %{phone_number}, please contact support if the error persists'
-      message_outside_messaging_window: 'Message not sent because the WhatsApp 24-hour customer service window is closed and no template parameters were provided. Send an approved template message instead.'
       reauthorization:
         generic: 'Failed to reauthorize WhatsApp. Please try again.'
         not_supported: 'Reauthorization is not supported for this type of WhatsApp channel.'

config/routes.rb

@@ -66,8 +66,7 @@ Rails.application.routes.draw do
             resources :assistants do
               member do
                 post :playground
-                get :metrics
-                get :faq_stats
+                get :stats
                 get :summary
                 get :drilldown
               end
@@ -230,7 +229,6 @@ Rails.application.routes.draw do
             end
             member do
               post :start
-              post :retry, action: :retry_import
               post :abandon
               get :error_logs
               get :skip_logs

enterprise/app/builders/captain/assistant_credit_usage_builder.rb

@@ -0,0 +1,124 @@
+# Computes per-assistant credit consumption (from agent_sessions) for the
+# Captain Overview page: window totals for the trend plus a per-day series for
+# the usage chart.
+#
+# Days are bucketed on the viewer's calendar day, so a completed day's sum never
+# changes. Those sums are cached per (assistant, timezone, date) with a long TTL
+# and only the still-accruing current day is computed live on every request.
+class Captain::AssistantCreditUsageBuilder
+  CACHE_TTL = 30.days
+
+  # Credit tracking on agent_sessions began on this date. Earlier days are still
+  # returned in `daily` but with a nil value ("no data recorded") and are
+  # excluded from the queried span and the window totals.
+  DATA_EPOCH = Date.new(2026, 7, 17)
+
+  def initialize(assistant, window)
+    @assistant = assistant
+    @window = window
+  end
+
+  # => { current:, previous:, daily: [{ date:, value: }, ...] }
+  # `daily` covers the current window only; `previous` is the total for the
+  # preceding equal-length day span, used for the trend. A nil daily value means
+  # the day predates credit tracking.
+  def build
+    sums = daily_sums(recorded_dates)
+
+    {
+      current: total(sums, current_dates),
+      previous: total(sums, previous_dates),
+      daily: current_dates.map { |date| { date: date, value: sums[date] } }
+    }
+  end
+
+  private
+
+  attr_reader :assistant, :window
+
+  # Credit usage is bucketed on calendar days (unlike the rolling timestamp
+  # windows of the other metrics): a day-count range means exactly N calendar
+  # days ending today in the viewer's timezone, and month ranges follow the
+  # window's calendar boundaries.
+  def current_dates
+    @current_dates ||= if day_range?
+                         (today - (window.range.to_i - 1))..today
+                       else
+                         to_dates(window.current)
+                       end
+  end
+
+  # Day ranges derive the comparison span from the current dates (the N days
+  # right before them), since current_dates redefines the window as N calendar
+  # days ending today. Named month ranges follow window.previous so the trend
+  # compares the same periods as every other overview metric.
+  def previous_dates
+    if day_range?
+      (current_dates.first - current_dates.count)..(current_dates.first - 1)
+    else
+      to_dates(window.previous)
+    end
+  end
+
+  def day_range?
+    window.range.match?(/\A\d+\z/)
+  end
+
+  def to_dates(range)
+    range.first.in_time_zone(timezone).to_date..range.last.in_time_zone(timezone).to_date
+  end
+
+  # Pre-epoch days are never queried or cached; they fall out of the sums hash
+  # and surface as nil daily values / zero contribution to totals. While the
+  # previous span is fully pre-epoch this leaves `previous` at 0, so the
+  # frontend hides the trend.
+  def recorded_dates
+    (previous_dates.to_a + current_dates.to_a).select { |date| date >= DATA_EPOCH }
+  end
+
+  def daily_sums(dates)
+    cacheable = dates.select { |date| date < today }
+    cached = Rails.cache.read_multi(*cacheable.map { |date| cache_key(date) })
+
+    missing = dates.reject { |date| cached.key?(cache_key(date)) }
+    live = live_sums(missing)
+    missing.each do |date|
+      Rails.cache.write(cache_key(date), live[date], expires_in: CACHE_TTL) if date < today
+    end
+
+    dates.index_with { |date| cached.fetch(cache_key(date)) { live[date] } }
+  end
+
+  # One grouped scan covering the span of the requested dates. Zero-filled via
+  # groupdate so cache entries exist even for days without sessions.
+  def live_sums(dates)
+    return {} if dates.empty?
+
+    span = dates.min.in_time_zone(timezone).beginning_of_day..dates.max.in_time_zone(timezone).end_of_day
+    Captain::AgentSession
+      .where(assistant_id: assistant.id, created_at: span)
+      .group_by_day(:created_at, time_zone: timezone, range: span)
+      .sum(:credits_consumed)
+      .to_h { |day, value| [day.to_date, (value || 0).round(2)] }
+  end
+
+  def total(sums, dates)
+    dates.sum { |date| sums[date].to_f }.round(2)
+  end
+
+  def today
+    @today ||= Time.current.in_time_zone(timezone).to_date
+  end
+
+  def timezone
+    window.timezone
+  end
+
+  def cache_key(date)
+    "captain_credit_usage/#{assistant.id}/#{timezone_name}/#{date}"
+  end
+
+  def timezone_name
+    @timezone_name ||= timezone.is_a?(String) ? timezone : timezone.name
+  end
+end

enterprise/app/builders/captain/assistant_stats_builder.rb

@@ -37,23 +37,6 @@ class Captain::AssistantStatsBuilder
     build_metrics(current, previous)
   end
 
-  # Approved/pending FAQ counts and the document total in a single round trip.
-  def faq_stats
-    approved, pending, documents = Captain::AssistantResponse.by_assistant(assistant.id).reorder(nil).pick(
-      Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['approved']})"),
-      Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['pending']})"),
-      Arel.sql("(SELECT COUNT(*) FROM captain_documents WHERE assistant_id = #{assistant.id.to_i})")
-    )
-    total = approved + pending
-
-    {
-      approved: approved,
-      pending: pending,
-      documents: documents,
-      coverage: total.zero? ? 0 : (approved.to_f / total * 100).round
-    }
-  end
-
   private
 
   attr_reader :window
@@ -73,7 +56,9 @@ class Captain::AssistantStatsBuilder
       handoff_rate: pack(current[:handoff], previous[:handoff], :point),
       hours_saved: pack(current[:hours_saved], previous[:hours_saved], :percent),
       reopen_rate: pack(current[:reopen], previous[:reopen], :point),
-      conversation_depth: pack(current[:depth], previous[:depth], :absolute)
+      conversation_depth: pack(current[:depth], previous[:depth], :absolute),
+      credit_usage: credit_usage,
+      knowledge: knowledge
     }
   end
 
@@ -89,7 +74,7 @@ class Captain::AssistantStatsBuilder
       auto_resolution: rate(resolution[:resolved], handled),
       handoff: rate(resolution[:handoff], handled),
       hours_saved: (public_count * SECONDS_SAVED_PER_REPLY / 3600.0).round,
-      reopen: reopen_rate(range, resolution[:resolved]),
+      reopen: reopen_rate(range),
       depth: depth_conversations.zero? ? 0 : (public_count.to_f / depth_conversations).round(1)
     }
   end
@@ -174,9 +159,7 @@ class Captain::AssistantStatsBuilder
   # derived from the assistant's handled conversations (not current inbox membership) so a later
   # inbox reassignment doesn't drop historical resolves, and covers both the evaluated (inference)
   # and time-based (bot) resolve paths so the denominator matches auto_resolution_rate.
-  def reopen_rate(range, resolved_count)
-    return 0 if resolved_count.zero?
-
+  def reopen_rate(range)
     resolved_scope = account.reporting_events
                             .where(name: RESOLVED_EVENT_NAMES, created_at: range,
                                    conversation_id: handled_scope(range).select(:conversation_id))
@@ -196,7 +179,31 @@ class Captain::AssistantStatsBuilder
                              'ON resolves.conversation_id = reporting_events.conversation_id ' \
                              'AND reporting_events.event_end_time >= resolves.event_end_time')
                       .distinct.count('reporting_events.conversation_id')
-    rate(reopened, resolved_count)
+    rate(reopened, resolved_scope.distinct.count(:conversation_id))
+  end
+
+  # Credits consumed by the assistant's agent sessions: window totals packed like
+  # the other trend metrics, plus the per-day series for the usage chart.
+  def credit_usage
+    usage = Captain::AssistantCreditUsageBuilder.new(assistant, window).build
+    pack(usage[:current], usage[:previous], :percent).merge(daily: usage[:daily])
+  end
+
+  # Approved/pending FAQ counts and the document total in a single round trip.
+  def knowledge
+    approved, pending, documents = Captain::AssistantResponse.by_assistant(assistant.id).reorder(nil).pick(
+      Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['approved']})"),
+      Arel.sql("COUNT(*) FILTER (WHERE status = #{Captain::AssistantResponse.statuses['pending']})"),
+      Arel.sql("(SELECT COUNT(*) FROM captain_documents WHERE assistant_id = #{assistant.id.to_i})")
+    )
+    total = approved + pending
+
+    {
+      approved: approved,
+      pending: pending,
+      documents: documents,
+      coverage: total.zero? ? 0 : (approved.to_f / total * 100).round
+    }
   end
 
   def rate(numerator, denominator)

enterprise/app/builders/captain/assistant_stats_window.rb

@@ -14,7 +14,7 @@ class Captain::AssistantStatsWindow
   DEFAULT_RANGE = '30'.freeze
   ALLOWED_RANGES = %w[7 30 90 this_month last_month].freeze
 
-  attr_reader :range
+  attr_reader :range, :timezone
 
   def initialize(range = DEFAULT_RANGE, timezone_offset = nil)
     @range = ALLOWED_RANGES.include?(range.to_s) ? range.to_s : DEFAULT_RANGE

enterprise/app/controllers/api/v1/accounts/captain/assistants_controller.rb

@@ -1,7 +1,7 @@
 class Api::V1::Accounts::Captain::AssistantsController < Api::V1::Accounts::BaseController
   before_action -> { check_authorization(Captain::Assistant) }
 
-  before_action :set_assistant, only: [:show, :update, :destroy, :playground, :metrics, :faq_stats, :summary, :drilldown]
+  before_action :set_assistant, only: [:show, :update, :destroy, :playground, :stats, :summary, :drilldown]
 
   def index
     @assistants = account_assistants.ordered
@@ -42,14 +42,10 @@ class Api::V1::Accounts::Captain::AssistantsController < Api::V1::Accounts::Base
     @tools = assistant.available_agent_tools
   end
 
-  def metrics
+  def stats
     render json: Captain::AssistantStatsBuilder.new(@assistant, params[:range], params[:timezone_offset]).metrics
   end
 
-  def faq_stats
-    render json: Captain::AssistantStatsBuilder.new(@assistant).faq_stats
-  end
-
   def summary
     window = Captain::AssistantStatsWindow.new(params[:range], params[:timezone_offset])
     result = cached_or_generated_summary(window, summary_stats)

enterprise/app/controllers/enterprise/api/v1/accounts/agents_controller.rb

@@ -1,8 +1,6 @@
 module Enterprise::Api::V1::Accounts::AgentsController
   def create
     super
-    return if @agent.blank?
-
     associate_agent_with_custom_role
   end
 

enterprise/app/policies/captain/assistant_policy.rb

@@ -7,11 +7,7 @@ class Captain::AssistantPolicy < ApplicationPolicy
     true
   end
 
-  def metrics?
-    true
-  end
-
-  def faq_stats?
+  def stats?
     true
   end
 

package.json

@@ -34,7 +34,7 @@
     "@amplitude/analytics-browser": "^2.11.10",
     "@breezystack/lamejs": "^1.2.7",
     "@chatwoot/ninja-keys": "1.2.3",
-    "@chatwoot/prosemirror-schema": "1.3.23",
+    "@chatwoot/prosemirror-schema": "1.3.22",
     "@chatwoot/utils": "^0.0.56",
     "@formkit/core": "^1.7.2",
     "@formkit/vue": "^1.7.2",
@@ -86,6 +86,9 @@
     "mitt": "^3.0.1",
     "opus-recorder": "^8.0.5",
     "pinia": "^3.0.4",
+    "prosemirror-commands": "^1.7.1",
+    "prosemirror-inputrules": "^1.4.0",
+    "prosemirror-schema-list": "^1.5.1",
     "qrcode": "^1.5.4",
     "semver": "7.6.3",
     "snakecase-keys": "^8.0.1",

pnpm-lock.yaml

@@ -25,8 +25,8 @@ importers:
         specifier: 1.2.3
         version: 1.2.3
       '@chatwoot/prosemirror-schema':
-        specifier: 1.3.23
-        version: 1.3.23
+        specifier: 1.3.22
+        version: 1.3.22
       '@chatwoot/utils':
         specifier: ^0.0.56
         version: 0.0.56
@@ -180,6 +180,15 @@ importers:
       pinia:
         specifier: ^3.0.4
         version: 3.0.4(typescript@5.6.2)(vue@3.5.12(typescript@5.6.2))
+      prosemirror-commands:
+        specifier: ^1.7.1
+        version: 1.7.1
+      prosemirror-inputrules:
+        specifier: ^1.4.0
+        version: 1.4.0
+      prosemirror-schema-list:
+        specifier: ^1.5.1
+        version: 1.5.1
       qrcode:
         specifier: ^1.5.4
         version: 1.5.4
@@ -452,8 +461,8 @@ packages:
   '@chatwoot/ninja-keys@1.2.3':
     resolution: {integrity: sha512-xM8d9P5ikDMZm2WbaCTk/TW5HFauylrU3cJ75fq5je6ixKwyhl/0kZbVN/vbbZN4+AUX/OaSIn6IJbtCgIF67g==}
 
-  '@chatwoot/prosemirror-schema@1.3.23':
-    resolution: {integrity: sha512-jGxbWELCdlVI64BJiE1wT84ekJHYDXXKiluQIKT3aKPEjPwMR48umKF3A0yHjKoR7IIxCC9oM77TvXOA0ebLtw==}
+  '@chatwoot/prosemirror-schema@1.3.22':
+    resolution: {integrity: sha512-0r+PT8xhQLCKCpoV9k9XVTTRECs/0Nr37wbcLsRS7yvc7WkF9FY05z2hGCRJReWmTOcmmshHtb042LVP+MyB/w==}
 
   '@chatwoot/utils@0.0.56':
     resolution: {integrity: sha512-A6dmPLfTSrW4qYNY73btyi4PqpfzcXRSaucscZTQdzNqF6G/QUdgnBmHtho8HeiYby/kSHXaSxLJj+0dx3yEQQ==}
@@ -3992,6 +4001,9 @@ packages:
   prosemirror-tables@1.5.0:
     resolution: {integrity: sha512-VMx4zlYWm7aBlZ5xtfJHpqa3Xgu3b7srV54fXYnXgsAcIGRqKSrhiK3f89omzzgaAgAtDOV4ImXnLKhVfheVNQ==}
 
+  prosemirror-transform@1.10.0:
+    resolution: {integrity: sha512-9UOgFSgN6Gj2ekQH5CTDJ8Rp/fnKR2IkYfGdzzp5zQMFsS4zDllLVx/+jGcX86YlACpG7UR5fwAXiWzxqWtBTg==}
+
   prosemirror-transform@1.12.0:
     resolution: {integrity: sha512-GxboyN4AMIsoHNtz5uf2r2Ru551i5hWeCMD6E2Ib4Eogqoub0NflniaBPVQ4MrGE5yZ8JV9tUHg9qcZTTrcN4w==}
 
@@ -5124,7 +5136,7 @@ snapshots:
       hotkeys-js: 3.8.7
       lit: 2.2.6
 
-  '@chatwoot/prosemirror-schema@1.3.23':
+  '@chatwoot/prosemirror-schema@1.3.22':
     dependencies:
       markdown-it-sup: 2.0.0
       prosemirror-commands: 1.7.1
@@ -9023,7 +9035,7 @@ snapshots:
     dependencies:
       prosemirror-model: 1.22.3
       prosemirror-state: 1.4.3
-      prosemirror-transform: 1.12.0
+      prosemirror-transform: 1.10.0
 
   prosemirror-state@1.4.3:
     dependencies:
@@ -9039,6 +9051,10 @@ snapshots:
       prosemirror-transform: 1.12.0
       prosemirror-view: 1.34.1
 
+  prosemirror-transform@1.10.0:
+    dependencies:
+      prosemirror-model: 1.22.3
+
   prosemirror-transform@1.12.0:
     dependencies:
       prosemirror-model: 1.22.3

spec/builders/agent_builder_spec.rb

@@ -23,12 +23,6 @@ RSpec.describe AgentBuilder, type: :model do
   end
 
   describe '#perform' do
-    it 'locks the account while checking and creating the agent' do
-      expect(account).to receive(:with_lock).and_call_original
-
-      agent_builder.perform
-    end
-
     context 'when user does not exist' do
       it 'creates a new user' do
         expect { agent_builder.perform }.to change(User, :count).by(1)
@@ -73,17 +67,5 @@ RSpec.describe AgentBuilder, type: :model do
         expect(user.encrypted_password).not_to be_empty
       end
     end
-
-    context 'when the account has reached its agent limit' do
-      before do
-        allow(account).to receive(:usage_limits).and_return({ agents: account.account_users.count })
-      end
-
-      it 'raises a limit exceeded error without creating a user' do
-        expect { agent_builder.perform }.to raise_error(described_class::LimitExceededError, described_class::LIMIT_EXCEEDED_MESSAGE)
-
-        expect(User.from_email(email)).to be_nil
-      end
-    end
   end
 end

spec/controllers/api/v1/accounts/integrations/linear_controller_spec.rb

@@ -13,9 +13,7 @@ RSpec.describe 'Linear Integration API', type: :request do
   end
 
   describe 'DELETE /api/v1/accounts/:account_id/integrations/linear' do
-    let(:admin) { create(:user, account: account, role: :administrator) }
-
-    it 'deletes the linear integration when the user is an administrator' do
+    it 'deletes the linear integration' do
       # Stub the HTTP call to Linear's revoke endpoint
       allow(HTTParty).to receive(:post).with(
         'https://api.linear.app/oauth/revoke',
@@ -23,19 +21,11 @@ RSpec.describe 'Linear Integration API', type: :request do
       ).and_return(instance_double(HTTParty::Response, success?: true))
 
       delete "/api/v1/accounts/#{account.id}/integrations/linear",
-             headers: admin.create_new_auth_token,
+             headers: agent.create_new_auth_token,
              as: :json
       expect(response).to have_http_status(:ok)
       expect(account.hooks.count).to eq(0)
     end
-
-    it 'returns unauthorized for an agent and keeps the integration' do
-      delete "/api/v1/accounts/#{account.id}/integrations/linear",
-             headers: agent.create_new_auth_token,
-             as: :json
-      expect(response).to have_http_status(:unauthorized)
-      expect(account.hooks.count).to eq(1)
-    end
   end
 
   describe 'GET /api/v1/accounts/:account_id/integrations/linear/teams' do

spec/controllers/api/v1/accounts/integrations/shopify_controller_spec.rb

@@ -159,33 +159,19 @@ RSpec.describe 'Shopify Integration API', type: :request do
   end
 
   describe 'DELETE /api/v1/accounts/:account_id/integrations/shopify' do
-    let(:admin) { create(:user, account: account, role: :administrator) }
-
     before do
       create(:integrations_hook, :shopify, account: account)
     end
 
-    context 'when it is an administrator' do
+    context 'when it is an authenticated user' do
       it 'deletes the shopify integration' do
-        expect do
-          delete "/api/v1/accounts/#{account.id}/integrations/shopify",
-                 headers: admin.create_new_auth_token,
-                 as: :json
-        end.to change { account.hooks.count }.by(-1)
-
-        expect(response).to have_http_status(:ok)
-      end
-    end
-
-    context 'when it is an agent' do
-      it 'returns unauthorized and keeps the integration' do
         expect do
           delete "/api/v1/accounts/#{account.id}/integrations/shopify",
                  headers: agent.create_new_auth_token,
                  as: :json
-        end.not_to(change { account.hooks.count })
+        end.to change { account.hooks.count }.by(-1)
 
-        expect(response).to have_http_status(:unauthorized)
+        expect(response).to have_http_status(:ok)
       end
     end
 

spec/enterprise/builders/captain/assistant_stats_builder_spec.rb

@@ -27,7 +27,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
 
       expect(metrics.keys).to contain_exactly(
         :conversations_handled, :auto_resolution_rate, :handoff_rate,
-        :hours_saved, :reopen_rate, :conversation_depth
+        :hours_saved, :reopen_rate, :conversation_depth, :credit_usage, :knowledge
       )
       expect(metrics[:conversations_handled]).to include(:current, :previous, :trend)
     end
@@ -229,7 +229,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
     end
   end
 
-  describe '#faq_stats' do
+  describe '#metrics knowledge' do
     before do
       create_list(:captain_assistant_response, 3, assistant: assistant, account: account, status: :approved)
       create(:captain_assistant_response, assistant: assistant, account: account, status: :pending)
@@ -237,7 +237,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
     end
 
     it 'returns approved, pending, document counts and coverage' do
-      knowledge = described_class.new(assistant).faq_stats
+      knowledge = described_class.new(assistant, '30').metrics[:knowledge]
 
       expect(knowledge).to eq(approved: 3, pending: 1, documents: 2, coverage: 75)
     end
@@ -245,7 +245,7 @@ RSpec.describe Captain::AssistantStatsBuilder do
     it 'reports zero coverage when there are no responses' do
       Captain::AssistantResponse.where(assistant: assistant).delete_all
 
-      knowledge = described_class.new(assistant).faq_stats
+      knowledge = described_class.new(assistant, '30').metrics[:knowledge]
 
       expect(knowledge[:coverage]).to eq(0)
     end

spec/enterprise/controllers/api/v1/accounts/agents_controller_spec.rb

@@ -21,27 +21,6 @@ RSpec.describe 'Agents API', type: :request do
         expect(response).to have_http_status(:payment_required)
         expect(response.body).to include('Account limit exceeded. Please purchase more licenses')
       end
-
-      it 'prevents adding an agent if the last seat is consumed before creation' do
-        account.update!(limits: { agents: account.account_users.count + 1 })
-        competing_agent_created = false
-
-        allow(AgentBuilder).to receive(:new).and_wrap_original do |method, *args|
-          unless competing_agent_created
-            create(:user, account: account, role: :agent)
-            competing_agent_created = true
-          end
-
-          method.call(*args)
-        end
-
-        post "/api/v1/accounts/#{account.id}/agents", params: params, headers: admin.create_new_auth_token, as: :json
-
-        expect(response).to have_http_status(:payment_required)
-        expect(response.body).to include('Account limit exceeded. Please purchase more licenses')
-        expect(User.from_email(params[:email])).to be_nil
-        expect(account.account_users.count).to eq(account.usage_limits[:agents])
-      end
     end
   end
 

spec/enterprise/policies/captain/assistant_policy_spec.rb

@@ -12,7 +12,7 @@ RSpec.describe Captain::AssistantPolicy, type: :policy do
   let(:administrator_context) { { user: administrator, account: account, account_user: account.account_users.first } }
   let(:agent_context) { { user: agent, account: account, account_user: account.account_users.first } }
 
-  permissions :index?, :show?, :playground?, :metrics?, :faq_stats? do
+  permissions :index?, :show?, :playground? do
     context 'when administrator' do
       it { expect(assistant_policy).to permit(administrator_context, assistant) }
     end

spec/jobs/webhooks/whatsapp_events_job_spec.rb

@@ -345,94 +345,6 @@ RSpec.describe Webhooks::WhatsappEventsJob do
       end.not_to change(Conversation, :count)
     end
 
-    it 'finds channel using normalized Brazil phone number when display_phone_number is missing the 9 digit' do
-      brazil_channel = create(:channel_whatsapp, phone_number: '+5541999887766', provider: 'whatsapp_cloud',
-                                                 sync_templates: false, validate_provider_config: false)
-      wb_params = {
-        object: 'whatsapp_business_account',
-        entry: [{
-          changes: [{
-            value: {
-              metadata: {
-                phone_number_id: brazil_channel.provider_config['phone_number_id'],
-                display_phone_number: '554199887766'
-              }
-            }
-          }]
-        }]
-      }
-      allow(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).and_return(process_service)
-      expect(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).with(inbox: brazil_channel.inbox, params: wb_params)
-      job.perform_now(wb_params)
-    end
-
-    it 'finds channel using normalized Argentina phone number when display_phone_number has extra 9 digit' do
-      argentina_channel = create(:channel_whatsapp, phone_number: '+541112345678', provider: 'whatsapp_cloud',
-                                                    sync_templates: false, validate_provider_config: false)
-      wb_params = {
-        object: 'whatsapp_business_account',
-        entry: [{
-          changes: [{
-            value: {
-              metadata: {
-                phone_number_id: argentina_channel.provider_config['phone_number_id'],
-                display_phone_number: '5491112345678'
-              }
-            }
-          }]
-        }]
-      }
-      allow(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).and_return(process_service)
-      expect(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).with(inbox: argentina_channel.inbox, params: wb_params)
-      job.perform_now(wb_params)
-    end
-
-    it 'finds channel when display_phone_number contains formatting characters' do
-      formatted_channel = create(:channel_whatsapp, phone_number: '+14155552671', provider: 'whatsapp_cloud',
-                                                    sync_templates: false, validate_provider_config: false)
-      wb_params = {
-        object: 'whatsapp_business_account',
-        entry: [{
-          changes: [{
-            value: {
-              metadata: {
-                phone_number_id: formatted_channel.provider_config['phone_number_id'],
-                display_phone_number: '+1 415-555-2671'
-              }
-            }
-          }]
-        }]
-      }
-      allow(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).and_return(process_service)
-      expect(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).with(inbox: formatted_channel.inbox, params: wb_params)
-      job.perform_now(wb_params)
-    end
-
-    it 'prefers the phone_number_id match when a raw display_phone_number collision exists' do
-      normalized_channel = create(:channel_whatsapp, phone_number: '+5541999887766', provider: 'whatsapp_cloud',
-                                                     sync_templates: false, validate_provider_config: false)
-      create(:channel_whatsapp, phone_number: '+554199887766', provider: 'whatsapp_cloud',
-                                sync_templates: false, validate_provider_config: false).tap do |raw_channel|
-        raw_channel.update!(provider_config: raw_channel.provider_config.merge('phone_number_id' => 'other-id'))
-      end
-      wb_params = {
-        object: 'whatsapp_business_account',
-        entry: [{
-          changes: [{
-            value: {
-              metadata: {
-                phone_number_id: normalized_channel.provider_config['phone_number_id'],
-                display_phone_number: '554199887766'
-              }
-            }
-          }]
-        }]
-      }
-      allow(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).and_return(process_service)
-      expect(Whatsapp::IncomingMessageWhatsappCloudService).to receive(:new).with(inbox: normalized_channel.inbox, params: wb_params)
-      job.perform_now(wb_params)
-    end
-
     it 'will not enque Whatsapp::IncomingMessageWhatsappCloudService when invalid phone number id' do
       other_channel = create(:channel_whatsapp, phone_number: '+1987654', provider: 'whatsapp_cloud', sync_templates: false,
                                                 validate_provider_config: false)

spec/models/data_import_spec.rb

@@ -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

spec/requests/api/v1/accounts/data_imports_spec.rb

@@ -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
       )

spec/services/data_imports/intercom/importer_spec.rb

@@ -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 }
@@ -116,275 +116,6 @@ RSpec.describe DataImports::Intercom::Importer do
     expect(DataImportMapping.where(data_import: data_import).count).to eq(5)
   end
 
-  it 'writes a normal conversation with one message insert and one mapping upsert', :aggregate_failures do
-    importer = described_class.new(data_import: data_import)
-    message_batch_sizes = []
-    mapping_batch_sizes = []
-    mapping_upsert_options = []
-    allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
-      message_batch_sizes << records.size
-      method.call(records, **kwargs)
-    end
-    allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
-      mapping_batch_sizes << records.size
-      mapping_upsert_options << kwargs
-      method.call(records, **kwargs)
-    end
-    expect(importer).not_to receive(:create_message)
-
-    importer.perform
-
-    expect(message_batch_sizes).to eq([3])
-    expect(mapping_batch_sizes).to eq([3])
-    expect(mapping_upsert_options).to contain_exactly(
-      include(unique_by: described_class::MESSAGE_MAPPING_UNIQUE_INDEX, record_timestamps: false)
-    )
-    expect(account.messages.count).to eq(3)
-    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-  end
-
-  it 'preserves provider order when imported messages share a timestamp' do
-    equal_timestamp_conversation = conversation_payload.deep_dup
-    equal_timestamp_conversation['conversation_parts']['conversation_parts'].each do |part|
-      part['created_at'] = conversation_payload['created_at']
-      part['updated_at'] = conversation_payload['created_at']
-    end
-    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(equal_timestamp_conversation)
-
-    described_class.new(data_import: data_import).perform
-
-    conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
-    expect(conversation.messages.order(:created_at, :id).pluck(:source_id)).to eq(
-      %w[
-        intercom:conversation:conversation_1:source:source_1
-        intercom:conversation:conversation_1:part:part_1
-        intercom:conversation:conversation_1:part:part_2
-      ]
-    )
-  end
-
-  it 'writes conversations in batches of at most 100 messages', :aggregate_failures do
-    bulk_conversation = conversation_payload.deep_dup
-    template_part = bulk_conversation.dig('conversation_parts', 'conversation_parts').first
-    bulk_conversation['conversation_parts']['conversation_parts'] = Array.new(205) do |index|
-      template_part.merge(
-        'id' => "part_#{index + 1}",
-        'created_at' => 1_700_000_100 + index,
-        'updated_at' => 1_700_000_100 + index
-      )
-    end
-    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(bulk_conversation)
-    message_batch_sizes = []
-    mapping_batch_sizes = []
-    allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
-      message_batch_sizes << records.size
-      method.call(records, **kwargs)
-    end
-    allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
-      mapping_batch_sizes << records.size
-      method.call(records, **kwargs)
-    end
-    importer = described_class.new(data_import: data_import)
-    expect(importer).to receive(:update_conversation_activity).once.and_call_original
-
-    importer.import_conversations_page
-
-    expect(message_batch_sizes).to eq([100, 100, 6])
-    expect(mapping_batch_sizes).to eq([100, 100, 6])
-    expect(account.messages.order(:created_at).pick(:source_id)).to eq('intercom:conversation:conversation_1:source:source_1')
-    expect(account.messages.count).to eq(206)
-    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(206)
-  end
-
-  context 'when a bulk message chunk fails' do
-    it 'retries a query timeout once and keeps the successful bulk result', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs|
-        mapping_attempts += 1
-        raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
-
-        method.call(records, **kwargs)
-      end
-      expect(importer).not_to receive(:fallback_message_entries)
-
-      importer.perform
-
-      expect(mapping_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(account.messages.count).to eq(3)
-      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-    end
-
-    it 'falls back individually after the query timeout retry is exhausted', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      reindex_transaction_depths = []
-      transaction_depth_before_import = Message.connection.open_transactions
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:fallback_message_entries).and_call_original
-      allow(importer).to receive(:create_message).and_call_original
-      allow(importer).to receive(:reindex_message_for_search).and_wrap_original do |method, message|
-        reindex_transaction_depths << Message.connection.open_transactions
-        method.call(message)
-      end
-      allow(DataImportMapping).to receive(:upsert_all) do
-        mapping_attempts += 1
-        raise ActiveRecord::QueryCanceled, 'statement timeout'
-      end
-
-      importer.perform
-
-      expect(mapping_attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(importer).to have_received(:fallback_message_entries).once
-      expect(importer).to have_received(:create_message).exactly(3).times
-      expect(account.messages.count).to eq(3)
-      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-      expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import))
-    end
-
-    it 'falls back immediately for a non-timeout database error', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      mapping_attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:fallback_message_entries).and_call_original
-      allow(DataImportMapping).to receive(:upsert_all) do
-        mapping_attempts += 1
-        raise ActiveRecord::StatementInvalid, 'bulk mapping failed'
-      end
-
-      importer.perform
-
-      expect(mapping_attempts).to eq(1)
-      expect(importer).not_to have_received(:sleep)
-      expect(importer).to have_received(:fallback_message_entries).once
-      expect(account.messages.count).to eq(3)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-    end
-
-    it 'refreshes fallback entries when another worker repairs a message', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
-      allow(importer).to receive(:fallback_message_entries).and_wrap_original do |method, conversation, contact, batch_builder, entries|
-        source_entry = entries.first
-        message = create(
-          :message,
-          account: account,
-          inbox: conversation.inbox,
-          conversation: conversation,
-          source_id: "intercom:#{source_entry.source_id}",
-          created_at: Time.zone.at(source_entry.part['created_at']),
-          updated_at: Time.zone.at(source_entry.part['created_at'])
-        )
-        DataImportMapping.create!(
-          account: account,
-          data_import: data_import,
-          source_provider: 'intercom',
-          source_object_type: 'message',
-          source_object_id: source_entry.source_id,
-          chatwoot_record_type: 'Message',
-          chatwoot_record_id: message.id,
-          metadata: {}
-        )
-        method.call(conversation, contact, batch_builder, entries)
-      end
-
-      importer.perform
-
-      source_id = 'intercom:conversation:conversation_1:source:source_1'
-      expect(account.messages.where(source_id: source_id).count).to eq(1)
-      expect(account.messages.count).to eq(3)
-      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-    end
-
-    it 'retries a fallback refresh timeout after the lock transaction rolls back', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      refresh_attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
-      allow(DataImports::Intercom::MessageBatchBuilder).to receive(:new).and_wrap_original do |method, **kwargs|
-        method.call(**kwargs).tap do |batch_builder|
-          allow(batch_builder).to receive(:refresh).and_wrap_original do |refresh, entries|
-            refresh_attempts += 1
-            raise ActiveRecord::QueryCanceled, 'statement timeout' if refresh_attempts == 1
-
-            refresh.call(entries)
-          end
-        end
-      end
-
-      importer.perform
-
-      expect(refresh_attempts).to eq(4)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(account.messages.count).to eq(3)
-      expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3)
-      expect(data_import.import_errors).to be_empty
-      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
-    end
-  end
-
-  it 'isolates a malformed message payload while importing valid messages', :aggregate_failures do
-    malformed_conversation = conversation_payload.deep_dup
-    malformed_conversation['source']['subject'] = nil
-    malformed_conversation['source']['body'] = nil
-    malformed_conversation.dig('conversation_parts', 'conversation_parts').first['created_at'] = { 'unexpected' => true }
-    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(malformed_conversation)
-    importer = described_class.new(data_import: data_import)
-    expect(importer).not_to receive(:fallback_message_entries)
-
-    importer.perform
-
-    expect(account.messages.pluck(:source_id)).to eq(['intercom:conversation:conversation_1:part:part_2'])
-    error = data_import.import_errors.find_by!(
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1'
-    )
-    expect(error).to have_attributes(
-      error_code: described_class::InvalidMessagePayloadError.name,
-      message: 'Intercom message created_at must be a Unix timestamp'
-    )
-    expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1)
-    expect(data_import.reload).to be_completed_with_errors
-    expect(data_import.stats.dig('messages', 'imported')).to eq(1)
-    expect(data_import.stats.dig('errors', 'count')).to eq(1)
-  end
-
-  it 'isolates an out-of-range message timestamp while importing valid messages', :aggregate_failures do
-    malformed_conversation = conversation_payload.deep_dup
-    malformed_conversation['source']['subject'] = nil
-    malformed_conversation['source']['body'] = nil
-    malformed_conversation.dig('conversation_parts', 'conversation_parts').first['created_at'] = Float::INFINITY
-    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(malformed_conversation)
-    importer = described_class.new(data_import: data_import)
-    expect(importer).not_to receive(:fallback_message_entries)
-
-    importer.perform
-
-    expect(account.messages.pluck(:source_id)).to eq(['intercom:conversation:conversation_1:part:part_2'])
-    error = data_import.import_errors.find_by!(
-      source_object_type: 'message',
-      source_object_id: 'conversation:conversation_1:part:part_1'
-    )
-    expect(error).to have_attributes(
-      error_code: described_class::InvalidMessagePayloadError.name,
-      message: 'Intercom message created_at must be a Unix timestamp'
-    )
-    expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1)
-    expect(data_import.reload).to be_completed_with_errors
-    expect(data_import.stats.dig('messages', 'imported')).to eq(1)
-    expect(data_import.stats.dig('errors', 'count')).to eq(1)
-  end
-
   it 'imports historical records without dispatching record events or outbound side effects', :aggregate_failures do
     dispatched_events = []
     allow(Rails.configuration.dispatcher).to receive(:dispatch) do |event_name, *_args|
@@ -457,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
 
@@ -507,19 +203,13 @@ RSpec.describe DataImports::Intercom::Importer do
     allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true)
     allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false)
     reindexed_message_ids = []
-    reindex_transaction_depths = []
-    transaction_depth_before_import = Message.connection.open_transactions
     original_reindex_for_search = Message.instance_method(:reindex_for_search)
-    Message.define_method(:reindex_for_search) do
-      reindexed_message_ids << id
-      reindex_transaction_depths << self.class.connection.open_transactions
-    end
+    Message.define_method(:reindex_for_search) { reindexed_message_ids << id }
     Message.__send__(:private, :reindex_for_search)
 
     described_class.new(data_import: data_import).perform
 
     expect(reindexed_message_ids).to match_array(Message.where(account_id: account.id).pluck(:id))
-    expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import))
   ensure
     Message.define_method(:reindex_for_search, original_reindex_for_search)
     Message.__send__(:private, :reindex_for_search)
@@ -577,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' } }
       )
@@ -595,165 +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 message batches' 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(400) 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)
-        empty_result = described_class::MessageBatchResult.new(
-          imported_entries: [],
-          skipped_entries: [],
-          current_entries: [],
-          previous_entries: [],
-          messages: []
-        )
-        allow(importer).to receive(:bulk_write_message_entries) do
-          travel 30.seconds
-          empty_result
-        end
-
-        importer.import_conversations_page
-
-        expect(heartbeat_times).to eq([started_at + 1.minute, started_at + 2.minutes])
-        expect(importer).to have_received(:bulk_write_message_entries).exactly(5).times
-      end
-    end
-
-    it 'stops batches 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 })
-        long_conversation = conversation_payload.deep_dup
-        template_part = long_conversation.dig('conversation_parts', 'conversation_parts').first
-        long_conversation['conversation_parts']['conversation_parts'] = Array.new(100) do |index|
-          template_part.merge('id' => "part_#{index + 1}")
-        end
-        allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(long_conversation)
-        allow(data_import).to receive(:touch).and_call_original
-        importer = described_class.new(data_import: data_import, run_id: run_id)
-        batch_write_count = 0
-        allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args|
-          result = method.call(*args)
-          batch_write_count += 1
-          if batch_write_count == 1
-            DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-          end
-          result
-        end
-
-        importer.import_conversations_page
-
-        expect(importer).to have_received(:bulk_write_message_entries).once
-        expect(data_import).not_to have_received(:touch)
-        expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0)
-        expect(account.messages.count).to eq(100)
-      end
-    end
-
-    it 'does not persist stale stats when a newer run takes over during the final batch', :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(:bulk_write_message_entries).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          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 'does not persist stale stats when a newer run takes over during the final prefetch fallback entry', :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(:sleep)
-      allow(DataImports::Intercom::MessageBatchBuilder).to receive(:new).and_wrap_original do |method, **kwargs|
-        method.call(**kwargs).tap do |batch_builder|
-          allow(batch_builder).to receive(:perform).and_wrap_original do |perform, *args|
-            raise ActiveRecord::QueryCanceled, 'statement timeout' if args.empty?
-
-            perform.call(*args)
-          end
-        end
-      end
-      allow(importer).to receive(:import_unprepared_message).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          source_entry = args.last
-          next unless source_entry.dig(:part, 'id') == 'part_2'
-
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-        end
-      end
-      expect(importer).not_to receive(:update_conversation_activity)
-
-      importer.import_conversations_page
-
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(importer).to have_received(:import_unprepared_message).exactly(3).times
-      expect(account.messages.count).to eq(3)
-      expect(data_import.reload.stats).to include(
-        'conversations' => include('imported' => 0),
-        'messages' => include('imported' => 0)
-      )
-    end
-
-    it 'stops individual fallback entries when a newer 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 })
-      importer = described_class.new(data_import: data_import, run_id: run_id)
-      allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed')
-      allow(importer).to receive(:fallback_message_entry).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
-        end
-      end
-
-      importer.import_conversations_page
-
-      expect(importer).to have_received(:fallback_message_entry).once
-      expect(account.messages.count).to eq(1)
-      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(:bulk_write_message_entries).and_wrap_original do |method, *args|
-          method.call(*args).tap do
-            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('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:|
@@ -791,11 +322,10 @@ 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(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk mapping failed')
-      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
-        raise StandardError, 'mapping failed' if entry.source_id == 'conversation:conversation_1:source:source_1'
+      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(entry, message)
+        method.call(object_type, source_id, record, metadata: metadata)
       end
 
       importer.import_conversations_page
@@ -808,81 +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(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable')
-      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(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk unavailable')
-      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
@@ -918,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
@@ -969,60 +404,6 @@ RSpec.describe DataImports::Intercom::Importer do
         'message' => 3
       )
       expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
-      expect(
-        DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message').distinct.pluck(:data_import_id)
-      ).to eq([data_import.id])
-    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
@@ -1048,34 +429,6 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(next_data_import.import_errors.skip_logs.where(source_object_type: 'message')).to be_empty
       message_mappings = DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message')
       expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3)
-      expect(message_mappings.distinct.pluck(:data_import_id)).to eq([next_data_import.id])
-    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
@@ -1140,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
@@ -1392,42 +741,6 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1)
     end
 
-    it 'retries skip-log reconciliation after a query timeout', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      attempts = 0
-      allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_skipped_message_log).and_wrap_original do |method, *args|
-        attempts += 1
-        raise ActiveRecord::QueryCanceled, 'statement timeout' if attempts == 1
-
-        method.call(*args)
-      end
-
-      importer.perform
-
-      expect(attempts).to eq(2)
-      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
-      expect(data_import.reload).to be_completed
-      expect(data_import.stats.dig('messages', 'skipped')).to eq(1)
-    end
-
-    it 'isolates a persistent skip-log reconciliation failure to the message', :aggregate_failures do
-      importer = described_class.new(data_import: data_import)
-      allow(importer).to receive(:record_skipped_message_log).and_raise(ActiveRecord::StatementInvalid, 'skip log failed')
-
-      importer.perform
-
-      item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1')
-      error = data_import.import_errors.find_by!(
-        source_object_type: 'message',
-        source_object_id: 'conversation:conversation_1:part:blank_part'
-      )
-      expect(item).to be_imported
-      expect(error).to have_attributes(error_code: 'ActiveRecord::StatementInvalid', message: 'skip log failed')
-      expect(data_import.reload).to be_completed_with_errors
-      expect(data_import.stats.dig('errors', 'count')).to eq(1)
-    end
-
     it 'records the skip log again for a later import run', :aggregate_failures do
       described_class.new(data_import: data_import).perform
       next_data_import = create(
@@ -1563,29 +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(:bulk_write_message_entries).and_wrap_original do |method, *args|
-        method.call(*args).tap do
-          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
@@ -1632,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' => {
@@ -1654,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.any? { |record| record[:source_id] == 'intercom:conversation:conversation_1:part:bad_part' }
-          insert_attempts << 'intercom:conversation:conversation_1:part:bad_part'
-          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
@@ -1678,9 +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.size).to eq(2)
-      expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1)
-      expect(account.messages.find_by(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_present
     end
   end
 end

spec/services/data_imports/intercom/message_batch_builder_spec.rb

@@ -1,205 +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 'refreshes classifications while preserving source positions' do
-    batch = builder.perform
-    target_entry = batch.entries.second
-    message = create(
-      :message,
-      account: account,
-      conversation: conversation,
-      inbox: conversation.inbox,
-      source_id: "intercom:#{target_entry.source_id}"
-    )
-    DataImportMapping.create!(
-      account: account,
-      data_import: data_import,
-      source_provider: 'intercom',
-      source_object_type: 'message',
-      source_object_id: target_entry.source_id,
-      chatwoot_record_type: 'Message',
-      chatwoot_record_id: message.id,
-      metadata: {}
-    )
-
-    refreshed_batch = builder.refresh(batch.entries)
-
-    expect(refreshed_batch.entries.map(&:position)).to eq([0, 1, 2])
-    expect(refreshed_batch.entries.second).to have_attributes(classification: :current_import, message: message)
-  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

spec/services/data_imports/intercom/retry_service_spec.rb

@@ -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

spec/services/whatsapp/send_on_whatsapp_service_spec.rb

@@ -72,21 +72,6 @@ describe Whatsapp::SendOnWhatsappService do
         expect(message.reload.source_id).to eq('123456789')
       end
 
-      it 'fails a free-form message without contacting the provider when outside the 24 hour limit' do
-        create(:message, message_type: :incoming, content: 'test', created_at: 25.hours.ago,
-                         conversation: conversation, account: conversation.account)
-        message = create(:message, message_type: :outgoing, content: 'test',
-                                   conversation: conversation, account: conversation.account)
-
-        expect(Whatsapp::TemplateProcessorService).not_to receive(:new)
-
-        described_class.new(message: message).perform
-
-        expect(message.reload.status).to eq('failed')
-        expect(message.external_error).to eq(I18n.t('errors.whatsapp.message_outside_messaging_window'))
-        expect(a_request(:post, 'https://waba.360dialog.io/v1/messages')).not_to have_been_made
-      end
-
       it 'marks message as failed when template name is blank' do
         processor = instance_double(Whatsapp::TemplateProcessorService)
         allow(Whatsapp::TemplateProcessorService).to receive(:new).and_return(processor)