refactor: use MutexApplicationJob

app/jobs/mutex_application_job.rb

@@ -0,0 +1,35 @@
+# MutexApplicationJob serves as a base class for jobs that require distributed locking mechanisms.
+# It abstracts the locking logic using Redis and ensures that a block of code can be executed with
+# mutual exclusion.
+#
+# The primary mechanism provided is the `with_lock` method, which accepts a key format and associated
+# arguments. This method attempts to acquire a lock using the generated key, and if successful, it
+# executes the provided block of code. If the lock cannot be acquired, it raises a LockAcquisitionError.
+#
+# To use this class, inherit from MutexApplicationJob and make use of the `with_lock` method in the
+# `perform` method of the derived job class.
+#
+# Also see, retry mechanism here: https://edgeapi.rubyonrails.org/classes/ActiveJob/Exceptions/ClassMethods.html#method-i-retry_on
+#
+class MutexApplicationJob < ApplicationJob
+  class LockAcquisitionError < StandardError; end
+
+  def with_lock(key_format, *args)
+    lock_key = format(key_format, *args)
+    lock_manager = Redis::LockManager.new
+
+    if lock_manager.locked?(lock_key)
+      Rails.logger.error "[#{self.class.name}] Failed to acquire lock on attempt #{executions + 1}: #{lock_key}"
+      raise LockAcquisitionError, "Failed to acquire lock for key: #{lock_key}"
+    end
+
+    begin
+      lock_manager.lock(lock_key)
+      Rails.logger.info "[#{self.class.name}] Acquired lock for: #{lock_key}"
+      yield
+    ensure
+      # Ensure that the lock is released even if there's an error in processing
+      lock_manager.unlock(lock_key)
+    end
+  end
+end

app/jobs/send_on_slack_job.rb

@@ -1,7 +1,10 @@
-class SendOnSlackJob < ApplicationJob
+class SendOnSlackJob < MutexApplicationJob
   queue_as :medium
+  retry_on LockAcquisitionError, wait: 3.seconds, attempts: 5
 
   def perform(message, hook)
-    Integrations::Slack::SendOnSlackService.new(message: message, hook: hook).perform
+    with_lock(::Redis::Alfred::SLACK_MESSAGE_MUTEX, sender_id: message.sender_id, reference_id: hook.reference_id) do
+      Integrations::Slack::SendOnSlackService.new(message: message, hook: hook).perform
+    end
   end
 end

app/jobs/webhooks/facebook_events_job.rb

@@ -1,8 +1,12 @@
 class Webhooks::FacebookEventsJob < ApplicationJob
   queue_as :default
+  retry_on LockAcquisitionError, wait: 2.seconds, attempts: 5
 
   def perform(message)
     response = ::Integrations::Facebook::MessageParser.new(message)
-    ::Integrations::Facebook::MessageCreator.new(response).perform
+
+    with_lock(::Redis::Alfred::FACEBOOK_MESSAGE_MUTEX, sender_id: response.sender_id, recipient_id: response.recipient_id) do
+      ::Integrations::Facebook::MessageCreator.new(response).perform
+    end
   end
 end

lib/redis/redis_keys.rb

@@ -33,4 +33,9 @@ module Redis::RedisKeys
   MESSAGE_SOURCE_KEY = 'MESSAGE_SOURCE_KEY::%<id>s'.freeze
   CUSTOM_FILTER_RECORDS_COUNT_KEY = 'CUSTOM_FILTER::%<account_id>d::%<user_id>d::%<filter_id>d'.freeze
   OPENAI_CONVERSATION_KEY = 'OPEN_AI_CONVERSATION_KEY::%<event_name>s::%<conversation_id>d::%<updated_at>d'.freeze
+
+  ## Sempahores / Locks
+  # 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
+  SLACK_MESSAGE_MUTEX = 'SLACK_MESSAGE_LOCK::%<sender_id>s::%<reference_id>s'.freeze
 end