FEATURE: Create and update thread memberships (#21501)

When the user sends a message in a thread, we want to
create a membership for them in the background (default
to notification level of Watching) so we can track whether
they have read the thread.

Then, for now since we don't have granular message reading/
scrolling in the thread panel, we just update the thread
last_read_message_id for the user to the latest reply in the
thread when they open the thread panel. This at least will
mark the thread as read.

In future PRs we want to show the blue dot indicator in various
places in the UI for unread threads which will also require
some MessageBus functionality.

This takes into account the same issue fixed for channels
in ae3231e1406d4ccf1d048ef8a8d8f744f840896b

plugins/chat/app/services/chat/action/reset_user_last_read_channel_message.rb

@@ -4,7 +4,7 @@ module Chat
   module Action
     class ResetUserLastReadChannelMessage
       # @param [Array] last_read_message_ids The message IDs to match with the
-      #   last_read_message_ids in UserChannelMembership which will be reset
+      #   last_read_message_ids in UserChatChannelMembership which will be reset
       #   to NULL or the most recent non-deleted message in the channel to
       #   update read state.
       # @param [Integer] channel_ids The channel IDs of the memberships to update,

plugins/chat/app/services/chat/action/reset_user_last_read_thread_message.rb

@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+module Chat
+  module Action
+    class ResetUserLastReadThreadMessage
+      # @param [Array] last_read_message_ids The message IDs to match with the
+      #   last_read_message_ids in UserChatThreadMembership which will be reset
+      #   to NULL or the most recent non-deleted message in the thread to
+      #   update read state.
+      # @param [Integer] thread_ids The thread IDs of the memberships to update,
+      #   this is used to find the latest non-deleted message in the thread.
+      def self.call(last_read_message_ids, thread_ids)
+        sql = <<~SQL
+         -- update the last_read_message_id to the most recent
+         -- non-deleted message in the thread so unread counts are correct.
+         -- the cte row_number is necessary to only return a single row
+         -- for each thread to prevent additional data being returned
+         WITH cte AS (
+           SELECT * FROM (
+             SELECT id, thread_id, row_number() OVER (
+                 PARTITION BY thread_id ORDER BY created_at DESC, id DESC
+               ) AS row_number
+             FROM chat_messages
+             WHERE deleted_at IS NULL AND thread_id IN (:thread_ids) AND chat_messages.id NOT IN (
+               SELECT original_message_id FROM chat_threads WHERE thread_id IN (:thread_ids)
+             )
+           ) AS recent_messages
+           WHERE recent_messages.row_number = 1
+         )
+         UPDATE user_chat_thread_memberships
+         SET last_read_message_id = cte.id
+         FROM cte
+         WHERE user_chat_thread_memberships.last_read_message_id IN (:last_read_message_ids)
+         AND cte.thread_id = user_chat_thread_memberships.thread_id;
+
+          -- then reset all last_read_message_ids to null
+          -- for the cases where all messages in the thread were
+          -- already deleted
+          UPDATE user_chat_thread_memberships
+          SET last_read_message_id = NULL
+          WHERE last_read_message_id IN (:last_read_message_ids);
+        SQL
+
+        DB.exec(sql, last_read_message_ids: last_read_message_ids, thread_ids: thread_ids)
+      end
+    end
+  end
+end

plugins/chat/app/services/chat/trash_message.rb

@@ -59,6 +59,9 @@ module Chat
 
     def update_tracking_state(message:, **)
       ::Chat::Action::ResetUserLastReadChannelMessage.call([message.id], [message.chat_channel_id])
+      if message.thread_id.present?
+        ::Chat::Action::ResetUserLastReadThreadMessage.call([message.id], [message.thread_id])
+      end
     end
 
     def update_thread_reply_cache(message:, **)

plugins/chat/app/services/chat/update_user_thread_last_read.rb

@@ -3,8 +3,9 @@
 module Chat
   # Service responsible for marking messages in a thread
   # as read. For now this just marks any mentions in the thread
-  # as read, but as we add user tracking state to threads it
-  # will work in a similar way to Chat::UpdateUserLastRead
+  # as read and marks the entire thread as read.
+  # As we add finer-grained user tracking state to threads it
+  # will work in a similar way to Chat::UpdateUserLastRead.
   #
   # @example
   #  Chat::UpdateUserThreadLastRead.call(channel_id: 2, thread_id: 3, guardian: guardian)
@@ -22,6 +23,7 @@ module Chat
     model :thread
     policy :invalid_access
     step :mark_associated_mentions_as_read
+    step :mark_thread_read
     step :publish_new_last_read_to_clients
 
     # @!visibility private
@@ -42,6 +44,23 @@ module Chat
       guardian.can_join_chat_channel?(thread.channel)
     end
 
+    # NOTE: In future we will pass in the last_read_message_id
+    # to the service and this query will be unnecessary.
+    def mark_thread_read(thread:, guardian:, **)
+      query = <<~SQL
+        UPDATE user_chat_thread_memberships
+        SET last_read_message_id = (
+          SELECT id FROM chat_messages
+          WHERE thread_id = :thread_id
+          AND deleted_at IS NULL
+          ORDER BY created_at DESC, id DESC
+          LIMIT 1
+        )
+        WHERE user_id = :user_id AND thread_id = :thread_id
+      SQL
+      DB.exec(query, thread_id: thread.id, user_id: guardian.user.id)
+    end
+
     def mark_associated_mentions_as_read(thread:, guardian:, **)
       ::Chat::Action::MarkMentionsRead.call(
         guardian.user,

plugins/chat/lib/chat/message_creator.rb

@@ -68,7 +68,7 @@ module Chat
           @staged_id,
           staged_thread_id: @staged_thread_id,
         )
-        resolved_thread&.increment_replies_count_cache
+        post_process_resolved_thread
         Jobs.enqueue(Jobs::Chat::ProcessMessage, { chat_message_id: @chat_message.id })
         Chat::Notifier.notify_new(chat_message: @chat_message, timestamp: @chat_message.created_at)
         @chat_channel.touch(:last_message_sent_at)
@@ -228,5 +228,11 @@ module Chat
     def resolved_thread
       @existing_thread || @chat_message.thread
     end
+
+    def post_process_resolved_thread
+      return if resolved_thread.blank?
+      resolved_thread.increment_replies_count_cache
+      Chat::UserChatThreadMembership.find_or_create_by!(user: @user, thread: resolved_thread)
+    end
   end
 end

plugins/chat/spec/components/chat/message_creator_spec.rb

@@ -440,6 +440,23 @@ describe Chat::MessageCreator do
         expect(message.thread.original_message_user).to eq(reply_message.user)
       end
 
+      it "creates a user thread membership" do
+        message = nil
+        expect {
+          message =
+            described_class.create(
+              chat_channel: public_chat_channel,
+              user: user1,
+              content: "this is a message",
+              in_reply_to_id: reply_message.id,
+            ).chat_message
+        }.to change { Chat::UserChatThreadMembership.count }
+
+        expect(
+          Chat::UserChatThreadMembership.exists?(user: user1, thread: message.thread),
+        ).to be_truthy
+      end
+
       context "when threading is enabled" do
         it "publishes the new thread" do
           public_chat_channel.update!(threading_enabled: true)
@@ -522,6 +539,33 @@ describe Chat::MessageCreator do
           expect(message.reload.thread).to eq(existing_thread)
         end
 
+        it "creates a user thread membership if one does not exist" do
+          expect {
+            described_class.create(
+              chat_channel: public_chat_channel,
+              user: user1,
+              content: "this is a message",
+              thread_id: existing_thread.id,
+            ).chat_message
+          }.to change { Chat::UserChatThreadMembership.count }
+
+          expect(
+            Chat::UserChatThreadMembership.exists?(user: user1, thread: existing_thread),
+          ).to be_truthy
+        end
+
+        it "does not create a thread membership if one exists" do
+          Fabricate(:user_chat_thread_membership, user: user1, thread: existing_thread)
+          expect {
+            described_class.create(
+              chat_channel: public_chat_channel,
+              user: user1,
+              content: "this is a message",
+              thread_id: existing_thread.id,
+            ).chat_message
+          }.not_to change { Chat::UserChatThreadMembership.count }
+        end
+
         it "errors when the thread ID is for a different channel" do
           other_channel_thread = Fabricate(:chat_thread, channel: Fabricate(:chat_channel))
           result =

plugins/chat/spec/services/actions/reset_user_last_read_thread_message_spec.rb

@@ -0,0 +1,83 @@
+# frozen_string_literal: true
+
+RSpec.describe Chat::Action::ResetUserLastReadThreadMessage do
+  fab!(:thread_1) { Fabricate(:chat_thread) }
+  fab!(:thread_2) { Fabricate(:chat_thread) }
+  fab!(:message_1) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_1.channel,
+      thread: thread_1,
+      created_at: 1.hour.ago,
+    )
+  end
+  fab!(:message_2) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_1.channel,
+      thread: thread_1,
+      created_at: 2.seconds.ago,
+    )
+  end
+  fab!(:message_3) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_1.channel,
+      thread: thread_1,
+      created_at: 3.minutes.ago,
+    )
+  end
+  fab!(:message_4) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_2.channel,
+      thread: thread_2,
+      created_at: 30.seconds.ago,
+    )
+  end
+  fab!(:message_5) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_2.channel,
+      thread: thread_2,
+      created_at: 3.seconds.ago,
+    )
+  end
+  fab!(:message_6) do
+    Fabricate(
+      :chat_message,
+      chat_channel: thread_2.channel,
+      thread: thread_2,
+      created_at: 1.day.ago,
+    )
+  end
+  fab!(:membership_1) do
+    Fabricate(:user_chat_thread_membership, thread: thread_1, last_read_message_id: message_3.id)
+  end
+  fab!(:membership_2) do
+    Fabricate(:user_chat_thread_membership, thread: thread_2, last_read_message_id: message_6.id)
+  end
+
+  context "when there are non-deleted messages left in the thread" do
+    before do
+      message_3.trash!
+      message_6.trash!
+    end
+
+    it "sets the matching membership last_read_message_ids to the most recently created message ID" do
+      described_class.call([message_3.id, message_6.id], [thread_1.id, thread_2.id])
+      expect(membership_1.reload.last_read_message_id).to eq(message_2.id)
+      expect(membership_2.reload.last_read_message_id).to eq(message_5.id)
+    end
+  end
+
+  context "when there are no more non-deleted messages left in the thread (excluding the original message)" do
+    before { [message_1, message_2, message_4, message_5].each(&:trash!) }
+
+    it "sets the matching membership last_read_message_ids to NULL" do
+      described_class.call([message_3.id, message_6.id], [thread_1.id, thread_2.id])
+      expect(membership_1.reload.last_read_message_id).to be_nil
+      expect(membership_2.reload.last_read_message_id).to be_nil
+    end
+  end
+end

plugins/chat/spec/services/chat/trash_message_spec.rb

@@ -119,6 +119,35 @@ RSpec.describe Chat::TrashMessage do
             result
             expect(thread.replies_count_cache).to eq(4)
           end
+
+          it "updates the tracking to the last non-deleted thread message for users whose last_read_message_id was the trashed message" do
+            other_message =
+              Fabricate(:chat_message, chat_channel: message.chat_channel, thread: thread)
+            membership_1 =
+              Fabricate(:user_chat_thread_membership, thread: thread, last_read_message: message)
+            membership_2 =
+              Fabricate(:user_chat_thread_membership, thread: thread, last_read_message: message)
+            membership_3 =
+              Fabricate(
+                :user_chat_thread_membership,
+                thread: thread,
+                last_read_message: other_message,
+              )
+            result
+            expect(membership_1.reload.last_read_message_id).to eq(other_message.id)
+            expect(membership_2.reload.last_read_message_id).to eq(other_message.id)
+            expect(membership_3.reload.last_read_message_id).to eq(other_message.id)
+          end
+
+          it "updates the tracking to nil when there are no other messages left in the thread" do
+            membership_1 =
+              Fabricate(:user_chat_thread_membership, thread: thread, last_read_message: message)
+            membership_2 =
+              Fabricate(:user_chat_thread_membership, thread: thread, last_read_message: message)
+            result
+            expect(membership_1.reload.last_read_message_id).to be_nil
+            expect(membership_2.reload.last_read_message_id).to be_nil
+          end
         end
 
         context "when message is already deleted" do

plugins/chat/spec/services/chat/update_user_thread_last_read_spec.rb

@@ -12,6 +12,8 @@ RSpec.describe Chat::UpdateUserThreadLastRead do
     fab!(:current_user) { Fabricate(:user) }
     fab!(:channel) { Fabricate(:chat_channel) }
     fab!(:thread) { Fabricate(:chat_thread, channel: channel) }
+    fab!(:thread_reply_1) { Fabricate(:chat_message, chat_channel: channel, thread: thread) }
+    fab!(:thread_reply_2) { Fabricate(:chat_message, chat_channel: channel, thread: thread) }
 
     let(:guardian) { Guardian.new(current_user) }
     let(:params) { { guardian: guardian, channel_id: channel.id, thread_id: thread.id } }
@@ -85,6 +87,17 @@ RSpec.describe Chat::UpdateUserThreadLastRead do
         it "publishes new last read to clients" do
           expect(messages.map(&:channel)).to include("/chat/user-tracking-state/#{current_user.id}")
         end
+
+        context "when the user is a member of the thread" do
+          fab!(:membership) do
+            Fabricate(:user_chat_thread_membership, user: current_user, thread: thread)
+          end
+
+          it "updates the last_read_message_id of the thread" do
+            result
+            expect(membership.reload.last_read_message_id).to eq(thread.replies.last.id)
+          end
+        end
       end
     end
   end