DEV: Migrate Chat::MessageCreator to a service (#22390)

Currently, the logic for creating a new chat message is scattered
between a controller and an “old” service.

This patch address this issue by creating a new service (using the “new”
sevice object system) encapsulating all the necessary logic.
(authorization, publishing events, etc.)

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

@@ -19,4 +19,29 @@ class Chat::Api::ChannelMessagesController < Chat::ApiController
       on_model_not_found(:message) { raise Discourse::NotFound }
     end
   end
+
+  def create
+    Chat::MessageRateLimiter.run!(current_user)
+
+    with_service(Chat::CreateMessage) do
+      on_success { render json: success_json.merge(message_id: result[:message].id) }
+      on_failed_policy(:no_silenced_user) { raise Discourse::InvalidAccess }
+      on_model_not_found(:channel) { raise Discourse::NotFound }
+      on_failed_policy(:allowed_to_join_channel) { raise Discourse::InvalidAccess }
+      on_model_not_found(:channel_membership) { raise Discourse::InvalidAccess }
+      on_failed_policy(:ensure_reply_consistency) { raise Discourse::NotFound }
+      on_failed_policy(:allowed_to_create_message_in_channel) do |policy|
+        render_json_error(policy.reason)
+      end
+      on_failed_policy(:ensure_valid_thread_for_channel) do
+        render_json_error(I18n.t("chat.errors.thread_invalid_for_channel"))
+      end
+      on_failed_policy(:ensure_thread_matches_parent) do
+        render_json_error(I18n.t("chat.errors.thread_does_not_match_parent"))
+      end
+      on_model_errors(:message) do |model|
+        render_json_error(model.errors.map(&:full_message).join(", "))
+      end
+    end
+  end
 end

plugins/chat/app/controllers/chat/chat_controller.rb

@@ -77,81 +77,6 @@ module Chat
       end
     end
 
-    def create_message
-      raise Discourse::InvalidAccess if current_user.silenced?
-
-      Chat::MessageRateLimiter.run!(current_user)
-
-      @user_chat_channel_membership =
-        Chat::ChannelMembershipManager.new(@chat_channel).find_for_user(current_user)
-      raise Discourse::InvalidAccess unless @user_chat_channel_membership
-
-      reply_to_msg_id = params[:in_reply_to_id]
-      if reply_to_msg_id.present?
-        rm = Chat::Message.find(reply_to_msg_id)
-        raise Discourse::NotFound if rm.chat_channel_id != @chat_channel.id
-      end
-
-      content = params[:message]
-
-      chat_message_creator =
-        Chat::MessageCreator.create(
-          chat_channel: @chat_channel,
-          user: current_user,
-          in_reply_to_id: reply_to_msg_id,
-          content: content,
-          staged_id: params[:staged_id],
-          upload_ids: params[:upload_ids],
-          thread_id: params[:thread_id],
-          staged_thread_id: params[:staged_thread_id],
-        )
-
-      return render_json_error(chat_message_creator.error) if chat_message_creator.failed?
-
-      if !chat_message_creator.chat_message.thread_id.present?
-        @user_chat_channel_membership.update!(
-          last_read_message_id: chat_message_creator.chat_message.id,
-        )
-      end
-
-      if @chat_channel.direct_message_channel?
-        # If any of the channel users is ignoring, muting, or preventing DMs from
-        # the current user then we should not auto-follow the channel once again or
-        # publish the new channel.
-        allowed_user_ids =
-          UserCommScreener.new(
-            acting_user: current_user,
-            target_user_ids:
-              @chat_channel.user_chat_channel_memberships.where(following: false).pluck(:user_id),
-          ).allowing_actor_communication
-
-        allowed_user_ids << current_user.id if !@user_chat_channel_membership.following
-
-        if allowed_user_ids.any?
-          Chat::Publisher.publish_new_channel(@chat_channel, User.where(id: allowed_user_ids))
-
-          @chat_channel
-            .user_chat_channel_memberships
-            .where(user_id: allowed_user_ids)
-            .update_all(following: true)
-        end
-      end
-
-      message =
-        (
-          if @user_chat_channel_membership.last_read_message_id &&
-               chat_message_creator.chat_message.in_thread?
-            Chat::Message.find(@user_chat_channel_membership.last_read_message_id)
-          else
-            chat_message_creator.chat_message
-          end
-        )
-
-      Chat::Publisher.publish_user_tracking_state!(current_user, @chat_channel, message)
-
-      render json: success_json.merge(message_id: chat_message_creator.chat_message.id)
-    end
-
     def edit_message
       chat_message_updater =
         Chat::MessageUpdater.update(

plugins/chat/app/controllers/chat/incoming_webhooks_controller.rb

@@ -2,6 +2,8 @@
 
 module Chat
   class IncomingWebhooksController < ::ApplicationController
+    include Chat::WithServiceHelper
+
     requires_plugin Chat::PLUGIN_NAME
 
     WEBHOOK_MESSAGES_PER_MINUTE_LIMIT = 10
@@ -50,20 +52,37 @@ module Chat
     private
 
     def process_webhook_payload(text:, key:)
-      validate_message_length(text)
       webhook = find_and_rate_limit_webhook(key)
+      webhook.chat_channel.add(Discourse.system_user)
 
-      chat_message_creator =
+      with_service(
-        Chat::MessageCreator.create(
+        Chat::CreateMessage,
-          chat_channel: webhook.chat_channel,
+        chat_channel_id: webhook.chat_channel_id,
-          user: Discourse.system_user,
+        guardian: Discourse.system_user.guardian,
-          content: text,
+        message: text,
         incoming_chat_webhook: webhook,
-        )
+      ) do
-      if chat_message_creator.failed?
+        on_success { render json: success_json }
-        render_json_error(chat_message_creator.error)
+        on_failed_contract do |contract|
-      else
+          raise Discourse::InvalidParameters.new(contract.errors.full_messages)
-        render json: success_json
+        end
+        on_failed_policy(:no_silenced_user) { raise Discourse::InvalidAccess }
+        on_model_not_found(:channel) { raise Discourse::NotFound }
+        on_failed_policy(:allowed_to_join_channel) { raise Discourse::InvalidAccess }
+        on_model_not_found(:channel_membership) { raise Discourse::InvalidAccess }
+        on_failed_policy(:ensure_reply_consistency) { raise Discourse::NotFound }
+        on_failed_policy(:allowed_to_create_message_in_channel) do |policy|
+          render_json_error(policy.reason)
+        end
+        on_failed_policy(:ensure_valid_thread_for_channel) do
+          render_json_error(I18n.t("chat.errors.thread_invalid_for_channel"))
+        end
+        on_failed_policy(:ensure_thread_matches_parent) do
+          render_json_error(I18n.t("chat.errors.thread_does_not_match_parent"))
+        end
+        on_model_errors(:message) do |model|
+          render_json_error(model.errors.map(&:full_message).join(", "))
+        end
       end
     end
 
@@ -81,13 +100,6 @@ module Chat
       webhook
     end
 
-    def validate_message_length(message)
-      return if message.length <= SiteSetting.chat_maximum_message_length
-      raise Discourse::InvalidParameters.new(
-              "Body cannot be over #{SiteSetting.chat_maximum_message_length} characters",
-            )
-    end
-
     # The webhook POST body can be in 3 different formats:
     #
     # * { text: "message text" }, which is the most basic method, and also mirrors Slack payloads

plugins/chat/app/models/chat/message.rb

@@ -13,9 +13,9 @@ module Chat
 
     belongs_to :chat_channel, class_name: "Chat::Channel"
     belongs_to :user
-    belongs_to :in_reply_to, class_name: "Chat::Message"
+    belongs_to :in_reply_to, class_name: "Chat::Message", autosave: true
     belongs_to :last_editor, class_name: "User"
-    belongs_to :thread, class_name: "Chat::Thread", optional: true
+    belongs_to :thread, class_name: "Chat::Thread", optional: true, autosave: true
 
     has_many :replies,
              class_name: "Chat::Message",

plugins/chat/app/policies/chat/channel/message_creation_policy.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+class Chat::Channel::MessageCreationPolicy < PolicyBase
+  class DirectMessageStrategy
+    class << self
+      def call(guardian, channel)
+        guardian.can_create_channel_message?(channel) || guardian.can_create_direct_message?
+      end
+
+      def reason(*)
+        I18n.t("chat.errors.user_cannot_send_direct_messages")
+      end
+    end
+  end
+
+  class CategoryStrategy
+    class << self
+      def call(guardian, channel)
+        guardian.can_create_channel_message?(channel)
+      end
+
+      def reason(_, channel)
+        I18n.t("chat.errors.channel_new_message_disallowed.#{channel.status}")
+      end
+    end
+  end
+
+  attr_reader :strategy
+
+  delegate :channel, to: :context
+
+  def initialize(*)
+    super
+    @strategy = CategoryStrategy
+    @strategy = DirectMessageStrategy if channel.direct_message_channel?
+  end
+
+  def call
+    strategy.call(guardian, channel)
+  end
+
+  def reason
+    strategy.reason(guardian, channel)
+  end
+end

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

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+module Chat
+  module Action
+    class PublishAndFollowDirectMessageChannel
+      attr_reader :channel_membership
+
+      delegate :chat_channel, :user, to: :channel_membership
+
+      def self.call(...)
+        new(...).call
+      end
+
+      def initialize(channel_membership:)
+        @channel_membership = channel_membership
+      end
+
+      def call
+        return unless chat_channel.direct_message_channel?
+        return if users_allowing_communication.none?
+
+        chat_channel
+          .user_chat_channel_memberships
+          .where(user: users_allowing_communication)
+          .update_all(following: true)
+        Chat::Publisher.publish_new_channel(chat_channel, users_allowing_communication)
+      end
+
+      private
+
+      def users_allowing_communication
+        @users_allowing_communication ||= User.where(id: user_ids).to_a
+      end
+
+      def user_ids
+        UserCommScreener.new(
+          acting_user: user,
+          target_user_ids:
+            chat_channel.user_chat_channel_memberships.where(following: false).pluck(:user_id),
+        ).allowing_actor_communication + Array.wrap(current_user_id)
+      end
+
+      def current_user_id
+        return if channel_membership.following?
+        user.id
+      end
+    end
+  end
+end

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

@@ -0,0 +1,191 @@
+# frozen_string_literal: true
+
+module Chat
+  # Service responsible for creating a new message.
+  #
+  # @example
+  #  Chat::CreateMessage.call(chat_channel_id: 2, guardian: guardian, message: "A new message")
+  #
+  class CreateMessage
+    include Service::Base
+
+    # @!method call(chat_channel_id:, guardian:, in_reply_to_id:, message:, staged_id:, upload_ids:, thread_id:, staged_thread_id:, incoming_chat_webhook:)
+    #   @param guardian [Guardian]
+    #   @param chat_channel_id [Integer]
+    #   @param message [String]
+    #   @param in_reply_to_id [Integer] ID of a message to reply to
+    #   @param thread_id [Integer] ID of a thread to reply to
+    #   @param upload_ids [Array<Integer>] IDs of uploaded documents
+    #   @param staged_id [String] arbitrary string that will be sent back to the client
+    #   @param staged_thread_id [String] arbitrary string that will be sent back to the client (for a new thread)
+    #   @param incoming_chat_webhook [Chat::IncomingWebhook]
+
+    policy :no_silenced_user
+    contract
+    model :channel
+    policy :allowed_to_join_channel
+    policy :allowed_to_create_message_in_channel, class_name: Chat::Channel::MessageCreationPolicy
+    model :channel_membership
+    model :reply, optional: true
+    policy :ensure_reply_consistency
+    model :thread, optional: true
+    policy :ensure_valid_thread_for_channel
+    policy :ensure_thread_matches_parent
+    model :uploads, optional: true
+    model :message, :instantiate_message
+    transaction do
+      step :save_message
+      step :delete_drafts
+      step :post_process_thread
+      step :create_webhook_event
+      step :update_channel_last_message
+      step :update_membership_last_read
+      step :process_direct_message_channel
+    end
+    step :publish_new_thread
+    step :publish_new_message_events
+    step :publish_user_tracking_state
+
+    class Contract
+      attribute :chat_channel_id, :string
+      attribute :in_reply_to_id, :string
+      attribute :message, :string
+      attribute :staged_id, :string
+      attribute :upload_ids, :array
+      attribute :thread_id, :string
+      attribute :staged_thread_id, :string
+      attribute :incoming_chat_webhook
+
+      validates :chat_channel_id, presence: true
+      validates :message, presence: true, if: -> { upload_ids.blank? }
+    end
+
+    private
+
+    def no_silenced_user(guardian:, **)
+      !guardian.is_silenced?
+    end
+
+    def fetch_channel(contract:, **)
+      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 fetch_channel_membership(guardian:, channel:, **)
+      Chat::ChannelMembershipManager.new(channel).find_for_user(guardian.user)
+    end
+
+    def fetch_reply(contract:, **)
+      Chat::Message.find_by(id: contract.in_reply_to_id)
+    end
+
+    def ensure_reply_consistency(channel:, contract:, reply:, **)
+      return true if contract.in_reply_to_id.blank?
+      reply&.chat_channel == channel
+    end
+
+    def fetch_thread(contract:, reply:, channel:, **)
+      return Chat::Thread.find_by(id: contract.thread_id) if contract.thread_id.present?
+      return unless reply
+      reply.thread ||
+        reply.build_thread(
+          original_message: reply,
+          original_message_user: reply.user,
+          channel: channel,
+        )
+    end
+
+    def ensure_valid_thread_for_channel(thread:, contract:, channel:, **)
+      return true if contract.thread_id.blank?
+      thread&.channel == channel
+    end
+
+    def ensure_thread_matches_parent(thread:, reply:, **)
+      return true unless thread && reply
+      reply.thread == thread
+    end
+
+    def fetch_uploads(contract:, guardian:, **)
+      return [] if !SiteSetting.chat_allow_uploads
+      guardian.user.uploads.where(id: contract.upload_ids)
+    end
+
+    def instantiate_message(channel:, guardian:, contract:, uploads:, thread:, reply:, **)
+      channel.chat_messages.new(
+        user: guardian.user,
+        last_editor: guardian.user,
+        in_reply_to: reply,
+        message: contract.message,
+        uploads: uploads,
+        thread: thread,
+      )
+    end
+
+    def save_message(message:, **)
+      message.cook
+      message.save!
+      message.create_mentions
+    end
+
+    def delete_drafts(channel:, guardian:, **)
+      Chat::Draft.where(user: guardian.user, chat_channel: channel).destroy_all
+    end
+
+    def post_process_thread(thread:, message:, guardian:, **)
+      return unless thread
+
+      thread.update!(last_message: message)
+      thread.increment_replies_count_cache
+      thread.add(guardian.user).update!(last_read_message: message)
+      thread.add(thread.original_message_user)
+    end
+
+    def create_webhook_event(contract:, message:, **)
+      return if contract.incoming_chat_webhook.blank?
+      message.create_chat_webhook_event(incoming_chat_webhook: contract.incoming_chat_webhook)
+    end
+
+    def update_channel_last_message(channel:, message:, **)
+      return if message.in_thread?
+      channel.update!(last_message: message)
+    end
+
+    def update_membership_last_read(channel_membership:, message:, **)
+      return if message.in_thread?
+      channel_membership.update!(last_read_message: message)
+    end
+
+    def process_direct_message_channel(channel_membership:, **)
+      Chat::Action::PublishAndFollowDirectMessageChannel.call(
+        channel_membership: channel_membership,
+      )
+    end
+
+    def publish_new_thread(reply:, contract:, channel:, thread:, **)
+      return unless channel.threading_enabled?
+      return unless reply&.thread_id_previously_changed?(from: nil)
+      Chat::Publisher.publish_thread_created!(channel, reply, thread.id, contract.staged_thread_id)
+    end
+
+    def publish_new_message_events(channel:, message:, contract:, guardian:, **)
+      Chat::Publisher.publish_new!(
+        channel,
+        message,
+        contract.staged_id,
+        staged_thread_id: contract.staged_thread_id,
+      )
+      Jobs.enqueue(Jobs::Chat::ProcessMessage, { chat_message_id: message.id })
+      Chat::Notifier.notify_new(chat_message: message, timestamp: message.created_at)
+      DiscourseEvent.trigger(:chat_message_created, message, channel, guardian.user)
+    end
+
+    def publish_user_tracking_state(message:, channel:, channel_membership:, guardian:, **)
+      message_to_publish = message
+      message_to_publish = channel_membership.last_read_message || message if message.in_thread?
+      Chat::Publisher.publish_user_tracking_state!(guardian.user, channel, message_to_publish)
+    end
+  end
+end

plugins/chat/app/services/service/base.rb

@@ -370,7 +370,7 @@ module Service
 
     # @!visibility private
     def fail!(message)
-      step_name = caller_locations(1, 1)[0].label
+      step_name = caller_locations(1, 1)[0].base_label
       context["result.step.#{step_name}"].fail(error: message)
       context.fail!
     end

plugins/chat/config/routes.rb

@@ -79,7 +79,7 @@ Chat::Engine.routes.draw do
   put "/user_chat_enabled/:user_id" => "chat#set_user_chat_status"
   put "/:chat_channel_id/invite" => "chat#invite_users"
   post "/drafts" => "chat#set_draft"
-  post "/:chat_channel_id" => "chat#create_message"
+  post "/:chat_channel_id" => "api/channel_messages#create"
   put "/flag" => "chat#flag"
   get "/emojis" => "emojis#index"
 

plugins/chat/lib/chat/channel_fetcher.rb

@@ -248,7 +248,7 @@ module Chat
     def self.find_with_access_check(channel_id_or_slug, guardian)
       base_channel_relation = Chat::Channel.includes(:chatable)
 
-      if guardian.user.staff?
+      if guardian.is_staff?
         base_channel_relation = base_channel_relation.includes(:chat_channel_archive)
       end
 

plugins/chat/lib/chat/message_creator.rb

@@ -1,249 +0,0 @@
-# frozen_string_literal: true
-module Chat
-  class MessageCreator
-    attr_reader :error, :chat_message
-
-    def self.create(opts)
-      instance = new(**opts)
-      instance.create
-      instance
-    end
-
-    def initialize(
-      chat_channel:,
-      in_reply_to_id: nil,
-      thread_id: nil,
-      staged_thread_id: nil,
-      user:,
-      content:,
-      staged_id: nil,
-      incoming_chat_webhook: nil,
-      upload_ids: nil,
-      created_at: nil
-    )
-      @chat_channel = chat_channel
-      @user = user
-      @guardian = Guardian.new(user)
-
-      # NOTE: We confirm this exists and the user can access it in the ChatController,
-      # but in future the checks should be here
-      @in_reply_to_id = in_reply_to_id
-      @content = content
-      @staged_id = staged_id
-      @incoming_chat_webhook = incoming_chat_webhook
-      @upload_ids = upload_ids || []
-      @thread_id = thread_id
-      @staged_thread_id = staged_thread_id
-      @error = nil
-
-      @chat_message =
-        Chat::Message.new(
-          chat_channel: @chat_channel,
-          user_id: @user.id,
-          last_editor_id: @user.id,
-          in_reply_to_id: @in_reply_to_id,
-          message: @content,
-          created_at: created_at,
-        )
-    end
-
-    def create
-      begin
-        validate_channel_status!
-        @chat_message.uploads = get_uploads
-        validate_message!
-        validate_reply_chain!
-        validate_existing_thread!
-
-        @chat_message.thread_id = @existing_thread&.id
-        @chat_message.cook
-        @chat_message.save!
-        @chat_message.create_mentions
-
-        create_chat_webhook_event
-        create_thread
-        Chat::Draft.where(user_id: @user.id, chat_channel_id: @chat_channel.id).destroy_all
-        post_process_resolved_thread
-        update_channel_last_message
-        Chat::Publisher.publish_new!(
-          @chat_channel,
-          @chat_message,
-          @staged_id,
-          staged_thread_id: @staged_thread_id,
-        )
-        Jobs.enqueue(Jobs::Chat::ProcessMessage, { chat_message_id: @chat_message.id })
-        Chat::Notifier.notify_new(chat_message: @chat_message, timestamp: @chat_message.created_at)
-        DiscourseEvent.trigger(:chat_message_created, @chat_message, @chat_channel, @user)
-      rescue => error
-        @error = error
-      end
-    end
-
-    def failed?
-      @error.present?
-    end
-
-    private
-
-    def validate_channel_status!
-      return if @guardian.can_create_channel_message?(@chat_channel)
-
-      if @chat_channel.direct_message_channel? && !@guardian.can_create_direct_message?
-        raise StandardError.new(I18n.t("chat.errors.user_cannot_send_direct_messages"))
-      else
-        raise StandardError.new(
-                I18n.t("chat.errors.channel_new_message_disallowed.#{@chat_channel.status}"),
-              )
-      end
-    end
-
-    def validate_reply_chain!
-      return if @in_reply_to_id.blank?
-
-      @original_message_id = DB.query_single(<<~SQL).last
-      WITH RECURSIVE original_message_finder( id, in_reply_to_id )
-      AS (
-        -- start with the message id we want to find the parents of
-        SELECT id, in_reply_to_id
-        FROM chat_messages
-        WHERE id = #{@in_reply_to_id}
-
-        UNION ALL
-
-        -- get the chain of direct parents of the message
-        -- following in_reply_to_id
-        SELECT cm.id, cm.in_reply_to_id
-        FROM original_message_finder rm
-        JOIN chat_messages cm ON rm.in_reply_to_id = cm.id
-      )
-      SELECT id FROM original_message_finder
-
-      -- this makes it so only the root parent ID is returned, we can
-      -- exclude this to return all parents in the chain
-      WHERE in_reply_to_id IS NULL;
-    SQL
-
-      if @original_message_id.blank?
-        raise StandardError.new(I18n.t("chat.errors.original_message_not_found"))
-      end
-
-      @original_message = Chat::Message.with_deleted.find_by(id: @original_message_id)
-      if @original_message&.trashed?
-        raise StandardError.new(I18n.t("chat.errors.original_message_not_found"))
-      end
-    end
-
-    def validate_existing_thread!
-      return if @staged_thread_id.present? && @thread_id.blank?
-
-      return if @thread_id.blank?
-      @existing_thread = Chat::Thread.find(@thread_id)
-
-      if @existing_thread.channel_id != @chat_channel.id
-        raise StandardError.new(I18n.t("chat.errors.thread_invalid_for_channel"))
-      end
-
-      reply_to_thread_mismatch =
-        @chat_message.in_reply_to&.thread_id &&
-          @chat_message.in_reply_to.thread_id != @existing_thread.id
-      original_message_has_no_thread = @original_message && @original_message.thread_id.blank?
-      original_message_thread_mismatch =
-        @original_message && @original_message.thread_id != @existing_thread.id
-      if reply_to_thread_mismatch || original_message_has_no_thread ||
-           original_message_thread_mismatch
-        raise StandardError.new(I18n.t("chat.errors.thread_does_not_match_parent"))
-      end
-    end
-
-    def validate_message!
-      return if @chat_message.valid?
-      raise StandardError.new(@chat_message.errors.map(&:full_message).join(", "))
-    end
-
-    def create_chat_webhook_event
-      return if @incoming_chat_webhook.blank?
-      Chat::WebhookEvent.create(
-        chat_message: @chat_message,
-        incoming_chat_webhook: @incoming_chat_webhook,
-      )
-    end
-
-    def get_uploads
-      return [] if @upload_ids.blank? || !SiteSetting.chat_allow_uploads
-
-      ::Upload.where(id: @upload_ids, user_id: @user.id)
-    end
-
-    def create_thread
-      return if @in_reply_to_id.blank?
-      return if @chat_message.in_thread? && !@staged_thread_id.present?
-
-      if @original_message.thread
-        thread = @original_message.thread
-      else
-        thread =
-          Chat::Thread.create!(
-            original_message: @chat_message.in_reply_to,
-            original_message_user: @chat_message.in_reply_to.user,
-            channel: @chat_message.chat_channel,
-          )
-        @chat_message.in_reply_to.thread_id = thread.id
-      end
-
-      @chat_message.thread_id = thread.id
-
-      # NOTE: We intentionally do not try to correct thread IDs within the chain
-      # if they are incorrect, and only set the thread ID of messages where the
-      # thread ID is NULL. In future we may want some sync/background job to correct
-      # any inconsistencies.
-      DB.exec(<<~SQL)
-        WITH RECURSIVE thread_updater AS (
-          SELECT cm.id, cm.in_reply_to_id
-          FROM chat_messages cm
-          WHERE cm.in_reply_to_id IS NULL AND cm.id = #{@original_message_id}
-
-          UNION ALL
-
-          SELECT cm.id, cm.in_reply_to_id
-          FROM chat_messages cm
-          JOIN thread_updater ON cm.in_reply_to_id = thread_updater.id
-        )
-        UPDATE chat_messages
-        SET thread_id = #{thread.id}
-        FROM thread_updater
-        WHERE thread_id IS NULL AND chat_messages.id = thread_updater.id
-      SQL
-
-      if @chat_message.chat_channel.threading_enabled
-        Chat::Publisher.publish_thread_created!(
-          @chat_message.chat_channel,
-          @chat_message.in_reply_to,
-          thread.id,
-          @staged_thread_id,
-        )
-      end
-    end
-
-    def resolved_thread
-      @existing_thread || @chat_message.thread
-    end
-
-    def post_process_resolved_thread
-      return if resolved_thread.blank?
-
-      resolved_thread.update!(last_message: @chat_message)
-      resolved_thread.increment_replies_count_cache
-      current_user_thread_membership = resolved_thread.add(@user)
-      current_user_thread_membership.update!(last_read_message_id: @chat_message.id)
-
-      if resolved_thread.original_message_user != @user
-        resolved_thread.add(resolved_thread.original_message_user)
-      end
-    end
-
-    def update_channel_last_message
-      return if @chat_message.thread_reply?
-      @chat_channel.update!(last_message: @chat_message)
-    end
-  end
-end

plugins/chat/lib/chat/message_mover.rb

@@ -186,10 +186,11 @@ module Chat
     end
 
     def add_moved_placeholder(destination_channel, first_moved_message)
-      Chat::MessageCreator.create(
+      @source_channel.add(Discourse.system_user)
-        chat_channel: @source_channel,
+      Chat::CreateMessage.call(
-        user: Discourse.system_user,
+        chat_channel_id: @source_channel.id,
-        content:
+        guardian: Discourse.system_user.guardian,
+        message:
           I18n.t(
             "chat.channel.messages_moved",
             count: @source_message_ids.length,

plugins/chat/lib/discourse_dev/category_channel.rb

@@ -45,8 +45,10 @@ module DiscourseDev
         Faker::Number
           .between(from: 20, to: 80)
           .times do
-            Chat::MessageCreator.create(
+            Chat::CreateMessage.call(
-              { user: users.sample, chat_channel: channel, content: Faker::Lorem.paragraph },
+              guardian: users.sample.guardian,
+              chat_channel_id: channel.id,
+              message: Faker::Lorem.paragraph,
             )
           end
       end

plugins/chat/lib/discourse_dev/message.rb

@@ -24,11 +24,11 @@ module DiscourseDev
         ::Chat::UserChatChannelMembership.where(chat_channel: channel).order("RANDOM()").first
       user = membership.user
 
-      { user: user, content: Faker::Lorem.paragraph, chat_channel: channel }
+      { guardian: user.guardian, message: Faker::Lorem.paragraph, chat_channel_id: channel.id }
     end
 
     def create!
-      Chat::MessageCreator.create(data)
+      Chat::CreateMessage.call(data)
     end
   end
 end

plugins/chat/lib/discourse_dev/thread.rb

@@ -26,11 +26,11 @@ module DiscourseDev
       user = membership.user
 
       om =
-        Chat::MessageCreator.create(
+        Chat::CreateMessage.call(
-          user: user,
+          guardian: user.guardian,
-          content: Faker::Lorem.paragraph,
+          message: Faker::Lorem.paragraph,
-          chat_channel: channel,
+          chat_channel_id: channel.id,
-        ).chat_message
+        ).message
 
       { original_message_user: user, original_message: om, channel: channel }
     end
@@ -45,11 +45,11 @@ module DiscourseDev
             .first
             .user
         @message_count.times do
-          Chat::MessageCreator.create(
+          Chat::CreateMessage.call(
             {
-              user: user,
+              guardian: user.guardian,
-              chat_channel: thread.channel,
+              chat_channel_id: thread.channel_id,
-              content: Faker::Lorem.paragraph,
+              message: Faker::Lorem.paragraph,
               thread_id: thread.id,
             },
           )

plugins/chat/plugin.rb

@@ -417,14 +417,14 @@ after_initialize do
         placeholders = { channel_name: channel.title(sender) }.merge(context["placeholders"] || {})
 
         creator =
-          Chat::MessageCreator.create(
+          Chat::CreateMessage.call(
             chat_channel: channel,
-            user: sender,
+            guardian: sender.guardian,
             content: utils.apply_placeholders(fields.dig("message", "value"), placeholders),
           )
 
-        if creator.failed?
+        if creator.failure?
-          Rails.logger.warn "[discourse-automation] Chat message failed to send, error was: #{creator.error}"
+          Rails.logger.warn "[discourse-automation] Chat message failed to send:\n#{creator.inspect_steps.inspect}\n#{creator.inspect_steps.error}"
         end
       end
     end
@@ -432,7 +432,12 @@ after_initialize do
 
   add_api_key_scope(
     :chat,
-    { create_message: { actions: %w[chat/chat#create_message], params: %i[chat_channel_id] } },
+    {
+      create_message: {
+        actions: %w[chat/api/channel_messages#create],
+        params: %i[chat_channel_id],
+      },
+    },
   )
 
   # Dark mode email styles

plugins/chat/spec/fabricators/chat_fabricator.rb

@@ -54,13 +54,13 @@ Fabricator(:chat_message, class_name: "Chat::Message") do
 
   initialize_with do |transients|
     Fabricate(
-      transients[:use_service] ? :service_chat_message : :no_service_chat_message,
+      transients[:use_service] ? :chat_message_with_service : :chat_message_without_service,
       **to_params,
     )
   end
 end
 
-Fabricator(:no_service_chat_message, class_name: "Chat::Message") do
+Fabricator(:chat_message_without_service, class_name: "Chat::Message") do
   user
   chat_channel
   message { Faker::Lorem.paragraph_by_chars(number: 500) }
@@ -69,8 +69,14 @@ Fabricator(:no_service_chat_message, class_name: "Chat::Message") do
   after_create { |message, attrs| message.create_mentions }
 end
 
-Fabricator(:service_chat_message, class_name: "Chat::MessageCreator") do
+Fabricator(:chat_message_with_service, class_name: "Chat::CreateMessage") do
-  transient :chat_channel, :user, :message, :in_reply_to, :thread, :upload_ids
+  transient :chat_channel,
+            :user,
+            :message,
+            :in_reply_to,
+            :thread,
+            :upload_ids,
+            :incoming_chat_webhook
 
   initialize_with do |transients|
     channel =
@@ -80,14 +86,15 @@ Fabricator(:service_chat_message, class_name: "Chat::MessageCreator") do
     Group.refresh_automatic_groups!
     channel.add(user)
 
-    resolved_class.create(
+    resolved_class.call(
-      chat_channel: channel,
+      chat_channel_id: channel.id,
-      user: user,
+      guardian: user.guardian,
-      content: transients[:message] || Faker::Lorem.paragraph_by_chars(number: 500),
+      message: transients[:message] || Faker::Lorem.paragraph_by_chars(number: 500),
       thread_id: transients[:thread]&.id,
       in_reply_to_id: transients[:in_reply_to]&.id,
       upload_ids: transients[:upload_ids],
-    ).chat_message
+      incoming_chat_webhook: transients[:incoming_chat_webhook],
+    ).message
   end
 end
 

plugins/chat/spec/integration/thread_replies_count_cache_accuracy_spec.rb

@@ -6,7 +6,14 @@ RSpec.describe "Chat::Thread replies_count cache accuracy" do
   fab!(:user) { Fabricate(:user) }
   fab!(:thread) { Fabricate(:chat_thread) }
 
-  before { SiteSetting.chat_enabled = true }
+  let(:guardian) { user.guardian }
+
+  before do
+    SiteSetting.chat_enabled = true
+    thread.add(user)
+    thread.channel.add(user)
+    Group.refresh_automatic_groups!
+  end
 
   it "keeps an accurate replies_count cache" do
     freeze_time
@@ -17,26 +24,26 @@ RSpec.describe "Chat::Thread replies_count cache accuracy" do
 
     # Create 5 replies
     5.times do |i|
-      Chat::MessageCreator.create(
+      Chat::CreateMessage.call(
-        chat_channel: thread.channel,
+        chat_channel_id: thread.channel_id,
-        user: user,
+        guardian: guardian,
         thread_id: thread.id,
-        content: "Hello world #{i}",
+        message: "Hello world #{i}",
       )
     end
 
     # The job only runs to completion if the cache has not been recently
     # updated, so the DB count will only be 1.
-    expect(thread.replies_count_cache).to eq(5)
+    expect(thread.reload.replies_count_cache).to eq(5)
     expect(thread.reload.replies_count).to eq(1)
 
     # Travel to the future so the cache expires.
     travel_to 6.minutes.from_now
-    Chat::MessageCreator.create(
+    Chat::CreateMessage.call(
-      chat_channel: thread.channel,
+      chat_channel_id: thread.channel_id,
-      user: user,
+      guardian: guardian,
       thread_id: thread.id,
-      content: "Hello world now that time has passed",
+      message: "Hello world now that time has passed",
     )
     expect(thread.replies_count_cache).to eq(6)
     expect(thread.reload.replies_count).to eq(6)
@@ -47,7 +54,7 @@ RSpec.describe "Chat::Thread replies_count cache accuracy" do
     Chat::TrashMessage.call(
       message_id: message_to_destroy.id,
       channel_id: thread.channel_id,
-      guardian: Guardian.new(user),
+      guardian: guardian,
     )
     expect(thread.replies_count_cache).to eq(5)
     expect(thread.reload.replies_count).to eq(5)
@@ -57,7 +64,7 @@ RSpec.describe "Chat::Thread replies_count cache accuracy" do
     Chat::RestoreMessage.call(
       message_id: message_to_destroy.id,
       channel_id: thread.channel_id,
-      guardian: Guardian.new(user),
+      guardian: guardian,
     )
     expect(thread.replies_count_cache).to eq(6)
     expect(thread.reload.replies_count).to eq(6)

plugins/chat/spec/mailers/user_notifications_spec.rb

@@ -170,7 +170,11 @@ describe UserNotifications do
           # Sometimes it's not enough to just fabricate a message
           # and we have to create it like here. In this case all the necessary
           # db records for mentions and notifications will be created under the hood.
-          Chat::MessageCreator.create(chat_channel: channel, user: sender, content: content)
+          Chat::CreateMessage.call(
+            chat_channel_id: channel.id,
+            guardian: sender.guardian,
+            message: content,
+          )
         end
 
         it "returns email for @all mention by default" do

plugins/chat/spec/plugin_helper.rb

@@ -37,17 +37,16 @@ module ChatSystemHelpers
       thread_id = i.zero? ? nil : last_message.thread_id
       last_user = ((users - [last_user]).presence || users).sample
       creator =
-        Chat::MessageCreator.new(
+        Chat::CreateMessage.call(
-          chat_channel: channel,
+          chat_channel_id: channel.id,
           in_reply_to_id: in_reply_to,
           thread_id: thread_id,
-          user: last_user,
+          guardian: last_user.guardian,
-          content: Faker::Lorem.paragraph,
+          message: Faker::Lorem.paragraph,
         )
-      creator.create
 
-      raise creator.error if creator.error
+      raise "#{creator.inspect_steps.inspect}\n\n#{creator.inspect_steps.error}" if creator.failure?
-      last_message = creator.chat_message
+      last_message = creator.message
     end
 
     last_message.thread.set_replies_count_cache(messages_count - 1, update_db: true)

plugins/chat/spec/policies/chat/channel/message_creation_policy_spec.rb

@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+RSpec.describe Chat::Channel::MessageCreationPolicy do
+  subject(:policy) { described_class.new(context) }
+
+  fab!(:user) { Fabricate(:user) }
+
+  let(:guardian) { user.guardian }
+  let(:context) { Service::Base::Context.build(channel: channel, guardian: guardian) }
+
+  describe "#call" do
+    subject(:result) { policy.call }
+
+    context "when channel is a direct message one" do
+      fab!(:channel) { Fabricate(:direct_message_channel) }
+
+      context "when user can't create a message in this channel" do
+        before { channel.closed!(Discourse.system_user) }
+
+        context "when user can't create direct messages" do
+          it "returns false" do
+            expect(result).to be_falsey
+          end
+        end
+
+        context "when user can create direct messages" do
+          before { user.groups << Group.find(Group::AUTO_GROUPS[:trust_level_1]) }
+
+          it "returns true" do
+            expect(result).to be_truthy
+          end
+        end
+      end
+
+      context "when user can create a message in this channel" do
+        it "returns true" do
+          expect(result).to be_truthy
+        end
+      end
+    end
+
+    context "when channel is a category one" do
+      fab!(:channel) { Fabricate(:chat_channel) }
+
+      context "when user can't create a message in this channel" do
+        before { channel.closed!(Discourse.system_user) }
+
+        it "returns false" do
+          expect(result).to be_falsey
+        end
+      end
+
+      context "when user can create a message in this channel" do
+        it "returns true" do
+          expect(result).to be_truthy
+        end
+      end
+    end
+  end
+
+  describe "#reason" do
+    subject(:reason) { policy.reason }
+
+    context "when channel is a direct message one" do
+      fab!(:channel) { Fabricate(:direct_message_channel) }
+
+      it "returns a message related to direct messages" do
+        expect(reason).to eq(I18n.t("chat.errors.user_cannot_send_direct_messages"))
+      end
+    end
+
+    context "when channel is a category one" do
+      let!(:channel) { Fabricate(:chat_channel, status: status) }
+
+      context "when channel is closed" do
+        let(:status) { :closed }
+
+        it "returns a proper message" do
+          expect(reason).to eq(I18n.t("chat.errors.channel_new_message_disallowed.closed"))
+        end
+      end
+
+      context "when channel is archived" do
+        let(:status) { :archived }
+
+        it "returns a proper message" do
+          expect(reason).to eq(I18n.t("chat.errors.channel_new_message_disallowed.archived"))
+        end
+      end
+
+      context "when channel is read-only" do
+        let(:status) { :read_only }
+
+        it "returns a proper message" do
+          expect(reason).to eq(I18n.t("chat.errors.channel_new_message_disallowed.read_only"))
+        end
+      end
+    end
+  end
+end

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

@@ -197,4 +197,268 @@ RSpec.describe Chat::Api::ChannelMessagesController do
       it_behaves_like "chat_message_restoration"
     end
   end
+
+  describe "#create" do
+    fab!(:user) { Fabricate(:user) }
+    fab!(:category) { Fabricate(:category) }
+
+    let(:message) { "This is a message" }
+
+    describe "for category" do
+      fab!(:chat_channel) { Fabricate(:category_channel, chatable: category) }
+
+      context "when current user is silenced" do
+        before do
+          chat_channel.add(user)
+          sign_in(user)
+          UserSilencer.new(user).silence
+        end
+
+        it "raises invalid acces" do
+          post "/chat/#{chat_channel.id}.json", params: { message: message }
+          expect(response.status).to eq(403)
+        end
+      end
+
+      it "errors for regular user when chat is staff-only" do
+        sign_in(user)
+        SiteSetting.chat_allowed_groups = Group::AUTO_GROUPS[:staff]
+
+        post "/chat/#{chat_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(403)
+      end
+
+      it "errors when the user isn't following the channel" do
+        sign_in(user)
+
+        post "/chat/#{chat_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(403)
+      end
+
+      it "errors when the user is not staff and the channel is not open" do
+        Fabricate(:user_chat_channel_membership, chat_channel: chat_channel, user: user)
+        sign_in(user)
+
+        chat_channel.update(status: :closed)
+        post "/chat/#{chat_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(422)
+        expect(response.parsed_body["errors"]).to include(
+          I18n.t("chat.errors.channel_new_message_disallowed.closed"),
+        )
+      end
+
+      it "errors when the user is staff and the channel is not open or closed" do
+        Fabricate(:user_chat_channel_membership, chat_channel: chat_channel, user: admin)
+        sign_in(admin)
+
+        chat_channel.update(status: :closed)
+        post "/chat/#{chat_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(200)
+
+        chat_channel.update(status: :read_only)
+        post "/chat/#{chat_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(422)
+        expect(response.parsed_body["errors"]).to include(
+          I18n.t("chat.errors.channel_new_message_disallowed.read_only"),
+        )
+      end
+
+      context "when the regular user is following the channel" do
+        fab!(:message_1) { Fabricate(:chat_message, chat_channel: chat_channel) }
+        fab!(:membership) do
+          Chat::UserChatChannelMembership.create(
+            user: user,
+            chat_channel: chat_channel,
+            following: true,
+            last_read_message_id: message_1.id,
+          )
+        end
+
+        it "sends a message for regular user when staff-only is disabled and they are following channel" do
+          sign_in(user)
+
+          expect { post "/chat/#{chat_channel.id}.json", params: { message: message } }.to change {
+            Chat::Message.count
+          }.by(1)
+          expect(response.status).to eq(200)
+          expect(Chat::Message.last.message).to eq(message)
+        end
+
+        it "updates the last_read_message_id for the user who sent the message" do
+          sign_in(user)
+          post "/chat/#{chat_channel.id}.json", params: { message: message }
+          expect(response.status).to eq(200)
+          expect(membership.reload.last_read_message_id).to eq(Chat::Message.last.id)
+        end
+
+        it "publishes user tracking state using the new chat message as the last_read_message_id" do
+          sign_in(user)
+          messages =
+            MessageBus.track_publish(
+              Chat::Publisher.user_tracking_state_message_bus_channel(user.id),
+            ) { post "/chat/#{chat_channel.id}.json", params: { message: message } }
+          expect(response.status).to eq(200)
+          expect(messages.first.data["last_read_message_id"]).to eq(Chat::Message.last.id)
+        end
+
+        context "when sending a message in a staged thread" do
+          it "creates the thread and publishes with the staged id" do
+            sign_in(user)
+            chat_channel.update!(threading_enabled: true)
+
+            messages =
+              MessageBus.track_publish do
+                post "/chat/#{chat_channel.id}.json",
+                     params: {
+                       message: message,
+                       in_reply_to_id: message_1.id,
+                       staged_thread_id: "stagedthreadid",
+                     }
+              end
+
+            expect(response.status).to eq(200)
+
+            thread_event = messages.find { |m| m.data["type"] == "thread_created" }
+            expect(thread_event.data["staged_thread_id"]).to eq("stagedthreadid")
+            expect(Chat::Thread.find(thread_event.data["thread_id"])).to be_persisted
+
+            sent_event = messages.find { |m| m.data["type"] == "sent" }
+            expect(sent_event.data["staged_thread_id"]).to eq("stagedthreadid")
+          end
+        end
+
+        context "when sending a message in a thread" do
+          fab!(:thread) do
+            Fabricate(:chat_thread, channel: chat_channel, original_message: message_1)
+          end
+
+          before { sign_in(user) }
+
+          it "does not update the last_read_message_id for the user who sent the message" do
+            post "/chat/#{chat_channel.id}.json", params: { message: message, thread_id: thread.id }
+            expect(response.status).to eq(200)
+            expect(membership.reload.last_read_message_id).to eq(message_1.id)
+          end
+
+          it "publishes user tracking state using the old membership last_read_message_id" do
+            messages =
+              MessageBus.track_publish(
+                Chat::Publisher.user_tracking_state_message_bus_channel(user.id),
+              ) do
+                post "/chat/#{chat_channel.id}.json",
+                     params: {
+                       message: message,
+                       thread_id: thread.id,
+                     }
+              end
+            expect(response.status).to eq(200)
+            expect(messages.first.data["last_read_message_id"]).to eq(message_1.id)
+          end
+
+          context "when thread is not part of the provided channel" do
+            let!(:another_channel) { Fabricate(:category_channel) }
+
+            before do
+              Fabricate(:user_chat_channel_membership, chat_channel: another_channel, user: user)
+            end
+
+            it "returns an error" do
+              post "/chat/#{another_channel.id}.json",
+                   params: {
+                     message: message,
+                     thread_id: thread.id,
+                   }
+              expect(response).to have_http_status :unprocessable_entity
+              expect(response.parsed_body["errors"]).to include(
+                /thread is not part of the provided channel/i,
+              )
+            end
+          end
+
+          context "when provided thread does not match `reply_to_id`" do
+            let!(:another_thread) { Fabricate(:chat_thread, channel: chat_channel) }
+
+            it "returns an error" do
+              post "/chat/#{chat_channel.id}.json",
+                   params: {
+                     message: message,
+                     in_reply_to_id: message_1.id,
+                     thread_id: another_thread.id,
+                   }
+              expect(response).to have_http_status :unprocessable_entity
+              expect(response.parsed_body["errors"]).to include(/does not match parent message/)
+            end
+          end
+        end
+      end
+    end
+
+    describe "for direct message" do
+      fab!(:user1) { Fabricate(:user) }
+      fab!(:user2) { Fabricate(:user) }
+      fab!(:chatable) { Fabricate(:direct_message, users: [user1, user2]) }
+      fab!(:direct_message_channel) { Fabricate(:direct_message_channel, chatable: chatable) }
+
+      it "forces users to follow the channel" do
+        direct_message_channel.remove(user2)
+
+        Chat::Publisher.expects(:publish_new_channel).once
+
+        sign_in(user1)
+
+        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+
+        expect(Chat::UserChatChannelMembership.find_by(user_id: user2.id).following).to be true
+      end
+
+      it "doesn’t call publish new channel when already following" do
+        Chat::Publisher.expects(:publish_new_channel).never
+
+        sign_in(user1)
+
+        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+      end
+
+      it "errors when the user is not part of the direct message channel" do
+        Chat::DirectMessageUser.find_by(user: user1, direct_message: chatable).destroy!
+        sign_in(user1)
+        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(403)
+
+        Chat::UserChatChannelMembership.find_by(user_id: user2.id).update!(following: true)
+        sign_in(user2)
+        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+        expect(response.status).to eq(200)
+      end
+
+      context "when current user is silenced" do
+        before do
+          sign_in(user1)
+          UserSilencer.new(user1).silence
+        end
+
+        it "raises invalid acces" do
+          post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+          expect(response.status).to eq(403)
+        end
+      end
+
+      context "if any of the direct message users is ignoring the acting user" do
+        before do
+          IgnoredUser.create!(user: user2, ignored_user: user1, expiring_at: 1.day.from_now)
+        end
+
+        it "does not force them to follow the channel or send a publish_new_channel message" do
+          direct_message_channel.remove(user2)
+
+          Chat::Publisher.expects(:publish_new_channel).never
+
+          sign_in(user1)
+          post "/chat/#{direct_message_channel.id}.json", params: { message: message }
+
+          expect(Chat::UserChatChannelMembership.find_by(user_id: user2.id).following).to be false
+        end
+      end
+    end
+  end
 end

plugins/chat/spec/requests/chat/incoming_webhooks_controller_spec.rb

@@ -6,7 +6,10 @@ RSpec.describe Chat::IncomingWebhooksController do
   fab!(:chat_channel) { Fabricate(:category_channel) }
   fab!(:webhook) { Fabricate(:incoming_chat_webhook, chat_channel: chat_channel) }
 
-  before { SiteSetting.chat_debug_webhook_payloads = true }
+  before do
+    SiteSetting.chat_enabled = true
+    SiteSetting.chat_debug_webhook_payloads = true
+  end
 
   let(:valid_payload) { { text: "A new signup woo!" } }
 
@@ -35,7 +38,7 @@ RSpec.describe Chat::IncomingWebhooksController do
            params: {
              text: "$" * (SiteSetting.chat_maximum_message_length + 1),
            }
-      expect(response.status).to eq(400)
+      expect(response.status).to eq(422)
     end
 
     it "creates a new chat message" do

plugins/chat/spec/requests/chat_controller_spec.rb

@@ -69,236 +69,6 @@ RSpec.describe Chat::ChatController do
     end
   end
 
-  describe "#create_message" do
-    let(:message) { "This is a message" }
-
-    describe "for category" do
-      fab!(:chat_channel) { Fabricate(:category_channel, chatable: category) }
-
-      context "when current user is silenced" do
-        before do
-          Chat::UserChatChannelMembership.create(
-            user: user,
-            chat_channel: chat_channel,
-            following: true,
-          )
-          sign_in(user)
-          UserSilencer.new(user).silence
-        end
-
-        it "raises invalid acces" do
-          post "/chat/#{chat_channel.id}.json", params: { message: message }
-          expect(response.status).to eq(403)
-        end
-      end
-
-      it "errors for regular user when chat is staff-only" do
-        sign_in(user)
-        SiteSetting.chat_allowed_groups = Group::AUTO_GROUPS[:staff]
-
-        post "/chat/#{chat_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(403)
-      end
-
-      it "errors when the user isn't following the channel" do
-        sign_in(user)
-
-        post "/chat/#{chat_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(403)
-      end
-
-      it "errors when the user is not staff and the channel is not open" do
-        Fabricate(:user_chat_channel_membership, chat_channel: chat_channel, user: user)
-        sign_in(user)
-
-        chat_channel.update(status: :closed)
-        post "/chat/#{chat_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(422)
-        expect(response.parsed_body["errors"]).to include(
-          I18n.t("chat.errors.channel_new_message_disallowed.closed"),
-        )
-      end
-
-      it "errors when the user is staff and the channel is not open or closed" do
-        Fabricate(:user_chat_channel_membership, chat_channel: chat_channel, user: admin)
-        sign_in(admin)
-
-        chat_channel.update(status: :closed)
-        post "/chat/#{chat_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(200)
-
-        chat_channel.update(status: :read_only)
-        post "/chat/#{chat_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(422)
-        expect(response.parsed_body["errors"]).to include(
-          I18n.t("chat.errors.channel_new_message_disallowed.read_only"),
-        )
-      end
-
-      context "when the regular user is following the channel" do
-        fab!(:message_1) { Fabricate(:chat_message, chat_channel: chat_channel) }
-        fab!(:membership) do
-          Chat::UserChatChannelMembership.create(
-            user: user,
-            chat_channel: chat_channel,
-            following: true,
-            last_read_message_id: message_1.id,
-          )
-        end
-
-        it "sends a message for regular user when staff-only is disabled and they are following channel" do
-          sign_in(user)
-
-          expect { post "/chat/#{chat_channel.id}.json", params: { message: message } }.to change {
-            Chat::Message.count
-          }.by(1)
-          expect(response.status).to eq(200)
-          expect(Chat::Message.last.message).to eq(message)
-        end
-
-        it "updates the last_read_message_id for the user who sent the message" do
-          sign_in(user)
-          post "/chat/#{chat_channel.id}.json", params: { message: message }
-          expect(response.status).to eq(200)
-          expect(membership.reload.last_read_message_id).to eq(Chat::Message.last.id)
-        end
-
-        it "publishes user tracking state using the new chat message as the last_read_message_id" do
-          sign_in(user)
-          messages =
-            MessageBus.track_publish(
-              Chat::Publisher.user_tracking_state_message_bus_channel(user.id),
-            ) { post "/chat/#{chat_channel.id}.json", params: { message: message } }
-          expect(response.status).to eq(200)
-          expect(messages.first.data["last_read_message_id"]).to eq(Chat::Message.last.id)
-        end
-
-        context "when sending a message in a staged thread" do
-          it "creates the thread and publishes with the staged id" do
-            sign_in(user)
-            chat_channel.update!(threading_enabled: true)
-
-            messages =
-              MessageBus.track_publish do
-                post "/chat/#{chat_channel.id}.json",
-                     params: {
-                       message: message,
-                       in_reply_to_id: message_1.id,
-                       staged_thread_id: "stagedthreadid",
-                     }
-              end
-
-            expect(response.status).to eq(200)
-
-            thread_event = messages.find { |m| m.data["type"] == "thread_created" }
-            expect(thread_event.data["staged_thread_id"]).to eq("stagedthreadid")
-            expect(Chat::Thread.find(thread_event.data["thread_id"])).to be_persisted
-
-            sent_event = messages.find { |m| m.data["type"] == "sent" }
-            expect(sent_event.data["staged_thread_id"]).to eq("stagedthreadid")
-          end
-        end
-
-        context "when sending a message in a thread" do
-          fab!(:thread) do
-            Fabricate(:chat_thread, channel: chat_channel, original_message: message_1)
-          end
-
-          it "does not update the last_read_message_id for the user who sent the message" do
-            sign_in(user)
-            post "/chat/#{chat_channel.id}.json", params: { message: message, thread_id: thread.id }
-            expect(response.status).to eq(200)
-            expect(membership.reload.last_read_message_id).to eq(message_1.id)
-          end
-
-          it "publishes user tracking state using the old membership last_read_message_id" do
-            sign_in(user)
-            messages =
-              MessageBus.track_publish(
-                Chat::Publisher.user_tracking_state_message_bus_channel(user.id),
-              ) do
-                post "/chat/#{chat_channel.id}.json",
-                     params: {
-                       message: message,
-                       thread_id: thread.id,
-                     }
-              end
-            expect(response.status).to eq(200)
-            expect(messages.first.data["last_read_message_id"]).to eq(message_1.id)
-          end
-        end
-      end
-    end
-
-    describe "for direct message" do
-      fab!(:user1) { Fabricate(:user) }
-      fab!(:user2) { Fabricate(:user) }
-      fab!(:chatable) { Fabricate(:direct_message, users: [user1, user2]) }
-      fab!(:direct_message_channel) { Fabricate(:direct_message_channel, chatable: chatable) }
-
-      it "forces users to follow the channel" do
-        direct_message_channel.remove(user2)
-
-        Chat::Publisher.expects(:publish_new_channel).once
-
-        sign_in(user1)
-
-        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-
-        expect(Chat::UserChatChannelMembership.find_by(user_id: user2.id).following).to be true
-      end
-
-      it "doesn’t call publish new channel when already following" do
-        Chat::Publisher.expects(:publish_new_channel).never
-
-        sign_in(user1)
-
-        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-      end
-
-      it "errors when the user is not part of the direct message channel" do
-        Chat::DirectMessageUser.find_by(user: user1, direct_message: chatable).destroy!
-        sign_in(user1)
-        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(403)
-
-        Chat::UserChatChannelMembership.find_by(user_id: user2.id).update!(following: true)
-        sign_in(user2)
-        post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-        expect(response.status).to eq(200)
-      end
-
-      context "when current user is silenced" do
-        before do
-          sign_in(user1)
-          UserSilencer.new(user1).silence
-        end
-
-        it "raises invalid acces" do
-          post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-          expect(response.status).to eq(403)
-        end
-      end
-
-      context "if any of the direct message users is ignoring the acting user" do
-        before do
-          IgnoredUser.create!(user: user2, ignored_user: user1, expiring_at: 1.day.from_now)
-        end
-
-        it "does not force them to follow the channel or send a publish_new_channel message" do
-          direct_message_channel.remove(user2)
-
-          Chat::Publisher.expects(:publish_new_channel).never
-
-          sign_in(user1)
-          post "/chat/#{direct_message_channel.id}.json", params: { message: message }
-
-          expect(Chat::UserChatChannelMembership.find_by(user_id: user2.id).following).to be false
-        end
-      end
-    end
-  end
-
   describe "#rebake" do
     fab!(:chat_message) { Fabricate(:chat_message, chat_channel: chat_channel, user: user) }
 

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

@@ -0,0 +1,60 @@
+# frozen_string_literal: true
+
+RSpec.describe Chat::Action::PublishAndFollowDirectMessageChannel do
+  subject(:action) { described_class.call(channel_membership: membership) }
+
+  fab!(:user) { Fabricate(:user) }
+
+  let(:membership) { user.user_chat_channel_memberships.last }
+
+  before { channel.add(user) }
+
+  context "when channel is not a direct message one" do
+    fab!(:channel) { Fabricate(:chat_channel) }
+
+    it "does not publish anything" do
+      Chat::Publisher.expects(:publish_new_channel).never
+      action
+    end
+
+    it "does not update memberships" do
+      expect { action }.not_to change {
+        channel.user_chat_channel_memberships.where(following: true).count
+      }
+    end
+  end
+
+  context "when channel is a direct message one" do
+    fab!(:channel) { Fabricate(:direct_message_channel) }
+
+    context "when no users allow communication" do
+      it "does not publish anything" do
+        Chat::Publisher.expects(:publish_new_channel).never
+        action
+      end
+
+      it "does not update memberships" do
+        expect { action }.not_to change {
+          channel.user_chat_channel_memberships.where(following: true).count
+        }
+      end
+    end
+
+    context "when at least one user allows communication" do
+      let(:users) { channel.user_chat_channel_memberships.map(&:user) }
+
+      before { channel.user_chat_channel_memberships.update_all(following: false) }
+
+      it "publishes the channel" do
+        Chat::Publisher.expects(:publish_new_channel).with(channel, includes(*users))
+        action
+      end
+
+      it "sets autofollow for these users" do
+        expect { action }.to change {
+          channel.user_chat_channel_memberships.where(following: true).count
+        }.by(3)
+      end
+    end
+  end
+end

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

@@ -0,0 +1,378 @@
+# frozen_string_literal: true
+
+RSpec.describe Chat::CreateMessage do
+  describe described_class::Contract, type: :model do
+    subject(:contract) { described_class.new(upload_ids: upload_ids) }
+
+    let(:upload_ids) { nil }
+
+    it { is_expected.to validate_presence_of :chat_channel_id }
+
+    context "when uploads are not provided" do
+      it { is_expected.to validate_presence_of :message }
+    end
+
+    context "when uploads are provided" do
+      let(:upload_ids) { "2,3" }
+
+      it { is_expected.not_to validate_presence_of :message }
+    end
+  end
+
+  describe ".call" do
+    subject(:result) { described_class.call(params) }
+
+    fab!(:user) { Fabricate(:user) }
+    fab!(:other_user) { Fabricate(:user) }
+    fab!(:channel) { Fabricate(:chat_channel, threading_enabled: true) }
+    fab!(:thread) { Fabricate(:chat_thread, channel: channel) }
+    fab!(:upload) { Fabricate(:upload, user: user) }
+    fab!(:draft) { Fabricate(:chat_draft, user: user, chat_channel: channel) }
+
+    let(:guardian) { user.guardian }
+    let(:content) { "A new message @#{other_user.username_lower}" }
+    let(:params) do
+      { guardian: guardian, chat_channel_id: channel.id, message: content, upload_ids: [upload.id] }
+    end
+    let(:message) { result[:message].reload }
+
+    shared_examples "creating a new message" do
+      it "saves the message" do
+        expect { result }.to change { Chat::Message.count }.by(1)
+        expect(message).to have_attributes(message: content)
+      end
+
+      it "cooks the message" do
+        expect(message).to be_cooked
+      end
+
+      it "creates mentions" do
+        expect { result }.to change { Chat::Mention.count }.by(1)
+      end
+
+      context "when coming from a webhook" do
+        let(:incoming_webhook) { Fabricate(:incoming_chat_webhook, chat_channel: channel) }
+
+        before { params[:incoming_chat_webhook] = incoming_webhook }
+
+        it "creates a webhook event" do
+          expect { result }.to change { Chat::WebhookEvent.count }.by(1)
+        end
+      end
+
+      it "attaches uploads" do
+        expect(message.uploads).to match_array(upload)
+      end
+
+      it "deletes drafts" do
+        expect { result }.to change { Chat::Draft.count }.by(-1)
+      end
+
+      it "publishes the new message" do
+        Chat::Publisher.expects(:publish_new!).with(
+          channel,
+          instance_of(Chat::Message),
+          nil,
+          staged_thread_id: nil,
+        )
+        result
+      end
+
+      it "enqueues a job to process message" do
+        result
+        expect_job_enqueued(job: Jobs::Chat::ProcessMessage, args: { chat_message_id: message.id })
+      end
+
+      it "notifies the new message" do
+        result
+        expect_job_enqueued(
+          job: Jobs::Chat::SendMessageNotifications,
+          args: {
+            chat_message_id: message.id,
+            timestamp: message.created_at.iso8601(6),
+            reason: "new",
+          },
+        )
+      end
+
+      it "triggers a Discourse event" do
+        DiscourseEvent.expects(:trigger).with(
+          :chat_message_created,
+          instance_of(Chat::Message),
+          channel,
+          user,
+        )
+        result
+      end
+
+      it "processes the direct message channel" do
+        Chat::Action::PublishAndFollowDirectMessageChannel.expects(:call).with(
+          channel_membership: membership,
+        )
+        result
+      end
+    end
+
+    shared_examples "a message in a thread" do
+      let(:thread_membership) { Chat::UserChatThreadMembership.find_by(user: user) }
+      let(:original_user) { thread.original_message_user }
+
+      before do
+        Chat::UserChatThreadMembership.where(user: original_user).delete_all
+        Discourse.redis.flushdb # for replies count cache
+      end
+
+      it "increments the replies count" do
+        result
+        expect(thread.replies_count_cache).to eq(1)
+      end
+
+      it "adds current user to the thread" do
+        expect { result }.to change { Chat::UserChatThreadMembership.where(user: user).count }.by(1)
+      end
+
+      it "sets last_read_message on the thread membership" do
+        result
+        expect(thread_membership.last_read_message).to eq message
+      end
+
+      it "adds original message user to the thread" do
+        expect { result }.to change {
+          Chat::UserChatThreadMembership.where(user: original_user).count
+        }.by(1)
+      end
+
+      it "publishes user tracking state" do
+        Chat::Publisher.expects(:publish_user_tracking_state!).with(user, channel, existing_message)
+        result
+      end
+
+      it "doesn't update channel last_message attribute" do
+        expect { result }.not_to change { channel.reload.last_message }
+      end
+
+      it "updates thread last_message attribute" do
+        result
+        expect(thread.reload.last_message).to eq message
+      end
+
+      it "doesn't update last_read_message attribute on the channel membership" do
+        expect { result }.not_to change { membership.reload.last_read_message }
+      end
+    end
+
+    context "when user is silenced" do
+      before { UserSilencer.new(user).silence }
+
+      it { is_expected.to fail_a_policy(:no_silenced_user) }
+    end
+
+    context "when user is not silenced" do
+      context "when mandatory parameters are missing" do
+        before { params[:chat_channel_id] = "" }
+
+        it { is_expected.to fail_a_contract }
+      end
+
+      context "when mandatory parameters are present" do
+        context "when channel model is not found" do
+          before { params[:chat_channel_id] = -1 }
+
+          it { is_expected.to fail_to_find_a_model(:channel) }
+        end
+
+        context "when channel model is found" do
+          context "when user can't join channel" do
+            let(:guardian) { Guardian.new }
+
+            it { is_expected.to fail_a_policy(:allowed_to_join_channel) }
+          end
+
+          context "when user can join channel" do
+            before { user.groups << Group.find(Group::AUTO_GROUPS[:trust_level_1]) }
+
+            context "when user can't create a message in the channel" do
+              before { channel.closed!(Discourse.system_user) }
+
+              it { is_expected.to fail_a_policy(:allowed_to_create_message_in_channel) }
+            end
+
+            context "when user can create a message in the channel" do
+              context "when user is not a member of the channel" do
+                it { is_expected.to fail_to_find_a_model(:channel_membership) }
+              end
+
+              context "when user is a member of the channel" do
+                fab!(:existing_message) { Fabricate(:chat_message, chat_channel: channel) }
+
+                let(:membership) { Chat::UserChatChannelMembership.last }
+
+                before do
+                  channel.add(user).update!(last_read_message: existing_message)
+                  DiscourseEvent.stubs(:trigger)
+                end
+
+                context "when message is a reply" do
+                  before { params[:in_reply_to_id] = reply_to.id }
+
+                  context "when reply is not part of the channel" do
+                    fab!(:reply_to) { Fabricate(:chat_message) }
+
+                    it { is_expected.to fail_a_policy(:ensure_reply_consistency) }
+                  end
+
+                  context "when reply is part of the channel" do
+                    fab!(:reply_to) { Fabricate(:chat_message, chat_channel: channel) }
+
+                    context "when reply is in a thread" do
+                      fab!(:thread) do
+                        Fabricate(:chat_thread, channel: channel, original_message: reply_to)
+                      end
+
+                      it_behaves_like "creating a new message"
+                      it_behaves_like "a message in a thread"
+
+                      it "assigns the thread to the new message" do
+                        expect(message).to have_attributes(
+                          in_reply_to: an_object_having_attributes(thread: thread),
+                          thread: thread,
+                        )
+                      end
+
+                      it "does not publish the existing thread" do
+                        Chat::Publisher.expects(:publish_thread_created!).never
+                        result
+                      end
+                    end
+
+                    context "when reply is not in a thread" do
+                      let(:thread) { Chat::Thread.last }
+
+                      it_behaves_like "creating a new message"
+                      it_behaves_like "a message in a thread" do
+                        let(:original_user) { reply_to.user }
+                      end
+
+                      it "creates a new thread" do
+                        expect { result }.to change { Chat::Thread.count }.by(1)
+                        expect(message).to have_attributes(
+                          in_reply_to: an_object_having_attributes(thread: thread),
+                          thread: thread,
+                        )
+                      end
+
+                      context "when threading is enabled in channel" do
+                        it "publishes the new thread" do
+                          Chat::Publisher.expects(:publish_thread_created!).with(
+                            channel,
+                            reply_to,
+                            instance_of(Integer),
+                            nil,
+                          )
+                          result
+                        end
+                      end
+
+                      context "when threading is disabled in channel" do
+                        before { channel.update!(threading_enabled: false) }
+
+                        it "does not publish the new thread" do
+                          Chat::Publisher.expects(:publish_thread_created!).never
+                          result
+                        end
+                      end
+                    end
+                  end
+                end
+
+                context "when a thread is provided" do
+                  before { params[:thread_id] = thread.id }
+
+                  context "when thread is not part of the provided channel" do
+                    let(:thread) { Fabricate(:chat_thread) }
+
+                    it { is_expected.to fail_a_policy(:ensure_valid_thread_for_channel) }
+                  end
+
+                  context "when thread is part of the provided channel" do
+                    let(:thread) { Fabricate(:chat_thread, channel: channel) }
+
+                    context "when replying to an existing message" do
+                      let(:reply_to) { Fabricate(:chat_message, chat_channel: channel) }
+
+                      context "when reply thread does not match the provided thread" do
+                        let!(:another_thread) do
+                          Fabricate(:chat_thread, channel: channel, original_message: reply_to)
+                        end
+
+                        before { params[:in_reply_to_id] = reply_to.id }
+
+                        it { is_expected.to fail_a_policy(:ensure_thread_matches_parent) }
+                      end
+
+                      context "when reply thread matches the provided thread" do
+                        before { reply_to.update!(thread: thread) }
+
+                        it_behaves_like "creating a new message"
+                        it_behaves_like "a message in a thread"
+
+                        it "does not publish the thread" do
+                          Chat::Publisher.expects(:publish_thread_created!).never
+                          result
+                        end
+                      end
+                    end
+
+                    context "when not replying to an existing message" do
+                      it_behaves_like "creating a new message"
+                      it_behaves_like "a message in a thread"
+
+                      it "does not publish the thread" do
+                        Chat::Publisher.expects(:publish_thread_created!).never
+                        result
+                      end
+                    end
+                  end
+                end
+
+                context "when nor thread nor reply is provided" do
+                  context "when message is not valid" do
+                    let(:content) { "a" * (SiteSetting.chat_maximum_message_length + 1) }
+
+                    it { is_expected.to fail_with_an_invalid_model(:message) }
+                  end
+
+                  context "when message is valid" do
+                    it_behaves_like "creating a new message"
+
+                    it "updates membership last_read_message attribute" do
+                      expect { result }.to change { membership.reload.last_read_message }
+                    end
+
+                    it "updates channel last_message attribute" do
+                      result
+                      expect(channel.reload.last_message).to eq message
+                    end
+
+                    it "publishes user tracking state" do
+                      Chat::Publisher
+                        .expects(:publish_user_tracking_state!)
+                        .with(user, channel, existing_message)
+                        .never
+                      Chat::Publisher.expects(:publish_user_tracking_state!).with(
+                        user,
+                        channel,
+                        instance_of(Chat::Message),
+                      )
+                      result
+                    end
+                  end
+                end
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end