1 files changed, 84 insertions, 65 deletions
diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb
index b72bb82d3..f62f78a79 100644
--- a/app/services/fan_out_on_write_service.rb
+++ b/app/services/fan_out_on_write_service.rb
@@ -3,107 +3,126 @@
 class FanOutOnWriteService < BaseService
   # Push a status into home and mentions feeds
   # @param [Status] status
-  def call(status)
-    raise Mastodon::RaceConditionError if status.visibility.nil?
-
-    deliver_to_self(status) if status.account.local?
-
-    if status.direct_visibility?
-      deliver_to_mentioned_followers(status)
-      deliver_to_own_conversation(status)
-    elsif status.limited_visibility?
-      deliver_to_mentioned_followers(status)
-    else
-      deliver_to_followers(status)
-      deliver_to_lists(status)
-    end
+  # @param [Hash] options
+  # @option options [Boolean] update
+  # @option options [Array<Integer>] silenced_account_ids
+  def call(status, options = {})
+    @status    = status
+    @account   = status.account
+    @options   = options
+
+    check_race_condition!
+
+    fan_out_to_local_recipients!
+    fan_out_to_public_streams! if broadcastable?
+  end
 
-    return if status.account.silenced? || !status.public_visibility? || status.reblog?
+  private
 
-    render_anonymous_payload(status)
+  def check_race_condition!
+    # I don't know why but at some point we had an issue where
+    # this service was being executed with status objects
+    # that had a null visibility - which should not be possible
+    # since the column in the database is not nullable.
+    #
+    # This check re-queues the service to be run at a later time
+    # with the full object, if something like it occurs
 
-    deliver_to_hashtags(status)
+    raise Mastodon::RaceConditionError if @status.visibility.nil?
+  end
 
-    return if status.reply? && status.in_reply_to_account_id != status.account_id
+  def fan_out_to_local_recipients!
+    deliver_to_self!
+    notify_mentioned_accounts!
 
-    deliver_to_public(status)
-    deliver_to_media(status) if status.media_attachments.any?
+    case @status.visibility.to_sym
+    when :public, :unlisted, :private
+      deliver_to_all_followers!
+      deliver_to_lists!
+    when :limited
+      deliver_to_mentioned_followers!
+    else
+      deliver_to_mentioned_followers!
+      deliver_to_conversation!
+    end
   end
 
-  private
+  def fan_out_to_public_streams!
+    broadcast_to_hashtag_streams!
+    broadcast_to_public_streams!
+  end
 
-  def deliver_to_self(status)
-    Rails.logger.debug "Delivering status #{status.id} to author"
-    FeedManager.instance.push_to_home(status.account, status)
+  def deliver_to_self!
+    FeedManager.instance.push_to_home(@account, @status, update: update?) if @account.local?
   end
 
-  def deliver_to_followers(status)
-    Rails.logger.debug "Delivering status #{status.id} to followers"
+  def notify_mentioned_accounts!
+    @status.active_mentions.where.not(id: @options[:silenced_account_ids] || []).joins(:account).merge(Account.local).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
+      LocalNotificationWorker.push_bulk(mentions) do |mention|
+        [mention.account_id, mention.id, 'Mention', :mention]
+      end
+    end
+  end
 
-    status.account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
+  def deliver_to_all_followers!
+    @account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
       FeedInsertWorker.push_bulk(followers) do |follower|
-        [status.id, follower.id, :home]
+        [@status.id, follower.id, :home, update: update?]
       end
     end
   end
 
-  def deliver_to_lists(status)
-    Rails.logger.debug "Delivering status #{status.id} to lists"
-
-    status.account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
+  def deliver_to_lists!
+    @account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
       FeedInsertWorker.push_bulk(lists) do |list|
-        [status.id, list.id, :list]
+        [@status.id, list.id, :list, update: update?]
       end
     end
   end
 
-  def deliver_to_mentioned_followers(status)
-    Rails.logger.debug "Delivering status #{status.id} to limited followers"
-
-    status.mentions.joins(:account).merge(status.account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
+  def deliver_to_mentioned_followers!
+    @status.mentions.joins(:account).merge(@account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
       FeedInsertWorker.push_bulk(mentions) do |mention|
-        [status.id, mention.account_id, :home]
+        [@status.id, mention.account_id, :home, update: update?]
       end
     end
   end
 
-  def render_anonymous_payload(status)
-    @payload = InlineRenderer.render(status, nil, :status)
-    @payload = Oj.dump(event: :update, payload: @payload)
+  def broadcast_to_hashtag_streams!
+    @status.tags.pluck(:name).each do |hashtag|
+      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", anonymous_payload)
+      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", anonymous_payload) if @status.local?
+    end
   end
 
-  def deliver_to_hashtags(status)
-    Rails.logger.debug "Delivering status #{status.id} to hashtags"
+  def broadcast_to_public_streams!
+    return if @status.reply? && @status.in_reply_to_account_id != @account.id
+
+    Redis.current.publish('timeline:public', anonymous_payload)
+    Redis.current.publish(@status.local? ? 'timeline:public:local' : 'timeline:public:remote', anonymous_payload)
 
-    status.tags.pluck(:name).each do |hashtag|
-      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
-      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if status.local?
+    if @status.media_attachments.any?
+      Redis.current.publish('timeline:public:media', anonymous_payload)
+      Redis.current.publish(@status.local? ? 'timeline:public:local:media' : 'timeline:public:remote:media', anonymous_payload)
     end
   end
 
-  def deliver_to_public(status)
-    Rails.logger.debug "Delivering status #{status.id} to public timeline"
-
-    Redis.current.publish('timeline:public', @payload)
-    if status.local?
-      Redis.current.publish('timeline:public:local', @payload)
-    else
-      Redis.current.publish('timeline:public:remote', @payload)
-    end
+  def deliver_to_conversation!
+    AccountConversation.add_status(@account, @status) unless update?
   end
 
-  def deliver_to_media(status)
-    Rails.logger.debug "Delivering status #{status.id} to media timeline"
+  def anonymous_payload
+    @anonymous_payload ||= Oj.dump(
+      event: update? ? :'status.update' : :update,
+      payload: InlineRenderer.render(@status, nil, :status)
+    )
+  end
 
-    Redis.current.publish('timeline:public:media', @payload)
-    if status.local?
-      Redis.current.publish('timeline:public:local:media', @payload)
-    else
-      Redis.current.publish('timeline:public:remote:media', @payload)
-    end
+  def update?
+    @is_update
   end
 
-  def deliver_to_own_conversation(status)
-    AccountConversation.add_status(status.account, status)
+  def broadcastable?
+    @status.public_visibility? && !@status.reblog? && !@account.silenced?
   end
 end