Add ability to follow hashtags (#18809)

app/services/fan_out_on_write_service.rb

@@ -16,6 +16,7 @@ class FanOutOnWriteService < BaseService
     check_race_condition!
 
     fan_out_to_local_recipients!
+    fan_out_to_public_recipients! if broadcastable?
     fan_out_to_public_streams! if broadcastable?
   end
 
@@ -50,6 +51,10 @@ class FanOutOnWriteService < BaseService
     end
   end
 
+  def fan_out_to_public_recipients!
+    deliver_to_hashtag_followers!
+  end
+
   def fan_out_to_public_streams!
     broadcast_to_hashtag_streams!
     broadcast_to_public_streams!
@@ -83,6 +88,14 @@ class FanOutOnWriteService < BaseService
     end
   end
 
+  def deliver_to_hashtag_followers!
+    TagFollow.where(tag_id: @status.tags.map(&:id)).select(:id, :account_id).reorder(nil).find_in_batches do |follows|
+      FeedInsertWorker.push_bulk(follows) do |follow|
+        [@status.id, follow.account_id, 'tags', { 'update' => update? }]
+      end
+    end
+  end
+
   def deliver_to_lists!
     @account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
       FeedInsertWorker.push_bulk(lists) do |list|
@@ -100,7 +113,7 @@ class FanOutOnWriteService < BaseService
   end
 
   def broadcast_to_hashtag_streams!
-    @status.tags.pluck(:name).each do |hashtag|
+    @status.tags.map(&:name).each do |hashtag|
       redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", anonymous_payload)
       redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", anonymous_payload) if @status.local?
     end