Dalan/app/services/event_mailer.rb

# frozen_string_literal: true

require 'bunny'
require 'ostruct'

class EventMailer
  Error = Class.new(StandardError)

  class VerificationError < Error; end

  def initialize(events, exchanges, keychain)
    @exchanges = exchanges
    @keychain = keychain
    @events = events

    Kernel.at_exit { unlisten }
  end

  def call
    listen
  end

  private

  def listen
    unlisten

    @bunny_session = Bunny::Session.new(rabbitmq_credentials).tap do |session|
      session.start
      Kernel.at_exit { session.stop }
    end

    @bunny_channel = @bunny_session.channel
    # Delete old queue if some exists
    @bunny_channel.queue_delete('barong.postmaster.event.mailer') if @bunny_session.queue_exists?('barong.postmaster.event.mailer')

    # Define fanout exchanges which will broadcast
    # all the messages they receives to all the queues they know
    retry_exchange = @bunny_channel.fanout('barong.event.mailer.retry.exchange')
    main_exchange = @bunny_channel.fanout('barong.event.mailer.main.exchange')

    queue = @bunny_channel.queue('barong.event.mailer.main', auto_delete: false, durable: true,
      arguments: {
        :'x-dead-letter-exchange' => retry_exchange.name,
      }
    )
    queue.bind(main_exchange)

    retry_queue = @bunny_channel.queue('barong.event.mailer.retry', auto_delete: false, durable: true,
      arguments: {
        :'x-dead-letter-exchange' => main_exchange.name,
        :'x-message-ttl' => 120000   # will trigger retry every 2 minutes
    })
    retry_queue.bind(retry_exchange)

    @events.each do |event|
      exchange_name = @exchanges[event[:exchange].to_sym][:name]
      exchange = @bunny_channel.direct(exchange_name)

      queue.bind(exchange, routing_key: event[:key])
    end

    Rails.logger.info { 'Listening for events.' }
    queue.subscribe(manual_ack: true, block: true, &method(:handle_message))
  end

  def unlisten
    if @bunny_session || @bunny_channel
      Rails.logger.info { 'No longer listening for events.' }
    end

    @bunny_channel&.work_pool&.kill
    @bunny_session&.stop
  ensure
    @bunny_channel = nil
    @bunny_session = nil
  end

  def algorithm_verification_options(signer)
    { algorithms: @keychain[signer][:algorithm] }
  end

  def jwt_public_key(signer)
    OpenSSL::PKey.read(Base64.urlsafe_decode64(@keychain[signer][:value]))
  end

  def rabbitmq_credentials
    if Barong::App.config.event_api_rabbitmq_url.present?
      Barong::App.config.event_api_rabbitmq_url
    else
      {
        host: Barong::App.config.event_api_rabbitmq_host,
        port: Barong::App.config.event_api_rabbitmq_port,
        username: Barong::App.config.event_api_rabbitmq_username,
        password: Barong::App.config.event_api_rabbitmq_password
      }
    end
  end

  def handle_message(delivery_info, _metadata, payload)
    Rails.logger.info { "Start handling a message" }
    Rails.logger.info { "\nPayload: \n #{payload} \n\n Metadata: \n #{_metadata} \n\n Delivery info: \n #{delivery_info} \n" }
    exchange = @exchanges.select { |_, ex| ex[:name] == delivery_info[:exchange] }

    # In case of retry message
    # we should get exchange name from _metadata info
    if exchange.empty?
      exchange_name = _metadata[:headers]['x-death'][1]['exchange']
      exchange = @exchanges.select { |_, ex| ex[:name] == exchange_name }
    end

    exchange_id = exchange.keys.first.to_s
    signer      = exchange[exchange_id.to_sym][:signer]

    result = verify_jwt(payload, signer.to_sym)

    raise VerificationError, "Failed to verify signature from #{signer}." \
      unless result[:verified].include?(signer.to_sym)

    config = @events.select do |event|
      event[:key] == delivery_info[:routing_key] &&
        event[:exchange] == exchange_id
    end.first

    event = result[:payload].fetch(:event)
    obj   = JSON.parse(event.to_json, object_class: OpenStruct)

    user  = User.includes(:profiles).find_by(uid: obj.record.user.uid)
    language = user.language.downcase.to_sym
    Rails.logger.info { "User #{user.email} has '#{language}' email language" }
    template_config = config[:templates].transform_keys(&:downcase)

    unless template_config.keys.include?(language)
      Rails.logger.error { "Language #{language} is not supported. Skipping." }
      return
    end

    if config[:expression].present? && skip_event(event, config[:expression])
      Rails.logger.info { "Event #{obj.name} skipped" }
      return
    end

    params = {
      logo: Barong::App.config.smtp_logo_link,
      subject: template_config[language][:subject],
      template_name: template_config[language][:template_path],
      record: obj.record,
      changes: obj.changes,
      user: user
    }

    Postmaster.process_payload(params).deliver_now

    # Acknowledges a message
    # Acknowledged message is completely removed from the queue
    @bunny_channel.ack(delivery_info.delivery_tag)
  rescue StandardError => e
    Rails.logger.error { e.inspect }

    if e.is_a?(JWT::ExpiredSignature) || e.is_a?(JWT::VerificationError) || e.is_a?(VerificationError)
      # Acknowledges a message
      @bunny_channel.ack(delivery_info.delivery_tag)
    else
      # Rejects a message
      # A rejected message dropped by RabbitMQ and goes to dead letter exchange queue
      @bunny_channel.reject(delivery_info.delivery_tag)
    end

    unlisten if db_connection_error?(e)
  end

  def verify_jwt(payload, signer)
    options = algorithm_verification_options(signer)
    JWT::Multisig.verify_jwt JSON.parse(payload), { signer => jwt_public_key(signer) },
                             options.compact
  end

  def skip_event(event, expression)
    # valid operators: and / or / not
    operator = expression.keys.first.downcase
    # { field_name: field_value }
    values = expression[operator]

    # return array of boolean [false, true]
    res = values.keys.map do |field_name|
      safe_dig(event, field_name.to_s.split('.')) == values[field_name]
    end

    # all? works as AND operator, any? works as OR operator
    return false if (operator == :and && res.all?) || (operator == :or && res.any?) ||
                    (operator == :not && !res.all?)

    return true if operator == :not && res.all?

    true
  end

  def db_connection_error?(exception)
    exception.is_a?(Mysql2::Error::ConnectionError) || exception.cause.is_a?(Mysql2::Error)
  end

  def safe_dig(hash, keypath, default = nil)
    stringified_hash = JSON.parse(hash.to_json)
    stringified_keypath = keypath.map(&:to_s)

    stringified_keypath.reduce(stringified_hash) do |accessible, key|
      return default unless accessible.is_a? Hash
      return default unless accessible.key? key

      accessible[key]
    end
  end

  class << self
    def call(*args)
      new(*args).call
    end
  end
end