Use to_json call for raw event strings (#38215)

app/lib/access_token_extension.rb

@@ -24,6 +24,6 @@ module AccessTokenExtension
   end
 
   def push_to_streaming_api
-    redis.publish("timeline:access_token:#{id}", Oj.dump(event: :kill)) if revoked? || destroyed?
+    redis.publish("timeline:access_token:#{id}", { event: :kill }.to_json) if revoked? || destroyed?
   end
 end

app/lib/application_extension.rb

@@ -35,7 +35,7 @@ module ApplicationExtension
 
   def close_streaming_sessions(resource_owner = nil)
     # TODO: #28793 Combine into a single topic
-    payload = Oj.dump(event: :kill)
+    payload = { event: :kill }.to_json
     scope = access_tokens
     scope = scope.where(resource_owner_id: resource_owner.id) unless resource_owner.nil?
     scope.in_batches do |tokens|

app/lib/feed_manager.rb

@@ -90,7 +90,7 @@ class FeedManager
   def unpush_from_home(account, status, update: false)
     return false unless remove_from_feed(:home, account.id, status, aggregate_reblogs: account.user&.aggregates_reblogs?)
 
-    redis.publish("timeline:#{account.id}", Oj.dump(event: :delete, payload: status.id.to_s)) unless update
+    redis.publish("timeline:#{account.id}", { event: :delete, payload: status.id.to_s }.to_json) unless update
     true
   end
 
@@ -117,7 +117,7 @@ class FeedManager
   def unpush_from_list(list, status, update: false)
     return false unless remove_from_feed(:list, list.id, status, aggregate_reblogs: list.account.user&.aggregates_reblogs?)
 
-    redis.publish("timeline:list:#{list.id}", Oj.dump(event: :delete, payload: status.id.to_s)) unless update
+    redis.publish("timeline:list:#{list.id}", { event: :delete, payload: status.id.to_s }.to_json) unless update
     true
   end
 

app/models/concerns/account/suspensions.rb

@@ -35,7 +35,7 @@ module Account::Suspensions
 
     # This terminates all connections for the given account with the streaming
     # server:
-    redis.publish("timeline:system:#{id}", Oj.dump(event: :kill)) if local?
+    redis.publish("timeline:system:#{id}", { event: :kill }.to_json) if local?
   end
 
   def unsuspend!

app/models/custom_filter.rb

@@ -115,8 +115,8 @@ class CustomFilter < ApplicationRecord
     @should_invalidate_cache = false
 
     Rails.cache.delete("filters:v3:#{account_id}")
-    redis.publish("timeline:#{account_id}", Oj.dump(event: :filters_changed))
-    redis.publish("timeline:system:#{account_id}", Oj.dump(event: :filters_changed))
+    redis.publish("timeline:#{account_id}", { event: :filters_changed }.to_json)
+    redis.publish("timeline:system:#{account_id}", { event: :filters_changed }.to_json)
   end
 
   private

app/models/user.rb

@@ -173,7 +173,7 @@ class User < ApplicationRecord
 
     # This terminates all connections for the given account with the streaming
     # server:
-    redis.publish("timeline:system:#{account.id}", Oj.dump(event: :kill))
+    redis.publish("timeline:system:#{account.id}", { event: :kill }.to_json)
   end
 
   def enable!
@@ -347,7 +347,7 @@ class User < ApplicationRecord
       # Revoke each access token for the Streaming API, since `update_all``
       # doesn't trigger ActiveRecord Callbacks:
       # TODO: #28793 Combine into a single topic
-      payload = Oj.dump(event: :kill)
+      payload = { event: :kill }.to_json
       redis.pipelined do |pipeline|
         batch.ids.each do |id|
           pipeline.publish("timeline:access_token:#{id}", payload)

app/services/batched_remove_status_service.rb

@@ -82,7 +82,7 @@ class BatchedRemoveStatusService < BaseService
   def unpush_from_public_timelines(status, pipeline)
     return unless status.public_visibility? && status.id > @status_id_cutoff
 
-    payload = Oj.dump(event: :delete, payload: status.id.to_s)
+    payload = { event: :delete, payload: status.id.to_s }.to_json
 
     pipeline.publish('timeline:public', payload)
     pipeline.publish(status.local? ? 'timeline:public:local' : 'timeline:public:remote', payload)

app/services/notify_service.rb

@@ -259,7 +259,7 @@ class NotifyService < BaseService
   end
 
   def push_to_streaming_api!
-    redis.publish("timeline:#{@recipient.id}:notifications", Oj.dump(event: :notification, payload: InlineRenderer.render(@notification, @recipient, :notification)))
+    redis.publish("timeline:#{@recipient.id}:notifications", { event: :notification, payload: InlineRenderer.render(@notification, @recipient, :notification) }.to_json)
   end
 
   def subscribed_to_streaming_api?

app/services/remove_status_service.rb

@@ -14,7 +14,7 @@ class RemoveStatusService < BaseService
   # @option  [Boolean] :original_removed
   # @option  [Boolean] :skip_streaming
   def call(status, **options)
-    @payload  = Oj.dump(event: :delete, payload: status.id.to_s)
+    @payload  = { event: :delete, payload: status.id.to_s }.to_json
     @status   = status
     @account  = status.account
     @options  = options

app/workers/publish_announcement_reaction_worker.rb

@@ -11,7 +11,7 @@ class PublishAnnouncementReactionWorker
     reaction ||= announcement.announcement_reactions.new(name: name)
 
     payload = InlineRenderer.render(reaction, nil, :reaction).tap { |h| h[:announcement_id] = announcement_id.to_s }
-    payload = Oj.dump(event: :'announcement.reaction', payload: payload)
+    payload = { 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}")

app/workers/publish_scheduled_announcement_worker.rb

@@ -12,7 +12,7 @@ class PublishScheduledAnnouncementWorker
     @announcement.publish! unless @announcement.published?
 
     payload = InlineRenderer.render(@announcement, nil, :announcement)
-    payload = Oj.dump(event: :announcement, payload: payload)
+    payload = { event: :announcement, payload: payload }.to_json
 
     FeedManager.instance.with_active_accounts do |account|
       redis.publish("timeline:#{account.id}", payload) if redis.exists?("subscribed:timeline:#{account.id}")

app/workers/push_conversation_worker.rb

@@ -9,7 +9,7 @@ class PushConversationWorker
     message      = InlineRenderer.render(conversation, conversation.account, :conversation)
     timeline_id  = "timeline:direct:#{conversation.account_id}"
 
-    redis.publish(timeline_id, Oj.dump(event: :conversation, payload: message))
+    redis.publish(timeline_id, { event: :conversation, payload: message }.to_json)
   rescue ActiveRecord::RecordNotFound
     true
   end

app/workers/unfilter_notifications_worker.rb

@@ -39,7 +39,7 @@ class UnfilterNotificationsWorker
   end
 
   def push_streaming_event!
-    redis.publish("timeline:#{@recipient.id}:notifications", Oj.dump(event: :notifications_merged, payload: '1'))
+    redis.publish("timeline:#{@recipient.id}:notifications", { event: :notifications_merged, payload: '1' }.to_json)
   end
 
   def subscribed_to_streaming_api?

app/workers/unpublish_announcement_worker.rb

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

spec/lib/feed_manager_spec.rb

@@ -546,7 +546,7 @@ RSpec.describe FeedManager do
       allow(redis).to receive_messages(publish: nil)
       subject.unpush_from_home(receiver, status)
 
-      deletion = Oj.dump(event: :delete, payload: status.id.to_s)
+      deletion = { event: :delete, payload: status.id.to_s }.to_json
       expect(redis).to have_received(:publish).with("timeline:#{receiver.id}", deletion)
     end
   end

spec/models/user_spec.rb

@@ -403,7 +403,7 @@ RSpec.describe User do
       expect(user).to have_attributes(disabled: true)
 
       expect(redis)
-        .to have_received(:publish).with("timeline:system:#{user.account.id}", Oj.dump(event: :kill)).once
+        .to have_received(:publish).with("timeline:system:#{user.account.id}", { event: :kill }.to_json).once
     end
   end
 
@@ -445,7 +445,7 @@ RSpec.describe User do
       expect { web_push_subscription.reload }
         .to raise_error(ActiveRecord::RecordNotFound)
       expect(redis_pipeline_stub)
-        .to have_received(:publish).with("timeline:access_token:#{access_token.id}", Oj.dump(event: :kill)).once
+        .to have_received(:publish).with("timeline:access_token:#{access_token.id}", { event: :kill }.to_json).once
     end
 
     def remove_activated_sessions

spec/requests/api/v2/filters_spec.rb

@@ -222,7 +222,7 @@ RSpec.describe 'Filters' do
 
         expect(keyword.reload.keyword).to eq 'updated'
 
-        expect(redis).to have_received(:publish).with("timeline:#{user.account.id}", Oj.dump(event: :filters_changed)).once
+        expect(redis).to have_received(:publish).with("timeline:#{user.account.id}", { event: :filters_changed }.to_json).once
       end
     end
 

spec/services/remove_status_service_spec.rb

@@ -40,7 +40,7 @@ RSpec.describe RemoveStatusService, :inline_jobs do
         .to_not include(status.id)
 
       expect(redis)
-        .to have_received(:publish).with('timeline:public:media', Oj.dump(event: :delete, payload: status.id.to_s))
+        .to have_received(:publish).with('timeline:public:media', { event: :delete, payload: status.id.to_s }.to_json)
 
       expect(delete_delivery(hank, status))
         .to have_been_made.once