WIP more incoming tx logic

lib/plugins.js

@@ -8,6 +8,9 @@ var POLLING_RATE = 60 * 1000; // poll each minute
 var REAP_RATE = 5 * 1000;
 var PENDING_TIMEOUT = 70 * 1000;
 
+// TODO: might have to update this if user is allowed to extend monitoring time
+var DEPOSIT_TIMEOUT = 120 * 1000;
+
 var db = null;
 
 var tickerPlugin = null;
@@ -238,6 +241,10 @@ function reapTx(row) {
 }
 
 function reapTxs() {
+  db.removeOldPending(DEPOSIT_TIMEOUT);
+
+  // NOTE: No harm in processing old pending tx, we don't need to wait for
+  // removeOldPending to complete.
   db.pendingTxs(PENDING_TIMEOUT, function(err, results) {
     if (err) return logger.warn(err);
     var rows = results.rows;
@@ -293,6 +300,11 @@ exports.cashOut = function cashOut(deviceFingerprint, tx, cb) {
   });
 };
 
+exports.dispenseStatus = function dispenseStatus(deviceFingerprint, sessionId,
+    cb) {
+  db.fetchDispenseStatus(deviceFingerprint, sessionId, cb);
+};
+
 exports.dispenseAck = function dispenseAck(deviceFingerprint, tx) {
 };
 

lib/postgresql_interface.js

@@ -153,10 +153,24 @@ function computeSendAmount(tx, totals) {
   return result;
 }
 
+exports.removeOldPending = function removeOldPending(timeoutMS) {
+  connect(function(err, client, done) {
+    var sql = 'DELETE FROM pending_transactions ' +
+      'WHERE incoming AND extract(EPOCH FROM now() - created) > $1)';
+    var timeoutS = timeoutMS / 1000;
+    var values = [timeoutS];
+    query(client, sql, values, function(err) {
+      done();
+      if (err) logger.error(err);
+    });
+  });
+};
+
 exports.pendingTxs = function pendingTxs(timeoutMS, cb) {
   connect(function(err, client, done) {
-    var sql = 'SELECT * FROM pending_transactions ' +
+    var sql = 'SELECT *, extract(EPOCH FROM now() - created) AS age ' +
-      'WHERE (incoming OR EXTRACT(EPOCH FROM now() - created > $2) ' +
+      'FROM pending_transactions ' +
+      'WHERE (incoming OR age > $2) ' +
       'ORDER BY created ASC';
     var timeoutS = timeoutMS / 1000;
     var values = [timeoutS];
@@ -303,7 +317,7 @@ exports.addInitialIncoming = function addInitialIncoming(deviceFingerprint, tx,
     async.waterfall([
       async.apply(silentQuery, client, 'BEGIN', null),
       async.apply(addPendingTx, client, deviceFingerprint, tx.sessionId,
-        tx.incoming, tx.currencyCode, tx.toAddress),
+        true, tx.currencyCode, tx.toAddress),
       async.apply(insertTx, client, tx, tx.satoshis, tx.fiat,
         'initial_request', 'pending')
     ], function(err) {
@@ -320,6 +334,56 @@ exports.addInitialIncoming = function addInitialIncoming(deviceFingerprint, tx,
 
 };
 
+function lastTxStatus(client, deviceFingerprint, sessionId, cb) {
+  var sql = 'SELECT satoshis, source FROM transactions ' +
+    'WHERE device_fingerprint=$1 AND session_id=$2 AND incoming=$3 ' +
+    'ORDER BY id DESC LIMIT 1';
+  var values = [deviceFingerprint, sessionId, true];
+
+  query(client, sql, values, cb);
+}
+
+function initialRequest(client, deviceFingerprint, sessionId, cb) {
+  var sql = 'SELECT fiat, satoshis FROM transactions ' +
+    'WHERE device_fingerprint=$1 AND session_id=$2 AND incoming=$3 ' +
+    'AND status=$4';
+  var values = [deviceFingerprint, sessionId, true, 'initial_request'];
+
+  query(client, sql, values, cb);
+}
+
+exports.dispenseStatus = function dispenseStatus(deviceFingerprint, sessionId,
+    cb) {
+
+  // NOTE: select for both device_fingerprint and session_id for security.
+  // Don't want to allow operators to read other machines' transactions.
+  connect(function(err, client, done) {
+    if (err) return cb(err);
+    async.parallel([
+      async.apply(client, initialRequest, deviceFingerprint, sessionId),
+      async.apply(client, lastTxStatus, deviceFingerprint, sessionId)
+    ], function(_err, results) {
+      done();
+      if (_err) return cb(_err);
+
+      var pending = (results[0].rows.length === 1) &&
+        (results[1].rows.length === 1) &&
+        (results[1].rows[0].status == 'deposit');
+      if (!pending) return cb(null, null);
+
+      var requiredSatoshis = results[0].rows[0].requiredSatoshis;
+      var lastTx = results[1].rows[1];
+
+      // TODO: handle multiple deposits
+      var status = (lastTx.satoshis < requiredSatoshis) ?
+        'insufficientFunds' :
+        lastTx.source;
+      cb(null, status);
+    });
+  });
+
+};
+
 /*
 exports.decrementCartridges =
     function decrementCartridges(fingerprint, cartridge1, cartridge2, cb) {

lib/routes.js

@@ -61,23 +61,28 @@ function poll(req, res) {
   var complianceSettings = config.exchanges.settings.compliance;
   var fiatCommission = config.exchanges.settings.fiatCommission ||
     config.exchanges.settings.commission;
-  var response = {
+  var sessionId = req.get('session-id');
-    err: null,
-    rate: rate * config.exchanges.settings.commission,
-    fiatRate: fiatRate / fiatCommission,
-    fiat: fiatBalance,
-    locale: config.brain.locale,
-    txLimit: parseInt(complianceSettings.maximum.limit, 10),
-    dispenseStatus: plugins.dispenseStatus(fingerprint),
-    idVerificationEnabled: complianceSettings.idVerificationEnabled,
-    cartridges: cartridges,
-    twoWayMode: cartridges ? true : false
-  };
 
-  if (response.idVerificationEnabled)
+  plugins.dispenseStatus(fingerprint, sessionId, function(err, dispenseStatus) {
-    response.idVerificationLimit = complianceSettings.idVerificationLimit;
+    if (err) return logger.error(err);
+    var response = {
+      err: null,
+      rate: rate * config.exchanges.settings.commission,
+      fiatRate: fiatRate / fiatCommission,
+      fiat: fiatBalance,
+      locale: config.brain.locale,
+      txLimit: parseInt(complianceSettings.maximum.limit, 10),
+      dispenseStatus: dispenseStatus,
+      idVerificationEnabled: complianceSettings.idVerificationEnabled,
+      cartridges: cartridges,
+      twoWayMode: cartridges ? true : false
+    };
 
-  res.json(response);
+    if (response.idVerificationEnabled)
+      response.idVerificationLimit = complianceSettings.idVerificationLimit;
+
+    res.json(response);
+  });
 }
 
 function trade(req, res) {