fix(instagram): process webhook events after mutex retry exhaustion (#14647)

Instagram webhooks use a Redis mutex to protect the first-message path
for a contact/account pair. When multiple webhook events for the same
Instagram contact arrive at nearly the same time, they can all enter the
“find or create” flow together.

There are two pieces involved:

`ContactInbox` maps the external Instagram scoped user id to a Chatwoot
contact inside an inbox. That side already has a unique `(inbox_id,
source_id)` index and retry handling, so duplicate contact-inbox
creation is mostly protected at the database layer.

`Conversation` creation is more fragile. The message builder looks for
an existing active conversation for the contact/inbox and creates one
when none is found. If two first-message jobs run concurrently, both can
see “no active conversation yet” and both can create a conversation. The
mutex exists to bump those jobs apart long enough for the first one to
create the conversation, so the next one appends to it instead of
creating a duplicate.

In production, the old behavior could turn that race dampener into a
drop path. Once `Webhooks::InstagramEventsJob` exhausted its lock
retries, ActiveJob stopped retrying with
`MutexApplicationJob::LockAcquisitionError`. Since ActiveJob retries are
not FIFO, later jobs could still win the lock while older jobs kept
getting deferred until exhaustion.

## Mutex Application Job Changes

This adds `retry_on_lock_conflict` as a small wrapper around the
existing `retry_on LockAcquisitionError` pattern.

The new API keeps the default behavior intact, but lets jobs explicitly
define what should happen after lock retry exhaustion:

```ruby
retry_on_lock_conflict wait: 1.second,
                       attempts: 3,
                       on_exhaustion: :process_without_lock
```

The fallback receives the original job arguments, so job classes do not
need to reach into ActiveJob internals like `arguments.first`.

## Instagram Fallback

Instagram now processes the webhook payload even after the mutex retry
window is exhausted.

This matches what the mutex was meant to do in the first place: bump
concurrent events apart long enough to avoid duplicate conversation
creation, not permanently block or drop webhook events. If a job cannot
acquire the lock after a few retries, it continues through the normal
Instagram processing path without the mutex.

## Retry And Lock Tuning

The Instagram lock retry window now uses deterministic backoff:

```ruby
retry_on_lock_conflict wait: ->(executions) { executions.seconds },
                       attempts: 3,
                       on_exhaustion: :process_without_lock
```

The lock TTL is also set explicitly:

```ruby
with_lock(key, 3.seconds)
```

This matters because Rails treats `attempts` as total executions, not
retries after the first execution. With a fixed `wait: 1.second` and
`attempts: 3`, the exhaustion handler can run roughly two seconds after
the first lock conflict. That is earlier than a 3-second lock TTL, so
the fallback could process without the lock while the original mutex is
still valid. That would reopen the duplicate-conversation race the lock
is meant to dampen.

Using a proc wait makes the timing predictable and avoids Rails jitter
for this retry path. The first conflict waits about 1 second, the second
waits about 2 seconds, and the final attempt happens around the 3-second
mark. That lines the retry window up with the Redis lock TTL before
falling back to `process_without_lock`.

The protected race window is intentionally small. `ContactInbox`
creation already has database uniqueness protection, while conversation
creation is the softer `find active conversation || create` path. We
only need to give the first job enough time to create the conversation
state that later jobs should reuse.

The mutex still does not preserve message order, and ActiveJob retries
are not FIFO. A longer retry window would mostly add latency for hot
Instagram contacts without making ordering more correct. After the lock
TTL has elapsed, processing without the lock is the better tradeoff than
dead-lettering customer messages.

app/jobs/mutex_application_job.rb

@@ -14,6 +14,19 @@
 class MutexApplicationJob < ApplicationJob
   class LockAcquisitionError < StandardError; end
 
+  def self.retry_on_lock_conflict(wait:, attempts:, on_exhaustion: :raise)
+    retry_on LockAcquisitionError, wait: wait, attempts: attempts do |job, error|
+      raise error if on_exhaustion == :raise
+
+      job.public_send(on_exhaustion, *job.arguments)
+    end
+  end
+
+  # Redis::LockManager#unlock is not owner-checked. If a job runs past the TTL,
+  # Redis can expire the key, a newer job can acquire it, and the older job can
+  # then delete the newer job's lock on unlock. Current mutex users treat locks as
+  # short race dampeners, so this is acceptable for now. Future iterations should
+  # move Redis::LockManager to token-checked unlocks.
   def with_lock(lock_key, timeout = Redis::LockManager::LOCK_TIMEOUT)
     lock_manager = Redis::LockManager.new
 

app/jobs/webhooks/instagram_events_job.rb

@@ -1,6 +1,14 @@
 class Webhooks::InstagramEventsJob < MutexApplicationJob
   queue_as :default
-  retry_on LockAcquisitionError, wait: 1.second, attempts: 8
+  # This lock is only a short race dampener for first-message conversation creation.
+  # ContactInbox creation is already protected by a unique index, but conversation
+  # lookup is `find active conversation || create`, so concurrent first messages from
+  # the same IG contact can create duplicate conversations.
+  #
+  # ActiveJob retries are not FIFO, so a longer retry window does not preserve message
+  # order. Use deterministic backoff so the final attempt happens after the 3s lock TTL,
+  # then process without the lock instead of dropping the webhook.
+  retry_on_lock_conflict wait: ->(executions) { executions.seconds }, attempts: 3, on_exhaustion: :process_without_lock
 
   # @return [Array] We will support further events like reaction or seen in future
   SUPPORTED_EVENTS = [:message, :read].freeze
@@ -9,11 +17,19 @@ class Webhooks::InstagramEventsJob < MutexApplicationJob
     @entries = entries
 
     key = format(::Redis::Alfred::IG_MESSAGE_MUTEX, sender_id: contact_instagram_id, ig_account_id: ig_account_id)
-    with_lock(key) do
+    # Keep the lock TTL just long enough for the first job to fetch profile data and
+    # create the contact/conversation. A longer TTL would add user-visible latency for
+    # hot contacts without giving us ordering guarantees.
+    with_lock(key, 3.seconds) do
       process_entries(entries)
     end
   end
 
+  def process_without_lock(entries)
+    Rails.logger.warn("[#{self.class.name}] Processing without lock after lock retry exhaustion")
+    process_entries(entries)
+  end
+
   # https://developers.facebook.com/docs/messenger-platform/instagram/features/webhook
   def process_entries(entries)
     entries.each do |entry|

spec/jobs/mutex_application_job_spec.rb

@@ -55,4 +55,57 @@ RSpec.describe MutexApplicationJob do
       end.to raise_error(StandardError)
     end
   end
+
+  describe '.retry_on_lock_conflict' do
+    let(:job_class) do
+      Class.new(described_class) do
+        retry_on_lock_conflict wait: 1.second, attempts: 1, on_exhaustion: :process_without_lock
+
+        attr_reader :fallback_args
+
+        def perform(lock_key, _payload)
+          with_lock(lock_key) { raise 'lock should not be acquired' }
+        end
+
+        def process_without_lock(lock_key, payload)
+          @fallback_args = [lock_key, payload]
+        end
+      end
+    end
+
+    let(:payload) { { 'message' => 'hello' } }
+
+    before do
+      stub_const('LockConflictTestJob', job_class)
+    end
+
+    it 'runs the configured handler with the original job arguments when lock retries are exhausted' do
+      allow(lock_manager).to receive(:lock).with(lock_key, Redis::LockManager::LOCK_TIMEOUT).and_return(false)
+
+      job = job_class.new(lock_key, payload)
+
+      expect { job.perform_now }.not_to raise_error
+      expect(job.fallback_args).to eq([lock_key, payload])
+    end
+
+    context 'without an exhaustion handler' do
+      let(:job_class) do
+        Class.new(described_class) do
+          retry_on_lock_conflict wait: 1.second, attempts: 1
+
+          def perform(lock_key)
+            with_lock(lock_key) { raise 'lock should not be acquired' }
+          end
+        end
+      end
+
+      it 'raises the lock acquisition error when retries are exhausted' do
+        allow(lock_manager).to receive(:lock).with(lock_key, Redis::LockManager::LOCK_TIMEOUT).and_return(false)
+
+        expect do
+          job_class.perform_now(lock_key)
+        end.to raise_error(StandardError) { |error| expect(error.class.name).to eq('MutexApplicationJob::LockAcquisitionError') }
+      end
+    end
+  end
 end