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

Merge branch 'codex/cw-7519-intercom-shorter-jobs-heartbeats' into codex/cw-7615-intercom-bulk-message-writes

Browse Source
This commit is contained in:
Sony Mathew
2026-07-23 23:18:17 +05:30
parent ed080ac253 9a489d38c7
commit 0f439f2921
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
@@ -954,7 +954,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 fail_item(item, error)
spec/services/data_imports/intercom/importer_spec.rb
+18 -16
View File
@@ -939,29 +939,31 @@ RSpec.describe DataImports::Intercom::Importer do
).to eq([data_import.id])
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 'keeps skipped contact stats when a query timeout is retried after the item update', :aggregate_failures do