Add spec for Pubsubhubbub::DistributionWorker. PuSH-deliver public items (#2954)

to all subscribers. IDN-normalize callback URLs for subscriptions on insert.

app/services/pubsubhubbub/subscribe_service.rb

@@ -6,9 +6,9 @@ class Pubsubhubbub::SubscribeService < BaseService
   attr_reader :account, :callback, :secret, :lease_seconds
 
   def call(account, callback, secret, lease_seconds)
-    @account = account
-    @callback = callback
-    @secret = secret
+    @account       = account
+    @callback      = Addressable::URI.parse(callback).normalize.to_s
+    @secret        = secret
     @lease_seconds = lease_seconds
 
     process_subscribe
@@ -52,7 +52,7 @@ class Pubsubhubbub::SubscribeService < BaseService
   end
 
   def blocked_domain?
-    DomainBlock.blocked? Addressable::URI.parse(callback).normalize.host
+    DomainBlock.blocked? Addressable::URI.parse(callback).host
   end
 
   def locate_subscription

app/services/pubsubhubbub/unsubscribe_service.rb

@@ -1,11 +1,11 @@
 # frozen_string_literal: true
 
 class Pubsubhubbub::UnsubscribeService < BaseService
-  attr_reader :account, :callback_url
+  attr_reader :account, :callback
 
-  def call(account, callback_url)
-    @account = account
-    @callback_url = callback_url
+  def call(account, callback)
+    @account  = account
+    @callback = Addressable::URI.parse(callback).normalize.to_s
 
     process_unsubscribe
   end
@@ -26,6 +26,6 @@ class Pubsubhubbub::UnsubscribeService < BaseService
   end
 
   def subscription
-    @_subscription ||= Subscription.find_by(account: account, callback_url: callback_url)
+    @_subscription ||= Subscription.find_by(account: account, callback_url: callback)
   end
 end

app/workers/pubsubhubbub/distribution_worker.rb

@@ -10,15 +10,21 @@ class Pubsubhubbub::DistributionWorker
 
     return if stream_entry.status&.direct_visibility?
 
-    account = stream_entry.account
-    payload = AtomSerializer.render(AtomSerializer.new.feed(account, [stream_entry]))
-    domains = account.followers_domains
+    @account = stream_entry.account
+    @payload = AtomSerializer.render(AtomSerializer.new.feed(@account, [stream_entry]))
+    @domains = @account.followers_domains
 
-    Subscription.where(account: account).active.select('id, callback_url').find_each do |subscription|
-      next unless domains.include?(Addressable::URI.parse(subscription.callback_url).host)
-      Pubsubhubbub::DeliveryWorker.perform_async(subscription.id, payload)
+    Subscription.where(account: @account).active.select('id, callback_url').find_each do |subscription|
+      next if stream_entry.hidden? && !allowed_to_receive?(subscription.callback_url)
+      Pubsubhubbub::DeliveryWorker.perform_async(subscription.id, @payload)
     end
   rescue ActiveRecord::RecordNotFound
     true
   end
+
+  private
+
+  def allowed_to_receive?(callback_url)
+    @domains.include?(Addressable::URI.parse(callback_url).host)
+  end
 end

spec/workers/pubsubhubbub/distribution_worker_spec.rb

@@ -0,0 +1,46 @@
+require 'rails_helper'
+
+describe Pubsubhubbub::DistributionWorker do
+  subject { Pubsubhubbub::DistributionWorker.new }
+
+  let!(:alice) { Fabricate(:account, username: 'alice') }
+  let!(:bob) { Fabricate(:account, username: 'bob', domain: 'example2.com') }
+  let!(:anonymous_subscription) { Fabricate(:subscription, account_id: alice.id, callback_url: 'http://example1.com', confirmed: true, lease_seconds: 3600) }
+  let!(:subscription_with_follower) { Fabricate(:subscription, account_id: alice.id, callback_url: 'http://example2.com', confirmed: true, lease_seconds: 3600) }
+
+  before do
+    bob.follow!(alice)
+  end
+
+  describe 'with public status' do
+    let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :public) }
+
+    it 'delivers payload to all subscriptions' do
+      allow(Pubsubhubbub::DeliveryWorker).to receive(:perform_async)
+      subject.perform(status.stream_entry.id)
+      expect(Pubsubhubbub::DeliveryWorker).to have_received(:perform_async).with(subscription_with_follower.id, /.*/)
+      expect(Pubsubhubbub::DeliveryWorker).to have_received(:perform_async).with(anonymous_subscription.id, /.*/)
+    end
+  end
+
+  describe 'with private status' do
+    let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :private) }
+
+    it 'delivers payload only to subscriptions with followers' do
+      allow(Pubsubhubbub::DeliveryWorker).to receive(:perform_async)
+      subject.perform(status.stream_entry.id)
+      expect(Pubsubhubbub::DeliveryWorker).to have_received(:perform_async).with(subscription_with_follower.id, /.*/)
+      expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:perform_async).with(anonymous_subscription.id, /.*/)
+    end
+  end
+
+  describe 'with direct status' do
+    let(:status) { Fabricate(:status, account: alice, text: 'Hello', visibility: :direct) }
+
+    it 'does not deliver payload' do
+      allow(Pubsubhubbub::DeliveryWorker).to receive(:perform_async)
+      subject.perform(status.stream_entry.id)
+      expect(Pubsubhubbub::DeliveryWorker).to_not have_received(:perform_async)
+    end
+  end
+end