4 files changed, 42 insertions, 18 deletions

diff --git a/app/workers/pubsubhubbub/confirmation_worker.rb b/app/workers/pubsubhubbub/confirmation_worker.rb
index 9186c5d7d..e1ccfb99c 100644
--- a/app/workers/pubsubhubbub/confirmation_worker.rb
+++ b/app/workers/pubsubhubbub/confirmation_worker.rb
@@ -60,9 +60,7 @@ class Pubsubhubbub::ConfirmationWorker
   end
 
   def callback_get_with_params
-    HTTP.headers(user_agent: 'Mastodon/PubSubHubbub')
-        .timeout(:per_operation, write: 20, connect: 20, read: 50)
-        .get(subscription.callback_url, params: callback_params)
+    Request.new(:get, subscription.callback_url, params: callback_params).perform
   end
 
   def callback_response_body
@@ -71,10 +69,10 @@ class Pubsubhubbub::ConfirmationWorker
 
   def callback_params
     {
-      'hub.topic' => account_url(subscription.account, format: :atom),
-      'hub.mode' => mode,
-      'hub.challenge' => challenge,
-      'hub.lease_seconds' => subscription.lease_seconds,
+      'hub.topic': account_url(subscription.account, format: :atom),
+      'hub.mode': mode,
+      'hub.challenge': challenge,
+      'hub.lease_seconds': subscription.lease_seconds,
     }
   end
 
diff --git a/app/workers/pubsubhubbub/delivery_worker.rb b/app/workers/pubsubhubbub/delivery_worker.rb
index 981838f33..2e1101b93 100644
--- a/app/workers/pubsubhubbub/delivery_worker.rb
+++ b/app/workers/pubsubhubbub/delivery_worker.rb
@@ -33,9 +33,9 @@ class Pubsubhubbub::DeliveryWorker
   end
 
   def callback_post_payload
-    HTTP.timeout(:per_operation, write: 50, connect: 20, read: 50)
-        .headers(headers)
-        .post(subscription.callback_url, body: payload)
+    request = Request.new(:post, subscription.callback_url, body: payload)
+    request.add_headers(headers)
+    request.perform
   end
 
   def blocked_domain?
@@ -43,18 +43,17 @@ class Pubsubhubbub::DeliveryWorker
   end
 
   def host
-    Addressable::URI.parse(subscription.callback_url).normalize.host
+    Addressable::URI.parse(subscription.callback_url).normalized_host
   end
 
   def headers
     {
-      'User-Agent' => 'Mastodon/PubSubHubbub',
       'Content-Type' => 'application/atom+xml',
-      'Link' => link_headers,
+      'Link' => link_header,
     }.merge(signature_headers.to_h)
   end
 
-  def link_headers
+  def link_header
     LinkHeader.new([hub_link_header, self_link_header]).to_s
   end
 
diff --git a/app/workers/pubsubhubbub/distribution_worker.rb b/app/workers/pubsubhubbub/distribution_worker.rb
index b41cec90d..7592354cc 100644
--- a/app/workers/pubsubhubbub/distribution_worker.rb
+++ b/app/workers/pubsubhubbub/distribution_worker.rb
@@ -35,16 +35,16 @@ class Pubsubhubbub::DistributionWorker
     @payload = AtomSerializer.render(AtomSerializer.new.feed(@account, stream_entries))
     @domains = @account.followers.domains
 
-    Pubsubhubbub::DeliveryWorker.push_bulk(@subscriptions.reject { |s| !allowed_to_receive?(s.callback_url) }) do |subscription|
+    Pubsubhubbub::DeliveryWorker.push_bulk(@subscriptions.reject { |s| !allowed_to_receive?(s.callback_url, s.domain) }) do |subscription|
       [subscription.id, @payload]
     end
   end
 
   def active_subscriptions
-    Subscription.where(account: @account).active.select('id, callback_url')
+    Subscription.where(account: @account).active.select('id, callback_url, domain')
   end
 
-  def allowed_to_receive?(callback_url)
-    @domains.include?(Addressable::URI.parse(callback_url).host)
+  def allowed_to_receive?(callback_url, domain)
+    (!domain.nil? && @domains.include?(domain)) || @domains.include?(Addressable::URI.parse(callback_url).host)
   end
 end
diff --git a/app/workers/web_push_notification_worker.rb b/app/workers/web_push_notification_worker.rb
new file mode 100644
index 000000000..0568a3e02
--- /dev/null
+++ b/app/workers/web_push_notification_worker.rb
@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+class WebPushNotificationWorker
+  include Sidekiq::Worker
+
+  sidekiq_options backtrace: true
+
+  def perform(recipient_id, notification_id)
+    recipient = Account.find(recipient_id)
+    notification = Notification.find(notification_id)
+
+    sessions_with_subscriptions = recipient.user.session_activations.reject { |session| session.web_push_subscription.nil? }
+
+    sessions_with_subscriptions.each do |session|
+      begin
+        session.web_push_subscription.push(notification)
+      rescue Webpush::InvalidSubscription, Webpush::ExpiredSubscription
+        # Subscription expiration is not currently implemented in any browser
+        session.web_push_subscription.destroy!
+        session.web_push_subscription = nil
+        session.save!
+      rescue Webpush::PayloadTooLarge => e
+        Rails.logger.error(e)
+      end
+    end
+  end
+end