diff --git a/app/assets/javascripts/discourse/app/controllers/user-private-messages.js b/app/assets/javascripts/discourse/app/controllers/user-private-messages.js
index f50f9bad576..2ff2c1fc95e 100644
--- a/app/assets/javascripts/discourse/app/controllers/user-private-messages.js
+++ b/app/assets/javascripts/discourse/app/controllers/user-private-messages.js
@@ -66,12 +66,28 @@ export default Controller.extend({
     return pmView === VIEW_NAME_WARNINGS && !viewingSelf && !isAdmin;
   },
 
-  @discourseComputed("model.groups")
-  inboxes(groups) {
-    const groupsWithMessages = groups?.filter((group) => {
-      return group.has_messages;
-    });
+  @discourseComputed("pmTopicTrackingState.newIncoming.[]", "selectedInbox")
+  newLinkText() {
+    return this._linkText("new");
+  },
 
+  @discourseComputed("selectedInbox", "pmTopicTrackingState.newIncoming.[]")
+  unreadLinkText() {
+    return this._linkText("unread");
+  },
+
+  _linkText(type) {
+    const count = this.pmTopicTrackingState?.lookupCount(type) || 0;
+
+    if (count === 0) {
+      return I18n.t(`user.messages.${type}`);
+    } else {
+      return I18n.t(`user.messages.${type}_with_count`, { count });
+    }
+  },
+
+  @discourseComputed("model.groupsWithMessages")
+  inboxes(groupsWithMessages) {
     if (!groupsWithMessages || groupsWithMessages.length === 0) {
       return [];
     }
diff --git a/app/assets/javascripts/discourse/app/controllers/user-topics-list.js b/app/assets/javascripts/discourse/app/controllers/user-topics-list.js
index 4b3adfba487..195adb4fb00 100644
--- a/app/assets/javascripts/discourse/app/controllers/user-topics-list.js
+++ b/app/assets/javascripts/discourse/app/controllers/user-topics-list.js
@@ -1,8 +1,6 @@
 import Controller, { inject as controller } from "@ember/controller";
-import discourseComputed, {
-  observes,
-  on,
-} from "discourse-common/utils/decorators";
+import discourseComputed, { observes } from "discourse-common/utils/decorators";
+import { reads } from "@ember/object/computed";
 import BulkTopicSelection from "discourse/mixins/bulk-topic-selection";
 import { action } from "@ember/object";
 import Topic from "discourse/models/topic";
@@ -18,14 +16,9 @@ export default Controller.extend(BulkTopicSelection, {
 
   hideCategory: false,
   showPosters: false,
-  incomingCount: 0,
   channel: null,
   tagsForUser: null,
-
-  @on("init")
-  _initialize() {
-    this.newIncoming = [];
-  },
+  pmTopicTrackingState: null,
 
   saveScrollPosition() {
     this.session.set("topicListScrollPosition", $(window).scrollTop());
@@ -36,10 +29,7 @@ export default Controller.extend(BulkTopicSelection, {
     this.set("application.showFooter", !this.get("model.canLoadMore"));
   },
 
-  @discourseComputed("incomingCount")
-  hasIncoming(incomingCount) {
-    return incomingCount > 0;
-  },
+  incomingCount: reads("pmTopicTrackingState.newIncoming.length"),
 
   @discourseComputed("filter", "model.topics.length")
   showResetNew(filter, hasTopics) {
@@ -51,31 +41,16 @@ export default Controller.extend(BulkTopicSelection, {
     return filter === UNREAD_FILTER && hasTopics;
   },
 
-  subscribe(channel) {
-    this.set("channel", channel);
-
-    this.messageBus.subscribe(channel, (data) => {
-      if (this.newIncoming.indexOf(data.topic_id) === -1) {
-        this.newIncoming.push(data.topic_id);
-        this.incrementProperty("incomingCount");
-      }
-    });
+  subscribe() {
+    this.pmTopicTrackingState?.trackIncoming(
+      this.inbox,
+      this.filter,
+      this.group
+    );
   },
 
   unsubscribe() {
-    const channel = this.channel;
-    if (channel) {
-      this.messageBus.unsubscribe(channel);
-    }
-    this._resetTracking();
-    this.set("channel", null);
-  },
-
-  _resetTracking() {
-    this.setProperties({
-      newIncoming: [],
-      incomingCount: 0,
-    });
+    this.pmTopicTrackingState?.resetTracking();
   },
 
   @action
@@ -105,8 +80,8 @@ export default Controller.extend(BulkTopicSelection, {
 
   @action
   showInserted() {
-    this.model.loadBefore(this.newIncoming);
-    this._resetTracking();
+    this.model.loadBefore(this.pmTopicTrackingState.newIncoming);
+    this.pmTopicTrackingState.resetTracking();
     return false;
   },
 });
diff --git a/app/assets/javascripts/discourse/app/models/private-message-topic-tracking-state.js b/app/assets/javascripts/discourse/app/models/private-message-topic-tracking-state.js
new file mode 100644
index 00000000000..4fcb8f96925
--- /dev/null
+++ b/app/assets/javascripts/discourse/app/models/private-message-topic-tracking-state.js
@@ -0,0 +1,196 @@
+import EmberObject from "@ember/object";
+import {
+  ARCHIVE_FILTER,
+  INBOX_FILTER,
+  NEW_FILTER,
+  UNREAD_FILTER,
+} from "discourse/routes/build-private-messages-route";
+import { NotificationLevels } from "discourse/lib/notification-levels";
+
+// See private_message_topic_tracking_state.rb for documentation
+const PrivateMessageTopicTrackingState = EmberObject.extend({
+  CHANNEL_PREFIX: "/private-message-topic-tracking-state",
+
+  inbox: null,
+  filter: null,
+  activeGroup: null,
+
+  startTracking(data) {
+    this.states = new Map();
+    this.newIncoming = [];
+    this._loadStates(data);
+    this.establishChannels();
+  },
+
+  establishChannels() {
+    this.messageBus.subscribe(
+      this._userChannel(this.user.id),
+      this._processMessage.bind(this)
+    );
+
+    this.user.groupsWithMessages?.forEach((group) => {
+      this.messageBus.subscribe(
+        this._groupChannel(group.id),
+        this._processMessage.bind(this)
+      );
+    });
+  },
+
+  stopTracking() {
+    this.messageBus.unsubscribe(this._userChannel(this.user.id));
+
+    this.user.groupsWithMessages?.forEach((group) => {
+      this.messageBus.unsubscribe(this._groupChannel(group.id));
+    });
+  },
+
+  lookupCount(type) {
+    const typeFilterFn = type === "new" ? this._isNew : this._isUnread;
+    let filterFn;
+
+    if (this.inbox === "user") {
+      filterFn = this._isPersonal.bind(this);
+    } else if (this.inbox === "group") {
+      filterFn = this._isGroup.bind(this);
+    }
+
+    return Array.from(this.states.values()).filter((topic) => {
+      return typeFilterFn(topic) && (!filterFn || filterFn(topic));
+    }).length;
+  },
+
+  trackIncoming(inbox, filter, group) {
+    this.setProperties({ inbox, filter, activeGroup: group });
+  },
+
+  resetTracking() {
+    if (this.inbox) {
+      this.set("newIncoming", []);
+    }
+  },
+
+  _userChannel(userId) {
+    return `${this.CHANNEL_PREFIX}/user/${userId}`;
+  },
+
+  _groupChannel(groupId) {
+    return `${this.CHANNEL_PREFIX}/group/${groupId}`;
+  },
+
+  _isNew(topic) {
+    return (
+      !topic.last_read_post_number &&
+      ((topic.notification_level !== 0 && !topic.notification_level) ||
+        topic.notification_level >= NotificationLevels.TRACKING) &&
+      !topic.is_seen
+    );
+  },
+
+  _isUnread(topic) {
+    return (
+      topic.last_read_post_number &&
+      topic.last_read_post_number < topic.highest_post_number &&
+      topic.notification_level >= NotificationLevels.TRACKING
+    );
+  },
+
+  _isPersonal(topic) {
+    const groups = this.user.groups;
+
+    if (groups.length === 0) {
+      return true;
+    }
+
+    return !groups.some((group) => {
+      return topic.group_ids?.includes(group.id);
+    });
+  },
+
+  _isGroup(topic) {
+    return this.user.groups.some((group) => {
+      return (
+        group.name === this.activeGroup.name &&
+        topic.group_ids?.includes(group.id)
+      );
+    });
+  },
+
+  _processMessage(message) {
+    switch (message.message_type) {
+      case "new_topic":
+        this._modifyState(message.topic_id, message.payload);
+
+        if (
+          [NEW_FILTER, INBOX_FILTER].includes(this.filter) &&
+          this._shouldDisplayMessageForInbox(message)
+        ) {
+          this._notifyIncoming(message.topic_id);
+        }
+
+        break;
+      case "unread":
+        this._modifyState(message.topic_id, message.payload);
+
+        if (
+          [UNREAD_FILTER, INBOX_FILTER].includes(this.filter) &&
+          this._shouldDisplayMessageForInbox(message)
+        ) {
+          this._notifyIncoming(message.topic_id);
+        }
+
+        break;
+      case "archive":
+        if (
+          [INBOX_FILTER, ARCHIVE_FILTER].includes(this.filter) &&
+          ["user", "all"].includes(this.inbox)
+        ) {
+          this._notifyIncoming(message.topic_id);
+        }
+        break;
+      case "group_archive":
+        if (
+          [INBOX_FILTER, ARCHIVE_FILTER].includes(this.filter) &&
+          (this.inbox === "all" || this._displayMessageForGroupInbox(message))
+        ) {
+          this._notifyIncoming(message.topic_id);
+        }
+    }
+  },
+
+  _displayMessageForGroupInbox(message) {
+    return (
+      this.inbox === "group" &&
+      message.payload.group_ids.includes(this.activeGroup.id)
+    );
+  },
+
+  _shouldDisplayMessageForInbox(message) {
+    return (
+      this.inbox === "all" ||
+      this._displayMessageForGroupInbox(message) ||
+      (this.inbox === "user" &&
+        (message.payload.group_ids.length === 0 ||
+          this.currentUser.groups.filter((group) => {
+            return message.payload.group_ids.includes(group.id);
+          }).length === 0))
+    );
+  },
+
+  _notifyIncoming(topicId) {
+    if (this.newIncoming.indexOf(topicId) === -1) {
+      this.newIncoming.pushObject(topicId);
+    }
+  },
+
+  _loadStates(data) {
+    (data || []).forEach((topic) => {
+      this._modifyState(topic.topic_id, topic);
+    });
+  },
+
+  _modifyState(topicId, data) {
+    this.states.set(topicId, data);
+  },
+});
+
+export default PrivateMessageTopicTrackingState;
diff --git a/app/assets/javascripts/discourse/app/models/user.js b/app/assets/javascripts/discourse/app/models/user.js
index fc28a3b31f0..775b40db8ee 100644
--- a/app/assets/javascripts/discourse/app/models/user.js
+++ b/app/assets/javascripts/discourse/app/models/user.js
@@ -1,7 +1,7 @@
 import EmberObject, { computed, get, getProperties } from "@ember/object";
 import cookie, { removeCookie } from "discourse/lib/cookie";
 import { defaultHomepage, escapeExpression } from "discourse/lib/utilities";
-import { equal, gt, or } from "@ember/object/computed";
+import { equal, filterBy, gt, or } from "@ember/object/computed";
 import getURL, { getURLWithCDN } from "discourse-common/lib/get-url";
 import { A } from "@ember/array";
 import Badge from "discourse/models/badge";
@@ -562,6 +562,8 @@ const User = RestModel.extend({
     });
   },
 
+  groupsWithMessages: filterBy("groups", "has_messages", true),
+
   @discourseComputed("filteredGroups", "numGroupsToDisplay")
   displayGroups(filteredGroups, numGroupsToDisplay) {
     const groups = filteredGroups.slice(0, numGroupsToDisplay);
diff --git a/app/assets/javascripts/discourse/app/routes/build-private-messages-group-route.js b/app/assets/javascripts/discourse/app/routes/build-private-messages-group-route.js
index c82d36ccdf4..a96df1dc42b 100644
--- a/app/assets/javascripts/discourse/app/routes/build-private-messages-group-route.js
+++ b/app/assets/javascripts/discourse/app/routes/build-private-messages-group-route.js
@@ -57,16 +57,16 @@ export default (inboxType, filter) => {
 
     setupController() {
       this._super.apply(this, arguments);
-      this.controllerFor("user-private-messages").set("group", this.group);
-      this.controllerFor("user-topics-list").set("group", this.group);
 
-      if (filter) {
-        this.controllerFor("user-topics-list").subscribe(
-          `/private-messages/group/${this.get(
-            "groupName"
-          ).toLowerCase()}/${filter}`
-        );
-      }
+      const userTopicsListController = this.controllerFor("user-topics-list");
+      userTopicsListController.set("group", this.group);
+
+      userTopicsListController.set(
+        "pmTopicTrackingState.activeGroup",
+        this.group
+      );
+
+      this.controllerFor("user-private-messages").set("group", this.group);
     },
 
     dismissReadOptions() {
diff --git a/app/assets/javascripts/discourse/app/routes/build-private-messages-route.js b/app/assets/javascripts/discourse/app/routes/build-private-messages-route.js
index cf2a8c8ae38..c2608d7606c 100644
--- a/app/assets/javascripts/discourse/app/routes/build-private-messages-route.js
+++ b/app/assets/javascripts/discourse/app/routes/build-private-messages-route.js
@@ -6,6 +6,8 @@ import { action } from "@ember/object";
 
 export const NEW_FILTER = "new";
 export const UNREAD_FILTER = "unread";
+export const INBOX_FILTER = "inbox";
+export const ARCHIVE_FILTER = "archive";
 
 // A helper to build a user topic list route
 export default (inboxType, path, filter) => {
@@ -42,13 +44,13 @@ export default (inboxType, path, filter) => {
     setupController() {
       this._super.apply(this, arguments);
 
-      if (filter) {
-        this.controllerFor("user-topics-list").subscribe(
-          `/private-messages/${filter}`
-        );
-      }
+      const userPrivateMessagesController = this.controllerFor(
+        "user-private-messages"
+      );
 
-      this.controllerFor("user-topics-list").setProperties({
+      const userTopicsListController = this.controllerFor("user-topics-list");
+
+      userTopicsListController.setProperties({
         hideCategory: true,
         showPosters: true,
         tagsForUser: this.modelFor("user").get("username_lower"),
@@ -57,9 +59,13 @@ export default (inboxType, path, filter) => {
         filter: filter,
         group: null,
         inbox: inboxType,
+        pmTopicTrackingState:
+          userPrivateMessagesController.pmTopicTrackingState,
       });
 
-      this.controllerFor("user-private-messages").setProperties({
+      userTopicsListController.subscribe();
+
+      userPrivateMessagesController.setProperties({
         archive: false,
         pmView: inboxType,
         group: null,
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-archive.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-archive.js
index d12d153a5bb..9861501713e 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-archive.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-archive.js
@@ -1,7 +1,9 @@
-import createPMRoute from "discourse/routes/build-private-messages-route";
+import createPMRoute, {
+  ARCHIVE_FILTER,
+} from "discourse/routes/build-private-messages-route";
 
 export default createPMRoute(
-  "archive",
+  "all",
   "private-messages-all-archive",
-  "archive"
+  ARCHIVE_FILTER
 );
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-group-archive.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-group-archive.js
index 726f6e15017..de84bea4147 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-group-archive.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-group-archive.js
@@ -1,3 +1,4 @@
 import createPMRoute from "discourse/routes/build-private-messages-group-route";
+import { ARCHIVE_FILTER } from "discourse/routes/build-private-messages-route";
 
-export default createPMRoute("group", "archive");
+export default createPMRoute("group", ARCHIVE_FILTER);
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-group.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-group.js
index ce17c2fc0a5..e8c39ac074c 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-group.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-group.js
@@ -1,3 +1,4 @@
 import createPMRoute from "discourse/routes/build-private-messages-group-route";
+import { INBOX_FILTER } from "discourse/routes/build-private-messages-route";
 
-export default createPMRoute("group", "inbox");
+export default createPMRoute("group", INBOX_FILTER);
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-index.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-index.js
index fa4024d5289..aab9f5a98ba 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-index.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-index.js
@@ -1,3 +1,5 @@
-import createPMRoute from "discourse/routes/build-private-messages-route";
+import createPMRoute, {
+  INBOX_FILTER,
+} from "discourse/routes/build-private-messages-route";
 
-export default createPMRoute("all", "private-messages-all", "inbox");
+export default createPMRoute("all", "private-messages-all", INBOX_FILTER);
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-personal-archive.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-personal-archive.js
index 17ab77b75cb..a6128847f7c 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-personal-archive.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-personal-archive.js
@@ -1,3 +1,9 @@
-import createPMRoute from "discourse/routes/build-private-messages-route";
+import createPMRoute, {
+  ARCHIVE_FILTER,
+} from "discourse/routes/build-private-messages-route";
 
-export default createPMRoute("user", "private-messages-archive", "archive");
+export default createPMRoute(
+  "user",
+  "private-messages-archive",
+  ARCHIVE_FILTER
+);
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages-personal.js b/app/assets/javascripts/discourse/app/routes/user-private-messages-personal.js
index bedfe937aba..fd0a1f8c10e 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages-personal.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages-personal.js
@@ -1,3 +1,5 @@
-import createPMRoute from "discourse/routes/build-private-messages-route";
+import createPMRoute, {
+  INBOX_FILTER,
+} from "discourse/routes/build-private-messages-route";
 
-export default createPMRoute("user", "private-messages", "inbox");
+export default createPMRoute("user", "private-messages", INBOX_FILTER);
diff --git a/app/assets/javascripts/discourse/app/routes/user-private-messages.js b/app/assets/javascripts/discourse/app/routes/user-private-messages.js
index 34e8b488c1d..9c0673865e0 100644
--- a/app/assets/javascripts/discourse/app/routes/user-private-messages.js
+++ b/app/assets/javascripts/discourse/app/routes/user-private-messages.js
@@ -1,6 +1,9 @@
 import Composer from "discourse/models/composer";
 import DiscourseRoute from "discourse/routes/discourse";
 import Draft from "discourse/models/draft";
+import { ajax } from "discourse/lib/ajax";
+import { popupAjaxError } from "discourse/lib/ajax-error";
+import PrivateMessageTopicTrackingState from "discourse/models/private-message-topic-tracking-state";
 
 export default DiscourseRoute.extend({
   renderTemplate() {
@@ -8,11 +11,36 @@ export default DiscourseRoute.extend({
   },
 
   model() {
-    return this.modelFor("user");
+    const user = this.modelFor("user");
+    return ajax(`/u/${user.username}/private-message-topic-tracking-state`)
+      .then((pmTopicTrackingStateData) => {
+        return {
+          user,
+          pmTopicTrackingStateData,
+        };
+      })
+      .catch((e) => {
+        popupAjaxError(e);
+        return { user };
+      });
   },
 
-  setupController(controller, user) {
-    controller.set("model", user);
+  setupController(controller, model) {
+    const user = model.user;
+
+    const pmTopicTrackingState = PrivateMessageTopicTrackingState.create({
+      messageBus: controller.messageBus,
+      user,
+    });
+
+    pmTopicTrackingState.startTracking(model.pmTopicTrackingStateData);
+
+    controller.setProperties({
+      model: user,
+      pmTopicTrackingState,
+    });
+
+    this.set("pmTopicTrackingState", pmTopicTrackingState);
 
     if (this.currentUser) {
       const composerController = this.controllerFor("composer");
@@ -30,6 +58,10 @@ export default DiscourseRoute.extend({
     }
   },
 
+  deactivate() {
+    this.pmTopicTrackingState.stopTracking();
+  },
+
   actions: {
     refresh() {
       this.refresh();
diff --git a/app/assets/javascripts/discourse/app/templates/user-topics-list.hbs b/app/assets/javascripts/discourse/app/templates/user-topics-list.hbs
index e6b0a5d237c..4e9d4d138e3 100644
--- a/app/assets/javascripts/discourse/app/templates/user-topics-list.hbs
+++ b/app/assets/javascripts/discourse/app/templates/user-topics-list.hbs
@@ -13,7 +13,7 @@
       showDismissRead=showDismissRead
       resetNew=(action "resetNew")}}
 
-  {{#if hasIncoming}}
+  {{#if (gt incomingCount 0)}}
     <div class="show-mores">
       <a tabindex="0" href {{action "showInserted"}} class="alert alert-info clickable">
         {{count-i18n key="topic_count_" suffix="latest" count=incomingCount}}
diff --git a/app/assets/javascripts/discourse/app/templates/user/messages.hbs b/app/assets/javascripts/discourse/app/templates/user/messages.hbs
index 415a6ae11a5..fbe1bf4f2f3 100644
--- a/app/assets/javascripts/discourse/app/templates/user/messages.hbs
+++ b/app/assets/javascripts/discourse/app/templates/user/messages.hbs
@@ -32,13 +32,13 @@
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.new" model}}
-          {{i18n "user.messages.new"}}
+        {{#link-to "userPrivateMessages.new" model class="new"}}
+          {{newLinkText}}
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.unread" model}}
-          {{i18n "user.messages.unread"}}
+        {{#link-to "userPrivateMessages.unread" model class="unread"}}
+          {{unreadLinkText}}
         {{/link-to}}
       </li>
       <li>
@@ -55,13 +55,13 @@
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.groupNew" group.name}}
-          {{i18n "user.messages.new"}}
+        {{#link-to "userPrivateMessages.groupNew" group.name class="new"}}
+          {{newLinkText}}
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.groupUnread" group.name}}
-          {{i18n "user.messages.unread"}}
+        {{#link-to "userPrivateMessages.groupUnread" group.name class="unread"}}
+          {{unreadLinkText}}
         {{/link-to}}
       </li>
       <li>
@@ -83,13 +83,13 @@
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.personalNew" model}}
-          {{i18n "user.messages.new"}}
+        {{#link-to "userPrivateMessages.personalNew" model class="new"}}
+          {{newLinkText}}
         {{/link-to}}
       </li>
       <li>
-        {{#link-to "userPrivateMessages.personalUnread" model}}
-          {{i18n "user.messages.unread"}}
+        {{#link-to "userPrivateMessages.personalUnread" model class="unread"}}
+          {{unreadLinkText}}
         {{/link-to}}
       </li>
       <li>
@@ -101,7 +101,7 @@
 
     {{#if displayGlobalFilters}}
       {{#if pmTaggingEnabled}}
-        <li class="noGlyph tags">
+        <li class="tags">
           {{#link-to "userPrivateMessages.tags" model}}
             {{i18n "user.messages.tags"}}
           {{/link-to}}
diff --git a/app/assets/javascripts/discourse/tests/acceptance/user-private-messages-test.js b/app/assets/javascripts/discourse/tests/acceptance/user-private-messages-test.js
index 32e00a8cfaf..dac9e7d27be 100644
--- a/app/assets/javascripts/discourse/tests/acceptance/user-private-messages-test.js
+++ b/app/assets/javascripts/discourse/tests/acceptance/user-private-messages-test.js
@@ -1,10 +1,13 @@
+import { visit } from "@ember/test-helpers";
+import { test } from "qunit";
+import I18n from "I18n";
 import {
   acceptance,
   count,
   exists,
+  publishToMessageBus,
+  query,
 } from "discourse/tests/helpers/qunit-helpers";
-import { visit } from "@ember/test-helpers";
-import { test } from "qunit";
 import selectKit from "discourse/tests/helpers/select-kit-helper";
 import { PERSONAL_INBOX } from "discourse/controllers/user-private-messages";
 
@@ -72,10 +75,13 @@ acceptance(
       [
         "/topics/private-messages-all-new/:username.json",
         "/topics/private-messages-all-unread/:username.json",
+        "/topics/private-messages-all-archive/:username.json",
         "/topics/private-messages-new/:username.json",
         "/topics/private-messages-unread/:username.json",
+        "/topics/private-messages-archive/:username.json",
         "/topics/private-messages-group/:username/:group_name/new.json",
         "/topics/private-messages-group/:username/:group_name/unread.json",
+        "/topics/private-messages-group/:username/:group_name/archive.json",
       ].forEach((url) => {
         server.get(url, () => {
           let topics;
@@ -153,6 +159,227 @@ acceptance(
       });
     });
 
+    const publishUnreadToMessageBus = function (group_ids) {
+      publishToMessageBus("/private-message-topic-tracking-state/user/5", {
+        topic_id: Math.random(),
+        message_type: "unread",
+        payload: {
+          last_read_post_number: 1,
+          highest_post_number: 2,
+          notification_level: 2,
+          group_ids: group_ids || [],
+        },
+      });
+    };
+
+    const publishNewToMessageBus = function (group_ids) {
+      publishToMessageBus("/private-message-topic-tracking-state/user/5", {
+        topic_id: Math.random(),
+        message_type: "new_topic",
+        payload: {
+          last_read_post_number: null,
+          highest_post_number: 1,
+          group_ids: group_ids || [],
+        },
+      });
+    };
+
+    const publishArchiveToMessageBus = function () {
+      publishToMessageBus("/private-message-topic-tracking-state/user/5", {
+        topic_id: Math.random(),
+        message_type: "archive",
+      });
+    };
+
+    const publishGroupArchiveToMessageBus = function (group_ids) {
+      publishToMessageBus(
+        `/private-message-topic-tracking-state/group/${group_ids[0]}`,
+        {
+          topic_id: Math.random(),
+          message_type: "group_archive",
+          payload: {
+            group_ids: group_ids,
+          },
+        }
+      );
+    };
+
+    test("incoming archive message on all and archive filter", async function (assert) {
+      for (const url of [
+        "/u/charlie/messages",
+        "/u/charlie/messages/archive",
+        "/u/charlie/messages/personal",
+        "/u/charlie/messages/personal/archive",
+      ]) {
+        await visit(url);
+
+        publishArchiveToMessageBus();
+
+        await visit(url); // wait for re-render
+
+        assert.ok(
+          exists(".show-mores"),
+          `${url} displays the topic incoming info`
+        );
+      }
+
+      for (const url of [
+        "/u/charlie/messages/group/awesome_group/archive",
+        "/u/charlie/messages/group/awesome_group",
+      ]) {
+        await visit(url);
+
+        publishArchiveToMessageBus();
+
+        await visit(url); // wait for re-render
+
+        assert.ok(
+          !exists(".show-mores"),
+          `${url} does not display the topic incoming info`
+        );
+      }
+    });
+
+    test("incoming group archive message on all and archive filter", async function (assert) {
+      for (const url of [
+        "/u/charlie/messages",
+        "/u/charlie/messages/archive",
+        "/u/charlie/messages/group/awesome_group",
+        "/u/charlie/messages/group/awesome_group/archive",
+      ]) {
+        await visit(url);
+
+        publishGroupArchiveToMessageBus([14]);
+
+        await visit(url); // wait for re-render
+
+        assert.ok(
+          exists(".show-mores"),
+          `${url} displays the topic incoming info`
+        );
+      }
+
+      for (const url of [
+        "/u/charlie/messages/personal",
+        "/u/charlie/messages/personal/archive",
+      ]) {
+        await visit(url);
+
+        publishGroupArchiveToMessageBus([14]);
+
+        await visit(url); // wait for re-render
+
+        assert.ok(
+          !exists(".show-mores"),
+          `${url} does not display the topic incoming info`
+        );
+      }
+    });
+
+    test("incoming unread and new messages on all filter", async function (assert) {
+      await visit("/u/charlie/messages");
+
+      publishUnreadToMessageBus();
+      publishNewToMessageBus();
+
+      await visit("/u/charlie/messages"); // wait for re-render
+
+      assert.equal(
+        query(".messages-nav li a.new").innerText.trim(),
+        I18n.t("user.messages.new_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.equal(
+        query(".messages-nav li a.unread").innerText.trim(),
+        I18n.t("user.messages.unread_with_count", { count: 1 }),
+        "displays the right count"
+      );
+    });
+
+    test("incoming new messages while viewing new", async function (assert) {
+      await visit("/u/charlie/messages/new");
+
+      publishNewToMessageBus();
+
+      await visit("/u/charlie/messages/new"); // wait for re-render
+
+      assert.equal(
+        query(".messages-nav li a.new").innerText.trim(),
+        I18n.t("user.messages.new_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.ok(exists(".show-mores"), "displays the topic incoming info");
+    });
+
+    test("incoming unread messages while viewing unread", async function (assert) {
+      await visit("/u/charlie/messages/unread");
+
+      publishUnreadToMessageBus();
+
+      await visit("/u/charlie/messages/unread"); // wait for re-render
+
+      assert.equal(
+        query(".messages-nav li a.unread").innerText.trim(),
+        I18n.t("user.messages.unread_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.ok(exists(".show-mores"), "displays the topic incoming info");
+    });
+
+    test("incoming unread messages while viewing group unread", async function (assert) {
+      await visit("/u/charlie/messages/group/awesome_group/unread");
+
+      publishUnreadToMessageBus([14]);
+      publishNewToMessageBus([14]);
+
+      await visit("/u/charlie/messages/group/awesome_group/unread"); // wait for re-render
+
+      assert.equal(
+        query(".messages-nav li a.unread").innerText.trim(),
+        I18n.t("user.messages.unread_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.equal(
+        query(".messages-nav li a.new").innerText.trim(),
+        I18n.t("user.messages.new_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.ok(exists(".show-mores"), "displays the topic incoming info");
+
+      await visit("/u/charlie/messages/unread");
+
+      assert.equal(
+        query(".messages-nav li a.unread").innerText.trim(),
+        I18n.t("user.messages.unread_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      assert.equal(
+        query(".messages-nav li a.new").innerText.trim(),
+        I18n.t("user.messages.new_with_count", { count: 1 }),
+        "displays the right count"
+      );
+
+      await visit("/u/charlie/messages/personal/unread");
+
+      assert.equal(
+        query(".messages-nav li a.unread").innerText.trim(),
+        I18n.t("user.messages.unread"),
+        "displays the right count"
+      );
+
+      assert.equal(
+        query(".messages-nav li a.new").innerText.trim(),
+        I18n.t("user.messages.new"),
+        "displays the right count"
+      );
+    });
+
     test("dismissing all unread messages", async function (assert) {
       await visit("/u/charlie/messages/unread");
 
diff --git a/app/assets/javascripts/discourse/tests/helpers/create-pretender.js b/app/assets/javascripts/discourse/tests/helpers/create-pretender.js
index be32ddfcdf3..96ae6e91b5c 100644
--- a/app/assets/javascripts/discourse/tests/helpers/create-pretender.js
+++ b/app/assets/javascripts/discourse/tests/helpers/create-pretender.js
@@ -217,6 +217,10 @@ export function applyDefaultHandlers(pretender) {
     });
   });
 
+  pretender.get("/u/:username/private-message-topic-tracking-state", () => {
+    return response([]);
+  });
+
   pretender.get("/topics/feature_stats.json", () => {
     return response({
       pinned_in_category_count: 0,
diff --git a/app/controllers/users_controller.rb b/app/controllers/users_controller.rb
index 83ee6ce9a34..f151b340264 100644
--- a/app/controllers/users_controller.rb
+++ b/app/controllers/users_controller.rb
@@ -316,6 +316,21 @@ class UsersController < ApplicationController
     render json: MultiJson.dump(serializer)
   end
 
+  def private_message_topic_tracking_state
+    user = fetch_user_from_params
+    guardian.ensure_can_edit!(user)
+
+    report = PrivateMessageTopicTrackingState.report(user)
+
+    serializer = ActiveModel::ArraySerializer.new(
+      report,
+      each_serializer: PrivateMessageTopicTrackingStateSerializer,
+      scope: guardian
+    )
+
+    render json: MultiJson.dump(serializer)
+  end
+
   def badge_title
     params.require(:user_badge_id)
 
diff --git a/app/jobs/regular/post_update_topic_tracking_state.rb b/app/jobs/regular/post_update_topic_tracking_state.rb
index 6eaaeb288b4..c3d4610f802 100644
--- a/app/jobs/regular/post_update_topic_tracking_state.rb
+++ b/app/jobs/regular/post_update_topic_tracking_state.rb
@@ -5,8 +5,21 @@ module Jobs
 
     def execute(args)
       post = Post.find_by(id: args[:post_id])
+      return if !post&.topic
 
-      if post && post.topic
+      topic = post.topic
+
+      if topic.private_message?
+        if post.post_number > 1
+          PrivateMessageTopicTrackingState.publish_unread(post)
+        end
+
+        TopicGroup.new_message_update(
+          topic.last_poster,
+          topic.id,
+          post.post_number
+        )
+      else
         TopicTrackingState.publish_unmuted(post.topic)
         if post.post_number > 1
           TopicTrackingState.publish_muted(post.topic)
diff --git a/app/models/group_archived_message.rb b/app/models/group_archived_message.rb
index 6d31b94d206..3b481c86f12 100644
--- a/app/models/group_archived_message.rb
+++ b/app/models/group_archived_message.rb
@@ -9,7 +9,7 @@ class GroupArchivedMessage < ActiveRecord::Base
     destroyed = GroupArchivedMessage.where(group_id: group_id, topic_id: topic_id).destroy_all
     trigger(:move_to_inbox, group_id, topic_id)
     MessageBus.publish("/topic/#{topic_id}", { type: "move_to_inbox" }, group_ids: [group_id])
-    publish_topic_tracking_state(topic)
+    publish_topic_tracking_state(topic, group_id)
     set_imap_sync(topic_id) if !opts[:skip_imap_sync] && destroyed.present?
   end
 
@@ -19,7 +19,7 @@ class GroupArchivedMessage < ActiveRecord::Base
     GroupArchivedMessage.create!(group_id: group_id, topic_id: topic_id)
     trigger(:archive_message, group_id, topic_id)
     MessageBus.publish("/topic/#{topic_id}", { type: "archived" }, group_ids: [group_id])
-    publish_topic_tracking_state(topic)
+    publish_topic_tracking_state(topic, group_id)
     set_imap_sync(topic_id) if !opts[:skip_imap_sync] && destroyed.blank?
   end
 
@@ -31,20 +31,18 @@ class GroupArchivedMessage < ActiveRecord::Base
     end
   end
 
-  private
-
-  def self.publish_topic_tracking_state(topic)
-    TopicTrackingState.publish_private_message(
-      topic, group_archive: true
-    )
-  end
-
   def self.set_imap_sync(topic_id)
     IncomingEmail.joins(:post)
       .where.not(imap_uid: nil)
       .where(topic_id: topic_id, posts: { post_number: 1 })
       .update_all(imap_sync: true)
   end
+  private_class_method :set_imap_sync
+
+  def self.publish_topic_tracking_state(topic, group_id)
+    PrivateMessageTopicTrackingState.publish_group_archived(topic, group_id)
+  end
+  private_class_method :publish_topic_tracking_state
 end
 
 # == Schema Information
diff --git a/app/models/private_message_topic_tracking_state.rb b/app/models/private_message_topic_tracking_state.rb
new file mode 100644
index 00000000000..da5cf921fa5
--- /dev/null
+++ b/app/models/private_message_topic_tracking_state.rb
@@ -0,0 +1,181 @@
+# frozen_string_literal: true
+
+# This class is used to mirror unread and new status for private messages between
+# server and client.
+#
+# On the server side, this class has two main responsibilities. The first is to
+# query the database for the initial state of a user's unread and new private
+# messages. The second is to publish message_bus messages to notify the client
+# of various topic events.
+#
+# On the client side, we have a `PrivateMessageTopicTrackingState` class as well
+# which will load the initial state into memory and subscribes to the relevant
+# message_bus messages. When a message is received, it modifies the in-memory
+# state based on the message type. The filtering for new and unread topics is
+# done on the client side based on the in-memory state in order to derive the
+# count of new and unread topics efficiently.
+class PrivateMessageTopicTrackingState
+  CHANNEL_PREFIX = "/private-message-topic-tracking-state"
+  NEW_MESSAGE_TYPE = "new_topic"
+  UNREAD_MESSAGE_TYPE = "unread"
+  ARCHIVE_MESSAGE_TYPE = "archive"
+  GROUP_ARCHIVE_MESSAGE_TYPE = "group_archive"
+
+  def self.report(user)
+    sql = new_and_unread_sql(user)
+
+    DB.query(
+      sql + "\n\n LIMIT :max_topics",
+      {
+        max_topics: TopicTrackingState::MAX_TOPICS,
+        min_new_topic_date: Time.at(SiteSetting.min_new_topics_time).to_datetime
+      }
+    )
+  end
+
+  def self.new_and_unread_sql(user)
+    sql = report_raw_sql(user, skip_unread: true)
+    sql << "\nUNION ALL\n\n"
+    sql << report_raw_sql(user, skip_new: true)
+  end
+
+  def self.report_raw_sql(user, skip_unread: false,
+                                skip_new: false)
+
+    unread =
+      if skip_unread
+        "1=0"
+      else
+        TopicTrackingState.unread_filter_sql(staff: user.staff?)
+      end
+
+    new =
+      if skip_new
+        "1=0"
+      else
+        TopicTrackingState.new_filter_sql
+      end
+
+    sql = +<<~SQL
+      SELECT
+        DISTINCT topics.id AS topic_id,
+        u.id AS user_id,
+        last_read_post_number,
+        tu.notification_level,
+        #{TopicTrackingState.highest_post_number_column_select(user.staff?)},
+        ARRAY(SELECT group_id FROM topic_allowed_groups WHERE topic_allowed_groups.topic_id = topics.id) AS group_ids
+      FROM topics
+      JOIN users u on u.id = #{user.id.to_i}
+      JOIN user_stats AS us ON us.user_id = u.id
+      JOIN user_options AS uo ON uo.user_id = u.id
+      LEFT JOIN group_users gu ON gu.user_id = u.id
+      LEFT JOIN topic_allowed_groups tag ON tag.topic_id = topics.id AND tag.group_id = gu.group_id
+      LEFT JOIN topic_users tu ON tu.topic_id = topics.id AND tu.user_id = u.id
+      LEFT JOIN topic_allowed_users tau ON tau.topic_id = topics.id AND tau.user_id = u.id
+      #{skip_new ? "" : "LEFT JOIN dismissed_topic_users ON dismissed_topic_users.topic_id = topics.id AND dismissed_topic_users.user_id = #{user.id.to_i}"}
+      WHERE (tau.topic_id IS NOT NULL OR tag.topic_id IS NOT NULL) AND
+        #{skip_unread ? "" : "topics.updated_at >= LEAST(us.first_unread_pm_at, gu.first_unread_pm_at) AND"}
+        topics.archetype = 'private_message' AND
+        ((#{unread}) OR (#{new})) AND
+        topics.deleted_at IS NULL
+    SQL
+  end
+
+  def self.publish_unread(post)
+    return unless post.topic.private_message?
+
+    scope = TopicUser
+      .tracking(post.topic_id)
+      .includes(user: :user_stat)
+
+    allowed_group_ids = post.topic.allowed_groups.pluck(:id)
+
+    group_ids =
+      if post.post_type == Post.types[:whisper]
+        [Group::AUTO_GROUPS[:staff]]
+      else
+        allowed_group_ids
+      end
+
+    if group_ids.present?
+      scope = scope
+        .joins("INNER JOIN group_users gu ON gu.user_id = topic_users.user_id")
+        .where("gu.group_id IN (?)", group_ids)
+    end
+
+    scope
+      .select([:user_id, :last_read_post_number, :notification_level])
+      .each do |tu|
+
+      message = {
+        topic_id: post.topic_id,
+        message_type: UNREAD_MESSAGE_TYPE,
+        payload: {
+          last_read_post_number: tu.last_read_post_number,
+          highest_post_number: post.post_number,
+          notification_level: tu.notification_level,
+          group_ids: allowed_group_ids
+        }
+      }
+
+      MessageBus.publish(self.user_channel(tu.user_id), message.as_json,
+        user_ids: [tu.user_id]
+      )
+    end
+  end
+
+  def self.publish_new(topic)
+    return unless topic.private_message?
+
+    message = {
+      message_type: NEW_MESSAGE_TYPE,
+      topic_id: topic.id,
+      payload: {
+        last_read_post_number: nil,
+        highest_post_number: 1,
+        group_ids: topic.allowed_groups.pluck(:id)
+      }
+    }.as_json
+
+    topic.allowed_users.pluck(:id).each do |user_id|
+      MessageBus.publish(self.user_channel(user_id), message, user_ids: [user_id])
+    end
+
+    topic.allowed_groups.pluck(:id).each do |group_id|
+      MessageBus.publish(self.group_channel(group_id), message, group_ids: [group_id])
+    end
+  end
+
+  def self.publish_group_archived(topic, group_id)
+    return unless topic.private_message?
+
+    message = {
+      message_type: GROUP_ARCHIVE_MESSAGE_TYPE,
+      topic_id: topic.id,
+      payload: {
+        group_ids: [group_id]
+      }
+    }.as_json
+
+    MessageBus.publish(self.group_channel(group_id), message, group_ids: [group_id])
+  end
+
+  def self.publish_user_archived(topic, user_id)
+    return unless topic.private_message?
+
+    message = {
+      message_type: ARCHIVE_MESSAGE_TYPE,
+      topic_id: topic.id,
+    }.as_json
+
+    MessageBus.publish(self.user_channel(user_id), message, user_ids: [user_id])
+  end
+
+  def self.user_channel(user_id)
+    "#{CHANNEL_PREFIX}/user/#{user_id}"
+  end
+
+  def self.group_channel(group_id)
+    "#{CHANNEL_PREFIX}/group/#{group_id}"
+  end
+end
diff --git a/app/models/topic_tracking_state.rb b/app/models/topic_tracking_state.rb
index 3051cbb8931..a5434399ff4 100644
--- a/app/models/topic_tracking_state.rb
+++ b/app/models/topic_tracking_state.rb
@@ -16,7 +16,6 @@
 #
 # See discourse/app/models/topic-tracking-state.js
 class TopicTrackingState
-
   include ActiveModel::SerializerSupport
 
   UNREAD_MESSAGE_TYPE = "unread"
@@ -510,60 +509,6 @@ class TopicTrackingState
     "#{staff ? "topics.highest_staff_post_number AS highest_post_number" : "topics.highest_post_number"}"
   end
 
-  def self.publish_private_message(topic, archive_user_id: nil,
-                                          post: nil,
-                                          group_archive: false)
-
-    return unless topic.private_message?
-    channels = {}
-
-    allowed_user_ids = topic.allowed_users.pluck(:id)
-
-    if post && allowed_user_ids.include?(post.user_id)
-      channels["/private-messages/sent"] = [post.user_id]
-    end
-
-    if archive_user_id
-      user_ids = [archive_user_id]
-
-      [
-        "/private-messages/archive",
-        "/private-messages/inbox",
-        "/private-messages/sent",
-      ].each do |channel|
-        channels[channel] = user_ids
-      end
-    end
-
-    if channels.except("/private-messages/sent").blank?
-      channels["/private-messages/inbox"] = allowed_user_ids
-    end
-
-    topic.allowed_groups.each do |group|
-      group_user_ids = group.users.pluck(:id)
-      next if group_user_ids.blank?
-      group_channels = []
-      channel_prefix = "/private-messages/group/#{group.name.downcase}"
-      group_channels << "#{channel_prefix}/inbox"
-      group_channels << "#{channel_prefix}/archive" if group_archive
-      group_channels.each { |channel| channels[channel] = group_user_ids }
-    end
-
-    message = {
-      topic_id: topic.id
-    }
-
-    channels.each do |channel, ids|
-      if ids.present?
-        MessageBus.publish(
-          channel,
-          message.as_json,
-          user_ids: ids
-        )
-      end
-    end
-  end
-
   def self.publish_read_indicator_on_write(topic_id, last_read_post_number, user_id)
     topic = Topic.includes(:allowed_groups).select(:highest_post_number, :archetype, :id).find_by(id: topic_id)
 
diff --git a/app/models/user_archived_message.rb b/app/models/user_archived_message.rb
index 7a51c2c8601..f5ee006dfbe 100644
--- a/app/models/user_archived_message.rb
+++ b/app/models/user_archived_message.rb
@@ -36,13 +36,10 @@ class UserArchivedMessage < ActiveRecord::Base
     end
   end
 
-  private
-
   def self.publish_topic_tracking_state(topic, user_id)
-    TopicTrackingState.publish_private_message(
-      topic, archive_user_id: user_id
-    )
+    PrivateMessageTopicTrackingState.publish_user_archived(topic, user_id)
   end
+  private_class_method :publish_topic_tracking_state
 end
 
 # == Schema Information
diff --git a/app/serializers/private_message_topic_tracking_state_serializer.rb b/app/serializers/private_message_topic_tracking_state_serializer.rb
new file mode 100644
index 00000000000..d08b3b78d67
--- /dev/null
+++ b/app/serializers/private_message_topic_tracking_state_serializer.rb
@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class PrivateMessageTopicTrackingStateSerializer < ApplicationSerializer
+  attributes :topic_id,
+             :highest_post_number,
+             :last_read_post_number,
+             :notification_level,
+             :group_ids
+end
diff --git a/config/locales/client.en.yml b/config/locales/client.en.yml
index 685689e0aa9..fcbb059ecc5 100644
--- a/config/locales/client.en.yml
+++ b/config/locales/client.en.yml
@@ -1184,7 +1184,13 @@ en:
         latest: "Latest"
         sent: "Sent"
         unread: "Unread"
+        unread_with_count:
+          one: "Unread (%{count})"
+          other: "Unread (%{count})"
         new: "New"
+        new_with_count:
+          one: "New (%{count})"
+          other: "New (%{count})"
         archive: "Archive"
         groups: "My Groups"
         move_to_inbox: "Move to Inbox"
diff --git a/config/routes.rb b/config/routes.rb
index c7737d9fce0..8a237167203 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -509,6 +509,7 @@ Discourse::Application.routes.draw do
       get "#{root_path}/:username/flagged-posts" => "users#show", constraints: { username: RouteFormat.username }
       get "#{root_path}/:username/deleted-posts" => "users#show", constraints: { username: RouteFormat.username }
       get "#{root_path}/:username/topic-tracking-state" => "users#topic_tracking_state", constraints: { username: RouteFormat.username }
+      get "#{root_path}/:username/private-message-topic-tracking-state" => "users#private_message_topic_tracking_state", constraints: { username: RouteFormat.username }
       get "#{root_path}/:username/profile-hidden" => "users#profile_hidden"
       put "#{root_path}/:username/feature-topic" => "users#feature_topic", constraints: { username: RouteFormat.username }
       put "#{root_path}/:username/clear-featured-topic" => "users#clear_featured_topic", constraints: { username: RouteFormat.username }
diff --git a/lib/post_jobs_enqueuer.rb b/lib/post_jobs_enqueuer.rb
index 4809fdc1e86..80d49c59751 100644
--- a/lib/post_jobs_enqueuer.rb
+++ b/lib/post_jobs_enqueuer.rb
@@ -20,11 +20,6 @@ class PostJobsEnqueuer
       after_topic_create
       make_visible
     end
-
-    if @topic.private_message?
-      TopicTrackingState.publish_private_message(@topic, post: @post)
-      TopicGroup.new_message_update(@topic.last_poster, @topic.id, @post.post_number)
-    end
   end
 
   private
@@ -46,6 +41,7 @@ class PostJobsEnqueuer
   end
 
   def make_visible
+    return if @topic.private_message?
     return unless SiteSetting.embed_unlisted?
     return unless @post.post_number > 1
     return if @topic.visible?
@@ -73,12 +69,18 @@ class PostJobsEnqueuer
     @topic.posters = @topic.posters_summary
     @topic.posts_count = 1
 
-    TopicTrackingState.publish_new(@topic)
+    klass =
+      if @topic.private_message?
+        PrivateMessageTopicTrackingState
+      else
+        TopicTrackingState
+      end
+
+    klass.publish_new(@topic)
   end
 
   def skip_after_create?
     @opts[:import_mode] ||
-      @topic.private_message? ||
       @post.post_type == Post.types[:moderator_action] ||
       @post.post_type == Post.types[:small_action]
   end
diff --git a/spec/models/group_archived_message_spec.rb b/spec/models/group_archived_message_spec.rb
new file mode 100644
index 00000000000..b186ea29394
--- /dev/null
+++ b/spec/models/group_archived_message_spec.rb
@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+describe GroupArchivedMessage do
+  fab!(:user) { Fabricate(:user) }
+  fab!(:user_2) { Fabricate(:user) }
+
+  fab!(:group) do
+    Fabricate(:group, messageable_level: Group::ALIAS_LEVELS[:everyone]).tap do |g|
+      g.add(user_2)
+    end
+  end
+
+  fab!(:group_message) do
+    create_post(
+      user: user,
+      target_group_names: [group.name],
+      archetype: Archetype.private_message
+    ).topic
+  end
+
+  describe '.move_to_inbox!' do
+    it 'should unarchive the topic correctly' do
+      described_class.archive!(group.id, group_message)
+
+      messages = MessageBus.track_publish(PrivateMessageTopicTrackingState.group_channel(group.id)) do
+        described_class.move_to_inbox!(group.id, group_message)
+      end
+
+      expect(messages.present?).to eq(true)
+
+      expect(GroupArchivedMessage.exists?(
+        topic: group_message,
+        group: group
+      )).to eq(false)
+    end
+  end
+
+  describe '.archive!' do
+    it 'should archive the topic correctly' do
+      messages = MessageBus.track_publish(PrivateMessageTopicTrackingState.group_channel(group.id)) do
+        described_class.archive!(group.id, group_message)
+      end
+
+      expect(GroupArchivedMessage.exists?(
+        topic: group_message,
+        group: group
+      )).to eq(true)
+
+      expect(messages.present?).to eq(true)
+    end
+  end
+end
diff --git a/spec/models/private_message_topic_tracking_state_spec.rb b/spec/models/private_message_topic_tracking_state_spec.rb
new file mode 100644
index 00000000000..164001bedd7
--- /dev/null
+++ b/spec/models/private_message_topic_tracking_state_spec.rb
@@ -0,0 +1,206 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+describe PrivateMessageTopicTrackingState do
+  fab!(:user) { Fabricate(:user) }
+  fab!(:user_2) { Fabricate(:user) }
+
+  fab!(:group) do
+    Fabricate(:group, messageable_level: Group::ALIAS_LEVELS[:everyone]).tap do |g|
+      g.add(user_2)
+    end
+  end
+
+  fab!(:group_message) do
+    create_post(
+      user: user,
+      target_group_names: [group.name],
+      archetype: Archetype.private_message
+    ).topic
+  end
+
+  fab!(:private_message) do
+    create_post(
+      user: user,
+      target_usernames: [user_2.username],
+      archetype: Archetype.private_message
+    ).topic
+  end
+
+  fab!(:private_message_2) do
+    create_post(
+      user: user,
+      target_usernames: [Fabricate(:user).username],
+      archetype: Archetype.private_message
+    ).topic
+  end
+
+  describe '.report' do
+    it 'returns the right tracking state' do
+      TopicUser.find_by(user: user_2, topic: group_message).update!(
+        last_read_post_number: 1
+      )
+
+      expect(described_class.report(user_2).map(&:topic_id))
+        .to contain_exactly(private_message.id)
+
+      create_post(user: user, topic: group_message)
+
+      report = described_class.report(user_2)
+
+      expect(report.map(&:topic_id)).to contain_exactly(
+        group_message.id,
+        private_message.id
+      )
+
+      state = report.first
+
+      expect(state.topic_id).to eq(private_message.id)
+      expect(state.user_id).to eq(user_2.id)
+      expect(state.last_read_post_number).to eq(nil)
+      expect(state.notification_level).to eq(NotificationLevels.all[:watching])
+      expect(state.highest_post_number).to eq(1)
+      expect(state.group_ids).to eq([])
+
+      expect(report.last.group_ids).to contain_exactly(group.id)
+    end
+
+    it 'returns the right tracking state when topics contain whispers' do
+      TopicUser.find_by(user: user_2, topic: private_message).update!(
+        last_read_post_number: 1
+      )
+
+      create_post(
+        raw: "this is a test post",
+        topic: private_message,
+        post_type: Post.types[:whisper],
+        user: Fabricate(:admin)
+      )
+
+      expect(described_class.report(user_2).map(&:topic_id))
+        .to contain_exactly(group_message.id)
+
+      user_2.grant_admin!
+
+      tracking_state = described_class.report(user_2)
+
+      expect(tracking_state.map { |topic| [topic.topic_id, topic.highest_post_number] })
+        .to contain_exactly(
+          [group_message.id, 1],
+          [private_message.id, 2]
+        )
+    end
+
+    it 'returns the right tracking state when topics have been dismissed' do
+      DismissedTopicUser.create!(
+        user_id: user_2.id,
+        topic_id: group_message.id
+      )
+
+      expect(described_class.report(user_2).map(&:topic_id))
+        .to contain_exactly(private_message.id)
+    end
+  end
+
+  describe '.publish_new' do
+    it 'should publish the right message_bus message' do
+      messages = MessageBus.track_publish do
+        described_class.publish_new(private_message)
+      end
+
+      expect(messages.map(&:channel)).to contain_exactly(
+        described_class.user_channel(user.id),
+        described_class.user_channel(user_2.id)
+      )
+
+      data = messages.find do |message|
+        message.channel == described_class.user_channel(user.id)
+      end.data
+
+      expect(data['message_type']).to eq(described_class::NEW_MESSAGE_TYPE)
+    end
+
+    it 'should publish the right message_bus message for a group message' do
+      messages = MessageBus.track_publish do
+        described_class.publish_new(group_message)
+      end
+
+      expect(messages.map(&:channel)).to contain_exactly(
+        described_class.group_channel(group.id),
+        described_class.user_channel(user.id)
+      )
+
+      data = messages.find do |message|
+        message.channel == described_class.group_channel(group.id)
+      end.data
+
+      expect(data['message_type']).to eq(described_class::NEW_MESSAGE_TYPE)
+      expect(data['topic_id']).to eq(group_message.id)
+      expect(data['payload']['last_read_post_number']).to eq(nil)
+      expect(data['payload']['highest_post_number']).to eq(1)
+      expect(data['payload']['group_ids']).to eq([group.id])
+    end
+  end
+
+  describe '.publish_unread' do
+    it 'should publish the right message_bus message' do
+      messages = MessageBus.track_publish do
+        described_class.publish_unread(private_message.first_post)
+      end
+
+      expect(messages.map(&:channel)).to contain_exactly(
+        described_class.user_channel(user.id),
+        described_class.user_channel(user_2.id)
+      )
+
+      data = messages.find do |message|
+        message.channel == described_class.user_channel(user.id)
+      end.data
+
+      expect(data['message_type']).to eq(described_class::UNREAD_MESSAGE_TYPE)
+      expect(data['topic_id']).to eq(private_message.id)
+      expect(data['payload']['last_read_post_number']).to eq(1)
+      expect(data['payload']['highest_post_number']).to eq(1)
+      expect(data['payload']['notification_level'])
+        .to eq(NotificationLevels.all[:watching])
+      expect(data['payload']['group_ids']).to eq([])
+    end
+  end
+
+  describe '.publish_user_archived' do
+    it 'should publish the right message_bus message' do
+      message = MessageBus.track_publish(described_class.user_channel(user.id)) do
+        described_class.publish_user_archived(private_message, user.id)
+      end.first
+
+      data = message.data
+
+      expect(data['topic_id']).to eq(private_message.id)
+      expect(data['message_type']).to eq(described_class::ARCHIVE_MESSAGE_TYPE)
+    end
+  end
+
+  describe '.publish_group_archived' do
+    it 'should publish the right message_bus message' do
+      user_3 = Fabricate(:user)
+      group.add(user_3)
+
+      messages = MessageBus.track_publish do
+        described_class.publish_group_archived(group_message, group.id)
+      end
+
+      expect(messages.map(&:channel)).to contain_exactly(
+        described_class.group_channel(group.id)
+      )
+
+      data = messages.find do |message|
+        message.channel == described_class.group_channel(group.id)
+      end.data
+
+      expect(data['message_type']).to eq(described_class::GROUP_ARCHIVE_MESSAGE_TYPE)
+      expect(data['topic_id']).to eq(group_message.id)
+      expect(data['payload']['group_ids']).to contain_exactly(group.id)
+    end
+  end
+end
diff --git a/spec/models/topic_tracking_state_spec.rb b/spec/models/topic_tracking_state_spec.rb
index ddcfc37c312..00472bb44c7 100644
--- a/spec/models/topic_tracking_state_spec.rb
+++ b/spec/models/topic_tracking_state_spec.rb
@@ -208,186 +208,6 @@ describe TopicTrackingState do
     end
   end
 
-  describe '#publish_private_message' do
-    fab!(:admin) { Fabricate(:admin) }
-
-    describe 'normal topic' do
-      it 'should publish the right message' do
-        allowed_users = private_message_topic.allowed_users
-
-        messages = MessageBus.track_publish do
-          TopicTrackingState.publish_private_message(private_message_topic)
-        end
-
-        expect(messages.count).to eq(1)
-
-        message = messages.first
-
-        expect(message.channel).to eq('/private-messages/inbox')
-        expect(message.data["topic_id"]).to eq(private_message_topic.id)
-        expect(message.user_ids).to contain_exactly(*allowed_users.map(&:id))
-      end
-    end
-
-    describe 'topic with groups' do
-      fab!(:group1) { Fabricate(:group, users: [Fabricate(:user)]) }
-      fab!(:group2) { Fabricate(:group, users: [Fabricate(:user), Fabricate(:user)]) }
-
-      before do
-        [group1, group2].each do |group|
-          private_message_topic.allowed_groups << group
-        end
-      end
-
-      it "should publish the right message" do
-        messages = MessageBus.track_publish do
-          TopicTrackingState.publish_private_message(
-            private_message_topic
-          )
-        end
-
-        expect(messages.map(&:channel)).to contain_exactly(
-          '/private-messages/inbox',
-          "/private-messages/group/#{group1.name}/inbox",
-          "/private-messages/group/#{group2.name}/inbox"
-        )
-
-        message = messages.find do |m|
-          m.channel == '/private-messages/inbox'
-        end
-
-        expect(message.data["topic_id"]).to eq(private_message_topic.id)
-        expect(message.user_ids).to eq(private_message_topic.allowed_users.map(&:id))
-
-        [group1, group2].each do |group|
-          message = messages.find do |m|
-            m.channel == "/private-messages/group/#{group.name}/inbox"
-          end
-
-          expect(message.data["topic_id"]).to eq(private_message_topic.id)
-          expect(message.user_ids).to eq(group.users.map(&:id))
-        end
-      end
-
-      describe "archiving topic" do
-        it "should publish the right message" do
-          messages = MessageBus.track_publish do
-            TopicTrackingState.publish_private_message(
-              private_message_topic,
-              group_archive: true
-            )
-          end
-
-          expect(messages.map(&:channel)).to contain_exactly(
-            '/private-messages/inbox',
-            "/private-messages/group/#{group1.name}/inbox",
-            "/private-messages/group/#{group1.name}/archive",
-            "/private-messages/group/#{group2.name}/inbox",
-            "/private-messages/group/#{group2.name}/archive",
-          )
-
-          message = messages.find { |m| m.channel == '/private-messages/inbox' }
-
-          expect(message.data["topic_id"]).to eq(private_message_topic.id)
-          expect(message.user_ids).to eq(private_message_topic.allowed_users.map(&:id))
-
-          [group1, group2].each do |group|
-            [
-              "/private-messages/group/#{group.name}/inbox",
-              "/private-messages/group/#{group.name}/archive"
-            ].each do |channel|
-              message = messages.find { |m| m.channel == channel }
-              expect(message.data["topic_id"]).to eq(private_message_topic.id)
-              expect(message.user_ids).to eq(group.users.map(&:id))
-            end
-          end
-        end
-      end
-    end
-
-    describe 'topic with new post' do
-      let(:user) { private_message_topic.allowed_users.last }
-
-      let!(:post) do
-        Fabricate(:post,
-          topic: private_message_topic,
-          user: user
-        )
-      end
-
-      let!(:group) do
-        group = Fabricate(:group, users: [Fabricate(:user)])
-        private_message_topic.allowed_groups << group
-        group
-      end
-
-      it 'should publish the right message' do
-        messages = MessageBus.track_publish do
-          TopicTrackingState.publish_private_message(
-            private_message_topic,
-            post: post
-          )
-        end
-
-        expected_channels = [
-          '/private-messages/inbox',
-          '/private-messages/sent',
-          "/private-messages/group/#{group.name}/inbox"
-        ]
-
-        expect(messages.map(&:channel)).to contain_exactly(*expected_channels)
-
-        expected_channels.zip([
-          private_message_topic.allowed_users.map(&:id),
-          [user.id],
-          [group.users.first.id]
-        ]).each do |channel, user_ids|
-          message = messages.find { |m| m.channel == channel }
-
-          expect(message.data["topic_id"]).to eq(private_message_topic.id)
-          expect(message.user_ids).to eq(user_ids)
-        end
-      end
-    end
-
-    describe 'archived topic' do
-      it 'should publish the right message' do
-        messages = MessageBus.track_publish do
-          TopicTrackingState.publish_private_message(
-            private_message_topic,
-            archive_user_id: private_message_post.user_id,
-          )
-        end
-
-        expected_channels = [
-          "/private-messages/archive",
-          "/private-messages/inbox",
-          "/private-messages/sent",
-        ]
-
-        expect(messages.map(&:channel)).to eq(expected_channels)
-
-        expected_channels.each do |channel|
-          message = messages.find { |m| m.channel = channel }
-          expect(message.data["topic_id"]).to eq(private_message_topic.id)
-          expect(message.user_ids).to eq([private_message_post.user_id])
-        end
-      end
-    end
-
-    describe 'for a regular topic' do
-      it 'should not publish any message' do
-        topic.allowed_users << Fabricate(:user)
-
-        messages = MessageBus.track_publish do
-          TopicTrackingState.publish_private_message(topic)
-        end
-
-        expect(messages).to eq([])
-      end
-    end
-  end
-
   describe '#publish_read_private_message' do
     fab!(:group) { Fabricate(:group) }
     let(:read_topic_key) { "/private-messages/unread-indicator/#{group_message.id}" }
diff --git a/spec/models/user_archived_message_spec.rb b/spec/models/user_archived_message_spec.rb
index d225b5bb360..dcfd7743e17 100644
--- a/spec/models/user_archived_message_spec.rb
+++ b/spec/models/user_archived_message_spec.rb
@@ -3,20 +3,51 @@
 require 'rails_helper'
 
 describe UserArchivedMessage do
-  it 'Does not move archived muted messages back to inbox' do
-    user = Fabricate(:admin)
-    user2 = Fabricate(:admin)
+  fab!(:user) { Fabricate(:user) }
+  fab!(:user_2) { Fabricate(:user) }
 
-    topic = create_post(user: user,
-                        skip_validations: true,
-                        target_usernames: [user2.username, user.username].join(","),
-                        archetype: Archetype.private_message).topic
-
-    UserArchivedMessage.archive!(user.id, topic)
-    expect(topic.message_archived?(user)).to eq(true)
-
-    TopicUser.change(user.id, topic.id, notification_level: TopicUser.notification_levels[:muted])
-    UserArchivedMessage.move_to_inbox!(user.id, topic)
-    expect(topic.message_archived?(user)).to eq(true)
+  fab!(:private_message) do
+    create_post(
+      user: user,
+      skip_validations: true,
+      target_usernames: [user_2.username, user.username].join(","),
+      archetype: Archetype.private_message
+    ).topic
   end
+
+  describe '.move_to_inbox!' do
+    it 'moves topic back to inbox correctly' do
+      UserArchivedMessage.archive!(user.id, private_message)
+
+      expect do
+        messages = MessageBus.track_publish(PrivateMessageTopicTrackingState.user_channel(user.id)) do
+          UserArchivedMessage.move_to_inbox!(user.id, private_message)
+        end
+
+        expect(messages.present?).to eq(true)
+      end.to change { private_message.message_archived?(user) }.from(true).to(false)
+    end
+
+    it 'does not move archived muted messages back to inbox' do
+      UserArchivedMessage.archive!(user.id, private_message)
+
+      expect(private_message.message_archived?(user)).to eq(true)
+
+      TopicUser.change(user.id, private_message.id, notification_level: TopicUser.notification_levels[:muted])
+      UserArchivedMessage.move_to_inbox!(user.id, private_message)
+
+      expect(private_message.message_archived?(user)).to eq(true)
+    end
+  end
+
+  describe '.archive' do
+    it 'archives message correctly' do
+      messages = MessageBus.track_publish(PrivateMessageTopicTrackingState.user_channel(user.id)) do
+        UserArchivedMessage.archive!(user.id, private_message)
+      end
+
+      expect(messages.present?).to eq(true)
+    end
+  end
+
 end
diff --git a/spec/requests/users_controller_spec.rb b/spec/requests/users_controller_spec.rb
index b22fe7349d3..abb36a7befe 100644
--- a/spec/requests/users_controller_spec.rb
+++ b/spec/requests/users_controller_spec.rb
@@ -5064,6 +5064,43 @@ describe UsersController do
     end
   end
 
+  describe "#private_message_topic_tracking_state" do
+    fab!(:user) { Fabricate(:user) }
+    fab!(:user_2) { Fabricate(:user) }
+
+    fab!(:private_message) do
+      create_post(
+        user: user,
+        target_usernames: [user_2.username],
+        archetype: Archetype.private_message
+      ).topic
+    end
+
+    before do
+      sign_in(user_2)
+    end
+
+    it 'does not allow an unauthorized user to access the state of another user' do
+      get "/u/#{user.username}/private-message-topic-tracking-state.json"
+
+      expect(response.status).to eq(403)
+    end
+
+    it 'returns the right response' do
+      get "/u/#{user_2.username}/private-message-topic-tracking-state.json"
+
+      expect(response.status).to eq(200)
+
+      topic_state = response.parsed_body.first
+
+      expect(topic_state["topic_id"]).to eq(private_message.id)
+      expect(topic_state["highest_post_number"]).to eq(1)
+      expect(topic_state["last_read_post_number"]).to eq(nil)
+      expect(topic_state["notification_level"]).to eq(NotificationLevels.all[:watching])
+      expect(topic_state["group_ids"]).to eq([])
+    end
+  end
+
   def create_second_factor_security_key
     sign_in(user)
     stub_secure_session_confirmed