Add support for editing for published statuses (#16697)

* Add support for editing for published statuses

* Fix references to stripped-out code

* Various fixes and improvements

* Further fixes and improvements

* Fix updates being potentially sent to unauthorized recipients

* Various fixes and improvements

* Fix wrong words in test

* Fix notifying accounts that were tagged but were not in the audience

* Fix mistake

6 files changed, 395 insertions, 158 deletions

diff --git a/app/services/activitypub/fetch_remote_poll_service.rb b/app/services/activitypub/fetch_remote_poll_service.rb
index 1c79ecf11..1829e791c 100644
--- a/app/services/activitypub/fetch_remote_poll_service.rb
+++ b/app/services/activitypub/fetch_remote_poll_service.rb
@@ -8,6 +8,6 @@ class ActivityPub::FetchRemotePollService < BaseService
 
     return unless supported_context?(json)
 
-    ActivityPub::ProcessPollService.new.call(poll, json)
+    ActivityPub::ProcessStatusUpdateService.new.call(poll.status, json)
   end
 end
diff --git a/app/services/activitypub/process_poll_service.rb b/app/services/activitypub/process_poll_service.rb
deleted file mode 100644
index d83e614d8..000000000
--- a/app/services/activitypub/process_poll_service.rb
+++ /dev/null
@@ -1,64 +0,0 @@
-# frozen_string_literal: true
-
-class ActivityPub::ProcessPollService < BaseService
-  include JsonLdHelper
-
-  def call(poll, json)
-    @json = json
-
-    return unless 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
-
-    voters_count = @json['votersCount']
-
-    latest_options = items.filter_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 },
-        voters_count: voters_count
-      )
-    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 expected_type?
-    equals_or_includes_any?(@json['type'], %w(Question))
-  end
-end
diff --git a/app/services/activitypub/process_status_update_service.rb b/app/services/activitypub/process_status_update_service.rb
new file mode 100644
index 000000000..e3e9b9d6a
--- /dev/null
+++ b/app/services/activitypub/process_status_update_service.rb
@@ -0,0 +1,275 @@
+# frozen_string_literal: true
+
+class ActivityPub::ProcessStatusUpdateService < BaseService
+  include JsonLdHelper
+
+  def call(status, json)
+    @json                      = json
+    @status_parser             = ActivityPub::Parser::StatusParser.new(@json)
+    @uri                       = @status_parser.uri
+    @status                    = status
+    @account                   = status.account
+    @media_attachments_changed = false
+
+    # Only native types can be updated at the moment
+    return if !expected_type? || already_updated_more_recently?
+
+    # Only allow processing one create/update per status at a time
+    RedisLock.acquire(lock_options) do |lock|
+      if lock.acquired?
+        Status.transaction do
+          create_previous_edit!
+          update_media_attachments!
+          update_poll!
+          update_immediate_attributes!
+          update_metadata!
+          create_edit!
+        end
+
+        queue_poll_notifications!
+        reset_preview_card!
+        broadcast_updates!
+      else
+        raise Mastodon::RaceConditionError
+      end
+    end
+  end
+
+  private
+
+  def update_media_attachments!
+    previous_media_attachments = @status.media_attachments.to_a
+    next_media_attachments     = []
+
+    as_array(@json['attachment']).each do |attachment|
+      media_attachment_parser = ActivityPub::Parser::MediaAttachmentParser.new(attachment)
+
+      next if media_attachment_parser.remote_url.blank? || next_media_attachments.size > 4
+
+      begin
+        media_attachment   = previous_media_attachments.find { |previous_media_attachment| previous_media_attachment.remote_url == media_attachment_parser.remote_url }
+        media_attachment ||= MediaAttachment.new(account: @account, remote_url: media_attachment_parser.remote_url)
+
+        # If a previously existing media attachment was significantly updated, mark
+        # media attachments as changed even if none were added or removed
+        if media_attachment_parser.significantly_changes?(media_attachment)
+          @media_attachments_changed = true
+        end
+
+        media_attachment.description          = media_attachment_parser.description
+        media_attachment.focus                = media_attachment_parser.focus
+        media_attachment.thumbnail_remote_url = media_attachment_parser.thumbnail_remote_url
+        media_attachment.blurhash             = media_attachment_parser.blurhash
+        media_attachment.save!
+
+        next_media_attachments << media_attachment
+
+        next if unsupported_media_type?(media_attachment_parser.file_content_type) || skip_download?
+
+        RedownloadMediaWorker.perform_async(media_attachment.id) if media_attachment.remote_url_previously_changed? || media_attachment.thumbnail_remote_url_previously_changed?
+      rescue Addressable::URI::InvalidURIError => e
+        Rails.logger.debug "Invalid URL in attachment: #{e}"
+      end
+    end
+
+    removed_media_attachments = previous_media_attachments - next_media_attachments
+    added_media_attachments   = next_media_attachments - previous_media_attachments
+
+    MediaAttachment.where(id: removed_media_attachments.map(&:id)).update_all(status_id: nil)
+    MediaAttachment.where(id: added_media_attachments.map(&:id)).update_all(status_id: @status.id)
+
+    @media_attachments_changed = true if removed_media_attachments.positive? || added_media_attachments.positive?
+  end
+
+  def update_poll!
+    previous_poll        = @status.preloadable_poll
+    @previous_expires_at = previous_poll&.expires_at
+    poll_parser          = ActivityPub::Parser::PollParser.new(@json)
+
+    if poll_parser.valid?
+      poll = previous_poll || @account.polls.new(status: @status)
+
+      # If for some reasons the options were changed, it invalidates all previous
+      # votes, so we need to remove them
+      if poll_parser.significantly_changes?(poll)
+        @media_attachments_changed = true
+        poll.votes.delete_all unless poll.new_record?
+      end
+
+      poll.last_fetched_at = Time.now.utc
+      poll.options         = poll_parser.options
+      poll.multiple        = poll_parser.multiple
+      poll.expires_at      = poll_parser.expires_at
+      poll.voters_count    = poll_parser.voters_count
+      poll.cached_tallies  = poll_parser.cached_tallies
+      poll.save!
+
+      @status.poll_id = poll.id
+    elsif previous_poll.present?
+      previous_poll.destroy!
+      @media_attachments_changed = true
+      @status.poll_id = nil
+    end
+  end
+
+  def update_immediate_attributes!
+    @status.text         = @status_parser.text || ''
+    @status.spoiler_text = @status_parser.spoiler_text || ''
+    @status.sensitive    = @account.sensitized? || @status_parser.sensitive || false
+    @status.language     = @status_parser.language || detected_language
+    @status.edited_at    = @status_parser.edited_at || Time.now.utc
+
+    @status.save!
+  end
+
+  def update_metadata!
+    @raw_tags     = []
+    @raw_mentions = []
+    @raw_emojis   = []
+
+    as_array(@json['tag']).each do |tag|
+      if equals_or_includes?(tag['type'], 'Hashtag')
+        @raw_tags << tag['name']
+      elsif equals_or_includes?(tag['type'], 'Mention')
+        @raw_mentions << tag['href']
+      elsif equals_or_includes?(tag['type'], 'Emoji')
+        @raw_emojis << tag
+      end
+    end
+
+    update_tags!
+    update_mentions!
+    update_emojis!
+  end
+
+  def update_tags!
+    @status.tags = Tag.find_or_create_by_names(@raw_tags)
+  end
+
+  def update_mentions!
+    previous_mentions = @status.active_mentions.includes(:account).to_a
+    current_mentions  = []
+
+    @raw_mentions.each do |href|
+      next if href.blank?
+
+      account   = ActivityPub::TagManager.instance.uri_to_resource(href, Account)
+      account ||= ActivityPub::FetchRemoteAccountService.new.call(href)
+
+      next if account.nil?
+
+      mention   = previous_mentions.find { |x| x.account_id == account.id }
+      mention ||= account.mentions.new(status: @status)
+
+      current_mentions << mention
+    end
+
+    current_mentions.each do |mention|
+      mention.save if mention.new_record?
+    end
+
+    # If previous mentions are no longer contained in the text, convert them
+    # to silent mentions, since withdrawing access from someone who already
+    # received a notification might be more confusing
+    removed_mentions = previous_mentions - current_mentions
+
+    Mention.where(id: removed_mentions.map(&:id)).update_all(silent: true) unless removed_mentions.empty?
+  end
+
+  def update_emojis!
+    return if skip_download?
+
+    @raw_emojis.each do |raw_emoji|
+      custom_emoji_parser = ActivityPub::Parser::CustomEmojiParser.new(raw_emoji)
+
+      next if custom_emoji_parser.shortcode.blank? || custom_emoji_parser.image_remote_url.blank?
+
+      emoji = CustomEmoji.find_by(shortcode: custom_emoji_parser.shortcode, domain: @account.domain)
+
+      next unless emoji.nil? || custom_emoji_parser.image_remote_url != emoji.image_remote_url || (custom_emoji_parser.updated_at && custom_emoji_parser.updated_at >= emoji.updated_at)
+
+      begin
+        emoji ||= CustomEmoji.new(domain: @account.domain, shortcode: custom_emoji_parser.shortcode, uri: custom_emoji_parser.uri)
+        emoji.image_remote_url = custom_emoji_parser.image_remote_url
+        emoji.save
+      rescue Seahorse::Client::NetworkingError => e
+        Rails.logger.warn "Error storing emoji: #{e}"
+      end
+    end
+  end
+
+  def expected_type?
+    equals_or_includes_any?(@json['type'], %w(Note Question))
+  end
+
+  def lock_options
+    { redis: Redis.current, key: "create:#{@uri}", autorelease: 15.minutes.seconds }
+  end
+
+  def detected_language
+    LanguageDetector.instance.detect(@status_parser.text, @account)
+  end
+
+  def create_previous_edit!
+    # We only need to create a previous edit when no previous edits exist, e.g.
+    # when the status has never been edited. For other cases, we always create
+    # an edit, so the step can be skipped
+
+    return if @status.edits.any?
+
+    @status.edits.create(
+      text: @status.text,
+      spoiler_text: @status.spoiler_text,
+      media_attachments_changed: false,
+      account_id: @account.id,
+      created_at: @status.created_at
+    )
+  end
+
+  def create_edit!
+    return unless @status.text_previously_changed? || @status.spoiler_text_previously_changed? || @media_attachments_changed
+
+    @status_edit = @status.edits.create(
+      text: @status.text,
+      spoiler_text: @status.spoiler_text,
+      media_attachments_changed: @media_attachments_changed,
+      account_id: @account.id,
+      created_at: @status.edited_at
+    )
+  end
+
+  def skip_download?
+    return @skip_download if defined?(@skip_download)
+
+    @skip_download ||= DomainBlock.reject_media?(@account.domain)
+  end
+
+  def unsupported_media_type?(mime_type)
+    mime_type.present? && !MediaAttachment.supported_mime_types.include?(mime_type)
+  end
+
+  def already_updated_more_recently?
+    @status.edited_at.present? && @status_parser.edited_at.present? && @status.edited_at > @status_parser.edited_at
+  end
+
+  def reset_preview_card!
+    @status.preview_cards.clear if @status.text_previously_changed? || @status.spoiler_text.present?
+    LinkCrawlWorker.perform_in(rand(1..59).seconds, @status.id) if @status.spoiler_text.blank?
+  end
+
+  def broadcast_updates!
+    ::DistributionWorker.perform_async(@status.id, update: true)
+  end
+
+  def queue_poll_notifications!
+    poll = @status.preloadable_poll
+
+    # If the poll had no expiration date set but now has, or now has a sooner
+    # expiration date, and people have voted, schedule a notification
+
+    return unless poll.present? && poll.expires_at.present? && poll.votes.exists?
+
+    PollExpirationNotifyWorker.remove_from_scheduled(poll.id) if @previous_expires_at.present? && @previous_expires_at > poll.expires_at
+    PollExpirationNotifyWorker.perform_at(poll.expires_at + 5.minutes, poll.id)
+  end
+end
diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb
index b72bb82d3..f62f78a79 100644
--- a/app/services/fan_out_on_write_service.rb
+++ b/app/services/fan_out_on_write_service.rb
@@ -3,107 +3,126 @@
 class FanOutOnWriteService < BaseService
   # Push a status into home and mentions feeds
   # @param [Status] status
-  def call(status)
-    raise Mastodon::RaceConditionError if status.visibility.nil?
+  # @param [Hash] options
+  # @option options [Boolean] update
+  # @option options [Array<Integer>] silenced_account_ids
+  def call(status, options = {})
+    @status    = status
+    @account   = status.account
+    @options   = options
 
-    deliver_to_self(status) if status.account.local?
+    check_race_condition!
 
-    if status.direct_visibility?
-      deliver_to_mentioned_followers(status)
-      deliver_to_own_conversation(status)
-    elsif status.limited_visibility?
-      deliver_to_mentioned_followers(status)
-    else
-      deliver_to_followers(status)
-      deliver_to_lists(status)
-    end
+    fan_out_to_local_recipients!
+    fan_out_to_public_streams! if broadcastable?
+  end
 
-    return if status.account.silenced? || !status.public_visibility? || status.reblog?
+  private
 
-    render_anonymous_payload(status)
+  def check_race_condition!
+    # I don't know why but at some point we had an issue where
+    # this service was being executed with status objects
+    # that had a null visibility - which should not be possible
+    # since the column in the database is not nullable.
+    #
+    # This check re-queues the service to be run at a later time
+    # with the full object, if something like it occurs
 
-    deliver_to_hashtags(status)
+    raise Mastodon::RaceConditionError if @status.visibility.nil?
+  end
 
-    return if status.reply? && status.in_reply_to_account_id != status.account_id
+  def fan_out_to_local_recipients!
+    deliver_to_self!
+    notify_mentioned_accounts!
 
-    deliver_to_public(status)
-    deliver_to_media(status) if status.media_attachments.any?
+    case @status.visibility.to_sym
+    when :public, :unlisted, :private
+      deliver_to_all_followers!
+      deliver_to_lists!
+    when :limited
+      deliver_to_mentioned_followers!
+    else
+      deliver_to_mentioned_followers!
+      deliver_to_conversation!
+    end
   end
 
-  private
+  def fan_out_to_public_streams!
+    broadcast_to_hashtag_streams!
+    broadcast_to_public_streams!
+  end
 
-  def deliver_to_self(status)
-    Rails.logger.debug "Delivering status #{status.id} to author"
-    FeedManager.instance.push_to_home(status.account, status)
+  def deliver_to_self!
+    FeedManager.instance.push_to_home(@account, @status, update: update?) if @account.local?
   end
 
-  def deliver_to_followers(status)
-    Rails.logger.debug "Delivering status #{status.id} to followers"
+  def notify_mentioned_accounts!
+    @status.active_mentions.where.not(id: @options[:silenced_account_ids] || []).joins(:account).merge(Account.local).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
+      LocalNotificationWorker.push_bulk(mentions) do |mention|
+        [mention.account_id, mention.id, 'Mention', :mention]
+      end
+    end
+  end
 
-    status.account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
+  def deliver_to_all_followers!
+    @account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
       FeedInsertWorker.push_bulk(followers) do |follower|
-        [status.id, follower.id, :home]
+        [@status.id, follower.id, :home, update: update?]
       end
     end
   end
 
-  def deliver_to_lists(status)
-    Rails.logger.debug "Delivering status #{status.id} to lists"
-
-    status.account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
+  def deliver_to_lists!
+    @account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
       FeedInsertWorker.push_bulk(lists) do |list|
-        [status.id, list.id, :list]
+        [@status.id, list.id, :list, update: update?]
       end
     end
   end
 
-  def deliver_to_mentioned_followers(status)
-    Rails.logger.debug "Delivering status #{status.id} to limited followers"
-
-    status.mentions.joins(:account).merge(status.account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
+  def deliver_to_mentioned_followers!
+    @status.mentions.joins(:account).merge(@account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
       FeedInsertWorker.push_bulk(mentions) do |mention|
-        [status.id, mention.account_id, :home]
+        [@status.id, mention.account_id, :home, update: update?]
       end
     end
   end
 
-  def render_anonymous_payload(status)
-    @payload = InlineRenderer.render(status, nil, :status)
-    @payload = Oj.dump(event: :update, payload: @payload)
+  def broadcast_to_hashtag_streams!
+    @status.tags.pluck(:name).each do |hashtag|
+      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", anonymous_payload)
+      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", anonymous_payload) if @status.local?
+    end
   end
 
-  def deliver_to_hashtags(status)
-    Rails.logger.debug "Delivering status #{status.id} to hashtags"
+  def broadcast_to_public_streams!
+    return if @status.reply? && @status.in_reply_to_account_id != @account.id
+
+    Redis.current.publish('timeline:public', anonymous_payload)
+    Redis.current.publish(@status.local? ? 'timeline:public:local' : 'timeline:public:remote', anonymous_payload)
 
-    status.tags.pluck(:name).each do |hashtag|
-      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
-      Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if status.local?
+    if @status.media_attachments.any?
+      Redis.current.publish('timeline:public:media', anonymous_payload)
+      Redis.current.publish(@status.local? ? 'timeline:public:local:media' : 'timeline:public:remote:media', anonymous_payload)
     end
   end
 
-  def deliver_to_public(status)
-    Rails.logger.debug "Delivering status #{status.id} to public timeline"
-
-    Redis.current.publish('timeline:public', @payload)
-    if status.local?
-      Redis.current.publish('timeline:public:local', @payload)
-    else
-      Redis.current.publish('timeline:public:remote', @payload)
-    end
+  def deliver_to_conversation!
+    AccountConversation.add_status(@account, @status) unless update?
   end
 
-  def deliver_to_media(status)
-    Rails.logger.debug "Delivering status #{status.id} to media timeline"
+  def anonymous_payload
+    @anonymous_payload ||= Oj.dump(
+      event: update? ? :'status.update' : :update,
+      payload: InlineRenderer.render(@status, nil, :status)
+    )
+  end
 
-    Redis.current.publish('timeline:public:media', @payload)
-    if status.local?
-      Redis.current.publish('timeline:public:local:media', @payload)
-    else
-      Redis.current.publish('timeline:public:remote:media', @payload)
-    end
+  def update?
+    @is_update
   end
 
-  def deliver_to_own_conversation(status)
-    AccountConversation.add_status(status.account, status)
+  def broadcastable?
+    @status.public_visibility? && !@status.reblog? && !@account.silenced?
   end
 end
diff --git a/app/services/process_mentions_service.rb b/app/services/process_mentions_service.rb
index 73dbb1834..9d239fc65 100644
--- a/app/services/process_mentions_service.rb
+++ b/app/services/process_mentions_service.rb
@@ -8,12 +8,23 @@ class ProcessMentionsService < BaseService
   # remote users
   # @param [Status] status
   def call(status)
-    return unless status.local?
+    @status = status
 
-    @status  = status
-    mentions = []
+    return unless @status.local?
 
-    status.text = status.text.gsub(Account::MENTION_RE) do |match|
+    @previous_mentions = @status.active_mentions.includes(:account).to_a
+    @current_mentions  = []
+
+    Status.transaction do
+      scan_text!
+      assign_mentions!
+    end
+  end
+
+  private
+
+  def scan_text!
+    @status.text = @status.text.gsub(Account::MENTION_RE) do |match|
       username, domain = Regexp.last_match(1).split('@')
 
       domain = begin
@@ -26,49 +37,45 @@ class ProcessMentionsService < BaseService
 
       mentioned_account = Account.find_remote(username, domain)
 
+      # If the account cannot be found or isn't the right protocol,
+      # first try to resolve it
       if mention_undeliverable?(mentioned_account)
         begin
-          mentioned_account = resolve_account_service.call(Regexp.last_match(1))
+          mentioned_account = ResolveAccountService.new.call(Regexp.last_match(1))
         rescue Webfinger::Error, HTTP::Error, OpenSSL::SSL::SSLError, Mastodon::UnexpectedResponseError
           mentioned_account = nil
         end
       end
 
+      # If after resolving it still isn't found or isn't the right
+      # protocol, then give up
       next match if mention_undeliverable?(mentioned_account) || mentioned_account&.suspended?
 
-      mention = mentioned_account.mentions.new(status: status)
-      mentions << mention if mention.save
+      mention   = @previous_mentions.find { |x| x.account_id == mentioned_account.id }
+      mention ||= mentioned_account.mentions.new(status: @status)
+
+      @current_mentions << mention
 
       "@#{mentioned_account.acct}"
     end
 
-    status.save!
-
-    mentions.each { |mention| create_notification(mention) }
+    @status.save!
   end
 
-  private
-
-  def mention_undeliverable?(mentioned_account)
-    mentioned_account.nil? || (!mentioned_account.local? && mentioned_account.ostatus?)
-  end
-
-  def create_notification(mention)
-    mentioned_account = mention.account
-
-    if mentioned_account.local?
-      LocalNotificationWorker.perform_async(mentioned_account.id, mention.id, mention.class.name, :mention)
-    elsif mentioned_account.activitypub?
-      ActivityPub::DeliveryWorker.perform_async(activitypub_json, mention.status.account_id, mentioned_account.inbox_url, { synchronize_followers: !mention.status.distributable? })
+  def assign_mentions!
+    @current_mentions.each do |mention|
+      mention.save if mention.new_record?
     end
-  end
 
-  def activitypub_json
-    return @activitypub_json if defined?(@activitypub_json)
-    @activitypub_json = Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status), ActivityPub::ActivitySerializer, signer: @status.account))
+    # If previous mentions are no longer contained in the text, convert them
+    # to silent mentions, since withdrawing access from someone who already
+    # received a notification might be more confusing
+    removed_mentions = @previous_mentions - @current_mentions
+
+    Mention.where(id: removed_mentions.map(&:id)).update_all(silent: true) unless removed_mentions.empty?
   end
 
-  def resolve_account_service
-    ResolveAccountService.new
+  def mention_undeliverable?(mentioned_account)
+    mentioned_account.nil? || (!mentioned_account.local? && !mentioned_account.activitypub?)
   end
 end
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index 3535b503b..bec95bb1b 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -87,7 +87,7 @@ class RemoveStatusService < BaseService
     # the author and wouldn't normally receive the delete
     # notification - so here, we explicitly send it to them
 
-    status_reach_finder = StatusReachFinder.new(@status)
+    status_reach_finder = StatusReachFinder.new(@status, unsafe: true)
 
     ActivityPub::DeliveryWorker.push_bulk(status_reach_finder.inboxes) do |inbox_url|
       [signed_activity_json, @account.id, inbox_url]