Refresh audience segment for 'active' user (#19470)

* Refresh an individual user's segmentation

* Refresh segment if creating a published article

* Quick refactor

* Refresh segment if update publishing

* Quick refactor

* Refresh segment if update user settings

* Refresh segments after user update + role change

* Using latest_article_updated_at

* Sidekiq uses JSON for arguments, needs basic types

* Fix test issues with last_updated_at and sidekiq params

* Tests update continues

* Consolidate dependencies

* Method names acknowledging manual exclusive

* Fix test name copypasta

* Rubocop

* Try to clarify Creator#series

* Try to avoid naming predicate

* Remove erroneously included helper

app/models/audience_segment.rb

@@ -18,7 +18,7 @@ class AudienceSegment < ApplicationRecord
   has_many :segmented_users, dependent: :destroy
   has_many :users, through: :segmented_users
 
-  after_validation :run_query
+  after_validation :persist_recently_active_users, unless: :manual?
 
   QUERIES = {
     manual: ->(scope = User) { scope.where(id: nil) },
@@ -40,6 +40,10 @@ class AudienceSegment < ApplicationRecord
     query_for_segment(segment_type)&.call(scope) || []
   end
 
+  def self.all_users_in_segment(segment_type, scope: User)
+    query_for_segment(segment_type)&.call(scope) || []
+  end
+
   def self.recently_active_users_in_segment(segment_type, scope: User.recently_active)
     scoped_users_in_segment(segment_type, scope: scope)
   end
@@ -54,11 +58,20 @@ class AudienceSegment < ApplicationRecord
       .map { |(id, type)| [I18n.t("models.#{model_name.i18n_key}.type_ofs.#{type}"), id] }
   end
 
-  def run_query
-    return if manual?
-
+  def persist_recently_active_users
     self.users = self.class.recently_active_users_in_segment(type_of)
   end
 
+  def all_users_in_segment
+    return users if manual?
+
+    self.class.all_users_in_segment(type_of)
+  end
+
+  def includes?(user_or_id)
+    user_id = user_or_id.respond_to?(:id) ? user_or_id.id : user_or_id
+    all_users_in_segment.exists?(user_id)
+  end
+
   alias refresh! save!
 end

app/models/user.rb

@@ -217,6 +217,7 @@ class User < ApplicationRecord
   before_validation :set_username
   before_validation :strip_payment_pointer
   before_create :create_users_settings_and_notification_settings_records
+  after_update :refresh_auto_audience_segments
   before_destroy :remove_from_mailchimp_newsletters, prepend: true
   before_destroy :destroy_follows, prepend: true
 
@@ -351,6 +352,10 @@ class User < ApplicationRecord
     end
   end
 
+  def refresh_auto_audience_segments
+    SegmentedUserRefreshWorker.perform_async(id)
+  end
+
   ##############################################################################
   #
   # Heads Up: Start Authorization Refactor
@@ -642,6 +647,7 @@ class User < ApplicationRecord
   def update_user_roles_cache(...)
     authorizer.clear_cache
     Rails.cache.delete("user-#{id}/has_trusted_role")
+    refresh_auto_audience_segments
     trusted?
   end
 end

app/models/users/setting.rb

@@ -31,12 +31,18 @@ module Users
 
     validate :validate_feed_url, if: :feed_url_changed?
 
+    after_update :refresh_auto_audience_segments
+
     def resolved_font_name
       config_font.gsub("default", Settings::UserExperience.default_font)
     end
 
     private
 
+    def refresh_auto_audience_segments
+      user.refresh_auto_audience_segments
+    end
+
     def validate_feed_url
       return if feed_url.blank?
 

app/services/articles/creator.rb

@@ -6,26 +6,30 @@ module Articles
 
     def initialize(user, article_params)
       @user = user
-      @article_params = article_params
+      @article_params = normalize_params(article_params)
     end
 
     def call
       rate_limit!
 
-      article = save_article
-
-      if article.persisted?
-        # Subscribe author to notifications for all comments on their article.
-        NotificationSubscription.create(user: user, notifiable_id: article.id, notifiable_type: "Article",
-                                        config: "all_comments")
+      create_article.tap do
+        subscribe_author if article.persisted?
+        refresh_auto_audience_segments if article.published?
       end
-
-      article
     end
 
     private
 
-    attr_reader :user, :article_params
+    attr_reader :article, :user, :article_params
+
+    def normalize_params(original_params)
+      original_params.except(:tags).tap do |params|
+        # convert tags from array to a string
+        if (tags = original_params[:tags]).present?
+          params[:tag_list] = tags.join(", ")
+        end
+      end
+    end
 
     def rate_limit!
       rate_limit_to_use = if user.decorate.considered_new?
@@ -37,22 +41,31 @@ module Articles
       user.rate_limiter.check_limit!(rate_limit_to_use)
     end
 
-    def save_article
-      series = article_params[:series]
-      tags = article_params[:tags]
+    def refresh_auto_audience_segments
+      user.refresh_auto_audience_segments
+    end
 
-      # convert tags from array to a string
-      if tags.present?
-        article_params.delete(:tags)
-        article_params[:tag_list] = tags.join(", ")
+    def create_article
+      @article = Article.create(article_params) do |article|
+        article.user_id = user.id
+        article.show_comments = true
+        article.collection = series if series.present?
       end
+    end
 
-      article = Article.new(article_params)
-      article.user_id = user.id
-      article.show_comments = true
-      article.collection = Collection.find_series(series, user) if series.present?
-      article.save
-      article
+    def series
+      @series ||= if article_params[:series].blank?
+                    []
+                  else
+                    Collection.find_series(article_params[:series], user)
+                  end
+    end
+
+    # Subscribe author to notifications for all comments on their article.
+    def subscribe_author
+      NotificationSubscription.create(user: user,
+                                      notifiable: article,
+                                      config: "all_comments")
     end
   end
 end

app/services/articles/updater.rb

@@ -9,53 +9,80 @@ module Articles
     def initialize(user, article, article_params)
       @user = user
       @article = article
-      @article_params = article_params
+      @article_params = normalize_params(article_params)
     end
 
     def call
       user.rate_limiter.check_limit!(:article_update)
+      success = article.update(article_params)
 
-      # updated edited time only if already published and not edited by an admin
-      update_edited_at = article.user == user && article.published
-      # remove published_at values received from a user if an articles was published before (has past published_at)
-      # published_at will remain as it was in this case
-      article_params.delete :published_at if article.published_at && !article.scheduled?
-
-      attrs = Articles::Attributes.new(article_params, article.user)
-        .for_update(update_edited_at: update_edited_at)
-
-      success = article.update(attrs)
       if success
         user.rate_limiter.track_limit_by_action(:article_update)
 
-        if article.published && article.saved_change_to_published.blank?
-          # If the article has already been published and is only being updated, then we need to create
-          # mentions and send notifications to mentioned users inline via the Mentions::CreateAll service.
-          Mentions::CreateAll.call(article)
-        end
-
-        # Remove any associated notifications if Article is unpublished
-        if article.saved_changes["published"] == [true, false]
-          Notification.remove_all_by_action_without_delay(notifiable_ids: article.id, notifiable_type: "Article",
-                                                          action: "Published")
-          ContextNotification.delete_by(context_id: article.id, context_type: "Article",
-                                        action: "Published")
-
-          if article.comments.exists?
-            Notification.remove_all(notifiable_ids: article.comments.ids,
-                                    notifiable_type: "Comment")
-          end
-          if article.mentions.exists?
-            Notification.remove_all(notifiable_ids: article.mentions.ids,
-                                    notifiable_type: "Mention")
-          end
-        end
+        remove_all_notifications if became_unpublished?
+        send_to_mentioned_users_and_followers if remains_published?
+        refresh_auto_audience_segments if became_published?
       end
+
       Result.new(success: success, article: article.decorate)
     end
 
     private
 
     attr_reader :user, :article, :article_params
+
+    def normalize_params(original_params)
+      article_params = original_params.dup
+
+      # updated edited time only if already published and not edited by an admin
+      update_edited_at = article.user == user && article.published
+
+      # remove published_at values received from a user if an articles was published before (has past published_at)
+      # published_at will remain as it was in this case
+      article_params.delete :published_at if article.published_at && !article.scheduled?
+
+      # NOTE: It's surprising that this is article.user and not @user
+      Articles::Attributes.new(article_params, article.user)
+        .for_update(update_edited_at: update_edited_at)
+    end
+
+    def refresh_auto_audience_segments
+      user.refresh_auto_audience_segments
+    end
+
+    def became_published?
+      article.published? && !article.published_previously_was
+    end
+
+    def remains_published?
+      article.published && article.saved_change_to_published.blank?
+    end
+
+    def became_unpublished?
+      article.saved_changes["published"] == [true, false]
+    end
+
+    # If the article has already been published and is only being updated, then we need to create
+    # mentions and send notifications to mentioned users inline via the Mentions::CreateAll service.
+    def send_to_mentioned_users_and_followers
+      Mentions::CreateAll.call(article)
+    end
+
+    # Remove any associated notifications if Article is unpublished
+    def remove_all_notifications
+      Notification.remove_all_by_action_without_delay(notifiable_ids: article.id, notifiable_type: "Article",
+                                                      action: "Published")
+      ContextNotification.delete_by(context_id: article.id, context_type: "Article",
+                                    action: "Published")
+
+      if article.comments.exists?
+        Notification.remove_all(notifiable_ids: article.comments.ids,
+                                notifiable_type: "Comment")
+      end
+      return unless article.mentions.exists?
+
+      Notification.remove_all(notifiable_ids: article.mentions.ids,
+                              notifiable_type: "Mention")
+    end
   end
 end

app/workers/segmented_user_refresh_worker.rb

@@ -0,0 +1,30 @@
+class SegmentedUserRefreshWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: :low_priority
+
+  # Ignore manually-updated segments, remove user if they no longer belong,
+  # add the user if they now belong
+  def perform(user_or_id)
+    user_id = user_or_id.respond_to?(:id) ? user_or_id.id : user_or_id
+    automated_segments = AudienceSegment.not_manual
+    matching = SegmentedUser.where(user_id: user_id).pluck(:audience_segment_id)
+
+    current_segments = automated_segments.where(id: matching)
+    future_segments = automated_segments.each_with_object([]) do |audience_segment, collection|
+      collection << audience_segment if audience_segment.includes?(user_id)
+    end
+
+    # In current / not in future => delete user from segment
+    SegmentedUser
+      .where(user_id: user_id)
+      .delete_by(audience_segment: (current_segments - future_segments))
+
+    # In future / not in current => add user to segment
+    (future_segments - current_segments).each do |audience_segment|
+      audience_segment.segmented_users.create! user_id: user_id
+    end
+
+    future_segments
+  end
+end

spec/factories/articles.rb

@@ -62,6 +62,27 @@ FactoryBot.define do
     end
   end
 
+  factory :published_article, class: "Article" do
+    transient do
+      past_published_at { 3.days.ago }
+    end
+
+    title { generate(:title) }
+    published { true }
+    tag_list { "javascript, html, discuss" }
+    co_author_ids { [] }
+    body_markdown { Faker::Hipster.paragraph(sentence_count: 2) }
+  end
+
+  factory :unpublished_article, class: "Article" do
+    title { generate(:title) }
+    published { false }
+    published_at { nil }
+    tag_list { "javascript, html, discuss" }
+    co_author_ids { [] }
+    body_markdown { Faker::Hipster.paragraph(sentence_count: 2) }
+  end
+
   trait :video do
     after(:create) do |article|
       article.update_columns(

spec/models/user_spec.rb

@@ -33,6 +33,7 @@ RSpec.describe User do
 
   before do
     omniauth_mock_providers_payload
+    allow(SegmentedUserRefreshWorker).to receive(:perform_async)
     allow(Settings::Authentication).to receive(:providers).and_return(Authentication::Providers.available)
   end
 
@@ -366,6 +367,13 @@ RSpec.describe User do
         expect(new_user.reload.notifications.count).to eq(0)
       end
     end
+
+    it "refreshes user segments" do
+      expect(user).to be_persisted
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+      user.update name: "New Name Here"
+      expect(SegmentedUserRefreshWorker).to have_received(:perform_async).with(user.id)
+    end
   end
 
   context "when callbacks are triggered after commit" do
@@ -414,6 +422,15 @@ RSpec.describe User do
     end
   end
 
+  context "when a new role is added" do
+    let(:without_role) { create(:user, roles: []) }
+
+    it "refreshes user segments" do
+      without_role.add_role :trusted
+      expect(SegmentedUserRefreshWorker).to have_received(:perform_async).with(without_role.id)
+    end
+  end
+
   describe "user registration", vcr: { cassette_name: "fastly_sloan" } do
     let(:user) { create(:user) }
 

spec/models/users/setting_spec.rb

@@ -4,6 +4,10 @@ RSpec.describe Users::Setting do
   let(:user) { create(:user) }
   let(:setting) { described_class.find_by(user_id: user.id) }
 
+  before do
+    allow(SegmentedUserRefreshWorker).to receive(:perform_async)
+  end
+
   describe "validations" do
     subject { setting }
 
@@ -133,5 +137,20 @@ RSpec.describe Users::Setting do
       expect(user_comic_sans.setting.resolved_font_name).to eq("comic_sans")
     end
   end
+
+  context "when updating a setting" do
+    it "refreshes user segment" do
+      setting.experience_level = setting.experience_level.to_i + 1
+      setting.save!
+      expect(SegmentedUserRefreshWorker).to have_received(:perform_async).with(user.id)
+    end
+  end
+
+  context "when creating from scratch" do
+    it "does not refresh user segment" do
+      create(:user).setting
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+    end
+  end
 end
 # rubocop:enable Layout/LineLength

spec/services/articles/creator_spec.rb

@@ -3,7 +3,10 @@ require "rails_helper"
 RSpec.describe Articles::Creator, type: :service do
   let(:user) { create(:user) }
 
-  before { allow(FeatureFlag).to receive(:enabled?).with(:consistent_rendering, any_args).and_return(true) }
+  before do
+    allow(FeatureFlag).to receive(:enabled?).with(:consistent_rendering, any_args).and_return(true)
+    allow(SegmentedUserRefreshWorker).to receive(:perform_async)
+  end
 
   context "when valid attributes" do
     let(:valid_attributes) { attributes_for(:article) }
@@ -56,4 +59,31 @@ RSpec.describe Articles::Creator, type: :service do
       end.not_to change(NotificationSubscription, :count)
     end
   end
+
+  context "when creating a published article" do
+    let(:article_params) { attributes_for(:article, published: true) }
+
+    it "refreshes user segments" do
+      described_class.call(user, article_params)
+      expect(SegmentedUserRefreshWorker).to have_received(:perform_async).with(user.id)
+    end
+  end
+
+  context "when creating a not-yet-published article" do
+    let(:article_params) { attributes_for(:article, published: false, published_at: 5.days.from_now) }
+
+    it "does not refresh user segments" do
+      described_class.call(user, article_params)
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+    end
+  end
+
+  context "when creating a non-published article" do
+    let(:article_params) { attributes_for(:article, published: false) }
+
+    it "does not refresh user segments" do
+      described_class.call(user, article_params)
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+    end
+  end
 end

spec/services/articles/updater_spec.rb

@@ -6,7 +6,10 @@ RSpec.describe Articles::Updater, type: :service do
   let(:attributes) { { body_markdown: "sample" } }
   let(:draft) { create(:article, user: user, published: false, published_at: nil) }
 
-  before { allow(FeatureFlag).to receive(:enabled?).with(:consistent_rendering, any_args).and_return(true) }
+  before do
+    allow(FeatureFlag).to receive(:enabled?).with(:consistent_rendering, any_args).and_return(true)
+    allow(SegmentedUserRefreshWorker).to receive(:perform_async)
+  end
 
   it "updates an article" do
     described_class.call(user, article, attributes)
@@ -290,4 +293,37 @@ description:\ntags: heytag\n---\n\nHey this is the article"
       end
     end
   end
+
+  context "when publishing a previously unpublished article" do
+    let(:attributes) { { published: true } }
+    let(:markdown) { "body text here, no frontmatter" }
+    let(:unpublished) { create(:unpublished_article, user: user) }
+
+    it "refreshes user segments" do
+      described_class.call(user, unpublished, attributes)
+      expect(SegmentedUserRefreshWorker).to have_received(:perform_async).with(user.id)
+    end
+  end
+
+  context "when unpublished a previously published article" do
+    let(:attributes) { { published: false } }
+    let(:markdown) { "body text here, no frontmatter" }
+    let(:published) { create(:published_article, :past, user: user) }
+
+    it "does not refresh user segments" do
+      described_class.call(user, published, attributes)
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+    end
+  end
+
+  context "when setting published_at for a previously unpublished article" do
+    let(:attributes) { { published_at: 5.days.from_now } }
+    let(:markdown) { "body text here, no frontmatter" }
+    let(:unpublished) { create(:unpublished_article, user: user) }
+
+    it "does not refresh user segments" do
+      described_class.call(user, unpublished, attributes)
+      expect(SegmentedUserRefreshWorker).not_to have_received(:perform_async)
+    end
+  end
 end

spec/workers/segmented_user_refresh_worker_spec.rb

@@ -0,0 +1,39 @@
+require "rails_helper"
+
+RSpec.describe SegmentedUserRefreshWorker, type: :worker do
+  let(:worker) { subject }
+
+  # Sort of contriving a scenario where...
+  #   User had not posted, then posted
+  # So that:
+  #   User was in group A and B, now belongs to B and C
+  let!(:group_a) { create(:audience_segment, type_of: "no_posts_yet") }
+  let!(:group_b) { create(:audience_segment, type_of: "trusted") }
+  let!(:group_c) { create(:audience_segment, type_of: "posted") }
+  let(:user) { create(:user, :trusted) }
+  let(:user_id) { user.id }
+
+  include_examples "#enqueues_on_correct_queue", "low_priority"
+
+  it "can confirm the scenario baseline by refreshing" do
+    worker.perform(user_id)
+    expect(group_a.segmented_users.pluck(:user_id)).to include(user_id)
+    expect(group_b.segmented_users.pluck(:user_id)).to include(user_id)
+    expect(group_c.segmented_users.pluck(:user_id)).not_to include(user_id)
+  end
+
+  context "when scenario is pre-built" do
+    before do
+      group_a.segmented_users.create! user: user
+      group_b.segmented_users.create! user: user
+      user.update articles_count: 1
+    end
+
+    it "refreshes a user's segmentation by id" do
+      worker.perform(user_id)
+      expect(group_a.segmented_users.pluck(:user_id)).not_to include(user_id)
+      expect(group_b.segmented_users.pluck(:user_id)).to include(user_id)
+      expect(group_c.segmented_users.pluck(:user_id)).to include(user_id)
+    end
+  end
+end