From 002c6763440c65bd6587dbe5b8e820b3713f406c Mon Sep 17 00:00:00 2001
From: Martin Brennan <mjrbrennan@gmail.com>
Date: Mon, 31 May 2021 09:22:28 +1000
Subject: [PATCH] DEV: Topic tracking state improvements (#12958)

Original PR which had to be reverted **https://github.com/discourse/discourse/pull/12555**. See the description there for what this PR is achieving, plus below.

The issue with the original PR is addressed in https://github.com/discourse/discourse/pull/12958/commits/92ef54f4020111ffacb0f2a27da5d5c2855f9d5d

If you went to the `x unread` link for a tag Chrome would freeze up and possibly crash, or eventually unfreeze after nearly 10 mins. Other routes for unread/new were similarly slow. From profiling the issue was the `sync` function of `topic-tracking-state.js`, which calls down to `isNew` which in turn calls `moment`, a change I had made in the PR above. The time it takes locally with ~1400 topics in the tracking state is 2.3 seconds.

To solve this issue, I have moved these calculations for "created in new period" and "unread not too old" into the tracking state serializer.

When I was looking at the profiler I also noticed this issue which was just compounding the problem. Every time we modify topic tracking state we recalculate the sidebar tracking/everything/tag counts. However this calls `forEachTracked` and `countTags` which can be quite expensive as they go through the whole tracking state (and were also calling the removed moment functions).

I added some logs and this was being called 30 times when navigating to a new /unread route because  `sync` is being called from `build-topic-route` (one for each topic loaded due to pagination). So I just added a debounce here and it makes things even faster.

Finally, I changed topic tracking state to use a Map so our counts of the state keys is faster (Maps have .size whereas objects you have to do Object.keys(obj) which is O(n).)
---
 .../app/components/discovery-topics-list.js   |  15 +-
 .../app/mixins/bulk-topic-selection.js        |   4 +-
 .../app/models/topic-tracking-state.js        | 794 ++++++++++++------
 .../discourse/tests/helpers/site-settings.js  |   1 +
 .../widgets/hamburger-menu-test.js            |  24 +-
 .../tests/unit/models/nav-item-test.js        |   6 +-
 .../unit/models/topic-tracking-state-test.js  | 772 ++++++++++++++---
 app/controllers/application_controller.rb     |   4 +-
 app/controllers/topics_controller.rb          |   2 +-
 app/controllers/users_controller.rb           |   4 +-
 app/models/topic_tracking_state.rb            | 251 ++++--
 app/serializers/current_user_serializer.rb    |   5 +
 .../topic_tracking_state_serializer.rb        |  14 +-
 lib/topic_query.rb                            |  27 +-
 14 files changed, 1442 insertions(+), 481 deletions(-)

diff --git a/app/assets/javascripts/discourse/app/components/discovery-topics-list.js b/app/assets/javascripts/discourse/app/components/discovery-topics-list.js
index 8665981311e..9b414b91198 100644
--- a/app/assets/javascripts/discourse/app/components/discovery-topics-list.js
+++ b/app/assets/javascripts/discourse/app/components/discovery-topics-list.js
@@ -21,8 +21,19 @@ const DiscoveryTopicsListComponent = Component.extend(UrlRefresh, LoadMore, {
     }
   },
 
-  @observes("topicTrackingState.states")
-  _updateTopics() {
+  @on("didInsertElement")
+  _monitorTrackingState() {
+    this.stateChangeCallbackId = this.topicTrackingState.onStateChange(
+      this._updateTrackingTopics.bind(this)
+    );
+  },
+
+  @on("willDestroyElement")
+  _removeTrackingStateChangeMonitor() {
+    this.topicTrackingState.offStateChange(this.stateChangeCallbackId);
+  },
+
+  _updateTrackingTopics() {
     this.topicTrackingState.updateTopics(this.model.topics);
   },
 
diff --git a/app/assets/javascripts/discourse/app/mixins/bulk-topic-selection.js b/app/assets/javascripts/discourse/app/mixins/bulk-topic-selection.js
index f2a11771ee2..edcae9b7f32 100644
--- a/app/assets/javascripts/discourse/app/mixins/bulk-topic-selection.js
+++ b/app/assets/javascripts/discourse/app/mixins/bulk-topic-selection.js
@@ -51,9 +51,7 @@ export default Mixin.create({
 
       promise.then((result) => {
         if (result && result.topic_ids) {
-          const tracker = this.topicTrackingState;
-          result.topic_ids.forEach((t) => tracker.removeTopic(t));
-          tracker.incrementMessageCount();
+          this.topicTrackingState.removeTopics(result.topic_ids);
         }
 
         this.send("closeModal");
diff --git a/app/assets/javascripts/discourse/app/models/topic-tracking-state.js b/app/assets/javascripts/discourse/app/models/topic-tracking-state.js
index b00e8427ba6..8b424ec007e 100644
--- a/app/assets/javascripts/discourse/app/models/topic-tracking-state.js
+++ b/app/assets/javascripts/discourse/app/models/topic-tracking-state.js
@@ -1,11 +1,11 @@
 import EmberObject, { get } from "@ember/object";
 import discourseComputed, { on } from "discourse-common/utils/decorators";
 import Category from "discourse/models/category";
+import { deepEqual, deepMerge } from "discourse-common/lib/object";
 import DiscourseURL from "discourse/lib/url";
 import { NotificationLevels } from "discourse/lib/notification-levels";
 import PreloadStore from "discourse/lib/preload-store";
 import User from "discourse/models/user";
-import { deepEqual } from "discourse-common/lib/object";
 import { isEmpty } from "@ember/utils";
 
 function isNew(topic) {
@@ -13,6 +13,7 @@ function isNew(topic) {
     topic.last_read_post_number === null &&
     ((topic.notification_level !== 0 && !topic.notification_level) ||
       topic.notification_level >= NotificationLevels.TRACKING) &&
+    topic.created_in_new_period &&
     isUnseen(topic)
   );
 }
@@ -21,7 +22,8 @@ function isUnread(topic) {
   return (
     topic.last_read_post_number !== null &&
     topic.last_read_post_number < topic.highest_post_number &&
-    topic.notification_level >= NotificationLevels.TRACKING
+    topic.notification_level >= NotificationLevels.TRACKING &&
+    topic.unread_not_too_old
   );
 }
 
@@ -48,105 +50,47 @@ const TopicTrackingState = EmberObject.extend({
   _setup() {
     this.unreadSequence = [];
     this.newSequence = [];
-    this.states = {};
+    this.states = new Map();
+    this.messageIncrementCallbacks = {};
+    this.stateChangeCallbacks = {};
+    this._trackedTopicLimit = 4000;
   },
 
+  /**
+   * Subscribe to MessageBus channels which are used for publishing changes
+   * to the tracking state. Each message received will modify state for
+   * a particular topic.
+   *
+   * See app/models/topic_tracking_state.rb for the data payloads published
+   * to each of the channels.
+   *
+   * @method establishChannels
+   */
   establishChannels() {
-    const tracker = this;
-
-    const process = (data) => {
-      if (["muted", "unmuted"].includes(data.message_type)) {
-        tracker.trackMutedOrUnmutedTopic(data);
-        return;
-      }
-
-      tracker.pruneOldMutedAndUnmutedTopics();
-
-      if (tracker.isMutedTopic(data.topic_id)) {
-        return;
-      }
-
-      if (
-        this.siteSettings.mute_all_categories_by_default &&
-        !tracker.isUnmutedTopic(data.topic_id)
-      ) {
-        return;
-      }
-
-      if (data.message_type === "delete") {
-        tracker.removeTopic(data.topic_id);
-        tracker.incrementMessageCount();
-      }
-
-      if (["new_topic", "latest"].includes(data.message_type)) {
-        const muted_category_ids = User.currentProp("muted_category_ids");
-        if (
-          muted_category_ids &&
-          muted_category_ids.includes(data.payload.category_id)
-        ) {
-          return;
-        }
-      }
-
-      if (["new_topic", "latest"].includes(data.message_type)) {
-        const mutedTagIds = User.currentProp("muted_tag_ids");
-        if (
-          hasMutedTags(
-            data.payload.topic_tag_ids,
-            mutedTagIds,
-            this.siteSettings
-          )
-        ) {
-          return;
-        }
-      }
-
-      if (data.message_type === "latest") {
-        tracker.notify(data);
-      }
-
-      if (data.message_type === "dismiss_new") {
-        tracker.dismissNewTopic(data);
-      }
-
-      if (["new_topic", "unread", "read"].includes(data.message_type)) {
-        tracker.notify(data);
-        const old = tracker.states["t" + data.topic_id];
-        if (!deepEqual(old, data.payload)) {
-          tracker.states["t" + data.topic_id] = data.payload;
-          tracker.notifyPropertyChange("states");
-          tracker.incrementMessageCount();
-        }
-      }
-    };
-
-    this.messageBus.subscribe("/new", process);
-    this.messageBus.subscribe("/latest", process);
+    this.messageBus.subscribe("/new", this._processChannelPayload.bind(this));
+    this.messageBus.subscribe(
+      "/latest",
+      this._processChannelPayload.bind(this)
+    );
     if (this.currentUser) {
       this.messageBus.subscribe(
         "/unread/" + this.currentUser.get("id"),
-        process
+        this._processChannelPayload.bind(this)
       );
     }
 
     this.messageBus.subscribe("/delete", (msg) => {
-      const old = tracker.states["t" + msg.topic_id];
-      if (old) {
-        old.deleted = true;
-      }
-      tracker.incrementMessageCount();
+      this.modifyStateProp(msg, "deleted", true);
+      this.incrementMessageCount();
     });
 
     this.messageBus.subscribe("/recover", (msg) => {
-      const old = tracker.states["t" + msg.topic_id];
-      if (old) {
-        delete old.deleted;
-      }
-      tracker.incrementMessageCount();
+      this.modifyStateProp(msg, "deleted", false);
+      this.incrementMessageCount();
     });
 
     this.messageBus.subscribe("/destroy", (msg) => {
-      tracker.incrementMessageCount();
+      this.incrementMessageCount();
       const currentRoute = DiscourseURL.router.currentRoute.parent;
       if (
         currentRoute.name === "topic" &&
@@ -181,17 +125,6 @@ const TopicTrackingState = EmberObject.extend({
     this.currentUser && this.currentUser.set(key, topics);
   },
 
-  dismissNewTopic(data) {
-    data.payload.topic_ids.forEach((k) => {
-      const topic = this.states[`t${k}`];
-      this.states[`t${k}`] = Object.assign({}, topic, {
-        is_seen: true,
-      });
-    });
-    this.notifyPropertyChange("states");
-    this.incrementMessageCount();
-  },
-
   pruneOldMutedAndUnmutedTopics() {
     const now = Date.now();
     let mutedTopics = this.mutedTopics().filter(
@@ -213,22 +146,50 @@ const TopicTrackingState = EmberObject.extend({
     return !!this.unmutedTopics().findBy("topicId", topicId);
   },
 
+  /**
+   * Updates the topic's last_read_post_number to the highestSeen post
+   * number, as long as the topic is being tracked.
+   *
+   * Calls onStateChange callbacks.
+   *
+   * @params {Number|String} topicId - The ID of the topic to set last_read_post_number for.
+   * @params {Number} highestSeen - The post number of the topic that should be
+   *                                used for last_read_post_number.
+   * @method updateSeen
+   */
   updateSeen(topicId, highestSeen) {
     if (!topicId || !highestSeen) {
       return;
     }
-    const state = this.states["t" + topicId];
+    const state = this.findState(topicId);
+    if (!state) {
+      return;
+    }
+
     if (
-      state &&
-      (!state.last_read_post_number ||
-        state.last_read_post_number < highestSeen)
+      !state.last_read_post_number ||
+      state.last_read_post_number < highestSeen
     ) {
-      state.last_read_post_number = highestSeen;
+      this.modifyStateProp(topicId, "last_read_post_number", highestSeen);
       this.incrementMessageCount();
     }
   },
 
-  notify(data) {
+  /**
+   * Used to count incoming topics which will be displayed in a message
+   * at the top of the topic list, if hasIncoming is true (which is if
+   * incomingCount > 0).
+   *
+   * This will do nothing unless resetTracking or trackIncoming has been
+   * called; newIncoming will be null instead of an array. trackIncoming
+   * is called by various topic routes, as is resetTracking.
+   *
+   * @method notifyIncoming
+   * @param {Object} data - The data sent by TopicTrackingState to MessageBus
+   *                        which includes the message_type, payload of the topic,
+   *                        and the topic_id.
+   */
+  notifyIncoming(data) {
     if (!this.newIncoming) {
       return;
     }
@@ -240,6 +201,9 @@ const TopicTrackingState = EmberObject.extend({
     const filterCategory = this.filterCategory;
     const categoryId = data.payload && data.payload.category_id;
 
+    // if we have a filter category currently and it is not the
+    // same as the topic category from the payload, then do nothing
+    // because it doesn't need to be counted as incoming
     if (filterCategory && filterCategory.get("id") !== categoryId) {
       const category = categoryId && Category.findById(categoryId);
       if (
@@ -250,46 +214,67 @@ const TopicTrackingState = EmberObject.extend({
       }
     }
 
+    // always count a new_topic as incoming
     if (
       ["all", "latest", "new"].includes(filter) &&
       data.message_type === "new_topic"
     ) {
-      this.addIncoming(data.topic_id);
+      this._addIncoming(data.topic_id);
     }
 
+    // count an unread topic as incoming
     if (["all", "unread"].includes(filter) && data.message_type === "unread") {
-      const old = this.states["t" + data.topic_id];
+      const old = this.findState(data);
+
+      // the highest post number is equal to last read post number here
+      // because the state has already been modified based on the /unread
+      // messageBus message
       if (!old || old.highest_post_number === old.last_read_post_number) {
-        this.addIncoming(data.topic_id);
+        this._addIncoming(data.topic_id);
       }
     }
 
+    // always add incoming if looking at the latest list and a latest channel
+    // message comes through
     if (filter === "latest" && data.message_type === "latest") {
-      this.addIncoming(data.topic_id);
+      this._addIncoming(data.topic_id);
     }
 
+    // hasIncoming relies on this count
     this.set("incomingCount", this.newIncoming.length);
   },
 
-  addIncoming(topicId) {
-    if (this.newIncoming.indexOf(topicId) === -1) {
-      this.newIncoming.push(topicId);
-    }
-  },
-
+  /**
+   * Resets the number of incoming topics to 0 and flushes the new topics
+   * from the array. Without calling this or trackIncoming the notifyIncoming
+   * method will do nothing.
+   *
+   * @method resetTracking
+   */
   resetTracking() {
     this.newIncoming = [];
     this.set("incomingCount", 0);
   },
 
-  // track how many new topics came for this filter
+  /**
+   * Track how many new topics came for the specified filter.
+   *
+   * Related/intertwined with notifyIncoming; the filter and filterCategory
+   * set here is used to determine whether or not to add incoming counts
+   * based on message types of incoming MessageBus messages (via establishChannels)
+   *
+   * @method trackIncoming
+   * @param {String} filter - Valid values are all, categories, and any topic list
+   *                          filters e.g. latest, unread, new. As well as this
+   *                          specific category and tag URLs like /tag/test/l/latest
+   *                          or c/cat/subcat/6/l/latest.
+   */
   trackIncoming(filter) {
     this.newIncoming = [];
-    const split = filter.split("/");
 
+    const split = filter.split("/");
     if (split.length >= 4) {
       filter = split[split.length - 1];
-      // c/cat/subcat/6/l/latest
       let category = Category.findSingleBySlug(
         split.splice(1, split.length - 4).join("/")
       );
@@ -302,145 +287,126 @@ const TopicTrackingState = EmberObject.extend({
     this.set("incomingCount", 0);
   },
 
+  /**
+   * Used to determine whether toshow the message at the top of the topic list
+   * e.g. "see 1 new or updated topic"
+   *
+   * @method incomingCount
+   */
   @discourseComputed("incomingCount")
   hasIncoming(incomingCount) {
     return incomingCount && incomingCount > 0;
   },
 
-  removeTopic(topic_id) {
-    delete this.states["t" + topic_id];
+  /**
+   * Removes the topic ID provided from the tracker state.
+   *
+   * Calls onStateChange callbacks.
+   *
+   * @param {Number|String} topicId - The ID of the topic to remove from state.
+   * @method removeTopic
+   */
+  removeTopic(topicId) {
+    this.states.delete(this._stateKey(topicId));
+    this._afterStateChange();
   },
 
-  // If we have a cached topic list, we can update it from our tracking information.
+  /**
+   * Removes multiple topics from the state at once, and increments
+   * the message count.
+   *
+   * Calls onStateChange callbacks.
+   *
+   * @param {Array} topicIds - The IDs of the topic to removes from state.
+   * @method removeTopics
+   */
+  removeTopics(topicIds) {
+    topicIds.forEach((topicId) => this.removeTopic(topicId));
+    this.incrementMessageCount();
+    this._afterStateChange();
+  },
+
+  /**
+   * If we have a cached topic list, we can update it from our tracking information
+   * if the last_read_post_number or is_seen property does not match what the
+   * cached topic has.
+   *
+   * @method updateTopics
+   * @param {Array} topics - An array of Topic models.
+   */
   updateTopics(topics) {
     if (isEmpty(topics)) {
       return;
     }
 
-    const states = this.states;
-    topics.forEach((t) => {
-      const state = states["t" + t.get("id")];
+    topics.forEach((topic) => {
+      const state = this.findState(topic.get("id"));
 
-      if (state) {
-        const lastRead = t.get("last_read_post_number");
-        const isSeen = t.get("is_seen");
-        if (
-          lastRead !== state.last_read_post_number ||
-          isSeen !== state.is_seen
-        ) {
-          const postsCount = t.get("posts_count");
-          let newPosts = postsCount - state.highest_post_number,
-            unread = postsCount - state.last_read_post_number;
+      if (!state) {
+        return;
+      }
 
-          if (newPosts < 0) {
-            newPosts = 0;
-          }
-          if (!state.last_read_post_number) {
-            unread = 0;
-          }
-          if (unread < 0) {
-            unread = 0;
-          }
+      const lastRead = topic.get("last_read_post_number");
+      const isSeen = topic.get("is_seen");
 
-          t.setProperties({
-            highest_post_number: state.highest_post_number,
-            last_read_post_number: state.last_read_post_number,
-            new_posts: newPosts,
-            unread: unread,
-            is_seen: state.is_seen,
-            unseen: !state.last_read_post_number && isUnseen(state),
-          });
+      if (
+        lastRead !== state.last_read_post_number ||
+        isSeen !== state.is_seen
+      ) {
+        const postsCount = topic.get("posts_count");
+        let newPosts = postsCount - state.highest_post_number,
+          unread = postsCount - state.last_read_post_number;
+
+        if (newPosts < 0) {
+          newPosts = 0;
         }
+        if (!state.last_read_post_number) {
+          unread = 0;
+        }
+        if (unread < 0) {
+          unread = 0;
+        }
+
+        topic.setProperties({
+          highest_post_number: state.highest_post_number,
+          last_read_post_number: state.last_read_post_number,
+          new_posts: newPosts,
+          unread: unread,
+          is_seen: state.is_seen,
+          unseen: !state.last_read_post_number && isUnseen(state),
+        });
       }
     });
   },
 
+  /**
+   * Uses the provided topic list to apply changes to the in-memory topic
+   * tracking state, remove state as required, and also compensate for missing
+   * in-memory state.
+   *
+   * Any state changes will make a callback to all state change callbacks defined
+   * via onStateChange and all message increment callbacks defined via onMessageIncrement
+   *
+   * @method sync
+   * @param {TopicList} list
+   * @param {String} filter - The filter used for the list e.g. new/unread
+   * @param {Object} queryParams - The query parameters for the list e.g. page
+   */
   sync(list, filter, queryParams) {
-    const tracker = this,
-      states = tracker.states;
-
     if (!list || !list.topics) {
       return;
     }
 
-    // compensate for delayed "new" topics
-    // client side we know they are not new, server side we think they are
-    for (let i = list.topics.length - 1; i >= 0; i--) {
-      const state = states["t" + list.topics[i].id];
-      if (state && state.last_read_post_number > 0) {
-        if (filter === "new") {
-          list.topics.splice(i, 1);
-        } else {
-          list.topics[i].set("unseen", false);
-          list.topics[i].set("dont_sync", true);
-        }
-      }
-    }
+    // make sure any server-side state matches reality in the client side
+    this._fixDelayedServerState(list, filter);
 
-    list.topics.forEach(function (topic) {
-      const row = tracker.states["t" + topic.id] || {};
-      row.topic_id = topic.id;
-      row.notification_level = topic.notification_level;
+    // make sure all the state is up to date with what is accurate
+    // from the server
+    list.topics.forEach(this._syncStateFromListTopic.bind(this));
 
-      if (topic.unseen) {
-        row.last_read_post_number = null;
-      } else if (topic.unread || topic.new_posts) {
-        row.last_read_post_number =
-          topic.highest_post_number -
-          ((topic.unread || 0) + (topic.new_posts || 0));
-      } else {
-        if (!topic.dont_sync) {
-          delete tracker.states["t" + topic.id];
-        }
-        return;
-      }
-
-      row.highest_post_number = topic.highest_post_number;
-      if (topic.category) {
-        row.category_id = topic.category.id;
-      }
-
-      if (topic.tags) {
-        row.tags = topic.tags;
-      }
-
-      tracker.states["t" + topic.id] = row;
-    });
-
-    // Correct missing states, safeguard in case message bus is corrupt
-    let shouldCompensate =
-      (filter === "new" || filter === "unread") && !list.more_topics_url;
-
-    if (shouldCompensate && queryParams) {
-      Object.keys(queryParams).forEach((k) => {
-        if (k !== "ascending" && k !== "order") {
-          shouldCompensate = false;
-        }
-      });
-    }
-
-    if (shouldCompensate) {
-      const ids = {};
-      list.topics.forEach((r) => (ids["t" + r.id] = true));
-
-      Object.keys(tracker.states).forEach((k) => {
-        // we are good if we are on the list
-        if (ids[k]) {
-          return;
-        }
-
-        const v = tracker.states[k];
-
-        if (filter === "unread" && isUnread(v)) {
-          // pretend read
-          v.last_read_post_number = v.highest_post_number;
-        }
-
-        if (filter === "new" && isNew(v)) {
-          // pretend not new
-          v.last_read_post_number = 1;
-        }
-      });
+    // correct missing states, safeguard in case message bus is corrupt
+    if (this._shouldCompensateState(list, filter, queryParams)) {
+      this._correctMissingState(list, filter);
     }
 
     this.incrementMessageCount();
@@ -448,6 +414,27 @@ const TopicTrackingState = EmberObject.extend({
 
   incrementMessageCount() {
     this.incrementProperty("messageCount");
+    Object.values(this.messageIncrementCallbacks).forEach((cb) => cb());
+  },
+
+  _generateCallbackId() {
+    return Math.random().toString(12).substr(2, 9);
+  },
+
+  onMessageIncrement(cb) {
+    let callbackId = this._generateCallbackId();
+    this.messageIncrementCallbacks[callbackId] = cb;
+    return callbackId;
+  },
+
+  onStateChange(cb) {
+    let callbackId = this._generateCallbackId();
+    this.stateChangeCallbacks[callbackId] = cb;
+    return callbackId;
+  },
+
+  offStateChange(callbackId) {
+    delete this.stateChangeCallbacks[callbackId];
   },
 
   getSubCategoryIds(categoryId) {
@@ -471,11 +458,11 @@ const TopicTrackingState = EmberObject.extend({
       : this.getSubCategoryIds(categoryId);
     const mutedCategoryIds =
       this.currentUser && this.currentUser.muted_category_ids;
-    let filter = type === "new" ? isNew : isUnread;
+    let filterFn = type === "new" ? isNew : isUnread;
 
-    return Object.values(this.states).filter(
+    return Array.from(this.states.values()).filter(
       (topic) =>
-        filter(topic) &&
+        filterFn(topic) &&
         topic.archetype !== "private_message" &&
         !topic.deleted &&
         (!categoryId || subcategoryIds.has(topic.category_id)) &&
@@ -499,46 +486,78 @@ const TopicTrackingState = EmberObject.extend({
     );
   },
 
-  forEachTracked(fn) {
-    Object.values(this.states).forEach((topic) => {
-      if (topic.archetype !== "private_message" && !topic.deleted) {
-        let newTopic = isNew(topic);
-        let unreadTopic = isUnread(topic);
-        if (newTopic || unreadTopic) {
-          fn(topic, newTopic, unreadTopic);
-        }
-      }
+  /**
+   * Calls the provided callback for each of the currenty tracked topics
+   * we have in state.
+   *
+   * @method forEachTracked
+   * @param {Function} fn - The callback function to call with the topic,
+   *                        newTopic which is a boolean result of isNew,
+   *                        and unreadTopic which is a boolean result of
+   *                        isUnread.
+   */
+  forEachTracked(fn, opts = {}) {
+    this._trackedTopics(opts).forEach((trackedTopic) => {
+      fn(trackedTopic.topic, trackedTopic.newTopic, trackedTopic.unreadTopic);
     });
   },
 
-  countTags(tags) {
+  /**
+   * Using the array of tags provided, tallys up all topics via forEachTracked
+   * that we are tracking, separated into new/unread/total.
+   *
+   * Total is only counted if opts.includeTotal is specified.
+   *
+   * Output (from input ["pending", "bug"]):
+   *
+   * {
+   *   pending: { unreadCount: 6, newCount: 1, totalCount: 10 },
+   *   bug: { unreadCount: 0, newCount: 4, totalCount: 20 }
+   * }
+   *
+   * @method countTags
+   * @param opts - Valid options:
+   *                 * includeTotal - When true, a totalCount is incremented for
+   *                                all topics matching a tag.
+   */
+  countTags(tags, opts = {}) {
     let counts = {};
 
     tags.forEach((tag) => {
       counts[tag] = { unreadCount: 0, newCount: 0 };
-    });
-
-    this.forEachTracked((topic, newTopic, unreadTopic) => {
-      if (topic.tags) {
-        tags.forEach((tag) => {
-          if (topic.tags.indexOf(tag) > -1) {
-            if (unreadTopic) {
-              counts[tag].unreadCount++;
-            }
-            if (newTopic) {
-              counts[tag].newCount++;
-            }
-          }
-        });
+      if (opts.includeTotal) {
+        counts[tag].totalCount = 0;
       }
     });
 
+    this.forEachTracked(
+      (topic, newTopic, unreadTopic) => {
+        if (topic.tags && topic.tags.length > 0) {
+          tags.forEach((tag) => {
+            if (topic.tags.indexOf(tag) > -1) {
+              if (unreadTopic) {
+                counts[tag].unreadCount++;
+              }
+              if (newTopic) {
+                counts[tag].newCount++;
+              }
+
+              if (opts.includeTotal) {
+                counts[tag].totalCount++;
+              }
+            }
+          });
+        }
+      },
+      { includeAll: opts.includeTotal }
+    );
+
     return counts;
   },
 
   countCategory(category_id, tagId) {
     let sum = 0;
-    Object.values(this.states).forEach((topic) => {
+    for (let topic of this.states.values()) {
       if (
         topic.category_id === category_id &&
         !topic.deleted &&
@@ -550,7 +569,7 @@ const TopicTrackingState = EmberObject.extend({
             ? 1
             : 0;
       }
-    });
+    }
     return sum;
   },
 
@@ -577,21 +596,272 @@ const TopicTrackingState = EmberObject.extend({
   },
 
   loadStates(data) {
-    const states = this.states;
+    (data || []).forEach((topic) => {
+      this.modifyState(topic, topic);
+    });
+  },
 
-    // I am taking some shortcuts here to avoid 500 gets for a large list
-    if (data) {
-      data.forEach((topic) => {
-        states["t" + topic.topic_id] = topic;
+  modifyState(topic, data) {
+    this.states.set(this._stateKey(topic), data);
+    this._afterStateChange();
+  },
+
+  modifyStateProp(topic, prop, data) {
+    const state = this.findState(topic);
+    if (state) {
+      state[prop] = data;
+      this._afterStateChange();
+    }
+  },
+
+  findState(topicOrId) {
+    return this.states.get(this._stateKey(topicOrId));
+  },
+
+  /*
+   * private
+   */
+
+  // fix delayed "new" topics by removing the now seen
+  // topic from the list (for the "new" list) or setting the topic
+  // to "seen" for other lists.
+  //
+  // client side we know they are not new, server side we think they are.
+  // this can happen if the list is cached or the update to the state
+  // for a particular seen topic has not yet reached the server.
+  _fixDelayedServerState(list, filter) {
+    for (let index = list.topics.length - 1; index >= 0; index--) {
+      const state = this.findState(list.topics[index].id);
+      if (state && state.last_read_post_number > 0) {
+        if (filter === "new") {
+          list.topics.splice(index, 1);
+        } else {
+          list.topics[index].set("unseen", false);
+          list.topics[index].set("prevent_sync", true);
+        }
+      }
+    }
+  },
+
+  // this updates the topic in the state to match the
+  // topic from the list (e.g. updates category, highest read post
+  // number, tags etc.)
+  _syncStateFromListTopic(topic) {
+    const state = this.findState(topic.id) || {};
+
+    // make a new copy so we aren't modifying the state object directly while
+    // we make changes
+    const newState = { ...state };
+
+    newState.topic_id = topic.id;
+    newState.notification_level = topic.notification_level;
+
+    // see ListableTopicSerializer for unread/unseen/new_posts and other
+    // topic property logic
+    if (topic.unseen) {
+      newState.last_read_post_number = null;
+    } else if (topic.unread || topic.new_posts) {
+      newState.last_read_post_number =
+        topic.highest_post_number -
+        ((topic.unread || 0) + (topic.new_posts || 0));
+    } else {
+      // remove the topic if it is no longer unread/new (it has been seen)
+      // and if there are too many topics in memory
+      if (!topic.prevent_sync && this._maxStateSizeReached()) {
+        this.removeTopic(topic.id);
+      }
+      return;
+    }
+
+    newState.highest_post_number = topic.highest_post_number;
+    if (topic.category) {
+      newState.category_id = topic.category.id;
+    }
+
+    if (topic.tags) {
+      newState.tags = topic.tags;
+    }
+
+    this.modifyState(topic.id, newState);
+  },
+
+  // this stops sync of tracking state when list is filtered, in the past this
+  // would cause the tracking state to become inconsistent.
+  _shouldCompensateState(list, filter, queryParams) {
+    let shouldCompensate =
+      (filter === "new" || filter === "unread") && !list.more_topics_url;
+
+    if (shouldCompensate && queryParams) {
+      Object.keys(queryParams).forEach((k) => {
+        if (k !== "ascending" && k !== "order") {
+          shouldCompensate = false;
+        }
       });
     }
+
+    return shouldCompensate;
+  },
+
+  // any state that is not in the provided list must be updated
+  // based on the filter selected so we do not have any incorrect
+  // state in the list
+  _correctMissingState(list, filter) {
+    const ids = {};
+    list.topics.forEach((topic) => (ids[this._stateKey(topic.id)] = true));
+
+    for (let topicKey of this.states.keys()) {
+      // if the topic is already in the list then there is
+      // no compensation needed; we already have latest state
+      // from the backend
+      if (ids[topicKey]) {
+        return;
+      }
+
+      const newState = { ...this.findState(topicKey) };
+      if (filter === "unread" && isUnread(newState)) {
+        // pretend read. if unread, the highest_post_number will be greater
+        // than the last_read_post_number
+        newState.last_read_post_number = newState.highest_post_number;
+      }
+
+      if (filter === "new" && isNew(newState)) {
+        // pretend not new. if the topic is new, then last_read_post_number
+        // will be null.
+        newState.last_read_post_number = 1;
+      }
+
+      this.modifyState(topicKey, newState);
+    }
+  },
+
+  // processes the data sent via messageBus, called by establishChannels
+  _processChannelPayload(data) {
+    if (["muted", "unmuted"].includes(data.message_type)) {
+      this.trackMutedOrUnmutedTopic(data);
+      return;
+    }
+
+    this.pruneOldMutedAndUnmutedTopics();
+
+    if (this.isMutedTopic(data.topic_id)) {
+      return;
+    }
+
+    if (
+      this.siteSettings.mute_all_categories_by_default &&
+      !this.isUnmutedTopic(data.topic_id)
+    ) {
+      return;
+    }
+
+    if (["new_topic", "latest"].includes(data.message_type)) {
+      const muted_category_ids = User.currentProp("muted_category_ids");
+      if (
+        muted_category_ids &&
+        muted_category_ids.includes(data.payload.category_id)
+      ) {
+        return;
+      }
+    }
+
+    if (["new_topic", "latest"].includes(data.message_type)) {
+      const mutedTagIds = User.currentProp("muted_tag_ids");
+      if (
+        hasMutedTags(data.payload.topic_tag_ids, mutedTagIds, this.siteSettings)
+      ) {
+        return;
+      }
+    }
+
+    const old = this.findState(data);
+
+    if (data.message_type === "latest") {
+      this.notifyIncoming(data);
+
+      if ((old && old.tags) !== data.payload.tags) {
+        this.modifyStateProp(data, "tags", data.payload.tags);
+        this.incrementMessageCount();
+      }
+    }
+
+    if (data.message_type === "dismiss_new") {
+      this._dismissNewTopics(data.payload.topic_ids);
+    }
+
+    if (["new_topic", "unread", "read"].includes(data.message_type)) {
+      this.notifyIncoming(data);
+      if (!deepEqual(old, data.payload)) {
+        if (data.message_type === "read") {
+          let mergeData = {};
+
+          // we have to do this because the "read" event does not
+          // include tags; we don't want them to be overridden
+          if (old) {
+            mergeData = {
+              tags: old.tags,
+              topic_tag_ids: old.topic_tag_ids,
+            };
+          }
+
+          this.modifyState(data, deepMerge(data.payload, mergeData));
+        } else {
+          this.modifyState(data, data.payload);
+        }
+        this.incrementMessageCount();
+      }
+    }
+  },
+
+  _dismissNewTopics(topicIds) {
+    topicIds.forEach((topicId) => {
+      this.modifyStateProp(topicId, "is_seen", true);
+    });
+    this.incrementMessageCount();
+  },
+
+  _addIncoming(topicId) {
+    if (this.newIncoming.indexOf(topicId) === -1) {
+      this.newIncoming.push(topicId);
+    }
+  },
+
+  _trackedTopics(opts = {}) {
+    return Array.from(this.states.values())
+      .map((topic) => {
+        if (topic.archetype !== "private_message" && !topic.deleted) {
+          let newTopic = isNew(topic);
+          let unreadTopic = isUnread(topic);
+          if (newTopic || unreadTopic || opts.includeAll) {
+            return { topic, newTopic, unreadTopic };
+          }
+        }
+      })
+      .compact();
+  },
+
+  _stateKey(topicOrId) {
+    if (typeof topicOrId === "number") {
+      return `t${topicOrId}`;
+    } else if (typeof topicOrId === "string" && topicOrId.indexOf("t") > -1) {
+      return topicOrId;
+    } else {
+      return `t${topicOrId.topic_id}`;
+    }
+  },
+
+  _afterStateChange() {
+    this.notifyPropertyChange("states");
+    Object.values(this.stateChangeCallbacks).forEach((cb) => cb());
+  },
+
+  _maxStateSizeReached() {
+    return this.states.size >= this._trackedTopicLimit;
   },
 });
 
 export function startTracking(tracking) {
   const data = PreloadStore.get("topicTrackingStates");
   tracking.loadStates(data);
-  tracking.initialStatesLength = data && data.length;
   tracking.establishChannels();
   PreloadStore.remove("topicTrackingStates");
 }
diff --git a/app/assets/javascripts/discourse/tests/helpers/site-settings.js b/app/assets/javascripts/discourse/tests/helpers/site-settings.js
index 3726d181087..2b86811813b 100644
--- a/app/assets/javascripts/discourse/tests/helpers/site-settings.js
+++ b/app/assets/javascripts/discourse/tests/helpers/site-settings.js
@@ -98,6 +98,7 @@ const ORIGINAL_SETTINGS = {
   unicode_usernames: false,
   secure_media: false,
   external_emoji_url: "",
+  remove_muted_tags_from_latest: "always",
 };
 
 let siteSettings = Object.assign({}, ORIGINAL_SETTINGS);
diff --git a/app/assets/javascripts/discourse/tests/integration/widgets/hamburger-menu-test.js b/app/assets/javascripts/discourse/tests/integration/widgets/hamburger-menu-test.js
index 52fd461819f..b07bdd9c5c5 100644
--- a/app/assets/javascripts/discourse/tests/integration/widgets/hamburger-menu-test.js
+++ b/app/assets/javascripts/discourse/tests/integration/widgets/hamburger-menu-test.js
@@ -171,20 +171,22 @@ discourseModule(
             } else if (unreadCategoryIds.length === 0) {
               unreadCategoryIds.push(c.id);
               for (let i = 0; i < 5; i++) {
-                c.topicTrackingState.states["t123" + i] = {
+                c.topicTrackingState.modifyState(123 + i, {
                   category_id: c.id,
                   last_read_post_number: 1,
                   highest_post_number: 2,
                   notification_level: NotificationLevels.TRACKING,
-                };
+                  unread_not_too_old: true,
+                });
               }
             } else {
               unreadCategoryIds.splice(0, 0, c.id);
               for (let i = 0; i < 10; i++) {
-                c.topicTrackingState.states["t321" + i] = {
+                c.topicTrackingState.modifyState(321 + i, {
                   category_id: c.id,
                   last_read_post_number: null,
-                };
+                  created_in_new_period: true,
+                });
               }
               return false;
             }
@@ -195,7 +197,11 @@ discourseModule(
       },
 
       test(assert) {
-        assert.equal(queryAll(".category-link").length, maxCategoriesToDisplay);
+        assert.equal(
+          queryAll(".category-link").length,
+          maxCategoriesToDisplay,
+          "categories displayed limited by header_dropdown_category_count"
+        );
 
         categoriesByCount = categoriesByCount.filter(
           (c) => !mutedCategoryIds.includes(c.id)
@@ -211,8 +217,12 @@ discourseModule(
         assert.equal(
           queryAll(".category-link .category-name").text(),
           ids
-            .map((i) => categoriesByCount.find((c) => c.id === i).name)
-            .join("")
+            .map(
+              (id) =>
+                categoriesByCount.find((category) => category.id === id).name
+            )
+            .join(""),
+          "top categories are in the correct order"
         );
       },
     });
diff --git a/app/assets/javascripts/discourse/tests/unit/models/nav-item-test.js b/app/assets/javascripts/discourse/tests/unit/models/nav-item-test.js
index 329aa30fa68..409eae4d463 100644
--- a/app/assets/javascripts/discourse/tests/unit/models/nav-item-test.js
+++ b/app/assets/javascripts/discourse/tests/unit/models/nav-item-test.js
@@ -40,7 +40,11 @@ module("Unit | Model | nav-item", function (hooks) {
     assert.equal(navItem.get("count"), 0, "it has no count by default");
 
     const tracker = navItem.get("topicTrackingState");
-    tracker.states["t1"] = { topic_id: 1, last_read_post_number: null };
+    tracker.modifyState("t1", {
+      topic_id: 1,
+      last_read_post_number: null,
+      created_in_new_period: true,
+    });
     tracker.incrementMessageCount();
 
     assert.equal(
diff --git a/app/assets/javascripts/discourse/tests/unit/models/topic-tracking-state-test.js b/app/assets/javascripts/discourse/tests/unit/models/topic-tracking-state-test.js
index 56542dfb959..d8d5935d072 100644
--- a/app/assets/javascripts/discourse/tests/unit/models/topic-tracking-state-test.js
+++ b/app/assets/javascripts/discourse/tests/unit/models/topic-tracking-state-test.js
@@ -1,12 +1,20 @@
-import { module, test } from "qunit";
+import { test } from "qunit";
+import DiscourseURL from "discourse/lib/url";
+import { getProperties } from "@ember/object";
 import Category from "discourse/models/category";
+import MessageBus from "message-bus-client";
+import {
+  discourseModule,
+  publishToMessageBus,
+} from "discourse/tests/helpers/qunit-helpers";
 import { NotificationLevels } from "discourse/lib/notification-levels";
 import TopicTrackingState from "discourse/models/topic-tracking-state";
 import User from "discourse/models/user";
+import Topic from "discourse/models/topic";
 import createStore from "discourse/tests/helpers/create-store";
 import sinon from "sinon";
 
-module("Unit | Model | topic-tracking-state", function (hooks) {
+discourseModule("Unit | Model | topic-tracking-state", function (hooks) {
   hooks.beforeEach(function () {
     this.clock = sinon.useFakeTimers(new Date(2012, 11, 31, 12, 0).getTime());
   });
@@ -16,18 +24,20 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
   });
 
   test("tag counts", function (assert) {
-    const state = TopicTrackingState.create();
+    const trackingState = TopicTrackingState.create();
 
-    state.loadStates([
+    trackingState.loadStates([
       {
         topic_id: 1,
         last_read_post_number: null,
-        tags: ["foo", "new"],
+        tags: ["foo", "baz"],
+        created_in_new_period: true,
       },
       {
         topic_id: 2,
         last_read_post_number: null,
-        tags: ["new"],
+        tags: ["baz"],
+        created_in_new_period: true,
       },
       {
         topic_id: 3,
@@ -38,15 +48,17 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
         topic_id: 4,
         last_read_post_number: 1,
         highest_post_number: 7,
-        tags: ["unread"],
+        tags: ["pending"],
         notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
       },
       {
         topic_id: 5,
         last_read_post_number: 1,
         highest_post_number: 7,
-        tags: ["bar", "unread"],
+        tags: ["bar", "pending"],
         notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
       },
       {
         topic_id: 6,
@@ -57,18 +69,100 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       },
     ]);
 
-    const states = state.countTags(["new", "unread"]);
+    const tagCounts = trackingState.countTags(["baz", "pending"]);
 
-    assert.equal(states["new"].newCount, 2, "new counts");
-    assert.equal(states["new"].unreadCount, 0, "new counts");
-    assert.equal(states["unread"].unreadCount, 2, "unread counts");
-    assert.equal(states["unread"].newCount, 0, "unread counts");
+    assert.equal(tagCounts["baz"].newCount, 2, "baz tag new counts");
+    assert.equal(tagCounts["baz"].unreadCount, 0, "baz tag unread counts");
+    assert.equal(
+      tagCounts["pending"].unreadCount,
+      2,
+      "pending tag unread counts"
+    );
+    assert.equal(tagCounts["pending"].newCount, 0, "pending tag new counts");
+  });
+
+  test("tag counts - with total", function (assert) {
+    const trackingState = TopicTrackingState.create();
+
+    trackingState.loadStates([
+      {
+        topic_id: 1,
+        last_read_post_number: null,
+        tags: ["foo", "baz"],
+        created_in_new_period: true,
+      },
+      {
+        topic_id: 2,
+        last_read_post_number: null,
+        tags: ["baz"],
+        created_in_new_period: true,
+      },
+      {
+        topic_id: 3,
+        last_read_post_number: null,
+        tags: ["random"],
+      },
+      {
+        topic_id: 4,
+        last_read_post_number: 1,
+        highest_post_number: 7,
+        tags: ["pending"],
+        notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
+      },
+      {
+        topic_id: 5,
+        last_read_post_number: 1,
+        highest_post_number: 7,
+        tags: ["bar", "pending"],
+        notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
+      },
+      {
+        topic_id: 6,
+        last_read_post_number: 1,
+        highest_post_number: 7,
+        tags: null,
+        notification_level: NotificationLevels.TRACKING,
+      },
+      {
+        topic_id: 7,
+        last_read_post_number: 7,
+        highest_post_number: 7,
+        tags: ["foo", "baz"],
+      },
+      {
+        topic_id: 8,
+        last_read_post_number: 4,
+        highest_post_number: 4,
+        tags: ["pending"],
+        notification_level: NotificationLevels.TRACKING,
+      },
+      {
+        topic_id: 9,
+        last_read_post_number: 88,
+        highest_post_number: 88,
+        tags: ["pending"],
+        notification_level: NotificationLevels.TRACKING,
+      },
+    ]);
+
+    const states = trackingState.countTags(["baz", "pending"], {
+      includeTotal: true,
+    });
+
+    assert.equal(states["baz"].newCount, 2, "baz tag new counts");
+    assert.equal(states["baz"].unreadCount, 0, "baz tag unread counts");
+    assert.equal(states["baz"].totalCount, 3, "baz tag total counts");
+    assert.equal(states["pending"].unreadCount, 2, "pending tag unread counts");
+    assert.equal(states["pending"].newCount, 0, "pending tag new counts");
+    assert.equal(states["pending"].totalCount, 4, "pending tag total counts");
   });
 
   test("forEachTracked", function (assert) {
-    const state = TopicTrackingState.create();
+    const trackingState = TopicTrackingState.create();
 
-    state.loadStates([
+    trackingState.loadStates([
       {
         topic_id: 1,
         last_read_post_number: null,
@@ -83,22 +177,25 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
         topic_id: 3,
         last_read_post_number: null,
         tags: ["random"],
+        created_in_new_period: true,
       },
       {
         topic_id: 4,
         last_read_post_number: 1,
         highest_post_number: 7,
         category_id: 7,
-        tags: ["unread"],
+        tags: ["bug"],
         notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
       },
       {
         topic_id: 5,
         last_read_post_number: 1,
         highest_post_number: 7,
-        tags: ["bar", "unread"],
+        tags: ["bar", "bug"],
         category_id: 7,
         notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
       },
       {
         topic_id: 6,
@@ -114,7 +211,7 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       sevenUnread = 0,
       sevenNew = 0;
 
-    state.forEachTracked((topic, isNew, isUnread) => {
+    trackingState.forEachTracked((topic, isNew, isUnread) => {
       if (topic.category_id === 7) {
         if (isNew) {
           sevenNew += 1;
@@ -134,17 +231,17 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       }
     });
 
-    assert.equal(randomNew, 1, "random new");
-    assert.equal(randomUnread, 0, "random unread");
-    assert.equal(sevenNew, 0, "seven unread");
-    assert.equal(sevenUnread, 2, "seven unread");
+    assert.equal(randomNew, 1, "random tag new");
+    assert.equal(randomUnread, 0, "random tag unread");
+    assert.equal(sevenNew, 0, "category seven new");
+    assert.equal(sevenUnread, 2, "category seven unread");
   });
 
-  test("sync", function (assert) {
-    const state = TopicTrackingState.create();
-    state.states["t111"] = { last_read_post_number: null };
+  test("sync - delayed new topics for backend list are removed", function (assert) {
+    const trackingState = TopicTrackingState.create();
+    trackingState.loadStates([{ last_read_post_number: null, topic_id: 111 }]);
 
-    state.updateSeen(111, 7);
+    trackingState.updateSeen(111, 7);
     const list = {
       topics: [
         {
@@ -156,7 +253,7 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       ],
     };
 
-    state.sync(list, "new");
+    trackingState.sync(list, "new");
     assert.equal(
       list.topics.length,
       0,
@@ -164,6 +261,495 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
     );
   });
 
+  test("sync - delayed unread topics for backend list are marked seen", function (assert) {
+    const trackingState = TopicTrackingState.create();
+    trackingState.loadStates([{ last_read_post_number: null, topic_id: 111 }]);
+
+    trackingState.updateSeen(111, 7);
+    const list = {
+      topics: [
+        Topic.create({
+          highest_post_number: null,
+          id: 111,
+          unread: 10,
+          new_posts: 10,
+          unseen: true,
+          prevent_sync: false,
+        }),
+      ],
+    };
+
+    trackingState.sync(list, "unread");
+    assert.equal(
+      list.topics[0].unseen,
+      false,
+      "expect unread topic to be marked as seen"
+    );
+    assert.equal(
+      list.topics[0].prevent_sync,
+      true,
+      "expect unread topic to be marked as prevent_sync"
+    );
+  });
+
+  test("sync - remove topic from state for performance if it is seen and has no unread or new posts and there are too many tracked topics in memory", function (assert) {
+    const trackingState = TopicTrackingState.create();
+    trackingState.loadStates([{ topic_id: 111 }, { topic_id: 222 }]);
+    trackingState.set("_trackedTopicLimit", 1);
+
+    const list = {
+      topics: [
+        Topic.create({
+          id: 111,
+          unseen: false,
+          seen: true,
+          unread: 0,
+          new_posts: 0,
+          prevent_sync: false,
+        }),
+      ],
+    };
+
+    trackingState.sync(list, "unread");
+    assert.notOk(
+      trackingState.states.has("t111"),
+      "expect state for topic 111 to be deleted"
+    );
+
+    trackingState.loadStates([{ topic_id: 111 }, { topic_id: 222 }]);
+    trackingState.set("_trackedTopicLimit", 5);
+    trackingState.sync(list, "unread");
+    assert.ok(
+      trackingState.states.has("t111"),
+      "expect state for topic 111 not to be deleted"
+    );
+  });
+
+  test("sync - updates state to match list topic for unseen and unread/new topics", function (assert) {
+    const trackingState = TopicTrackingState.create();
+    trackingState.loadStates([
+      { topic_id: 111, last_read_post_number: 0 },
+      { topic_id: 222, last_read_post_number: 1 },
+    ]);
+
+    const list = {
+      topics: [
+        Topic.create({
+          id: 111,
+          unseen: true,
+          seen: false,
+          unread: 0,
+          new_posts: 0,
+          highest_post_number: 20,
+          category_id: 1,
+          tags: ["pending"],
+        }),
+        Topic.create({
+          id: 222,
+          unseen: false,
+          seen: true,
+          unread: 3,
+          new_posts: 0,
+          highest_post_number: 20,
+        }),
+      ],
+    };
+
+    trackingState.sync(list, "unread");
+
+    let state111 = trackingState.findState(111);
+    let state222 = trackingState.findState(222);
+    assert.equal(
+      state111.last_read_post_number,
+      null,
+      "unseen topics get last_read_post_number reset to null"
+    );
+    assert.propEqual(
+      getProperties(state111, "highest_post_number", "tags", "category_id"),
+      { highest_post_number: 20, tags: ["pending"], category_id: 1 },
+      "highest_post_number, category, and tags are set for a topic"
+    );
+    assert.equal(
+      state222.last_read_post_number,
+      17,
+      "last_read_post_number is highest_post_number - (unread + new)"
+    );
+  });
+
+  test("sync - states missing from the topic list are updated based on the selected filter", function (assert) {
+    const trackingState = TopicTrackingState.create();
+    trackingState.loadStates([
+      {
+        topic_id: 111,
+        last_read_post_number: 4,
+        highest_post_number: 5,
+        notification_level: NotificationLevels.TRACKING,
+        unread_not_too_old: true,
+      },
+      {
+        topic_id: 222,
+        last_read_post_number: null,
+        seen: false,
+        notification_level: NotificationLevels.TRACKING,
+        created_in_new_period: true,
+      },
+    ]);
+
+    const list = {
+      topics: [],
+    };
+
+    trackingState.sync(list, "unread");
+    assert.equal(
+      trackingState.findState(111).last_read_post_number,
+      5,
+      "last_read_post_number set to highest post number to pretend read"
+    );
+
+    trackingState.sync(list, "new");
+    assert.equal(
+      trackingState.findState(222).last_read_post_number,
+      1,
+      "last_read_post_number set to 1 to pretend not new"
+    );
+  });
+
+  discourseModule(
+    "establishChannels - /unread/:userId MessageBus channel payloads processed",
+    function (unreadHooks) {
+      let trackingState;
+      let unreadTopicPayload = {
+        topic_id: 111,
+        message_type: "unread",
+        payload: {
+          topic_id: 111,
+          category_id: 123,
+          topic_tag_ids: [44],
+          tags: ["pending"],
+          last_read_post_number: 4,
+          highest_post_number: 10,
+          created_at: "2012-11-31 12:00:00 UTC",
+          archetype: "regular",
+          notification_level: NotificationLevels.TRACKING,
+        },
+      };
+      let currentUser;
+
+      unreadHooks.beforeEach(function () {
+        currentUser = User.create({
+          username: "chuck",
+        });
+        User.resetCurrent(currentUser);
+
+        trackingState = TopicTrackingState.create({
+          messageBus: MessageBus,
+          currentUser,
+          siteSettings: this.siteSettings,
+        });
+        trackingState.establishChannels();
+        trackingState.loadStates([
+          {
+            topic_id: 111,
+            last_read_post_number: 4,
+            highest_post_number: 4,
+            notification_level: NotificationLevels.TRACKING,
+          },
+        ]);
+      });
+
+      test("message count is incremented and callback is called", function (assert) {
+        let messageIncrementCalled = false;
+        trackingState.onMessageIncrement(() => {
+          messageIncrementCalled = true;
+        });
+        publishToMessageBus(`/unread/${currentUser.id}`, unreadTopicPayload);
+        assert.equal(
+          trackingState.messageCount,
+          1,
+          "message count incremented"
+        );
+        assert.equal(
+          messageIncrementCalled,
+          true,
+          "message increment callback called"
+        );
+      });
+
+      test("state is modified and callback is called", function (assert) {
+        let stateCallbackCalled = false;
+        trackingState.onStateChange(() => {
+          stateCallbackCalled = true;
+        });
+        publishToMessageBus(`/unread/${currentUser.id}`, unreadTopicPayload);
+        assert.deepEqual(
+          trackingState.findState(111),
+          {
+            topic_id: 111,
+            category_id: 123,
+            topic_tag_ids: [44],
+            tags: ["pending"],
+            last_read_post_number: 4,
+            highest_post_number: 10,
+            notification_level: NotificationLevels.TRACKING,
+            created_at: "2012-11-31 12:00:00 UTC",
+            archetype: "regular",
+          },
+          "topic state updated"
+        );
+        assert.equal(stateCallbackCalled, true, "state change callback called");
+      });
+
+      test("adds incoming so it is counted in topic lists", function (assert) {
+        trackingState.trackIncoming("all");
+        publishToMessageBus(`/unread/${currentUser.id}`, unreadTopicPayload);
+        assert.deepEqual(
+          trackingState.newIncoming,
+          [111],
+          "unread topic is incoming"
+        );
+        assert.equal(
+          trackingState.incomingCount,
+          1,
+          "incoming count is increased"
+        );
+      });
+
+      test("dismisses new topic", function (assert) {
+        trackingState.loadStates([
+          {
+            last_read_post_number: null,
+            topic_id: 112,
+            notification_level: NotificationLevels.TRACKING,
+            category_id: 1,
+            is_seen: false,
+            tags: ["foo"],
+          },
+        ]);
+
+        publishToMessageBus(`/unread/${currentUser.id}`, {
+          message_type: "dismiss_new",
+          payload: { topic_ids: [112] },
+        });
+        assert.equal(trackingState.findState(112).is_seen, true);
+      });
+
+      test("marks a topic as read", function (assert) {
+        trackingState.loadStates([
+          {
+            last_read_post_number: null,
+            topic_id: 112,
+            notification_level: NotificationLevels.TRACKING,
+            category_id: 1,
+            is_seen: false,
+            tags: ["foo"],
+          },
+        ]);
+        publishToMessageBus(`/unread/${currentUser.id}`, {
+          message_type: "read",
+          topic_id: 112,
+          payload: {
+            topic_id: 112,
+            last_read_post_number: 4,
+            highest_post_number: 4,
+            notification_level: NotificationLevels.TRACKING,
+          },
+        });
+        assert.propEqual(
+          getProperties(
+            trackingState.findState(112),
+            "highest_post_number",
+            "last_read_post_number"
+          ),
+          { highest_post_number: 4, last_read_post_number: 4 },
+          "highest_post_number and last_read_post_number are set for a topic"
+        );
+        assert.deepEqual(
+          trackingState.findState(112).tags,
+          ["foo"],
+          "tags are not accidentally cleared"
+        );
+      });
+    }
+  );
+
+  discourseModule(
+    "establishChannels - /new MessageBus channel payloads processed",
+    function (establishChannelsHooks) {
+      let trackingState;
+      let newTopicPayload = {
+        topic_id: 222,
+        message_type: "new_topic",
+        payload: {
+          topic_id: 222,
+          category_id: 123,
+          topic_tag_ids: [44],
+          tags: ["pending"],
+          last_read_post_number: null,
+          highest_post_number: 1,
+          created_at: "2012-11-31 12:00:00 UTC",
+          archetype: "regular",
+        },
+      };
+      let currentUser;
+
+      establishChannelsHooks.beforeEach(function () {
+        currentUser = User.create({
+          username: "chuck",
+        });
+        User.resetCurrent(currentUser);
+
+        trackingState = TopicTrackingState.create({
+          messageBus: MessageBus,
+          currentUser,
+          siteSettings: this.siteSettings,
+        });
+        trackingState.establishChannels();
+      });
+
+      test("topics in muted categories do not get added to the state", function (assert) {
+        trackingState.currentUser.set("muted_category_ids", [123]);
+        publishToMessageBus("/new", newTopicPayload);
+        assert.equal(
+          trackingState.findState(222),
+          null,
+          "the new topic is not in the state"
+        );
+      });
+
+      test("topics in muted tags do not get added to the state", function (assert) {
+        trackingState.currentUser.set("muted_tag_ids", [44]);
+        publishToMessageBus("/new", newTopicPayload);
+        assert.equal(
+          trackingState.findState(222),
+          null,
+          "the new topic is not in the state"
+        );
+      });
+
+      test("message count is incremented and callback is called", function (assert) {
+        let messageIncrementCalled = false;
+        trackingState.onMessageIncrement(() => {
+          messageIncrementCalled = true;
+        });
+        publishToMessageBus("/new", newTopicPayload);
+        assert.equal(
+          trackingState.messageCount,
+          1,
+          "message count incremented"
+        );
+        assert.equal(
+          messageIncrementCalled,
+          true,
+          "message increment callback called"
+        );
+      });
+
+      test("state is modified and callback is called", function (assert) {
+        let stateCallbackCalled = false;
+        trackingState.onStateChange(() => {
+          stateCallbackCalled = true;
+        });
+        publishToMessageBus("/new", newTopicPayload);
+        assert.deepEqual(
+          trackingState.findState(222),
+          {
+            topic_id: 222,
+            category_id: 123,
+            topic_tag_ids: [44],
+            tags: ["pending"],
+            last_read_post_number: null,
+            highest_post_number: 1,
+            created_at: "2012-11-31 12:00:00 UTC",
+            archetype: "regular",
+          },
+          "new topic loaded into state"
+        );
+        assert.equal(stateCallbackCalled, true, "state change callback called");
+      });
+
+      test("adds incoming so it is counted in topic lists", function (assert) {
+        trackingState.trackIncoming("all");
+        publishToMessageBus("/new", newTopicPayload);
+        assert.deepEqual(
+          trackingState.newIncoming,
+          [222],
+          "new topic is incoming"
+        );
+        assert.equal(
+          trackingState.incomingCount,
+          1,
+          "incoming count is increased"
+        );
+      });
+    }
+  );
+
+  test("establishChannels - /delete MessageBus channel payloads processed", function (assert) {
+    const trackingState = TopicTrackingState.create({ messageBus: MessageBus });
+    trackingState.establishChannels();
+
+    trackingState.loadStates([
+      {
+        topic_id: 111,
+        deleted: false,
+      },
+    ]);
+
+    publishToMessageBus("/delete", { topic_id: 111 });
+
+    assert.equal(
+      trackingState.findState(111).deleted,
+      true,
+      "marks the topic as deleted"
+    );
+    assert.equal(trackingState.messageCount, 1, "increments message count");
+  });
+
+  test("establishChannels - /recover MessageBus channel payloads processed", function (assert) {
+    const trackingState = TopicTrackingState.create({ messageBus: MessageBus });
+    trackingState.establishChannels();
+
+    trackingState.loadStates([
+      {
+        topic_id: 111,
+        deleted: true,
+      },
+    ]);
+
+    publishToMessageBus("/recover", { topic_id: 111 });
+
+    assert.equal(
+      trackingState.findState(111).deleted,
+      false,
+      "marks the topic as not deleted"
+    );
+    assert.equal(trackingState.messageCount, 1, "increments message count");
+  });
+
+  test("establishChannels - /destroy MessageBus channel payloads processed", function (assert) {
+    sinon.stub(DiscourseURL, "router").value({
+      currentRoute: { parent: { name: "topic", params: { id: 111 } } },
+    });
+    sinon.stub(DiscourseURL, "redirectTo");
+
+    const trackingState = TopicTrackingState.create({ messageBus: MessageBus });
+    trackingState.establishChannels();
+    trackingState.loadStates([
+      {
+        topic_id: 111,
+        deleted: false,
+      },
+    ]);
+
+    publishToMessageBus("/destroy", { topic_id: 111 });
+
+    assert.equal(trackingState.messageCount, 1, "increments message count");
+    assert.ok(
+      DiscourseURL.redirectTo.calledWith("/"),
+      "redirect to / because topic is destroyed"
+    );
+  });
+
   test("subscribe to category", function (assert) {
     const store = createStore();
     const darth = store.createRecord("category", { id: 1, slug: "darth" }),
@@ -176,53 +762,53 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
 
     sinon.stub(Category, "list").returns(categoryList);
 
-    const state = TopicTrackingState.create();
+    const trackingState = TopicTrackingState.create();
 
-    state.trackIncoming("c/darth/1/l/latest");
+    trackingState.trackIncoming("c/darth/1/l/latest");
 
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 1,
       payload: { category_id: 2, topic_id: 1 },
     });
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 2,
       payload: { category_id: 3, topic_id: 2 },
     });
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 3,
       payload: { category_id: 1, topic_id: 3 },
     });
 
     assert.equal(
-      state.get("incomingCount"),
+      trackingState.get("incomingCount"),
       2,
       "expect to properly track incoming for category"
     );
 
-    state.resetTracking();
-    state.trackIncoming("c/darth/luke/2/l/latest");
+    trackingState.resetTracking();
+    trackingState.trackIncoming("c/darth/luke/2/l/latest");
 
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 1,
       payload: { category_id: 2, topic_id: 1 },
     });
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 2,
       payload: { category_id: 3, topic_id: 2 },
     });
-    state.notify({
+    trackingState.notifyIncoming({
       message_type: "new_topic",
       topic_id: 3,
       payload: { category_id: 1, topic_id: 3 },
     });
 
     assert.equal(
-      state.get("incomingCount"),
+      trackingState.get("incomingCount"),
       1,
       "expect to properly track incoming for subcategory"
     );
@@ -243,10 +829,10 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
     });
     sinon.stub(Category, "list").returns([foo, bar, baz]);
 
-    const state = TopicTrackingState.create();
-    assert.deepEqual(Array.from(state.getSubCategoryIds(1)), [1, 2, 3]);
-    assert.deepEqual(Array.from(state.getSubCategoryIds(2)), [2, 3]);
-    assert.deepEqual(Array.from(state.getSubCategoryIds(3)), [3]);
+    const trackingState = TopicTrackingState.create();
+    assert.deepEqual(Array.from(trackingState.getSubCategoryIds(1)), [1, 2, 3]);
+    assert.deepEqual(Array.from(trackingState.getSubCategoryIds(2)), [2, 3]);
+    assert.deepEqual(Array.from(trackingState.getSubCategoryIds(3)), [3]);
   });
 
   test("countNew", function (assert) {
@@ -276,79 +862,59 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       muted_category_ids: [4],
     });
 
-    const state = TopicTrackingState.create({ currentUser });
+    const trackingState = TopicTrackingState.create({ currentUser });
 
-    assert.equal(state.countNew(1), 0);
-    assert.equal(state.countNew(2), 0);
-    assert.equal(state.countNew(3), 0);
+    assert.equal(trackingState.countNew(1), 0);
+    assert.equal(trackingState.countNew(2), 0);
+    assert.equal(trackingState.countNew(3), 0);
 
-    state.states["t112"] = {
+    trackingState.states.set("t112", {
       last_read_post_number: null,
       id: 112,
       notification_level: NotificationLevels.TRACKING,
       category_id: 2,
-    };
+      created_in_new_period: true,
+    });
 
-    assert.equal(state.countNew(1), 1);
-    assert.equal(state.countNew(1, undefined, true), 0);
-    assert.equal(state.countNew(1, "missing-tag"), 0);
-    assert.equal(state.countNew(2), 1);
-    assert.equal(state.countNew(3), 0);
+    assert.equal(trackingState.countNew(1), 1);
+    assert.equal(trackingState.countNew(1, undefined, true), 0);
+    assert.equal(trackingState.countNew(1, "missing-tag"), 0);
+    assert.equal(trackingState.countNew(2), 1);
+    assert.equal(trackingState.countNew(3), 0);
 
-    state.states["t113"] = {
+    trackingState.states.set("t113", {
       last_read_post_number: null,
       id: 113,
       notification_level: NotificationLevels.TRACKING,
       category_id: 3,
       tags: ["amazing"],
-    };
+      created_in_new_period: true,
+    });
 
-    assert.equal(state.countNew(1), 2);
-    assert.equal(state.countNew(2), 2);
-    assert.equal(state.countNew(3), 1);
-    assert.equal(state.countNew(3, "amazing"), 1);
-    assert.equal(state.countNew(3, "missing"), 0);
+    assert.equal(trackingState.countNew(1), 2);
+    assert.equal(trackingState.countNew(2), 2);
+    assert.equal(trackingState.countNew(3), 1);
+    assert.equal(trackingState.countNew(3, "amazing"), 1);
+    assert.equal(trackingState.countNew(3, "missing"), 0);
 
-    state.states["t111"] = {
+    trackingState.states.set("t111", {
       last_read_post_number: null,
       id: 111,
       notification_level: NotificationLevels.TRACKING,
       category_id: 1,
-    };
+      created_in_new_period: true,
+    });
 
-    assert.equal(state.countNew(1), 3);
-    assert.equal(state.countNew(2), 2);
-    assert.equal(state.countNew(3), 1);
+    assert.equal(trackingState.countNew(1), 3);
+    assert.equal(trackingState.countNew(2), 2);
+    assert.equal(trackingState.countNew(3), 1);
 
-    state.states["t115"] = {
+    trackingState.states.set("t115", {
       last_read_post_number: null,
       id: 115,
       category_id: 4,
-    };
-    assert.equal(state.countNew(4), 0);
-  });
-
-  test("dismissNew", function (assert) {
-    let currentUser = User.create({
-      username: "chuck",
     });
-
-    const state = TopicTrackingState.create({ currentUser });
-
-    state.states["t112"] = {
-      last_read_post_number: null,
-      id: 112,
-      notification_level: NotificationLevels.TRACKING,
-      category_id: 1,
-      is_seen: false,
-      tags: ["foo"],
-    };
-
-    state.dismissNewTopic({
-      message_type: "dismiss_new",
-      payload: { topic_ids: [112] },
-    });
-    assert.equal(state.states["t112"].is_seen, true);
+    assert.equal(trackingState.countNew(4), 0);
   });
 
   test("mute and unmute topic", function (assert) {
@@ -357,21 +923,27 @@ module("Unit | Model | topic-tracking-state", function (hooks) {
       muted_category_ids: [],
     });
 
-    const state = TopicTrackingState.create({ currentUser });
+    const trackingState = TopicTrackingState.create({ currentUser });
 
-    state.trackMutedOrUnmutedTopic({ topic_id: 1, message_type: "muted" });
+    trackingState.trackMutedOrUnmutedTopic({
+      topic_id: 1,
+      message_type: "muted",
+    });
     assert.equal(currentUser.muted_topics[0].topicId, 1);
 
-    state.trackMutedOrUnmutedTopic({ topic_id: 2, message_type: "unmuted" });
+    trackingState.trackMutedOrUnmutedTopic({
+      topic_id: 2,
+      message_type: "unmuted",
+    });
     assert.equal(currentUser.unmuted_topics[0].topicId, 2);
 
-    state.pruneOldMutedAndUnmutedTopics();
-    assert.equal(state.isMutedTopic(1), true);
-    assert.equal(state.isUnmutedTopic(2), true);
+    trackingState.pruneOldMutedAndUnmutedTopics();
+    assert.equal(trackingState.isMutedTopic(1), true);
+    assert.equal(trackingState.isUnmutedTopic(2), true);
 
     this.clock.tick(60000);
-    state.pruneOldMutedAndUnmutedTopics();
-    assert.equal(state.isMutedTopic(1), false);
-    assert.equal(state.isUnmutedTopic(2), false);
+    trackingState.pruneOldMutedAndUnmutedTopics();
+    assert.equal(trackingState.isMutedTopic(1), false);
+    assert.equal(trackingState.isUnmutedTopic(2), false);
   });
 });
diff --git a/app/controllers/application_controller.rb b/app/controllers/application_controller.rb
index d75586a458f..abae103a2ec 100644
--- a/app/controllers/application_controller.rb
+++ b/app/controllers/application_controller.rb
@@ -608,7 +608,9 @@ class ApplicationController < ActionController::Base
   def preload_current_user_data
     store_preloaded("currentUser", MultiJson.dump(CurrentUserSerializer.new(current_user, scope: guardian, root: false)))
     report = TopicTrackingState.report(current_user)
-    serializer = ActiveModel::ArraySerializer.new(report, each_serializer: TopicTrackingStateSerializer)
+    serializer = ActiveModel::ArraySerializer.new(
+      report, each_serializer: TopicTrackingStateSerializer, scope: guardian
+    )
     store_preloaded("topicTrackingStates", MultiJson.dump(serializer))
   end
 
diff --git a/app/controllers/topics_controller.rb b/app/controllers/topics_controller.rb
index 4e66f3e7889..010c51bd7f3 100644
--- a/app/controllers/topics_controller.rb
+++ b/app/controllers/topics_controller.rb
@@ -919,7 +919,7 @@ class TopicsController < ApplicationController
       topic_ids = params[:topic_ids].map { |t| t.to_i }
     elsif params[:filter] == 'unread'
       tq = TopicQuery.new(current_user)
-      topics = TopicQuery.unread_filter(tq.joined_topic_user, current_user.id, staff: guardian.is_staff?).listable_topics
+      topics = TopicQuery.unread_filter(tq.joined_topic_user, staff: guardian.is_staff?).listable_topics
       topics = TopicQuery.tracked_filter(topics, current_user.id) if params[:tracked].to_s == "true"
 
       if params[:category_id]
diff --git a/app/controllers/users_controller.rb b/app/controllers/users_controller.rb
index 5d7a37e9f98..8c7171be171 100644
--- a/app/controllers/users_controller.rb
+++ b/app/controllers/users_controller.rb
@@ -308,7 +308,9 @@ class UsersController < ApplicationController
     guardian.ensure_can_edit!(user)
 
     report = TopicTrackingState.report(user)
-    serializer = ActiveModel::ArraySerializer.new(report, each_serializer: TopicTrackingStateSerializer)
+    serializer = ActiveModel::ArraySerializer.new(
+      report, each_serializer: TopicTrackingStateSerializer, scope: guardian
+    )
 
     render json: MultiJson.dump(serializer)
   end
diff --git a/app/models/topic_tracking_state.rb b/app/models/topic_tracking_state.rb
index 52dba4823d6..4e03dc6661b 100644
--- a/app/models/topic_tracking_state.rb
+++ b/app/models/topic_tracking_state.rb
@@ -1,10 +1,20 @@
 # 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 the massage bus
+# 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
-
+# 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
@@ -13,6 +23,13 @@ class TopicTrackingState
   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"
+  MAX_TOPICS = 5000
 
   attr_accessor :user_id,
                 :topic_id,
@@ -20,20 +37,15 @@ class TopicTrackingState
                 :last_read_post_number,
                 :created_at,
                 :category_id,
-                :notification_level
+                :notification_level,
+                :tags
 
   def self.publish_new(topic)
     return unless topic.regular?
 
-    tags, tag_ids = nil
+    tag_ids, tags = nil
     if SiteSetting.tagging_enabled
-      topic.tags.pluck(:id, :name).each do |id, name|
-        tags ||= []
-        tag_ids ||= []
-
-        tags << name
-        tag_ids << id
-      end
+      tag_ids, tags = topic.tags.pluck(:id, :name).transpose
     end
 
     payload = {
@@ -43,6 +55,7 @@ class TopicTrackingState
       topic_id: topic.id,
       category_id: topic.category_id,
       archetype: topic.archetype,
+      created_in_new_period: true
     }
 
     if tags
@@ -52,7 +65,7 @@ class TopicTrackingState
 
     message = {
       topic_id: topic.id,
-      message_type: "new_topic",
+      message_type: NEW_TOPIC_MESSAGE_TYPE,
       payload: payload
     }
 
@@ -65,17 +78,26 @@ class TopicTrackingState
   def self.publish_latest(topic, staff_only = false)
     return unless topic.regular?
 
+    tag_ids, tags = nil
+    if SiteSetting.tagging_enabled
+      tag_ids, tags = topic.tags.pluck(:id, :name).transpose
+    end
+
     message = {
       topic_id: topic.id,
       message_type: LATEST_MESSAGE_TYPE,
       payload: {
         bumped_at: topic.bumped_at,
         category_id: topic.category_id,
-        archetype: topic.archetype,
-        topic_tag_ids: topic.tags.pluck(:id)
+        archetype: topic.archetype
       }
     }
 
+    if tags
+      message[:payload][:tags] = tags
+      message[:payload][:topic_tag_ids] = tag_ids
+    end
+
     group_ids =
       if staff_only
         [Group::AUTO_GROUPS[:staff]]
@@ -133,25 +155,33 @@ class TopicTrackingState
       end
 
     tags = nil
+    tag_ids = nil
     if include_tags_in_report?
-      tags = post.topic.tags.pluck(:name)
+      tag_ids, tags = post.topic.tags.pluck(:id, :name).transpose
     end
 
     TopicUser
       .tracking(post.topic_id)
+      .includes(user: :user_stat)
       .select([:user_id, :last_read_post_number, :notification_level])
       .each do |tu|
 
       payload = {
         last_read_post_number: tu.last_read_post_number,
         highest_post_number: post.post_number,
+        updated_at: post.topic.updated_at,
         created_at: post.created_at,
         category_id: post.topic.category_id,
         notification_level: tu.notification_level,
-        archetype: post.topic.archetype
+        archetype: post.topic.archetype,
+        first_unread_at: tu.user.user_stat.first_unread_at,
+        unread_not_too_old: true
       }
 
-      payload[:tags] = tags if tags
+      if tags
+        payload[:tags] = tags
+        payload[:topic_tag_ids] = tag_ids
+      end
 
       message = {
         topic_id: post.topic_id,
@@ -169,7 +199,7 @@ class TopicTrackingState
 
     message = {
       topic_id: topic.id,
-      message_type: "recover"
+      message_type: RECOVER_MESSAGE_TYPE
     }
 
     MessageBus.publish("/recover", message.as_json, group_ids: group_ids)
@@ -181,7 +211,7 @@ class TopicTrackingState
 
     message = {
       topic_id: topic.id,
-      message_type: "delete"
+      message_type: DELETE_MESSAGE_TYPE
     }
 
     MessageBus.publish("/delete", message.as_json, group_ids: group_ids)
@@ -192,7 +222,7 @@ class TopicTrackingState
 
     message = {
       topic_id: topic.id,
-      message_type: "destroy"
+      message_type: DESTROY_MESSAGE_TYPE
     }
 
     MessageBus.publish("/destroy", message.as_json, group_ids: group_ids)
@@ -203,7 +233,7 @@ class TopicTrackingState
 
     message = {
       topic_id: topic_id,
-      message_type: "read",
+      message_type: READ_MESSAGE_TYPE,
       payload: {
         last_read_post_number: last_read_post_number,
         highest_post_number: highest_post_number,
@@ -217,7 +247,7 @@ class TopicTrackingState
 
   def self.publish_dismiss_new(user_id, topic_ids: [])
     message = {
-      message_type: "dismiss_new",
+      message_type: DISMISS_NEW_MESSAGE_TYPE,
       payload: {
         topic_ids: topic_ids
       }
@@ -225,6 +255,18 @@ class TopicTrackingState
     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(staff: false)
+    TopicQuery.unread_filter(Topic, staff: staff).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
@@ -247,18 +289,32 @@ class TopicTrackingState
     @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)
-    # 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
     tag_ids = muted_tag_ids(user)
+    sql = new_and_unread_sql(topic_id, user, tag_ids)
+    sql = tags_included_wrapped_sql(sql)
 
-    sql = +report_raw_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
+    )
+
+    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,
@@ -280,74 +336,80 @@ class TopicTrackingState
       user: user,
       muted_tag_ids: tag_ids
     )
+  end
 
+  def self.tags_included_wrapped_sql(sql)
     if SiteSetting.tagging_enabled && TopicTrackingState.include_tags_in_report?
-      sql = <<~SQL
-        WITH X AS (#{sql})
+      return <<~SQL
+        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 = X.topic_id
+             WHERE topic_id = tags_included_cte.topic_id
           ) tags
-        FROM X
+        FROM tags_included_cte
       SQL
     end
 
-    DB.query(
-      sql,
-        user_id: user.id,
-        topic_id: topic_id,
-        min_new_topic_date: Time.at(SiteSetting.min_new_topics_time).to_datetime
-    )
+    sql
   end
 
   def self.muted_tag_ids(user)
     TagUser.lookup(user, :muted).pluck(:tag_id)
   end
 
-  def self.report_raw_sql(opts = nil)
-    opts ||= {}
-
+  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,
+    select: nil,
+    custom_state_filter: nil
+  )
     unread =
-      if opts[:skip_unread]
+      if skip_unread
         "1=0"
       else
-        TopicQuery
-          .unread_filter(Topic, -999, staff: opts && opts[:staff])
-          .where_clause.ast.to_sql
-          .gsub("-999", ":user_id")
+        unread_filter_sql
       end
 
-    filter_old_unread =
-      if opts[:filter_old_unread]
+    filter_old_unread_sql =
+      if filter_old_unread
         " topics.updated_at >= us.first_unread_at AND "
       else
         ""
       end
 
     new =
-      if opts[:skip_new]
+      if skip_new
         "1=0"
       else
-        TopicQuery.new_filter(Topic, "xxx").where_clause.ast.to_sql.gsub!("'xxx'", treat_as_new_topic_clause) +
-          " AND topics.created_at > :min_new_topic_date" +
-          " AND dismissed_topic_users.id IS NULL"
+        new_filter_sql
       end
 
-    select = (opts[:select]) || "
-           u.id AS user_id,
-           topics.id AS topic_id,
+    select_sql = select || "
+           u.id as user_id,
+           topics.id as topic_id,
            topics.created_at,
-           #{opts[:staff] ? "highest_staff_post_number highest_post_number" : "highest_post_number"},
+           topics.updated_at,
+           #{staff ? "highest_staff_post_number highest_post_number" : "highest_post_number"},
            last_read_post_number,
-           c.id AS category_id,
-           tu.notification_level"
+           c.id as category_id,
+           tu.notification_level,
+           us.first_unread_at"
 
     category_filter =
-      if opts[:admin]
+      if admin
         ""
       else
-        append = "OR u.admin" if !opts.key?(:admin)
+        append = "OR u.admin" if !admin
         <<~SQL
           (
            NOT c.read_restricted #{append} OR c.id IN (
@@ -360,18 +422,18 @@ class TopicTrackingState
       end
 
     visibility_filter =
-      if opts[:staff]
+      if staff
         ""
       else
-        append = "OR u.admin OR u.moderator" if !opts.key?(:staff)
+        append = "OR u.admin OR u.moderator" if !staff
         "(topics.visible #{append}) AND"
       end
 
     tags_filter = ""
 
-    if (muted_tag_ids = opts[:muted_tag_ids]).present? && ['always', 'only_muted'].include?(SiteSetting.remove_muted_tags_from_latest)
+    if muted_tag_ids.present? && ['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[#{opts[:muted_tag_ids].join(',')}]"
+      muted_tags_array_sql = "ARRAY[#{muted_tag_ids.join(',')}]"
 
       if SiteSetting.remove_muted_tags_from_latest == 'always'
         tags_filter = <<~SQL
@@ -389,34 +451,37 @@ class TopicTrackingState
     end
 
     sql = +<<~SQL
-    SELECT #{select}
-    FROM topics
-    JOIN users u on u.id = :user_id
-    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
-    LEFT JOIN category_users ON category_users.category_id = topics.category_id AND category_users.user_id = #{opts[:user].id}
-    LEFT JOIN dismissed_topic_users ON dismissed_topic_users.topic_id = topics.id AND dismissed_topic_users.user_id = #{opts[:user].id}
-    WHERE u.id = :user_id AND
-          #{filter_old_unread}
-          topics.archetype <> 'private_message' AND
-          ((#{unread}) OR (#{new})) AND
-          #{visibility_filter}
-          #{tags_filter}
-          topics.deleted_at IS NULL AND
-          #{category_filter}
-          NOT (
-            last_read_post_number IS NULL AND
-            COALESCE(category_users.notification_level, #{CategoryUser.default_notification_level}) = #{CategoryUser.notification_levels[:muted]}
-          )
-SQL
+      SELECT #{select_sql}
+      FROM topics
+      JOIN users u on u.id = :user_id
+      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
+      LEFT JOIN category_users ON category_users.category_id = topics.category_id AND category_users.user_id = #{user.id}
+      LEFT JOIN dismissed_topic_users ON dismissed_topic_users.topic_id = topics.id AND dismissed_topic_users.user_id = #{user.id}
+      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"}
+              (
+                COALESCE(category_users.notification_level, #{CategoryUser.default_notification_level}) = #{CategoryUser.notification_levels[:muted]}
+                AND tu.notification_level <= #{TopicUser.notification_levels[:regular]}
+              )
+            )
+    SQL
 
-    if opts[:topic_id]
+    if topic_id
       sql << " AND topics.id = :topic_id"
     end
 
-    unless opts[:skip_order]
+    unless skip_order
       sql << " ORDER BY topics.bumped_at DESC"
     end
 
diff --git a/app/serializers/current_user_serializer.rb b/app/serializers/current_user_serializer.rb
index 6b6f015876e..8d7965819f5 100644
--- a/app/serializers/current_user_serializer.rb
+++ b/app/serializers/current_user_serializer.rb
@@ -44,6 +44,7 @@ class CurrentUserSerializer < BasicUserSerializer
              :read_faq,
              :automatically_unpin_topics,
              :mailing_list_mode,
+             :treat_as_new_topic_start_date,
              :previous_visit_at,
              :seen_notification_id,
              :primary_group_id,
@@ -278,6 +279,10 @@ class CurrentUserSerializer < BasicUserSerializer
     object.user_option.mailing_list_mode
   end
 
+  def treat_as_new_topic_start_date
+    object.user_option.treat_as_new_topic_start_date
+  end
+
   def skip_new_user_tips
     object.user_option.skip_new_user_tips
   end
diff --git a/app/serializers/topic_tracking_state_serializer.rb b/app/serializers/topic_tracking_state_serializer.rb
index b200ca82e30..524d3ec5731 100644
--- a/app/serializers/topic_tracking_state_serializer.rb
+++ b/app/serializers/topic_tracking_state_serializer.rb
@@ -6,5 +6,17 @@ class TopicTrackingStateSerializer < ApplicationSerializer
              :last_read_post_number,
              :created_at,
              :category_id,
-             :notification_level
+             :notification_level,
+             :created_in_new_period,
+             :unread_not_too_old
+
+  def created_in_new_period
+    return true if !scope
+    object.created_at >= scope.user.user_option.treat_as_new_topic_start_date
+  end
+
+  def unread_not_too_old
+    return true if object.first_unread_at.blank?
+    object.updated_at >= object.first_unread_at
+  end
 end
diff --git a/lib/topic_query.rb b/lib/topic_query.rb
index d769c1d1d1f..5a648b9f270 100644
--- a/lib/topic_query.rb
+++ b/lib/topic_query.rb
@@ -331,7 +331,6 @@ class TopicQuery
 
     list = TopicQuery.unread_filter(
       list,
-      user.id,
       staff: user.staff?
     )
 
@@ -379,14 +378,20 @@ class TopicQuery
     end
   end
 
-  def self.new_filter(list, treat_as_new_topic_start_date)
-    list.where("topics.created_at >= :created_at", created_at: treat_as_new_topic_start_date)
+  def self.new_filter(list, treat_as_new_topic_start_date: nil, treat_as_new_topic_clause_sql: nil)
+    if treat_as_new_topic_start_date
+      list = list.where("topics.created_at >= :created_at", created_at: treat_as_new_topic_start_date)
+    else
+      list = list.where("topics.created_at >= #{treat_as_new_topic_clause_sql}")
+    end
+
+    list
       .where("tu.last_read_post_number IS NULL")
       .where("COALESCE(tu.notification_level, :tracking) >= :tracking", tracking: TopicUser.notification_levels[:tracking])
   end
 
-  def self.unread_filter(list, user_id, opts)
-    col_name = opts[:staff] ? "highest_staff_post_number" : "highest_post_number"
+  def self.unread_filter(list, staff: false)
+    col_name = staff ? "highest_staff_post_number" : "highest_post_number"
 
     list
       .where("tu.last_read_post_number < topics.#{col_name}")
@@ -516,7 +521,6 @@ class TopicQuery
   def unread_results(options = {})
     result = TopicQuery.unread_filter(
         default_results(options.reverse_merge(unordered: true)),
-        @user&.id,
         staff: @user&.staff?)
       .order('CASE WHEN topics.user_id = tu.user_id THEN 1 ELSE 2 END')
 
@@ -547,7 +551,10 @@ class TopicQuery
   def new_results(options = {})
     # TODO does this make sense or should it be ordered on created_at
     #  it is ordering on bumped_at now
-    result = TopicQuery.new_filter(default_results(options.reverse_merge(unordered: true)), @user.user_option.treat_as_new_topic_start_date)
+    result = TopicQuery.new_filter(
+      default_results(options.reverse_merge(unordered: true)),
+      treat_as_new_topic_start_date: @user.user_option.treat_as_new_topic_start_date
+    )
     result = remove_muted_topics(result, @user)
     result = remove_muted_categories(result, @user, exclude: options[:category])
     result = remove_muted_tags(result, @user, options)
@@ -979,14 +986,16 @@ class TopicQuery
 
   def new_messages(params)
     TopicQuery
-      .new_filter(messages_for_groups_or_user(params[:my_group_ids]), Time.at(SiteSetting.min_new_topics_time).to_datetime)
+      .new_filter(
+        messages_for_groups_or_user(params[:my_group_ids]),
+        treat_as_new_topic_start_date: Time.at(SiteSetting.min_new_topics_time).to_datetime
+      )
       .limit(params[:count])
   end
 
   def unread_messages(params)
     query = TopicQuery.unread_filter(
       messages_for_groups_or_user(params[:my_group_ids]),
-      @user.id,
       staff: @user.staff?
     )