Initial commit

app/workers/amqp/base.rb

@@ -0,0 +1,12 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class Base
+      def is_db_connection_error?(exception)
+        exception.is_a?(Mysql2::Error::ConnectionError) || exception.cause.is_a?(Mysql2::Error)
+      end
+    end
+  end
+end

app/workers/amqp/deposit_coin_address.rb

@@ -0,0 +1,58 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class DepositCoinAddress < Base
+      def process(payload)
+        payload.symbolize_keys!
+
+        member = Member.find_by_id(payload[:member_id])
+        unless member
+          Rails.logger.warn do
+            'Unable to generate deposit address.'\
+            "Member with id: #{payload[:member_id]} doesn't exist"
+          end
+          return
+        end
+
+        wallet = Wallet.find_by_id(payload[:wallet_id])
+
+        unless wallet
+          Rails.logger.warn do
+            'Unable to generate deposit address.'\
+            "Deposit Wallet with id: #{payload[:wallet_id]} doesn't exist"
+          end
+          return
+        end
+
+        wallet_service = WalletService.new(wallet)
+
+        member.payment_address(wallet.id).tap do |pa|
+          pa.with_lock do
+            next if pa.address.present?
+
+            # Supply address ID in case of BitGo address generation if it exists.
+            result = wallet_service.create_address!(member.uid, pa.details.merge(updated_at: pa.updated_at))
+
+            if result.present?
+              pa.update!(address: result[:address],
+                        secret:  result[:secret],
+                        details: result.fetch(:details, {}).merge(pa.details))
+            end
+          end
+
+          pa.trigger_address_event unless pa.address.blank?
+        end
+
+      # Don't re-enqueue this job in case of error.
+      # The system is designed in such way that when user will
+      # request list of accounts system will ask to generate address again (if it is not generated of course).
+      rescue StandardError => e
+        raise e if is_db_connection_error?(e)
+
+        report_exception(e)
+      end
+    end
+  end
+end

app/workers/amqp/influx_writer.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class InfluxWriter < Base
+      def process(payload, metadata, _delivery_info)
+        case metadata[:headers]['type']
+        when 'local'
+          trade = Trade.new payload
+          trade.write_to_influx
+        when 'upstream'
+          trade = Trade.new payload.merge(total: payload['price'].to_d * payload['amount'].to_d)
+          trade.write_to_influx
+        end
+      end
+    end
+  end
+end

app/workers/amqp/matching.rb

@@ -0,0 +1,160 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class Matching < Base
+
+      class DryrunError < StandardError
+        attr :engine
+
+        def initialize(engine)
+          @engine = engine
+        end
+      end
+
+      def initialize(options={})
+        @options = options
+        reload 'all'
+      end
+
+      def process(payload, metadata, delivery_info)
+        payload.symbolize_keys!
+
+        case payload[:action]
+        when 'submit'
+          submit build_order(payload[:order])
+        when 'cancel'
+          cancel build_order(payload[:order])
+        when 'reload'
+          reload payload[:market]
+        when 'new'
+          initialize_engine Market.find(payload[:market])
+        else
+          Rails.logger.fatal { "Unknown action: #{payload[:action]}" }
+        end
+      end
+
+      def submit(order)
+        engines[order.market].submit(order)
+      end
+
+      def cancel(order)
+        engines[order.market].cancel(order)
+      end
+
+      def reload(market)
+        if market == 'all'
+          # NOTE: Run matching engine for disabled markets.
+          Market.find_each(&method(:initialize_engine))
+          Rails.logger.info { "All engines reloaded." }
+        else
+          initialize_engine Market.find(market)
+          Rails.logger.info { "#{market} engine reloaded." }
+        end
+      rescue DryrunError => e
+        # stop started engines
+        engines.each {|id, engine| engine.shift_gears(:dryrun) unless engine == e.engine }
+
+        Rails.logger.fatal { "#{market} engine failed to start. Matched during dryrun:" }
+        e.engine.queue.each do |trade|
+          Rails.logger.info { trade[1].inspect }
+        end
+      end
+
+      def build_order(attrs)
+        ::Matching::OrderBookManager.build_order attrs
+      end
+
+      def initialize_engine(market)
+        engine = create_engine(market)
+        load_orders(market)
+        engine.initializing = false
+        engine.publish_snapshot
+        start_engine(market)
+      end
+
+      def create_engine(market)
+        engines[market.id] = ::Matching::Engine.new(market, @options)
+      end
+
+      def load_orders(market)
+        ::Order.active.with_market(market.id).order('id asc').each do |order|
+          submit build_order(order.to_matching_attributes)
+        end
+      end
+
+      def start_engine(market)
+        engine = engines[market.id]
+        if engine.mode == :dryrun
+          if engine.queue.empty?
+            engine.shift_gears :run
+          else
+            accept = ENV['ACCEPT_MINUTES'] ? ENV['ACCEPT_MINUTES'].to_i : 30
+            order_ids = engine.queue
+              .map {|args| [args[1][:trade][:maker_order_id], args[1][:trade][:taker_order_id]] }
+              .flatten.uniq
+
+            orders = Order.where('created_at < ?', accept.minutes.ago).where(id: order_ids)
+            if orders.exists?
+              # there're very old orders matched, need human intervention
+              raise DryrunError, engine
+            else
+              # only buffered orders matched, just publish trades and continue
+              engine.queue.each {|args| ::AMQP::Queue.enqueue(*args) }
+              engine.shift_gears :run
+            end
+          end
+        else
+          Rails.logger.info { "#{market.id} engine already started. mode=#{engine.mode}" }
+        end
+      end
+
+      def engines
+        @engines ||= {}
+      end
+
+      # dump limit orderbook
+      def on_usr1
+        engines.each do |id, eng|
+          dump_file = File.join('/', 'tmp', "limit_orderbook_#{id}")
+          limit_orders = eng.limit_orders
+
+          File.open(dump_file, 'w') do |f|
+            f.puts "ASK"
+            limit_orders[:ask].keys.reverse.each do |k|
+              f.puts k.to_s('F')
+              limit_orders[:ask][k].each {|o| f.puts "\t#{o.label}" }
+            end
+            f.puts "-"*40
+            limit_orders[:bid].keys.reverse.each do |k|
+              f.puts k.to_s('F')
+              limit_orders[:bid][k].each {|o| f.puts "\t#{o.label}" }
+            end
+            f.puts "BID"
+          end
+
+          puts "#{id} limit orderbook dumped to #{dump_file}."
+        end
+      end
+
+      # dump market orderbook
+      def on_usr2
+        engines.each do |id, eng|
+          dump_file = File.join('/', 'tmp', "market_orderbook_#{id}")
+          market_orders = eng.market_orders
+
+          File.open(dump_file, 'w') do |f|
+            f.puts "ASK"
+            market_orders[:ask].each {|o| f.puts "\t#{o.label}" }
+            f.puts "-"*40
+            market_orders[:bid].each {|o| f.puts "\t#{o.label}" }
+            f.puts "BID"
+          end
+
+          puts "#{id} market orderbook dumped to #{dump_file}."
+        end
+      end
+    end
+  end
+end

app/workers/amqp/order_processor.rb

@@ -0,0 +1,33 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class OrderProcessor < Base
+      def initialize
+        Order.where(state: ::Order::PENDING).find_each do |order|
+          Order.submit(order.id)
+        rescue StandardError => e
+          ::AMQP::Queue.enqueue(:trade_error, e.message)
+          report_exception_to_screen(e)
+
+          raise e if is_db_connection_error?(e)
+        end
+      end
+
+      def process(payload)
+        case payload['action']
+        when 'submit'
+          Order.submit(payload.dig('order', 'id'))
+        when 'cancel'
+          Order.cancel(payload.dig('order', 'id'))
+        end
+      rescue StandardError => e
+        ::AMQP::Queue.enqueue(:trade_error, e.message)
+        report_exception_to_screen(e)
+
+        raise e if is_db_connection_error?(e)
+      end
+    end
+  end
+end

app/workers/amqp/trade_executor.rb

@@ -0,0 +1,12 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class TradeExecutor < Base
+      def process(payload)
+        ::Matching::Executor.new(payload.symbolize_keys).process
+      end
+    end
+  end
+end

app/workers/amqp/withdraw_coin.rb

@@ -0,0 +1,95 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module AMQP
+    class WithdrawCoin < Base
+      def initialize
+        @logger = TaggedLogger.new(Rails.logger, worker: __FILE__)
+      end
+
+      def process(payload)
+        payload.symbolize_keys!
+
+        @logger.warn id: payload[:id], message: 'Received request for processing withdraw.'
+
+        withdraw = Withdraw.find_by_id(payload[:id])
+        if withdraw.blank?
+          @logger.warn id: payload[:id], message: 'The withdraw with such ID doesn\'t exist in database.'
+          return
+        end
+
+        withdraw.with_lock do
+          unless withdraw.processing?
+            @logger.warn id: withdraw.id,
+                         message: 'The withdraw is being processed by another worker or has already been processed.'
+            return
+          end
+
+          if withdraw.rid.blank?
+            @logger.warn id: withdraw.id,
+                         message: 'The destination address doesn\'t exist.'
+            withdraw.fail!
+            return
+          end
+
+          @logger.warn id: withdraw.id,
+                       amount: withdraw.amount.to_s('F'),
+                       fee: withdraw.fee.to_s('F'),
+                       currency: withdraw.currency.code.upcase,
+                       rid: withdraw.rid,
+                       message: 'Sending witdraw.'
+
+          wallet = Wallet.active.joins(:currencies)
+                         .find_by(currencies: { id: withdraw.currency_id }, kind: :hot)
+
+          unless wallet
+            @logger.warn id: withdraw.id,
+                         currency: withdraw.currency.code.upcase,
+                         message: 'Can\'t find active hot wallet for currency.'
+            withdraw.skip!
+            return
+          end
+
+          balance = wallet.current_balance(withdraw.currency)
+          if balance == 'N/A' || balance < withdraw.amount
+            @logger.warn id: withdraw.id,
+                         balance: balance.to_s,
+                         amount: withdraw.amount.to_s,
+                         message: 'The withdraw skipped because wallet balance is not sufficient or amount greater than wallet max_balance.'
+            return withdraw.skip!
+          end
+
+          @logger.warn id: withdraw.id,
+                       message: 'Sending request to Wallet Service.'
+
+          wallet_service = WalletService.new(wallet)
+          transaction = wallet_service.build_withdrawal!(withdraw)
+
+          @logger.warn id: withdraw.id,
+                       tid: transaction.hash,
+                       message: 'The currency API accepted withdraw and assigned transaction ID.'
+
+          @logger.warn id: withdraw.id,
+                       message: 'Updating withdraw state in database.'
+
+          withdraw.txid = transaction.hash
+          withdraw.dispatch
+          withdraw.save!
+
+          @logger.warn id: withdraw.id, message: 'Withdrawal has processed'
+
+        rescue StandardError => e
+          @logger.warn id: withdraw.id, message: 'Failed to process withdrawal. See exception details below.'
+          report_exception(e)
+          withdraw.err! e
+
+          raise e if is_db_connection_error?(e)
+
+          @logger.warn id: withdraw.id,
+                       message: 'Setting withdrawal state to errored.'
+        end
+      end
+    end
+  end
+end

app/workers/daemons/base.rb

@@ -0,0 +1,66 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module Daemons
+    class Base
+      class GetLockError < StandardError; end
+      class << self; attr_accessor :sleep_time end
+
+      attr_accessor :running
+      attr_reader :logger
+
+      def initialize
+        @running = true
+        @logger = Rails.logger
+      end
+
+      def stop
+        @running = false
+      end
+
+      def run
+        while running
+          begin
+            process
+          rescue ScriptError => e
+            raise e if is_db_connection_error?(e)
+
+            report_exception(e)
+          end
+          wait
+        end
+      end
+
+      def process
+        method_not_implemented
+      end
+
+      def wait
+        Kernel.sleep self.class.sleep_time
+      end
+
+      def is_db_connection_error?(exception)
+        exception.is_a?(Mysql2::Error::ConnectionError) || exception.cause.is_a?(Mysql2::Error)
+      end
+
+      def lock(klass, timeout)
+        res = ActiveRecord::Base.connection.exec_query("SELECT GET_LOCK('Peatio_#{klass}',#{timeout})")
+
+        # response from this query will look like this [{"GET_LOCK(id,10)"=>1}]
+        # returns 1 if the lock was obtained successfully, 0 if the attempt timed out
+        if res.to_a[0].values[0] == 1
+          begin
+            yield
+          rescue StandardError => e
+            report_exception(e)
+          ensure
+            ActiveRecord::Base.connection.exec_query("SELECT RELEASE_LOCK('Peatio_#{klass}')")
+          end
+        else
+          raise GetLockError, "Peatio_#{klass} is already running"
+        end
+      end
+    end
+  end
+end

app/workers/daemons/blockchain.rb

@@ -0,0 +1,114 @@
+# encoding: UTF-8
+# frozen_string_literal: true
+
+module Workers
+  module Daemons
+    class Blockchain < Base
+      class Runner
+        attr_reader :ts, :thread
+
+        def initialize(blockchain, ts)
+          @blockchain = blockchain
+          @ts = ts
+          @thread = nil
+        end
+
+        def start
+          @thread ||= Thread.new do
+            bc_service = BlockchainService.new(@blockchain)
+
+            Rails.logger.info { "Processing #{@blockchain.name} blocks." }
+
+            loop do
+              begin
+                # Reset blockchain_service state.
+                bc_service.reset!
+
+                if @blockchain.reload.height + @blockchain.min_confirmations >= bc_service.latest_block_number
+                  Rails.logger.info { "Skip synchronization. No new blocks detected, height: #{@blockchain.height}, latest_block: #{bc_service.latest_block_number}." }
+                  Rails.logger.info { "Sleeping for 10 seconds" }
+                  sleep(10)
+                  next
+                end
+
+                from_block = @blockchain.height || 0
+
+                (from_block..bc_service.latest_block_number).each do |block_id|
+                  Rails.logger.info { "Started processing #{@blockchain.key} block number #{block_id}." }
+                  block_json = bc_service.process_block(block_id)
+                  Rails.logger.info { "Fetch #{block_json.transactions.count} transactions in block number #{block_id}." }
+                  bc_service.update_height(block_id)
+                  Rails.logger.info { "Finished processing #{@blockchain.key} block number #{block_id}." }
+                end
+              rescue StandardError => e
+                report_exception(e)
+                Rails.logger.warn { "Error: #{e}. Sleeping for 10 seconds" }
+                sleep(10)
+              end
+            end
+          end
+        end
+
+        def stop
+          @thread&.kill
+        end
+      end
+
+      def run
+        @runner_pool = ::Blockchain.active.each_with_object({}) do |b, pool|
+          max_ts = [b.currencies.maximum(:updated_at), b.updated_at].compact.max.to_i
+
+          logger.warn { "Creating the runner for #{b.key}" }
+          pool[b.key] = Runner.new(b, max_ts).tap(&:start)
+        end
+
+        while running
+          begin
+            # Stop disabled blockchains runners first.
+            (@runner_pool.keys - ::Blockchain.active.pluck(:key)).each do |b_key|
+              logger.warn { "Stopping the runner for #{b_key} (blockchain is not active anymore)" }
+              @runner_pool.delete(b_key).stop
+            end
+
+            # Recreate active blockchain runners by comparing runner &
+            # maximum blockchain & currencies updated_at timestamp.
+            ::Blockchain.active.each do |b|
+              max_ts = [b.currencies.maximum(:updated_at), b.updated_at].compact.max.to_i
+
+              if @runner_pool[b.key].blank?
+                logger.warn { "Starting the new runner for #{b.key} (no runner found in pool)" }
+                @runner_pool[b.key] = Runner.new(b, max_ts).tap(&:start)
+              elsif @runner_pool[b.key].ts < max_ts
+                logger.warn { "Recreating a runner for #{b.key} (#{Time.at(@runner_pool[b.key].ts)} < #{Time.at(max_ts)})" }
+                @runner_pool.delete(b.key).stop
+                @runner_pool[b.key] = Runner.new(b, max_ts).tap(&:start)
+              else
+                logger.warn { "The runner for #{b.key} is up to date (#{Time.at(@runner_pool[b.key].ts)} >= #{Time.at(max_ts)})" }
+              end
+            end
+
+            logger.info { "Current runners timestamps:" }
+            logger.info do
+              @runner_pool.transform_values(&:ts)
+            end
+
+            # Check for blockchain config changes in 30 seconds.
+            sleep 30
+
+          rescue StandardError => e
+            raise e if is_db_connection_error?(e)
+
+            report_exception(e)
+            Rails.logger.warn { "Error: #{e}. Sleeping for 10 seconds" }
+            sleep(10)
+          end
+        end
+      end
+
+      def stop
+        @running = false
+        @runner_pool.each { |_bc_key, runner| runner.stop }
+      end
+    end
+  end
+end

app/workers/daemons/cron_job.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+# K-line point is represented as array of 5 numbers:
+# [timestamp, open_price, max_price, min_price, last_price, period_volume]
+
+module Workers
+  module Daemons
+    class CronJob < Base
+      JOBS = [Jobs::Cron::KLine, Jobs::Cron::Ticker, Jobs::Cron::StatsMemberPnl, Jobs::Cron::AML, Jobs::Cron::Refund, Jobs::Cron::WalletBalances, Jobs::Cron::ReferralBonus].freeze
+
+      def run
+        JOBS.map { |j| Thread.new { process(j) } }.map(&:join)
+      end
+
+      def process(service)
+        service.process while running
+      end
+    end
+  end
+end

app/workers/daemons/deposit.rb

@@ -0,0 +1,94 @@
+# frozen_string_literal: true
+
+module Workers
+  module Daemons
+    class Deposit < Base
+      self.sleep_time = 60
+
+      def process
+        # Process deposits with `processing` state each minute
+        ::Deposit.processing.each do |deposit|
+          Rails.logger.info { "Starting processing coin deposit with id: #{deposit.id}." }
+
+          wallet = Wallet.deposit_wallet(deposit.currency_id)
+          unless wallet
+            Rails.logger.warn { "Can't find active deposit wallet for currency with code: #{deposit.currency_id}."}
+            next
+          end
+          service = WalletService.new(wallet)
+          # Check if adapter has prepare_deposit_collection! implementation
+          if service.adapter.class.instance_methods(false).include?(:prepare_deposit_collection!)
+            begin
+              # Process fee collection for tokens
+              collect_fee(deposit)
+              # Will be processed after fee collection
+              next if deposit.fee_processing?
+            rescue StandardError => e
+              Rails.logger.error { "Failed to collect deposit fee #{deposit.id}. See exception details below." }
+              report_exception(e)
+              raise e if is_db_connection_error?(e)
+
+              next
+            end
+          end
+
+          process_deposit(deposit)
+        end
+
+        # Process deposits with `fee_processing` state that already collected fees for collection
+        ::Deposit.fee_processing.where('updated_at < ?', 5.minute.ago).each do |deposit|
+          Rails.logger.info { "Starting processing token deposit with id: #{deposit.id}." }
+
+          process_deposit(deposit)
+        end
+      end
+
+      def process_deposit(deposit)
+        if deposit.spread.blank?
+          deposit.spread_between_wallets!
+          Rails.logger.warn { "The deposit was spreaded in the next way: #{deposit.spread}"}
+        end
+
+        wallet = Wallet.deposit_wallet(deposit.currency_id)
+        service = WalletService.new(wallet)
+
+        transactions = service.collect_deposit!(deposit, deposit.spread_to_transactions)
+
+        if transactions.present?
+          # Save txids in deposit spread.
+          deposit.update!(spread: transactions.map(&:as_json))
+
+          Rails.logger.warn { "The API accepted deposit collection and assigned transaction ID: #{transactions.map(&:as_json)}." }
+
+          deposit.dispatch!
+        else
+          deposit.skip!
+          "Skipped deposit with txid: #{deposit.txid} with amount: #{deposit.amount}"\
+          " to #{deposit.address}"
+        end
+      rescue StandardError => e
+        Rails.logger.error { "Failed to collect deposit #{deposit.id}. See exception details below." }
+        report_exception(e)
+
+        raise e if is_db_connection_error?(e)
+      end
+
+      def collect_fee(deposit)
+        if deposit.spread.blank?
+          deposit.spread_between_wallets!
+          Rails.logger.warn { "The deposit was spreaded in the next way: #{deposit.spread}"}
+        end
+
+        fee_wallet = Wallet.active.fee.find_by(blockchain_key: deposit.currency.blockchain_key)
+        unless fee_wallet
+          Rails.logger.warn { "Can't find active fee wallet for currency with code: #{deposit.currency_id}." }
+          return
+        end
+
+        transactions = WalletService.new(fee_wallet).deposit_collection_fees!(deposit, deposit.spread_to_transactions)
+        deposit.fee_process! if transactions.present?
+        Rails.logger.warn { "The API accepted token deposit collection fee and assigned transaction ID: #{transactions.map(&:as_json)}." }
+      end
+    end
+  end
+end

app/workers/daemons/upstream.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module Workers
+  module Daemons
+    class Upstream < Base
+      def run
+        Engine.all.map { |e| Thread.new { process(e) } }.map(&:join)
+      end
+
+      def process(engine)
+        EM.synchrony do
+          upstream = Peatio::Upstream.registry[engine.driver]
+          engine.markets.each do |market|
+            target = if market.data.present? && market.data['target'].present?
+                       market.data['target']
+                     else
+                       market.id
+                     end
+
+            configs = engine.data.merge('source' => market.id, 'amqp' => ::AMQP::Queue, 'target' => target)
+
+            upstream.new(configs).ws_connect
+            Rails.logger.info "Upstream with driver #{engine.driver} for #{market.id} started"
+          rescue StandardError => e
+            report_exception(e)
+            next
+          end
+        rescue Peatio::AdapterRegistry::NotRegisteredAdapterError => e
+          report_exception(e)
+        end
+      end
+
+      def stop
+        puts 'Shutting down'
+        @shutdown = true
+        exit(42)
+      end
+    end
+  end
+end