fix(whatsapp): Prevent duplicate conversations from concurrent uploads (#14060)

When a WhatsApp contact starts a new conversation by sending multiple
images at once (an album), each image arrives as a separate webhook.
Because no conversation exists yet, the concurrent workers each pass the
"does a conversation exist?" check and each create their own
conversation — producing one conversation per image instead of one
grouped conversation.

This fix serializes webhook processing per `(inbox, contact)` using a
Redis lock at the job level, so only one webhook at a time can create
the initial conversation for a given contact. Concurrent workers retry
with backoff and append to the same conversation once the lock is
released.

## Closes

- Closes #13261

## How to test

1. On a WhatsApp inbox, ensure there is no active (open) conversation
with a specific test contact — resolve or delete any existing one.
2. From a phone, select 6+ images in the WhatsApp gallery and send them
as a single album to the Chatwoot-connected number.
3. Open the Chatwoot dashboard and confirm exactly **one** new
conversation is created, with all images grouped under it.
4. Repeat the test with a mix of attachment types (XMLs, PDFs, images)
sent in rapid succession — still one conversation.

## What changed

- New Redis key `WHATSAPP_MESSAGE_CREATE_LOCK::<inbox_id>::<sender_id>`
in `lib/redis/redis_keys.rb`.
- `Webhooks::WhatsappEventsJob` now inherits from `MutexApplicationJob`
and wraps event processing in `with_lock(key)`, matching the pattern
already used by `FacebookEventsJob`, `InstagramEventsJob`, and
`TiktokEventsJob`.
- Uses `retry_on LockAcquisitionError, wait: 1.second, attempts: 8` so
concurrent webhooks retry until the lock is free instead of poll-waiting
inside the service.
- Sender ID is derived from the webhook payload (contact's `from`, or
`to` for SMB echo events); status-only webhooks bypass the lock.
- Issue 1 from the report (same `source_id` redelivery) was already
handled previously by `Whatsapp::MessageDedupLock` (atomic `SET NX EX`);
no changes needed there.

---------

Co-authored-by: Muhsin <12408980+muhsin-k@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/jobs/webhooks/whatsapp_events_job.rb

@@ -1,5 +1,9 @@
-class Webhooks::WhatsappEventsJob < ApplicationJob
+class Webhooks::WhatsappEventsJob < MutexApplicationJob
   queue_as :low
+  # Retry budget (19 × 2s = 38s) must exceed the 30s lock TTL set in `perform`, otherwise
+  # a webhook that arrives just after the lock is acquired can exhaust retries before the
+  # holder finishes and silently drop its message.
+  retry_on LockAcquisitionError, wait: 2.seconds, attempts: 20
 
   def perform(params = {})
     channel = find_channel_from_whatsapp_business_payload(params)
@@ -9,6 +13,20 @@ class Webhooks::WhatsappEventsJob < ApplicationJob
       return
     end
 
+    sender_id = contact_sender_id(params)
+    return process_events(channel, params) if sender_id.blank?
+
+    # Album uploads arrive as separate concurrent webhooks. Serialize per (inbox, contact)
+    # so the first webhook creates the conversation and the rest append to it.
+    # 30s TTL covers the attachment download + transaction — the default 1s expires
+    # mid-processing and lets a concurrent webhook re-acquire before the first commit.
+    key = format(::Redis::Alfred::WHATSAPP_MESSAGE_MUTEX, inbox_id: channel.inbox.id, sender_id: sender_id)
+    with_lock(key, 30.seconds) do
+      process_events(channel, params)
+    end
+  end
+
+  def process_events(channel, params)
     if message_echo_event?(params)
       handle_message_echo(channel, params)
     else
@@ -69,6 +87,16 @@ class Webhooks::WhatsappEventsJob < ApplicationJob
 
   private
 
+  # Echo payloads reverse the fields — `from` is the business number and `to` is the contact.
+  # Returns nil for status-only webhooks so they bypass the lock.
+  def contact_sender_id(params)
+    value = params.dig(:entry, 0, :changes, 0, :value) || params
+    message = (value[:messages] || value[:message_echoes])&.first
+    return if message.blank?
+
+    message[:to] || message[:from]
+  end
+
   def channel_is_inactive?(channel)
     return true if channel.blank?
     return true if channel.reauthorization_required?

lib/redis/redis_keys.rb

@@ -43,6 +43,7 @@ module Redis::RedisKeys
   TIKTOK_REFRESH_TOKEN_MUTEX = 'TIKTOK_REFRESH_TOKEN_LOCK::%<channel_id>s'.freeze
   SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<conversation_id>s::%<reference_id>s'.freeze
   EMAIL_MESSAGE_MUTEX = 'EMAIL_CHANNEL_LOCK::%<inbox_id>s'.freeze
+  WHATSAPP_MESSAGE_MUTEX = 'WHATSAPP_MESSAGE_CREATE_LOCK::%<inbox_id>s::%<sender_id>s'.freeze
   CRM_PROCESS_MUTEX = 'CRM_PROCESS_MUTEX::%<hook_id>s'.freeze
   CAPTAIN_DOCUMENT_SYNC_MUTEX = 'CAPTAIN_DOCUMENT_SYNC_LOCK::%<document_id>s'.freeze