fix(voice): tighten WhatsApp call lifecycle handling and per-call locking

enterprise/app/jobs/enterprise/webhooks/whatsapp_events_job.rb

@@ -8,19 +8,13 @@ module Enterprise::Webhooks::WhatsappEventsJob
 
   private
 
-  # OSS contact_sender_id returns nil for `field: 'calls'` payloads, which makes
-  # the parent job bypass the per-(inbox, sender) mutex. That lets a fast
-  # `terminate` webhook be processed before the `connect` transaction commits,
-  # silently dropping the terminate and stranding the call as `ringing`. Falling
-  # through to `call:<id>` reuses the existing mutex to serialize connect and
-  # terminate for the same call.
+  # Call webhooks don't share the message-mutex sender_id; we lock per-call_id
+  # inside `handle_call_events` instead so multi-call batches don't share a
+  # single lock keyed on the first call's id.
   def contact_sender_id(params)
-    super.presence || call_event_sender_id(params)
-  end
+    return nil if call_event?(params)
 
-  def call_event_sender_id(params)
-    call_id = params.dig(:entry, 0, :changes, 0, :value, :calls, 0, :id)
-    call_id.present? ? "call:#{call_id}" : nil
+    super
   end
 
   def call_event?(params)
@@ -32,18 +26,21 @@ module Enterprise::Webhooks::WhatsappEventsJob
     message&.dig(:type) == 'interactive' && message&.dig(:interactive, :type) == 'call_permission_reply'
   end
 
+  # Acquire a per-call_id mutex around each call payload so that connect /
+  # terminate webhooks for the same call are serialized — even when Meta
+  # batches multiple calls in one webhook envelope.
   def handle_call_events(channel, params)
-    Whatsapp::IncomingCallService.new(
-      inbox: channel.inbox,
-      params: extract_call_params(params)
-    ).perform
+    calls = params.dig(:entry, 0, :changes, 0, :value, :calls) || []
+    calls.each do |call_payload|
+      lock_key = format(::Redis::Alfred::WHATSAPP_MESSAGE_MUTEX,
+                        inbox_id: channel.inbox.id, sender_id: "call:#{call_payload[:id]}")
+      with_lock(lock_key, 30.seconds) do
+        Whatsapp::IncomingCallService.new(inbox: channel.inbox, params: { calls: [call_payload] }).perform
+      end
+    end
   end
 
   def handle_call_permission_reply(channel, params)
     Whatsapp::CallPermissionReplyService.new(inbox: channel.inbox, params: params).perform
   end
-
-  def extract_call_params(params)
-    params.dig(:entry, 0, :changes, 0, :value) || {}
-  end
 end

enterprise/app/services/whatsapp/call_permission_reply_service.rb

@@ -37,8 +37,16 @@ class Whatsapp::CallPermissionReplyService
          .first&.contact
   end
 
+  # Pick the conversation that actually requested permission, not just any open
+  # one — a contact with multiple open threads in the same inbox would otherwise
+  # have the wrong conversation cleared and broadcast.
   def find_active_conversation(contact)
-    inbox.conversations.where(contact: contact).where.not(status: :resolved).order(:created_at).last
+    inbox.conversations
+         .where(contact: contact)
+         .where.not(status: :resolved)
+         .where("additional_attributes ->> 'call_permission_requested_at' IS NOT NULL")
+         .order(:created_at)
+         .last
   end
 
   def clear_permission_flag(conversation)

enterprise/app/services/whatsapp/incoming_call_service.rb

@@ -21,13 +21,21 @@ class Whatsapp::IncomingCallService
 
   def handle_call_connect(call_payload)
     existing = Call.whatsapp.find_by(provider_call_id: call_payload[:id])
-    existing ? handle_outbound_connect(existing, call_payload) : handle_inbound_connect(call_payload)
+    if existing&.outgoing?
+      handle_outbound_connect(existing, call_payload)
+    elsif existing
+      Rails.logger.info "[WHATSAPP CALL] Duplicate inbound connect for #{call_payload[:id]}; ignoring"
+    else
+      handle_inbound_connect(call_payload)
+    end
   rescue ActiveRecord::RecordNotUnique
     Rails.logger.warn "[WHATSAPP CALL] Duplicate provider_call_id received: #{call_payload[:id]}"
   end
 
   def handle_outbound_connect(call, call_payload)
-    return if call.in_progress?
+    # `in_progress?` skips duplicate connect deliveries; `terminal?` stops a
+    # delayed connect from reopening an already-ended call.
+    return if call.in_progress? || call.terminal?
 
     sdp_answer = fix_sdp_setup(call_payload.dig(:session, :sdp))
     call.update!(status: 'in_progress', started_at: Time.current,
@@ -66,7 +74,9 @@ class Whatsapp::IncomingCallService
   end
 
   def answered?(call, duration)
-    call.in_progress? || duration.to_i.positive? || call.accepted_by_agent_id.present?
+    # `accepted_by_agent_id` only signals an answered call for INBOUND — outbound
+    # calls have the initiating agent set before the contact picks up.
+    call.in_progress? || duration.to_i.positive? || (call.incoming? && call.accepted_by_agent_id.present?)
   end
 
   def update_conversation_call_status(conversation, call_status, direction)

spec/enterprise/services/whatsapp/call_permission_reply_service_spec.rb

@@ -59,4 +59,18 @@ describe Whatsapp::CallPermissionReplyService do
 
     expect(ActionCable.server).not_to have_received(:broadcast)
   end
+
+  it 'targets the conversation that requested permission, not just any open one' do
+    other_open = create(:conversation, account: account, inbox: inbox, contact: contact, contact_inbox: contact_inbox,
+                                       status: :open, additional_attributes: {})
+    allow(ActionCable.server).to receive(:broadcast)
+
+    described_class.new(inbox: inbox, params: reply_params(response: 'accept')).perform
+
+    expect(other_open.reload.additional_attributes).to eq({})
+    expect(ActionCable.server).to have_received(:broadcast).with(
+      "account_#{account.id}",
+      hash_including(data: hash_including(conversation_id: conversation.id))
+    )
+  end
 end

spec/enterprise/services/whatsapp/incoming_call_service_spec.rb

@@ -101,6 +101,63 @@ describe Whatsapp::IncomingCallService do
     end
   end
 
+  describe 'duplicate inbound connect' do
+    let!(:call) do
+      conversation = create(:conversation, account: account, inbox: inbox)
+      create(:call, account: account, inbox: inbox, conversation: conversation, contact: conversation.contact,
+                    provider: :whatsapp, direction: :incoming, status: 'ringing', provider_call_id: provider_call_id)
+    end
+
+    it 'logs and ignores rather than treating it as outbound' do
+      allow(Rails.logger).to receive(:info)
+      allow(ActionCable.server).to receive(:broadcast)
+      params = call_payload(event: 'connect', session: { sdp: 'sdp_x', sdp_type: 'offer' })
+
+      described_class.new(inbox: inbox, params: params).perform
+
+      expect(call.reload).to have_attributes(status: 'ringing')
+      expect(Rails.logger).to have_received(:info).with(/Duplicate inbound connect/)
+      expect(ActionCable.server).not_to have_received(:broadcast)
+    end
+  end
+
+  describe 'connect arriving after terminal status' do
+    let!(:call) do
+      conversation = create(:conversation, account: account, inbox: inbox)
+      create(:call, account: account, inbox: inbox, conversation: conversation, contact: conversation.contact,
+                    provider: :whatsapp, direction: :outgoing, status: 'completed', provider_call_id: provider_call_id)
+    end
+
+    it 'does not reopen a completed outbound call' do
+      allow(ActionCable.server).to receive(:broadcast)
+      params = call_payload(event: 'connect', session: { sdp: 'late_sdp', sdp_type: 'answer' })
+
+      described_class.new(inbox: inbox, params: params).perform
+
+      expect(call.reload.status).to eq('completed')
+      expect(ActionCable.server).not_to have_received(:broadcast)
+    end
+  end
+
+  describe 'unanswered outbound call terminate' do
+    let!(:agent) { create(:user, account: account) }
+    let!(:call) do
+      conversation = create(:conversation, account: account, inbox: inbox)
+      create(:call, account: account, inbox: inbox, conversation: conversation, contact: conversation.contact,
+                    provider: :whatsapp, direction: :outgoing, status: 'ringing',
+                    accepted_by_agent: agent, provider_call_id: provider_call_id)
+    end
+
+    it 'marks the call as no_answer even though accepted_by_agent_id is set' do
+      allow(ActionCable.server).to receive(:broadcast)
+      params = call_payload(event: 'terminate', duration: 0, terminate_reason: 'no_answer')
+
+      described_class.new(inbox: inbox, params: params).perform
+
+      expect(call.reload.status).to eq('no_answer')
+    end
+  end
+
   describe 'unknown event' do
     it 'logs a warning and does not raise' do
       allow(Rails.logger).to receive(:warn)