chatwoot/app/jobs/hook_job.rb

class HookJob < MutexApplicationJob
  retry_on LockAcquisitionError, wait: 3.seconds, attempts: 3

  queue_as :medium

  INTEGRATION_PROCESSORS = {
    'slack' => :process_slack_integration,
    'dialogflow' => :process_dialogflow_integration,
    'google_translate' => :google_translate_integration,
    'leadsquared' => :process_leadsquared_integration_with_lock,
    'linear' => :process_linear_integration
  }.freeze

  def perform(hook, event_name, event_data = {})
    return if hook.disabled?

    processor = INTEGRATION_PROCESSORS[hook.app_id]
    send(processor, hook, event_name, event_data) if processor
  rescue StandardError => e
    Rails.logger.error e
  end

  private

  def process_slack_integration(hook, event_name, event_data)
    message = event_data[:message]

    case event_name
    when 'message.created'
      if message.attachments.blank?
        ::SendOnSlackJob.perform_later(message, hook)
      else
        ::SendOnSlackJob.set(wait: 2.seconds).perform_later(message, hook)
      end
    when 'message.updated'
      # Only interactive bot messages store responses via content_attributes (submitted_values / submitted_email).
      # Skip other content types to avoid unnecessary job enqueues on every message update.
      return unless message.content_type.in?(Integrations::Slack::UpdateSlackMessageService::SUPPORTED_CONTENT_TYPES)
      # Guard against redundant Slack updates when unrelated attributes change (e.g. status)
      # while submitted_values is already present on the message.
      return unless event_data[:previous_changes]&.key?('content_attributes')

      ::UpdateSlackMessageJob.perform_later(message, hook)
    end
  end

  def process_dialogflow_integration(hook, event_name, event_data)
    return unless ['message.created', 'message.updated'].include?(event_name)

    Integrations::Dialogflow::ProcessorService.new(event_name: event_name, hook: hook, event_data: event_data).perform
  end

  def google_translate_integration(hook, event_name, event_data)
    return unless ['message.created'].include?(event_name)

    message = event_data[:message]
    Integrations::GoogleTranslate::DetectLanguageService.new(hook: hook, message: message).perform
  end

  def process_linear_integration(hook, event_name, event_data)
    return unless event_name == 'message.created'

    message = event_data[:message]
    Integrations::Linear::AutoLinkService.new(account: hook.account, message: message).perform
  end

  def process_leadsquared_integration_with_lock(hook, event_name, event_data)
    # Why do we need a mutex here? glad you asked
    # When a new conversation is created. We get a contact created event, immediately followed by
    # a contact updated event, and then a conversation created event.
    # This all happens within milliseconds of each other.
    # Now each of these subsequent event handlers need to have a leadsquared lead created and the contact to have the ID.
    # If the lead data is not present, we try to search the API and create a new lead if it doesn't exist.
    # This gives us a bad race condition that allows the API to create multiple leads for the same contact.
    #
    # This would have not been a problem if the email and phone number were unique identifiers for contacts at LeadSquared
    # But then this is configurable in the LeadSquared settings, and may or may not be unique.
    valid_event_names = ['contact.updated', 'conversation.created', 'conversation.resolved']
    return unless valid_event_names.include?(event_name)
    return unless hook.feature_allowed?

    key = format(::Redis::Alfred::CRM_PROCESS_MUTEX, hook_id: hook.id)
    with_lock(key) do
      process_leadsquared_integration(hook, event_name, event_data)
    end
  end

  def process_leadsquared_integration(hook, event_name, event_data)
    # Process the event with the processor service
    processor = Crm::Leadsquared::ProcessorService.new(hook)

    case event_name
    when 'contact.updated'
      processor.handle_contact(event_data[:contact])
    when 'conversation.created'
      processor.handle_conversation_created(event_data[:conversation])
    when 'conversation.resolved'
      processor.handle_conversation_resolved(event_data[:conversation])
    end
  end
end