diff --git a/app/services/whatsapp/zapi_handlers/helpers.rb b/app/services/whatsapp/zapi_handlers/helpers.rb index e80afe8a8..f26a039d9 100644 --- a/app/services/whatsapp/zapi_handlers/helpers.rb +++ b/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 diff --git a/app/services/whatsapp/zapi_handlers/received_callback.rb b/app/services/whatsapp/zapi_handlers/received_callback.rb index e89edf5e5..fb3e4e4ad 100644 --- a/app/services/whatsapp/zapi_handlers/received_callback.rb +++ b/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 - Rails.logger.warn "Contact not found for message: #{raw_message_id}" - return + unless @contact + Rails.logger.warn "Contact not found for message: #{raw_message_id}" + 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) - return unless existing_contact_inbox + existing_contact = inbox.account.contacts.find_by(phone_number: "+#{phone}") + 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_inbox.contact.update!(identifier: identifier) + existing_contact.update!(identifier: identifier) + existing_contact_inbox&.update!(source_id: source_id) end end diff --git a/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb b/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb index 219fd8bf4..da8e1f12b 100644 --- a/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb +++ b/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 - expect(new_contact_inbox.source_id).to eq('123456789') - expect(new_contact_inbox.contact).to eq(existing_contact) + existing_contact_inbox = existing_contact.contact_inboxes.find_by(inbox: inbox) + expect(existing_contact_inbox.source_id).to eq('123456789') 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)