diff --git a/enterprise/app/jobs/enterprise/webhooks/whatsapp_events_job.rb b/enterprise/app/jobs/enterprise/webhooks/whatsapp_events_job.rb index 2a0db6f98..b8053c1a1 100644 --- a/enterprise/app/jobs/enterprise/webhooks/whatsapp_events_job.rb +++ b/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:` 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 diff --git a/enterprise/app/services/whatsapp/call_permission_reply_service.rb b/enterprise/app/services/whatsapp/call_permission_reply_service.rb index 95798b8d7..5f3c883fc 100644 --- a/enterprise/app/services/whatsapp/call_permission_reply_service.rb +++ b/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) diff --git a/enterprise/app/services/whatsapp/incoming_call_service.rb b/enterprise/app/services/whatsapp/incoming_call_service.rb index 7e3d58e1e..d3685e8a3 100644 --- a/enterprise/app/services/whatsapp/incoming_call_service.rb +++ b/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) diff --git a/spec/enterprise/services/whatsapp/call_permission_reply_service_spec.rb b/spec/enterprise/services/whatsapp/call_permission_reply_service_spec.rb index d8a2aca78..d52c1bb60 100644 --- a/spec/enterprise/services/whatsapp/call_permission_reply_service_spec.rb +++ b/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 diff --git a/spec/enterprise/services/whatsapp/incoming_call_service_spec.rb b/spec/enterprise/services/whatsapp/incoming_call_service_spec.rb index 6ca3aa489..bf416bb29 100644 --- a/spec/enterprise/services/whatsapp/incoming_call_service_spec.rb +++ b/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)