11 files changed, 59 insertions, 14 deletions
diff --git a/app/workers/admin/suspension_worker.rb b/app/workers/admin/suspension_worker.rb
index 38761f3b9..7ef2b35ec 100644
--- a/app/workers/admin/suspension_worker.rb
+++ b/app/workers/admin/suspension_worker.rb
@@ -3,6 +3,8 @@
 class Admin::SuspensionWorker
   include Sidekiq::Worker
 
+  sidekiq_options queue: 'pull'
+
   def perform(account_id)
     SuspendAccountService.new.call(Account.find(account_id))
   end
diff --git a/app/workers/after_remote_follow_request_worker.rb b/app/workers/after_remote_follow_request_worker.rb
index 1f2db3061..928069211 100644
--- a/app/workers/after_remote_follow_request_worker.rb
+++ b/app/workers/after_remote_follow_request_worker.rb
@@ -13,5 +13,7 @@ class AfterRemoteFollowRequestWorker
 
     follow_request.destroy
     FollowService.new.call(follow_request.account, updated_account.acct)
+  rescue ActiveRecord::RecordNotFound
+    true
   end
 end
diff --git a/app/workers/after_remote_follow_worker.rb b/app/workers/after_remote_follow_worker.rb
index bdd2c2a91..d12fa3454 100644
--- a/app/workers/after_remote_follow_worker.rb
+++ b/app/workers/after_remote_follow_worker.rb
@@ -13,5 +13,7 @@ class AfterRemoteFollowWorker
 
     follow.destroy
     FollowService.new.call(follow.account, updated_account.acct)
+  rescue ActiveRecord::RecordNotFound
+    true
   end
 end
diff --git a/app/workers/application_worker.rb b/app/workers/application_worker.rb
new file mode 100644
index 000000000..436f24763
--- /dev/null
+++ b/app/workers/application_worker.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class ApplicationWorker
+  def info(message)
+    Rails.logger.info("#{self.class.name} - #{message}")
+  end
+end
diff --git a/app/workers/distribution_worker.rb b/app/workers/distribution_worker.rb
index f4e738d80..f7953689b 100644
--- a/app/workers/distribution_worker.rb
+++ b/app/workers/distribution_worker.rb
@@ -1,14 +1,11 @@
 # frozen_string_literal: true
 
-class DistributionWorker
+class DistributionWorker < ApplicationWorker
   include Sidekiq::Worker
 
   def perform(status_id)
-    status = Status.find(status_id)
-
-    FanOutOnWriteService.new.call(status)
-    WarmCacheService.new.call(status)
+    FanOutOnWriteService.new.call(Status.find(status_id))
   rescue ActiveRecord::RecordNotFound
-    true
+    info("Couldn't find the status")
   end
 end
diff --git a/app/workers/import_worker.rb b/app/workers/import_worker.rb
index 7cf29fb53..d5a33cada 100644
--- a/app/workers/import_worker.rb
+++ b/app/workers/import_worker.rb
@@ -46,7 +46,7 @@ class ImportWorker
 
       begin
         FollowService.new.call(from_account, row[0])
-      rescue Goldfinger::Error, HTTP::Error, OpenSSL::SSL::SSLError
+      rescue Mastodon::NotPermittedError, ActiveRecord::RecordNotFound, Goldfinger::Error, HTTP::Error, OpenSSL::SSL::SSLError
         next
       end
     end
diff --git a/app/workers/pubsubhubbub/delivery_worker.rb b/app/workers/pubsubhubbub/delivery_worker.rb
index 15005bc80..8412be4b7 100644
--- a/app/workers/pubsubhubbub/delivery_worker.rb
+++ b/app/workers/pubsubhubbub/delivery_worker.rb
@@ -13,6 +13,9 @@ class Pubsubhubbub::DeliveryWorker
   def perform(subscription_id, payload)
     subscription = Subscription.find(subscription_id)
     headers      = {}
+    host         = Addressable::URI.parse(subscription.callback_url).host
+
+    return if DomainBlock.blocked?(host)
 
     headers['User-Agent']      = 'Mastodon/PubSubHubbub'
     headers['Link']            = LinkHeader.new([[api_push_url, [%w(rel hub)]], [account_url(subscription.account, format: :atom), [%w(rel self)]]]).to_s
@@ -22,6 +25,7 @@ class Pubsubhubbub::DeliveryWorker
                    .headers(headers)
                    .post(subscription.callback_url, body: payload)
 
+    return subscription.destroy! if response.code > 299 && response.code < 500 && response.code != 429 # HTTP 4xx means error is not temporary, except for 429 (throttling)
     raise "Delivery failed for #{subscription.callback_url}: HTTP #{response.code}" unless response.code > 199 && response.code < 300
 
     subscription.touch(:last_successful_delivery_at)
diff --git a/app/workers/pubsubhubbub/distribution_worker.rb b/app/workers/pubsubhubbub/distribution_worker.rb
index 82ff257af..68ca0f870 100644
--- a/app/workers/pubsubhubbub/distribution_worker.rb
+++ b/app/workers/pubsubhubbub/distribution_worker.rb
@@ -10,14 +10,10 @@ class Pubsubhubbub::DistributionWorker
 
     return if stream_entry.hidden?
 
-    account  = stream_entry.account
-    renderer = AccountsController.renderer.new(method: 'get', http_host: Rails.configuration.x.local_domain, https: Rails.configuration.x.use_https)
-    payload  = renderer.render(:show, assigns: { account: account, entries: [stream_entry] }, formats: [:atom])
-    # domains  = account.followers_domains
+    account = stream_entry.account
+    payload = AtomSerializer.render(AtomSerializer.new.feed(account, [stream_entry]))
 
     Subscription.where(account: account).active.select('id, callback_url').find_each do |subscription|
-      host = Addressable::URI.parse(subscription.callback_url).host
-      next if DomainBlock.blocked?(host) # || !domains.include?(host)
       Pubsubhubbub::DeliveryWorker.perform_async(subscription.id, payload)
     end
   rescue ActiveRecord::RecordNotFound
diff --git a/app/workers/push_update_worker.rb b/app/workers/push_update_worker.rb
new file mode 100644
index 000000000..fbcdcf634
--- /dev/null
+++ b/app/workers/push_update_worker.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+class PushUpdateWorker
+  include Sidekiq::Worker
+
+  def perform(account_id, status_id)
+    account = Account.find(account_id)
+    status  = Status.find(status_id)
+    message = InlineRenderer.render(status, account, 'api/v1/statuses/show')
+
+    Redis.current.publish("timeline:#{account.id}", Oj.dump(event: :update, payload: message, queued_at: (Time.now.to_f * 1000.0).to_i))
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+end
diff --git a/app/workers/remote_profile_update_worker.rb b/app/workers/remote_profile_update_worker.rb
new file mode 100644
index 000000000..b91dc3466
--- /dev/null
+++ b/app/workers/remote_profile_update_worker.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+class RemoteProfileUpdateWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'pull'
+
+  def perform(account_id, body, resubscribe)
+    account = Account.find(account_id)
+
+    xml = Nokogiri::XML(body)
+    xml.encoding = 'utf-8'
+
+    author_container = xml.at_xpath('/xmlns:feed', xmlns: TagManager::XMLNS) || xml.at_xpath('/xmlns:entry', xmlns: TagManager::XMLNS)
+
+    UpdateRemoteProfileService.new.call(author_container, account, resubscribe)
+  rescue ActiveRecord::RecordNotFound
+    true
+  end
+end
diff --git a/app/workers/salmon_worker.rb b/app/workers/salmon_worker.rb
index fc95ce47f..d37d40432 100644
--- a/app/workers/salmon_worker.rb
+++ b/app/workers/salmon_worker.rb
@@ -7,7 +7,7 @@ class SalmonWorker
 
   def perform(account_id, body)
     ProcessInteractionService.new.call(body, Account.find(account_id))
-  rescue ActiveRecord::RecordNotFound
+  rescue Nokogiri::XML::XPath::SyntaxError, ActiveRecord::RecordNotFound
     true
   end
 end