try to match pending messages by content

as gmail api changes the message id

Author: dutow

app/services/email_ingestor.rb

@@ -20,6 +20,20 @@ def ingest_raw(raw_message, mailing_list:, fallback_threading: false, trust_date
       return existing_message
     end
 
+    pending_echo = find_pending_echo(
+      reply_to_message_id: reply_to_message_id,
+      from_addresses: extract_from_emails(m),
+      echo_body: body,
+      echo_sent_at: sent_at
+    )
+    if pending_echo
+      pending_echo.update_columns(message_id: message_id) if pending_echo.message_id != message_id
+      update_existing_message(pending_echo, body: body, sent_at: sent_at,
+                              reply_to_message_id: reply_to_message_id, update_existing: update_existing)
+      associate_mailing_list(pending_echo, mailing_list)
+      return pending_echo
+    end
+
     import_log = ""
 
     from = build_from_aliases(m, sent_at)
@@ -263,6 +277,53 @@ def clean_reference(ref)
     MessageIdNormalizer.normalize(ref)
   end
 
+  # Gmail rewrites Message-IDs on send, so the echo from the list has a different
+  # id than the pending row we created. Match heuristically on parent reference,
+  # sender, body fragment, and time window.
+  PENDING_ECHO_WINDOW = 24.hours
+  PENDING_ECHO_MIN_BODY = 20
+
+  def find_pending_echo(reply_to_message_id:, from_addresses:, echo_body:, echo_sent_at:)
+    return nil if from_addresses.empty?
+
+    scope = Message.pending
+    scope = scope.where(reply_to_message_id: reply_to_message_id) if reply_to_message_id.present?
+
+    reference_time = echo_sent_at || Time.current
+    scope = scope.where(sent_at: (reference_time - PENDING_ECHO_WINDOW)..(reference_time + PENDING_ECHO_WINDOW))
+
+    sender_alias_ids = Alias.where("LOWER(email) IN (?)", from_addresses).pluck(:id)
+    return nil if sender_alias_ids.empty?
+    scope = scope.where(sender_id: sender_alias_ids)
+
+    candidates = scope.to_a
+    return nil if candidates.empty?
+
+    echo_normalized = normalize_for_body_match(echo_body)
+    return nil if echo_normalized.blank?
+
+    matches = candidates.select do |c|
+      pending_normalized = normalize_for_body_match(c.body)
+      next false if pending_normalized.length < PENDING_ECHO_MIN_BODY
+      echo_normalized.include?(pending_normalized)
+    end
+
+    matches.size == 1 ? matches.first : nil
+  end
+
+  def normalize_for_body_match(text)
+    text.to_s.gsub(/\s+/, " ").strip
+  end
+
+  def extract_from_emails(mail)
+    addrs = mail.from
+    return [] unless addrs
+    addrs = [ addrs ] if addrs.is_a?(String)
+    addrs.compact.map { |a| a.to_s.downcase.strip }.reject(&:blank?)
+  rescue NoMethodError
+    []
+  end
+
   def update_default_alias_for_person(alias_record)
     return unless alias_record&.person
     person = alias_record.person

spec/services/email_ingestor_spec.rb

@@ -265,6 +265,108 @@
     end
   end
 
+  describe "pending echo with Gmail-rewritten Message-ID" do
+    let(:user)   { create(:user) }
+    let(:sender) { create(:alias, user: user, email: "sender@example.com", name: "Sender") }
+    let(:list)   { create(:mailing_list) }
+    let(:parent_id) { "parent-thread@example.com" }
+    let(:topic)  { create(:topic, creator: sender) }
+    let!(:parent_msg) {
+      Message.create!(
+        topic: topic, sender: sender, sender_person_id: sender.person_id,
+        subject: "Topic", body: "parent body",
+        message_id: parent_id, state: Message::STATE_SENT,
+        created_at: 1.hour.ago
+      )
+    }
+    let(:pending_body) { "Hello, this is my reply body with enough length to match." }
+    let!(:pending) {
+      Message.create!(
+        topic: topic, sender: sender, sender_person_id: sender.person_id,
+        reply_to: parent_msg, reply_to_message_id: parent_id,
+        subject: "Re: Topic", body: pending_body,
+        message_id: "<placeholder-uuid@hackorum.dev>",
+        state: Message::STATE_PENDING,
+        sent_at: 30.seconds.ago,
+        created_at: 30.seconds.ago
+      )
+    }
+
+    let(:gmail_echo_id) { "CABx_real_gmail_id@mail.gmail.com" }
+    let(:echo_body) { "#{pending_body}\n\n-- \nList footer goes here" }
+    let(:raw_echo) {
+      <<~EML
+        From: Sender <sender@example.com>
+        To: pgsql-test@example.com
+        Subject: [PGSQL] Re: Topic
+        Message-ID: <#{gmail_echo_id}>
+        In-Reply-To: <#{parent_id}>
+        References: <#{parent_id}>
+        Date: #{Time.current.rfc2822}
+
+        #{echo_body}
+      EML
+    }
+
+    it "flips pending to sent and rewrites message_id" do
+      described_class.new.ingest_raw(raw_echo, mailing_list: list)
+      pending.reload
+      expect(pending.state).to eq(Message::STATE_SENT)
+      expect(pending.message_id).to eq(gmail_echo_id)
+    end
+
+    it "does not create a duplicate message" do
+      expect {
+        described_class.new.ingest_raw(raw_echo, mailing_list: list)
+      }.not_to change { Message.count }
+    end
+
+    it "associates the mailing list with the pending row" do
+      described_class.new.ingest_raw(raw_echo, mailing_list: list)
+      expect(pending.message_mailing_lists.where(mailing_list: list)).to exist
+    end
+
+    it "leaves pending intact when echo body does not contain pending body" do
+      raw = raw_echo.sub(echo_body, "completely unrelated content here that is long enough")
+      expect {
+        described_class.new.ingest_raw(raw, mailing_list: list)
+      }.to change { Message.count }.by(1)
+      expect(pending.reload.state).to eq(Message::STATE_PENDING)
+    end
+
+    it "skips heuristic when sender email differs" do
+      raw = raw_echo.sub("sender@example.com", "someone-else@example.com")
+      expect {
+        described_class.new.ingest_raw(raw, mailing_list: list)
+      }.to change { Message.count }.by(1)
+      expect(pending.reload.state).to eq(Message::STATE_PENDING)
+    end
+
+    it "skips heuristic outside the time window" do
+      pending.update_columns(sent_at: 3.days.ago, created_at: 3.days.ago)
+      expect {
+        described_class.new.ingest_raw(raw_echo, mailing_list: list)
+      }.to change { Message.count }.by(1)
+      expect(pending.reload.state).to eq(Message::STATE_PENDING)
+    end
+
+    it "skips heuristic when two pending rows match ambiguously" do
+      Message.create!(
+        topic: topic, sender: sender, sender_person_id: sender.person_id,
+        reply_to: parent_msg, reply_to_message_id: parent_id,
+        subject: "Re: Topic", body: pending_body,
+        message_id: "<another-placeholder@hackorum.dev>",
+        state: Message::STATE_PENDING,
+        sent_at: 20.seconds.ago,
+        created_at: 20.seconds.ago
+      )
+      expect {
+        described_class.new.ingest_raw(raw_echo, mailing_list: list)
+      }.to change { Message.count }.by(1)
+      expect(pending.reload.state).to eq(Message::STATE_PENDING)
+    end
+  end
+
   describe "#fallback_thread_lookup scoped to list" do
     let(:ingestor) { described_class.new }
     let(:hackers_list) { create(:mailing_list, identifier: "pgsql-hackers", display_name: "hackers") }