Merge branch 'codex/cw-7615-intercom-message-batches' into codex/cw-7615-intercom-bulk-message-writes

# Conflicts: # app/services/data_imports/intercom/importer.rb # app/services/data_imports/intercom/message_batch_builder.rb # spec/services/data_imports/intercom/importer_spec.rb
2026-07-23 14:44:36 +05:30
parent 0410211e23 d238df2375
commit 5d479d36e1
3 changed files with 184 additions and 30 deletions
@@ -173,6 +173,7 @@ 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)

      update_conversation_activity(mapped_conversation)
@@ -184,7 +185,11 @@ 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)
-    increment_stat('conversations', 'imported') unless already_handled
+    if already_handled
+      reconcile_item_stats('conversation')
+    else
+      increment_stat('conversations', 'imported')
+    end

    return unless import_conversation_messages(conversation, chatwoot_conversation, contact)

@@ -205,8 +210,8 @@ class DataImports::Intercom::Importer
  end

  def continue_import_with_heartbeat?
-    return true if @data_import.updated_at > HEARTBEAT_INTERVAL.ago
    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
@@ -408,7 +413,12 @@ class DataImports::Intercom::Importer
      conversation: chatwoot_conversation,
      source_conversation: conversation
    )
-    batch = with_query_timeout_retry { batch_builder.perform }
+    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|
@@ -417,6 +427,8 @@ class DataImports::Intercom::Importer
      import_message_batch(chatwoot_conversation, contact, batch_builder, entries)
      return false if @import_stopped
    end
+    return false if import_stopped?
+
    true
  end

@@ -535,6 +547,26 @@ class DataImports::Intercom::Importer
    increment_stat('messages', 'skipped') unless already_recorded
  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
+    true
+  end
+
+  def import_unprepared_message(conversation, contact, batch_builder, source_entry)
+    entry = with_query_timeout_retry { batch_builder.perform([source_entry]).entries.first }
+    import_message(conversation, contact, entry)
+  rescue StandardError => e
+    fail_message(conversation, source_entry[:source_id], source_entry[:part], e)
+  end
+
  def import_message(conversation, contact, entry)
    with_query_timeout_retry do
      case entry.classification
@@ -804,19 +836,21 @@ class DataImports::Intercom::Importer
  end

  def skip_already_imported_item(item, mapping, already_handled:)
-    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
-    )
+    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
    increment_stat(stat_group_for(item.source_object_type), 'skipped') unless already_handled
  end

@@ -1088,17 +1122,19 @@ class DataImports::Intercom::Importer
  def reconcile_dirty_stats
    return if @dirty_stat_groups.empty?

-    @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}"
+    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
-    persist_stats
    @dirty_stat_groups.clear
  end

@@ -38,8 +38,8 @@ class DataImports::Intercom::MessageBatchBuilder
    @source_conversation = source_conversation
  end

-  def perform
-    classify(ordered_source_entries)
+  def perform(source_entries = unprepared_entries)
+    classify(source_entries)
  end

  def refresh(entries)
@@ -48,6 +48,10 @@ class DataImports::Intercom::MessageBatchBuilder
    end)
  end

+  def unprepared_entries
+    ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) }
+  end
+
  private

  def classify(source_entries)
@@ -112,7 +116,7 @@ class DataImports::Intercom::MessageBatchBuilder
  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]
+    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(
@@ -292,6 +292,28 @@ RSpec.describe DataImports::Intercom::Importer do
    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
+
  it 'indexes imported messages for advanced search' do
    allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true)
    allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false)
@@ -434,19 +456,62 @@ RSpec.describe DataImports::Intercom::Importer do
        allow(importer).to receive(:bulk_write_message_entries).and_wrap_original do |method, *args|
          result = method.call(*args)
          batch_write_count += 1
-          data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) if 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).twice
+        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 '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:|
@@ -667,6 +732,55 @@ RSpec.describe DataImports::Intercom::Importer do
      ).to eq([data_import.id])
    end

+    it 'reconciles skipped conversation stats when a superseded run is retried' 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 })
+
+      freeze_time do
+        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)
+          next_data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => next_run_id })
+          travel 1.minute
+        end
+
+        importer.import_conversations_page
+        expect(next_data_import.reload.stats.dig('conversations', '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)
+      end
+    end
+
+    it 'keeps skipped contact stats when a query timeout is retried after the item update', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+      importer = described_class.new(data_import: next_data_import)
+      contact_log_attempts = 0
+      allow(importer).to receive(:sleep)
+      allow(importer).to receive(:record_already_imported_log).and_wrap_original do |method, **attributes|
+        if attributes[:source_object_type] == 'contact'
+          contact_log_attempts += 1
+          raise ActiveRecord::QueryCanceled, 'statement timeout' if contact_log_attempts == 1
+        end
+
+        method.call(**attributes)
+      end
+
+      importer.import_contacts_page
+
+      contact_item = next_data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(contact_log_attempts).to eq(2)
+      expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once
+      expect(contact_item).to be_skipped
+      expect(next_data_import.reload.stats.dig('contacts', 'skipped')).to eq(1)
+      expect(next_data_import.import_errors.skip_logs.exists?(data_import_item: contact_item)).to be(true)
+    end
+
    it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do
      described_class.new(data_import: data_import).perform
      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')