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

This reverts commit 94c3bbc2d1.

At this current point in time, we do not have enough data on whether
this centralisation is the trade-offs of coupling features into a single
channel.

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

@@ -108,19 +108,17 @@ export default {
         user.notification_channel_position
       );
 
-      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);
+      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");
-            break;
-        }
       });
 
       const site = container.lookup("site:main");

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

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

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

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

app/models/user.rb

@@ -657,38 +657,16 @@ 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)
-    publish_updates(type: PUBLISH_DO_NOT_STATUS_TYPE, payload: { ends_at: ends_at&.httpdate })
+    MessageBus.publish("/do-not-disturb/#{id}", { ends_at: ends_at&.httpdate }, user_ids: [id])
   end
 
   def publish_user_status(status)
-    publish_updates(
-      type: PUBLISH_USER_STATUS_TYPE,
-      payload: status ? { description: status.description, emoji: status.emoji } : nil
-    )
+    payload = status ?
+                { description: status.description, emoji: status.emoji } :
+                nil
+
+    MessageBus.publish("/user-status/#{id}", payload, user_ids: [id])
   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
 
-      User.publish_updates(
-        user_id: user_id,
-        type: User::PUBLISH_DRAFTS_TYPE,
-        payload: {
+      MessageBus.publish(
+        "/user-drafts/#{user_id}",
+        {
           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.publish_updates_channel(admin.id),
-            User.publish_updates_channel(admin.id),
-            User.publish_updates_channel(admin.id),
+            "/user-drafts/#{admin.id}",
+            "/user-drafts/#{admin.id}",
+            "/user-drafts/#{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.publish_updates_channel(p.user_id) }
+        draft_count = messages.find { |m| m.channel == "/user-drafts/#{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,21 @@ 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-updates/#{user.id}") do
+    messages = MessageBus.track_publish("/user-drafts/#{user.id}") do
       Draft.set(user, Draft::NEW_TOPIC, 0, "data")
     end
 
-    expect(messages.first.data[:payload][:draft_count]).to eq(1)
-    expect(messages.first.data[:payload][:has_topic_draft]).to eq(true)
+    expect(messages.first.data[:draft_count]).to eq(1)
+    expect(messages.first.data[:has_topic_draft]).to eq(true)
+    expect(messages.first.user_ids).to contain_exactly(user.id)
 
-    messages = MessageBus.track_publish("/user-updates/#{user.id}") do
+    messages = MessageBus.track_publish("/user-drafts/#{user.id}") do
       Draft.where(user: user).destroy_all
     end
 
-    expect(messages.first.data[:payload][:draft_count]).to eq(0)
-    expect(messages.first.data[:payload][:has_topic_draft]).to eq(false)
+    expect(messages.first.data[:draft_count]).to eq(0)
+    expect(messages.first.data[:has_topic_draft]).to eq(false)
+    expect(messages.first.user_ids).to contain_exactly(user.id)
   end
 
   describe '#stream' do

spec/requests/user_status_controller_spec.rb

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

spec/services/user_notification_schedule_processor_spec.rb

@@ -143,13 +143,12 @@ 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
-        messages = MessageBus.track_publish(User.publish_updates_channel(user.id)) do
+        messages = MessageBus.track_publish("/do-not-disturb/#{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].data[: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