rodrigo/iachat
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

release v4.12.0-fazer-ai.47 (#259)

Browse Source 
* fix(whatsapp): add idempotent message sending to prevent duplicates on timeout retry

When sending media messages via Baileys, Net::ReadTimeout causes Sidekiq to
retry the job, potentially sending the same message multiple times. This adds
a chatwootMessageId parameter to the Baileys API request, enabling server-side
deduplication via Redis. Also increases HTTP timeout to 120s and channel lock
to 130s to reduce false timeouts.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: address review feedback

- Use error.class.name assertions for parallel/reloading safety
- Assert reconnect endpoint was not called on 409 (stronger assertion)

* fix: address review feedback (round 2)

- Only release channel lock in ensure if it was actually acquired (prevents
  clearing another worker's lock on timeout)
- Assert chatwootMessageId in reproduction spec body matcher

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Gabriel Jablonski 2026-04-08 20:52:26 -03:00 committed by GitHub
parent dba5379d5e
commit 8cf6e8907f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 197 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
app/helpers/baileys_helper.rb
View File

@ -1,6 +1,6 @@
module BaileysHelper module BaileysHelper
CHANNEL_LOCK_ON_OUTGOING_MESSAGE_KEY = 'BAILEYS::CHANNEL_LOCK_ON_OUTGOING_MESSAGE::%<channel_id>s'.freeze CHANNEL_LOCK_ON_OUTGOING_MESSAGE_KEY = 'BAILEYS::CHANNEL_LOCK_ON_OUTGOING_MESSAGE::%<channel_id>s'.freeze
CHANNEL_LOCK_ON_OUTGOING_MESSAGE_TIMEOUT = 60.seconds CHANNEL_LOCK_ON_OUTGOING_MESSAGE_TIMEOUT = 130.seconds
def baileys_extract_message_timestamp(timestamp) def baileys_extract_message_timestamp(timestamp)
# NOTE: Timestamp might be in this format {"low"=>1748003165, "high"=>0, "unsigned"=>true} # NOTE: Timestamp might be in this format {"low"=>1748003165, "high"=>0, "unsigned"=>true}
@ -35,7 +35,7 @@ module BaileysHelper
yield yield
ensure ensure
baileys_clear_channel_lock_on_outgoing_message(channel_id) baileys_clear_channel_lock_on_outgoing_message(channel_id) if lock_acquired
end end
private private

10
app/services/whatsapp/providers/whatsapp_baileys_service.rb
View File

@ -4,6 +4,7 @@ class Whatsapp::Providers::WhatsappBaileysService < Whatsapp::Providers::BaseSer
class MessageContentTypeNotSupported < StandardError; end class MessageContentTypeNotSupported < StandardError; end
class ProviderUnavailableError < StandardError; end class ProviderUnavailableError < StandardError; end
class GroupParticipantNotAllowedError < StandardError; end class GroupParticipantNotAllowedError < StandardError; end
class MessageAlreadyProcessingError < StandardError; end
DEFAULT_CLIENT_NAME = ENV.fetch('BAILEYS_PROVIDER_DEFAULT_CLIENT_NAME', nil) DEFAULT_CLIENT_NAME = ENV.fetch('BAILEYS_PROVIDER_DEFAULT_CLIENT_NAME', nil)
DEFAULT_URL = ENV.fetch('BAILEYS_PROVIDER_DEFAULT_URL', nil) DEFAULT_URL = ENV.fetch('BAILEYS_PROVIDER_DEFAULT_URL', nil)
@ -567,10 +568,13 @@ class Whatsapp::Providers::WhatsappBaileysService < Whatsapp::Providers::BaseSer
headers: api_headers, headers: api_headers,
body: { body: {
jid: remote_jid, jid: remote_jid,
messageContent: @message_content messageContent: @message_content,
}.to_json chatwootMessageId: @message.id
}.to_json,
timeout: 120
) )
raise MessageAlreadyProcessingError if response.code == 409
raise ProviderUnavailableError unless process_response(response) raise ProviderUnavailableError unless process_response(response)
update_external_created_at(response) update_external_created_at(response)
@ -816,6 +820,8 @@ class Whatsapp::Providers::WhatsappBaileysService < Whatsapp::Providers::BaseSer
define_method(method_name) do |*args, **kwargs, &block| define_method(method_name) do |*args, **kwargs, &block|
original_method.bind_call(self, *args, **kwargs, &block) original_method.bind_call(self, *args, **kwargs, &block)
rescue MessageAlreadyProcessingError
raise
rescue StandardError => e rescue StandardError => e
handle_channel_error handle_channel_error
raise e raise e

4
spec/helpers/baileys_helper_spec.rb
View File

@ -74,7 +74,7 @@ RSpec.describe BaileysHelper do
end end
context 'when lock is not acquired within timeout' do context 'when lock is not acquired within timeout' do
it 'still executes the block and clears the lock' do it 'still executes the block but does not clear the lock' do
freeze_time freeze_time
allow(Redis::Alfred).to receive(:set).and_return(false) allow(Redis::Alfred).to receive(:set).and_return(false)
allow(self).to receive(:sleep) { travel_to 1.second.from_now } allow(self).to receive(:sleep) { travel_to 1.second.from_now }
@ -83,7 +83,7 @@ RSpec.describe BaileysHelper do
expect(Redis::Alfred).to have_received(:set) expect(Redis::Alfred).to have_received(:set)
.with(lock_key, 1, nx: true, ex: timeout) .with(lock_key, 1, nx: true, ex: timeout)
.exactly(BaileysHelper::CHANNEL_LOCK_ON_OUTGOING_MESSAGE_TIMEOUT.to_i) .exactly(BaileysHelper::CHANNEL_LOCK_ON_OUTGOING_MESSAGE_TIMEOUT.to_i)
expect(Redis::Alfred).to have_received(:delete).once.with(lock_key) expect(Redis::Alfred).not_to have_received(:delete)
end end
end end
end end

281
spec/services/whatsapp/providers/whatsapp_baileys_service_spec.rb
View File

@ -205,10 +205,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { text: 'Please rate us http://example.com/survey' } messageContent: { text: 'Please rate us http://example.com/survey' }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -240,10 +240,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { fileName: 'image.png', caption: message.content, image: base64_image } messageContent: { fileName: 'image.png', caption: message.content, image: base64_image }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -261,10 +261,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { fileName: 'image.png', image: base64_image } messageContent: { fileName: 'image.png', image: base64_image }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -296,10 +296,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { fileName: 'audio.wav', caption: message.content, audio: base64_audio } messageContent: { fileName: 'audio.wav', caption: message.content, audio: base64_audio }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -318,10 +318,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { fileName: 'audio.wav', caption: message.content, audio: base64_audio, ptt: true } messageContent: { fileName: 'audio.wav', caption: message.content, audio: base64_audio, ptt: true }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -340,10 +340,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { text: message.content } messageContent: { text: message.content }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -360,10 +360,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { text: message.content } messageContent: { text: message.content }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -393,13 +393,13 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { react: { key: { id: message.source_id, messageContent: { react: { key: { id: message.source_id,
remoteJid: test_send_jid, remoteJid: test_send_jid,
fromMe: true }, fromMe: true },
text: '👍' } } text: '👍' } }
}.to_json }, reaction)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -419,13 +419,13 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { react: { key: { id: message.source_id, messageContent: { react: { key: { id: message.source_id,
remoteJid: test_send_jid, remoteJid: test_send_jid,
fromMe: false }, fromMe: false },
text: '👍' } } text: '👍' } }
}.to_json }, reaction)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -457,20 +457,20 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { messageContent: {
text: 'Reply text', text: 'Reply text',
quotedMessage: { quotedMessage: {
key: { key: {
id: 'original_msg_123', id: 'original_msg_123',
remoteJid: test_send_jid, remoteJid: test_send_jid,
fromMe: true fromMe: true
}, },
message: { conversation: 'Original text' } message: { conversation: 'Original text' }
} }
} }
}.to_json }, reply_message)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -492,20 +492,20 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { messageContent: {
text: 'Reply to incoming', text: 'Reply to incoming',
quotedMessage: { quotedMessage: {
key: { key: {
id: 'incoming_msg_456', id: 'incoming_msg_456',
remoteJid: test_send_jid, remoteJid: test_send_jid,
fromMe: false fromMe: false
}, },
message: { conversation: 'Incoming text' } message: { conversation: 'Incoming text' }
} }
} }
}.to_json }, reply_message)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -530,20 +530,20 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { messageContent: {
text: 'Nice image!', text: 'Nice image!',
quotedMessage: { quotedMessage: {
key: { key: {
id: 'image_msg_789', id: 'image_msg_789',
remoteJid: test_send_jid, remoteJid: test_send_jid,
fromMe: false fromMe: false
}, },
message: { imageMessage: { caption: 'Check this image' } } message: { imageMessage: { caption: 'Check this image' } }
} }
} }
}.to_json }, reply_message)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -562,10 +562,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: test_send_jid, jid: test_send_jid,
messageContent: { text: 'Regular message' } messageContent: { text: 'Regular message' }
}.to_json }, regular_message)
) )
.to_return( .to_return(
status: 200, status: 200,
@ -586,10 +586,10 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, request_path) stub_request(:post, request_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: group_jid, jid: group_jid,
messageContent: { text: message.content } messageContent: { text: message.content }
}.to_json })
) )
.to_return( .to_return(
status: 200, status: 200,
@ -621,6 +621,23 @@ describe Whatsapp::Providers::WhatsappBaileysService do
end.to raise_error(Whatsapp::Providers::WhatsappBaileysService::ProviderUnavailableError) end.to raise_error(Whatsapp::Providers::WhatsappBaileysService::ProviderUnavailableError)
end end
end end
context 'when server returns 409 (message already processing)' do
it 'raises MessageAlreadyProcessingError without triggering channel reconnection' do
stub_request(:post, request_path)
.to_return(status: 409, body: 'Message is already being processed')
setup_url = "#{whatsapp_channel.provider_config['provider_url']}/connections/#{whatsapp_channel.phone_number}"
expect do
service.send_message(test_send_phone_number, message)
end.to(raise_error do |error|
expect(error.class.name).to eq('Whatsapp::Providers::WhatsappBaileysService::MessageAlreadyProcessingError')
end)
expect(WebMock).not_to have_requested(:post, setup_url)
end
end
end end
describe '#media_url' do describe '#media_url' do
@ -1104,16 +1121,16 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, send_message_path) stub_request(:post, send_message_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: group_jid, jid: group_jid,
messageContent: { messageContent: {
text: 'World!', text: 'World!',
quotedMessage: { quotedMessage: {
key: { id: 'incoming_group_msg', remoteJid: group_jid, fromMe: false, participant: participant_lid }, key: { id: 'incoming_group_msg', remoteJid: group_jid, fromMe: false, participant: participant_lid },
message: { conversation: 'Hello' } message: { conversation: 'Hello' }
} }
} }
}.to_json }, reply_message)
) )
.to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json) .to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json)
@ -1131,16 +1148,16 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, send_message_path) stub_request(:post, send_message_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: group_jid, jid: group_jid,
messageContent: { messageContent: {
text: 'World!', text: 'World!',
quotedMessage: { quotedMessage: {
key: { id: 'outgoing_group_msg', remoteJid: group_jid, fromMe: true }, key: { id: 'outgoing_group_msg', remoteJid: group_jid, fromMe: true },
message: { conversation: 'Hello' } message: { conversation: 'Hello' }
} }
} }
}.to_json }, reply_message)
) )
.to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json) .to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json)
@ -1157,15 +1174,15 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, send_message_path) stub_request(:post, send_message_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: group_jid, jid: group_jid,
messageContent: { messageContent: {
react: { react: {
key: { id: original_message.source_id, remoteJid: group_jid, fromMe: false, participant: participant_lid }, key: { id: original_message.source_id, remoteJid: group_jid, fromMe: false, participant: participant_lid },
text: '👍' text: '👍'
} }
} }
}.to_json }, reaction)
) )
.to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json) .to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json)
@ -1182,15 +1199,15 @@ describe Whatsapp::Providers::WhatsappBaileysService do
stub_request(:post, send_message_path) stub_request(:post, send_message_path)
.with( .with(
headers: stub_headers(whatsapp_channel), headers: stub_headers(whatsapp_channel),
body: { body: send_message_body({
jid: group_jid, jid: group_jid,
messageContent: { messageContent: {
react: { react: {
key: { id: original_message.source_id, remoteJid: group_jid, fromMe: true }, key: { id: original_message.source_id, remoteJid: group_jid, fromMe: true },
text: '❤️' text: '❤️'
} }
} }
}.to_json }, reaction)
) )
.to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json) .to_return(status: 200, headers: { 'Content-Type' => 'application/json' }, body: result_body.to_json)
@ -1801,4 +1818,8 @@ describe Whatsapp::Providers::WhatsappBaileysService do
'x-api-key' => channel.provider_config['api_key'] 'x-api-key' => channel.provider_config['api_key']
} }
end end
def send_message_body(hash, msg = message)
hash.merge(chatwootMessageId: msg.id).to_json
end
end end

34
spec/services/whatsapp/send_on_whatsapp_service_spec.rb
View File

@ -432,6 +432,40 @@ describe Whatsapp::SendOnWhatsappService do
expect(message.reload.source_id).to eq('msg_group') expect(message.reload.source_id).to eq('msg_group')
end end
describe 'duplicate send on Net::ReadTimeout retry' do
let(:send_message_url) do
"#{whatsapp_channel.provider_config['provider_url']}/connections/#{whatsapp_channel.phone_number}/send-message"
end
let(:setup_url) do
"#{whatsapp_channel.provider_config['provider_url']}/connections/#{whatsapp_channel.phone_number}"
end
let(:success_body) { { data: { key: { id: 'wa_msg_123' }, messageTimestamp: '123' } }.to_json }
before do
conversation.contact.update!(phone_number: '+123456789')
create(:message, message_type: :incoming, content: 'hi', conversation: conversation)
stub_request(:post, setup_url).to_return(status: 200, body: '', headers: {})
end
it 'sends the message twice when first attempt times out and job is retried' do
message = create(:message, message_type: :outgoing, content: 'test', conversation: conversation, source_id: nil)
stub = stub_request(:post, send_message_url)
.with { |req| JSON.parse(req.body)['chatwootMessageId'] == message.id }
.to_raise(Net::ReadTimeout.new('Net::ReadTimeout'))
.then
.to_return(status: 200, body: success_body, headers: { 'Content-Type' => 'application/json' })
expect { SendReplyJob.perform_now(message.id) }.to(raise_error { |e| expect(e.class.name).to eq('Net::ReadTimeout') })
expect(message.reload.source_id).to be_nil
SendReplyJob.perform_now(message.id)
expect(message.reload.source_id).to eq('wa_msg_123')
expect(stub).to have_been_requested.twice
end
end
end end
context 'when provider is zapi' do context 'when provider is zapi' do