b686d14044
When businesses use WhatsApp Business App (co-existence mode) or Instagram App or TikTok alongside Chatwoot, messages sent from the native apps were not synced properly back to Chatwoot. This left agents with an incomplete conversation history and no visibility into responses sent outside the dashboard. Additionally, if these echo messages did arrive, they appeared as "Sent by: Bot" in the UI since they had no sender, making it confusing for agents. This PR subscribes to WhatsApp `smb_message_echoes` webhook events and routes them through the existing service with an `outgoing_echo` flag, mirroring how Instagram already handles echoes. On the Instagram side, echo messages now also carry the `external_echo` content attribute and `delivered` status. On the frontend, messages with `externalEcho` are distinguished from bot messages showing a "Native app" avatar and an advisory note encouraging agents to reply from Chatwoot to maintain the service window. <img width="1518" height="524" alt="CleanShot 2026-01-29 at 13 37 57@2x" src="https://github.com/user-attachments/assets/5aa0b552-6382-441f-96aa-9a62ca716e4a" /> Fixes https://linear.app/chatwoot/issue/CW-4204/display-messages-not-sent-from-chatwoot-in-case-of-outgoing-echo Fixes https://linear.app/chatwoot/issue/PLA-33/incoming-from-me-messages-from-whatsapp-business-app-are-not-falling
429 lines
16 KiB
Ruby
429 lines
16 KiB
Ruby
# == Schema Information
|
|
#
|
|
# Table name: messages
|
|
#
|
|
# id :integer not null, primary key
|
|
# additional_attributes :jsonb
|
|
# content :text
|
|
# content_attributes :json
|
|
# content_type :integer default("text"), not null
|
|
# external_source_ids :jsonb
|
|
# message_type :integer not null
|
|
# private :boolean default(FALSE), not null
|
|
# processed_message_content :text
|
|
# sender_type :string
|
|
# sentiment :jsonb
|
|
# status :integer default("sent")
|
|
# created_at :datetime not null
|
|
# updated_at :datetime not null
|
|
# account_id :integer not null
|
|
# conversation_id :integer not null
|
|
# inbox_id :integer not null
|
|
# sender_id :bigint
|
|
# source_id :text
|
|
#
|
|
# Indexes
|
|
#
|
|
# idx_messages_account_content_created (account_id,content_type,created_at)
|
|
# index_messages_on_account_created_type (account_id,created_at,message_type)
|
|
# index_messages_on_account_id (account_id)
|
|
# index_messages_on_account_id_and_inbox_id (account_id,inbox_id)
|
|
# index_messages_on_additional_attributes_campaign_id (((additional_attributes -> 'campaign_id'::text))) USING gin
|
|
# index_messages_on_content (content) USING gin
|
|
# index_messages_on_conversation_account_type_created (conversation_id,account_id,message_type,created_at)
|
|
# index_messages_on_conversation_id (conversation_id)
|
|
# index_messages_on_created_at (created_at)
|
|
# index_messages_on_inbox_id (inbox_id)
|
|
# index_messages_on_sender_type_and_sender_id (sender_type,sender_id)
|
|
# index_messages_on_source_id (source_id)
|
|
#
|
|
|
|
class Message < ApplicationRecord
|
|
searchkick callbacks: false if ChatwootApp.advanced_search_allowed?
|
|
|
|
include MessageFilterHelpers
|
|
include Liquidable
|
|
NUMBER_OF_PERMITTED_ATTACHMENTS = 15
|
|
|
|
TEMPLATE_PARAMS_SCHEMA = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'template_params': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'name': { 'type': 'string' },
|
|
'category': { 'type': 'string' },
|
|
'language': { 'type': 'string' },
|
|
'namespace': { 'type': 'string' },
|
|
'processed_params': { 'type': 'object' }
|
|
},
|
|
'required': %w[name]
|
|
}
|
|
}
|
|
}.to_json.freeze
|
|
|
|
before_validation :ensure_content_type
|
|
before_validation :prevent_message_flooding
|
|
before_save :ensure_processed_message_content
|
|
before_save :ensure_in_reply_to
|
|
|
|
validates :account_id, presence: true
|
|
validates :inbox_id, presence: true
|
|
validates :conversation_id, presence: true
|
|
validates_with ContentAttributeValidator
|
|
validates_with JsonSchemaValidator,
|
|
schema: TEMPLATE_PARAMS_SCHEMA,
|
|
attribute_resolver: ->(record) { record.additional_attributes }
|
|
|
|
validates :content_type, presence: true
|
|
validates :content, length: { maximum: 150_000 }
|
|
validates :processed_message_content, length: { maximum: 150_000 }
|
|
|
|
# when you have a temperory id in your frontend and want it echoed back via action cable
|
|
attr_accessor :echo_id
|
|
|
|
enum message_type: { incoming: 0, outgoing: 1, activity: 2, template: 3 }
|
|
enum content_type: {
|
|
text: 0,
|
|
input_text: 1,
|
|
input_textarea: 2,
|
|
input_email: 3,
|
|
input_select: 4,
|
|
cards: 5,
|
|
form: 6,
|
|
article: 7,
|
|
incoming_email: 8,
|
|
input_csat: 9,
|
|
integrations: 10,
|
|
sticker: 11,
|
|
voice_call: 12
|
|
}
|
|
enum status: { sent: 0, delivered: 1, read: 2, failed: 3 }
|
|
# [:submitted_email, :items, :submitted_values] : Used for bot message types
|
|
# [:email] : Used by conversation_continuity incoming email messages
|
|
# [:in_reply_to] : Used to reply to a particular tweet in threads
|
|
# [:deleted] : Used to denote whether the message was deleted by the agent
|
|
# [:external_created_at] : Can specify if the message was created at a different timestamp externally
|
|
# [:external_error : Can specify if the message creation failed due to an error at external API
|
|
# [:data] : Used for structured content types such as voice_call
|
|
store :content_attributes, accessors: [:submitted_email, :items, :submitted_values, :email, :in_reply_to, :deleted,
|
|
:external_created_at, :story_sender, :story_id, :external_error,
|
|
:translations, :in_reply_to_external_id, :is_unsupported, :data], coder: JSON
|
|
|
|
store :external_source_ids, accessors: [:slack], coder: JSON, prefix: :external_source_id
|
|
|
|
scope :created_since, ->(datetime) { where('created_at > ?', datetime) }
|
|
scope :chat, -> { where.not(message_type: :activity).where(private: false) }
|
|
scope :non_activity_messages, -> { where.not(message_type: :activity).reorder('created_at desc') }
|
|
scope :today, -> { where("date_trunc('day', created_at) = ?", Date.current) }
|
|
scope :voice_calls, -> { where(content_type: :voice_call) }
|
|
|
|
# TODO: Get rid of default scope
|
|
# https://stackoverflow.com/a/1834250/939299
|
|
# if you want to change order, use `reorder`
|
|
default_scope { order(created_at: :asc) }
|
|
|
|
belongs_to :account
|
|
belongs_to :inbox
|
|
belongs_to :conversation, touch: true
|
|
belongs_to :sender, polymorphic: true, optional: true
|
|
|
|
has_many :attachments, dependent: :destroy, autosave: true, before_add: :validate_attachments_limit
|
|
has_one :csat_survey_response, dependent: :destroy_async
|
|
has_many :notifications, as: :primary_actor, dependent: :destroy_async
|
|
|
|
after_create_commit :execute_after_create_commit_callbacks
|
|
|
|
after_update_commit :dispatch_update_event
|
|
after_commit :reindex_for_search, if: :should_index?, on: [:create, :update]
|
|
|
|
def channel_token
|
|
@token ||= inbox.channel.try(:page_access_token)
|
|
end
|
|
|
|
def push_event_data
|
|
data = attributes.symbolize_keys.merge(
|
|
created_at: created_at.to_i,
|
|
message_type: message_type_before_type_cast,
|
|
conversation_id: conversation&.display_id,
|
|
conversation: conversation.present? ? conversation_push_event_data : nil
|
|
)
|
|
data[:echo_id] = echo_id if echo_id.present?
|
|
data[:attachments] = attachments.map(&:push_event_data) if attachments.present?
|
|
merge_sender_attributes(data)
|
|
end
|
|
|
|
def conversation_push_event_data
|
|
{
|
|
assignee_id: conversation.assignee_id,
|
|
unread_count: conversation.unread_incoming_messages.count,
|
|
last_activity_at: conversation.last_activity_at.to_i,
|
|
contact_inbox: { source_id: conversation.contact_inbox.source_id }
|
|
}
|
|
end
|
|
|
|
def merge_sender_attributes(data)
|
|
data[:sender] = sender.push_event_data if sender && !sender.is_a?(AgentBot)
|
|
data[:sender] = sender.push_event_data(inbox) if sender.is_a?(AgentBot)
|
|
data
|
|
end
|
|
|
|
def webhook_data
|
|
data = {
|
|
account: account.webhook_data,
|
|
additional_attributes: additional_attributes,
|
|
content_attributes: content_attributes,
|
|
content_type: content_type,
|
|
content: outgoing_content,
|
|
conversation: conversation.webhook_data,
|
|
created_at: created_at,
|
|
id: id,
|
|
inbox: inbox.webhook_data,
|
|
message_type: message_type,
|
|
private: private,
|
|
sender: sender.try(:webhook_data),
|
|
source_id: source_id
|
|
}
|
|
data[:attachments] = attachments.map(&:push_event_data) if attachments.present?
|
|
data
|
|
end
|
|
|
|
# Method to get content with survey URL for outgoing channel delivery
|
|
def outgoing_content
|
|
MessageContentPresenter.new(self).outgoing_content
|
|
end
|
|
|
|
def email_notifiable_message?
|
|
return false if private?
|
|
return false if %w[outgoing template].exclude?(message_type)
|
|
return false if template? && %w[input_csat text].exclude?(content_type)
|
|
|
|
true
|
|
end
|
|
|
|
def auto_reply_email?
|
|
return false unless incoming_email? || inbox.email?
|
|
|
|
content_attributes.dig(:email, :auto_reply) == true
|
|
end
|
|
|
|
def valid_first_reply?
|
|
return false unless human_response? && !private?
|
|
return false if conversation.first_reply_created_at.present?
|
|
return false if conversation.messages.outgoing
|
|
.where.not(sender_type: ['AgentBot', 'Captain::Assistant'])
|
|
.where.not(private: true)
|
|
.where("(additional_attributes->'campaign_id') is null").count > 1
|
|
|
|
true
|
|
end
|
|
|
|
def save_story_info(story_info)
|
|
self.content_attributes = content_attributes.merge(
|
|
{
|
|
story_id: story_info['id'],
|
|
story_sender: inbox.channel.instagram_id,
|
|
story_url: story_info['url']
|
|
}
|
|
)
|
|
save!
|
|
end
|
|
|
|
def send_update_event
|
|
Rails.configuration.dispatcher.dispatch(MESSAGE_UPDATED, Time.zone.now, message: self, performed_by: Current.executed_by,
|
|
previous_changes: previous_changes)
|
|
end
|
|
|
|
def should_index?
|
|
return false unless ChatwootApp.advanced_search_allowed?
|
|
return false unless incoming? || outgoing?
|
|
# For Chatwoot Cloud:
|
|
# - Enable indexing only if the account is paid.
|
|
# - The `advanced_search_indexing` feature flag is used only in the cloud.
|
|
#
|
|
# For Self-hosted:
|
|
# - Adding an extra feature flag here would cause confusion.
|
|
# - If the user has configured Elasticsearch, enabling `advanced_search`
|
|
# should automatically work without any additional flags.
|
|
return false if ChatwootApp.chatwoot_cloud? && !account.feature_enabled?('advanced_search_indexing')
|
|
|
|
true
|
|
end
|
|
|
|
def search_data
|
|
Messages::SearchDataPresenter.new(self).search_data
|
|
end
|
|
|
|
# Returns message content suitable for LLM consumption
|
|
# Falls back to audio transcription or attachment placeholder when content is nil
|
|
def content_for_llm
|
|
return content if content.present?
|
|
|
|
audio_transcription = attachments
|
|
.where(file_type: :audio)
|
|
.filter_map { |att| att.meta&.dig('transcribed_text') }
|
|
.join(' ')
|
|
.presence
|
|
return "[Voice Message] #{audio_transcription}" if audio_transcription.present?
|
|
|
|
'[Attachment]' if attachments.any?
|
|
end
|
|
|
|
private
|
|
|
|
def prevent_message_flooding
|
|
# Added this to cover the validation specs in messages
|
|
# We can revisit and see if we can remove this later
|
|
return if conversation.blank?
|
|
|
|
# there are cases where automations can result in message loops, we need to prevent such cases.
|
|
if conversation.messages.where('created_at >= ?', 1.minute.ago).count >= Limits.conversation_message_per_minute_limit
|
|
Rails.logger.error "Too many message: Account Id - #{account_id} : Conversation id - #{conversation_id}"
|
|
errors.add(:base, 'Too many messages')
|
|
end
|
|
end
|
|
|
|
def ensure_processed_message_content
|
|
text_content_quoted = content_attributes.dig(:email, :text_content, :quoted)
|
|
html_content_quoted = content_attributes.dig(:email, :html_content, :quoted)
|
|
|
|
message_content = text_content_quoted || html_content_quoted || content
|
|
self.processed_message_content = message_content&.truncate(150_000)
|
|
end
|
|
|
|
# fetch the in_reply_to message and set the external id
|
|
def ensure_in_reply_to
|
|
in_reply_to = content_attributes[:in_reply_to]
|
|
in_reply_to_external_id = content_attributes[:in_reply_to_external_id]
|
|
|
|
Messages::InReplyToMessageBuilder.new(
|
|
message: self,
|
|
in_reply_to: in_reply_to,
|
|
in_reply_to_external_id: in_reply_to_external_id
|
|
).perform
|
|
end
|
|
|
|
def ensure_content_type
|
|
self.content_type ||= Message.content_types[:text]
|
|
end
|
|
|
|
def execute_after_create_commit_callbacks
|
|
# rails issue with order of active record callbacks being executed https://github.com/rails/rails/issues/20911
|
|
reopen_conversation
|
|
set_conversation_activity
|
|
dispatch_create_events
|
|
send_reply
|
|
execute_message_template_hooks
|
|
update_contact_activity
|
|
end
|
|
|
|
def update_contact_activity
|
|
sender.update(last_activity_at: DateTime.now) if sender.is_a?(Contact)
|
|
end
|
|
|
|
def update_waiting_since
|
|
waiting_present = conversation.waiting_since.present?
|
|
|
|
if waiting_present && !private
|
|
if human_response?
|
|
Rails.configuration.dispatcher.dispatch(
|
|
REPLY_CREATED, Time.zone.now, waiting_since: conversation.waiting_since, message: self
|
|
)
|
|
conversation.update(waiting_since: nil)
|
|
elsif bot_response?
|
|
# Bot responses also clear waiting_since (simpler than checking on next customer message)
|
|
conversation.update(waiting_since: nil)
|
|
end
|
|
end
|
|
|
|
# Set waiting_since when customer sends a message (if currently blank)
|
|
conversation.update(waiting_since: created_at) if incoming? && conversation.waiting_since.blank?
|
|
end
|
|
|
|
def human_response?
|
|
# if the sender is not a user, it's not a human response
|
|
# if automation rule id is present, it's not a human response
|
|
# if campaign id is present, it's not a human response
|
|
# external echo messages are responses sent from the native app (WhatsApp Business, Instagram)
|
|
outgoing? &&
|
|
content_attributes['automation_rule_id'].blank? &&
|
|
additional_attributes['campaign_id'].blank? &&
|
|
(sender.is_a?(User) || content_attributes['external_echo'].present?)
|
|
end
|
|
|
|
def bot_response?
|
|
# Check if this is a response from AgentBot or Captain::Assistant
|
|
outgoing? && sender_type.in?(['AgentBot', 'Captain::Assistant'])
|
|
end
|
|
|
|
def dispatch_create_events
|
|
Rails.configuration.dispatcher.dispatch(MESSAGE_CREATED, Time.zone.now, message: self, performed_by: Current.executed_by)
|
|
|
|
if valid_first_reply?
|
|
Rails.configuration.dispatcher.dispatch(FIRST_REPLY_CREATED, Time.zone.now, message: self, performed_by: Current.executed_by)
|
|
conversation.update(first_reply_created_at: created_at, waiting_since: nil)
|
|
else
|
|
update_waiting_since
|
|
end
|
|
end
|
|
|
|
def dispatch_update_event
|
|
# ref: https://github.com/rails/rails/issues/44500
|
|
# we want to skip the update event if the message is not updated
|
|
return if previous_changes.blank?
|
|
|
|
send_update_event
|
|
end
|
|
|
|
def send_reply
|
|
# FIXME: Giving it few seconds for the attachment to be uploaded to the service
|
|
# active storage attaches the file only after commit
|
|
attachments.blank? ? ::SendReplyJob.perform_later(id) : ::SendReplyJob.set(wait: 2.seconds).perform_later(id)
|
|
end
|
|
|
|
def reopen_conversation
|
|
return if conversation.muted?
|
|
return unless incoming?
|
|
|
|
conversation.open! if conversation.snoozed?
|
|
|
|
reopen_resolved_conversation if conversation.resolved?
|
|
end
|
|
|
|
def reopen_resolved_conversation
|
|
# mark resolved bot conversation as pending to be reopened by bot processor service
|
|
if conversation.inbox.active_bot?
|
|
conversation.pending!
|
|
elsif conversation.inbox.api?
|
|
Current.executed_by = sender if reopened_by_contact?
|
|
conversation.open!
|
|
else
|
|
conversation.open!
|
|
end
|
|
end
|
|
|
|
def reopened_by_contact?
|
|
incoming? && !private? && Current.user.class != sender.class && sender.instance_of?(Contact)
|
|
end
|
|
|
|
def execute_message_template_hooks
|
|
::MessageTemplates::HookExecutionService.new(message: self).perform
|
|
end
|
|
|
|
def validate_attachments_limit(_attachment)
|
|
errors.add(:attachments, message: 'exceeded maximum allowed') if attachments.size >= NUMBER_OF_PERMITTED_ATTACHMENTS
|
|
end
|
|
|
|
def set_conversation_activity
|
|
# rubocop:disable Rails/SkipsModelValidations
|
|
conversation.update_columns(last_activity_at: created_at)
|
|
# rubocop:enable Rails/SkipsModelValidations
|
|
end
|
|
|
|
def reindex_for_search
|
|
reindex(mode: :async)
|
|
end
|
|
end
|
|
|
|
Message.prepend_mod_with('Message')
|