DEV: Centralize user updates to a single MessageBus channel. (#17058)

Introduces an interface to publish user updates on the server side and
helps to reduce the growing number of subscriptions on the client side.

app/assets/javascripts/discourse/app/initializers/subscribe-user-notifications.js

@@ -108,17 +108,19 @@ export default {
         user.notification_channel_position
       );
 
-      bus.subscribe(`/user-drafts/${user.id}`, (data) => {
-        user.updateDraftProperties(data);
-      });
-
-      bus.subscribe(`/do-not-disturb/${user.get("id")}`, (data) => {
-        user.updateDoNotDisturbStatus(data.ends_at);
-      });
-
-      bus.subscribe(`/user-status/${user.id}`, (data) => {
-        user.set("status", data);
-        appEvents.trigger("user-status:changed");
+      bus.subscribe(`/user-updates/${user.id}`, (data) => {
+        switch (data.type) {
+          case "drafts":
+            user.updateDraftProperties(data.payload);
+            break;
+          case "do_not_disturb":
+            user.updateDoNotDisturbStatus(data.payload.ends_at);
+            break;
+          case "user_status":
+            user.set("status", data.payload);
+            appEvents.trigger("user-status:changed");
+            break;
+        }
       });
 
       const site = container.lookup("site:main");

app/assets/javascripts/discourse/tests/acceptance/sidebar-topics-section-test.js

@@ -242,8 +242,11 @@ acceptance("Sidebar - Topics Section", function (needs) {
     async function (assert) {
       await visit("/t/280");
 
-      publishToMessageBus(`/user-drafts/${loggedInUser().id}`, {
-        draft_count: 1,
+      publishToMessageBus(`/user-updates/${loggedInUser().id}`, {
+        type: "drafts",
+        payload: {
+          draft_count: 1,
+        },
       });
 
       await settled();

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

@@ -17,13 +17,17 @@ acceptance("User Status", function (needs) {
 
   needs.pretender((server, helper) => {
     server.put("/user-status.json", () => {
-      publishToMessageBus(`/user-status/${userId}`, {
-        description: userStatus,
+      publishToMessageBus(`/user-updates/${userId}`, {
+        type: "user_status",
+        payload: { description: userStatus },
       });
       return helper.response({ success: true });
     });
     server.delete("/user-status.json", () => {
-      publishToMessageBus(`/user-status/${userId}`, null);
+      publishToMessageBus(`/user-updates/${userId}`, {
+        type: "user_status",
+        payload: null,
+      });
       return helper.response({ success: true });
     });
   });

app/models/user.rb

@@ -657,16 +657,38 @@ class User < ActiveRecord::Base
     MessageBus.publish("/notification/#{id}", payload, user_ids: [id])
   end
 
+  PUBLISH_USER_STATUS_TYPE = "user_status"
+  PUBLISH_DO_NOT_STATUS_TYPE = "do_not_disturb"
+  PUBLISH_DRAFTS_TYPE = "drafts"
+
+  def self.publish_updates_channel(user_id)
+    "/user-updates/#{user_id}"
+  end
+
+  def self.publish_updates(user_id:, type:, payload:)
+    MessageBus.publish(
+      publish_updates_channel(user_id),
+      {
+        type: type,
+        payload: payload
+      },
+      user_ids: [user_id]
+    )
+  end
+
+  def publish_updates(type:, payload:)
+    self.class.publish_updates(user_id: id, type: type, payload: payload)
+  end
+
   def publish_do_not_disturb(ends_at: nil)
-    MessageBus.publish("/do-not-disturb/#{id}", { ends_at: ends_at&.httpdate }, user_ids: [id])
+    publish_updates(type: PUBLISH_DO_NOT_STATUS_TYPE, payload: { ends_at: ends_at&.httpdate })
   end
 
   def publish_user_status(status)
-    payload = status ?
-                { description: status.description, emoji: status.emoji } :
-                nil
-
-    MessageBus.publish("/user-status/#{id}", payload, user_ids: [id])
+    publish_updates(
+      type: PUBLISH_USER_STATUS_TYPE,
+      payload: status ? { description: status.description, emoji: status.emoji } : nil
+    )
   end
 
   def password=(password)

app/models/user_stat.rb

@@ -214,13 +214,13 @@ class UserStat < ActiveRecord::Base
         RETURNING draft_count, (SELECT 1 FROM drafts WHERE user_id = :user_id AND draft_key = :new_topic)
       SQL
 
-      MessageBus.publish(
-        "/user-drafts/#{user_id}",
-        {
+      User.publish_updates(
+        user_id: user_id,
+        type: User::PUBLISH_DRAFTS_TYPE,
+        payload: {
           draft_count: draft_count,
           has_topic_draft: !!has_topic_draft
-        },
-        user_ids: [user_id]
+        }
       )
     else
       DB.exec <<~SQL

spec/lib/post_creator_spec.rb

@@ -174,9 +174,9 @@ describe PostCreator do
             "/latest",
             "/topic/#{created_post.topic_id}",
             "/topic/#{created_post.topic_id}",
-            "/user-drafts/#{admin.id}",
-            "/user-drafts/#{admin.id}",
-            "/user-drafts/#{admin.id}",
+            User.publish_updates_channel(admin.id),
+            User.publish_updates_channel(admin.id),
+            User.publish_updates_channel(admin.id),
           ].sort
         )
 
@@ -205,7 +205,7 @@ describe PostCreator do
         user_action = messages.find { |m| m.channel == "/u/#{p.user.username}" }
         expect(user_action).not_to eq(nil)
 
-        draft_count = messages.find { |m| m.channel == "/user-drafts/#{p.user_id}" }
+        draft_count = messages.find { |m| m.channel == User.publish_updates_channel(p.user_id) }
         expect(draft_count).not_to eq(nil)
 
         expect(messages.filter { |m| m.channel != "/distributed_hash" }.length).to eq(7)

spec/models/draft_spec.rb

@@ -179,19 +179,19 @@ describe Draft do
   it 'updates draft count when a draft is created or destroyed' do
     Draft.set(Fabricate(:user), Draft::NEW_TOPIC, 0, "data")
 
-    messages = MessageBus.track_publish("/user-drafts/#{user.id}") do
+    messages = MessageBus.track_publish("/user-updates/#{user.id}") do
       Draft.set(user, Draft::NEW_TOPIC, 0, "data")
     end
 
-    expect(messages.first.data[:draft_count]).to eq(1)
-    expect(messages.first.data[:has_topic_draft]).to eq(true)
+    expect(messages.first.data[:payload][:draft_count]).to eq(1)
+    expect(messages.first.data[:payload][:has_topic_draft]).to eq(true)
 
-    messages = MessageBus.track_publish("/user-drafts/#{user.id}") do
+    messages = MessageBus.track_publish("/user-updates/#{user.id}") do
       Draft.where(user: user).destroy_all
     end
 
-    expect(messages.first.data[:draft_count]).to eq(0)
-    expect(messages.first.data[:has_topic_draft]).to eq(false)
+    expect(messages.first.data[:payload][:draft_count]).to eq(0)
+    expect(messages.first.data[:payload][:has_topic_draft]).to eq(false)
   end
 
   describe '#stream' do

spec/requests/user_status_controller_spec.rb

@@ -50,11 +50,16 @@ describe UserStatusController do
 
       it "publishes to message bus" do
         status = "off to dentist"
-        messages = MessageBus.track_publish { put "/user-status.json", params: { description: status } }
+
+        messages = MessageBus.track_publish(User.publish_updates_channel(user.id)) do
+          put "/user-status.json", params: { description: status }
+          expect(response.status).to eq(200)
+        end
 
         expect(messages.size).to eq(1)
-        expect(messages[0].channel).to eq("/user-status/#{user.id}")
-        expect(messages[0].data[:description]).to eq(status)
+        expect(messages[0].data[:type]).to eq(User::PUBLISH_USER_STATUS_TYPE)
+        expect(messages[0].data[:payload][:description]).to eq(status)
+        expect(messages[0].user_ids).to eq([user.id])
       end
     end
   end
@@ -92,11 +97,16 @@ describe UserStatusController do
       end
 
       it "publishes to message bus" do
-        messages = MessageBus.track_publish { delete "/user-status.json" }
+        messages = MessageBus.track_publish(User.publish_updates_channel(user.id)) do
+          delete "/user-status.json"
+
+          expect(response.status).to eq(200)
+        end
 
         expect(messages.size).to eq(1)
-        expect(messages[0].channel).to eq("/user-status/#{user.id}")
-        expect(messages[0].data).to eq(nil)
+        expect(messages[0].data[:type]).to eq(User::PUBLISH_USER_STATUS_TYPE)
+        expect(messages[0].data[:payload]).to eq(nil)
+        expect(messages[0].user_ids).to eq([user.id])
       end
     end
   end

spec/services/user_notification_schedule_processor_spec.rb

@@ -143,12 +143,14 @@ describe UserNotificationScheduleProcessor do
       user.user_option.update(timezone: "UTC")
       schedule = standard_schedule
       travel_to Time.new(2020, 12, 31, 1, 0, 0, "+00:00") do
-        MessageBus.expects(:publish).with(
-          "/do-not-disturb/#{user.id}",
-          { ends_at: Time.new(2020, 12, 31, 7, 59, 0, "+00:00").httpdate },
-          user_ids: [user.id]
-        )
-        UserNotificationScheduleProcessor.create_do_not_disturb_timings_for(schedule)
+        messages = MessageBus.track_publish(User.publish_updates_channel(user.id)) do
+          UserNotificationScheduleProcessor.create_do_not_disturb_timings_for(schedule)
+        end
+
+        expect(messages.size).to eq(1)
+        expect(messages[0].data[:type]).to eq(User::PUBLISH_DO_NOT_STATUS_TYPE)
+        expect(messages[0].data[:payload][:ends_at]).to eq(Time.new(2020, 12, 31, 7, 59, 0, "+00:00").httpdate)
+        expect(messages[0].user_ids).to contain_exactly(user.id)
       end
     end
   end