fix(zapi): contact race condition (#149)

app/services/whatsapp/zapi_handlers/helpers.rb

@@ -25,4 +25,21 @@ module Whatsapp::ZapiHandlers::Helpers
     key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{raw_message_id}")
     Redis::Alfred.get(key)
   end
+
+  def with_zapi_contact_lock(phone, timeout: 5.seconds)
+    raise ArgumentError, 'A block is required for with_zapi_contact_lock' unless block_given?
+
+    start_time = Time.now.to_i
+    key = "ZAPI::CONTACT_LOCK::#{phone}"
+
+    while (Time.now.to_i - start_time) < timeout
+      break if Redis::Alfred.set(key, 1, nx: true, ex: timeout)
+
+      sleep(0.1)
+    end
+
+    yield
+  ensure
+    Redis::Alfred.delete(key)
+  end
 end

app/services/whatsapp/zapi_handlers/received_callback.rb

@@ -16,15 +16,17 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module
 
     return handle_edited_message if @raw_message[:isEdit]
 
-    set_contact
+    with_zapi_contact_lock(@raw_message[:phone]) do
+      set_contact
 
-    unless @contact
+      unless @contact
-      Rails.logger.warn "Contact not found for message: #{raw_message_id}"
+        Rails.logger.warn "Contact not found for message: #{raw_message_id}"
-      return
+        return
+      end
+
+      set_conversation
+      handle_create_message
     end
-
-    set_conversation
-    handle_create_message
   ensure
     clear_message_source_id_from_redis
   end
@@ -100,12 +102,14 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module
   def update_existing_contact_inbox(phone, source_id, identifier)
     # NOTE: This is useful when we create a new contact manually, so we don't have information about contact LID;
     # With this, when we receive a message from that contact, we can link it properly.
-    existing_contact_inbox = inbox.contact_inboxes.find_by(source_id: phone)
+    existing_contact = inbox.account.contacts.find_by(phone_number: "+#{phone}")
-    return unless existing_contact_inbox
+    return unless existing_contact
+
+    existing_contact_inbox = existing_contact.contact_inboxes.find_by(inbox_id: inbox.id)
 
     ActiveRecord::Base.transaction do
-      existing_contact_inbox.update!(source_id: source_id)
+      existing_contact.update!(identifier: identifier)
-      existing_contact_inbox.contact.update!(identifier: identifier)
+      existing_contact_inbox&.update!(source_id: source_id)
     end
   end
 

spec/services/whatsapp/zapi_handlers/received_callback_spec.rb

@@ -365,14 +365,13 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do
         end.not_to change(Contact, :count)
       end
 
-      it 'creates new contact_inbox with new source_id for same contact' do
+      it 'updates existing contact_inbox with new source_id' do
         expect do
           service.perform
-        end.to change(ContactInbox, :count).by(1)
+        end.not_to change(ContactInbox, :count)
 
-        new_contact_inbox = ContactInbox.last
+        existing_contact_inbox = existing_contact.contact_inboxes.find_by(inbox: inbox)
-        expect(new_contact_inbox.source_id).to eq('123456789')
+        expect(existing_contact_inbox.source_id).to eq('123456789')
-        expect(new_contact_inbox.contact).to eq(existing_contact)
       end
     end
 
@@ -1214,6 +1213,42 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do
     end
   end
 
+  describe '#process_received_callback with locking' do
+    let(:params) do
+      {
+        type: 'ReceivedCallback',
+        messageId: 'msg_123',
+        momment: Time.current.to_i * 1000,
+        fromMe: false,
+        chatName: 'John Doe',
+        text: { message: 'Hello' },
+        phone: '5511987654321',
+        chatLid: '123456789@lid'
+      }
+    end
+
+    it 'acquires a lock on the contact phone number' do
+      allow(Redis::Alfred).to receive(:set).and_return(true)
+      allow(Redis::Alfred).to receive(:delete)
+
+      service.perform
+
+      expect(Redis::Alfred).to have_received(:set).with('ZAPI::CONTACT_LOCK::5511987654321', 1, nx: true, ex: 5.seconds)
+      expect(Redis::Alfred).to have_received(:delete).with('ZAPI::CONTACT_LOCK::5511987654321')
+    end
+
+    it 'waits for the lock if it is already acquired' do
+      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)
+
+      allow(service).to receive(:sleep).with(0.1)
+
+      service.perform
+
+      expect(service).to have_received(:sleep).with(0.1).once
+    end
+  end
+
   private
 
   def format_message_source_key(message_id)