FEATURE: Publish read topic tracking events for private messages. (#14274)

Follow-up to fc1fd1b416

app/assets/javascripts/discourse/app/models/private-message-topic-tracking-state.js

@@ -2,6 +2,7 @@ import EmberObject from "@ember/object";
 import { ajax } from "discourse/lib/ajax";
 import { on } from "discourse-common/utils/decorators";
 import { popupAjaxError } from "discourse/lib/ajax-error";
+import { deepEqual, deepMerge } from "discourse-common/lib/object";
 import {
   ARCHIVE_FILTER,
   INBOX_FILTER,
@@ -139,6 +140,15 @@ const PrivateMessageTopicTrackingState = EmberObject.extend({
         }
 
         break;
+      case "read":
+        this._modifyState(message.topic_id, message.payload);
+
+        if (
+          this.filter === UNREAD_FILTER &&
+          this._shouldDisplayMessageForInbox(message)
+        ) {
+          this._notifyIncoming(message.topic_id);
+        }
       case "unread":
         this._modifyState(message.topic_id, message.payload);
 
@@ -206,7 +216,14 @@ const PrivateMessageTopicTrackingState = EmberObject.extend({
   },
 
   _modifyState(topicId, data, opts = {}) {
-    this.states.set(topicId, data);
+    const oldState = this.states.get(topicId);
+    let newState = data;
+
+    if (oldState && !deepEqual(oldState, newState)) {
+      newState = deepMerge(oldState, newState);
+    }
+
+    this.states.set(topicId, newState);
 
     if (!opts.skipIncrement) {
       this.incrementProperty("statesModificationCounter");

app/assets/javascripts/discourse/tests/acceptance/user-private-messages-test.js

@@ -170,11 +170,26 @@ acceptance(
       });
     });
 
+    const publishReadToMessageBus = function (opts = {}) {
+      publishToMessageBus(
+        `/private-message-topic-tracking-state/user/${opts.userId || 5}`,
+        {
+          topic_id: opts.topicId,
+          message_type: "read",
+          payload: {
+            last_read_post_number: 2,
+            highest_post_number: 2,
+            notification_level: 2,
+          },
+        }
+      );
+    };
+
     const publishUnreadToMessageBus = function (opts = {}) {
       publishToMessageBus(
         `/private-message-topic-tracking-state/user/${opts.userId || 5}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "unread",
           payload: {
             last_read_post_number: 1,
@@ -190,7 +205,7 @@ acceptance(
       publishToMessageBus(
         `/private-message-topic-tracking-state/user/${opts.userId || 5}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "new_topic",
           payload: {
             last_read_post_number: null,
@@ -201,55 +216,55 @@ acceptance(
       );
     };
 
-    const publishArchiveToMessageBus = function (userId) {
+    const publishArchiveToMessageBus = function (opts) {
       publishToMessageBus(
-        `/private-message-topic-tracking-state/user/${userId || 5}`,
+        `/private-message-topic-tracking-state/user/${opts.userId || 5}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "archive",
         }
       );
     };
 
-    const publishGroupArchiveToMessageBus = function (groupIds) {
+    const publishGroupArchiveToMessageBus = function (opts) {
       publishToMessageBus(
-        `/private-message-topic-tracking-state/group/${groupIds[0]}`,
+        `/private-message-topic-tracking-state/group/${opts.groupIds[0]}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "group_archive",
           payload: {
-            group_ids: groupIds,
+            group_ids: opts.groupIds,
           },
         }
       );
     };
 
-    const publishGroupUnreadToMessageBus = function (groupIds) {
+    const publishGroupUnreadToMessageBus = function (opts) {
       publishToMessageBus(
-        `/private-message-topic-tracking-state/group/${groupIds[0]}`,
+        `/private-message-topic-tracking-state/group/${opts.groupIds[0]}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "unread",
           payload: {
             last_read_post_number: 1,
             highest_post_number: 2,
             notification_level: 2,
-            group_ids: groupIds || [],
+            group_ids: opts.groupIds || [],
           },
         }
       );
     };
 
-    const publishGroupNewToMessageBus = function (groupIds) {
+    const publishGroupNewToMessageBus = function (opts) {
       publishToMessageBus(
-        `/private-message-topic-tracking-state/group/${groupIds[0]}`,
+        `/private-message-topic-tracking-state/group/${opts.groupIds[0]}`,
         {
-          topic_id: Math.random(),
+          topic_id: opts.topicId,
           message_type: "new_topic",
           payload: {
             last_read_post_number: null,
             highest_post_number: 1,
-            group_ids: groupIds || [],
+            group_ids: opts.groupIds || [],
           },
         }
       );
@@ -264,7 +279,7 @@ acceptance(
       ]) {
         await visit(url);
 
-        publishArchiveToMessageBus();
+        publishArchiveToMessageBus({ topicId: 1 });
 
         await visit(url); // wait for re-render
 
@@ -280,7 +295,7 @@ acceptance(
       ]) {
         await visit(url);
 
-        publishArchiveToMessageBus();
+        publishArchiveToMessageBus({ topicId: 1 });
 
         await visit(url); // wait for re-render
 
@@ -291,6 +306,16 @@ acceptance(
       }
     });
 
+    test("incoming read message on unread filter", async function (assert) {
+      await visit("/u/charlie/messages/unread");
+
+      publishReadToMessageBus({ topicId: 1 });
+
+      await visit("/u/charlie/messages/unread"); // wait for re-render
+
+      assert.ok(exists(".show-mores"), `displays the topic incoming info`);
+    });
+
     test("incoming group archive message on all and archive filter", async function (assert) {
       for (const url of [
         "/u/charlie/messages",
@@ -300,7 +325,7 @@ acceptance(
       ]) {
         await visit(url);
 
-        publishGroupArchiveToMessageBus([14]);
+        publishGroupArchiveToMessageBus({ groupIds: [14], topicId: 1 });
 
         await visit(url); // wait for re-render
 
@@ -316,7 +341,7 @@ acceptance(
       ]) {
         await visit(url);
 
-        publishGroupArchiveToMessageBus([14]);
+        publishGroupArchiveToMessageBus({ groupIds: [14], topicId: 1 });
 
         await visit(url); // wait for re-render
 
@@ -330,8 +355,8 @@ acceptance(
     test("incoming unread and new messages on all filter", async function (assert) {
       await visit("/u/charlie/messages");
 
-      publishUnreadToMessageBus();
+      publishUnreadToMessageBus({ topicId: 1 });
-      publishNewToMessageBus();
+      publishNewToMessageBus({ topicId: 2 });
 
       await visit("/u/charlie/messages"); // wait for re-render
 
@@ -351,7 +376,7 @@ acceptance(
     test("incoming new messages while viewing new", async function (assert) {
       await visit("/u/charlie/messages/new");
 
-      publishNewToMessageBus();
+      publishNewToMessageBus({ topicId: 1 });
 
       await visit("/u/charlie/messages/new"); // wait for re-render
 
@@ -383,8 +408,8 @@ acceptance(
     test("incoming unread messages while viewing group unread", async function (assert) {
       await visit("/u/charlie/messages/group/awesome_group/unread");
 
-      publishUnreadToMessageBus({ groupIds: [14] });
+      publishUnreadToMessageBus({ groupIds: [14], topicId: 1 });
-      publishNewToMessageBus({ groupIds: [14] });
+      publishNewToMessageBus({ groupIds: [14], topicId: 2 });
 
       await visit("/u/charlie/messages/group/awesome_group/unread"); // wait for re-render
 
@@ -609,7 +634,7 @@ acceptance(
     test("suggested messages with new and unread", async function (assert) {
       await visit("/t/12");
 
-      publishNewToMessageBus({ userId: 5 });
+      publishNewToMessageBus({ userId: 5, topicId: 1 });
 
       await visit("/t/12"); // await re-render
 
@@ -619,7 +644,7 @@ acceptance(
         "displays the right browse more message"
       );
 
-      publishUnreadToMessageBus({ userId: 5 });
+      publishUnreadToMessageBus({ userId: 5, topicId: 2 });
 
       await visit("/t/12"); // await re-render
 
@@ -628,6 +653,16 @@ acceptance(
         "There is 1 unread and 1 new message remaining, or browse other personal messages",
         "displays the right browse more message"
       );
+
+      publishReadToMessageBus({ userId: 5, topicId: 2 });
+
+      await visit("/t/12"); // await re-render
+
+      assert.equal(
+        query(".suggested-topics-message").innerText.trim(),
+        "There is 1 new message remaining, or browse other personal messages",
+        "displays the right browse more message"
+      );
     });
 
     test("suggested messages for group messages without new or unread", async function (assert) {
@@ -643,7 +678,7 @@ acceptance(
     test("suggested messages for group messages with new and unread", async function (assert) {
       await visit("/t/13");
 
-      publishGroupNewToMessageBus([14]);
+      publishGroupNewToMessageBus({ groupIds: [14], topicId: 1 });
 
       await visit("/t/13"); // await re-render
 
@@ -653,7 +688,7 @@ acceptance(
         "displays the right browse more message"
       );
 
-      publishGroupUnreadToMessageBus([14]);
+      publishGroupUnreadToMessageBus({ groupIds: [14], topicId: 2 });
 
       await visit("/t/13"); // await re-render
 

app/models/concerns/topic_tracking_state_publishable.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module TopicTrackingStatePublishable
+  extend ActiveSupport::Concern
+
+  class_methods do
+    def publish_read_message(message_type:,
+                             channel_name:,
+                             topic_id:,
+                             user:,
+                             last_read_post_number:,
+                             notification_level: nil)
+
+      highest_post_number = DB.query_single(
+        "SELECT #{user.staff? ? "highest_staff_post_number" : "highest_post_number"} FROM topics WHERE id = ?",
+        topic_id
+      ).first
+
+      message = {
+        message_type: message_type,
+        topic_id: topic_id,
+        payload: {
+          last_read_post_number: last_read_post_number,
+          notification_level: notification_level,
+          highest_post_number: highest_post_number
+        }
+      }.as_json
+
+      MessageBus.publish(channel_name, message, user_ids: [user.id])
+    end
+  end
+end

app/models/private_message_topic_tracking_state.rb

@@ -15,9 +15,12 @@
 # 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
+  include TopicTrackingStatePublishable
+
   CHANNEL_PREFIX = "/private-message-topic-tracking-state"
   NEW_MESSAGE_TYPE = "new_topic"
   UNREAD_MESSAGE_TYPE = "unread"
+  READ_MESSAGE_TYPE = "read"
   ARCHIVE_MESSAGE_TYPE = "archive"
   GROUP_ARCHIVE_MESSAGE_TYPE = "group_archive"
 
@@ -185,6 +188,17 @@ class PrivateMessageTopicTrackingState
     MessageBus.publish(self.user_channel(user_id), message, user_ids: [user_id])
   end
 
+  def self.publish_read(topic_id, last_read_post_number, user, notification_level = nil)
+    self.publish_read_message(
+      message_type: READ_MESSAGE_TYPE,
+      channel_name: self.user_channel(user.id),
+      topic_id: topic_id,
+      user: user,
+      last_read_post_number: last_read_post_number,
+      notification_level: notification_level
+    )
+  end
+
   def self.user_channel(user_id)
     "#{CHANNEL_PREFIX}/user/#{user_id}"
   end

app/models/topic_tracking_state.rb

@@ -17,6 +17,7 @@
 # See discourse/app/models/topic-tracking-state.js
 class TopicTrackingState
   include ActiveModel::SerializerSupport
+  include TopicTrackingStatePublishable
 
   UNREAD_MESSAGE_TYPE = "unread"
   LATEST_MESSAGE_TYPE = "latest"
@@ -227,24 +228,14 @@ class TopicTrackingState
   end
 
   def self.publish_read(topic_id, last_read_post_number, user, notification_level = nil)
-    user_id = user.id
+    self.publish_read_message(
-
-    highest_post_number = DB.query_single(
-      "SELECT #{user.staff? ? "highest_staff_post_number" : "highest_post_number"} FROM topics WHERE id = ?",
-      topic_id
-    ).first
-
-    message = {
-      topic_id: topic_id,
       message_type: READ_MESSAGE_TYPE,
-      payload: {
+      channel_name: self.unread_channel_key(user.id),
+      topic_id: topic_id,
+      user: user,
       last_read_post_number: last_read_post_number,
-        highest_post_number: highest_post_number,
       notification_level: notification_level
-      }
+    )
-    }
-
-    MessageBus.publish(self.unread_channel_key(user_id), message.as_json, user_ids: [user_id])
   end
 
   def self.publish_dismiss_new(user_id, topic_ids: [])

app/models/topic_user.rb

@@ -259,7 +259,8 @@ class TopicUser < ActiveRecord::Base
 
     # Update the last read and the last seen post count, but only if it doesn't exist.
     # This would be a lot easier if psql supported some kind of upsert
-    UPDATE_TOPIC_USER_SQL = "UPDATE topic_users
+    UPDATE_TOPIC_USER_SQL = <<~SQL
+      UPDATE topic_users
       SET
         last_read_post_number = GREATEST(:post_number, tu.last_read_post_number),
         total_msecs_viewed = LEAST(tu.total_msecs_viewed + :msecs,86400000),
@@ -282,10 +283,11 @@ class TopicUser < ActiveRecord::Base
          tu.topic_id = :topic_id AND
          tu.user_id = :user_id
     RETURNING
-                                    topic_users.notification_level, tu.notification_level old_level, tu.last_read_post_number
+      topic_users.notification_level,
-                                "
+      tu.notification_level old_level,
-
+      tu.last_read_post_number,
-    UPDATE_TOPIC_USER_SQL_STAFF = UPDATE_TOPIC_USER_SQL.gsub("highest_post_number", "highest_staff_post_number")
+      t.archetype
+    SQL
 
     INSERT_TOPIC_USER_SQL = "INSERT INTO topic_users (user_id, topic_id, last_read_post_number, last_visited_at, first_visited_at, notification_level)
                   SELECT :user_id, :topic_id, :post_number, :now, :now, :new_status
@@ -296,8 +298,6 @@ class TopicUser < ActiveRecord::Base
                                    FROM topic_users AS ftu
                                    WHERE ftu.user_id = :user_id and ftu.topic_id = :topic_id)"
 
-    INSERT_TOPIC_USER_SQL_STAFF = INSERT_TOPIC_USER_SQL.gsub("highest_post_number", "highest_staff_post_number")
-
     def update_last_read(user, topic_id, post_number, new_posts_read, msecs, opts = {})
       return if post_number.blank?
       msecs = 0 if msecs.to_i < 0
@@ -312,26 +312,22 @@ class TopicUser < ActiveRecord::Base
         threshold: SiteSetting.default_other_auto_track_topics_after_msecs
       }
 
-      # 86400000 = 1 day
+      rows = DB.query(UPDATE_TOPIC_USER_SQL, args)
-      rows =
-        if user.staff?
-          DB.query(UPDATE_TOPIC_USER_SQL_STAFF, args)
-        else
-          DB.query(UPDATE_TOPIC_USER_SQL, args)
-        end
 
       if rows.length == 1
         before = rows[0].old_level.to_i
         after = rows[0].notification_level.to_i
         before_last_read = rows[0].last_read_post_number.to_i
+        archetype = rows[0].archetype
 
         if before_last_read < post_number
           # The user read at least one new post
-          TopicTrackingState.publish_read(
+          publish_read(
-            topic_id,
+            topic_id: topic_id,
-            post_number,
+            post_number: post_number,
-            user,
+            user: user,
-            after
+            notification_level: after,
+            private_message: archetype == Archetype.private_message
           )
         end
 
@@ -351,21 +347,21 @@ class TopicUser < ActiveRecord::Base
           args[:new_status] = notification_levels[:tracking]
         end
 
-        TopicTrackingState.publish_read(
+        publish_read(
-          topic_id,
+          topic_id: topic_id,
-          post_number,
+          post_number: post_number,
-          user,
+          user: user,
-          args[:new_status]
+          notification_level: args[:new_status],
+          private_message: Topic.exists?(
+            archetype: Archetype.private_message,
+            id: topic_id
+          )
         )
 
         user.update_posts_read!(new_posts_read, mobile: opts[:mobile])
 
         begin
-          if user.staff?
-            DB.exec(INSERT_TOPIC_USER_SQL_STAFF, args)
-          else
           DB.exec(INSERT_TOPIC_USER_SQL, args)
-          end
         rescue PG::UniqueViolation
           # if record is inserted between two statements this can happen
           # we retry once to avoid failing the req
@@ -381,6 +377,24 @@ class TopicUser < ActiveRecord::Base
       end
     end
 
+    private
+
+    def publish_read(topic_id:, post_number:, user:, notification_level: nil, private_message:)
+      klass =
+        if private_message
+          PrivateMessageTopicTrackingState
+        else
+          TopicTrackingState
+        end
+
+      klass.publish_read(
+        topic_id,
+        post_number,
+        user,
+        notification_level
+      )
+    end
+
   end
 
   # Update the cached topic_user.liked column based on data

spec/models/private_message_topic_tracking_state_spec.rb

@@ -203,4 +203,22 @@ describe PrivateMessageTopicTrackingState do
       expect(data['payload']['group_ids']).to contain_exactly(group.id)
     end
   end
+
+  describe '.publish_read' do
+    it 'should publish the right message_bus message' do
+      message = MessageBus.track_publish(described_class.user_channel(user.id)) do
+        PrivateMessageTopicTrackingState.publish_read(private_message.id, 1, user)
+      end.first
+
+      data = message.data
+
+      expect(message.user_ids).to contain_exactly(user.id)
+      expect(message.group_ids).to eq(nil)
+      expect(data["topic_id"]).to eq(private_message.id)
+      expect(data["message_type"]).to eq(described_class::READ_MESSAGE_TYPE)
+      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(nil)
+    end
+  end
 end

spec/models/topic_user_spec.rb

@@ -226,7 +226,14 @@ describe TopicUser do
         freeze_time tomorrow
 
         Fabricate(:post, topic: topic, user: user)
+        channel = TopicTrackingState.unread_channel_key(user.id)
+
+        messages = MessageBus.track_publish(channel) do
           TopicUser.update_last_read(user, topic.id, 2, 1, 0)
+        end
+
+        expect(messages.blank?).to eq(false)
+
         topic_user = TopicUser.get(topic, user)
 
         expect(topic_user.last_read_post_number).to eq(2)
@@ -270,6 +277,20 @@ describe TopicUser do
           .to eq(TopicUser.notification_levels[:regular])
       end
 
+      it 'should publish the right message_bus message' do
+        TopicUser.update_last_read(user, topic.id, 1, 1, 0)
+
+        Fabricate(:post, topic: topic, user: user)
+
+        channel = PrivateMessageTopicTrackingState.user_channel(user.id)
+
+        messages = MessageBus.track_publish(channel) do
+          TopicUser.update_last_read(user, topic.id, 2, 1, 0)
+        end
+
+        expect(messages.blank?).to eq(false)
+      end
+
       describe 'inviting a group' do
         let(:group) do
           Fabricate(:group,