1 files changed, 42 insertions, 21 deletions
diff --git a/app/services/notify_service.rb b/app/services/notify_service.rb
index a90f17cfd..d30b33876 100644
--- a/app/services/notify_service.rb
+++ b/app/services/notify_service.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 class NotifyService < BaseService
+  include Redisable
+
   def call(recipient, type, activity)
     @recipient    = recipient
     @activity     = activity
@@ -8,10 +10,15 @@ class NotifyService < BaseService
 
     return if recipient.user.nil? || blocked?
 
-    create_notification!
+    @notification.save!
+
+    # It's possible the underlying activity has been deleted
+    # between the save call and now
+    return if @notification.activity.nil?
+
     push_notification!
     push_to_conversation! if direct_message?
-    send_email! if email_enabled?
+    send_email! if email_needed?
   rescue ActiveRecord::RecordInvalid
     nil
   end
@@ -92,8 +99,8 @@ class NotifyService < BaseService
   end
 
   def blocked?
-    blocked   = @recipient.suspended?                            # Skip if the recipient account is suspended anyway
-    blocked ||= from_self? && @notification.type != :poll        # Skip for interactions with self
+    blocked   = @recipient.suspended?
+    blocked ||= from_self? && @notification.type != :poll
 
     return blocked if message? && from_staff?
 
@@ -117,38 +124,52 @@ class NotifyService < BaseService
     end
   end
 
-  def create_notification!
-    @notification.save!
+  def push_notification!
+    push_to_streaming_api! if subscribed_to_streaming_api?
+    push_to_web_push_subscriptions!
   end
 
-  def push_notification!
-    return if @notification.activity.nil?
+  def push_to_streaming_api!
+    redis.publish("timeline:#{@recipient.id}:notifications", Oj.dump(event: :notification, payload: InlineRenderer.render(@notification, @recipient, :notification)))
+  end
 
-    Redis.current.publish("timeline:#{@recipient.id}:notifications", Oj.dump(event: :notification, payload: InlineRenderer.render(@notification, @recipient, :notification)))
-    send_push_notifications!
+  def subscribed_to_streaming_api?
+    redis.exists?("subscribed:timeline:#{@recipient.id}") || redis.exists?("subscribed:timeline:#{@recipient.id}:notifications")
   end
 
   def push_to_conversation!
-    return if @notification.activity.nil?
     AccountConversation.add_status(@recipient, @notification.target_status)
   end
 
-  def send_push_notifications!
-    subscriptions_ids = ::Web::PushSubscription.where(user_id: @recipient.user.id)
-                                               .select { |subscription| subscription.pushable?(@notification) }
-                                               .map(&:id)
+  def push_to_web_push_subscriptions!
+    ::Web::PushNotificationWorker.push_bulk(web_push_subscriptions.select { |subscription| subscription.pushable?(@notification) }) { |subscription| [subscription.id, @notification.id] }
+  end
 
-    ::Web::PushNotificationWorker.push_bulk(subscriptions_ids) do |subscription_id|
-      [subscription_id, @notification.id]
-    end
+  def web_push_subscriptions
+    @web_push_subscriptions ||= ::Web::PushSubscription.where(user_id: @recipient.user.id).to_a
+  end
+
+  def subscribed_to_web_push?
+    web_push_subscriptions.any?
   end
 
   def send_email!
-    return if @notification.activity.nil?
-    NotificationMailer.public_send(@notification.type, @recipient, @notification).deliver_later(wait: 2.minutes)
+    NotificationMailer.public_send(@notification.type, @recipient, @notification).deliver_later(wait: 2.minutes) if NotificationMailer.respond_to?(@notification.type)
+  end
+
+  def email_needed?
+    (!recipient_online? || always_send_emails?) && send_email_for_notification_type?
+  end
+
+  def recipient_online?
+    subscribed_to_streaming_api? || subscribed_to_web_push?
+  end
+
+  def always_send_emails?
+    @recipient.user.settings.always_send_emails
   end
 
-  def email_enabled?
+  def send_email_for_notification_type?
     @recipient.user.settings.notification_emails[@notification.type.to_s]
   end
 end