FIX: IMAP post alerter race condition and code improvements (#11711)

This PR fixes a race condition with the IMAP notification code. In the `Email::Receiver` we call the `NewPostManager` to create the post and enqueue jobs and sends alerts via `PostAlerter`. However, if the post alerter reaches the `notify_pm_users` and the `group_notifying_via_smtp` method _before_ the incoming email is updated with the post and topic, we unnecessarily send a notification to the person who just posted. The result of this is that the IMAP syncer re-imports the email sent to the user about their own post, which looks like this in the group inbox:

To fix this, we skip the jobs enqueued by `NewPostManager` and only enqueue them with `PostJobsEnqueuer` manually _after_ the incoming email record has been updated with the post and topic.

Other improvements:

* Moved code to calculate email addresses from `IncomingEmail` records into the topic, with a group passed in, for easier testing and debugging. It is not the responsibility of the post alerter to figure this stuff out.
* Add shortcut methods on `IncomingEmail` to split or provide an empty array for to and cc addresses to avoid repetition.

app/models/incoming_email.rb

@@ -29,6 +29,14 @@ class IncomingEmail < ActiveRecord::Base
     SQL
   end
 
+  def to_addresses_split
+    self.to_addresses&.split(";") || []
+  end
+
+  def cc_addresses_split
+    self.cc_addresses&.split(";") || []
+  end
+
   def to_addresses=(to)
     if to&.is_a?(Array)
       to = to.map(&:downcase).join(";")

app/models/topic.rb

@@ -1632,6 +1632,34 @@ class Topic < ActiveRecord::Base
       .first
   end
 
+  def incoming_email_addresses(group: nil, received_before: Time.zone.now)
+    email_addresses = Set.new
+
+    # TODO(martin) Look at improving this N1, it will just get slower the
+    # more replies/incoming emails there are for the topic.
+    self.incoming_email.where("created_at <= ?", received_before).each do |incoming_email|
+      to_addresses = incoming_email.to_addresses_split
+      cc_addresses = incoming_email.cc_addresses_split
+      combined_addresses = [to_addresses, cc_addresses].flatten
+
+      # We only care about the emails addressed to the group or CC'd to the
+      # group if the group is present. If combined addresses is empty we do
+      # not need to do this check, and instead can proceed on to adding the
+      # from address.
+      if group.present? && combined_addresses.any?
+        next if combined_addresses.none? { |address| address =~ group.email_username_regex }
+      end
+
+      email_addresses.add(incoming_email.from_address)
+      email_addresses.merge(combined_addresses)
+    end
+
+    email_addresses.subtract([nil, ''])
+    email_addresses.delete(group.email_username) if group.present?
+
+    email_addresses.to_a
+  end
+
   private
 
   def invite_to_private_message(invited_by, target_user, guardian)

app/services/post_alerter.rb

@@ -416,8 +416,9 @@ class PostAlerter
     if opts[:skip_send_email_to]&.include?(user.email)
       skip_send_email = true
     elsif original_post.via_email && (incoming_email = original_post.incoming_email)
-      skip_send_email = contains_email_address?(incoming_email.to_addresses, user) ||
+      skip_send_email =
-        contains_email_address?(incoming_email.cc_addresses, user)
+        incoming_email.to_addresses_split.include?(user.email) ||
+        incoming_email.cc_addresses_split.include?(user.email)
     else
       skip_send_email = opts[:skip_send_email]
     end
@@ -458,11 +459,6 @@ class PostAlerter
     end
   end
 
-  def contains_email_address?(addresses, user)
-    return false if addresses.blank?
-    addresses.split(";").include?(user.email)
-  end
-
   def push_notification(user, payload)
     return if user.do_not_disturb?
 
@@ -557,9 +553,13 @@ class PostAlerter
   end
 
   def group_notifying_via_smtp(post)
-    return nil if !SiteSetting.enable_smtp ||
+    return nil if !SiteSetting.enable_smtp || post.post_type != Post.types[:regular]
-                  post.post_type != Post.types[:regular] ||
+
-                  post.incoming_email
+    # If the post already has an incoming email, it has been set in the
+    # Email::Receiver or via the GroupSmtpEmail job, and thus it was created
+    # via the IMAP/SMTP flow, so there is no need to notify those involved
+    # in the email chain again.
+    return nil if post.incoming_email.present?
 
     post.topic.allowed_groups
       .where.not(smtp_server: nil)
@@ -576,28 +576,10 @@ class PostAlerter
 
     # users who interacted with the post by _directly_ emailing the group
     if group = group_notifying_via_smtp(post)
-      group_email_regex = group.email_username_regex
+      email_addresses = post.topic.incoming_email_addresses(group: group)
-      email_addresses = Set[group.email_username]
 
-      post.topic.incoming_email.each do |incoming_email|
+      if email_addresses.any?
-        to_addresses = incoming_email.to_addresses&.split(';')
+        Jobs.enqueue(:group_smtp_email, group_id: group.id, post_id: post.id, email: email_addresses)
-        cc_addresses = incoming_email.cc_addresses&.split(';')
-
-        next if to_addresses&.none? { |address| address =~ group_email_regex } &&
-                cc_addresses&.none? { |address| address =~ group_email_regex }
-
-        email_addresses.add(incoming_email.from_address)
-        email_addresses.merge(to_addresses) if to_addresses.present?
-        email_addresses.merge(cc_addresses) if cc_addresses.present?
-      end
-
-      email_addresses.subtract([nil, ''])
-
-      if email_addresses.size > 1
-        Jobs.enqueue(:group_smtp_email,
-          group_id: group.id,
-          post_id: post.id,
-          email: email_addresses.to_a - [group.email_username])
       end
     end
 

lib/email/receiver.rb

@@ -1184,6 +1184,10 @@ module Email
         options[:post_type] = Post.types[:whisper]
       end
 
+      # To avoid race conditions with the post alerter and Group SMTP
+      # emails, we skip the jobs here and enqueue them only _after_
+      # the incoming email has been updated with the post and topic.
+      options[:skip_jobs] = true
       result = NewPostManager.new(user, options).perform
 
       errors = result.errors.full_messages
@@ -1203,6 +1207,13 @@ module Email
         if result.post.topic&.private_message? && !is_bounce?
           add_other_addresses(result.post, user)
         end
+
+        # Alert the people involved in the topic now that the incoming email
+        # has been linked to the post.
+        PostJobsEnqueuer.new(result.post, result.post.topic, options[:topic_id].blank?,
+                             import_mode: options[:import_mode],
+                             post_alert_options: options[:post_alert_options]
+                            ).enqueue_jobs
       end
 
       result.post

lib/imap/sync.rb

@@ -309,8 +309,8 @@ module Imap
       tags = Set.new
 
       # "Plus" part from the destination email address
-      to_addresses = incoming_email.to_addresses&.split(";") || []
+      to_addresses = incoming_email.to_addresses_split
-      cc_addresses = incoming_email.cc_addresses&.split(";") || []
+      cc_addresses = incoming_email.cc_addresses_split
       (to_addresses + cc_addresses).each do |address|
         if plus_part = address&.scan(group_email_regex)&.first&.first
           tags.add("plus:#{plus_part[1..-1]}") if plus_part.length > 0

spec/models/topic_spec.rb

@@ -2744,4 +2744,57 @@ describe Topic do
       expect(topic.like_count).to eq(1)
     end
   end
+
+  describe "#incoming_email_addresses" do
+    fab!(:group) do
+      Fabricate(
+        :group,
+        smtp_server: "imap.gmail.com",
+        smtp_port: 587,
+        email_username: "discourse@example.com",
+        email_password: "discourse@example.com"
+      )
+    end
+
+    fab!(:topic) do
+      Fabricate(:private_message_topic,
+        topic_allowed_groups: [
+          Fabricate.build(:topic_allowed_group, group: group)
+        ]
+      )
+    end
+
+    let!(:incoming1) do
+      Fabricate(:incoming_email, to_addresses: "discourse@example.com", from_address: "johnsmith@user.com", topic: topic, post: topic.posts.first, created_at: 20.minutes.ago)
+    end
+    let!(:incoming2) do
+      Fabricate(:incoming_email, from_address: "discourse@example.com", to_addresses: "johnsmith@user.com", topic: topic, post: Fabricate(:post, topic: topic), created_at: 10.minutes.ago)
+    end
+    let!(:incoming3) do
+      Fabricate(:incoming_email, to_addresses: "discourse@example.com", from_address: "johnsmith@user.com", topic: topic, post: topic.posts.first, cc_addresses: "otherguy@user.com", created_at: 2.minutes.ago)
+    end
+    let!(:incoming4) do
+      Fabricate(:incoming_email, to_addresses: "unrelated@test.com", from_address: "discourse@example.com", topic: topic, post: topic.posts.first, created_at: 1.minutes.ago)
+    end
+
+    it "returns an array of all the incoming email addresses" do
+      expect(topic.incoming_email_addresses).to match_array(
+        ["discourse@example.com", "johnsmith@user.com", "otherguy@user.com", "unrelated@test.com"]
+      )
+    end
+
+    it "returns an array of all the incoming email addresses where incoming was received before X" do
+      expect(topic.incoming_email_addresses(received_before: 5.minutes.ago)).to match_array(
+        ["discourse@example.com", "johnsmith@user.com"]
+      )
+    end
+
+    context "when the group is present" do
+      it "excludes incoming emails that are not to or CCd to the group" do
+        expect(topic.incoming_email_addresses(group: group)).not_to include(
+          "unrelated@test.com"
+        )
+      end
+    end
+  end
 end

spec/services/post_alerter_spec.rb

@@ -1244,7 +1244,7 @@ describe PostAlerter do
     end
   end
 
-  context "SMTP" do
+  context "SMTP (group_smtp_email)" do
     before do
       SiteSetting.enable_smtp = true
       Jobs.run_immediately!