Use timeline subscribed? checks

app/lib/feed_manager.rb

@@ -355,7 +355,7 @@ class FeedManager
   # @param [String] timeline_key
   # @return [Boolean]
   def push_update_required?(timeline_key)
-    redis.exists?("subscribed:#{timeline_key}")
+    Timeline.subscribed?(timeline_key)
   end
 
   # Check if the account is blocking or muting any of the given accounts

app/models/account_conversation.rb

@@ -123,7 +123,7 @@ class AccountConversation < ApplicationRecord
   end
 
   def subscribed_to_timeline?
-    redis.exists?("subscribed:#{streaming_channel}")
+    Timeline.subscribed?(streaming_channel)
   end
 
   def streaming_channel

app/services/fan_out_on_write_service.rb

@@ -173,6 +173,6 @@ class FanOutOnWriteService < BaseService
   end
 
   def subscribed_to_streaming_api?(account_id)
-    redis.exists?("subscribed:timeline:#{account_id}") || redis.exists?("subscribed:timeline:#{account_id}:notifications")
+    Timeline.subscribed?("timeline:#{account_id}") || Timeline.subscribed?("timeline:#{account_id}:notifications")
   end
 end

app/services/notify_service.rb

@@ -260,7 +260,7 @@ class NotifyService < BaseService
   end
 
   def subscribed_to_streaming_api?
-    redis.exists?("subscribed:timeline:#{@recipient.id}") || redis.exists?("subscribed:timeline:#{@recipient.id}:notifications")
+    Timeline.subscribed?("timeline:#{@recipient.id}") || Timeline.subscribed?("timeline:#{@recipient.id}:notifications")
   end
 
   def push_to_conversation!

app/workers/publish_announcement_reaction_worker.rb

@@ -14,7 +14,7 @@ class PublishAnnouncementReactionWorker
     payload = Oj.dump(event: :'announcement.reaction', payload: payload)
 
     FeedManager.instance.with_active_accounts do |account|
-      redis.publish("timeline:#{account.id}", payload) if redis.exists?("subscribed:timeline:#{account.id}")
+      redis.publish("timeline:#{account.id}", payload) if Timeline.subscribed?("timeline:#{account.id}")
     end
   rescue ActiveRecord::RecordNotFound
     true

app/workers/publish_scheduled_announcement_worker.rb

@@ -15,7 +15,7 @@ class PublishScheduledAnnouncementWorker
     payload = Oj.dump(event: :announcement, payload: payload)
 
     FeedManager.instance.with_active_accounts do |account|
-      redis.publish("timeline:#{account.id}", payload) if redis.exists?("subscribed:timeline:#{account.id}")
+      redis.publish("timeline:#{account.id}", payload) if Timeline.subscribed?("timeline:#{account.id}")
     end
   end
 

app/workers/unfilter_notifications_worker.rb

@@ -58,6 +58,6 @@ class UnfilterNotificationsWorker
   end
 
   def subscribed_to_streaming_api?
-    redis.exists?("subscribed:timeline:#{@recipient.id}") || redis.exists?("subscribed:timeline:#{@recipient.id}:notifications")
+    Timeline.subscribed?("timeline:#{@recipient.id}") || Timeline.subscribed?("timeline:#{@recipient.id}:notifications")
   end
 end

app/workers/unpublish_announcement_worker.rb

@@ -8,7 +8,7 @@ class UnpublishAnnouncementWorker
     payload = Oj.dump(event: :'announcement.delete', payload: announcement_id.to_s)
 
     FeedManager.instance.with_active_accounts do |account|
-      redis.publish("timeline:#{account.id}", payload) if redis.exists?("subscribed:timeline:#{account.id}")
+      redis.publish("timeline:#{account.id}", payload) if Timeline.subscribed?("timeline:#{account.id}")
     end
   end
 end

spec/workers/unfilter_notifications_worker_spec.rb

@@ -14,14 +14,14 @@ RSpec.describe UnfilterNotificationsWorker do
     follow_request = sender.request_follow!(recipient)
     Fabricate(:notification, filtered: true, from_account: sender, account: recipient, type: :follow_request, activity: follow_request)
     allow(redis).to receive(:publish)
-    allow(redis).to receive(:exists?).and_return(false)
+    allow(Timeline).to receive(:subscribed?).and_return(false)
   end
 
   shared_examples 'shared behavior' do
     context 'when this is the last pending merge job and the user is subscribed to streaming' do
       before do
         redis.set("notification_unfilter_jobs:#{recipient.id}", 1)
-        allow(redis).to receive(:exists?).with("subscribed:timeline:#{recipient.id}").and_return(true)
+        allow(Timeline).to receive(:subscribed?).with("timeline:#{recipient.id}").and_return(true)
       end
 
       it 'unfilters notifications, adds private messages to conversations, and pushes to redis' do
@@ -37,7 +37,7 @@ RSpec.describe UnfilterNotificationsWorker do
     context 'when this is not last pending merge job and the user is subscribed to streaming' do
       before do
         redis.set("notification_unfilter_jobs:#{recipient.id}", 2)
-        allow(redis).to receive(:exists?).with("subscribed:timeline:#{recipient.id}").and_return(true)
+        allow(Timeline).to receive(:subscribed?).with("timeline:#{recipient.id}").and_return(true)
       end
 
       it 'unfilters notifications, adds private messages to conversations, and does not push to redis' do