Compare commits

...
11 changed files with 257 additions and 147 deletions
+15
View File
@@ -0,0 +1,15 @@
class ConversationReplyEmailJob < ApplicationJob
queue_as :mailers
def perform(conversation_id, last_queued_id)
conversation = Conversation.find(conversation_id)
if conversation.messages.incoming&.last&.content_type == 'incoming_email'
ConversationReplyMailer.with(account: conversation.account).reply_without_summary(conversation, last_queued_id).deliver_later
else
ConversationReplyMailer.with(account: conversation.account).reply_with_summary(conversation, last_queued_id).deliver_later
end
Redis::Alfred.delete(format(::Redis::Alfred::CONVERSATION_MAILER_KEY, conversation_id: conversation.id))
end
end
@@ -12,7 +12,7 @@ class Messages::SendEmailNotificationService
# the worker never manages to clean up.
return unless Redis::Alfred.set(conversation_mail_key, message.id, nx: true, ex: 1.hour.to_i)
ConversationReplyEmailWorker.perform_in(2.minutes, conversation.id, message.id)
ConversationReplyEmailJob.set(wait: 2.minutes).perform_later(conversation.id, message.id)
end
private
@@ -1,29 +0,0 @@
# TODO: lets move this to active job, since thats what we use over all
class ConversationReplyEmailWorker
include Sidekiq::Worker
sidekiq_options queue: :mailers
def perform(conversation_id, last_queued_id)
@conversation = Conversation.find(conversation_id)
# send the email
if @conversation.messages.incoming&.last&.content_type == 'incoming_email'
ConversationReplyMailer.with(account: @conversation.account).reply_without_summary(@conversation, last_queued_id).deliver_later
else
ConversationReplyMailer.with(account: @conversation.account).reply_with_summary(@conversation, last_queued_id).deliver_later
end
# delete the redis set from the first new message on the conversation
Redis::Alfred.delete(conversation_mail_key)
end
private
def email_inbox?
@conversation.inbox&.inbox_type == 'Email'
end
def conversation_mail_key
format(::Redis::Alfred::CONVERSATION_MAILER_KEY, conversation_id: @conversation.id)
end
end
+2 -1
View File
@@ -9,7 +9,8 @@ end
# Logs whenever a job is pulled off Redis for execution.
class ChatwootDequeuedLogger
def call(_worker, job, queue)
Sidekiq.logger.info("Dequeued #{job['class']} #{job['jid']} from #{queue}")
payload = job['args'].first
Sidekiq.logger.info("Dequeued #{job['wrapped']} #{payload['job_id']} from #{queue}")
yield
end
end
+8 -5
View File
@@ -47,7 +47,7 @@ class Captain::Document < ApplicationRecord
after_create_commit :enqueue_crawl_job
after_create_commit :update_document_usage
after_destroy :update_document_usage
after_commit :enqueue_response_builder_job, on: :update, if: :should_enqueue_response_builder?
after_commit :enqueue_response_builder_job
scope :ordered, -> { order(created_at: :desc) }
scope :for_account, ->(account_id) { where(account_id: account_id) }
@@ -94,15 +94,18 @@ class Captain::Document < ApplicationRecord
end
def enqueue_response_builder_job
return if status != 'available'
return unless should_enqueue_response_builder?
Captain::Documents::ResponseBuilderJob.perform_later(self)
end
def should_enqueue_response_builder?
# Only enqueue when status changes to available
# Avoid re-enqueueing when metadata is updated by the job itself
saved_change_to_status? && status == 'available'
return false if destroyed?
return false unless available?
return saved_change_to_status? if pdf_document?
(saved_change_to_content? || saved_change_to_status?) && content.present?
end
def update_document_usage
@@ -1,8 +1,6 @@
require 'rails_helper'
RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
include ActiveJob::TestHelper
let!(:inbox) { create(:inbox) }
let!(:resolvable_pending_conversation) { create(:conversation, inbox: inbox, last_activity_at: 2.hours.ago, status: :pending) }
@@ -14,6 +12,7 @@ RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
before do
create(:captain_inbox, inbox: inbox, captain_assistant: captain_assistant)
stub_const('Limits::BULK_ACTIONS_LIMIT', 2)
inbox.reload
end
it 'queues the job' do
@@ -22,7 +21,7 @@ RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
end
it 'resolves only the eligible pending conversations' do
perform_enqueued_jobs { described_class.perform_later(inbox) }
described_class.perform_now(inbox)
expect(resolvable_pending_conversation.reload.status).to eq('resolved')
expect(recent_pending_conversation.reload.status).to eq('pending')
@@ -34,7 +33,7 @@ RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
captain_assistant.update!(config: { 'resolution_message' => custom_message })
expect do
perform_enqueued_jobs { described_class.perform_later(inbox) }
described_class.perform_now(inbox)
end.to change { resolvable_pending_conversation.messages.outgoing.reload.count }.by(1)
outgoing_message = resolvable_pending_conversation.messages.outgoing.last
@@ -44,7 +43,7 @@ RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
it 'creates an outgoing message with default auto resolution message if not configured' do
captain_assistant.update!(config: {})
perform_enqueued_jobs { described_class.perform_later(inbox) }
described_class.perform_now(inbox)
outgoing_message = resolvable_pending_conversation.messages.outgoing.last
expect(outgoing_message.content).to eq(
I18n.t('conversations.activity.auto_resolution_message')
@@ -52,11 +51,17 @@ RSpec.describe Captain::InboxPendingConversationsResolutionJob, type: :job do
end
it 'adds the correct activity message after resolution by Captain' do
perform_enqueued_jobs { described_class.perform_later(inbox) }
activity_message = resolvable_pending_conversation.messages.activity.last
expect(activity_message).not_to be_nil
expect(activity_message.content).to eq(
I18n.t('conversations.activity.captain.resolved', user_name: captain_assistant.name)
)
described_class.perform_now(inbox)
expected_content = I18n.t('conversations.activity.captain.resolved', user_name: captain_assistant.name)
expect(Conversations::ActivityMessageJob)
.to have_been_enqueued.with(
resolvable_pending_conversation,
{
account_id: resolvable_pending_conversation.account_id,
inbox_id: resolvable_pending_conversation.inbox_id,
message_type: :activity,
content: expected_content
}
)
end
end
@@ -82,4 +82,161 @@ RSpec.describe Captain::Document, type: :model do
end
end
end
describe 'response builder job callback' do
before { clear_enqueued_jobs }
describe 'non-PDF documents' do
it 'enqueues when created with available status and content' do
expect do
create(:captain_document, assistant: assistant, account: account, status: :available)
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue when created available without content' do
expect do
create(:captain_document, assistant: assistant, account: account, status: :available, content: nil)
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'enqueues when status transitions to available with existing content' do
document = create(:captain_document, assistant: assistant, account: account, status: :in_progress)
expect do
document.update!(status: :available)
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue when status transitions to available without content' do
document = create(
:captain_document,
assistant: assistant,
account: account,
status: :in_progress,
content: nil
)
expect do
document.update!(status: :available)
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'enqueues when content is populated on an available document' do
document = create(
:captain_document,
assistant: assistant,
account: account,
status: :available,
content: nil
)
clear_enqueued_jobs
expect do
document.update!(content: 'Fresh content from crawl')
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'enqueues when content changes on an available document' do
document = create(
:captain_document,
assistant: assistant,
account: account,
status: :available,
content: 'Initial content'
)
clear_enqueued_jobs
expect do
document.update!(content: 'Updated crawl content')
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue when content is cleared on an available document' do
document = create(
:captain_document,
assistant: assistant,
account: account,
status: :available,
content: 'Initial content'
)
clear_enqueued_jobs
expect do
document.update!(content: nil)
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue for metadata-only updates' do
document = create(:captain_document, assistant: assistant, account: account, status: :available)
clear_enqueued_jobs
expect do
document.update!(metadata: { 'title' => 'Updated Again' })
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue while document remains in progress' do
document = create(:captain_document, assistant: assistant, account: account, status: :in_progress)
expect do
document.update!(metadata: { 'title' => 'Updated' })
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
end
describe 'PDF documents' do
def build_pdf_document(status:, content:)
build(
:captain_document,
assistant: assistant,
account: account,
status: status,
content: content
).tap do |doc|
doc.pdf_file.attach(
io: StringIO.new('PDF content'),
filename: 'sample.pdf',
content_type: 'application/pdf'
)
end
end
it 'enqueues when created available without content' do
document = build_pdf_document(status: :available, content: nil)
expect do
document.save!
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'enqueues when status transitions to available' do
document = build_pdf_document(status: :in_progress, content: nil)
document.save!
clear_enqueued_jobs
expect do
document.update!(status: :available)
end.to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
it 'does not enqueue when content updates without status change' do
document = build_pdf_document(status: :available, content: nil)
document.save!
clear_enqueued_jobs
expect do
document.update!(content: 'Extracted PDF text')
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
end
it 'does not enqueue when the document is destroyed' do
document = create(:captain_document, assistant: assistant, account: account, status: :available)
clear_enqueued_jobs
expect do
document.destroy!
end.not_to have_enqueued_job(Captain::Documents::ResponseBuilderJob)
end
end
end
@@ -0,0 +1,33 @@
require 'rails_helper'
RSpec.describe ConversationReplyEmailJob, type: :job do
let(:conversation) { create(:conversation) }
let(:mailer) { double }
let(:mailer_action) { double }
before do
allow(Conversation).to receive(:find).and_return(conversation)
allow(ConversationReplyMailer).to receive(:with).and_return(mailer)
allow(mailer).to receive(:reply_with_summary).and_return(mailer_action)
allow(mailer).to receive(:reply_without_summary).and_return(mailer_action)
allow(mailer_action).to receive(:deliver_later).and_return(true)
end
it 'enqueues on mailers queue' do
ActiveJob::Base.queue_adapter = :test
expect do
described_class.perform_later(conversation.id, 123)
end.to have_enqueued_job(described_class).on_queue('mailers')
end
it 'calls reply_with_summary when last incoming message was not email' do
described_class.perform_now(conversation.id, 123)
expect(mailer).to have_received(:reply_with_summary)
end
it 'calls reply_without_summary when last incoming message was email' do
create(:message, conversation: conversation, message_type: :incoming, content_type: 'incoming_email')
described_class.perform_now(conversation.id, 123)
expect(mailer).to have_received(:reply_without_summary)
end
end
+8 -25
View File
@@ -327,14 +327,11 @@ RSpec.describe Message do
message.conversation.contact.update!(email: 'test@example.com')
message.message_type = 'outgoing'
# Perform jobs inline to test full integration
perform_enqueued_jobs do
message.save!
ActiveJob::Base.queue_adapter = :test
allow(Redis::Alfred).to receive(:set).and_return(true)
perform_enqueued_jobs(only: SendReplyJob) do
expect { message.save! }.to have_enqueued_job(ConversationReplyEmailJob).with(message.conversation.id, kind_of(Integer)).on_queue('mailers')
end
# Verify the email worker is eventually scheduled through the service
jobs_for_conversation_count = ConversationReplyEmailWorker.jobs.count { |job| job['args'].first == message.conversation.id }
expect(jobs_for_conversation_count).to eq(1)
end
it 'does not schedule email for website channel if continuity is disabled' do
@@ -345,15 +342,8 @@ RSpec.describe Message do
message.conversation.contact.update!(email: 'test@example.com')
message.message_type = 'outgoing'
initial_job_count = ConversationReplyEmailWorker.jobs.count { |job| job['args'].first == message.conversation.id }
perform_enqueued_jobs do
message.save!
end
# No new jobs should be scheduled for this conversation
jobs_for_conversation_count = ConversationReplyEmailWorker.jobs.count { |job| job['args'].first == message.conversation.id }
expect(jobs_for_conversation_count).to eq(initial_job_count)
ActiveJob::Base.queue_adapter = :test
expect { message.save! }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
it 'does not schedule email for private notes' do
@@ -363,15 +353,8 @@ RSpec.describe Message do
message.private = true
message.message_type = 'outgoing'
initial_job_count = ConversationReplyEmailWorker.jobs.count { |job| job['args'].first == message.conversation.id }
perform_enqueued_jobs do
message.save!
end
# No new jobs should be scheduled for this conversation
jobs_for_conversation_count = ConversationReplyEmailWorker.jobs.count { |job| job['args'].first == message.conversation.id }
expect(jobs_for_conversation_count).to eq(initial_job_count)
ActiveJob::Base.queue_adapter = :test
expect { message.save! }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
it 'calls SendReplyJob for all channels' do
@@ -14,17 +14,11 @@ describe Messages::SendEmailNotificationService do
before do
conversation.contact.update!(email: 'test@example.com')
allow(Redis::Alfred).to receive(:set).and_return(true)
allow(ConversationReplyEmailWorker).to receive(:perform_in)
ActiveJob::Base.queue_adapter = :test
end
it 'schedules ConversationReplyEmailWorker' do
service.perform
expect(ConversationReplyEmailWorker).to have_received(:perform_in).with(
2.minutes,
conversation.id,
message.id
)
it 'enqueues ConversationReplyEmailJob' do
expect { service.perform }.to have_enqueued_job(ConversationReplyEmailJob).with(conversation.id, message.id).on_queue('mailers')
end
it 'atomically sets redis key to prevent duplicate emails' do
@@ -40,10 +34,8 @@ describe Messages::SendEmailNotificationService do
allow(Redis::Alfred).to receive(:set).and_return(false)
end
it 'does not schedule worker' do
service.perform
expect(ConversationReplyEmailWorker).not_to have_received(:perform_in)
it 'does not enqueue job' do
expect { service.perform }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
it 'attempts atomic set once' do
@@ -62,7 +54,7 @@ describe Messages::SendEmailNotificationService do
conversation.contact.update!(email: 'test@example.com')
end
it 'prevents duplicate workers under race conditions' do
it 'prevents duplicate jobs under race conditions' do
# Create 5 threads that simultaneously try to enqueue workers for the same conversation
threads = Array.new(5) do
Thread.new do
@@ -73,24 +65,24 @@ describe Messages::SendEmailNotificationService do
threads.each(&:join)
# Only ONE worker should be scheduled despite 5 concurrent attempts
jobs_for_conversation = ConversationReplyEmailWorker.jobs.select { |job| job['args'].first == conversation.id }
# Only ONE job should be scheduled despite 5 concurrent attempts
jobs_for_conversation = ActiveJob::Base.queue_adapter.enqueued_jobs.select do |job|
job[:job] == ConversationReplyEmailJob && job[:args].first == conversation.id
end
expect(jobs_for_conversation.size).to eq(1)
end
end
context 'when email notification should not be sent' do
before do
allow(ConversationReplyEmailWorker).to receive(:perform_in)
ActiveJob::Base.queue_adapter = :test
end
context 'when message is not email notifiable' do
let(:message) { create(:message, conversation: conversation, message_type: 'incoming') }
it 'does not schedule worker' do
service.perform
expect(ConversationReplyEmailWorker).not_to have_received(:perform_in)
it 'does not enqueue job' do
expect { service.perform }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
end
@@ -102,10 +94,8 @@ describe Messages::SendEmailNotificationService do
conversation.contact.update!(email: nil)
end
it 'does not schedule worker' do
service.perform
expect(ConversationReplyEmailWorker).not_to have_received(:perform_in)
it 'does not enqueue job' do
expect { service.perform }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
end
@@ -117,10 +107,8 @@ describe Messages::SendEmailNotificationService do
conversation.contact.update!(email: 'test@example.com')
end
it 'does not schedule worker' do
service.perform
expect(ConversationReplyEmailWorker).not_to have_received(:perform_in)
it 'does not enqueue job' do
expect { service.perform }.not_to have_enqueued_job(ConversationReplyEmailJob)
end
end
end
@@ -1,46 +0,0 @@
require 'rails_helper'
Sidekiq::Testing.fake!
RSpec.describe ConversationReplyEmailWorker, type: :worker do
let(:conversation) { build(:conversation, display_id: nil) }
let(:message) { build(:message, conversation: conversation, content_type: 'incoming_email', inbox: conversation.inbox) }
let(:mailer) { double }
let(:mailer_action) { double }
describe 'testing ConversationSummaryEmailWorker' do
before do
conversation.save!
allow(Conversation).to receive(:find).and_return(conversation)
allow(ConversationReplyMailer).to receive(:with).and_return(mailer)
allow(ConversationReplyMailer).to receive(:with).and_return(mailer)
allow(mailer).to receive(:reply_with_summary).and_return(mailer_action)
allow(mailer).to receive(:reply_without_summary).and_return(mailer_action)
allow(mailer_action).to receive(:deliver_later).and_return(true)
end
it 'worker jobs are enqueued in the mailers queue' do
described_class.perform_async
expect(described_class.queue).to eq(:mailers)
end
it 'goes into the jobs array for testing environment' do
expect do
described_class.perform_async
end.to change(described_class.jobs, :size).by(1)
described_class.new.perform(1, message.id)
end
context 'with actions performed by the worker' do
it 'calls ConversationSummaryMailer#reply_with_summary when last incoming message was not email' do
described_class.new.perform(1, message.id)
expect(mailer).to have_received(:reply_with_summary)
end
it 'calls ConversationSummaryMailer#reply_without_summary when last incoming message was from email' do
message.save!
described_class.new.perform(1, message.id)
expect(mailer).to have_received(:reply_without_summary)
end
end
end
end