akadmin/chatwoot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(imports): reconcile skipped message stats on retry

Browse Source
This commit is contained in:
Sony Mathew
2026-07-23 14:47:41 +05:30
parent 2ae36d53c3
commit ae57547b84
2 changed files with 24 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/data_imports/intercom/importer.rb
+6 -1
View File
@@ -703,7 +703,12 @@ 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
increment_stat('messages', 'skipped') unless already_recorded
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
end
def message_mapping_handled?(mapping, part)
spec/services/data_imports/intercom/importer_spec.rb
+18 -16
View File
@@ -520,29 +520,31 @@ RSpec.describe DataImports::Intercom::Importer do
expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
end
it 'reconciles skipped conversation stats when a superseded run is retried' do
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 })
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 = 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'
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)
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 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do