From a962308a23b91152da8079c69e46f67ea62c6d8e Mon Sep 17 00:00:00 2001
From: Shivam Mishra <scm.mymail@gmail.com>
Date: Mon, 21 Aug 2023 15:02:43 +0700
Subject: [PATCH] feat: add mutex for slack send job

---
 app/jobs/send_on_slack_job.rb | 19 ++++++++++++++++++-
 lib/redis/redis_keys.rb       |  5 +++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/app/jobs/send_on_slack_job.rb b/app/jobs/send_on_slack_job.rb
index e9b84731b..da535ed3c 100644
--- a/app/jobs/send_on_slack_job.rb
+++ b/app/jobs/send_on_slack_job.rb
@@ -1,7 +1,24 @@
 class SendOnSlackJob < ApplicationJob
+  class LockAcquisitionError < StandardError; end
+
   queue_as :medium
+  retry_on LockAcquisitionError, wait: 3.seconds, attempts: 5
 
   def perform(message, hook)
-    Integrations::Slack::SendOnSlackService.new(message: message, hook: hook).perform
+    lock_key = format(::Redis::Alfred::SLACK_MESSAGE_MUTEX, sender_id: message.sender_id, reference_id: hook.reference_id)
+    lock_manager = Redis::LockManager.new
+
+    if lock_manager.locked?(lock_key)
+      Rails.logger.error "[SendOnSlackJob] Failed to acquire lock on attempt #{executions + 1}: #{lock_key}"
+      raise LockAcquisitionError, "Failed to acquire lock for key: #{lock_key}"
+    end
+
+    begin
+      Rails.logger.info "[SendOnSlackJob] Acquired lock for: #{lock_key}"
+      Integrations::Slack::SendOnSlackService.new(message: message, hook: hook).perform
+    ensure
+      # Ensure that the lock is released even if there's an error in processing
+      lock_manager.unlock(lock_key)
+    end
   end
 end
diff --git a/lib/redis/redis_keys.rb b/lib/redis/redis_keys.rb
index bf024b349..65f966037 100644
--- a/lib/redis/redis_keys.rb
+++ b/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