|
|
|
@ -6,31 +6,54 @@ class FeedManager |
|
|
|
|
include Singleton |
|
|
|
|
include Redisable |
|
|
|
|
|
|
|
|
|
# Maximum number of items stored in a single feed |
|
|
|
|
MAX_ITEMS = 400 |
|
|
|
|
|
|
|
|
|
# Must be <= MAX_ITEMS or the tracking sets will grow forever |
|
|
|
|
# Number of items in the feed since last reblog of status |
|
|
|
|
# before the new reblog will be inserted. Must be <= MAX_ITEMS |
|
|
|
|
# or the tracking sets will grow forever |
|
|
|
|
REBLOG_FALLOFF = 40 |
|
|
|
|
|
|
|
|
|
# Execute block for every active account |
|
|
|
|
# @yield [Account] |
|
|
|
|
# @return [void] |
|
|
|
|
def with_active_accounts(&block) |
|
|
|
|
Account.joins(:user).where('users.current_sign_in_at > ?', User::ACTIVE_DURATION.ago).find_each(&block) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Redis key of a feed |
|
|
|
|
# @param [Symbol] type |
|
|
|
|
# @param [Integer] id |
|
|
|
|
# @param [Symbol] subtype |
|
|
|
|
# @return [String] |
|
|
|
|
def key(type, id, subtype = nil) |
|
|
|
|
return "feed:#{type}:#{id}" unless subtype |
|
|
|
|
|
|
|
|
|
"feed:#{type}:#{id}:#{subtype}" |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def filter?(timeline_type, status, receiver_id) |
|
|
|
|
if timeline_type == :home |
|
|
|
|
filter_from_home?(status, receiver_id, build_crutches(receiver_id, [status])) |
|
|
|
|
elsif timeline_type == :mentions |
|
|
|
|
filter_from_mentions?(status, receiver_id) |
|
|
|
|
# Check if the status should not be added to a feed |
|
|
|
|
# @param [Symbol] timeline_type |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Account|List] receiver |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def filter?(timeline_type, status, receiver) |
|
|
|
|
case timeline_type |
|
|
|
|
when :home |
|
|
|
|
filter_from_home?(status, receiver.id, build_crutches(receiver.id, [status])) |
|
|
|
|
when :list |
|
|
|
|
filter_from_list?(status, receiver) || filter_from_home?(status, receiver.account_id, build_crutches(receiver.account_id, [status])) |
|
|
|
|
when :mentions |
|
|
|
|
filter_from_mentions?(status, receiver.id) |
|
|
|
|
else |
|
|
|
|
false |
|
|
|
|
end |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Add a status to a home feed and send a streaming API update |
|
|
|
|
# @param [Account] account |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def push_to_home(account, status) |
|
|
|
|
return false unless add_to_feed(:home, account.id, status, account.user&.aggregates_reblogs?) |
|
|
|
|
|
|
|
|
@ -39,6 +62,10 @@ class FeedManager |
|
|
|
|
true |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Remove a status from a home feed and send a streaming API update |
|
|
|
|
# @param [Account] account |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def unpush_from_home(account, status) |
|
|
|
|
return false unless remove_from_feed(:home, account.id, status, account.user&.aggregates_reblogs?) |
|
|
|
|
|
|
|
|
@ -46,21 +73,22 @@ class FeedManager |
|
|
|
|
true |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Add a status to a list feed and send a streaming API update |
|
|
|
|
# @param [List] list |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def push_to_list(list, status) |
|
|
|
|
if status.reply? && status.in_reply_to_account_id != status.account_id |
|
|
|
|
should_filter = status.in_reply_to_account_id != list.account_id |
|
|
|
|
should_filter &&= !list.show_all_replies? |
|
|
|
|
should_filter &&= !(list.show_list_replies? && ListAccount.where(list_id: list.id, account_id: status.in_reply_to_account_id).exists?) |
|
|
|
|
return false if should_filter |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
return false unless add_to_feed(:list, list.id, status, list.account.user&.aggregates_reblogs?) |
|
|
|
|
return false if filter_from_list?(status, list) || !add_to_feed(:list, list.id, status, list.account.user&.aggregates_reblogs?) |
|
|
|
|
|
|
|
|
|
trim(:list, list.id) |
|
|
|
|
PushUpdateWorker.perform_async(list.account_id, status.id, "timeline:list:#{list.id}") if push_update_required?("timeline:list:#{list.id}") |
|
|
|
|
true |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Remove a status from a list feed and send a streaming API update |
|
|
|
|
# @param [List] list |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def unpush_from_list(list, status) |
|
|
|
|
return false unless remove_from_feed(:list, list.id, status, list.account.user&.aggregates_reblogs?) |
|
|
|
|
|
|
|
|
@ -68,36 +96,39 @@ class FeedManager |
|
|
|
|
true |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def trim(type, account_id) |
|
|
|
|
timeline_key = key(type, account_id) |
|
|
|
|
reblog_key = key(type, account_id, 'reblogs') |
|
|
|
|
# Fill a home feed with an account's statuses |
|
|
|
|
# @param [Account] from_account |
|
|
|
|
# @param [Account] into_account |
|
|
|
|
# @return [void] |
|
|
|
|
def merge_into_home(from_account, into_account) |
|
|
|
|
timeline_key = key(:home, into_account.id) |
|
|
|
|
aggregate = into_account.user&.aggregates_reblogs? |
|
|
|
|
query = from_account.statuses.where(visibility: [:public, :unlisted, :private]).includes(:preloadable_poll, reblog: :account).limit(FeedManager::MAX_ITEMS / 4) |
|
|
|
|
|
|
|
|
|
# Remove any items past the MAX_ITEMS'th entry in our feed |
|
|
|
|
redis.zremrangebyrank(timeline_key, 0, -(FeedManager::MAX_ITEMS + 1)) |
|
|
|
|
if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS / 4 |
|
|
|
|
oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true).first.last.to_i |
|
|
|
|
query = query.where('id > ?', oldest_home_score) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Get the score of the REBLOG_FALLOFF'th item in our feed, and stop |
|
|
|
|
# tracking anything after it for deduplication purposes. |
|
|
|
|
falloff_rank = FeedManager::REBLOG_FALLOFF |
|
|
|
|
falloff_range = redis.zrevrange(timeline_key, falloff_rank, falloff_rank, with_scores: true) |
|
|
|
|
falloff_score = falloff_range&.first&.last&.to_i |
|
|
|
|
statuses = query.to_a |
|
|
|
|
crutches = build_crutches(into_account.id, statuses) |
|
|
|
|
|
|
|
|
|
return if falloff_score.nil? |
|
|
|
|
statuses.each do |status| |
|
|
|
|
next if filter_from_home?(status, into_account.id, crutches) |
|
|
|
|
|
|
|
|
|
# Get any reblogs we might have to clean up after. |
|
|
|
|
redis.zrangebyscore(reblog_key, 0, falloff_score).each do |reblogged_id| |
|
|
|
|
# Remove it from the set of reblogs we're tracking *first* to avoid races. |
|
|
|
|
redis.zrem(reblog_key, reblogged_id) |
|
|
|
|
# Just drop any set we might have created to track additional reblogs. |
|
|
|
|
# This means that if this reblog is deleted, we won't automatically insert |
|
|
|
|
# another reblog, but also that any new reblog can be inserted into the |
|
|
|
|
# feed. |
|
|
|
|
redis.del(key(type, account_id, "reblogs:#{reblogged_id}")) |
|
|
|
|
add_to_feed(:home, into_account.id, status, aggregate) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
trim(:home, into_account.id) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def merge_into_timeline(from_account, into_account) |
|
|
|
|
timeline_key = key(:home, into_account.id) |
|
|
|
|
aggregate = into_account.user&.aggregates_reblogs? |
|
|
|
|
# Fill a list feed with an account's statuses |
|
|
|
|
# @param [Account] from_account |
|
|
|
|
# @param [List] list |
|
|
|
|
# @return [void] |
|
|
|
|
def merge_into_list(from_account, list) |
|
|
|
|
timeline_key = key(:list, list.id) |
|
|
|
|
aggregate = list.account.user&.aggregates_reblogs? |
|
|
|
|
query = from_account.statuses.where(visibility: [:public, :unlisted, :private]).includes(:preloadable_poll, reblog: :account).limit(FeedManager::MAX_ITEMS / 4) |
|
|
|
|
|
|
|
|
|
if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS / 4 |
|
|
|
@ -106,18 +137,22 @@ class FeedManager |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
statuses = query.to_a |
|
|
|
|
crutches = build_crutches(into_account.id, statuses) |
|
|
|
|
crutches = build_crutches(list.account_id, statuses) |
|
|
|
|
|
|
|
|
|
statuses.each do |status| |
|
|
|
|
next if filter_from_home?(status, into_account.id, crutches) |
|
|
|
|
next if filter_from_home?(status, list.account_id, crutches) || filter_from_list?(status, list) |
|
|
|
|
|
|
|
|
|
add_to_feed(:home, into_account.id, status, aggregate) |
|
|
|
|
add_to_feed(:list, list.id, status, aggregate) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
trim(:home, into_account.id) |
|
|
|
|
trim(:list, list.id) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def unmerge_from_timeline(from_account, into_account) |
|
|
|
|
# Remove an account's statuses from a home feed |
|
|
|
|
# @param [Account] from_account |
|
|
|
|
# @param [Account] into_account |
|
|
|
|
# @return [void] |
|
|
|
|
def unmerge_from_home(from_account, into_account) |
|
|
|
|
timeline_key = key(:home, into_account.id) |
|
|
|
|
oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 |
|
|
|
|
|
|
|
|
@ -126,14 +161,31 @@ class FeedManager |
|
|
|
|
end |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def clear_from_timeline(account, target_account) |
|
|
|
|
# Clear from timeline all statuses from or mentionning target_account |
|
|
|
|
# Remove an account's statuses from a list feed |
|
|
|
|
# @param [Account] from_account |
|
|
|
|
# @param [List] list |
|
|
|
|
# @return [void] |
|
|
|
|
def unmerge_from_list(from_account, list) |
|
|
|
|
timeline_key = key(:list, list.id) |
|
|
|
|
oldest_list_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 |
|
|
|
|
|
|
|
|
|
from_account.statuses.select('id, reblog_of_id').where('id > ?', oldest_list_score).reorder(nil).find_each do |status| |
|
|
|
|
remove_from_feed(:list, list.id, status, list.account.user&.aggregates_reblogs?) |
|
|
|
|
end |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Clear all statuses from or mentioning target_account from a home feed |
|
|
|
|
# @param [Account] account |
|
|
|
|
# @param [Account] target_account |
|
|
|
|
# @return [void] |
|
|
|
|
def clear_from_home(account, target_account) |
|
|
|
|
timeline_key = key(:home, account.id) |
|
|
|
|
timeline_status_ids = redis.zrange(timeline_key, 0, -1) |
|
|
|
|
statuses = Status.where(id: timeline_status_ids).select(:id, :reblog_of_id, :account_id).to_a |
|
|
|
|
reblogged_ids = Status.where(id: statuses.map(&:reblog_of_id).compact, account: target_account).pluck(:id) |
|
|
|
|
with_mentions_ids = Mention.active.where(status_id: statuses.flat_map { |s| [s.id, s.reblog_of_id] }.compact, account: target_account).pluck(:status_id) |
|
|
|
|
target_statuses = statuses.filter do |status| |
|
|
|
|
|
|
|
|
|
target_statuses = statuses.select do |status| |
|
|
|
|
status.account_id == target_account.id || reblogged_ids.include?(status.reblog_of_id) || with_mentions_ids.include?(status.id) || with_mentions_ids.include?(status.reblog_of_id) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
@ -142,7 +194,10 @@ class FeedManager |
|
|
|
|
end |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
def populate_feed(account) |
|
|
|
|
# Populate home feed of account from scratch |
|
|
|
|
# @param [Account] account |
|
|
|
|
# @return [void] |
|
|
|
|
def populate_home(account) |
|
|
|
|
limit = FeedManager::MAX_ITEMS / 2 |
|
|
|
|
aggregate = account.user&.aggregates_reblogs? |
|
|
|
|
timeline_key = key(:home, account.id) |
|
|
|
@ -177,15 +232,59 @@ class FeedManager |
|
|
|
|
|
|
|
|
|
private |
|
|
|
|
|
|
|
|
|
def push_update_required?(timeline_id) |
|
|
|
|
redis.exists?("subscribed:#{timeline_id}") |
|
|
|
|
# Trim a feed to maximum size by removing older items |
|
|
|
|
# @param [Symbol] type |
|
|
|
|
# @param [Integer] timeline_id |
|
|
|
|
# @return [void] |
|
|
|
|
def trim(type, timeline_id) |
|
|
|
|
timeline_key = key(type, timeline_id) |
|
|
|
|
reblog_key = key(type, timeline_id, 'reblogs') |
|
|
|
|
|
|
|
|
|
# Remove any items past the MAX_ITEMS'th entry in our feed |
|
|
|
|
redis.zremrangebyrank(timeline_key, 0, -(FeedManager::MAX_ITEMS + 1)) |
|
|
|
|
|
|
|
|
|
# Get the score of the REBLOG_FALLOFF'th item in our feed, and stop |
|
|
|
|
# tracking anything after it for deduplication purposes. |
|
|
|
|
falloff_rank = FeedManager::REBLOG_FALLOFF |
|
|
|
|
falloff_range = redis.zrevrange(timeline_key, falloff_rank, falloff_rank, with_scores: true) |
|
|
|
|
falloff_score = falloff_range&.first&.last&.to_i |
|
|
|
|
|
|
|
|
|
return if falloff_score.nil? |
|
|
|
|
|
|
|
|
|
# Get any reblogs we might have to clean up after. |
|
|
|
|
redis.zrangebyscore(reblog_key, 0, falloff_score).each do |reblogged_id| |
|
|
|
|
# Remove it from the set of reblogs we're tracking *first* to avoid races. |
|
|
|
|
redis.zrem(reblog_key, reblogged_id) |
|
|
|
|
# Just drop any set we might have created to track additional reblogs. |
|
|
|
|
# This means that if this reblog is deleted, we won't automatically insert |
|
|
|
|
# another reblog, but also that any new reblog can be inserted into the |
|
|
|
|
# feed. |
|
|
|
|
redis.del(key(type, timeline_id, "reblogs:#{reblogged_id}")) |
|
|
|
|
end |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if there is a streaming API client connected |
|
|
|
|
# for the given feed |
|
|
|
|
# @param [String] timeline_key |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def push_update_required?(timeline_key) |
|
|
|
|
redis.exists?("subscribed:#{timeline_key}") |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if the account is blocking or muting any of the given accounts |
|
|
|
|
# @param [Integer] receiver_id |
|
|
|
|
# @param [Array<Integer>] account_ids |
|
|
|
|
# @param [Symbol] context |
|
|
|
|
def blocks_or_mutes?(receiver_id, account_ids, context) |
|
|
|
|
Block.where(account_id: receiver_id, target_account_id: account_ids).any? || |
|
|
|
|
(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?) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if status should not be added to the home feed |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Integer] receiver_id |
|
|
|
|
# @param [Hash] crutches |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def filter_from_home?(status, receiver_id, crutches) |
|
|
|
|
return false if receiver_id == status.account_id |
|
|
|
|
return true if status.reply? && (status.in_reply_to_id.nil? || status.in_reply_to_account_id.nil?) |
|
|
|
@ -218,6 +317,11 @@ class FeedManager |
|
|
|
|
false |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if status should not be added to the mentions feed |
|
|
|
|
# @see NotifyService |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Integer] receiver_id |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def filter_from_mentions?(status, receiver_id) |
|
|
|
|
return true if receiver_id == status.account_id |
|
|
|
|
return true if phrase_filtered?(status, receiver_id, :notifications) |
|
|
|
@ -234,6 +338,27 @@ class FeedManager |
|
|
|
|
should_filter |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if status should not be added to the list feed |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [List] list |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def filter_from_list?(status, list) |
|
|
|
|
if status.reply? && status.in_reply_to_account_id != status.account_id |
|
|
|
|
should_filter = status.in_reply_to_account_id != list.account_id |
|
|
|
|
should_filter &&= !list.show_all_replies? |
|
|
|
|
should_filter &&= !(list.show_list_replies? && ListAccount.where(list_id: list.id, account_id: status.in_reply_to_account_id).exists?) |
|
|
|
|
|
|
|
|
|
return !!should_filter |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
false |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Check if the status hits a phrase filter |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Integer] receiver_id |
|
|
|
|
# @param [Symbol] context |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def phrase_filtered?(status, receiver_id, context) |
|
|
|
|
active_filters = Rails.cache.fetch("filters:#{receiver_id}") { CustomFilter.where(account_id: receiver_id).active_irreversible.to_a }.to_a |
|
|
|
|
|
|
|
|
@ -269,6 +394,11 @@ class FeedManager |
|
|
|
|
# added, and false if it was not added to the feed. Note that this is |
|
|
|
|
# an internal helper: callers must call trim or push updates if |
|
|
|
|
# either action is appropriate. |
|
|
|
|
# @param [Symbol] timeline_type |
|
|
|
|
# @param [Integer] account_id |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Boolean] aggregate_reblogs |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def add_to_feed(timeline_type, account_id, status, aggregate_reblogs = true) |
|
|
|
|
timeline_key = key(timeline_type, account_id) |
|
|
|
|
reblog_key = key(timeline_type, account_id, 'reblogs') |
|
|
|
@ -312,6 +442,11 @@ class FeedManager |
|
|
|
|
# with reblogs, and returning true if a status was removed. As with |
|
|
|
|
# `add_to_feed`, this does not trigger push updates, so callers must |
|
|
|
|
# do so if appropriate. |
|
|
|
|
# @param [Symbol] timeline_type |
|
|
|
|
# @param [Integer] account_id |
|
|
|
|
# @param [Status] status |
|
|
|
|
# @param [Boolean] aggregate_reblogs |
|
|
|
|
# @return [Boolean] |
|
|
|
|
def remove_from_feed(timeline_type, account_id, status, aggregate_reblogs = true) |
|
|
|
|
timeline_key = key(timeline_type, account_id) |
|
|
|
|
reblog_key = key(timeline_type, account_id, 'reblogs') |
|
|
|
@ -346,6 +481,11 @@ class FeedManager |
|
|
|
|
redis.zrem(timeline_key, status.id) |
|
|
|
|
end |
|
|
|
|
|
|
|
|
|
# Pre-fetch various objects and relationships for given statuses that |
|
|
|
|
# are going to be checked by the filtering methods |
|
|
|
|
# @param [Integer] receiver_id |
|
|
|
|
# @param [Array<Status>] statuses |
|
|
|
|
# @return [Hash] |
|
|
|
|
def build_crutches(receiver_id, statuses) |
|
|
|
|
crutches = {} |
|
|
|
|
|
|
|
|
|