7 files changed, 41 insertions, 0 deletions
diff --git a/app/services/favourite_service.rb b/app/services/favourite_service.rb
index 781b03b40..2f280e03f 100644
--- a/app/services/favourite_service.rb
+++ b/app/services/favourite_service.rb
@@ -7,7 +7,9 @@ class FavouriteService < BaseService
   # @return [Favourite]
   def call(account, status)
     favourite = Favourite.create!(account: account, status: status)
+
     HubPingWorker.perform_async(account.id)
+    Pubsubhubbub::DistributionWorker.perform_async(favourite.stream_entry.id)
 
     if status.local?
       NotifyService.new.call(status.account, favourite)
diff --git a/app/services/follow_service.rb b/app/services/follow_service.rb
index a57e1b28a..09fa295e3 100644
--- a/app/services/follow_service.rb
+++ b/app/services/follow_service.rb
@@ -19,7 +19,10 @@ class FollowService < BaseService
     end
 
     merge_into_timeline(target_account, source_account)
+
     HubPingWorker.perform_async(source_account.id)
+    Pubsubhubbub::DistributionWorker.perform_async(follow.stream_entry.id)
+
     follow
   end
 
diff --git a/app/services/post_status_service.rb b/app/services/post_status_service.rb
index 76366e984..979a157e9 100644
--- a/app/services/post_status_service.rb
+++ b/app/services/post_status_service.rb
@@ -14,8 +14,11 @@ class PostStatusService < BaseService
     attach_media(status, options[:media_ids])
     process_mentions_service.call(status)
     process_hashtags_service.call(status)
+
     DistributionWorker.perform_async(status.id)
     HubPingWorker.perform_async(account.id)
+    Pubsubhubbub::DistributionWorker.perform_async(status.stream_entry.id)
+
     status
   end
 
diff --git a/app/services/pubsubhubbub/subscribe_service.rb b/app/services/pubsubhubbub/subscribe_service.rb
new file mode 100644
index 000000000..343376d77
--- /dev/null
+++ b/app/services/pubsubhubbub/subscribe_service.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+class Pubsubhubbub::SubscribeService < BaseService
+  def call(account, callback, secret, lease_seconds)
+    return ['Invalid topic URL', 422] if account.nil?
+    return ['Invalid callback URL', 422] unless !callback.blank? && callback =~ /\A#{URI.regexp(%w(http https))}\z/
+
+    subscription = Subscription.where(account: account, callback_url: callback).first_or_create!(account: account, callback_url: callback)
+    Pubsubhubbub::ConfirmationWorker.perform_async(subscription.id, 'subscribe', secret, lease_seconds)
+
+    ['', 202]
+  end
+end
diff --git a/app/services/pubsubhubbub/unsubscribe_service.rb b/app/services/pubsubhubbub/unsubscribe_service.rb
new file mode 100644
index 000000000..a2fdc548a
--- /dev/null
+++ b/app/services/pubsubhubbub/unsubscribe_service.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+class Pubsubhubbub::SubscribeService < BaseService
+  def call(account, callback)
+    return ['Invalid topic URL', 422] if account.nil?
+
+    subscription = Subscription.where(account: account, callback_url: callback)
+
+    unless subscription.nil?
+      Pubsubhubbub::ConfirmationWorker.perform_async(subscription.id, 'unsubscribe')
+    end
+
+    ['', 202]
+  end
+end
diff --git a/app/services/reblog_service.rb b/app/services/reblog_service.rb
index 6543d4ae7..39fdb4ea7 100644
--- a/app/services/reblog_service.rb
+++ b/app/services/reblog_service.rb
@@ -7,8 +7,10 @@ class ReblogService < BaseService
   # @return [Status]
   def call(account, reblogged_status)
     reblog = account.statuses.create!(reblog: reblogged_status, text: '')
+
     DistributionWorker.perform_async(reblog.id)
     HubPingWorker.perform_async(account.id)
+    Pubsubhubbub::DistributionWorker.perform_async(reblog.stream_entry.id)
 
     if reblogged_status.local?
       NotifyService.new.call(reblogged_status.account, reblog)
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index 689abc97b..058fd3f18 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -10,6 +10,9 @@ class RemoveStatusService < BaseService
     remove_from_public(status)
 
     status.destroy!
+
+    HubPingWorker.perform_async(status.account.id)
+    Pubsubhubbub::DistributionWorker.perform_async(status.stream_entry.id)
   end
 
   private