diff --git a/app/jobs/slack_notification_job.rb b/app/jobs/slack_notification_job.rb index 63f1c62e..e8925513 100644 --- a/app/jobs/slack_notification_job.rb +++ b/app/jobs/slack_notification_job.rb @@ -2,44 +2,128 @@ class SlackNotificationJob < ApplicationJob queue_as :default retry_on SlackClient::Error, wait: :polynomially_longer, attempts: 5 - discard_on SlackClient::PermanentError + + DEBOUNCE_WINDOW = 2.minutes + CACHE_EXPIRY = DEBOUNCE_WINDOW + 1.minute + # retry_on below can span several minutes across its 5 attempts of + # polynomially-longer backoff — give batch/notified state room to outlive + # every retry, not just the debounce wait, or a late retry falls back to + # the wrong window or forgets who it already notified. + RETRY_STATE_EXPIRY = 30.minutes + + # Coalesces a burst of comments on the same thread (e.g. an agent posting + # several in a row) into a single delayed job. The first call in a window + # records where the batch starts and schedules the send; later calls + # within the window are no-ops because a send is already scheduled — the + # eventual perform picks up everything created since the batch started. + # + # batch_start is the triggering comment's own created_at, not the dispatch + # time this method runs at — this method is always called from a job + # that's already been enqueued and dequeued for a comment that already + # exists, so Time.current here would be later than that comment's + # created_at and silently exclude it from its own notification. + def self.debounce(comment_thread_id:, comment_created_at: nil) + pending_key = pending_key(comment_thread_id) + started = Rails.cache.write(pending_key, true, expires_in: CACHE_EXPIRY, unless_exist: true) + return unless started + + Rails.cache.write(batch_start_key(comment_thread_id), comment_created_at, expires_in: RETRY_STATE_EXPIRY) + Rails.cache.delete(notified_key(comment_thread_id)) + set(wait: DEBOUNCE_WINDOW).perform_later(comment_thread_id: comment_thread_id) + end + + def self.pending_key(comment_thread_id) + "slack_notification_job:pending:#{comment_thread_id}" + end + + def self.batch_start_key(comment_thread_id) + "slack_notification_job:batch_start:#{comment_thread_id}" + end + + def self.notified_key(comment_thread_id) + "slack_notification_job:notified:#{comment_thread_id}" + end def perform(comment_thread_id:) + send_batch(comment_thread_id) + # Only release the pending flag once the batch fully sends. A transient + # error (see retry_on above) skips this line, keeping the flag set so a + # comment arriving mid-retry doesn't start a second, overlapping batch + # that clobbers this one's batch_start/notified cache state. + Rails.cache.delete(self.class.pending_key(comment_thread_id)) + end + + private + + def send_batch(comment_thread_id) return unless SlackClient.configured? - thread = CoPlan::CommentThread.find(comment_thread_id) + thread = CoPlan::CommentThread.find_by(id: comment_thread_id) + return unless thread + + batch_start = Rails.cache.read(self.class.batch_start_key(comment_thread_id)) || DEBOUNCE_WINDOW.ago + new_comments = thread.comments.kept.where("coplan_comments.created_at >= ?", batch_start).order(:created_at, :id).to_a + return if new_comments.empty? + plan = thread.plan - coplan_author = plan.created_by_user - first_comment = thread.comments.order(:created_at, :id).first + recipients = recipients_for(thread, plan, new_comments) + return if recipients.empty? - return unless first_comment - return if first_comment.author_type == "human" && first_comment.author_id == coplan_author.id + text = compose_message(thread, plan, new_comments) + notified_key = self.class.notified_key(comment_thread_id) + already_notified = Rails.cache.read(notified_key) || [] - return unless coplan_author.email.present? + # A transient error retries the whole job (see retry_on above); track who + # already got a DM this batch so a retry doesn't double-notify recipients + # that succeeded before the recipient that raised. + recipients.each do |user| + next unless user.email.present? + next if already_notified.include?(user.id) - text = compose_message(thread, plan) - SlackClient.send_dm(email: coplan_author.email, text: text) + send_dm(user, text) + already_notified << user.id + Rails.cache.write(notified_key, already_notified, expires_in: RETRY_STATE_EXPIRY) + end end - private + # Everyone who's part of the conversation: the plan owner, plus anyone + # who's commented in the thread. A participant is skipped only if every + # comment in this batch is their own — if someone else said something + # new, they still hear about it even if they also commented in the batch. + def recipients_for(thread, plan, new_comments) + participant_ids = thread.comments.kept.where(author_type: %w[human local_agent]).distinct.pluck(:author_id) + participant_ids = (participant_ids + [ plan.created_by_user_id ]).uniq + + users_by_id = CoPlan::User.where(id: participant_ids).index_by(&:id) + participant_ids.filter_map do |user_id| + next unless new_comments.any? { |c| c.author_id != user_id } + users_by_id[user_id] + end + end + + # A permanent failure for one recipient (e.g. no Slack account for their + # email) shouldn't stop the rest of the batch from being notified. + def send_dm(user, text) + SlackClient.send_dm(email: user.email, text: text) + rescue SlackClient::PermanentError => e + Rails.logger.warn("[SlackNotificationJob] skipping #{user.id}: #{e.message}") + end - def compose_message(thread, plan) - comment_body = first_comment_body(thread).truncate(300) + def compose_message(thread, plan, new_comments) plan_url = CoPlan::Engine.routes.url_helpers.plan_url(plan, **default_url_options) + latest_body = (new_comments.last.body_markdown || "").truncate(300) - lines = [ "New comment on *#{plan.title}*:" ] + lines = [ + new_comments.size == 1 ? "New comment on *#{plan.title}*:" : "#{new_comments.size} new comments on *#{plan.title}*:" + ] if thread.anchor_text.present? lines << "> _#{thread.anchor_text.truncate(120)}_" end - lines << "> #{comment_body}" + lines << "> #{latest_body}" lines << plan_url lines.join("\n") end - def first_comment_body(thread) - thread.comments.order(:created_at, :id).first&.body_markdown || "" - end - def default_url_options Rails.application.config.action_mailer.default_url_options || { host: "localhost", port: 3000 } end diff --git a/config/initializers/coplan.rb b/config/initializers/coplan.rb index 7c1b2f82..de7f8df7 100644 --- a/config/initializers/coplan.rb +++ b/config/initializers/coplan.rb @@ -43,7 +43,12 @@ config.notification_handler = ->(event, payload) { case event when :comment_created - SlackNotificationJob.perform_later(comment_thread_id: payload[:comment_thread_id]) + # Debounced: an agent posting several comments in a row should produce + # one Slack DM per recipient, not one per comment. + SlackNotificationJob.debounce( + comment_thread_id: payload[:comment_thread_id], + comment_created_at: payload[:comment_created_at] + ) end } diff --git a/engine/app/models/coplan/comment.rb b/engine/app/models/coplan/comment.rb index b15d3101..6b6b45de 100644 --- a/engine/app/models/coplan/comment.rb +++ b/engine/app/models/coplan/comment.rb @@ -11,7 +11,7 @@ class Comment < ApplicationRecord validates :agent_name, length: { maximum: 20 }, allow_nil: true before_save :rewrite_plain_mentions, if: :body_markdown_changed? - after_create_commit :notify_plan_author, if: :first_comment_in_thread? + after_create_commit :notify_thread_participants after_create_commit :track_comment_created # Runs on save (not just create) so adding a mention via edit also # notifies. ProcessMentions uses find_or_create_by to dedupe. @@ -49,13 +49,16 @@ def first_comment_in_thread? # IDs are random UUIDs, not insertion-ordered, so we can't compare them # with `id < ?`. after_create_commit guarantees the row is persisted, so # a total count of 1 reliably means this comment opened the thread. - # Memoized: both the notify and analytics callbacks ask. + # Memoized since the analytics callback re-asks in the same request. return @first_comment_in_thread if defined?(@first_comment_in_thread) @first_comment_in_thread = comment_thread.comments.count == 1 end - def notify_plan_author - CoPlan::NotificationJob.perform_later("comment_created", { comment_thread_id: comment_thread_id }) + def notify_thread_participants + CoPlan::NotificationJob.perform_later( + "comment_created", + { comment_thread_id: comment_thread_id, comment_created_at: created_at } + ) end def track_comment_created diff --git a/spec/jobs/slack_notification_job_spec.rb b/spec/jobs/slack_notification_job_spec.rb index 9d965650..419fda8f 100644 --- a/spec/jobs/slack_notification_job_spec.rb +++ b/spec/jobs/slack_notification_job_spec.rb @@ -2,18 +2,19 @@ RSpec.describe SlackNotificationJob, type: :job do let(:plan_author) { create(:coplan_user, email: "author@example.com") } - let(:commenter) { create(:coplan_user) } + let(:commenter) { create(:coplan_user, email: "commenter@example.com") } let(:plan) { create(:plan, created_by_user: plan_author) } let(:thread_record) do create(:comment_thread, plan: plan, plan_version: plan.current_plan_version, created_by_user: commenter) end - let!(:first_comment) do - thread_record.comments.create!( - author_type: "human", - author_id: commenter.id, body_markdown: "A comment body." - ) + + around do |example| + original_cache = Rails.cache + Rails.cache = ActiveSupport::Cache::MemoryStore.new + example.run + Rails.cache = original_cache end before do @@ -22,7 +23,14 @@ end describe "#perform" do - it "sends a DM to the plan author" do + let!(:first_comment) do + thread_record.comments.create!( + author_type: "human", + author_id: commenter.id, body_markdown: "A comment body." + ) + end + + it "sends a DM to the plan author for the thread's first comment" do described_class.perform_now(comment_thread_id: thread_record.id) expect(SlackClient).to have_received(:send_dm).with( @@ -42,15 +50,6 @@ ) end - it "includes first comment body in the message" do - described_class.perform_now(comment_thread_id: thread_record.id) - - expect(SlackClient).to have_received(:send_dm).with( - email: plan_author.email, - text: a_string_including("A comment body.") - ) - end - it "skips notification when first comment author is the plan author" do first_comment.update_columns(author_id: plan_author.id) @@ -67,7 +66,7 @@ expect(SlackClient).to have_received(:send_dm) end - it "skips notification when thread has no comments" do + it "skips notification when the thread has no comments in the batch window" do first_comment.destroy! described_class.perform_now(comment_thread_id: thread_record.id) @@ -82,31 +81,150 @@ expect(SlackClient).not_to have_received(:send_dm) end + end + + describe "replies" do + let(:other_participant) { create(:coplan_user, email: "other@example.com") } + + before do + # An hour-old comment, clearly outside this batch — simulates it + # having already been notified about in an earlier run. + travel_to 1.hour.ago do + thread_record.comments.create!( + author_type: "human", author_id: commenter.id, body_markdown: "First comment." + ) + end + Rails.cache.write(described_class.batch_start_key(thread_record.id), 30.minutes.ago) + end + + it "notifies the plan author and prior participants, excluding the replier" do + thread_record.comments.create!( + author_type: "human", author_id: other_participant.id, body_markdown: "Someone else chimes in." + ) + + described_class.perform_now(comment_thread_id: thread_record.id) + + expect(SlackClient).to have_received(:send_dm).with(email: plan_author.email, text: anything) + expect(SlackClient).to have_received(:send_dm).with(email: commenter.email, text: anything) + expect(SlackClient).not_to have_received(:send_dm).with(email: other_participant.email, text: anything) + end + + it "does not notify the replier about their own reply" do + thread_record.comments.create!( + author_type: "human", author_id: commenter.id, body_markdown: "Following up on my own thread." + ) + + described_class.perform_now(comment_thread_id: thread_record.id) + + expect(SlackClient).not_to have_received(:send_dm).with(email: commenter.email, text: anything) + end + end + + describe ".debounce" do + it "schedules exactly one delayed job for a burst of comments on the same thread" do + expect { + 3.times { described_class.debounce(comment_thread_id: thread_record.id) } + }.to have_enqueued_job(described_class).exactly(1).times.on_queue("default") + end + + it "coalesces every comment created during the debounce window into one message per recipient" do + Rails.cache.write(described_class.batch_start_key(thread_record.id), 30.minutes.ago) + other_participant = create(:coplan_user) + thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "First.") + thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "Second.") + thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "Third.") + + described_class.perform_now(comment_thread_id: thread_record.id) + + expect(SlackClient).to have_received(:send_dm).with( + email: plan_author.email, + text: a_string_including("3 new comments").and(a_string_including("Third.")) + ).once + end + + it "allows a new burst after the previous one has been sent" do + thread_record.comments.create!(author_type: "human", author_id: commenter.id, body_markdown: "First.") + + described_class.debounce(comment_thread_id: thread_record.id) + described_class.perform_now(comment_thread_id: thread_record.id) - it "enqueues on the default queue" do - thread_record # eagerly create to avoid callback enqueue inside expect block expect { - described_class.perform_later(comment_thread_id: thread_record.id) + described_class.debounce(comment_thread_id: thread_record.id) }.to have_enqueued_job(described_class).on_queue("default") end - it "discards permanent Slack errors without retrying" do - allow(SlackClient).to receive(:send_dm).and_raise(SlackClient::PermanentError, "users_not_found") + it "includes the comment that triggered the batch, not just later ones" do + # Regression test: batch_start must be the triggering comment's own + # created_at, not Time.current at the moment debounce runs — by the + # time this method is called, the comment already exists (that's what + # enqueued the job that led here), so Time.current would be later than + # created_at and would wrongly exclude this very comment. + comment = thread_record.comments.create!( + author_type: "human", author_id: commenter.id, body_markdown: "Kicks off the batch." + ) + + described_class.debounce(comment_thread_id: thread_record.id, comment_created_at: comment.created_at) + described_class.perform_now(comment_thread_id: thread_record.id) + + expect(SlackClient).to have_received(:send_dm).with( + email: plan_author.email, + text: a_string_including("Kicks off the batch.") + ) + end + end + + describe "error handling" do + let!(:first_comment) do + thread_record.comments.create!( + author_type: "human", author_id: commenter.id, body_markdown: "First comment." + ) + end + + it "skips a recipient with a permanent Slack error but keeps notifying others" do + other_participant = create(:coplan_user, email: "other@example.com") + thread_record.comments.create!( + author_type: "human", author_id: other_participant.id, body_markdown: "Reply." + ) + allow(SlackClient).to receive(:send_dm).with(email: plan_author.email, text: anything) + .and_raise(SlackClient::PermanentError, "users_not_found") expect { described_class.perform_now(comment_thread_id: thread_record.id) }.not_to raise_error - expect(SlackClient).to have_received(:send_dm).once + expect(SlackClient).to have_received(:send_dm).with(email: commenter.email, text: anything) end - it "retries on transient Slack errors" do - thread_record # eagerly create + it "retries the whole batch on a transient Slack error" do allow(SlackClient).to receive(:send_dm).and_raise(SlackClient::Error, "ratelimited") expect { described_class.perform_now(comment_thread_id: thread_record.id) }.to have_enqueued_job(described_class) end + + it "does not double-notify a recipient who already succeeded when a transient error forces a retry" do + other_participant = create(:coplan_user, email: "other@example.com") + thread_record.comments.create!( + author_type: "human", author_id: other_participant.id, body_markdown: "Reply." + ) + + call_log = [] + raised = false + allow(SlackClient).to receive(:send_dm) do |email:, text:| + call_log << email + if email == other_participant.email && !raised + raised = true + raise SlackClient::Error, "ratelimited" + end + end + + described_class.perform_now(comment_thread_id: thread_record.id) # fails partway through + described_class.perform_now(comment_thread_id: thread_record.id) # simulated retry + + expect(call_log.tally[plan_author.email]).to eq(1) + expect(call_log.tally[commenter.email]).to eq(1) + expect(call_log.tally[other_participant.email]).to eq(2) # one failed attempt, one success + end end end diff --git a/spec/models/comment_spec.rb b/spec/models/comment_spec.rb index 8454c546..ca9646dd 100644 --- a/spec/models/comment_spec.rb +++ b/spec/models/comment_spec.rb @@ -52,27 +52,16 @@ }.to have_enqueued_job(CoPlan::NotificationJob) end - it "does not enqueue NotificationJob for subsequent comments" do + # Every comment fires the event now, not just the thread opener — a + # reply should notify anyone already in the thread. It's up to the + # notification_handler (Slack DM job, debounced) to decide who actually + # hears about it and to coalesce a burst into one message. + it "enqueues NotificationJob for subsequent comments too" do create(:comment, comment_thread: thread_record) expect { create(:comment, comment_thread: thread_record) - }.not_to have_enqueued_job(CoPlan::NotificationJob) - end - - # Regression for COPLAN-30: `first_comment_in_thread?` used to compare - # UUIDs with `id < ?`, which is not insertion-ordered. A reply whose UUID - # sorts before the opener's was wrongly treated as the thread opener, - # firing a duplicate "new comment thread" notification for the plan author. - it "does not enqueue NotificationJob for a reply whose UUID sorts before the opener" do - high_id = "ffffffff-ffff-ffff-ffff-ffffffffffff" - low_id = "00000000-0000-0000-0000-000000000001" - - create(:comment, comment_thread: thread_record, id: high_id) - - expect { - create(:comment, comment_thread: thread_record, id: low_id) - }.not_to have_enqueued_job(CoPlan::NotificationJob) + }.to have_enqueued_job(CoPlan::NotificationJob) end end diff --git a/spec/models/coplan/comment_analytics_spec.rb b/spec/models/coplan/comment_analytics_spec.rb index fda5efec..9b1ebac0 100644 --- a/spec/models/coplan/comment_analytics_spec.rb +++ b/spec/models/coplan/comment_analytics_spec.rb @@ -43,8 +43,8 @@ expect(payload[:properties][:is_first_in_thread]).to be(false) end - # `first_comment_in_thread?` (used by notify_plan_author) compares UUID - # strings with `id < ?`, which is not insertion-ordered. The analytics + # `first_comment_in_thread?` (used by notify_thread_participants) compares + # UUID strings with `id < ?`, which is not insertion-ordered. The analytics # path uses a total-count check instead, so a reply whose UUID happens # to sort before the existing first comment still records is_first=false. it "is not fooled by a reply whose UUID sorts before earlier comments" do