DEV: allows stop/resume streaming on a message (#25774)

```ruby
ChatSDK::Message.start_stream(message_id: 1, guardian: guardian)
ChatSDK::Message.stream(raw: "foo", message_id: 1, guardian: guardian)
ChatSDK::Message.stream(raw: "bar", message_id: 1, guardian: guardian)
ChatSDK::Message.stop_stream(message_id: 1, guardian: guardian)
```

Generally speaking only admins or owners of the message can interact with a message.  Also note, Streaming to an existing message with a different user won't change the initial user of the message.

plugins/chat/app/services/chat/stop_message_streaming.rb

@@ -38,7 +38,8 @@ module Chat
     end
 
     def can_stop_streaming(guardian:, message:, **)
-      guardian.is_admin? || message.in_reply_to && message.in_reply_to.user_id == guardian.user.id
+      guardian.is_admin? || message.user.id == guardian.user.id ||
+        message.in_reply_to && message.in_reply_to.user_id == guardian.user.id
     end
 
     def stop_message_streaming(message:, **)

plugins/chat/assets/stylesheets/common/chat-message-text.scss

@@ -7,7 +7,7 @@
     }
 
     p::after {
-      margin-left: 3px;
+      margin-left: 1px;
       margin-bottom: -4px;
       content: "";
       width: 6px;

plugins/chat/lib/chat/guardian_extensions.rb

@@ -217,7 +217,7 @@ module Chat
     end
 
     def can_edit_chat?(message)
-      (message.user_id == @user.id && !@user.silenced?)
+      (message.user_id == @user.id && !@user.silenced?) || is_admin?
     end
 
     def can_react?

plugins/chat/lib/chat_sdk/channel.rb

@@ -26,10 +26,7 @@ module ChatSDK
         direction: "future",
       ) do
         on_success { result.messages }
-        on_failure do
-          p Chat::StepsInspector.new(result)
-          raise "Unexpected error"
-        end
+        on_failure { raise "Unexpected error" }
         on_failed_policy(:can_view_channel) { raise "Guardian can't view channel" }
         on_failed_policy(:target_message_exists) { raise "Target message doesn't exist" }
       end

plugins/chat/lib/chat_sdk/message.rb

@@ -17,13 +17,13 @@ module ChatSDK
     # @yield [helper, message] Offers a block with a helper and the message for streaming operations.
     # @yieldparam helper [Helper] The helper object for streaming operations.
     # @yieldparam message [Message] The newly created message object.
-    # @return [ChMessage] The created message object.
+    # @return [Chat::Message] The created message object.
     #
     # @example Creating a simple message
     #   ChatSDK::Message.create(raw: "Hello, world!", channel_id: 1, guardian: Guardian.new)
     #
     # @example Creating a message with a block for streaming
-    #   Message.create_with_stream(raw: "Streaming message", channel_id: 1, guardian: Guardian.new) do |helper, message|
+    #   ChatSDK::Message.create_with_stream(raw: "Streaming message", channel_id: 1, guardian: Guardian.new) do |helper, message|
     #     helper.stream(raw: "Continuation of the message")
     #   end
     def self.create(**params, &block)
@@ -40,6 +40,70 @@ module ChatSDK
       self.create(**params, streaming: true, &block)
     end
 
+    # Streams to a specific chat message.
+    #
+    # @param raw [String] text to append to the existing message.
+    # @param message_id [Integer] the ID of the message to stream.
+    # @param guardian [Guardian] an instance of the guardian class, representing the user's permissions.
+    # @return [Chat::Message] The message object.
+    # @example Streaming a message
+    #   ChatSDK::Message.stream(message_id: 42, guardian: guardian, raw: "text")
+    def self.stream(raw:, message_id:, guardian:, &block)
+      new.stream(raw: raw, message_id: message_id, guardian: guardian, &block)
+    end
+
+    # Starts streaming for a specific chat message.
+    #
+    # @param message_id [Integer] the ID of the message for which streaming should be stopped.
+    # @param guardian [Guardian] an instance of the guardian class, representing the user's permissions.
+    # @return [Chat::Message] The message object.
+    # @example Starting the streaming of a message
+    #   ChatSDK::Message.start_stream(message_id: 42, guardian: guardian)
+    def self.start_stream(message_id:, guardian:)
+      new.start_stream(message_id: message_id, guardian: guardian)
+    end
+
+    # Stops streaming for a specific chat message.
+    #
+    # @param message_id [Integer] the ID of the message for which streaming should be stopped.
+    # @param guardian [Guardian] an instance of the guardian class, representing the user's permissions.
+    # @return [Chat::Message] The message object.
+    # @example Stopping the streaming of a message
+    #   ChatSDK::Message.stop_stream(message_id: 42, guardian: guardian)
+    def self.stop_stream(message_id:, guardian:)
+      new.stop_stream(message_id: message_id, guardian: guardian)
+    end
+
+    def start_stream(message_id:, guardian:)
+      message = Chat::Message.find(message_id)
+      guardian.ensure_can_edit_chat!(message)
+      message.update!(streaming: true)
+      ::Chat::Publisher.publish_edit!(message.chat_channel, message.reload)
+      message
+    end
+
+    def stream(message_id:, raw:, guardian:, &block)
+      message = Chat::Message.find(message_id)
+      helper = StreamHelper.new(message, guardian)
+      helper.stream(raw: raw)
+      ::Chat::Publisher.publish_edit!(message.chat_channel, message.reload)
+      message
+    end
+
+    def stop_stream(message_id:, guardian:)
+      with_service(Chat::StopMessageStreaming, message_id: message_id, guardian: guardian) do
+        on_success { result.message }
+        on_model_not_found(:message) { raise "Couldn't find message with id: `#{message_id}`" }
+        on_failed_policy(:can_join_channel) do
+          raise "User with id: `#{guardian.user.id}` can't join this channel"
+        end
+        on_failed_policy(:can_stop_streaming) do
+          raise "User with id: `#{guardian.user.id}` can't stop streaming this message"
+        end
+        on_failure { raise "Unexpected error" }
+      end
+    end
+
     def create(
       raw:,
       channel_id:,
@@ -75,14 +139,11 @@ module ChatSDK
           end
           on_failed_contract { |contract| raise contract.errors.full_messages.join(", ") }
           on_success { result.message_instance }
-          on_failure do
-            p Chat::StepsInspector.new(result)
-            raise "Unexpected error"
-          end
+          on_failure { raise "Unexpected error" }
         end
 
       if streaming && block_given?
-        helper = Helper.new(message)
+        helper = StreamHelper.new(message, guardian)
         block.call(helper, message)
       end
 
@@ -95,30 +156,28 @@ module ChatSDK
     end
   end
 
-  class Helper
+  class StreamHelper
     include Chat::WithServiceHelper
 
     attr_reader :message
+    attr_reader :guardian
 
-    def initialize(message)
-      @message = message
+    def initialize(message, guardian)
+      @message = message.reload
+      @guardian = guardian
     end
 
     def stream(raw: nil)
-      return false unless self.message.reload.streaming
+      return false if !self.message.streaming
+      return false if !raw
 
       with_service(
         Chat::UpdateMessage,
         message_id: self.message.id,
-        message: raw ? self.message.reload.message + " " + raw : self.message.message,
-        guardian: self.message.user.guardian,
+        message: self.message.message + raw,
+        guardian: self.guardian,
         streaming: true,
-      ) do
-        on_failure do
-          p Chat::StepsInspector.new(result)
-          raise "Unexpected error"
-        end
-      end
+      ) { on_failure { raise "Unexpected error" } }
 
       self.message
     end

plugins/chat/lib/chat_sdk/thread.rb

@@ -48,10 +48,7 @@ module ChatSDK
         on_failed_policy(:ensure_thread_enabled) do
           raise "Threading is not enabled for this channel"
         end
-        on_failure do
-          p Chat::StepsInspector.new(result)
-          raise "Unexpected error"
-        end
+        on_failure { raise "Unexpected error" }
       end
     end
 
@@ -70,10 +67,7 @@ module ChatSDK
         end
         on_failed_contract { |contract| raise contract.errors.full_messages.join(", ") }
         on_success { result.thread_instance }
-        on_failure do
-          p Chat::StepsInspector.new(result)
-          raise "Unexpected error"
-        end
+        on_failure { raise "Unexpected error" }
       end
     end
   end

plugins/chat/spec/lib/chat/guardian_extensions_spec.rb

@@ -479,6 +479,30 @@ RSpec.describe Chat::GuardianExtensions do
       end
     end
 
+    describe "#can_edit_chat" do
+      fab!(:message) { Fabricate(:chat_message, chat_channel: channel) }
+
+      context "when user is staff" do
+        it "returns true" do
+          expect(staff_guardian.can_edit_chat?(message)).to eq(true)
+        end
+      end
+
+      context "when user is not staff" do
+        it "returns false" do
+          expect(guardian.can_edit_chat?(message)).to eq(false)
+        end
+      end
+
+      context "when user is owner of the message" do
+        fab!(:message) { Fabricate(:chat_message, chat_channel: channel, user: user) }
+
+        it "returns true" do
+          expect(guardian.can_edit_chat?(message)).to eq(true)
+        end
+      end
+    end
+
     describe "#can_delete_category?" do
       alias_matcher :be_able_to_delete_category, :be_can_delete_category
 

plugins/chat/spec/lib/chat_sdk/message_spec.rb

@@ -47,9 +47,9 @@ describe ChatSDK::Message do
 
     context "when membership is enforced" do
       it "works" do
+        SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]]
         params[:enforce_membership] = true
         params[:guardian] = Fabricate(:user).guardian
-        SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]]
 
         message = described_class.create(**params)
 
@@ -87,7 +87,7 @@ describe ChatSDK::Message do
 
           edit =
             MessageBus
-              .track_publish("/chat/#{channel_1.id}") { helper.stream(raw: "test") }
+              .track_publish("/chat/#{channel_1.id}") { helper.stream(raw: " test") }
               .find { |m| m.data["type"] == "edit" }
 
           expect(edit.data["chat_message"]["message"]).to eq("something test")
@@ -97,4 +97,88 @@ describe ChatSDK::Message do
       expect(created_message.message).to eq("something test")
     end
   end
+
+  describe ".stop_stream" do
+    fab!(:message_1) { Fabricate(:chat_message, message: "first") }
+
+    before do
+      SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]]
+      message_1.update!(streaming: true)
+    end
+
+    it "stop streaming message" do
+      described_class.stop_stream(message_id: message_1.id, guardian: message_1.user.guardian)
+
+      expect(message_1.reload.streaming).to eq(false)
+    end
+
+    context "when user can't stop streaming" do
+      it "fails" do
+        user = Fabricate(:user)
+        message_1.chat_channel.add(user)
+
+        expect {
+          described_class.stop_stream(message_id: message_1.id, guardian: user.guardian)
+        }.to raise_error("User with id: `#{user.id}` can't stop streaming this message")
+      end
+    end
+
+    context "when user can't join channel" do
+      fab!(:message_1) do
+        Fabricate(:chat_message, chat_channel: Fabricate(:private_category_channel))
+      end
+
+      it "fails" do
+        user = Fabricate(:user)
+
+        expect {
+          described_class.stop_stream(message_id: message_1.id, guardian: user.guardian)
+        }.to raise_error("User with id: `#{user.id}` can't join this channel")
+      end
+    end
+  end
+
+  describe ".start_stream" do
+    fab!(:message_1) { Fabricate(:chat_message, message: "first") }
+
+    it "enables streaming" do
+      initial_message = message_1.message
+
+      edit =
+        MessageBus
+          .track_publish("/chat/#{message_1.chat_channel.id}") do
+            described_class.start_stream(
+              message_id: message_1.id,
+              guardian: message_1.user.guardian,
+            )
+          end
+          .find { |m| m.data["type"] == "edit" }
+
+      expect(edit.data["chat_message"]["message"]).to eq("first")
+      expect(message_1.reload.streaming).to eq(true)
+    end
+  end
+
+  describe ".stream" do
+    fab!(:message_1) { Fabricate(:chat_message, message: "first") }
+    before { message_1.update!(streaming: true) }
+
+    it "streams" do
+      initial_message = message_1.message
+
+      edit =
+        MessageBus
+          .track_publish("/chat/#{message_1.chat_channel.id}") do
+            described_class.stream(
+              raw: " test",
+              message_id: message_1.id,
+              guardian: message_1.user.guardian,
+            )
+          end
+          .find { |m| m.data["type"] == "edit" }
+
+      expect(edit.data["chat_message"]["message"]).to eq("first test")
+      expect(message_1.reload.streaming).to eq(true)
+    end
+  end
 end