diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb
index 88ca3df17..385df025b 100644
--- a/app/services/data_imports/intercom/importer.rb
+++ b/app/services/data_imports/intercom/importer.rb
@@ -1,5 +1,7 @@
 # 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?
@@ -27,6 +29,7 @@ class DataImports::Intercom::Importer
     :current_entries,
     :previous_entries,
     :messages,
+    :failed_entries,
     keyword_init: true
   )
 
@@ -450,19 +453,43 @@ class DataImports::Intercom::Importer
     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)
-    bulk_write_message_entries(conversation, contact, batch_builder, entries)
-  rescue StandardError
+    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?
 
-      import_message(conversation, contact, entry)
+      message = fallback_message_entry(conversation, contact, batch_builder, entry)
+      reindex_message_for_search(message) if message.is_a?(Message)
     end
-    nil
+  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
+          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)
@@ -487,9 +514,23 @@ class DataImports::Intercom::Importer
     writable_entries = entries.select do |entry|
       %i[repairable_stale_mapping existing_message new_message].include?(entry.classification)
     end
-    content_by_source_id = writable_entries.to_h { |entry| [entry.source_id, content_for(entry.part)] }
+    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)
+      end
+      true
+    rescue InvalidMessagePayloadError => e
+      failed_entries << [entry, e]
+      false
+    end
     skipped_entries, imported_entries = writable_entries.partition { |entry| content_by_source_id[entry.source_id].blank? }
-    messages = insert_messages(conversation, contact, imported_entries, content_by_source_id)
+    messages = insert_messages(imported_entries, attributes_by_source_id)
     upsert_message_mappings(conversation, imported_entries, skipped_entries, messages)
 
     MessageBatchResult.new(
@@ -497,16 +538,15 @@ class DataImports::Intercom::Importer
       skipped_entries: skipped_entries,
       current_entries: grouped_entries.fetch(:current_import, []),
       previous_entries: grouped_entries.fetch(:previous_import, []),
-      messages: messages
+      messages: messages,
+      failed_entries: failed_entries
     )
   end
 
-  def insert_messages(conversation, contact, entries, content_by_source_id)
+  def insert_messages(entries, attributes_by_source_id)
     new_entries = entries.reject(&:message)
     if new_entries.present?
-      attributes = new_entries.map do |entry|
-        message_attributes(conversation, contact, entry.part, entry.source_id, content_by_source_id.fetch(entry.source_id))
-      end
+      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
@@ -516,6 +556,36 @@ class DataImports::Intercom::Importer
     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
+    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|
@@ -584,19 +654,23 @@ class DataImports::Intercom::Importer
     fail_message(conversation, source_entry[:source_id], source_entry[:part], e)
   end
 
-  def import_message(conversation, contact, entry)
-    with_query_timeout_retry do
-      case entry.classification
-      when :current_import
-        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
-      when :previous_import
-        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
-      when :repairable_stale_mapping, :existing_message, :new_message
-        create_message(conversation, contact, entry)
-      else
-        raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
+  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
@@ -607,15 +681,12 @@ class DataImports::Intercom::Importer
 
     attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
     message = entry.message
-    Message.transaction do
-      unless message
-        result = Message.insert_all!([attrs], returning: %w[id])
-        message = Message.find(result.rows.first.first)
-      end
-      record_message_mapping(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
 
diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb
index 205215812..212d90887 100644
--- a/spec/services/data_imports/intercom/importer_spec.rb
+++ b/spec/services/data_imports/intercom/importer_spec.rb
@@ -196,6 +196,169 @@ RSpec.describe DataImports::Intercom::Importer do
     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 '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|
@@ -493,7 +656,7 @@ RSpec.describe DataImports::Intercom::Importer do
       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(:import_message).and_wrap_original do |method, *args|
+      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
@@ -501,7 +664,7 @@ RSpec.describe DataImports::Intercom::Importer do
 
       importer.import_conversations_page
 
-      expect(importer).to have_received(:import_message).once
+      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
@@ -567,7 +730,7 @@ 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(StandardError, 'bulk mapping failed')
+      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'
 
@@ -1430,6 +1593,7 @@ 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