feed_manager.rb 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306
  1. # frozen_string_literal: true
  2. require 'singleton'
  3. class FeedManager
  4. include Singleton
  5. include Redisable
  6. MAX_ITEMS = 400
  7. # Must be <= MAX_ITEMS or the tracking sets will grow forever
  8. REBLOG_FALLOFF = 40
  9. def key(type, id, subtype = nil)
  10. return "feed:#{type}:#{id}" unless subtype
  11. "feed:#{type}:#{id}:#{subtype}"
  12. end
  13. def filter?(timeline_type, status, receiver_id)
  14. if timeline_type == :home
  15. filter_from_home?(status, receiver_id)
  16. elsif timeline_type == :mentions
  17. filter_from_mentions?(status, receiver_id)
  18. else
  19. false
  20. end
  21. end
  22. def push_to_home(account, status)
  23. return false unless add_to_feed(:home, account.id, status, account.user&.aggregates_reblogs?)
  24. trim(:home, account.id)
  25. PushUpdateWorker.perform_async(account.id, status.id, "timeline:#{account.id}") if push_update_required?("timeline:#{account.id}")
  26. true
  27. end
  28. def unpush_from_home(account, status)
  29. return false unless remove_from_feed(:home, account.id, status)
  30. redis.publish("timeline:#{account.id}", Oj.dump(event: :delete, payload: status.id.to_s))
  31. true
  32. end
  33. def push_to_list(list, status)
  34. if status.reply? && status.in_reply_to_account_id != status.account_id
  35. should_filter = status.in_reply_to_account_id != list.account_id
  36. should_filter &&= !ListAccount.where(list_id: list.id, account_id: status.in_reply_to_account_id).exists?
  37. return false if should_filter
  38. end
  39. return false unless add_to_feed(:list, list.id, status, list.account.user&.aggregates_reblogs?)
  40. trim(:list, list.id)
  41. PushUpdateWorker.perform_async(list.account_id, status.id, "timeline:list:#{list.id}") if push_update_required?("timeline:list:#{list.id}")
  42. true
  43. end
  44. def unpush_from_list(list, status)
  45. return false unless remove_from_feed(:list, list.id, status)
  46. redis.publish("timeline:list:#{list.id}", Oj.dump(event: :delete, payload: status.id.to_s))
  47. true
  48. end
  49. def trim(type, account_id)
  50. timeline_key = key(type, account_id)
  51. reblog_key = key(type, account_id, 'reblogs')
  52. # Remove any items past the MAX_ITEMS'th entry in our feed
  53. redis.zremrangebyrank(timeline_key, '0', (-(FeedManager::MAX_ITEMS + 1)).to_s)
  54. # Get the score of the REBLOG_FALLOFF'th item in our feed, and stop
  55. # tracking anything after it for deduplication purposes.
  56. falloff_rank = FeedManager::REBLOG_FALLOFF - 1
  57. falloff_range = redis.zrevrange(timeline_key, falloff_rank, falloff_rank, with_scores: true)
  58. falloff_score = falloff_range&.first&.last&.to_i || 0
  59. # Get any reblogs we might have to clean up after.
  60. redis.zrangebyscore(reblog_key, 0, falloff_score).each do |reblogged_id|
  61. # Remove it from the set of reblogs we're tracking *first* to avoid races.
  62. redis.zrem(reblog_key, reblogged_id)
  63. # Just drop any set we might have created to track additional reblogs.
  64. # This means that if this reblog is deleted, we won't automatically insert
  65. # another reblog, but also that any new reblog can be inserted into the
  66. # feed.
  67. redis.del(key(type, account_id, "reblogs:#{reblogged_id}"))
  68. end
  69. end
  70. def merge_into_timeline(from_account, into_account)
  71. timeline_key = key(:home, into_account.id)
  72. query = from_account.statuses.limit(FeedManager::MAX_ITEMS / 4)
  73. if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS / 4
  74. oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0
  75. query = query.where('id > ?', oldest_home_score)
  76. end
  77. query.each do |status|
  78. next if status.direct_visibility? || status.limited_visibility? || filter?(:home, status, into_account)
  79. add_to_feed(:home, into_account.id, status, into_account.user&.aggregates_reblogs?)
  80. end
  81. trim(:home, into_account.id)
  82. end
  83. def unmerge_from_timeline(from_account, into_account)
  84. timeline_key = key(:home, into_account.id)
  85. oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0
  86. from_account.statuses.select('id, reblog_of_id').where('id > ?', oldest_home_score).reorder(nil).find_each do |status|
  87. remove_from_feed(:home, into_account.id, status)
  88. end
  89. end
  90. def clear_from_timeline(account, target_account)
  91. timeline_key = key(:home, account.id)
  92. timeline_status_ids = redis.zrange(timeline_key, 0, -1)
  93. target_statuses = Status.where(id: timeline_status_ids, account: target_account)
  94. target_statuses.each do |status|
  95. unpush_from_home(account, status)
  96. end
  97. end
  98. def populate_feed(account)
  99. added = 0
  100. limit = FeedManager::MAX_ITEMS / 2
  101. max_id = nil
  102. loop do
  103. statuses = Status.as_home_timeline(account)
  104. .paginate_by_max_id(limit, max_id)
  105. break if statuses.empty?
  106. statuses.each do |status|
  107. next if filter_from_home?(status, account)
  108. added += 1 if add_to_feed(:home, account.id, status, account.user&.aggregates_reblogs?)
  109. end
  110. break unless added.zero?
  111. max_id = statuses.last.id
  112. end
  113. end
  114. private
  115. def push_update_required?(timeline_id)
  116. redis.exists("subscribed:#{timeline_id}")
  117. end
  118. def blocks_or_mutes?(receiver_id, account_ids, context)
  119. Block.where(account_id: receiver_id, target_account_id: account_ids).any? ||
  120. (context == :home ? Mute.where(account_id: receiver_id, target_account_id: account_ids).any? : Mute.where(account_id: receiver_id, target_account_id: account_ids, hide_notifications: true).any?)
  121. end
  122. def filter_from_home?(status, receiver_id)
  123. return false if receiver_id == status.account_id
  124. return true if status.reply? && (status.in_reply_to_id.nil? || status.in_reply_to_account_id.nil?)
  125. return true if phrase_filtered?(status, receiver_id, :home)
  126. check_for_blocks = status.active_mentions.pluck(:account_id)
  127. check_for_blocks.concat([status.account_id])
  128. if status.reblog?
  129. check_for_blocks.concat([status.reblog.account_id])
  130. check_for_blocks.concat(status.reblog.active_mentions.pluck(:account_id))
  131. end
  132. return true if blocks_or_mutes?(receiver_id, check_for_blocks, :home)
  133. if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply
  134. should_filter = !Follow.where(account_id: receiver_id, target_account_id: status.in_reply_to_account_id).exists? # and I'm not following the person it's a reply to
  135. should_filter &&= receiver_id != status.in_reply_to_account_id # and it's not a reply to me
  136. should_filter &&= status.account_id != status.in_reply_to_account_id # and it's not a self-reply
  137. return should_filter
  138. elsif status.reblog? # Filter out a reblog
  139. should_filter = Follow.where(account_id: receiver_id, target_account_id: status.account_id, show_reblogs: false).exists? # if the reblogger's reblogs are suppressed
  140. should_filter ||= Block.where(account_id: status.reblog.account_id, target_account_id: receiver_id).exists? # or if the author of the reblogged status is blocking me
  141. should_filter ||= AccountDomainBlock.where(account_id: receiver_id, domain: status.reblog.account.domain).exists? # or the author's domain is blocked
  142. return should_filter
  143. end
  144. false
  145. end
  146. def filter_from_mentions?(status, receiver_id)
  147. return true if receiver_id == status.account_id
  148. return true if phrase_filtered?(status, receiver_id, :notifications)
  149. # This filter is called from NotifyService, but already after the sender of
  150. # the notification has been checked for mute/block. Therefore, it's not
  151. # necessary to check the author of the toot for mute/block again
  152. check_for_blocks = status.active_mentions.pluck(:account_id)
  153. check_for_blocks.concat([status.in_reply_to_account]) if status.reply? && !status.in_reply_to_account_id.nil?
  154. should_filter = blocks_or_mutes?(receiver_id, check_for_blocks, :mentions) # Filter if it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked (or muted)
  155. should_filter ||= (status.account.silenced? && !Follow.where(account_id: receiver_id, target_account_id: status.account_id).exists?) # of if the account is silenced and I'm not following them
  156. should_filter
  157. end
  158. def phrase_filtered?(status, receiver_id, context)
  159. active_filters = Rails.cache.fetch("filters:#{receiver_id}") { CustomFilter.where(account_id: receiver_id).active_irreversible.to_a }.to_a
  160. active_filters.select! { |filter| filter.context.include?(context.to_s) && !filter.expired? }
  161. active_filters.map! do |filter|
  162. if filter.whole_word
  163. sb = filter.phrase =~ /\A[[:word:]]/ ? '\b' : ''
  164. eb = filter.phrase =~ /[[:word:]]\z/ ? '\b' : ''
  165. /(?mix:#{sb}#{Regexp.escape(filter.phrase)}#{eb})/
  166. else
  167. /#{Regexp.escape(filter.phrase)}/i
  168. end
  169. end
  170. return false if active_filters.empty?
  171. combined_regex = active_filters.reduce { |memo, obj| Regexp.union(memo, obj) }
  172. status = status.reblog if status.reblog?
  173. !combined_regex.match(Formatter.instance.plaintext(status)).nil? ||
  174. (status.spoiler_text.present? && !combined_regex.match(status.spoiler_text).nil?)
  175. end
  176. # Adds a status to an account's feed, returning true if a status was
  177. # added, and false if it was not added to the feed. Note that this is
  178. # an internal helper: callers must call trim or push updates if
  179. # either action is appropriate.
  180. def add_to_feed(timeline_type, account_id, status, aggregate_reblogs = true)
  181. timeline_key = key(timeline_type, account_id)
  182. reblog_key = key(timeline_type, account_id, 'reblogs')
  183. if status.reblog? && (aggregate_reblogs.nil? || aggregate_reblogs)
  184. # If the original status or a reblog of it is within
  185. # REBLOG_FALLOFF statuses from the top, do not re-insert it into
  186. # the feed
  187. rank = redis.zrevrank(timeline_key, status.reblog_of_id)
  188. return false if !rank.nil? && rank < FeedManager::REBLOG_FALLOFF
  189. reblog_rank = redis.zrevrank(reblog_key, status.reblog_of_id)
  190. if reblog_rank.nil?
  191. # This is not something we've already seen reblogged, so we
  192. # can just add it to the feed (and note that we're
  193. # reblogging it).
  194. redis.zadd(timeline_key, status.id, status.id)
  195. redis.zadd(reblog_key, status.id, status.reblog_of_id)
  196. else
  197. # Another reblog of the same status was already in the
  198. # REBLOG_FALLOFF most recent statuses, so we note that this
  199. # is an "extra" reblog, by storing it in reblog_set_key.
  200. reblog_set_key = key(timeline_type, account_id, "reblogs:#{status.reblog_of_id}")
  201. redis.sadd(reblog_set_key, status.id)
  202. return false
  203. end
  204. else
  205. # A reblog may reach earlier than the original status because of the
  206. # delay of the worker deliverying the original status, the late addition
  207. # by merging timelines, and other reasons.
  208. # If such a reblog already exists, just do not re-insert it into the feed.
  209. rank = redis.zrevrank(reblog_key, status.id)
  210. return false unless rank.nil?
  211. redis.zadd(timeline_key, status.id, status.id)
  212. end
  213. true
  214. end
  215. # Removes an individual status from a feed, correctly handling cases
  216. # with reblogs, and returning true if a status was removed. As with
  217. # `add_to_feed`, this does not trigger push updates, so callers must
  218. # do so if appropriate.
  219. def remove_from_feed(timeline_type, account_id, status)
  220. timeline_key = key(timeline_type, account_id)
  221. if status.reblog?
  222. # 1. If the reblogging status is not in the feed, stop.
  223. status_rank = redis.zrevrank(timeline_key, status.id)
  224. return false if status_rank.nil?
  225. # 2. Remove reblog from set of this status's reblogs.
  226. reblog_set_key = key(timeline_type, account_id, "reblogs:#{status.reblog_of_id}")
  227. redis.srem(reblog_set_key, status.id)
  228. # 3. Re-insert another reblog or original into the feed if one
  229. # remains in the set. We could pick a random element, but this
  230. # set should generally be small, and it seems ideal to show the
  231. # oldest potential such reblog.
  232. other_reblog = redis.smembers(reblog_set_key).map(&:to_i).min
  233. redis.zadd(timeline_key, other_reblog, other_reblog) if other_reblog
  234. # 4. Remove the reblogging status from the feed (as normal)
  235. # (outside conditional)
  236. else
  237. # If the original is getting deleted, no use for reblog references
  238. redis.del(key(timeline_type, account_id, "reblogs:#{status.id}"))
  239. end
  240. redis.zrem(timeline_key, status.id)
  241. end
  242. end