Support pushing and receiving updates to poll tallies (#10209)

* Process incoming poll tallies update

* Send Update on poll vote

* Do not send Updates for a poll more often than once every 3 minutes

* Include voters in people to notify of results update

* Schedule closing poll worker on poll creation

* Add new notification type for ending polls

* Add front-end support for ended poll notifications

* Fix UpdatePollSerializer

* Fix Updates not being triggered by local votes

* Fix tests failure

* Fix web push notifications for closing polls

* Minor cleanup

* Notify voters of both remote and local polls when those close

* Fix delivery of poll updates to mentioned accounts and voters

app/services/activitypub/fetch_remote_poll_service.rb

@@ -4,54 +4,7 @@ class ActivityPub::FetchRemotePollService < BaseService
   include JsonLdHelper
 
   def call(poll, on_behalf_of = nil)
-    @json = fetch_resource(poll.status.uri, true, on_behalf_of)
-
-    return unless supported_context? && expected_type?
-
-    expires_at = begin
-      if @json['closed'].is_a?(String)
-        @json['closed']
-      elsif !@json['closed'].nil? && !@json['closed'].is_a?(FalseClass)
-        Time.now.utc
-      else
-        @json['endTime']
-      end
-    end
-
-    items = begin
-      if @json['anyOf'].is_a?(Array)
-        @json['anyOf']
-      else
-        @json['oneOf']
-      end
-    end
-
-    latest_options = items.map { |item| item['name'].presence || item['content'] }
-
-    # If for some reasons the options were changed, it invalidates all previous
-    # votes, so we need to remove them
-    poll.votes.delete_all if latest_options != poll.options
-
-    begin
-      poll.update!(
-        last_fetched_at: Time.now.utc,
-        expires_at: expires_at,
-        options: latest_options,
-        cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 }
-      )
-    rescue ActiveRecord::StaleObjectError
-      poll.reload
-      retry
-    end
-  end
-
-  private
-
-  def supported_context?
-    super(@json)
-  end
-
-  def expected_type?
-    equals_or_includes_any?(@json['type'], %w(Question))
+    json = fetch_resource(poll.status.uri, true, on_behalf_of)
+    ActivityPub::ProcessPollService.new.call(poll, json)
   end
 end

app/services/activitypub/process_poll_service.rb

@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+class ActivityPub::ProcessPollService < BaseService
+  include JsonLdHelper
+
+  def call(poll, json)
+    @json = json
+    return unless supported_context? && expected_type?
+
+    previous_expires_at = poll.expires_at
+
+    expires_at = begin
+      if @json['closed'].is_a?(String)
+        @json['closed']
+      elsif !@json['closed'].nil? && !@json['closed'].is_a?(FalseClass)
+        Time.now.utc
+      else
+        @json['endTime']
+      end
+    end
+
+    items = begin
+      if @json['anyOf'].is_a?(Array)
+        @json['anyOf']
+      else
+        @json['oneOf']
+      end
+    end
+
+    latest_options = items.map { |item| item['name'].presence || item['content'] }
+
+    # If for some reasons the options were changed, it invalidates all previous
+    # votes, so we need to remove them
+    poll.votes.delete_all if latest_options != poll.options
+
+    begin
+      poll.update!(
+        last_fetched_at: Time.now.utc,
+        expires_at: expires_at,
+        options: latest_options,
+        cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 }
+      )
+    rescue ActiveRecord::StaleObjectError
+      poll.reload
+      retry
+    end
+
+    # If the poll had no expiration date set but now has, and people have voted,
+    # schedule a notification.
+    if previous_expires_at.nil? && poll.expires_at.present? && poll.votes.exists?
+      PollExpirationNotifyWorker.perform_at(poll.expires_at + 5.minutes, poll.id)
+    end
+  end
+
+  private
+
+  def supported_context?
+    super(@json)
+  end
+
+  def expected_type?
+    equals_or_includes_any?(@json['type'], %w(Question))
+  end
+end

app/services/notify_service.rb

@@ -38,6 +38,10 @@ class NotifyService < BaseService
     false
   end
 
+  def blocked_poll?
+    false
+  end
+
   def following_sender?
     return @following_sender if defined?(@following_sender)
     @following_sender = @recipient.following?(@notification.from_account) || @recipient.requested?(@notification.from_account)
@@ -88,7 +92,7 @@ class NotifyService < BaseService
 
   def blocked?
     blocked   = @recipient.suspended?                            # Skip if the recipient account is suspended anyway
-    blocked ||= from_self?                                       # Skip for interactions with self
+    blocked ||= from_self? unless @notification.type == :poll    # Skip for interactions with self
 
     return blocked if message? && from_staff?
 

app/services/post_status_service.rb

@@ -90,6 +90,7 @@ class PostStatusService < BaseService
     DistributionWorker.perform_async(@status.id)
     Pubsubhubbub::DistributionWorker.perform_async(@status.stream_entry.id)
     ActivityPub::DistributionWorker.perform_async(@status.id)
+    PollExpirationNotifyWorker.perform_at(@status.poll.expires_at, @status.poll.id) if @status.poll
   end
 
   def validate_media!

app/services/vote_service.rb

@@ -19,14 +19,17 @@ class VoteService < BaseService
       end
     end
 
-    return if @poll.account.local?
-
-    @votes.each do |vote|
-      ActivityPub::DeliveryWorker.perform_async(
-        build_json(vote),
-        @account.id,
-        @poll.account.inbox_url
-      )
+    if @poll.account.local?
+      ActivityPub::DistributePollUpdateWorker.perform_in(3.minutes, @poll.status.id) unless @poll.hide_totals
+    else
+      @votes.each do |vote|
+        ActivityPub::DeliveryWorker.perform_async(
+          build_json(vote),
+          @account.id,
+          @poll.account.inbox_url
+        )
+      end
+      PollExpirationNotifyWorker.perform_at(@poll.expires_at + 5.minutes, @poll.id) unless @poll.expires_at.nil?
     end
   end