fix: [CW-7001] Limit emails fetch (#14354)

This PR limits IMAP email fetching to 500 messages per sync run to avoid
expensive/long-running mailbox scans. It also filters out
already-imported emails and Chatwoot-generated notification emails
during the header fetch phase, before fetching full email bodies,
reducing unnecessary IMAP work.

Fixes #CW-7001 (issue) :
https://linear.app/chatwoot/issue/CW-7001/emails-not-syncing

app/services/imap/base_fetch_email_service.rb

@@ -1,6 +1,8 @@
 require 'net/imap'
 
 class Imap::BaseFetchEmailService
+  MAX_MESSAGES_PER_SYNC = 500
+
   pattr_initialize [:channel!, :interval]
 
   def fetch_emails
@@ -77,27 +79,49 @@ class Imap::BaseFetchEmailService
     Rails.logger.info "[IMAP::FETCH_EMAIL_SERVICE] Fetching mails from #{channel.email}, found #{seq_nums.length}."
 
     message_ids_with_seq = []
-    seq_nums.each_slice(10).each do |batch|
-      # Fetch only message-id only without mail body or contents.
-      batch_message_ids = imap_client.fetch(batch, 'BODY.PEEK[HEADER]')
-
-      # .fetch returns an array of Net::IMAP::FetchData or nil
-      # (instead of an empty array) if there is no matching message.
-      # Check
-      if batch_message_ids.blank?
-        Rails.logger.info "[IMAP::FETCH_EMAIL_SERVICE] Fetching the batch failed for #{channel.email}."
-        next
-      end
-
-      batch_message_ids.each do |data|
-        message_id = build_mail_from_string(data.attr['BODY[HEADER]']).message_id
-        message_ids_with_seq.push([data.seqno, message_id])
+    seq_nums.each_slice(MAX_MESSAGES_PER_SYNC).each do |batch|
+      append_message_ids_for_batch(batch, message_ids_with_seq)
+      if message_ids_with_seq.length >= MAX_MESSAGES_PER_SYNC
+        Rails.logger.info "[IMAP::FETCH_EMAIL_SERVICE] Reached MAX_MESSAGES_PER_SYNC=#{MAX_MESSAGES_PER_SYNC} for #{channel.email}, stopping sync."
+        break
       end
     end
 
     message_ids_with_seq
   end
 
+  def append_message_ids_for_batch(batch, message_ids_with_seq)
+    # Fetch only message-id only without mail body or contents.
+    batch_message_ids = imap_client.fetch(batch, 'BODY.PEEK[HEADER]')
+    Rails.logger.info "[IMAP::FETCH_EMAIL_SERVICE] Fetching the batch for #{channel.email}. Found #{batch_message_ids&.length} messages."
+
+    # .fetch returns an array of Net::IMAP::FetchData or nil
+    # (instead of an empty array) if there is no matching message.
+    if batch_message_ids.blank?
+      Rails.logger.info "[IMAP::FETCH_EMAIL_SERVICE] Fetching the batch failed for #{channel.email}."
+      return
+    end
+
+    batch_message_ids.each do |data|
+      entry = build_message_id_entry(data)
+      next if entry.nil?
+
+      message_ids_with_seq.push(entry)
+      break if message_ids_with_seq.length >= MAX_MESSAGES_PER_SYNC
+    end
+  end
+
+  def build_message_id_entry(data)
+    mail = build_mail_from_string(data.attr['BODY[HEADER]'])
+    return nil if MailPresenter.new(mail, channel.account).notification_email_from_chatwoot?
+
+    message_id = mail.message_id
+    return nil if message_id.blank?
+    return nil if email_already_present?(channel, message_id)
+
+    [data.seqno, message_id]
+  end
+
   # Sends a SEARCH command to search the mailbox for messages that were
   # created between yesterday (or given date) and today and returns message sequence numbers.
   # Return <message set>

spec/services/imap/fetch_email_service_spec.rb

@@ -7,6 +7,7 @@ RSpec.describe Imap::FetchEmailService do
   let(:imap_email_channel) { create(:channel_email, :imap_email, account: account) }
   let(:imap) { instance_double(Net::IMAP) }
   let(:eml_content_with_message_id) { Rails.root.join('spec/fixtures/files/only_text.eml').read }
+  let(:eml_content_without_message_id) { eml_content_with_message_id.sub(/^Message-ID:.*\n/, '') }
 
   describe '#perform' do
     before do
@@ -63,6 +64,34 @@ RSpec.describe Imap::FetchEmailService do
           expect(imap).not_to have_received(:fetch).with(1, 'RFC822')
         end
       end
+
+      it 'does not count emails without message ids toward the sync limit' do
+        travel_to '26.10.2020 10:00'.to_datetime do
+          email_object = create_inbound_email_from_fixture('only_text.eml')
+          max_messages_per_sync = Imap::BaseFetchEmailService::MAX_MESSAGES_PER_SYNC
+          empty_message_id_seq_nums = (1..max_messages_per_sync).to_a
+          valid_message_seq_num = max_messages_per_sync + 1
+          empty_message_id_headers = empty_message_id_seq_nums.map do |seq_num|
+            Net::IMAP::FetchData.new(seq_num, 'BODY[HEADER]' => eml_content_without_message_id)
+          end
+          valid_email_header = Net::IMAP::FetchData.new(valid_message_seq_num, 'BODY[HEADER]' => eml_content_with_message_id)
+          imap_fetch_mail = Net::IMAP::FetchData.new(valid_message_seq_num, 'RFC822' => eml_content_with_message_id)
+
+          allow(imap).to receive(:search).with(%w[SINCE 25-Oct-2020]).and_return(empty_message_id_seq_nums + [valid_message_seq_num])
+          allow(imap).to receive(:fetch).with(empty_message_id_seq_nums, 'BODY.PEEK[HEADER]').and_return(empty_message_id_headers)
+          allow(imap).to receive(:fetch).with([valid_message_seq_num], 'BODY.PEEK[HEADER]').and_return([valid_email_header])
+          allow(imap).to receive(:fetch).with(valid_message_seq_num, 'RFC822').and_return([imap_fetch_mail])
+          allow(imap).to receive(:logout)
+
+          result = described_class.new(channel: imap_email_channel).perform
+
+          expect(result.length).to eq 1
+          expect(result[0].message_id).to eq email_object.message_id
+          expect(imap).to have_received(:fetch).with(empty_message_id_seq_nums, 'BODY.PEEK[HEADER]')
+          expect(imap).to have_received(:fetch).with([valid_message_seq_num], 'BODY.PEEK[HEADER]')
+          expect(imap).to have_received(:fetch).with(valid_message_seq_num, 'RFC822')
+        end
+      end
     end
   end
 end