init.js

//require('newrelic');
var fs = require('fs');
var path = require('path');
var os = require('os');
var cluster = require('cluster');

var request = require('request');
var async = require('async');
var extend = require('extend');
var MiningRigRentalsAPI = require('miningrigrentals-api')
var process = require('process')

var PoolLogger = require('log4js');
var CliListener = require('./libs/cliListener.js');
var PoolWorker = require('./libs/poolWorker.js');
var PaymentProcessor = require('./libs/paymentProcessor.js');
var Website = require('./libs/website.js');
var ProfitSwitch = require('./libs/profitSwitch.js');

var algos = require('merged-pooler/lib/algoProperties.js');

MRRAPI = new MiningRigRentalsAPI(process.env.MRR_API_KEY, process.env.MRR_API_SECRET)


JSON.minify = JSON.minify || require("node-json-minify");

if (!fs.existsSync('config.json')){
    console.log('config.json file does not exist. Read the installation/setup instructions.');
    return;
}

var portalConfig = JSON.parse(JSON.minify(fs.readFileSync("config.json", {encoding: 'utf8'})));
var poolConfigs;

var logger = PoolLogger.getLogger();

logger.info('New Relic');
try {
//    require('newrelic');
    if (cluster.isMaster)
        logger.debug('NewRelic', 'Monitor', 'New Relic initiated');
} catch(e) {}


//Try to give process ability to handle 100k concurrent connections
try{
    var posix = require('posix');
    try {
    logger.info('Setting POSIX');
        posix.setrlimit('nofile', { soft: 100000, hard: 100000 });
    logger.info('POSIX Set');
    }
    catch(e){
    logger.info(e);
    logger.error('Must be ran as root');
        if (cluster.isMaster){
            logger.warn('POSIX', 'Connection Limit', '(Safe to ignore) Must be ran as root to increase resource limits');}
    }
    finally {
        // Find out which user used sudo through the environment variable
        var uid = parseInt(process.env.SUDO_UID);
        // Set our server's uid to that user
        if (uid) {
            process.setuid(uid);
        logger.info("UID Set");
            logger.debug('POSIX', 'Connection Limit', 'Raised to 100K concurrent connections, now running as non-root user: ' + process.getuid());
        logger.info('POSIX Msg');
        }
    }
}
catch(e){
    logger.info('POSIX Not Installed');
    if (cluster.isMaster)
        logger.debug('POSIX', 'Connection Limit', '(Safe to ignore) POSIX module not installed and resource (connection) limit was not raised');
}

logger.info('Run Workers');
if (cluster.isWorker){

    switch(process.env.workerType){
        case 'pool':
            new PoolWorker(logger);
            break;
        case 'paymentProcessor':
            new PaymentProcessor(logger);
            break;
        case 'website':
            new Website(logger);
            break;
        case 'profitSwitch':
            new ProfitSwitch(logger);
            break;
    }

    return;
}


//Read all pool configs from pool_configs and join them with their coin profile
var buildPoolConfigs = function(){
    var configs = {};
    var configDir = 'pool_configs/';

    var poolConfigFiles = [];


    /* Get filenames of pool config json files that are enabled */
    fs.readdirSync(configDir).forEach(function(file){
        if (!fs.existsSync(configDir + file) || path.extname(configDir + file) !== '.json') return;
        var poolOptions = JSON.parse(JSON.minify(fs.readFileSync(configDir + file, {encoding: 'utf8'})));
        if (!poolOptions.enabled) return;
        poolOptions.fileName = file;
        poolConfigFiles.push(poolOptions);
    });


    /* Ensure no pool uses any of the same ports as another pool */
    for (var i = 0; i < poolConfigFiles.length; i++){
        var ports = Object.keys(poolConfigFiles[i].ports);
        for (var f = 0; f < poolConfigFiles.length; f++){
            if (f === i) continue;
            var portsF = Object.keys(poolConfigFiles[f].ports);
            for (var g = 0; g < portsF.length; g++){
                if (ports.indexOf(portsF[g]) !== -1){
                    logger.error('Master', poolConfigFiles[f].fileName, 'Has same configured port of ' + portsF[g] + ' as ' + poolConfigFiles[i].fileName);
                    process.exit(1);
                    return;
                }
            }

            if (poolConfigFiles[f].coin === poolConfigFiles[i].coin){
                logger.error('Master', poolConfigFiles[f].fileName, 'Pool has same configured coin file coins/' + poolConfigFiles[f].coin + ' as ' + poolConfigFiles[i].fileName + ' pool');
                process.exit(1);
                return;
            }

        }
    }


    poolConfigFiles.forEach(function(poolOptions){

        poolOptions.coinFileName = poolOptions.coin;

       for (var i=0; i < poolOptions.auxes.length; i++){
            var auxFilePath = 'coins/' + poolOptions.auxes[i].coin;
            if (!fs.existsSync(auxFilePath)){
            logger.warn('Aux', poolOptions.auxes[i].coin, 'could not find file: ' + auxFilePath);
            return;
            }

            var auxProfile = JSON.parse(JSON.minify(fs.readFileSync(auxFilePath, {encoding: 'utf8'})));
            poolOptions.auxes[i].coin = auxProfile;
            poolOptions.auxes[i].coin.name = poolOptions.auxes[i].coin.name.toLowerCase();

            if (!(auxProfile.algorithm in algos)){
            logger.warn('Master', auxProfile.name, 'Cannot run a pool for unsupported algorithm "' + auxProfile.algorithm + '"');
            delete configs[poolOptions.auxcoin.name];
            }
        }

        var coinFilePath = 'coins/' + poolOptions.coinFileName;
        if (!fs.existsSync(coinFilePath)){
            logger.warn('Master', poolOptions.coinFileName, 'could not find file: ' + coinFilePath);
            return;
        }

        var coinProfile = JSON.parse(JSON.minify(fs.readFileSync(coinFilePath, {encoding: 'utf8'})));
        poolOptions.coin = coinProfile;
        poolOptions.coin.name = poolOptions.coin.name.toLowerCase();

        if (poolOptions.coin.name in configs){

            logger.warn('Master', poolOptions.fileName, 'coins/' + poolOptions.coinFileName
                + ' has same configured coin name ' + poolOptions.coin.name + ' as coins/'
                + configs[poolOptions.coin.name].coinFileName + ' used by pool config '
                + configs[poolOptions.coin.name].fileName);

            process.exit(1);
            return;
        }

        for (var option in portalConfig.defaultPoolConfigs){
            if (!(option in poolOptions)){
                var toCloneOption = portalConfig.defaultPoolConfigs[option];
                var clonedOption = {};
                if (toCloneOption.constructor === Object)
                    extend(true, clonedOption, toCloneOption);
                else
                    clonedOption = toCloneOption;
                poolOptions[option] = clonedOption;
            }
        }


        configs[poolOptions.coin.name] = poolOptions;

        if (!(coinProfile.algorithm in algos)){
            logger.warn('Master', coinProfile.name, 'Cannot run a pool for unsupported algorithm "' + coinProfile.algorithm + '"');
            delete configs[poolOptions.coin.name];
        }
    });
    return configs;
};

var buildAuxConfigs = function(){
    var configs = {};
    var configDir = 'aux_configs/';

    var poolConfigFiles = [];


    /* Get filenames of pool config json files that are enabled */
    fs.readdirSync(configDir).forEach(function(file){
        if (!fs.existsSync(configDir + file) || path.extname(configDir + file) !== '.json') return;
        var poolOptions = JSON.parse(JSON.minify(fs.readFileSync(configDir + file, {encoding: 'utf8'})));
        if (!poolOptions.enabled) return;
        poolOptions.fileName = file;
        poolConfigFiles.push(poolOptions);
    });

    poolConfigFiles.forEach(function(poolOptions){

        poolOptions.coinFileName = poolOptions.coin;

        var poolFilePath = 'coins/' + poolOptions.coinFileName;
        if (!fs.existsSync(poolFilePath)){
            logger.warn('Master', poolOptions.coinFileName, 'could not find file: ' + poolFilePath);
            return;
        }

        var poolProfile = JSON.parse(JSON.minify(fs.readFileSync(poolFilePath, {encoding: 'utf8'})));
        poolOptions.coin = poolProfile;
        poolOptions.coin.name = poolOptions.coin.name.toLowerCase();
        configs[poolOptions.coin.name] = poolOptions;

        for (var option in portalConfig.defaultPoolConfigs){
            if (!(option in poolOptions)){
                var toCloneOption = portalConfig.defaultPoolConfigs[option];
                var clonedOption = {};
                if (toCloneOption.constructor === Object)
                    extend(true, clonedOption, toCloneOption);
                else
                    clonedOption = toCloneOption;
                poolOptions[option] = clonedOption;
            }
        }

        if (!(poolProfile.algorithm in algos)){
            logger.warn('Master', coinProfile.name, 'Cannot run a pool for unsupported algorithm "' + coinProfile.algorithm + '"');
            delete configs[poolOptions.coin.name];
        }

    });
    return configs;
};

var spawnPoolWorkers = function(){

    Object.keys(poolConfigs).forEach(function(coin){
        var p = poolConfigs[coin];

        if (!Array.isArray(p.daemons) || p.daemons.length < 1){
            logger.error('Master', coin, 'No daemons configured so a pool cannot be started for this coin.');
            delete poolConfigs[coin];
        }
    });

    if (Object.keys(poolConfigs).length === 0){
        logger.warn('Master', 'PoolSpawner', 'No pool configs exists or are enabled in pool_configs folder. No pools spawned.');
        return;
    }


    var serializedConfigs = JSON.stringify(poolConfigs);

    var numForks = (function(){
        if (!portalConfig.clustering || !portalConfig.clustering.enabled)
            return 1;
        if (portalConfig.clustering.forks === 'auto')
            return os.cpus().length;
        if (!portalConfig.clustering.forks || isNaN(portalConfig.clustering.forks))
            return 1;
        return portalConfig.clustering.forks;
    })();

    var poolWorkers = {};

    var createPoolWorker = function(forkId){
        var worker = cluster.fork({
            workerType: 'pool',
            forkId: forkId,
            pools: serializedConfigs,
            portalConfig: JSON.stringify(portalConfig)
        });
        worker.forkId = forkId;
        worker.type = 'pool';
        poolWorkers[forkId] = worker;
        worker.on('exit', function(code, signal){
            logger.error('Master', 'PoolSpawner', 'Fork ' + forkId + ' died, spawning replacement worker...');
            setTimeout(function(){
                createPoolWorker(forkId);
            }, 2000);
        }).on('message', function(msg){
            switch(msg.type){
                case 'banIP':
                    Object.keys(cluster.workers).forEach(function(id) {
                        if (cluster.workers[id].type === 'pool'){
                            cluster.workers[id].send({type: 'banIP', ip: msg.ip});
                        }
                    });
                    break;
            }
        });
    };

    var i = 0;
    var spawnInterval = setInterval(function(){
        createPoolWorker(i);
        i++;
        if (i === numForks){
            clearInterval(spawnInterval);
            logger.debug('Master', 'PoolSpawner', 'Spawned ' + Object.keys(poolConfigs).length + ' pool(s) on ' + numForks + ' thread(s)');
        }
    }, 250);

};


var startCliListener = function(){

    var cliPort = portalConfig.cliPort;

    var listener = new CliListener(cliPort);
    listener.on('log', function(text){
        logger.debug('Master', 'CLI', text);
    }).on('command', function(command, params, options, reply){

        switch(command){
            case 'blocknotify':
                Object.keys(cluster.workers).forEach(function(id) {
                    cluster.workers[id].send({type: 'blocknotify', coin: params[0], hash: params[1]});
                });
                reply('Pool workers notified');
                break;
            case 'coinswitch':
                processCoinSwitchCommand(params, options, reply);
                break;
            case 'reloadpool':
                poolConfigs = buildPoolConfigs();
                Object.keys(cluster.workers).forEach(function(id) {
                    cluster.workers[id].send({type: 'reloadpool', coin: params[0], pools: JSON.stringify(poolConfigs) });
                });
                reply('reloaded pool ' + params[0]);
                break;
            default:
                reply('unrecognized command "' + command + '"');
                break;
        }
    }).start();
};


var processCoinSwitchCommand = function(params, options, reply){

    var logSystem = 'CLI';
    var logComponent = 'coinswitch';

    var replyError = function(msg){
        reply(msg);
        logger.error(logSystem, logComponent, msg);
    };

    if (!params[0]) {
        replyError('Coin name required');
        return;
    }

    if (!params[1] && !options.algorithm){
        replyError('If switch key is not provided then algorithm options must be specified');
        return;
    }
    else if (params[1] && !portalConfig.switching[params[1]]){
        replyError('Switch key not recognized: ' + params[1]);
        return;
    }
    else if (options.algorithm && !Object.keys(portalConfig.switching).filter(function(s){
        return portalConfig.switching[s].algorithm === options.algorithm;
    })[0]){
        replyError('No switching options contain the algorithm ' + options.algorithm);
        return;
    }

    var messageCoin = params[0].toLowerCase();
    var newCoin = Object.keys(poolConfigs).filter(function(p){
        return p.toLowerCase() === messageCoin;
    })[0];

    if (!newCoin){
        replyError('Switch message to coin that is not recognized: ' + messageCoin);
        return;
    }


    var switchNames = [];

    if (params[1]) {
        switchNames.push(params[1]);
    }
    else{
        for (var name in portalConfig.switching){
            if (portalConfig.switching[name].enabled && portalConfig.switching[name].algorithm === options.algorithm)
                switchNames.push(name);
        }
    }

    switchNames.forEach(function(name){
        if (poolConfigs[newCoin].coin.algorithm !== portalConfig.switching[name].algorithm){
            replyError('Cannot switch a '
                + portalConfig.switching[name].algorithm
                + ' algo pool to coin ' + newCoin + ' with ' + poolConfigs[newCoin].coin.algorithm + ' algo');
            return;
        }

        Object.keys(cluster.workers).forEach(function (id) {
            cluster.workers[id].send({type: 'coinswitch', coin: newCoin, switchName: name });
        });
    });

    reply('Switch message sent to pool workers');

};



var startPaymentProcessor = function(){

    var enabledForAny = false;
    for (var pool in poolConfigs){
        var p = poolConfigs[pool];
        var enabled = p.enabled && p.paymentProcessing && p.paymentProcessing.enabled;
        if (enabled){
            enabledForAny = true;
            break;
        }
    }

    if (!enabledForAny)
        return;

    var worker = cluster.fork({
        workerType: 'paymentProcessor',
        pools: JSON.stringify(poolConfigs)
    });
    worker.on('exit', function(code, signal){
        logger.error('Master', 'Payment Processor', 'Payment processor died, spawning replacement...');
        setTimeout(function(){
            startPaymentProcessor(poolConfigs);
        }, 2000);
    });
};

var startAuxPaymentProcessor = function(){

    var enabledForAny = false;
    for (var aux in auxConfigs){
        var p = auxConfigs[aux];
        var enabled = p.enabled && p.paymentProcessing && p.paymentProcessing.enabled;
        if (enabled){
            enabledForAny = true;
            break;
        }
    }

    if (!enabledForAny)
        return;

    var worker = cluster.fork({
        workerType: 'paymentProcessor',
        pools: JSON.stringify(auxConfigs)
    });
    worker.on('exit', function(code, signal){
        logger.error('Master', 'Auxilliary Payment Processor', 'Auxilliary Payment processor died, spawning replacement...');
        setTimeout(function(){
            startPaymentProcessor(auxConfigs);
        }, 2000);
    });
};

var startWebsite = function(){

    if (!portalConfig.website.enabled) return;

    var worker = cluster.fork({
        workerType: 'website',
        pools: JSON.stringify(poolConfigs),
        portalConfig: JSON.stringify(portalConfig)
    });
    worker.on('exit', function(code, signal){
        logger.error('Master', 'Website', 'Website process died, spawning replacement...');
        setTimeout(function(){
            startWebsite(portalConfig, poolConfigs);
        }, 2000);
    });
};


var startProfitSwitch = function(){

    if (!portalConfig.profitSwitch || !portalConfig.profitSwitch.enabled){
        //logger.error('Master', 'Profit', 'Profit auto switching disabled');
        return;
    }

    var worker = cluster.fork({
        workerType: 'profitSwitch',
        pools: JSON.stringify(poolConfigs),
        portalConfig: JSON.stringify(portalConfig)
    });
    worker.on('exit', function(code, signal){
        logger.error('Master', 'Profit', 'Profit switching process died, spawning replacement...');
        setTimeout(function(){
            startProfitSwitch(poolConfigs);
        }, 2000);
    });
};



(function init(){

    poolConfigs = buildPoolConfigs();

    auxConfigs = buildAuxConfigs();

    spawnPoolWorkers();
        setTimeout(function(){
    startPaymentProcessor();

    startAuxPaymentProcessor();

    startWebsite();

    startProfitSwitch();

    startCliListener();


    setInterval(updateDloaComment, 60 * 1000);

        }, 10000);
})();

var dloaCommentCache = 'api.alexandria.io/pool/'
var mrr_last10_cache = []
var mrr_last10_sum = 0
var mrr_last10_avg = 'nr'
function updateDloaComment() {
    async.parallel({
        mrr: function (callback) {
            MRRAPI.listRigs({type: 'scrypt'}, function (err, resp) {
              if (!!err)
                return callback(null, {
                  last_10: 'nr',
                  last_24hr: 'nr'
                })
              var body;
              try { body = JSON.parse(resp)} catch (e) { return callback(null, {last_10: 'nr', last_24hr: 'nr'}); }
              if (body && body['success']) {
                var last_10 = body['data']['info']['price']['last_10']
                last_10 = parseFloat(last_10)
                if (!isNaN(last_10)) {
                  mrr_last10_cache.push(last_10)
                  mrr_last10_sum += last_10
                  if (mrr_last10_cache.length > 1440) {
                    mrr_last10_sum -= mrr_last10_cache.shift()
                    mrr_last10_avg = mrr_last10_sum / 1440
                  }
                  callback(null, {
                    last_10: last_10,
                    last_24hr: mrr_last10_avg
                  });
                } else {
                    callback(null, {
                      last_10: 'nr',
                      last_24hr: 'nr'
                    });
                }
              } else {
                callback(null, {
                  last_10: 'nr',
                  last_24hr: 'nr'
                });
              }
            })
        },
        pool: function (callback) {
        request('https://api.alexandria.io/pool/api/stats', function (error, response, body) {
          if (!error && response.statusCode == 200) {
            try {
              var parsed = JSON.parse(body);
              return callback(null, {
                hashrate: parsed.pools.florincoin.hashrate
              });
            } catch (err) {
              console.error('Unable to parse Pool API response', err);
            }
          }
          callback(null, {
            hashrate: 'nr'
          });
        })
        },
        fmd: function (callback) {
        request('https://api.alexandria.io/flo-market-data/v1/getAll', function (error, response, body) {
          if (!error && response.statusCode == 200) {
            try {
              var parsed = JSON.parse(body);
              return callback(null, {
                weighted: parsed.weighted,
                usd: parsed.USD
              });
            } catch (err) {
              console.error('Unable to parse FMD API response', err);
            }
          }
          callback(null, {
            weighted: 'nr',
            usd: 'nr'
          });
        })
        },
        fbd: function (callback) {
        request('https://api.alexandria.io/florincoin/getMiningInfo', function (error, response, body) {
          if (!error && response.statusCode == 200) {
            try {
              var parsed = JSON.parse(body);
              return callback(null, {
                networkhashps: parsed.networkhashps
              });
            } catch (err) {
              console.error('Unable to parse HR API response', err);
            }
          }
          callback(null, {
            networkhashps: 'nr'
          });
        })
        },
        cmc: function (callback) {
        request('https://api.coinmarketcap.com/v1/ticker/litecoin/', function (error, response, body) {
          if (!error && response.statusCode == 200) {
            try {
              var parsed = JSON.parse(body);
              return callback(null, {
                ltc_usd: parsed[0].price_usd
              });
            } catch (err) {
                console.error('Unable to parse HR API response', err);
            }
          }
          callback(null, {
            ltc_usd: 'nr'
          });
        })
        }
    },
    function (err, results) {
        if (!err) {
            var api_arr = [
              'oip-historian-3',
              'FLmic78oU6eqXsTAaHGGdrFyY7FznjHfPU',
              results.mrr.last_10,
              results.mrr.last_24hr,
              results.pool.hashrate,
              results.fbd.networkhashps,
              results.fmd.weighted,
              results.fmd.usd,
              results.cmc.ltc_usd
            ];

            dloaCommentCache = api_arr.join(':');
        }
        else
            dloaCommentCache = 'api.alexandria.io/pool/:error';

        Object.keys(cluster.workers).forEach(function(id) {
          if (cluster.workers[id].type === 'pool'){
            cluster.workers[id].send({type: 'dloaComment', dloaCommentUnsigned: dloaCommentCache});
          }
         });
    });
}