Initial commit

app/services/event_mailer.rb

@@ -0,0 +1,219 @@
+# 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