Improve performance of deleting OAuth tokens

app/lib/application_extension.rb

@@ -25,8 +25,13 @@ module ApplicationExtension
 
   def push_to_streaming_api
     # TODO: #28793 Combine into a single topic
-    access_tokens.in_batches.each do |token|
+    payload = Oj.dump(event: :kill)
-      redis.publish("timeline:access_token:#{token.id}", Oj.dump(event: :kill))
+    access_tokens.in_batches do |tokens|
+      redis.pipelined do |pipeline|
+        tokens.ids.each do |id|
+          pipeline.publish("timeline:access_token:#{id}", payload)
+        end
+      end
     end
   end
 end

app/models/user.rb

@@ -363,8 +363,11 @@ 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
-      batch.each do |token|
+      payload = Oj.dump(event: :kill)
-        redis.publish("timeline:access_token:#{token.id}", Oj.dump(event: :kill))
+      redis.pipelined do |pipeline|
+        batch.ids.each do |id|
+          pipeline.publish("timeline:access_token:#{id}", payload)
+        end
       end
     end
   end

spec/models/user_spec.rb

@@ -438,8 +438,10 @@ RSpec.describe User do
     let!(:access_token) { Fabricate(:access_token, resource_owner_id: user.id) }
     let!(:web_push_subscription) { Fabricate(:web_push_subscription, access_token: access_token) }
 
+    let(:redis_pipeline_stub) { instance_double(Redis::Namespace, publish: nil) }
+
     before do
-      allow(redis).to receive_messages(publish: nil)
+      allow(redis).to receive(:pipelined).and_yield(redis_pipeline_stub)
       user.reset_password!
     end
 
@@ -456,7 +458,7 @@ RSpec.describe User do
     end
 
     it 'revokes streaming access for all access tokens' do
-      expect(redis).to have_received(:publish).with("timeline:access_token:#{access_token.id}", Oj.dump(event: :kill)).once
+      expect(redis_pipeline_stub).to have_received(:publish).with("timeline:access_token:#{access_token.id}", Oj.dump(event: :kill)).once
     end
 
     it 'removes push subscriptions' do