DEV: chat streaming (#25736)

This commit introduces the possibility to stream messages. To allow plugins to use streaming this commit also ships a `ChatSDK` library to allow to interact with few parts of discourse chat.

```ruby
ChatSDK::Message.create_with_stream(raw: "test") do |helper|
  5.times do |i|
    is_streaming = helper.stream(raw: "more #{i}")
    next if !is_streaming
    sleep 2
  end
end
```

This commit also introduces all the frontend parts:
- messages can now be marked as streaming
- when streaming their content will be updated when a new content is appended
- a special UI will be showing (a blinking indicator)
- a cancel button allows the user to stop the streaming, when cancelled `helper.stream(...)` will return `false`, and the plugin can decide exit early

config/initializers/000-zeitwerk.rb

@@ -44,6 +44,7 @@ Rails.autoloaders.each do |autoloader|
     "ssrf_detector" => "SSRFDetector",
     "http" => "HTTP",
     "gc_stat_instrumenter" => "GCStatInstrumenter",
+    "chat_sdk" => "ChatSDK",
   )
 end
 Rails.autoloaders.main.ignore(

plugins/chat/app/controllers/chat/api/channels_messages_streaming_controller.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+class Chat::Api::ChannelsMessagesStreamingController < Chat::Api::ChannelsController
+  def destroy
+    with_service(Chat::StopMessageStreaming) do
+      on_model_not_found(:message) { raise Discourse::NotFound }
+      on_failed_policy(:can_join_channel) { raise Discourse::InvalidAccess }
+      on_failed_policy(:can_stop_streaming) { raise Discourse::InvalidAccess }
+    end
+  end
+end

plugins/chat/app/helpers/chat/with_service_helper.rb

@@ -19,7 +19,10 @@ module Chat
     end
 
     def run_service(service, dependencies)
-      @_result = service.call(params.to_unsafe_h.merge(guardian: guardian, **dependencies))
+      params = self.try(:params) || ActionController::Parameters.new
+
+      @_result =
+        service.call(params.to_unsafe_h.merge(guardian: self.try(:guardian) || nil, **dependencies))
     end
 
     def default_actions_for_service

plugins/chat/app/serializers/chat/message_serializer.rb

@@ -12,6 +12,7 @@ module Chat
       deleted_by_id
       thread_id
       chat_channel_id
+      streaming
     ]
     attributes(
       *(

plugins/chat/app/services/chat/create_message.rb

@@ -58,6 +58,8 @@ module Chat
       attribute :staged_id, :string
       attribute :upload_ids, :array
       attribute :thread_id, :string
+      attribute :streaming, :boolean, default: false
+      attribute :enforce_membership, :boolean, default: false
       attribute :incoming_chat_webhook
       attribute :process_inline, :boolean, default: Rails.env.test?
 
@@ -75,12 +77,8 @@ module Chat
       Chat::Channel.find_by_id_or_slug(contract.chat_channel_id)
     end
 
-    def allowed_to_join_channel(guardian:, channel:, **)
-      guardian.can_join_chat_channel?(channel)
-    end
-
-    def enforce_system_membership(guardian:, channel:, **)
-      if guardian.user&.is_system_user?
+    def enforce_system_membership(guardian:, channel:, contract:, **)
+      if guardian.user&.is_system_user? || contract.enforce_membership
         channel.add(guardian.user)
 
         if channel.direct_message_channel?
@@ -89,6 +87,10 @@ module Chat
       end
     end
 
+    def allowed_to_join_channel(guardian:, channel:, **)
+      guardian.can_join_chat_channel?(channel)
+    end
+
     def fetch_channel_membership(guardian:, channel:, **)
       Chat::ChannelMembershipManager.new(channel).find_for_user(guardian.user)
     end
@@ -138,6 +140,7 @@ module Chat
         thread: thread,
         cooked: ::Chat::Message.cook(contract.message, user_id: guardian.user.id),
         cooked_version: ::Chat::Message::BAKED_VERSION,
+        streaming: contract.streaming,
       )
     end
 

plugins/chat/app/services/chat/list_channel_thread_messages.rb

@@ -11,10 +11,8 @@ module Chat
     include Service::Base
 
     # @!method call(guardian:)
-    #   @param [Integer] channel_id
     #   @param [Guardian] guardian
     #   @option optional_params [Integer] thread_id
-    #   @option optional_params [Integer] channel_id
     #   @return [Service::Base::Context]
 
     contract

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

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module Chat
+  # Service responsible for stopping streaming of a message.
+  #
+  # @example
+  #  Chat::StopMessageStreaming.call(message_id: 3, guardian: guardian)
+  #
+  class StopMessageStreaming
+    include ::Service::Base
+
+    # @!method call(message_id:, guardian:)
+    #   @param [Integer] message_id
+    #   @param [Guardian] guardian
+    #   @return [Service::Base::Context]
+    contract
+    model :message
+    policy :can_join_channel
+    policy :can_stop_streaming
+    step :stop_message_streaming
+    step :publish_message_streaming_state
+
+    # @!visibility private
+    class Contract
+      attribute :message_id, :integer
+
+      validates :message_id, presence: true
+    end
+
+    private
+
+    def fetch_message(contract:, **)
+      ::Chat::Message.find_by(id: contract.message_id)
+    end
+
+    def can_join_channel(guardian:, message:, **)
+      guardian.can_join_chat_channel?(message.chat_channel)
+    end
+
+    def can_stop_streaming(guardian:, message:, **)
+      guardian.is_admin? || message.in_reply_to && message.in_reply_to.user_id == guardian.user.id
+    end
+
+    def stop_message_streaming(message:, **)
+      message.update!(streaming: false)
+    end
+
+    def publish_message_streaming_state(guardian:, message:, contract:, **)
+      ::Chat::Publisher.publish_edit!(message.chat_channel, message)
+    end
+  end
+end

plugins/chat/app/services/chat/update_message.rb

@@ -38,6 +38,8 @@ module Chat
 
       attribute :upload_ids, :array
 
+      attribute :streaming, :boolean, default: false
+
       attribute :process_inline, :boolean, default: Rails.env.test?
     end
 
@@ -98,6 +100,8 @@ module Chat
     end
 
     def save_revision(message:, guardian:, **)
+      return false if message.streaming_before_last_save
+
       prev_message = message.message_before_last_save || message.message_was
       return if !should_create_revision(message, prev_message, guardian)
 
@@ -135,6 +139,7 @@ module Chat
       edit_timestamp = context.revision&.created_at&.iso8601(6) || Time.zone.now.iso8601(6)
 
       ::Chat::Publisher.publish_edit!(message.chat_channel, message)
+
       DiscourseEvent.trigger(:chat_message_edited, message, message.chat_channel, message.user)
 
       if contract.process_inline

plugins/chat/app/services/chat/update_thread.rb

@@ -7,7 +7,7 @@ module Chat
   # Only the thread title can be updated.
   #
   # @example
-  #  Chat::UpdateThread.call(thread_id: 88, channel_id: 2, guardian: guardian, title: "Restaurant for Saturday")
+  #  Chat::UpdateThread.call(thread_id: 88, guardian: guardian, title: "Restaurant for Saturday")
   #
   class UpdateThread
     include Service::Base
@@ -30,17 +30,16 @@ module Chat
     # @!visibility private
     class Contract
       attribute :thread_id, :integer
-      attribute :channel_id, :integer
       attribute :title, :string
 
-      validates :thread_id, :channel_id, presence: true
+      validates :thread_id, presence: true
       validates :title, length: { maximum: Chat::Thread::MAX_TITLE_LENGTH }
     end
 
     private
 
     def fetch_thread(contract:, **)
-      Chat::Thread.find_by(id: contract.thread_id, channel_id: contract.channel_id)
+      Chat::Thread.find_by(id: contract.thread_id)
     end
 
     def can_view_channel(guardian:, thread:, **)

plugins/chat/assets/javascripts/discourse/components/chat-message.gjs

@@ -2,6 +2,7 @@ import Component from "@glimmer/component";
 import { tracked } from "@glimmer/tracking";
 import { getOwner } from "@ember/application";
 import { Input } from "@ember/component";
+import { fn } from "@ember/helper";
 import { on } from "@ember/modifier";
 import { action } from "@ember/object";
 import didInsert from "@ember/render-modifiers/modifiers/did-insert";
@@ -484,6 +485,19 @@ export default class ChatMessage extends Component {
     return this.args.context === MESSAGE_CONTEXT_THREAD;
   }
 
+  get shouldRenderStopMessageStreamingButton() {
+    return (
+      this.args.message.streaming &&
+      (this.currentUser.admin ||
+        this.args.message.user.id === this.currentUser.id)
+    );
+  }
+
+  @action
+  stopMessageStreaming(message) {
+    this.chatApi.stopMessageStreaming(message.channel.id, message.id);
+  }
+
   #teardownMentionedUsers() {
     this.args.message.mentionedUsers.forEach((user) => {
       user.statusManager.stopTrackingStatus();
@@ -504,6 +518,7 @@ export default class ChatMessage extends Component {
           "chat-message-container"
           (if this.pane.selectingMessages "-selectable")
           (if @message.highlighted "-highlighted")
+          (if @message.streaming "-streaming")
           (if (eq @message.user.id this.currentUser.id) "is-by-current-user")
           (if @message.staged "-staged" "-persisted")
           (if @message.processed "-processed" "-not-processed")
@@ -607,6 +622,18 @@ export default class ChatMessage extends Component {
                   {{/if}}
                 </ChatMessageText>
 
+                {{#if this.shouldRenderStopMessageStreamingButton}}
+                  <div class="stop-streaming-btn-container">
+                    <DButton
+                      @class="stop-streaming-btn"
+                      @icon="stop-circle"
+                      @label="cancel"
+                      @action={{fn this.stopMessageStreaming @message}}
+                    />
+
+                  </div>
+                {{/if}}
+
                 <ChatMessageError
                   @message={{@message}}
                   @onRetry={{@resendStagedMessage}}

plugins/chat/assets/javascripts/discourse/lib/chat-channel-subscription-manager.js

@@ -147,6 +147,7 @@ export default class ChatChannelSubscriptionManager {
       message.excerpt = data.chat_message.excerpt;
       message.uploads = cloneJSON(data.chat_message.uploads || []);
       message.edited = data.chat_message.edited;
+      message.streaming = data.chat_message.streaming;
     }
   }
 

plugins/chat/assets/javascripts/discourse/lib/chat-channel-thread-subscription-manager.js

@@ -135,6 +135,7 @@ export default class ChatChannelThreadSubscriptionManager {
       message.excerpt = data.chat_message.excerpt;
       message.uploads = cloneJSON(data.chat_message.uploads || []);
       message.edited = data.chat_message.edited;
+      message.streaming = data.chat_message.streaming;
     }
   }
 

plugins/chat/assets/javascripts/discourse/models/chat-message.js

@@ -51,6 +51,7 @@ export default class ChatMessage {
   @tracked message;
   @tracked manager;
   @tracked deletedById;
+  @tracked streaming = false;
 
   @tracked _deletedAt;
   @tracked _cooked;
@@ -59,6 +60,7 @@ export default class ChatMessage {
   constructor(channel, args = {}) {
     this.id = args.id;
     this.channel = channel;
+    this.streaming = args.streaming;
     this.manager = args.manager;
     this.newest = args.newest || false;
     this.draftSaved = args.draftSaved || args.draft_saved || false;

plugins/chat/assets/javascripts/discourse/services/chat-api.js

@@ -193,6 +193,18 @@ export default class ChatApi extends Service {
     });
   }
 
+  /**
+   * Stop streaming of a message
+   * @param {number} channelId - ID of the channel.
+   * @param {number} messageId - ID of the message.
+   * @returns {Promise}
+   */
+  stopMessageStreaming(channelId, messageId) {
+    return this.#deleteRequest(
+      `/channels/${channelId}/messages/${messageId}/streaming`
+    );
+  }
+
   /**
    * Trashes (soft deletes) a chat message.
    * @param {number} channelId - ID of the channel.

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

@@ -0,0 +1,25 @@
+.chat-message-container.-streaming {
+  .chat-message-text {
+    @keyframes cursor-blink {
+      0% {
+        opacity: 0;
+      }
+    }
+
+    p::after {
+      margin-left: 3px;
+      margin-bottom: -4px;
+      content: "";
+      width: 6px;
+      height: 17px;
+      background: var(--primary);
+      display: inline-block;
+      animation: cursor-blink 0.5s steps(2) infinite;
+    }
+  }
+
+  .stop-streaming-btn {
+    margin-top: 0.5rem;
+    margin-bottom: 0.25rem;
+  }
+}

plugins/chat/assets/stylesheets/common/index.scss

@@ -68,3 +68,4 @@
 @import "chat-navbar";
 @import "chat-thread-title";
 @import "chat-audio-upload";
+@import "chat-message-text";

plugins/chat/config/routes.rb

@@ -18,6 +18,8 @@ Chat::Engine.routes.draw do
     get "/channels/:channel_id/messages" => "channel_messages#index"
     put "/channels/:channel_id/messages/:message_id" => "channel_messages#update"
     post "/channels/:channel_id/messages/moves" => "channels_messages_moves#create"
+    delete "/channels/:channel_id/messages/:message_id/streaming" =>
+             "channels_messages_streaming#destroy"
     post "/channels/:channel_id/invites" => "channels_invites#create"
     post "/channels/:channel_id/archives" => "channels_archives#create"
     get "/channels/:channel_id/memberships" => "channels_memberships#index"

plugins/chat/db/migrate/20240213175713_add_streaming_to_message.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddStreamingToMessage < ActiveRecord::Migration[7.0]
+  def change
+    add_column :chat_messages, :streaming, :boolean, null: false, default: false
+  end
+end

plugins/chat/lib/chat_sdk/channel.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module ChatSDK
+  class Channel
+    include Chat::WithServiceHelper
+
+    # Retrieves messages from a specified channel.
+    #
+    # @param channel_id [Integer] The ID of the chat channel from which to fetch messages.
+    # @param guardian [Guardian] The guardian object representing the user's permissions.
+    # @return [Array<ChMessage>] An array of message objects from the specified channel.
+    #
+    # @example Fetching messages from a channel with additional parameters
+    #   ChatSDK::Channel.messages(channel_id: 1, guardian: Guardian.new)
+    #
+    def self.messages(channel_id:, guardian:, **params)
+      new.messages(channel_id: channel_id, guardian: guardian, **params)
+    end
+
+    def messages(channel_id:, guardian:, **params)
+      with_service(
+        Chat::ListChannelMessages,
+        channel_id: channel_id,
+        guardian: guardian,
+        **params,
+        direction: "future",
+      ) do
+        on_success { result.messages }
+        on_failure do
+          p Chat::StepsInspector.new(result)
+          raise "Unexpected error"
+        end
+        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
+    end
+  end
+end

plugins/chat/lib/chat_sdk/message.rb

@@ -0,0 +1,126 @@
+# frozen_string_literal: true
+
+module ChatSDK
+  class Message
+    include Chat::WithServiceHelper
+
+    # Creates a new message in a chat channel.
+    #
+    # @param raw [String] The content of the message.
+    # @param channel_id [Integer] The ID of the chat channel.
+    # @param guardian [Guardian] The user's guardian object, for policy enforcement.
+    # @param in_reply_to_id [Integer, nil] The ID of the message this is in reply to (optional).
+    # @param thread_id [Integer, nil] The ID of the thread this message belongs to (optional).
+    # @param upload_ids [Array<Integer>, nil] The IDs of any uploads associated with the message (optional).
+    # @param streaming [Boolean] Whether the message is part of a streaming operation (default: false).
+    # @param enforce_membership [Boolean] Allows to ensure the guardian will be allowed in the channel (default: false).
+    # @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.
+    #
+    # @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|
+    #     helper.stream(raw: "Continuation of the message")
+    #   end
+    def self.create(**params, &block)
+      new.create(**params, &block)
+    end
+
+    # Creates a new message with streaming enabled by default.
+    #
+    # This method is a convenience wrapper around `create` with `streaming: true` set by default.
+    # It supports all the same parameters and block usage as `create`.
+    #
+    # @see #create
+    def self.create_with_stream(**params, &block)
+      self.create(**params, streaming: true, &block)
+    end
+
+    def create(
+      raw:,
+      channel_id:,
+      guardian:,
+      in_reply_to_id: nil,
+      thread_id: nil,
+      upload_ids: nil,
+      streaming: false,
+      enforce_membership: false,
+      &block
+    )
+      message =
+        with_service(
+          Chat::CreateMessage,
+          message: raw,
+          guardian: guardian,
+          chat_channel_id: channel_id,
+          in_reply_to_id: in_reply_to_id,
+          thread_id: thread_id,
+          upload_ids: upload_ids,
+          streaming: streaming,
+          enforce_membership: enforce_membership,
+        ) do
+          on_model_not_found(:channel) { raise "Couldn't find channel with id: `#{channel_id}`" }
+          on_model_not_found(:channel_membership) do
+            raise "User with id: `#{guardian.user.id}` has no membership to this channel"
+          end
+          on_failed_policy(:ensure_valid_thread_for_channel) do
+            raise "Couldn't find thread with id: `#{thread_id}`"
+          end
+          on_failed_policy(:allowed_to_join_channel) do
+            raise "User with id: `#{guardian.user.id}` can't join this channel"
+          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
+        end
+
+      if streaming && block_given?
+        helper = Helper.new(message)
+        block.call(helper, message)
+      end
+
+      message
+    ensure
+      if message && streaming
+        message.update!(streaming: false)
+        ::Chat::Publisher.publish_edit!(message.chat_channel, message.reload)
+      end
+    end
+  end
+
+  class Helper
+    include Chat::WithServiceHelper
+
+    attr_reader :message
+
+    def initialize(message)
+      @message = message
+    end
+
+    def stream(raw: nil)
+      return false unless self.message.reload.streaming
+
+      with_service(
+        Chat::UpdateMessage,
+        message_id: self.message.id,
+        message: raw ? self.message.reload.message + " " + raw : self.message.message,
+        guardian: self.message.user.guardian,
+        streaming: true,
+      ) do
+        on_failure do
+          p Chat::StepsInspector.new(result)
+          raise "Unexpected error"
+        end
+      end
+
+      self.message
+    end
+  end
+end

plugins/chat/lib/chat_sdk/thread.rb

@@ -0,0 +1,80 @@
+# frozen_string_literal: true
+
+module ChatSDK
+  class Thread
+    include Chat::WithServiceHelper
+
+    # Updates the title of a specified chat thread.
+    #
+    # @param title [String] The new title for the chat thread.
+    # @param thread_id [Integer] The ID of the chat thread to be updated.
+    # @param guardian [Guardian] The guardian object representing the user's permissions.
+    # @return [Chat::Thread] The updated thread object with the new title.
+    #
+    # @example Updating the title of a chat thread
+    #   ChatSDK::Thread.update_title(title: "New Thread Title", thread_id: 1, guardian: Guardian.new)
+    def self.update_title(**params)
+      new.update(title: params[:title], thread_id: params[:thread_id], guardian: params[:guardian])
+    end
+
+    def self.update(**params)
+      new.update(**params)
+    end
+
+    # Retrieves messages from a specified thread.
+    #
+    # @param thread_id [Integer] The ID of the chat thread from which to fetch messages.
+    # @param guardian [Guardian] The guardian object representing the user's permissions.
+    # @return [Array<Chat::Message>] An array of message objects from the specified thread.
+    #
+    # @example Fetching messages from a thread with additional parameters
+    #   ChatSDK::Thread.messages(thread_id: 1, guardian: Guardian.new)
+    #
+    def self.messages(thread_id:, guardian:, **params)
+      new.messages(thread_id: thread_id, guardian: guardian, **params)
+    end
+
+    def messages(thread_id:, guardian:, **params)
+      with_service(
+        Chat::ListChannelThreadMessages,
+        thread_id: thread_id,
+        guardian: guardian,
+        **params,
+        direction: "future",
+      ) do
+        on_success { result.messages }
+        on_failed_policy(:can_view_thread) { raise "Guardian can't view thread" }
+        on_failed_policy(:target_message_exists) { raise "Target message doesn't exist" }
+        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
+      end
+    end
+
+    def update(**params)
+      with_service(Chat::UpdateThread, **params) do
+        on_model_not_found(:channel) do
+          raise "Couldn’t find channel with id: `#{params[:channel_id]}`"
+        end
+        on_model_not_found(:thread) do
+          raise "Couldn’t find thread with id: `#{params[:thread_id]}`"
+        end
+        on_failed_policy(:can_view_channel) { raise "Guardian can't view channel" }
+        on_failed_policy(:can_edit_thread) { raise "Guardian can't edit thread" }
+        on_failed_policy(:threading_enabled_for_channel) do
+          raise "Threading is not enabled for this channel"
+        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
+      end
+    end
+  end
+end

plugins/chat/plugin.rb

@@ -24,6 +24,7 @@ register_svg_icon "clipboard"
 register_svg_icon "file-audio"
 register_svg_icon "file-video"
 register_svg_icon "file-image"
+register_svg_icon "stop-circle"
 
 # route: /admin/plugins/chat
 add_admin_route "chat.admin.title", "chat"

plugins/chat/spec/fabricators/chat_fabricator.rb

@@ -216,7 +216,7 @@ Fabricator(:chat_thread, class_name: "Chat::Thread") do
   original_message do |attrs|
     Fabricate(
       :chat_message,
-      chat_channel: attrs[:channel] || Fabricate(:chat_channel),
+      chat_channel: attrs[:channel] || Fabricate(:chat_channel, threading_enabled: true),
       user: attrs[:original_message_user] || Fabricate(:user),
       use_service: attrs[:use_service],
     )

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

@@ -0,0 +1,43 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+describe ChatSDK::Channel do
+  describe ".messages" do
+    fab!(:channel_1) { Fabricate(:chat_channel) }
+    fab!(:message_1) { Fabricate(:chat_message, chat_channel: channel_1) }
+    fab!(:message_2) { Fabricate(:chat_message, chat_channel: channel_1) }
+
+    let(:params) { { channel_id: channel_1.id, guardian: Discourse.system_user.guardian } }
+
+    it "loads the messages" do
+      messages = described_class.messages(**params)
+
+      expect(messages).to eq([message_1, message_2])
+    end
+
+    it "accepts page_size" do
+      messages = described_class.messages(**params, page_size: 1)
+
+      expect(messages).to eq([message_1])
+    end
+
+    context "when guardian can't see the channel" do
+      fab!(:channel_1) { Fabricate(:private_category_channel) }
+
+      it "fails" do
+        params[:guardian] = Fabricate(:user).guardian
+
+        expect { described_class.messages(**params) }.to raise_error("Guardian can't view channel")
+      end
+    end
+
+    context "when target_message doesn’t exist" do
+      it "fails" do
+        expect { described_class.messages(**params, target_message_id: -999) }.to raise_error(
+          "Target message doesn't exist",
+        )
+      end
+    end
+  end
+end

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

@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+describe ChatSDK::Message do
+  describe ".create" do
+    fab!(:channel_1) { Fabricate(:chat_channel) }
+
+    let(:guardian) { Discourse.system_user.guardian }
+    let(:params) do
+      { enforce_membership: false, raw: "something", channel_id: channel_1.id, guardian: guardian }
+    end
+
+    it "creates the message" do
+      message = described_class.create(**params)
+
+      expect(message.message).to eq("something")
+    end
+
+    context "when thread_id is present" do
+      fab!(:thread_1) { Fabricate(:chat_thread, channel: channel_1) }
+
+      it "creates the message in a thread" do
+        message = described_class.create(**params, thread_id: thread_1.id)
+
+        expect(message.thread_id).to eq(thread_1.id)
+      end
+    end
+
+    context "when channel doesn’t exist" do
+      it "fails" do
+        expect { described_class.create(**params, channel_id: -999) }.to raise_error(
+          "Couldn't find channel with id: `-999`",
+        )
+      end
+    end
+
+    context "when user can't join channel" do
+      it "fails" do
+        params[:guardian] = Fabricate(:user).guardian
+
+        expect { described_class.create(**params) }.to raise_error(
+          "User with id: `#{params[:guardian].user.id}` can't join this channel",
+        )
+      end
+    end
+
+    context "when membership is enforced" do
+      it "works" do
+        params[:enforce_membership] = true
+        params[:guardian] = Fabricate(:user).guardian
+        SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]]
+
+        message = described_class.create(**params)
+
+        expect(message.message).to eq("something")
+      end
+    end
+
+    context "when thread doesn't exist" do
+      it "fails" do
+        expect { described_class.create(**params, thread_id: -999) }.to raise_error(
+          "Couldn't find thread with id: `-999`",
+        )
+      end
+    end
+
+    context "when params are invalid" do
+      it "fails" do
+        expect { described_class.create(**params, raw: nil, channel_id: nil) }.to raise_error(
+          "Chat channel can't be blank, Message can't be blank",
+        )
+      end
+    end
+  end
+
+  describe ".create_with_stream" do
+    fab!(:channel_1) { Fabricate(:chat_channel) }
+
+    let(:guardian) { Discourse.system_user.guardian }
+    let(:params) { { raw: "something", channel_id: channel_1.id, guardian: guardian } }
+
+    it "allows streaming" do
+      created_message =
+        described_class.create_with_stream(**params) do |helper, message|
+          expect(message.streaming).to eq(true)
+
+          edit =
+            MessageBus
+              .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")
+        end
+
+      expect(created_message.streaming).to eq(false)
+      expect(created_message.message).to eq("something test")
+    end
+  end
+end

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

@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+describe ChatSDK::Thread do
+  describe ".update_title" do
+    fab!(:thread_1) { Fabricate(:chat_thread) }
+
+    let(:params) do
+      {
+        title: "New Title",
+        channel_id: thread_1.channel_id,
+        thread_id: thread_1.id,
+        guardian: Discourse.system_user.guardian,
+      }
+    end
+
+    it "changes the title" do
+      expect { described_class.update_title(**params) }.to change { thread_1.reload.title }.from(
+        thread_1.title,
+      ).to(params[:title])
+    end
+
+    context "when missing param" do
+      it "fails" do
+        params.delete(:thread_id)
+
+        expect { described_class.update_title(**params) }.to raise_error("Thread can't be blank")
+      end
+    end
+
+    context "when guardian can't see the channel" do
+      fab!(:thread_1) { Fabricate(:chat_thread, channel: Fabricate(:private_category_channel)) }
+
+      it "fails" do
+        params[:guardian] = Fabricate(:user).guardian
+
+        expect { described_class.update_title(**params) }.to raise_error(
+          "Guardian can't view channel",
+        )
+      end
+    end
+
+    context "when guardian can't edit the thread" do
+      it "fails" do
+        params[:guardian] = Fabricate(:user).guardian
+
+        expect { described_class.update_title(**params) }.to raise_error(
+          "Guardian can't edit thread",
+        )
+      end
+    end
+
+    context "when the threadind is not enabled" do
+      before { thread_1.channel.update!(threading_enabled: false) }
+
+      it "fails" do
+        expect { described_class.update_title(**params) }.to raise_error(
+          "Threading is not enabled for this channel",
+        )
+      end
+    end
+
+    context "when the thread doesn't exist" do
+      it "fails" do
+        params[:thread_id] = -999
+        expect { described_class.update_title(**params) }.to raise_error(
+          "Couldn’t find thread with id: `-999`",
+        )
+      end
+    end
+
+    context "when target_message doesn’t exist" do
+      it "fails" do
+        expect { described_class.messages(**params, target_message_id: -999) }.to raise_error(
+          "Target message doesn't exist",
+        )
+      end
+    end
+  end
+end

plugins/chat/spec/requests/chat/api/channels_messages_streaming_controller_spec.rb

@@ -0,0 +1,66 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+RSpec.describe Chat::Api::ChannelsMessagesStreamingController do
+  fab!(:channel_1) { Fabricate(:chat_channel) }
+  fab!(:current_user) { Fabricate(:user) }
+
+  before do
+    SiteSetting.chat_enabled = true
+    SiteSetting.chat_allowed_groups = Group::AUTO_GROUPS[:everyone]
+  end
+
+  describe "#destroy" do
+    before { sign_in(current_user) }
+
+    context "when chat is not enabled" do
+      it "returns a 404 error" do
+        SiteSetting.chat_enabled = false
+
+        delete "/chat/api/channels/-/messages/-/streaming"
+
+        expect(response.status).to eq(404)
+      end
+    end
+
+    context "when user is not logged" do
+      it "returns a 404 error" do
+        sign_out
+
+        delete "/chat/api/channels/-/messages/-/streaming"
+
+        expect(response.status).to eq(404)
+      end
+    end
+
+    context "when the message doesnt exist" do
+      it "returns a 404 error" do
+        delete "/chat/api/channels/#{channel_1.id}/messages/-999/streaming"
+
+        expect(response.status).to eq(404)
+      end
+    end
+
+    context "when the user can’t stop" do
+      fab!(:message_1) { Fabricate(:chat_message, chat_channel: channel_1) }
+
+      it "returns a 403 error" do
+        delete "/chat/api/channels/#{channel_1.id}/messages/#{message_1.id}/streaming"
+
+        expect(response.status).to eq(403)
+      end
+    end
+
+    context "when the user can stop" do
+      fab!(:current_user) { Fabricate(:admin) }
+      fab!(:message_1) { Fabricate(:chat_message, chat_channel: channel_1) }
+
+      it "returns a 200" do
+        delete "/chat/api/channels/#{channel_1.id}/messages/#{message_1.id}/streaming"
+
+        expect(response.status).to eq(200)
+      end
+    end
+  end
+end

plugins/chat/spec/services/chat/create_message_spec.rb

@@ -35,6 +35,7 @@ RSpec.describe Chat::CreateMessage do
     let(:context_post_ids) { nil }
     let(:params) do
       {
+        enforce_membership: false,
         guardian: guardian,
         chat_channel_id: channel.id,
         message: content,
@@ -212,6 +213,17 @@ RSpec.describe Chat::CreateMessage do
             it { is_expected.to be_a_success }
           end
 
+          context "when membership is enforced" do
+            fab!(:user) { Fabricate(:user) }
+
+            before do
+              SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]]
+              params[:enforce_membership] = true
+            end
+
+            it { is_expected.to be_a_success }
+          end
+
           context "when user can join channel" do
             before { user.groups << Group.find(Group::AUTO_GROUPS[:trust_level_1]) }
 

plugins/chat/spec/services/chat/stop_message_streaming_spec.rb

@@ -0,0 +1,102 @@
+# frozen_string_literal: true
+
+RSpec.describe Chat::StopMessageStreaming do
+  describe ".call" do
+    subject(:result) { described_class.call(params) }
+
+    let(:params) { { guardian: guardian } }
+    let(:guardian) { Guardian.new(current_user) }
+
+    fab!(:current_user) { Fabricate(:user) }
+    fab!(:channel_1) { Fabricate(:chat_channel) }
+
+    before { SiteSetting.chat_allowed_groups = [Group::AUTO_GROUPS[:everyone]] }
+
+    context "with valid params" do
+      fab!(:current_user) { Fabricate(:admin) }
+      fab!(:message_1) { Fabricate(:chat_message, chat_channel: channel_1, streaming: true) }
+
+      let(:params) { { guardian: guardian, channel_id: channel_1.id, message_id: message_1.id } }
+
+      it { is_expected.to be_a_success }
+
+      it "updates the streaming attribute to false" do
+        expect { result }.to change { message_1.reload.streaming }.to eq(false)
+      end
+
+      it "publishes an event" do
+        messages = MessageBus.track_publish { result }
+
+        expect(messages.find { |m| m.channel == "/chat/#{channel_1.id}" }.data).to include(
+          { "type" => "edit" },
+        )
+      end
+    end
+
+    context "when the channel_id is not provided" do
+      it { is_expected.to fail_a_contract }
+    end
+
+    context "when the message_id is not provided" do
+      let(:params) { { guardian: guardian, channel_id: channel_1.id } }
+
+      it { is_expected.to fail_a_contract }
+    end
+
+    context "when the message doesnt exist" do
+      let(:params) { { guardian: guardian, channel_id: channel_1.id, message_id: -999 } }
+
+      it { is_expected.to fail_to_find_a_model(:message) }
+    end
+
+    context "when the message is a reply" do
+      let(:params) { { guardian: guardian, channel_id: channel_1.id, message_id: reply.id } }
+
+      context "when the OM is from current user" do
+        fab!(:original_message) do
+          Fabricate(:chat_message, chat_channel: channel_1, user: current_user)
+        end
+        fab!(:reply) do
+          Fabricate(:chat_message, chat_channel: channel_1, in_reply_to: original_message)
+        end
+
+        it { is_expected.to be_a_success }
+      end
+
+      context "when the OM is not from current user" do
+        fab!(:original_message) do
+          Fabricate(:chat_message, chat_channel: channel_1, user: Fabricate(:user))
+        end
+        fab!(:reply) do
+          Fabricate(:chat_message, chat_channel: channel_1, in_reply_to: original_message)
+        end
+
+        context "when current user is a regular user" do
+          it { is_expected.to fail_a_policy(:can_stop_streaming) }
+        end
+
+        context "when current user is an admin" do
+          fab!(:current_user) { Fabricate(:admin) }
+
+          it { is_expected.to be_a_success }
+        end
+      end
+    end
+
+    context "when the message is not a reply" do
+      let(:params) { { guardian: guardian, channel_id: channel_1.id, message_id: message.id } }
+
+      fab!(:message) { Fabricate(:chat_message, chat_channel: channel_1) }
+
+      context "when current user is a regular user" do
+        it { is_expected.to fail_a_policy(:can_stop_streaming) }
+      end
+
+      context "when current user is an admin" do
+        fab!(:current_user) { Fabricate(:admin) }
+
+        it { is_expected.to be_a_success }
+      end
+    end
+  end
+end

plugins/chat/spec/services/chat/update_thread_spec.rb

@@ -2,7 +2,6 @@
 
 RSpec.describe Chat::UpdateThread do
   describe Chat::UpdateThread::Contract, type: :model do
-    it { is_expected.to validate_presence_of :channel_id }
     it { is_expected.to validate_presence_of :thread_id }
   end
 
@@ -17,9 +16,7 @@ RSpec.describe Chat::UpdateThread do
 
     let(:guardian) { Guardian.new(current_user) }
     let(:title) { "some new title :D" }
-    let(:params) do
-      { guardian: guardian, thread_id: thread.id, channel_id: thread.channel_id, title: title }
-    end
+    let(:params) { { guardian: guardian, thread_id: thread.id, title: title } }
 
     context "when all steps pass" do
       it "sets the service result as successful" do
@@ -53,12 +50,6 @@ RSpec.describe Chat::UpdateThread do
       it { is_expected.to fail_a_contract }
     end
 
-    context "when thread is not found because the channel ID differs" do
-      before { params[:thread_id] = other_thread.id }
-
-      it { is_expected.to fail_to_find_a_model(:thread) }
-    end
-
     context "when thread is not found" do
       before { thread.destroy! }