iachat/app/services/whatsapp/incoming_message_base_service.rb

# Mostly modeled after the intial implementation of the service based on 360 Dialog
# https://docs.360dialog.com/whatsapp-api/whatsapp-api/media
# https://developers.facebook.com/docs/whatsapp/api/media/
class Whatsapp::IncomingMessageBaseService # rubocop:disable Metrics/ClassLength
  include ::Whatsapp::IncomingMessageServiceHelpers

  pattr_initialize [:inbox!, :params!, :outgoing_echo]

  def perform
    processed_params

    if processed_params.try(:[], :statuses).present?
      process_statuses
    elsif messages_data.present?
      process_messages
    end
  end

  # Returns messages array for both regular messages and echo events
  def messages_data
    @processed_params&.dig(:messages) || @processed_params&.dig(:message_echoes)
  end

  private

  def process_messages # rubocop:disable Metrics/AbcSize,Metrics/MethodLength
    @lock_acquired = false

    # We don't support ephemeral message now, we need to skip processing the message
    # if the webhook event is an ephermal message or an unsupported message.
    # Reactions removed by the user arrive with an empty emoji and are skipped to match Baileys behavior.
    return if skip_message?

    # Multiple webhook event can be received against the same message due to misconfigurations in the Meta
    # business manager account. While we have not found the core reason yet, the following line ensure that
    # there are no duplicate messages created.
    return if find_message_by_source_id(messages_data.first[:id])

    @lock_acquired = acquire_message_processing_lock
    return unless @lock_acquired

    # Lock by contact phone to prevent race conditions when multiple messages
    # from the same contact arrive simultaneously (e.g., WhatsApp albums).
    with_contact_lock(contact_phone_for_lock) do
      # Re-check after acquiring lock to handle race conditions where an outgoing message
      # was sent from Chatwoot and the webhook arrived before source_id was saved
      next if find_message_by_source_id(messages_data.first[:id])

      # Reaction removals don't persist anything new, so peek for an existing
      # reaction row before set_contact: a removal webhook for a sender we
      # never stored has nothing to mark and shouldn't auto-create a contact
      # just to no-op. The match is sender-agnostic on purpose; the precise
      # filter happens inside `mark_existing_reaction_as_removed`.
      process_in_reply_to(messages_data.first)
      next if reaction_removal? && !existing_reaction_row?

      set_contact
      next if @contact.blank?

      # Reactions don't create a new Message row, so handle them outside the
      # transaction to avoid set_conversation opening/creating a stray thread
      # for a blank webhook. We also intentionally run this BEFORE
      # contact_processable? so blocked contacts can still reconcile an
      # existing reaction row.
      next mark_existing_reaction_as_removed if reaction_removal?

      next unless contact_processable?

      ActiveRecord::Base.transaction do
        set_conversation
        create_messages
      end
    end
  ensure
    # Clear lock AFTER transaction commits to prevent race conditions where another request
    # acquires the lock before this transaction is visible to other connections
    clear_message_source_id_from_redis if @lock_acquired
  end

  def skip_message?
    unprocessable_message_type?(message_type)
  end

  # For regular messages the contact phone is in :from; for echoes it's in :to.
  def contact_phone_for_lock
    outgoing_echo ? messages_data.first[:to] : messages_data.first[:from]
  end

  # Blocked contacts should not generate new incoming messages, but we still
  # accept echoes so outgoing messages tracked from native apps are preserved.
  def contact_processable?
    @contact.present? && !(@contact.blocked? && !outgoing_echo)
  end

  def process_statuses
    return unless find_message_by_source_id(@processed_params[:statuses].first[:id])

    update_message_with_status(@message, @processed_params[:statuses].first)
  rescue ArgumentError => e
    Rails.logger.error "Error while processing whatsapp status update #{e.message}"
  end

  def update_message_with_status(message, status)
    message.status = status[:status]
    if status[:status] == 'failed' && status[:errors].present?
      error = status[:errors]&.first
      message.external_error = "#{error[:code]}: #{error[:title]}"
    end
    message.save!
  end

  def create_messages
    message = messages_data.first
    log_error(message) && return if error_webhook_event?(message)

    message_type == 'contacts' ? create_contact_messages(message) : create_regular_message(message)
  end

  # Cloud delivers a reaction removal as a webhook with empty emoji. Our schema
  # keeps a single Message row per (target, sender) with `deleted` toggled on it,
  # so we update that row in place.
  #
  # Two paths converge here:
  # - Incoming: contact removed their reaction; mark the contact-owned row.
  # - Outgoing echo (multi-device, agent un-reacted from the connected phone):
  #   mark the senderless outgoing row. The Chatwoot-originated removal echo
  #   also lands here, but the active-only filter drops it (the controller
  #   already toggled the row to deleted) so it no-ops harmlessly.
  #
  # Lookup is intentionally NOT scoped to `@conversation`: the reaction may live
  # in an older/resolved thread, while `set_conversation` could have just picked
  # (or created) a different one for this webhook. Find the row globally, then
  # operate on its real `existing.conversation`.
  # Sender-agnostic existence check used to skip set_contact for removal
  # webhooks that have nothing to act on. Mirrors the inbox/in_reply_to scope
  # of `mark_existing_reaction_as_removed`.
  def existing_reaction_row?
    return false if @in_reply_to_external_id.blank?

    json_path = "(content_attributes#>>'{}')::jsonb"
    Message.where(inbox_id: inbox.id)
           .where("#{json_path}->>'is_reaction' = 'true'")
           .exists?(["#{json_path}->>'in_reply_to_external_id' = ?", @in_reply_to_external_id])
  end

  def mark_existing_reaction_as_removed # rubocop:disable Metrics/MethodLength,Metrics/AbcSize
    return if @in_reply_to_external_id.blank?

    json_path = "(content_attributes#>>'{}')::jsonb"
    # Scope by inbox so a colliding WhatsApp id from another inbox can't match
    # here and hand us back the wrong row.
    base = Message.where(inbox_id: inbox.id)
                  .where("#{json_path}->>'is_reaction' = 'true'")
                  .where("#{json_path}->>'in_reply_to_external_id' = ?", @in_reply_to_external_id)
    matches = if outgoing_echo
                # Multi-device: agent reacted via the connected phone, so the
                # local row has no agent (sender_id IS NULL) and is outgoing.
                base.where(sender_id: nil, sender_type: nil)
                    .where(message_type: Message.message_types[:outgoing])
              else
                base.where(sender: @contact)
              end
    # Active-only: when the only matches are already deleted, return nil so
    # the caller no-ops instead of re-deleting and bumping the conversation
    # for an echoed Chatwoot-originated removal.
    existing = matches.where.not(content: '')
                      .where("COALESCE(#{json_path}->>'deleted', 'false') != 'true'")
                      .reorder(created_at: :desc)
                      .first
    return if existing.nil?

    new_attrs = existing.content_attributes.merge('deleted' => true)
    existing.update!(content: '', content_attributes: new_attrs)
    target_conversation = existing.conversation
    # Refresh the chat list snapshot; cable MESSAGE_UPDATED only touches
    # chat.messages on the client, so the conversation card preview stays stale
    # without an explicit conversation.updated dispatch. Touch updated_at so
    # the frontend out-of-order guard can drop stale cables.
    target_conversation.update_columns(updated_at: Time.current) # rubocop:disable Rails/SkipsModelValidations
    target_conversation.dispatch_conversation_updated_event
  end

  def create_contact_messages(message)
    message['contacts'].each do |contact|
      # Pass source_id from parent message since contact objects don't have :id
      create_message(contact, source_id: message[:id])
      attach_contact(contact)
      @message.save!
    end
  end

  def create_regular_message(message)
    create_message(message, source_id: message[:id])
    attach_files
    attach_location if message_type == 'location'
    @message.save!
  end

  def set_contact
    if outgoing_echo
      set_contact_from_echo
    else
      set_contact_from_message
    end
  end

  def set_contact_from_echo
    # For echo messages, contact phone is in the 'to' field
    phone_number = messages_data.first[:to]
    waid = processed_waid(phone_number)

    contact_inbox = ::ContactInboxWithContactBuilder.new(
      source_id: waid,
      inbox: inbox,
      contact_attributes: { name: "+#{phone_number}", phone_number: "+#{phone_number}" }
    ).perform

    @contact_inbox = contact_inbox
    @contact = contact_inbox.contact
  end

  def set_contact_from_message
    contact_params = @processed_params[:contacts]&.first
    return if contact_params.blank?

    waid = processed_waid(contact_params[:wa_id])

    contact_inbox = ::ContactInboxWithContactBuilder.new(
      source_id: waid,
      inbox: inbox,
      contact_attributes: { name: contact_params.dig(:profile, :name), phone_number: "+#{messages_data.first[:from]}" }
    ).perform

    @contact_inbox = contact_inbox
    @contact = contact_inbox.contact

    # Update existing contact name if ProfileName is available and current name is just phone number
    update_contact_with_profile_name(contact_params)
  end

  def set_conversation
    # if lock to single conversation is disabled, we will create a new conversation if previous conversation is resolved
    @conversation = if @inbox.lock_to_single_conversation
                      @inbox.conversations.where(contact_id: @contact_inbox.contact_id).last
                    else
                      @contact_inbox.conversations
                                    .where.not(status: :resolved).last
                    end
    return if @conversation

    @conversation = ::Conversation.create!(conversation_params)
  end

  def attach_files
    return if %w[text button interactive location contacts reaction].include?(message_type)

    attachment_payload = messages_data.first[message_type.to_sym]
    @message.content ||= attachment_payload[:caption]

    attachment_file = download_attachment_file(attachment_payload)
    return if attachment_file.blank?

    @message.attachments.new(
      account_id: @message.account_id,
      file_type: file_content_type(message_type),
      file: {
        io: attachment_file,
        filename: attachment_payload[:filename].presence || attachment_file.original_filename,
        content_type: attachment_file.content_type
      },
      meta: ({ is_recorded_audio: true } if attachment_payload[:voice])
    )
  end

  def attach_location
    location = messages_data.first['location']
    location_name = location['name'] ? "#{location['name']}, #{location['address']}" : ''
    @message.attachments.new(
      account_id: @message.account_id,
      file_type: file_content_type(message_type),
      coordinates_lat: location['latitude'],
      coordinates_long: location['longitude'],
      fallback_title: location_name,
      external_url: location['url']
    )
  end

  def create_message(message, source_id: nil)
    @message = @conversation.messages.build(
      content: message_content(message),
      account_id: @inbox.account_id,
      inbox_id: @inbox.id,
      message_type: outgoing_echo ? :outgoing : :incoming,
      # Set status to :delivered for echo messages to prevent SendReplyJob from trying to send them
      status: outgoing_echo ? :delivered : :sent,
      sender: outgoing_echo ? nil : @contact,
      source_id: (source_id || message[:id]).to_s,
      content_attributes: build_content_attributes(message)
    )
  end

  def build_content_attributes(message)
    content_attrs = outgoing_echo ? { external_echo: true } : {}
    content_attrs[:in_reply_to_external_id] = @in_reply_to_external_id if @in_reply_to_external_id.present?
    content_attrs[:external_created_at] = message[:timestamp].to_i
    content_attrs[:is_reaction] = true if message_type == 'reaction'
    content_attrs
  end

  def attach_contact(contact)
    phones = contact[:phones]
    phones = [{ phone: 'Phone number is not available' }] if phones.blank?

    name_info = contact['name'] || {}
    contact_meta = {
      firstName: name_info['first_name'],
      lastName: name_info['last_name']
    }.compact

    phones.each do |phone|
      @message.attachments.new(
        account_id: @message.account_id,
        file_type: file_content_type(message_type),
        fallback_title: phone[:phone].to_s,
        meta: contact_meta
      )
    end
  end

  def update_contact_with_profile_name(contact_params)
    profile_name = contact_params.dig(:profile, :name)
    return if profile_name.blank?
    return if @contact.name == profile_name

    # Only update if current name exactly matches the phone number or formatted phone number
    return unless contact_name_matches_phone_number?

    @contact.update!(name: profile_name)
  end

  def contact_name_matches_phone_number?
    phone_number = "+#{messages_data.first[:from]}"
    formatted_phone_number = TelephoneNumber.parse(phone_number).international_number
    @contact.name == phone_number || @contact.name == formatted_phone_number
  end
end