From 9a780e8716fd94a48b1c4ba44b0e02d8ca4bf65c Mon Sep 17 00:00:00 2001 From: aakashb95 Date: Wed, 22 Jul 2026 13:14:48 +0530 Subject: [PATCH] fix(captain): serialize FAQ suggestion grouping --- .../jobs/captain/llm/conversation_faq_job.rb | 20 +++++++++++-- .../captain/llm/conversation_faq_service.rb | 22 +++++++++++---- lib/redis/redis_keys.rb | 1 + .../captain/llm/conversation_faq_job_spec.rb | 28 +++++++++++++++++++ 4 files changed, 63 insertions(+), 8 deletions(-) diff --git a/enterprise/app/jobs/captain/llm/conversation_faq_job.rb b/enterprise/app/jobs/captain/llm/conversation_faq_job.rb index 4948ccc48..9425c4b88 100644 --- a/enterprise/app/jobs/captain/llm/conversation_faq_job.rb +++ b/enterprise/app/jobs/captain/llm/conversation_faq_job.rb @@ -1,6 +1,10 @@ -class Captain::Llm::ConversationFaqJob < ApplicationJob +class Captain::Llm::ConversationFaqJob < MutexApplicationJob queue_as :low + LOCK_TIMEOUT = 10.minutes + + retry_on_lock_conflict wait: 30.seconds, attempts: 30 + def perform(conversation, assistant) inbox = conversation.inbox @@ -9,6 +13,18 @@ class Captain::Llm::ConversationFaqJob < ApplicationJob return if assistant.config['feature_faq'].blank? - Captain::Llm::ConversationFaqService.new(assistant, conversation).generate_suggestions + with_lock(lock_key(assistant, conversation), LOCK_TIMEOUT) do + Captain::Llm::ConversationFaqService.new(assistant, conversation).generate_suggestions + end + end + + private + + def lock_key(assistant, conversation) + format( + ::Redis::Alfred::CAPTAIN_CONVERSATION_FAQ_MUTEX, + assistant_id: assistant.id, + language: Captain::Llm::ConversationFaqService.language_for(conversation) + ) end end diff --git a/enterprise/app/services/captain/llm/conversation_faq_service.rb b/enterprise/app/services/captain/llm/conversation_faq_service.rb index 96da86ff7..e3ea9b4c8 100644 --- a/enterprise/app/services/captain/llm/conversation_faq_service.rb +++ b/enterprise/app/services/captain/llm/conversation_faq_service.rb @@ -6,6 +6,20 @@ class Captain::Llm::ConversationFaqService < Llm::BaseAiService LLM_FEATURE = 'conversation_faq_generation'.freeze FAQ_MATCH_MODEL = 'gpt-4.1-mini'.freeze + def self.language_for(conversation) + language = conversation.language.presence || conversation.account.locale.presence || I18n.default_locale.to_s + normalize_language(language) + end + + def self.normalize_language(language) + language.to_s.tr('-', '_').split('_').first.downcase + end + + def self.account_language_for(account) + normalize_language(account.locale.presence || I18n.default_locale.to_s) + end + private_class_method :normalize_language + def initialize(assistant, conversation) super(feature: LLM_FEATURE, account: conversation.account, fallback_model: Llm::Models.default_model_for(LLM_FEATURE)) @assistant = assistant @@ -180,15 +194,11 @@ class Captain::Llm::ConversationFaqService < Llm::BaseAiService end def faq_language - @faq_language ||= normalize_language(conversation.language.presence || conversation.account.locale.presence || I18n.default_locale.to_s) + @faq_language ||= self.class.language_for(conversation) end def account_language - @account_language ||= normalize_language(conversation.account.locale.presence || I18n.default_locale.to_s) - end - - def normalize_language(language) - language.to_s.tr('-', '_').split('_').first.downcase + @account_language ||= self.class.account_language_for(conversation.account) end def language_name(language) diff --git a/lib/redis/redis_keys.rb b/lib/redis/redis_keys.rb index b782270ef..c71f30f10 100644 --- a/lib/redis/redis_keys.rb +++ b/lib/redis/redis_keys.rb @@ -89,6 +89,7 @@ module Redis::RedisKeys WHATSAPP_MESSAGE_MUTEX = 'WHATSAPP_MESSAGE_CREATE_LOCK::%s::%s'.freeze CRM_PROCESS_MUTEX = 'CRM_PROCESS_MUTEX::%s'.freeze CAPTAIN_DOCUMENT_SYNC_MUTEX = 'CAPTAIN_DOCUMENT_SYNC_LOCK::%s'.freeze + CAPTAIN_CONVERSATION_FAQ_MUTEX = 'CAPTAIN_CONVERSATION_FAQ_LOCK::%s::%s'.freeze ## Auto Assignment Keys # Track conversation assignments to agents for rate limiting diff --git a/spec/enterprise/jobs/captain/llm/conversation_faq_job_spec.rb b/spec/enterprise/jobs/captain/llm/conversation_faq_job_spec.rb index 2bc13e45a..8925363e5 100644 --- a/spec/enterprise/jobs/captain/llm/conversation_faq_job_spec.rb +++ b/spec/enterprise/jobs/captain/llm/conversation_faq_job_spec.rb @@ -6,10 +6,14 @@ RSpec.describe Captain::Llm::ConversationFaqJob, type: :job do let(:assistant) { create(:captain_assistant, account: account, config: { feature_faq: true }) } let(:conversation) { create(:conversation, account: account, inbox: inbox, first_reply_created_at: Time.zone.now) } let(:faq_service) { instance_double(Captain::Llm::ConversationFaqService, generate_suggestions: []) } + let(:lock_manager) { instance_double(Redis::LockManager, lock: true, unlock: true) } + let(:lock_key) { "CAPTAIN_CONVERSATION_FAQ_LOCK::#{assistant.id}::en" } before do create(:captain_inbox, inbox: inbox, captain_assistant: assistant) conversation.update!(status: :resolved) + allow(Redis::LockManager).to receive(:new).and_return(lock_manager) + allow(Captain::Llm::ConversationFaqService).to receive(:new).and_return(faq_service) end describe '#perform' do @@ -25,5 +29,29 @@ RSpec.describe Captain::Llm::ConversationFaqJob, type: :job do described_class.perform_now(conversation, assistant) end + + it 'locks FAQ grouping for the assistant and normalized language' do + conversation.update!(additional_attributes: { conversation_language: 'pt-BR' }) + expected_key = "CAPTAIN_CONVERSATION_FAQ_LOCK::#{assistant.id}::pt" + + expect(lock_manager).to receive(:lock).with(expected_key, described_class::LOCK_TIMEOUT).and_return(true) + expect(lock_manager).to receive(:unlock).with(expected_key) + + described_class.perform_now(conversation, assistant) + end + + context 'when another job holds the grouping lock' do + before do + allow(lock_manager).to receive(:lock).with(lock_key, described_class::LOCK_TIMEOUT).and_return(false) + end + + it 'does not generate suggestions concurrently' do + expect(Captain::Llm::ConversationFaqService).not_to receive(:new) + + expect do + described_class.new.perform(conversation, assistant) + end.to raise_error(MutexApplicationJob::LockAcquisitionError) + end + end end end