Start the refactor

lib/app.js

@@ -25,7 +25,7 @@ var argv = require('optimist').argv;
 var app = express();
 var fs = require('fs');
 var LamassuConfig = require('lamassu-config');
-var atm = require('lamassu-atm-protocol');
+var atm = require('./protocol/atm-api.js');
 
 var conString, dbConfig, config;
 

lib/protocol/api/api.js

@@ -0,0 +1,145 @@
+'use strict';
+
+require('date-utils');
+
+//var async = require('async');
+var winston = require('winston');
+var logger = new (winston.Logger)({transports:[new (winston.transports.Console)()]});
+var path = require('path');
+
+var _transferExchange;
+var _tickerExchange;
+var _tradeExchange;
+var _rates = {};
+var _config;
+var _commission;
+var _config;
+var SATOSHI_FACTOR = Math.pow(10, 8);
+
+exports.ticker = require('./ticker');
+exports.trade = require('./trade');
+exports.send = require('./send');
+exports.balance = require('./balance');
+exports._tradeExchange = null;
+exports._transferExchange = null;
+
+exports.findExchange = function (name) {
+  var exchange;
+
+  try {
+    exchange = require('lamassu-' + name);
+  } catch (err) {
+    if (!err.message.match(/Cannot find module/)) throw err;
+    exchange = require(path.join(path.dirname(__dirname), 'exchanges', name));
+  }
+
+  return exchange;
+};
+
+exports.findTicker = function (name) {
+  var exchange = exports.findExchange(name);
+  return exchange.ticker || exchange;
+};
+
+exports.findTrader = function (name) {
+  var exchange = exports.findExchange(name);
+  return exchange.trader || exchange;
+};
+
+exports.findWallet = function (name) {
+  var exchange = exports.findExchange(name);
+  return exchange.wallet || exchange;
+};
+
+exports.triggerBalance = function triggerBalance() {
+  this.balance.triggerBalance();
+};
+
+exports.init = function(config) {
+  _config = config;
+
+  if (config.settings.lowBalanceMargin < 1) {
+    throw new Error('`settings.lowBalanceMargin` has to be >= 1');
+  }
+
+  var tickerExchangeCode = config.plugins.current.ticker;
+  var tickerExchangeConfig = config.plugins.settings[tickerExchangeCode] || {};
+  tickerExchangeConfig.currency = config.settings.currency;
+  _tickerExchange = exports.findTicker(tickerExchangeCode).factory(tickerExchangeConfig);
+
+  var tradeExchangeCode = config.plugins.current.trade;
+  if (tradeExchangeCode) {
+    var tradeExchangeConfig = config.plugins.settings[tradeExchangeCode];
+    _tradeExchange = exports.findTrader(tradeExchangeCode).factory(tradeExchangeConfig);
+  }
+
+  var transferExchangeCode = config.plugins.current.transfer;
+  var transferExchangeConfig = config.plugins.settings[transferExchangeCode];
+  _commission = config.settings.commission;
+  _transferExchange = exports.findWallet(transferExchangeCode).factory(transferExchangeConfig);
+
+  var doRequestTradeExchange = _tradeExchange && tradeExchangeCode !== transferExchangeCode;
+
+  exports._tradeExchange = _tradeExchange;
+  exports._transferExchange = _transferExchange;
+  exports.ticker.init(config, exports, _tickerExchange);
+  exports.trade.init(config, exports, _tradeExchange, exports.ticker);
+  exports.send.init(config, exports, _transferExchange, exports.ticker);
+  exports.balance.init(config, exports, _transferExchange,
+                       doRequestTradeExchange ? _tradeExchange : null);
+};
+
+/**
+ * return fiat balance
+ *
+ * in input to this function, balance has the following parameters...
+ *
+ * balance.transferBalance - in satoshis
+ * balance.tradeBalance    - in USD
+ *
+ * Have added conversion here, but this really needs to be thought through, lamassu-bitstamp should perhaps
+ * return balance in satoshis
+ */
+exports.fiatBalance = function(rate, balance, transferSatoshis, tradeFiat, callback) {
+  if (!rate || !balance) return 0;
+
+  // The rate is actually our commission times real rate.
+  rate = _commission * rate;
+
+  // `lowBalanceMargin` is our safety net. It's a number > 1, and we divide
+  // all our balances by it to provide a safety margin.
+  var lowBalanceMargin = _config.settings.lowBalanceMargin;
+
+  // `balance.transferBalance` is the balance of our transfer account (the one
+  // we use to send Bitcoins to clients). `transferSatoshis` is the number
+  // of satoshis we're expected to send for this transaction. By subtracting
+  // them, we get `adjustedTransferBalance`, amount of satoshis we'll have
+  // after the transaction.
+  var adjustedTransferBalance = balance.transferBalance - transferSatoshis;
+  
+  // Since `adjustedTransferBalance` is in Satoshis, we need to turn it into
+  // Bitcoins and then fiat to learn how much fiat currency we can exchange.
+  //
+  // Unit validity proof: [ $ ] = [ (B * 10^8) / 10^8 * $/B ]
+  //                      [ $ ] = [ B * $/B ]
+  //                      [ $ ] = [ $ ]
+  var fiatTransferBalance = ((adjustedTransferBalance / SATOSHI_FACTOR) * rate) / lowBalanceMargin;
+
+  // If this server is also configured to trade received fiat for Bitcoins,
+  // we also need to calculate if we have enough funds on our trade exchange.
+  if (balance.tradeBalance === null) return fiatTransferBalance;
+  var tradeBalance = balance.tradeBalance;
+
+  // We need to secure `tradeFiat` (amount of fiat in this transaction) and
+  // enough fiat to cover our trading queue (trades aren't executed immediately).
+  var adjustedFiat = tradeFiat + exports.trade.queueFiatBalance(rate);
+
+  // So we subtract `adjustedFiat` from `tradeBalance` and again, apply
+  // `lowBalanceMargin`.
+  var fiatTradeBalance = (tradeBalance - adjustedFiat) / lowBalanceMargin;
+
+  // And we return the smallest number.
+  return Math.min(fiatTransferBalance, fiatTradeBalance);
+};
+
+

lib/protocol/api/balance.js

@@ -0,0 +1,49 @@
+'use strict';
+
+var _transferExchange;
+var _tradeExchange;
+var _api;
+var _config;
+var _balance = null;
+var _balanceTriggers = [];
+
+var winston = require('winston');
+var logger = new (winston.Logger)({transports:[new (winston.transports.Console)()]});
+
+var async = require('async');
+
+exports.init = function(config, api, transferExchange, tradeExchange) {
+  _api = api;
+  _config = config;
+
+  _transferExchange = transferExchange;
+  _tradeExchange = tradeExchange;
+
+  _balanceTriggers = [function (cb) { _transferExchange.balance(cb); }];
+
+  if (tradeExchange)
+    _balanceTriggers.push(function(cb) { _tradeExchange.balance(cb); });
+
+  _pollBalance();
+  setInterval(_pollBalance, 60 * 1000);
+};
+
+exports.balance = function balance() {
+  return _balance;
+};
+
+exports.triggerBalance = _pollBalance;
+
+function _pollBalance() {
+  logger.info('collecting balance');
+  async.parallel(_balanceTriggers, function(err, results) {
+    if (err) return;
+
+    _balance = {
+      transferBalance: results[0],
+      tradeBalance: results.length === 2 ? results[1] : null,
+      timestamp: Date.now()    
+    };
+    logger.info('Balance update:', _balance);
+  });
+}

lib/protocol/api/send.js

@@ -0,0 +1,37 @@
+'use strict';
+
+var _transferExchange;
+var _api;
+var _config;
+var _conString = process.env.DATABASE_URL || 'postgres://lamassu:lamassu@localhost/lamassu';
+var _db = require('../db/postgresql_interface').factory(_conString);
+
+exports.init = function(config, api, transferExchange) {
+  _api = api;
+  _config = config;
+  _transferExchange = transferExchange;
+};
+
+exports.setDomain = function(domain) {
+  _transferExchange.setDomain(domain);
+};
+
+exports.sendBitcoins = function sendBitcoins(deviceFingerprint, tx, cb) {
+  _db.summonTransaction(deviceFingerprint, tx, function (err, isNew, txHash) {
+    if (err) return cb(err);
+    if (isNew) return _transferExchange.sendBitcoins(tx.toAddress, tx.satoshis, 
+          _config.settings.transactionFee, function(err, txHash) {
+        if (err) {
+          _db.reportTransactionError(tx, err);
+          return cb(err);
+        }
+        cb(null, txHash);
+        _db.completeTransaction(tx, txHash);
+        _api.triggerBalance();
+      });
+
+    // transaction exists, but txHash might be null, 
+    // in which case ATM should continue polling  
+    cb(null, txHash);
+  });  
+};

lib/protocol/api/ticker.js

@@ -0,0 +1,33 @@
+'use strict';
+
+require('date-utils');
+var winston = require('winston');
+var logger = new (winston.Logger)({transports:[new (winston.transports.Console)()]});
+
+var _tickerExchange;
+var _api;
+var _rates = {};
+
+var _pollRate = function(currency) {
+  logger.info('polling for rate...');
+  _tickerExchange.ticker(currency, function(err, rate) {
+    if (err) return;
+    logger.info('Rate update:', rate);
+    _rates[currency] = {rate: rate, timestamp: new Date()};
+  });
+};
+
+exports.init = function(config, api, tickerExchange) {
+  _api = api;
+  _tickerExchange = tickerExchange;
+
+  _pollRate(config.settings.currency);
+  setInterval(function () {
+    _pollRate(config.settings.currency);
+  }, 60 * 1000);
+};
+
+exports.rate = function(currency) {
+  if (!_rates[currency]) return null;
+  return _rates[currency];
+};

lib/protocol/api/trade.js

@@ -0,0 +1,100 @@
+'use strict';
+
+require('date-utils');
+var winston = require('winston');
+var _ = require('underscore');
+var logger = new (winston.Logger)({transports:[new (winston.transports.Console)()]});
+
+var _tradeExchange;
+var _ticker;
+var _tradeQueue = [];
+var _api;
+var _config;
+
+var SATOSHI_FACTOR = Math.pow(10, 8);
+
+var _consolidateTrades = function() {
+  var queue = _tradeQueue;
+  var tradeRec = {
+    fiat: 0,
+    satoshis: 0,
+    currency: 'USD'
+  };
+
+  while (true) {
+    var lastRec = queue.shift();
+    if (!lastRec) {
+      break;
+    }
+    tradeRec.fiat += lastRec.fiat;
+    tradeRec.satoshis += lastRec.satoshis;
+    tradeRec.currency = lastRec.currency;
+  }
+  return tradeRec;
+};
+
+
+
+/**
+ * TODO: add error reporting
+ */
+var _purchase = function(trade) {
+  _ticker.rate(trade.currency, function(err, rate) {
+    _tradeExchange.purchase(trade.satoshis, rate, function(err) {
+      _api.triggerBalance();
+    });
+  });
+};
+
+exports.init = function(config, api, tradeExchange, ticker) {
+  _config = config;
+  _api = api;
+  _tradeExchange = tradeExchange;
+  _ticker = ticker;
+
+  var interval = setInterval(function() {
+    exports.executeTrades();
+  }, _config.settings.tradeInterval);
+  interval.unref();
+};
+
+exports.trade = function(fiat, satoshis, currency, cb) {
+  _tradeQueue.push({fiat: fiat, satoshis: satoshis, currency: currency});
+  cb(null);
+};
+
+exports.queueFiatBalance = function(exchangeRate) {
+  var satoshis = _.reduce(_tradeQueue, function(memo, rec) {
+    return memo + rec.satoshis;
+  }, 0);
+  return (satoshis / SATOSHI_FACTOR) * exchangeRate;
+};
+
+exports.executeTrades = function() {
+  if (!_tradeExchange) return;
+
+  logger.info('checking for trades');
+
+  if (!_config.plugins.current.trade) {
+    logger.info('NO ENGINE');
+    return;
+  }
+
+  var trade = _consolidateTrades();
+  logger.info('consolidated: ' + JSON.stringify(trade));
+
+  if (trade.fiat === 0) {
+    logger.info('reject fiat 0');
+    return;
+  }
+
+  if (trade.fiat < _config.settings.minimumTradeFiat) {
+    // throw it back in the water
+    logger.info('reject fiat too small');
+    _tradeQueue.unshift(trade);
+    return;
+  }
+
+  logger.info('making a trade: %d', trade.satoshis / Math.pow(10,8));
+  _purchase(trade);
+};

lib/protocol/atm-api.js

@@ -0,0 +1,116 @@
+'use strict';
+
+var api = exports.api = require('./api/api');
+var _config;
+var _lamassuConfig;
+var _commission;
+
+// Make sure these are higher than polling interval
+// or there will be a lot of errors
+var STALE_TICKER  = 180000;
+var STALE_BALANCE = 180000;
+
+Error.prototype.toJSON = function () {
+  var self = this;
+  var ret = {};
+  Object.getOwnPropertyNames(self).forEach(function (key) {
+    ret[key] = self[key];
+  });
+  return ret;
+};
+
+var poll = function(req, res) {
+  if (req.device.unpair) {
+    return res.json({
+      unpair: true
+    });
+  }
+
+  var rateRec = api.ticker.rate(req.params.currency);
+  var satoshiBalanceRec = api.balance.balance();
+
+  if (rateRec === null || satoshiBalanceRec === null)
+    return res.json({err: 'Server initializing'});
+  if (Date.now() - rateRec.timestamp > STALE_TICKER)
+    return res.json({err: 'Stale ticker'});
+  if (Date.now() - rateRec.timestamp > STALE_BALANCE)
+    return res.json({err: 'Stale balance'});
+
+  var rate = rateRec.rate;
+
+  res.json({
+    err: null,
+    rate: rate * _commission,
+    fiat: api.fiatBalance(rate, satoshiBalanceRec, 0, 0),
+    currency: req.params.currency,
+    txLimit: parseInt(_config.exchanges.settings.compliance.maximum.limit, 10)
+  });
+};
+
+// TODO need to add in a UID for this trade
+var trade = function(req, res) {
+  api.trade.trade(req.body.fiat, req.body.satoshis, req.body.currency, function(err) {
+    res.json({err: err});
+  });
+};
+
+var send = function(req, res) {
+  var fingerprint = req.connection.getPeerCertificate().fingerprint;
+  api.send.sendBitcoins(fingerprint, req.body, function(err, txHash) {
+    res.json({err: err, txHash: txHash});
+  });
+};
+
+var configurations = function(req, res) {
+  res.json({
+    err: _config.exchanges && _config.exchanges.settings ? null : new Error('Settings Not Found!'),
+    results: _config.exchanges.settings
+  });
+};
+
+var pair = function(req, res) {
+  var token = req.body.token;
+  var name = req.body.name;
+
+  _lamassuConfig.pair(
+    token,
+    req.connection.getPeerCertificate().fingerprint,
+    name,
+    function(err) {
+      if (err) res.json(500, { err: err.message });
+      else res.json(200);
+    }
+  );
+};
+
+exports.init = function(app, config, lamassuConfig, authMiddleware) {
+  _config = config;
+  _lamassuConfig = lamassuConfig;
+  
+  api.init(_config.exchanges);
+
+  _commission = _config.exchanges.settings.commission;
+
+  exports._tradeExchange = api._tradeExchange;
+  exports._transferExchange = api._transferExchange;
+
+  app.get('/poll/:currency', authMiddleware, poll);
+  app.get('/config', authMiddleware, configurations);
+  app.post('/trade', authMiddleware, trade);
+  app.post('/send', authMiddleware, send);
+  app.post('/pair', pair);
+
+  lamassuConfig.on('configUpdate', function () {
+    _lamassuConfig.load(function(err, config) {
+      if (err) {
+        return console.error('Error while reloading config');
+      }
+
+      _config = config;
+      api.init(_config.exchanges);
+      console.log('Config reloaded');
+    });
+  });
+
+  return app;
+};

lib/protocol/db/postgresql_interface.js

@@ -0,0 +1,69 @@
+'use strict';
+
+var pg = require('pg');
+var PG_ERRORS = {
+  23505: 'uniqueViolation'
+};
+
+var PostgresqlInterface = function (conString) {
+  if (!conString) {
+    throw new Error('Postgres connection string is required');
+  }
+
+  this.client = new pg.Client(conString);
+
+  // TODO better logging
+  this.client.on('error', function (err) { console.log(err); });
+
+  this.client.connect();
+};
+PostgresqlInterface.factory = function factory(conString) { return new PostgresqlInterface(conString); };
+module.exports = PostgresqlInterface;
+
+PostgresqlInterface.prototype.summonTransaction = 
+  function summonTransaction(deviceFingerprint, tx, cb) {
+  // First do an INSERT
+  // If it worked, go ahead with transaction
+  // If duplicate, fetch status and return
+  var self = this;
+  this.client.query('INSERT INTO transactions (id, status, deviceFingerprint, ' +
+      'toAddress, satoshis, currencyCode, fiat) ' + 
+      'VALUES ($1, $2, $3, $4, $5, $6, $7)', [tx.txId, 'pending', deviceFingerprint,
+      tx.toAddress, tx.satoshis, tx.currencyCode, tx.fiat], 
+      function (err) {
+    if (err && PG_ERRORS[err.code] === 'uniqueViolation')
+      return self._fetchTransaction(tx.txId, cb);
+    if (err) return cb(err);
+    cb(null, true);
+  });
+};
+
+PostgresqlInterface.prototype.reportTransactionError = 
+  function reportTransactionError(tx, err) {
+  this.client.query('UPDATE transactions SET status=$1, error=$2 WHERE id=$3', 
+    ['failed', err.message, tx.txId]);
+};
+
+PostgresqlInterface.prototype.completeTransaction = 
+  function completeTransaction(tx, txHash) {
+  if (txHash)
+    this.client.query('UPDATE transactions SET txHash=$1, status=$2, completed=now() WHERE id=$3', 
+      [txHash, 'completed', tx.txId]);
+  else
+    this.client.query('UPDATE transactions SET status=$1, error=$2 WHERE id=$3', 
+      ['failed', 'No txHash received', tx.txId]);    
+};
+
+PostgresqlInterface.prototype._fetchTransaction = 
+    function _fetchTransaction(txId, cb) {
+  this.client.query('SELECT status, txHash FROM transaction WHERE id=$1', 
+      [txId], function (err, rows) {
+    if (err) return cb(err);
+
+    // This should never happen, since we already checked for existence
+    if (rows === 0) return cb(new Error('Couldn\'t find transaction.')); 
+
+    var result = rows[0];
+    cb(null, false, result.txHash);
+  });
+};

lib/protocol/exchanges/custom_ticker.js

@@ -0,0 +1,45 @@
+'use strict';
+
+// TODO: refactor this with bitpay_ticker.js
+
+var https = require('https');
+var _ = require('underscore');
+
+var CustomTicker = function(config) {
+  this.config = config;
+};
+
+CustomTicker.factory = function factory(config) {
+  return new CustomTicker(config);
+};
+
+CustomTicker.prototype.ticker = function ticker(currency, cb) {
+  var self = this;
+  https.get(this.config.uri, function(res) {
+    var buf = '';
+    res.setEncoding('utf8');
+    res.on('data', function(chunk) {
+      buf += chunk;
+    })
+    .on('end', function() {
+      var json = null;
+      try {
+        json = JSON.parse(buf);
+      } catch(e) {
+        cb(new Error('Couldn\'t parse JSON response'));
+        return;
+      }
+      var rec = _.findWhere(json, {code: currency});
+
+      if (!rec) {
+        cb(new Error('Currency not listed: ' + currency));
+        return;
+      }
+      cb(null, rec.rate);
+    });
+  }).on('error', function(e) {
+    cb(e);
+  });
+};
+
+module.exports = CustomTicker;

package.json

@@ -15,7 +15,19 @@
     "express": "~3.4.7",
     "optimist": "~0.6.0",
     "lamassu-config": "~0.2.0",
-    "lamassu-atm-protocol": "~0.2.0"
+    "lodash": "~2.4.1",
+    "async": "~0.2.9",
+    "deepmerge": "~0.2.7",
+    "underscore": "~1.5.2",
+    "error-create": "0.0.0",
+    "date-utils": "~1.2.15",
+    "bitstamp": "~0.1.3",
+    "winston": "~0.7.2",
+    "pg": "~2.11.1",
+    "lamassu-bitpay": "~0.0.1",
+    "lamassu-bitstamp": "~0.0.1",
+    "lamassu-mtgox": "~0.0.1",
+    "lamassu-blockchain": "0.0.4"
   },
   "repository": {
     "type": "git",