FIX: Mark threads read when threading enabled for a channel (#22458)

Since we create threads in the background regardless of whether
threading is enabled for a channel, we get the unexpected behaviour
of everyone having a lot of unread threads when threading is enabled
for the channel.

To counteract this, when the admin enables threads for a channel
we can just run a high priority background job to mark all threads
as read in the channel for all users, so they are essentially
starting from a clean slate.

plugins/chat/spec/jobs/regular/mark_all_channel_threads_read_spec.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+RSpec.describe Jobs::Chat::MarkAllChannelThreadsRead do
+  fab!(:channel) { Fabricate(:chat_channel, threading_enabled: true) }
+
+  context "when enable_experimental_chat_threaded_discussions is false" do
+    before { SiteSetting.enable_experimental_chat_threaded_discussions = false }
+
+    it "does nothing" do
+      Chat::Channel.any_instance.expects(:mark_all_threads_as_read).never
+      described_class.new.execute(channel_id: channel.id)
+    end
+  end
+
+  context "when enable_experimental_chat_threaded_discussions is true" do
+    fab!(:thread_1) { Fabricate(:chat_thread, channel: channel) }
+    fab!(:thread_2) { Fabricate(:chat_thread, channel: channel) }
+    fab!(:user_1) { Fabricate(:user) }
+    fab!(:user_2) { Fabricate(:user) }
+    fab!(:thread_1_message_1) { Fabricate(:chat_message, thread: thread_1) }
+    fab!(:thread_1_message_2) { Fabricate(:chat_message, thread: thread_1) }
+    fab!(:thread_1_message_3) { Fabricate(:chat_message, thread: thread_1) }
+    fab!(:thread_2_message_1) { Fabricate(:chat_message, thread: thread_2) }
+    fab!(:thread_2_message_2) { Fabricate(:chat_message, thread: thread_2) }
+
+    before do
+      SiteSetting.enable_experimental_chat_threaded_discussions = true
+      channel.add(user_1)
+      channel.add(user_2)
+      thread_1.add(user_1)
+      thread_2.add(user_2)
+    end
+
+    def unread_count(user)
+      Chat::ThreadUnreadsQuery.call(channel_ids: [channel.id], user_id: user.id).first.unread_count
+    end
+
+    it "marks all threads as read across all users in the channel" do
+      expect(unread_count(user_1)).to eq(3)
+      expect(unread_count(user_2)).to eq(2)
+      described_class.new.execute(channel_id: channel.id)
+      expect(unread_count(user_1)).to eq(0)
+      expect(unread_count(user_2)).to eq(0)
+    end
+  end
+end