fan_out_on_write_service.rb 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. # frozen_string_literal: true
  2. class FanOutOnWriteService < BaseService
  3. include Redisable
  4. # Push a status into home and mentions feeds
  5. # @param [Status] status
  6. # @param [Hash] options
  7. # @option options [Boolean] update
  8. # @option options [Array<Integer>] silenced_account_ids
  9. # @option options [Boolean] skip_notifications
  10. def call(status, options = {})
  11. @status = status
  12. @account = status.account
  13. @options = options
  14. check_race_condition!
  15. warm_payload_cache!
  16. fan_out_to_local_recipients!
  17. fan_out_to_public_recipients! if broadcastable?
  18. fan_out_to_public_streams! if broadcastable?
  19. end
  20. private
  21. def check_race_condition!
  22. # I don't know why but at some point we had an issue where
  23. # this service was being executed with status objects
  24. # that had a null visibility - which should not be possible
  25. # since the column in the database is not nullable.
  26. #
  27. # This check re-queues the service to be run at a later time
  28. # with the full object, if something like it occurs
  29. raise Mastodon::RaceConditionError if @status.visibility.nil?
  30. end
  31. def fan_out_to_local_recipients!
  32. deliver_to_self!
  33. unless @options[:skip_notifications]
  34. notify_mentioned_accounts!
  35. notify_about_update! if update?
  36. end
  37. case @status.visibility.to_sym
  38. when :public, :unlisted, :private
  39. deliver_to_all_followers!
  40. deliver_to_lists!
  41. when :limited
  42. deliver_to_mentioned_followers!
  43. else
  44. deliver_to_mentioned_followers!
  45. deliver_to_conversation!
  46. end
  47. end
  48. def fan_out_to_public_recipients!
  49. deliver_to_hashtag_followers!
  50. end
  51. def fan_out_to_public_streams!
  52. broadcast_to_hashtag_streams!
  53. broadcast_to_public_streams!
  54. end
  55. def deliver_to_self!
  56. FeedManager.instance.push_to_home(@account, @status, update: update?) if @account.local?
  57. end
  58. def notify_mentioned_accounts!
  59. @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|
  60. LocalNotificationWorker.push_bulk(mentions) do |mention|
  61. [mention.account_id, mention.id, 'Mention', 'mention']
  62. end
  63. next unless update?
  64. # This may result in duplicate update payloads, but this ensures clients
  65. # are aware of edits to posts only appearing in mention notifications
  66. # (e.g. private mentions or mentions by people they do not follow)
  67. PushUpdateWorker.push_bulk(mentions.filter { |mention| subscribed_to_streaming_api?(mention.account_id) }) do |mention|
  68. [mention.account_id, @status.id, "timeline:#{mention.account_id}:notifications", { 'update' => true }]
  69. end
  70. end
  71. end
  72. def notify_about_update!
  73. @status.reblogged_by_accounts.merge(Account.local).select(:id).reorder(nil).find_in_batches do |accounts|
  74. LocalNotificationWorker.push_bulk(accounts) do |account|
  75. [account.id, @status.id, 'Status', 'update']
  76. end
  77. end
  78. end
  79. def deliver_to_all_followers!
  80. @account.followers_for_local_distribution.select(:id).reorder(nil).find_in_batches do |followers|
  81. FeedInsertWorker.push_bulk(followers) do |follower|
  82. [@status.id, follower.id, 'home', { 'update' => update? }]
  83. end
  84. end
  85. end
  86. def deliver_to_hashtag_followers!
  87. TagFollow.where(tag_id: @status.tags.map(&:id)).select(:id, :account_id).reorder(nil).find_in_batches do |follows|
  88. FeedInsertWorker.push_bulk(follows) do |follow|
  89. [@status.id, follow.account_id, 'tags', { 'update' => update? }]
  90. end
  91. end
  92. end
  93. def deliver_to_lists!
  94. @account.lists_for_local_distribution.select(:id).reorder(nil).find_in_batches do |lists|
  95. FeedInsertWorker.push_bulk(lists) do |list|
  96. [@status.id, list.id, 'list', { 'update' => update? }]
  97. end
  98. end
  99. end
  100. def deliver_to_mentioned_followers!
  101. @status.mentions.joins(:account).merge(@account.followers_for_local_distribution).select(:id, :account_id).reorder(nil).find_in_batches do |mentions|
  102. FeedInsertWorker.push_bulk(mentions) do |mention|
  103. [@status.id, mention.account_id, 'home', { 'update' => update? }]
  104. end
  105. end
  106. end
  107. def broadcast_to_hashtag_streams!
  108. @status.tags.map(&:name).each do |hashtag|
  109. redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", anonymous_payload)
  110. redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", anonymous_payload) if @status.local?
  111. end
  112. end
  113. def broadcast_to_public_streams!
  114. return if @status.reply? && @status.in_reply_to_account_id != @account.id
  115. redis.publish('timeline:public', anonymous_payload)
  116. redis.publish(@status.local? ? 'timeline:public:local' : 'timeline:public:remote', anonymous_payload)
  117. if @status.with_media?
  118. redis.publish('timeline:public:media', anonymous_payload)
  119. redis.publish(@status.local? ? 'timeline:public:local:media' : 'timeline:public:remote:media', anonymous_payload)
  120. end
  121. end
  122. def deliver_to_conversation!
  123. AccountConversation.add_status(@account, @status) unless update?
  124. end
  125. def warm_payload_cache!
  126. Rails.cache.write("fan-out/#{@status.id}", rendered_status)
  127. end
  128. def anonymous_payload
  129. @anonymous_payload ||= Oj.dump(
  130. event: update? ? :'status.update' : :update,
  131. payload: rendered_status
  132. )
  133. end
  134. def rendered_status
  135. @rendered_status ||= InlineRenderer.render(@status, nil, :status)
  136. end
  137. def update?
  138. @options[:update]
  139. end
  140. def broadcastable?
  141. @status.public_visibility? && !@status.reblog? && !@account.silenced?
  142. end
  143. def subscribed_to_streaming_api?(account_id)
  144. redis.exists?("subscribed:timeline:#{account_id}") || redis.exists?("subscribed:timeline:#{account_id}:notifications")
  145. end
  146. end