Change Web Push API deliveries to use request pooling (#16014)

app/models/web/push_subscription.rb

@@ -24,81 +24,80 @@ class Web::PushSubscription < ApplicationRecord
   validates :key_p256dh, presence: true
   validates :key_auth, presence: true
 
-  def push(notification)
-    I18n.with_locale(associated_user&.locale || I18n.default_locale) do
-      push_payload(payload_for_notification(notification), 48.hours.seconds)
-    end
+  delegate :locale, to: :associated_user
+
+  def encrypt(payload)
+    Webpush::Encryption.encrypt(payload, key_p256dh, key_auth)
+  end
+
+  def audience
+    @audience ||= Addressable::URI.parse(endpoint).normalized_site
+  end
+
+  def crypto_key_header
+    p256ecdsa = vapid_key.public_key_for_push_header
+
+    "p256ecdsa=#{p256ecdsa}"
+  end
+
+  def authorization_header
+    jwt = JWT.encode({ aud: audience, exp: 24.hours.from_now.to_i, sub: "mailto:#{contact_email}" }, vapid_key.curve, 'ES256', typ: 'JWT')
+
+    "WebPush #{jwt}"
   end
 
   def pushable?(notification)
-    data&.key?('alerts') && ActiveModel::Type::Boolean.new.cast(data['alerts'][notification.type.to_s])
+    ActiveModel::Type::Boolean.new.cast(data&.dig('alerts', notification.type.to_s))
   end
 
   def associated_user
     return @associated_user if defined?(@associated_user)
 
-    @associated_user = if user_id.nil?
-                         session_activation.user
-                       else
-                         user
-                       end
+    @associated_user = begin
+      if user_id.nil?
+        session_activation.user
+      else
+        user
+      end
+    end
   end
 
   def associated_access_token
     return @associated_access_token if defined?(@associated_access_token)
 
-    @associated_access_token = if access_token_id.nil?
-                                 find_or_create_access_token.token
-                               else
-                                 access_token.token
-                               end
+    @associated_access_token = begin
+      if access_token_id.nil?
+        find_or_create_access_token.token
+      else
+        access_token.token
+      end
+    end
   end
 
   class << self
     def unsubscribe_for(application_id, resource_owner)
-      access_token_ids = Doorkeeper::AccessToken.where(application_id: application_id, resource_owner_id: resource_owner.id, revoked_at: nil)
-                                                .pluck(:id)
-
+      access_token_ids = Doorkeeper::AccessToken.where(application_id: application_id, resource_owner_id: resource_owner.id, revoked_at: nil).pluck(:id)
       where(access_token_id: access_token_ids).delete_all
     end
   end
 
   private
 
-  def push_payload(message, ttl = 5.minutes.seconds)
-    Webpush.payload_send(
-      message: Oj.dump(message),
-      endpoint: endpoint,
-      p256dh: key_p256dh,
-      auth: key_auth,
-      ttl: ttl,
-      ssl_timeout: 10,
-      open_timeout: 10,
-      read_timeout: 10,
-      vapid: {
-        subject: "mailto:#{::Setting.site_contact_email}",
-        private_key: Rails.configuration.x.vapid_private_key,
-        public_key: Rails.configuration.x.vapid_public_key,
-      }
-    )
-  end
-
-  def payload_for_notification(notification)
-    ActiveModelSerializers::SerializableResource.new(
-      notification,
-      serializer: Web::NotificationSerializer,
-      scope: self,
-      scope_name: :current_push_subscription
-    ).as_json
-  end
-
   def find_or_create_access_token
     Doorkeeper::AccessToken.find_or_create_for(
       application: Doorkeeper::Application.find_by(superapp: true),
-      resource_owner: session_activation.user_id,
+      resource_owner: user_id || session_activation.user_id,
       scopes: Doorkeeper::OAuth::Scopes.from_string('read write follow push'),
       expires_in: Doorkeeper.configuration.access_token_expires_in,
       use_refresh_token: Doorkeeper.configuration.refresh_token_enabled?
     )
   end
+
+  def vapid_key
+    @vapid_key ||= Webpush::VapidKey.from_keys(Rails.configuration.x.vapid_public_key, Rails.configuration.x.vapid_private_key)
+  end
+
+  def contact_email
+    @contact_email ||= ::Setting.site_contact_email
+  end
 end

app/workers/web/push_notification_worker.rb

@@ -3,22 +3,67 @@
 class Web::PushNotificationWorker
   include Sidekiq::Worker
 
-  sidekiq_options backtrace: true, retry: 5
+  sidekiq_options queue: 'push', retry: 5
+
+  TTL     = 48.hours.to_s
+  URGENCY = 'normal'
 
   def perform(subscription_id, notification_id)
-    subscription = ::Web::PushSubscription.find(subscription_id)
-    notification = Notification.find(notification_id)
+    @subscription = Web::PushSubscription.find(subscription_id)
+    @notification = Notification.find(notification_id)
 
-    subscription.push(notification) unless notification.activity.nil?
-  rescue Webpush::ResponseError => e
-    code = e.response.code.to_i
+    # Polymorphically associated activity could have been deleted
+    # in the meantime, so we have to double-check before proceeding
+    return unless @notification.activity.present? && @subscription.pushable?(@notification)
 
-    if (400..499).cover?(code) && ![408, 429].include?(code)
-      subscription.destroy!
-    else
-      raise e
+    payload = @subscription.encrypt(push_notification_json)
+
+    request_pool.with(@subscription.audience) do |http_client|
+      request = Request.new(:post, @subscription.endpoint, body: payload.fetch(:ciphertext), http_client: http_client)
+
+      request.add_headers(
+        'Content-Type'     => 'application/octet-stream',
+        'Ttl'              => TTL,
+        'Urgency'          => URGENCY,
+        'Content-Encoding' => 'aesgcm',
+        'Encryption'       => "salt=#{Webpush.encode64(payload.fetch(:salt)).delete('=')}",
+        'Crypto-Key'       => "dh=#{Webpush.encode64(payload.fetch(:server_public_key)).delete('=')};#{@subscription.crypto_key_header}",
+        'Authorization'    => @subscription.authorization_header
+      )
+
+      request.perform do |response|
+        # If the server responds with an error in the 4xx range
+        # that isn't about rate-limiting or timeouts, we can
+        # assume that the subscription is invalid or expired
+        # and must be removed
+
+        if (400..499).cover?(response.code) && ![408, 429].include?(response.code)
+          @subscription.destroy!
+        elsif !(200...300).cover?(response.code)
+          raise Mastodon::UnexpectedResponseError, response
+        end
+      end
     end
   rescue ActiveRecord::RecordNotFound
     true
   end
+
+  private
+
+  def push_notification_json
+    json = I18n.with_locale(@subscription.locale || I18n.default_locale) do
+      ActiveModelSerializers::SerializableResource.new(
+        @notification,
+        serializer: Web::NotificationSerializer,
+        scope: @subscription,
+        scope_name: :current_push_subscription
+      ).as_json
+    end
+
+    Oj.dump(json)
+  end
+
+  def request_pool
+    RequestPool.current
+  end
 end