discourse/app/models/topic_tracking_state.rb
Alan Guo Xiang Tan 2f5e66b6f8
PERF: Optimise TopicTrackingState.report query to speed up query (#22871)
In the query generated by `TopicTrackingState.report`, there are two
subqueies being executed. The first subquery fetches all the topics
that are new for a given user while the second subquery fetches all the topics with
unread posts for a given user. For the second subquery, there is a
filter `topics.updated_at >= user_stats.first_unread_at` which is used
as a performance optimisation to reduce the number of rows that PG has
to scan through the `topics` table.

However, we started to notice in production that the PG planner doesn't
always execute the filter first to reduce the number of rows that it has
to scan through. Running the following query in one of our production
instance,

```
EXPLAIN ANALYZE
SELECT
           DISTINCT topics.id as topic_id,
           u.id as user_id,
           topics.created_at,
           topics.updated_at,
           topics.highest_staff_post_number AS highest_post_number,
           last_read_post_number,
           c.id as category_id,
           c.topic_id AS category_topic_id,
           tu.notification_level,
           us.first_unread_at,
           GREATEST(
              CASE
              WHEN COALESCE(uo.new_topic_duration_minutes, 2880) = -1 THEN u.created_at
              WHEN COALESCE(uo.new_topic_duration_minutes, 2880) = -2 THEN COALESCE(
                u.previous_visit_at,u.created_at
              )
              ELSE ('2023-07-31 03:29:45.737630'::timestamp - INTERVAL '1 MINUTE' * COALESCE(uo.new_topic_duration_minutes, 2880))
              END, u.created_at, '2023-07-25 15:06:44'
           ) AS treat_as_new_topic_start_date
FROM topics
JOIN users u on u.id = 13455
JOIN user_stats AS us ON us.user_id = u.id
JOIN user_options AS uo ON uo.user_id = u.id
JOIN categories c ON c.id = topics.category_id
LEFT JOIN topic_users tu ON tu.topic_id = topics.id AND tu.user_id = u.id

WHERE u.id = 13455 AND
       topics.updated_at >= us.first_unread_at AND
      topics.archetype <> 'private_message' AND
      (("topics"."deleted_at" IS NULL AND (tu.last_read_post_number < topics.highest_staff_post_number) AND (COALESCE(tu.notification_level, 1) >= 2)) OR (1=0)) AND

      NOT (
  COALESCE((select array_agg(tag_id) from topic_tags where topic_tags.topic_id = topics.id), ARRAY[]::int[]) && ARRAY[451,452,453]
) AND

      topics.deleted_at IS NULL AND

      NOT (
        last_read_post_number IS NULL AND
        (
          topics.category_id IN (SELECT "categories"."id" FROM "categories" LEFT JOIN categories categories2 ON categories2.id = categories.parent_category_id LEFT JOIN category_users ON category_users.category_id = categories.id AND category_users.user_id = 13455 LEFT JOIN category_users category_users2 ON category_users2.category_id = categories2.id AND category_users2.user_id = 13455 WHERE ((category_users.id IS NULL AND COALESCE(category_users2.notification_level, 1) = 0) OR COALESCE(category_users.notification_level, 1) = 0))
          AND tu.notification_level <= 1
        )
      )
```

we get the following

```
                                                                                                                                                                                                                                                                                                                          QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Unique  (cost=201606.06..201608.15 rows=76 width=60) (actual time=91.279..91.294 rows=14 loops=1)
   ->  Sort  (cost=201606.06..201606.25 rows=76 width=60) (actual time=91.278..91.284 rows=14 loops=1)
         Sort Key: topics.id, topics.created_at, topics.updated_at, topics.highest_staff_post_number, tu.last_read_post_number, c.id, c.topic_id, tu.notification_level, us.first_unread_at, (GREATEST(CASE WHEN (COALESCE(uo.new_topic_duration_minutes, 2880) = '-1'::integer) THEN u.created_at WHEN (COALESCE(uo.new_topic_duration_minutes, 2880) = '-2'::integer) THEN COALESCE(u.previous_visit_at, u.created_at) ELSE ('2023-07-31 03:29:45.73763'::timestamp without time zone - ('00:01:00'::interval * (COALESCE(uo.new_topic_duration_minutes, 2880))::double precision)) END, u.created_at, '2023-07-25 15:06:44'::timestamp without time zone))
         Sort Method: quicksort  Memory: 26kB
         ->  Hash Join  (cost=97519.51..201603.69 rows=76 width=60) (actual time=87.662..91.268 rows=14 loops=1)
               Hash Cond: (topics.id = tu.topic_id)
               Join Filter: ((tu.last_read_post_number < topics.highest_staff_post_number) AND ((tu.last_read_post_number IS NOT NULL) OR (NOT (hashed SubPlan 2)) OR (tu.notification_level > 1)))
               Rows Removed by Join Filter: 10
               ->  Nested Loop  (cost=1.54..104075.36 rows=3511 width=68) (actual time=0.055..3.609 rows=548 loops=1)
                     ->  Nested Loop  (cost=1.13..25.20 rows=1 width=32) (actual time=0.027..0.033 rows=1 loops=1)
                           ->  Nested Loop  (cost=0.71..16.76 rows=1 width=28) (actual time=0.020..0.023 rows=1 loops=1)
                                 ->  Index Scan using users_pkey on users u  (cost=0.42..8.44 rows=1 width=20) (actual time=0.010..0.012 rows=1 loops=1)
                                       Index Cond: (id = 13455)
                                 ->  Index Scan using user_stats_pkey on user_stats us  (cost=0.29..8.31 rows=1 width=12) (actual time=0.008..0.010 rows=1 loops=1)
                                       Index Cond: (user_id = 13455)
                           ->  Index Scan using index_user_options_on_user_id_and_default_calendar on user_options uo  (cost=0.42..8.44 rows=1 width=8) (actual time=0.007..0.008 rows=1 loops=1)
                                 Index Cond: (user_id = 13455)
                     ->  Nested Loop  (cost=0.41..104015.12 rows=3504 width=36) (actual time=0.026..3.503 rows=548 loops=1)
                           ->  Seq Scan on categories c  (cost=0.00..13.73 rows=73 width=8) (actual time=0.003..0.039 rows=73 loops=1)
                           ->  Index Only Scan using index_topics_on_updated_at_public on topics  (cost=0.41..1424.20 rows=48 width=28) (actual time=0.012..0.046 rows=8 loops=73)
                                 Index Cond: ((updated_at >= us.first_unread_at) AND (category_id = c.id))
                                 Filter: (NOT (COALESCE((SubPlan 1), '{}'::integer[]) && '{451,452,453}'::integer[]))
                                 Heap Fetches: 553
                                 SubPlan 1
                                   ->  Aggregate  (cost=4.31..4.32 rows=1 width=32) (actual time=0.002..0.002 rows=1 loops=548)
                                         ->  Index Only Scan using index_topic_tags_on_topic_id_and_tag_id on topic_tags  (cost=0.29..4.31 rows=1 width=4) (actual time=0.002..0.002 rows=1 loops=548)
                                               Index Cond: (topic_id = topics.id)
                                               Heap Fetches: 178
               ->  Hash  (cost=97222.14..97222.14 rows=19914 width=16) (actual time=87.545..87.546 rows=42884 loops=1)
                     Buckets: 65536 (originally 32768)  Batches: 1 (originally 1)  Memory Usage: 2387kB
                     ->  Bitmap Heap Scan on topic_users tu  (cost=1217.47..97222.14 rows=19914 width=16) (actual time=14.419..78.286 rows=42884 loops=1)
                           Recheck Cond: (user_id = 13455)
                           Filter: (COALESCE(notification_level, 1) >= 2)
                           Rows Removed by Filter: 15839
                           Heap Blocks: exact=45285
                           ->  Bitmap Index Scan on index_topic_users_on_user_id_and_topic_id  (cost=0.00..1212.49 rows=59741 width=0) (actual time=6.448..6.448 rows=58723 loops=1)
                                 Index Cond: (user_id = 13455)
               SubPlan 2
                 ->  Nested Loop Left Join  (cost=0.74..46.90 rows=1 width=4) (never executed)
                       Join Filter: (category_users2.category_id = categories2.id)
                       Filter: (((category_users.id IS NULL) AND (COALESCE(category_users2.notification_level, 1) = 0)) OR (COALESCE(category_users.notification_level, 1) = 0))
                       ->  Nested Loop Left Join  (cost=0.45..32.31 rows=73 width=16) (never executed)
                             Join Filter: (category_users.category_id = categories.id)
                             ->  Nested Loop Left Join  (cost=0.15..18.45 rows=73 width=8) (never executed)
                                   ->  Seq Scan on categories  (cost=0.00..13.73 rows=73 width=8) (never executed)
                                   ->  Memoize  (cost=0.15..0.28 rows=1 width=4) (never executed)
                                         Cache Key: categories.parent_category_id
                                         Cache Mode: logical
                                         ->  Index Only Scan using categories_pkey on categories categories2  (cost=0.14..0.27 rows=1 width=4) (never executed)
                                               Index Cond: (id = categories.parent_category_id)
                                               Heap Fetches: 0
                             ->  Materialize  (cost=0.29..11.69 rows=2 width=12) (never executed)
                                   ->  Index Scan using idx_category_users_user_id_category_id on category_users  (cost=0.29..11.68 rows=2 width=12) (never executed)
                                         Index Cond: (user_id = 13455)
                       ->  Materialize  (cost=0.29..11.69 rows=2 width=8) (never executed)
                             ->  Index Scan using idx_category_users_user_id_category_id on category_users category_users2  (cost=0.29..11.68 rows=2 width=8) (never executed)
                                   Index Cond: (user_id = 13455)
 Planning Time: 1.740 ms
 Execution Time: 91.414 ms
(59 rows)
```

From the execution plan, we can see the most of the time is spent
joining about 42888 rows in the `topics` table to the `topic_users` table.
However, we know that we only have to scan through a
subset of the `topics` table because the user's last unread at is '2023-07-20 11:33:05'.
If we filter the `topics` table with `topics.updated_at >= '2023-07-20 11:33:05'`, this would only
return about 1500 rows.

From our testing in production, the PG planner is able to execute a
better query plan when we avoid the unnecessary joins on `user_stats` just to be
able to get the user's `UserStat#first_unread_at`. Instead, we can just
pass the value of `UserStat#first_unread_at` directly as a query
parameter.

```
EXPLAIN ANALYZE
SELECT
           DISTINCT topics.id as topic_id,
           u.id as user_id,
           topics.created_at,
           topics.updated_at,
           topics.highest_staff_post_number AS highest_post_number,
           last_read_post_number,
           c.id as category_id,
           c.topic_id AS category_topic_id,
           tu.notification_level,
           GREATEST(
              CASE
              WHEN COALESCE(uo.new_topic_duration_minutes, 2880) = -1 THEN u.created_at
              WHEN COALESCE(uo.new_topic_duration_minutes, 2880) = -2 THEN COALESCE(
                u.previous_visit_at,u.created_at
              )
              ELSE ('2023-07-31 03:29:45.737630'::timestamp - INTERVAL '1 MINUTE' * COALESCE(uo.new_topic_duration_minutes, 2880))
              END, u.created_at, '2023-07-25 15:06:44'
           ) AS treat_as_new_topic_start_date
FROM topics
JOIN users u on u.id = 13455
JOIN user_options AS uo ON uo.user_id = u.id
JOIN categories c ON c.id = topics.category_id
LEFT JOIN topic_users tu ON tu.topic_id = topics.id AND tu.user_id = u.id

WHERE u.id = 13455 AND
       topics.updated_at >= '2023-07-20 11:33:05' AND
      topics.archetype <> 'private_message' AND
      (("topics"."deleted_at" IS NULL AND (tu.last_read_post_number < topics.highest_staff_post_number) AND (COALESCE(tu.notification_level, 1) >= 2)) OR (1=0)) AND

      NOT (
  COALESCE((select array_agg(tag_id) from topic_tags where topic_tags.topic_id = topics.id), ARRAY[]::int[]) && ARRAY[451,452,453]
) AND

      topics.deleted_at IS NULL AND

      NOT (
        last_read_post_number IS NULL AND
        (
          topics.category_id IN (SELECT "categories"."id" FROM "categories" LEFT JOIN categories categories2 ON categories2.id = categories.parent_category_id LEFT JOIN category_users ON category_users.category_id = categories.id AND category_users.user_id = 13455 LEFT JOIN category_users category_users2 ON category_users2.category_id = categories2.id AND category_users2.user_id = 13455 WHERE ((category_users.id IS NULL AND COALESCE(category_users2.notification_level, 1) = 0) OR COALESCE(category_users.notification_level, 1) = 0))
          AND tu.notification_level <= 1
        )
      );
```

Note how the filter is now `topics.updated_at >= '2023-07-20 11:33:05'`
instead of `topics.updated_at >= us.first_unread_at`. The modified query
above generates the following execution plan.

```
                                                                                                                                                                                                                                                                                                                QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Unique  (cost=5189.86..5189.88 rows=1 width=52) (actual time=4.991..5.002 rows=14 loops=1)
   ->  Sort  (cost=5189.86..5189.86 rows=1 width=52) (actual time=4.990..4.994 rows=14 loops=1)
         Sort Key: topics.id, topics.created_at, topics.updated_at, topics.highest_staff_post_number, tu.last_read_post_number, c.id, c.topic_id, tu.notification_level, (GREATEST(CASE WHEN (COALESCE(uo.new_topic_duration_minutes, 2880) = '-1'::integer) THEN u.created_at WHEN (COALESCE(uo.new_topic_duration_minutes, 2880) = '-2'::integer) THEN COALESCE(u.previous_visit_at, u.created_at) ELSE ('2023-07-31 03:29:45.73763'::timestamp without time zone - ('00:01:00'::interval * (COALESCE(uo.new_topic_duration_minutes, 2880))::double precision)) END, u.created_at, '2023-07-25 15:06:44'::timestamp without time zone))
         Sort Method: quicksort  Memory: 26kB
         ->  Nested Loop  (cost=52.11..5189.85 rows=1 width=52) (actual time=0.093..4.974 rows=14 loops=1)
               ->  Nested Loop  (cost=51.70..5181.39 rows=1 width=60) (actual time=0.084..4.931 rows=14 loops=1)
                     ->  Nested Loop  (cost=51.28..5172.94 rows=1 width=44) (actual time=0.076..4.887 rows=14 loops=1)
                           ->  Nested Loop  (cost=0.41..1698.46 rows=59 width=36) (actual time=0.029..3.537 rows=548 loops=1)
                                 ->  Seq Scan on categories c  (cost=0.00..13.73 rows=73 width=8) (actual time=0.005..0.039 rows=73 loops=1)
                                 ->  Index Only Scan using index_topics_on_updated_at_public on topics  (cost=0.41..23.07 rows=1 width=28) (actual time=0.012..0.047 rows=8 loops=73)
                                       Index Cond: ((updated_at >= '2023-07-20 11:33:05'::timestamp without time zone) AND (category_id = c.id))
                                       Filter: (NOT (COALESCE((SubPlan 1), '{}'::integer[]) && '{451,452,453}'::integer[]))
                                       Heap Fetches: 552
                                       SubPlan 1
                                         ->  Aggregate  (cost=4.31..4.32 rows=1 width=32) (actual time=0.002..0.002 rows=1 loops=548)
                                               ->  Index Only Scan using index_topic_tags_on_topic_id_and_tag_id on topic_tags  (cost=0.29..4.31 rows=1 width=4) (actual time=0.002..0.002 rows=1 loops=548)
                                                     Index Cond: (topic_id = topics.id)
                                                     Heap Fetches: 178
                           ->  Index Scan using index_topic_users_on_user_id_and_topic_id on topic_users tu  (cost=50.86..58.88 rows=1 width=16) (actual time=0.002..0.002 rows=0 loops=548)
                                 Index Cond: ((user_id = 13455) AND (topic_id = topics.id))
                                 Filter: ((COALESCE(notification_level, 1) >= 2) AND (last_read_post_number < topics.highest_staff_post_number) AND ((last_read_post_number IS NOT NULL) OR (NOT (hashed SubPlan 2)) OR (notification_level > 1)))
                                 Rows Removed by Filter: 0
                                 SubPlan 2
                                   ->  Nested Loop Left Join  (cost=0.74..50.43 rows=1 width=4) (never executed)
                                         Join Filter: (category_users2.category_id = categories2.id)
                                         Filter: (((category_users.id IS NULL) AND (COALESCE(category_users2.notification_level, 1) = 0)) OR (COALESCE(category_users.notification_level, 1) = 0))
                                         ->  Nested Loop Left Join  (cost=0.45..35.84 rows=73 width=16) (never executed)
                                               Join Filter: (category_users.category_id = categories.id)
                                               ->  Nested Loop Left Join  (cost=0.15..21.97 rows=73 width=8) (never executed)
                                                     ->  Seq Scan on categories  (cost=0.00..13.73 rows=73 width=8) (never executed)
                                                     ->  Memoize  (cost=0.15..0.61 rows=1 width=4) (never executed)
                                                           Cache Key: categories.parent_category_id
                                                           Cache Mode: logical
                                                           ->  Index Only Scan using categories_pkey on categories categories2  (cost=0.14..0.60 rows=1 width=4) (never executed)
                                                                 Index Cond: (id = categories.parent_category_id)
                                                                 Heap Fetches: 0
                                               ->  Materialize  (cost=0.29..11.69 rows=2 width=12) (never executed)
                                                     ->  Index Scan using idx_category_users_user_id_category_id on category_users  (cost=0.29..11.68 rows=2 width=12) (never executed)
                                                           Index Cond: (user_id = 13455)
                                         ->  Materialize  (cost=0.29..11.69 rows=2 width=8) (never executed)
                                               ->  Index Scan using idx_category_users_user_id_category_id on category_users category_users2  (cost=0.29..11.68 rows=2 width=8) (never executed)
                                                     Index Cond: (user_id = 13455)
                     ->  Index Scan using users_pkey on users u  (cost=0.42..8.44 rows=1 width=20) (actual time=0.003..0.003 rows=1 loops=14)
                           Index Cond: (id = 13455)
               ->  Index Scan using index_user_options_on_user_id_and_default_calendar on user_options uo  (cost=0.42..8.44 rows=1 width=8) (actual time=0.002..0.002 rows=1 loops=14)
                     Index Cond: (user_id = 13455)
 Planning Time: 1.281 ms
 Execution Time: 5.092 ms
(48 rows)
```

With the new query, PG first does an index scan using the `index_topics_on_updated_at_public` index to filter away most of the topics making the subsequent joins much cheaper. Total query time has been reduced from ~90ms to ~5ms.

This optimisation will mostly affect users with very few/recent unread topics since a large `UserStat#firsts_unread_at` value will still mean scanning through a large portion of the `topics` table.
2023-07-31 12:21:41 +08:00

599 lines
18 KiB
Ruby

# frozen_string_literal: true
# This class is used to mirror unread and new status back to end users
# in JavaScript there is a mirror class that is kept in-sync using MessageBus
# the allows end users to always know which topics have unread posts in them
# and which topics are new. This is used in various places in the UI, such as
# counters, indicators, and messages at the top of topic lists, so the user
# knows there is something worth reading at a glance.
#
# The TopicTrackingState.report data is preloaded in ApplicationController
# for the current user under the topicTrackingStates key, and the existing
# state is loaded into memory on page load. From there the MessageBus is
# used to keep topic state up to date, as well as syncing with topics from
# corresponding lists fetched from the server (e.g. the /new, /latest,
# /unread topic lists).
#
# See discourse/app/models/topic-tracking-state.js
class TopicTrackingState
include ActiveModel::SerializerSupport
include TopicTrackingStatePublishable
UNREAD_MESSAGE_TYPE = "unread"
LATEST_MESSAGE_TYPE = "latest"
MUTED_MESSAGE_TYPE = "muted"
UNMUTED_MESSAGE_TYPE = "unmuted"
NEW_TOPIC_MESSAGE_TYPE = "new_topic"
RECOVER_MESSAGE_TYPE = "recover"
DELETE_MESSAGE_TYPE = "delete"
DESTROY_MESSAGE_TYPE = "destroy"
READ_MESSAGE_TYPE = "read"
DISMISS_NEW_MESSAGE_TYPE = "dismiss_new"
DISMISS_NEW_POSTS_MESSAGE_TYPE = "dismiss_new_posts"
MAX_TOPICS = 5000
NEW_MESSAGE_BUS_CHANNEL = "/new"
LATEST_MESSAGE_BUS_CHANNEL = "/latest"
UNREAD_MESSAGE_BUS_CHANNEL = "/unread"
RECOVER_MESSAGE_BUS_CHANNEL = "/recover"
DELETE_MESSAGE_BUS_CHANNEL = "/delete"
DESTROY_MESSAGE_BUS_CHANNEL = "/destroy"
def self.publish_new(topic)
return unless topic.regular?
tag_ids, tags = nil
tag_ids, tags = topic.tags.pluck(:id, :name).transpose if SiteSetting.tagging_enabled
payload = {
last_read_post_number: nil,
highest_post_number: 1,
created_at: topic.created_at,
category_id: topic.category_id,
archetype: topic.archetype,
created_in_new_period: true,
}
if tags
payload[:tags] = tags
payload[:topic_tag_ids] = tag_ids
end
message = { topic_id: topic.id, message_type: NEW_TOPIC_MESSAGE_TYPE, payload: payload }
group_ids = secure_category_group_ids(topic)
MessageBus.publish(NEW_MESSAGE_BUS_CHANNEL, message.as_json, group_ids: group_ids)
publish_read(topic.id, 1, topic.user)
end
def self.publish_latest(topic, whisper = false)
return unless topic.regular?
tag_ids, tags = nil
tag_ids, tags = topic.tags.pluck(:id, :name).transpose if SiteSetting.tagging_enabled
message = {
topic_id: topic.id,
message_type: LATEST_MESSAGE_TYPE,
payload: {
bumped_at: topic.bumped_at,
category_id: topic.category_id,
archetype: topic.archetype,
},
}
if tags
message[:payload][:tags] = tags
message[:payload][:topic_tag_ids] = tag_ids
end
group_ids =
if whisper
[Group::AUTO_GROUPS[:staff], *SiteSetting.whispers_allowed_group_ids]
else
secure_category_group_ids(topic)
end
MessageBus.publish(LATEST_MESSAGE_BUS_CHANNEL, message.as_json, group_ids: group_ids)
end
def self.unread_channel_key(user_id)
"/unread/#{user_id}"
end
def self.publish_muted(topic)
return unless topic.regular?
user_ids =
topic
.topic_users
.where(notification_level: NotificationLevels.all[:muted])
.joins(:user)
.where("users.last_seen_at > ?", 7.days.ago)
.order("users.last_seen_at DESC")
.limit(100)
.pluck(:user_id)
return if user_ids.blank?
message = { topic_id: topic.id, message_type: MUTED_MESSAGE_TYPE }
MessageBus.publish(LATEST_MESSAGE_BUS_CHANNEL, message.as_json, user_ids: user_ids)
end
def self.publish_unmuted(topic)
return unless topic.regular?
user_ids =
User
.watching_topic(topic)
.where("users.last_seen_at > ?", 7.days.ago)
.order("users.last_seen_at DESC")
.limit(100)
.pluck(:id)
return if user_ids.blank?
message = { topic_id: topic.id, message_type: UNMUTED_MESSAGE_TYPE }
MessageBus.publish(LATEST_MESSAGE_BUS_CHANNEL, message.as_json, user_ids: user_ids)
end
def self.publish_unread(post)
return unless post.topic.regular?
# TODO at high scale we are going to have to defer this,
# perhaps cut down to users that are around in the last 7 days as well
tags = nil
tag_ids = nil
tag_ids, tags = post.topic.tags.pluck(:id, :name).transpose if include_tags_in_report?
# We don't need to publish unread to the person who just made the post,
# this is why they are excluded from the initial scope.
scope =
TopicUser.tracking(post.topic_id).includes(user: :user_stat).where.not(user_id: post.user_id)
group_ids =
if post.post_type == Post.types[:whisper]
[Group::AUTO_GROUPS[:staff], *SiteSetting.whispers_allowed_group_ids]
else
post.topic.category && post.topic.category.secure_group_ids
end
if group_ids.present?
scope =
scope.joins("INNER JOIN group_users gu ON gu.user_id = topic_users.user_id").where(
"gu.group_id IN (?)",
group_ids,
)
end
user_ids = scope.pluck(:user_id)
return if user_ids.empty?
payload = {
highest_post_number: post.post_number,
updated_at: post.topic.updated_at,
created_at: post.created_at,
category_id: post.topic.category_id,
archetype: post.topic.archetype,
}
if tags
payload[:tags] = tags
payload[:topic_tag_ids] = tag_ids
end
message = { topic_id: post.topic_id, message_type: UNREAD_MESSAGE_TYPE, payload: payload }
MessageBus.publish(UNREAD_MESSAGE_BUS_CHANNEL, message.as_json, user_ids: user_ids)
end
def self.publish_recover(topic)
return unless topic.regular?
group_ids = secure_category_group_ids(topic)
message = { topic_id: topic.id, message_type: RECOVER_MESSAGE_TYPE }
MessageBus.publish(RECOVER_MESSAGE_BUS_CHANNEL, message.as_json, group_ids: group_ids)
end
def self.publish_delete(topic)
return unless topic.regular?
group_ids = secure_category_group_ids(topic)
message = { topic_id: topic.id, message_type: DELETE_MESSAGE_TYPE }
MessageBus.publish("/delete", message.as_json, group_ids: group_ids)
end
def self.publish_destroy(topic)
return unless topic.regular?
group_ids = secure_category_group_ids(topic)
message = { topic_id: topic.id, message_type: DESTROY_MESSAGE_TYPE }
MessageBus.publish(DESTROY_MESSAGE_BUS_CHANNEL, message.as_json, group_ids: group_ids)
end
def self.publish_read(topic_id, last_read_post_number, user, notification_level = nil)
self.publish_read_message(
message_type: READ_MESSAGE_TYPE,
channel_name: self.unread_channel_key(user.id),
topic_id: topic_id,
user: user,
last_read_post_number: last_read_post_number,
notification_level: notification_level,
)
end
def self.publish_dismiss_new(user_id, topic_ids: [])
message = { message_type: DISMISS_NEW_MESSAGE_TYPE, payload: { topic_ids: topic_ids } }
MessageBus.publish(self.unread_channel_key(user_id), message.as_json, user_ids: [user_id])
end
def self.publish_dismiss_new_posts(user_id, topic_ids: [])
message = { message_type: DISMISS_NEW_POSTS_MESSAGE_TYPE, payload: { topic_ids: topic_ids } }
MessageBus.publish(self.unread_channel_key(user_id), message.as_json, user_ids: [user_id])
end
def self.new_filter_sql
TopicQuery
.new_filter(Topic, treat_as_new_topic_clause_sql: treat_as_new_topic_clause)
.where_clause
.ast
.to_sql + " AND topics.created_at > :min_new_topic_date" +
" AND dismissed_topic_users.id IS NULL"
end
def self.unread_filter_sql(whisperer: false)
TopicQuery.unread_filter(Topic, whisperer: whisperer).where_clause.ast.to_sql
end
def self.treat_as_new_topic_clause
User
.where(
"GREATEST(CASE
WHEN COALESCE(uo.new_topic_duration_minutes, :default_duration) = :always THEN u.created_at
WHEN COALESCE(uo.new_topic_duration_minutes, :default_duration) = :last_visit THEN COALESCE(u.previous_visit_at,u.created_at)
ELSE (:now::timestamp - INTERVAL '1 MINUTE' * COALESCE(uo.new_topic_duration_minutes, :default_duration))
END, u.created_at, :min_date)",
treat_as_new_topic_params,
)
.where_clause
.ast
.to_sql
end
def self.treat_as_new_topic_params
{
now: DateTime.now,
last_visit: User::NewTopicDuration::LAST_VISIT,
always: User::NewTopicDuration::ALWAYS,
default_duration: SiteSetting.default_other_new_topic_duration_minutes,
min_date: Time.at(SiteSetting.min_new_topics_time).to_datetime,
}
end
def self.include_tags_in_report?
SiteSetting.tagging_enabled && (@include_tags_in_report || !SiteSetting.legacy_navigation_menu?)
end
def self.include_tags_in_report=(v)
@include_tags_in_report = v
end
# Sam: this is a hairy report, in particular I need custom joins and fancy conditions
# Dropping to sql_builder so I can make sense of it.
#
# Keep in mind, we need to be able to filter on a GROUP of users, and zero in on topic
# all our existing scope work does not do this
#
# This code needs to be VERY efficient as it is triggered via the message bus and may steal
# cycles from usual requests
def self.report(user, topic_id = nil)
tag_ids = muted_tag_ids(user)
sql = new_and_unread_sql(topic_id, user, tag_ids)
sql = tags_included_wrapped_sql(sql)
report =
DB.query(
sql + "\n\n LIMIT :max_topics",
{
user_id: user.id,
topic_id: topic_id,
min_new_topic_date: Time.at(SiteSetting.min_new_topics_time).to_datetime,
max_topics: TopicTrackingState::MAX_TOPICS,
user_first_unread_at: user.user_stat.first_unread_at,
}.merge(treat_as_new_topic_params),
)
report
end
def self.new_and_unread_sql(topic_id, user, tag_ids)
sql =
report_raw_sql(
topic_id: topic_id,
skip_unread: true,
skip_order: true,
staff: user.staff?,
admin: user.admin?,
whisperer: user.whisperer?,
user: user,
muted_tag_ids: tag_ids,
)
sql << "\nUNION ALL\n\n"
sql << report_raw_sql(
topic_id: topic_id,
skip_new: true,
skip_order: true,
staff: user.staff?,
filter_old_unread: true,
admin: user.admin?,
whisperer: user.whisperer?,
user: user,
muted_tag_ids: tag_ids,
)
end
def self.tags_included_wrapped_sql(sql)
return <<~SQL if SiteSetting.tagging_enabled && TopicTrackingState.include_tags_in_report?
WITH tags_included_cte AS (
#{sql}
)
SELECT *, (
SELECT ARRAY_AGG(name) from topic_tags
JOIN tags on tags.id = topic_tags.tag_id
WHERE topic_id = tags_included_cte.topic_id
) tags
FROM tags_included_cte
SQL
sql
end
def self.muted_tag_ids(user)
TagUser.lookup(user, :muted).pluck(:tag_id)
end
def self.report_raw_sql(
user:,
muted_tag_ids:,
topic_id: nil,
filter_old_unread: false,
skip_new: false,
skip_unread: false,
skip_order: false,
staff: false,
admin: false,
whisperer: false,
select: nil,
custom_state_filter: nil,
additional_join_sql: nil
)
unread =
if skip_unread
"1=0"
else
unread_filter_sql(whisperer: whisperer)
end
filter_old_unread_sql =
if filter_old_unread
" topics.updated_at >= :user_first_unread_at AND "
else
""
end
new =
if skip_new
"1=0"
else
new_filter_sql
end
category_topic_id_column_select =
if SiteSetting.show_category_definitions_in_topic_lists
""
else
"c.topic_id AS category_topic_id,"
end
select_sql =
select ||
"
DISTINCT topics.id as topic_id,
u.id as user_id,
topics.created_at,
topics.updated_at,
#{highest_post_number_column_select(whisperer)},
last_read_post_number,
c.id as category_id,
#{category_topic_id_column_select}
tu.notification_level,
GREATEST(
CASE
WHEN COALESCE(uo.new_topic_duration_minutes, :default_duration) = :always THEN u.created_at
WHEN COALESCE(uo.new_topic_duration_minutes, :default_duration) = :last_visit THEN COALESCE(
u.previous_visit_at,u.created_at
)
ELSE (:now::timestamp - INTERVAL '1 MINUTE' * COALESCE(uo.new_topic_duration_minutes, :default_duration))
END, u.created_at, :min_date
) AS treat_as_new_topic_start_date"
category_filter =
if admin
""
else
append = "OR u.admin" if !admin
<<~SQL
(
NOT c.read_restricted #{append} OR c.id IN (
SELECT c2.id FROM categories c2
JOIN category_groups cg ON cg.category_id = c2.id
JOIN group_users gu ON gu.user_id = :user_id AND cg.group_id = gu.group_id
WHERE c2.read_restricted )
) AND
SQL
end
visibility_filter =
if staff
""
else
append = "OR u.admin OR u.moderator" if !staff
"(topics.visible #{append}) AND"
end
tags_filter = ""
if muted_tag_ids.present? &&
%w[always only_muted].include?(SiteSetting.remove_muted_tags_from_latest)
existing_tags_sql =
"(select array_agg(tag_id) from topic_tags where topic_tags.topic_id = topics.id)"
muted_tags_array_sql = "ARRAY[#{muted_tag_ids.join(",")}]"
if SiteSetting.remove_muted_tags_from_latest == "always"
tags_filter = <<~SQL
NOT (
COALESCE(#{existing_tags_sql}, ARRAY[]::int[]) && #{muted_tags_array_sql}
) AND
SQL
else # only muted
tags_filter = <<~SQL
NOT (
COALESCE(#{existing_tags_sql}, ARRAY[-999]) <@ #{muted_tags_array_sql}
) AND
SQL
end
end
sql = +<<~SQL
SELECT #{select_sql}
FROM topics
JOIN users u on u.id = :user_id
JOIN user_options AS uo ON uo.user_id = u.id
JOIN categories c ON c.id = topics.category_id
LEFT JOIN topic_users tu ON tu.topic_id = topics.id AND tu.user_id = u.id
#{skip_new ? "" : "LEFT JOIN dismissed_topic_users ON dismissed_topic_users.topic_id = topics.id AND dismissed_topic_users.user_id = :user_id"}
#{additional_join_sql}
WHERE u.id = :user_id AND
#{filter_old_unread_sql}
topics.archetype <> 'private_message' AND
#{custom_state_filter ? custom_state_filter : "((#{unread}) OR (#{new})) AND"}
#{visibility_filter}
#{tags_filter}
topics.deleted_at IS NULL AND
#{category_filter}
NOT (
#{(skip_new && skip_unread) ? "" : "last_read_post_number IS NULL AND"}
(
topics.category_id IN (#{CategoryUser.muted_category_ids_query(user, include_direct: true).select("categories.id").to_sql})
AND tu.notification_level <= #{TopicUser.notification_levels[:regular]}
)
)
SQL
sql << " AND topics.id = :topic_id" if topic_id
sql << " ORDER BY topics.bumped_at DESC" unless skip_order
sql
end
def self.highest_post_number_column_select(whisperer)
"#{whisperer ? "topics.highest_staff_post_number AS highest_post_number" : "topics.highest_post_number"}"
end
def self.publish_read_indicator_on_write(topic_id, last_read_post_number, user_id)
topic =
Topic
.includes(:allowed_groups)
.select(:highest_post_number, :archetype, :id)
.find_by(id: topic_id)
if topic&.private_message?
groups = read_allowed_groups_of(topic)
update_topic_list_read_indicator(topic, groups, topic.highest_post_number, user_id, true)
end
end
def self.publish_read_indicator_on_read(topic_id, last_read_post_number, user_id)
topic =
Topic
.includes(:allowed_groups)
.select(:highest_post_number, :archetype, :id)
.find_by(id: topic_id)
if topic&.private_message?
groups = read_allowed_groups_of(topic)
post = Post.find_by(topic_id: topic.id, post_number: last_read_post_number)
trigger_post_read_count_update(post, groups, last_read_post_number, user_id)
update_topic_list_read_indicator(topic, groups, last_read_post_number, user_id, false)
end
end
def self.read_allowed_groups_of(topic)
topic
.allowed_groups
.joins(:group_users)
.where(publish_read_state: true)
.select("ARRAY_AGG(group_users.user_id) AS members", :name, :id)
.group("groups.id")
end
def self.update_topic_list_read_indicator(
topic,
groups,
last_read_post_number,
user_id,
write_event
)
return unless last_read_post_number == topic.highest_post_number
message = { topic_id: topic.id, show_indicator: write_event }.as_json
groups_to_update = []
groups.each do |group|
member = group.members.include?(user_id)
member_writing = (write_event && member)
non_member_reading = (!write_event && !member)
next if non_member_reading || member_writing
groups_to_update << group
end
return if groups_to_update.empty?
MessageBus.publish(
"/private-messages/unread-indicator/#{topic.id}",
message,
user_ids: groups_to_update.flat_map(&:members),
)
end
def self.trigger_post_read_count_update(post, groups, last_read_post_number, user_id)
return if !post
return if groups.empty?
opts = { readers_count: post.readers_count, reader_id: user_id }
post.publish_change_to_clients!(:read, opts)
end
def self.secure_category_group_ids(topic)
category = topic.category
return [Group::AUTO_GROUPS[:admins]] if category.nil?
if category.read_restricted
ids = [Group::AUTO_GROUPS[:admins]]
ids.push(*category.secure_group_ids)
ids.uniq
else
nil
end
end
private_class_method :secure_category_group_ids
end