LeadMagnet/leadchat
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat: mutex for InstagramEventsJob [CW-2447] (#7828)

Browse Source
This commit is contained in:
Shivam Mishra
2023-09-04 16:32:13 +07:00
committed by GitHub
parent 9ebabb9832
commit 336584c95a
3 changed files with 20 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/jobs/webhooks/facebook_events_job.rb
View File

@@ -1,6 +1,6 @@
class Webhooks::FacebookEventsJob < MutexApplicationJob class Webhooks::FacebookEventsJob < MutexApplicationJob
queue_as :default queue_as :default
retry_on LockAcquisitionError, wait: 1.second, attempts: 6 retry_on LockAcquisitionError, wait: 1.second, attempts: 8
def perform(message) def perform(message)
response = ::Integrations::Facebook::MessageParser.new(message) response = ::Integrations::Facebook::MessageParser.new(message)

21
app/jobs/webhooks/instagram_events_job.rb
View File

@@ -1,5 +1,6 @@
class Webhooks::InstagramEventsJob < ApplicationJob class Webhooks::InstagramEventsJob < MutexApplicationJob
queue_as :default queue_as :default
retry_on LockAcquisitionError, wait: 1.second, attempts: 8
include HTTParty include HTTParty
@@ -8,11 +9,17 @@ class Webhooks::InstagramEventsJob < ApplicationJob
# @return [Array] We will support further events like reaction or seen in future # @return [Array] We will support further events like reaction or seen in future
SUPPORTED_EVENTS = [:message].freeze SUPPORTED_EVENTS = [:message].freeze
# @see https://developers.facebook.com/docs/messenger-platform/instagram/features/webhook
def perform(entries) def perform(entries)
@entries = entries @entries = entries
@entries.each do |entry| with_lock(::Redis::Alfred::IG_MESSAGE_MUTEX, sender_id: sender_id, ig_account_id: ig_account_id) do
process_entries(entries)
end
end
# @see https://developers.facebook.com/docs/messenger-platform/instagram/features/webhook
def process_entries(entries)
entries.each do |entry|
entry = entry.with_indifferent_access entry = entry.with_indifferent_access
messages(entry).each do |messaging| messages(entry).each do |messaging|
send(@event_name, messaging) if event_name(messaging) send(@event_name, messaging) if event_name(messaging)
@@ -22,6 +29,14 @@ class Webhooks::InstagramEventsJob < ApplicationJob
private private
def ig_account_id
@entries&.first&.dig(:id)
end
def sender_id
@entries&.dig(0, :messaging, 0, :sender, :id)
end
def event_name(messaging) def event_name(messaging)
@event_name ||= SUPPORTED_EVENTS.find { |key| messaging.key?(key) } @event_name ||= SUPPORTED_EVENTS.find { |key| messaging.key?(key) }
end end

1
lib/redis/redis_keys.rb
View File

@@ -37,5 +37,6 @@ module Redis::RedisKeys
## Sempahores / Locks ## Sempahores / Locks
# We don't want to process messages from the same sender concurrently to prevent creating double conversations # We don't want to process messages from the same sender concurrently to prevent creating double conversations
FACEBOOK_MESSAGE_MUTEX = 'FB_MESSAGE_CREATE_LOCK::%<sender_id>s::%<recipient_id>s'.freeze FACEBOOK_MESSAGE_MUTEX = 'FB_MESSAGE_CREATE_LOCK::%<sender_id>s::%<recipient_id>s'.freeze
IG_MESSAGE_MUTEX = 'IG_MESSAGE_CREATE_LOCK::%<sender_id>s::%<ig_account_id>s'.freeze
SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<conversation_id>s::%<reference_id>s'.freeze SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<conversation_id>s::%<reference_id>s'.freeze
end end