From 4f8215bc5c59e6a03ecd7feac2d0638a304ef7e8 Mon Sep 17 00:00:00 2001 From: Sony Mathew <2040199+sony-mathew@users.noreply.github.com> Date: Thu, 23 Jul 2026 00:51:31 +0530 Subject: [PATCH] fix(imports): reload run state before heartbeat checks --- .../data_imports/intercom/importer.rb | 4 +++- .../data_imports/intercom/importer_spec.rb | 23 +++++++++++++++++-- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index 0d2371162..0dbda8ff7 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -197,8 +197,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 @@ -435,6 +435,8 @@ class DataImports::Intercom::Importer fail_message(chatwoot_conversation, message_source_id, part, e) end end + return false if import_stopped? + true end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index 0572d3d4a..cf54cd9fa 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -325,8 +325,7 @@ RSpec.describe DataImports::Intercom::Importer do allow(data_import).to receive(:touch).and_call_original importer = described_class.new(data_import: data_import, run_id: run_id) allow(importer).to receive(:create_message) do - data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) - travel 1.minute + DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) end importer.import_conversations_page @@ -337,6 +336,26 @@ RSpec.describe DataImports::Intercom::Importer do end end + it 'does not persist stale stats when a newer run takes over during the final part', :aggregate_failures do + run_id = 'intercom-run-1' + data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }) + importer = described_class.new(data_import: data_import, run_id: run_id) + allow(importer).to receive(:create_message).and_wrap_original do |method, *args| + method.call(*args).tap do + part = args[2] + next unless part['id'] == 'part_2' + + DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) + end + end + + importer.import_conversations_page + + conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1') + expect(conversation.messages.count).to eq(3) + expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0) + end + it 'reconciles conversation stats when a superseded run is retried', :aggregate_failures do freeze_time do run_id = 'intercom-run-1'