fix: whatsapp race condition (#185)

* feat: include external created at in whatsapp cloud message

* fix: whatsapp providers race condition

* refactor: remove redundant comments and streamline message processing lock acquisition

* fix: scope message source key by inbox.id to prevent race condition

* fix: scope message source key by inbox.id to prevent race condition

app/services/whatsapp/baileys_handlers/helpers.rb

@@ -197,9 +197,9 @@ module Whatsapp::BaileysHandlers::Helpers # rubocop:disable Metrics/ModuleLength
     Redis::Alfred.get(key)
   end
 
-  def cache_message_source_id_in_redis
+  def acquire_message_processing_lock
     key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{raw_message_id}")
-    ::Redis::Alfred.setex(key, true)
+    Redis::Alfred.set(key, true, nx: true, ex: 1.day)
   end
 
   def clear_message_source_id_from_redis

app/services/whatsapp/baileys_handlers/messages_upsert.rb

@@ -24,9 +24,10 @@ module Whatsapp::BaileysHandlers::MessagesUpsert # rubocop:disable Metrics/Modul
     return unless %w[lid user].include?(jid_type)
     return unless extract_from_jid(type: 'lid')
     return if ignore_message?
-    return if find_message_by_source_id(raw_message_id) || message_under_process?
+    return if find_message_by_source_id(raw_message_id)
+
+    return unless acquire_message_processing_lock
 
-    cache_message_source_id_in_redis
     set_contact
 
     unless @contact

app/services/whatsapp/incoming_message_base_service.rb

@@ -26,9 +26,10 @@ class Whatsapp::IncomingMessageBaseService
     # Multiple webhook event can be received against the same message due to misconfigurations in the Meta
     # business manager account. While we have not found the core reason yet, the following line ensure that
     # there are no duplicate messages created.
-    return if find_message_by_source_id(@processed_params[:messages].first[:id]) || message_under_process?
+    return if find_message_by_source_id(@processed_params[:messages].first[:id])
+
+    return unless acquire_message_processing_lock
 
-    cache_message_source_id_in_redis
     set_contact
     return unless @contact
 
@@ -154,7 +155,8 @@ class Whatsapp::IncomingMessageBaseService
       message_type: :incoming,
       sender: @contact,
       source_id: message[:id].to_s,
-      in_reply_to_external_id: @in_reply_to_external_id
+      in_reply_to_external_id: @in_reply_to_external_id,
+      external_created_at: message[:timestamp].to_i
     )
   end
 

app/services/whatsapp/incoming_message_service_helpers.rb

@@ -66,23 +66,23 @@ module Whatsapp::IncomingMessageServiceHelpers
   def find_message_by_source_id(source_id)
     return unless source_id
 
-    @message = Message.find_by(source_id: source_id)
+    @message = inbox.messages.find_by(source_id: source_id)
   end
 
   def message_under_process?
-    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id])
+    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}")
     Redis::Alfred.get(key)
   end
 
-  def cache_message_source_id_in_redis
-    return if @processed_params.try(:[], :messages).blank?
+  def acquire_message_processing_lock
+    return false if @processed_params.try(:[], :messages).blank?
 
-    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id])
-    ::Redis::Alfred.setex(key, true)
+    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}")
+    Redis::Alfred.set(key, true, nx: true, ex: 1.day)
   end
 
   def clear_message_source_id_from_redis
-    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id])
+    key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}")
     ::Redis::Alfred.delete(key)
   end
 end

app/services/whatsapp/zapi_handlers/helpers.rb

@@ -11,9 +11,9 @@ module Whatsapp::ZapiHandlers::Helpers
     !@raw_message[:fromMe]
   end
 
-  def cache_message_source_id_in_redis
+  def acquire_message_processing_lock
     key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{raw_message_id}")
-    Redis::Alfred.setex(key, true)
+    Redis::Alfred.set(key, true, nx: true, ex: 1.day)
   end
 
   def clear_message_source_id_from_redis

app/services/whatsapp/zapi_handlers/received_callback.rb

@@ -3,16 +3,19 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module
 
   private
 
-  def process_received_callback
+  def process_received_callback # rubocop:disable Metrics/MethodLength
     @raw_message = processed_params
     @message = nil
     @contact_inbox = nil
     @contact = nil
+    @lock_acquired = false
 
     return unless should_process_message?
-    return if find_message_by_source_id(raw_message_id) || message_under_process?
+    return if find_message_by_source_id(raw_message_id)
 
-    cache_message_source_id_in_redis
+    # Atomically acquire lock to prevent race conditions with concurrent webhook deliveries
+    @lock_acquired = acquire_message_processing_lock
+    return unless @lock_acquired
 
     return handle_edited_message if @raw_message[:isEdit]
 
@@ -28,7 +31,7 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module
       handle_create_message
     end
   ensure
-    clear_message_source_id_from_redis
+    clear_message_source_id_from_redis if @lock_acquired
   end
 
   def should_process_message?

spec/services/whatsapp/incoming_message_baileys_service_spec.rb

@@ -452,20 +452,25 @@ describe Whatsapp::IncomingMessageBaileysService do
           end
 
           it 'does not create a message if it is already being processed' do
-            allow(Redis::Alfred).to receive(:get).with(format_message_source_key('msg_123')).and_return(true)
+            # Simulate lock already acquired by returning false from SETNX
+            allow(Redis::Alfred).to receive(:set)
+              .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day)
+              .and_return(false)
 
             described_class.new(inbox: inbox, params: params).perform
 
             expect(inbox.conversations).to be_empty
           end
 
-          it 'caches the message source id in Redis and clears it' do
-            allow(Redis::Alfred).to receive(:setex).with(format_message_source_key('msg_123'), true)
+          it 'caches and clears message source id in Redis' do
+            allow(Redis::Alfred).to receive(:set)
+              .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day)
+              .and_return(true)
             allow(Redis::Alfred).to receive(:delete).with(format_message_source_key('msg_123'))
 
             described_class.new(inbox: inbox, params: params).perform
 
-            expect(Redis::Alfred).to have_received(:setex)
+            expect(Redis::Alfred).to have_received(:set)
             expect(Redis::Alfred).to have_received(:delete)
           end
         end

spec/services/whatsapp/incoming_message_service_spec.rb

@@ -24,6 +24,12 @@ describe Whatsapp::IncomingMessageService do
         expect(whatsapp_channel.inbox.messages.first.content).to eq('Test')
       end
 
+      it 'stores the external_created_at timestamp from the message' do
+        described_class.new(inbox: whatsapp_channel.inbox, params: params).perform
+        message = whatsapp_channel.inbox.messages.first
+        expect(message.external_created_at).to eq(1_633_034_394)
+      end
+
       it 'appends to last conversation when if conversation already exists' do
         contact_inbox = create(:contact_inbox, inbox: whatsapp_channel.inbox, source_id: params[:messages].first[:from])
         2.times.each { create(:conversation, inbox: whatsapp_channel.inbox, contact_inbox: contact_inbox) }
@@ -78,6 +84,123 @@ describe Whatsapp::IncomingMessageService do
         described_class.new(inbox: whatsapp_channel.inbox, params: params).perform
         expect(whatsapp_channel.inbox.messages.count).to eq(1)
       end
+
+      it 'will not create duplicate conversations when same message is received for new contact' do
+        # First call creates contact, conversation and message
+        described_class.new(inbox: whatsapp_channel.inbox, params: params).perform
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+
+        # Second call with same message ID should not create duplicate conversation
+        described_class.new(inbox: whatsapp_channel.inbox, params: params).perform
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+      end
+
+      it 'will not create duplicate conversations when same message is processed concurrently' do
+        # Simulate concurrent processing by having Redis key cleared before second call
+        # but message not yet visible in database due to transaction isolation
+        service1 = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+        service2 = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+
+        # Process first message
+        service1.perform
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+
+        # Even if we clear the Redis key manually, should still not create duplicates
+        key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: params[:messages].first[:id])
+        Redis::Alfred.delete(key)
+
+        # Second call should check database and find existing message
+        service2.perform
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+      end
+
+      it 'prevents race condition with atomic lock when two requests arrive simultaneously' do
+        # This test simulates the exact race condition that was causing duplicate conversations
+        # Two concurrent webhook deliveries for the same message should result in only one being processed
+        threads_started = Concurrent::CountDownLatch.new(2)
+
+        thread1 = Thread.new do
+          threads_started.count_down
+          threads_started.wait # Wait for both threads to be ready
+          service = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+          service.perform
+        end
+
+        thread2 = Thread.new do
+          threads_started.count_down
+          threads_started.wait # Wait for both threads to be ready
+          service = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+          service.perform
+        end
+
+        thread1.join
+        thread2.join
+
+        # Only one conversation and one message should exist
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+      end
+
+      it 'prevents duplicate when both requests pass the message_under_process? check before cache is set' do
+        # This test explicitly simulates the race condition timing:
+        # 1. Request A calls message_under_process? -> returns nil (no lock)
+        # 2. Request B calls message_under_process? -> returns nil (no lock yet!)
+        # 3. Request A calls cache_message_source_id_in_redis
+        # 4. Request B calls cache_message_source_id_in_redis
+        # 5. Both requests pass find_message_by_source_id (message not yet committed)
+        # 6. Both create messages -> DUPLICATE!
+        #
+        # With atomic lock (SETNX), only one can succeed.
+
+        # Key is scoped by inbox.id to prevent cross-inbox lock collisions
+        message_source_key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{whatsapp_channel.inbox.id}_#{params[:messages].first[:id]}")
+        lock_acquired = false
+
+        # Simulate atomic SETNX: only the first call returns true
+        allow(Redis::Alfred).to receive(:set).with(message_source_key, true, nx: true, ex: 1.day) do
+          if lock_acquired
+            false # Second caller fails to acquire lock
+          else
+            lock_acquired = true
+            true # First caller acquires lock
+          end
+        end
+
+        allow(Redis::Alfred).to receive(:delete).with(message_source_key)
+
+        # Run two services "simultaneously" (both pass the check before either sets the lock)
+        service1 = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+        service2 = described_class.new(inbox: whatsapp_channel.inbox, params: params)
+
+        # Mock find_message_by_source_id on specific instances (simulating uncommitted transaction)
+        allow(service1).to receive(:find_message_by_source_id).and_return(nil)
+        allow(service2).to receive(:find_message_by_source_id).and_return(nil)
+
+        service1.perform
+        service2.perform
+
+        # With atomic lock, only ONE should create a message
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+      end
+
+      it 'creates message in second inbox when same source_id exists in different inbox' do
+        # Create a message with same source_id in a different inbox
+        other_whatsapp_channel = create(:channel_whatsapp, sync_templates: false)
+        other_inbox = other_whatsapp_channel.inbox
+        other_contact = create(:contact, account: other_inbox.account)
+        other_contact_inbox = create(:contact_inbox, inbox: other_inbox, contact: other_contact)
+        other_conversation = create(:conversation, inbox: other_inbox, contact_inbox: other_contact_inbox)
+        create(:message, conversation: other_conversation, inbox: other_inbox, source_id: params[:messages].first[:id])
+
+        # Should still create message in our inbox since source_id check should be inbox-scoped
+        described_class.new(inbox: whatsapp_channel.inbox, params: params).perform
+        expect(whatsapp_channel.inbox.conversations.count).to eq(1)
+        expect(whatsapp_channel.inbox.messages.count).to eq(1)
+      end
     end
 
     context 'when unsupported message types' do
@@ -432,7 +555,8 @@ describe Whatsapp::IncomingMessageService do
                                     ] }] }.with_indifferent_access
 
         expect(Message.find_by(source_id: 'wamid.SDFADSf23sfasdafasdfa')).not_to be_present
-        key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: 'wamid.SDFADSf23sfasdafasdfa')
+        # Key is scoped by inbox.id to prevent cross-inbox lock collisions
+        key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{whatsapp_channel.inbox.id}_wamid.SDFADSf23sfasdafasdfa")
 
         Redis::Alfred.setex(key, true)
         expect(Redis::Alfred.get(key)).to be_truthy

spec/services/whatsapp/zapi_handlers/received_callback_spec.rb

@@ -825,9 +825,10 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do
       end
 
       it 'does not create message if it is already being processed' do
-        allow(Redis::Alfred).to receive(:get)
-          .with(format_message_source_key('duplicate_123'))
-          .and_return(true)
+        # Simulate lock already acquired by returning false from SETNX
+        allow(Redis::Alfred).to receive(:set)
+          .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day)
+          .and_return(false)
 
         expect do
           service.perform
@@ -835,16 +836,28 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do
       end
 
       it 'caches and clears message source id in Redis' do
-        allow(Redis::Alfred).to receive(:setex)
+        allow(Redis::Alfred).to receive(:set).and_return(true)
         allow(Redis::Alfred).to receive(:delete)
 
         service.perform
 
-        expect(Redis::Alfred).to have_received(:setex)
-          .with(format_message_source_key('duplicate_123'), true)
+        expect(Redis::Alfred).to have_received(:set)
+          .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day)
         expect(Redis::Alfred).to have_received(:delete)
           .with(format_message_source_key('duplicate_123'))
       end
+
+      it 'does not clear lock when acquisition fails' do
+        allow(Redis::Alfred).to receive(:set)
+          .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day)
+          .and_return(false)
+        allow(Redis::Alfred).to receive(:delete)
+
+        service.perform
+
+        expect(Redis::Alfred).not_to have_received(:delete)
+          .with(format_message_source_key('duplicate_123'))
+      end
     end
 
     context 'when attachment download fails' do
@@ -1238,6 +1251,10 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do
     end
 
     it 'waits for the lock if it is already acquired' do
+      # Stub the message processing lock to always succeed
+      allow(Redis::Alfred).to receive(:set)
+        .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day)
+        .and_return(true)
       allow(Redis::Alfred).to receive(:set).with('ZAPI::CONTACT_LOCK::5511987654321', 1, nx: true, ex: 5.seconds).and_return(false, true)
       allow(Redis::Alfred).to receive(:delete)