Merge pull request from GHSA-7w3c-p9j8-mq3x

* Ensure destruction of OAuth Applications notifies streaming

Due to doorkeeper using a dependent: delete_all relationship, the destroy of an OAuth Application bypassed the existing AccessTokenExtension callbacks for announcing destructing of access tokens.

* Ensure password resets revoke access to Streaming API

* Improve performance of deleting OAuth tokens

---------

Co-authored-by: Emelia Smith <ThisIsMissEm@users.noreply.github.com>

app/lib/application_extension.rb

@@ -4,12 +4,32 @@ module ApplicationExtension
   extend ActiveSupport::Concern
 
   included do
+    include Redisable
+
     validates :name, length: { maximum: 60 }
     validates :website, url: true, length: { maximum: 2_000 }, if: :website?
     validates :redirect_uri, length: { maximum: 2_000 }
+
+    # The relationship used between Applications and AccessTokens is using
+    # dependent: delete_all, which means the ActiveRecord callback in
+    # AccessTokenExtension is not run, so instead we manually announce to
+    # streaming that these tokens are being deleted.
+    before_destroy :push_to_streaming_api, prepend: true
   end
 
   def confirmation_redirect_uri
     redirect_uri.lines.first.strip
   end
+
+  def push_to_streaming_api
+    # TODO: #28793 Combine into a single topic
+    payload = 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

@@ -399,6 +399,16 @@ class User < ApplicationRecord
     Doorkeeper::AccessToken.by_resource_owner(self).in_batches do |batch|
       batch.update_all(revoked_at: Time.now.utc)
       Web::PushSubscription.where(access_token_id: batch).delete_all
+
+      # 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)
+      redis.pipelined do |pipeline|
+        batch.ids.each do |id|
+          pipeline.publish("timeline:access_token:#{id}", payload)
+        end
+      end
     end
   end