fix

Fixed: _aggregateAddressSummaryResult to handle duplicate tx data

lib/services/address/index.js

@@ -16,7 +16,7 @@ var utils = require('../../utils');
 var LRU = require('lru-cache');
 var XXHash = require('xxhash');
 
-const MAX_TX_QUERY_LIMIT = 1000;
+const MAX_TX_QUERY_LIMIT = 100;
 
 // See rationale about this cache at function getTxList(next)
 const TXID_LIST_CACHE_ITEMS = 250;              // nr of items (this translates to: consecutive 
@@ -177,10 +177,10 @@ AddressService.prototype.getAddressHistory = function(addresses, options, stream
   var self = this;
 
   options = options || {};
-  options.from = options.from || 0;
+  //options.from = options.from || 0; //Deprecated, use `after` option
-  options.to = options.to || 0xffffffff;
+  //options.to = options.to || 0xffffffff; //Deprecated, use `after` option
 
-  if(!callback){ //if only 3 args, then streamer is callback
+  if(typeof callback !== 'function'){ //if only 3 args, then streamer is callback
     callback = streamer;
     streamer = () => null; //NULL fn
   }
@@ -189,6 +189,13 @@ AddressService.prototype.getAddressHistory = function(addresses, options, stream
     options.queryMempool = true;
   }
 
+  //Quick support for `from` and `to` options (DEPRECATED! Not recommeded to use)
+  if( !_.isUndefined(options.from) || !_.isUndefined(options.to)) { 
+    options.old_support = true;
+    options.from = options.from || 0;
+    options.to = options.to || 0xffffffff; //Max value of to will actually be MAX_TX_QUERY_LIMIT
+  }
+
   if (_.isString(addresses)) {
     addresses = [addresses];
   }
@@ -202,13 +209,23 @@ AddressService.prototype.getAddressHistory = function(addresses, options, stream
 
     self._streamAddressSummary(address, options, function(err, tx){
 
-      results.totalCount++;
-
       if(err)
         return log.error(err);
 
-      if(!options.txNotNeeded && results.items.length < MAX_TX_QUERY_LIMIT)
+      if(!options.txNotNeeded) {
-        results.items.push(tx);
+        results.totalCount++;
+        
+        if(!results.items.some(x => x.txid() === tx.txid())) //push only if tx not already in array
+          results.items.unshift(tx); //using unshift, so that recent tx (low) are at front
+        
+        if(results.items.length > MAX_TX_QUERY_LIMIT) { //remove items from array when overflown
+          results.items.sort((a, b) => (b.__height || 0xffffffff) - (a.__height || 0xffffffff) || b.txid().localeCompare(a.txid()));
+          let del_count = options.old_support ? results.items.length : results.items.length - MAX_TX_QUERY_LIMIT;
+          let start_index = options.old_support ? MAX_TX_QUERY_LIMIT : 0;
+          results.items.splice(start_index, del_count);
+        }
+      
+      }
       
       streamer(null, tx);
 
@@ -216,11 +233,20 @@ AddressService.prototype.getAddressHistory = function(addresses, options, stream
 
   }, function(err) {
 
+    console.debug("END @getAddressHistory");
     if (err) {
       return callback(err);
     }
 
-    //TODO: sorting of tx list (results.items)
+    //sort items in desc block-height, then asc txid (if same height)
+    results.items.sort((a, b) => (b.__height || 0xffffffff) - (a.__height || 0xffffffff) || b.txid().localeCompare(a.txid()));
+    results.totalCount = parseInt(results.totalCount.toFixed());
+
+    //Quick support for `from` and `to` options (DEPRECATED! Not recommeded to use)
+    if(options.old_support) { 
+      results.items = results.items.slice(options.from, options.to);
+    }
+
     callback(null, results);
 
   })
@@ -254,7 +280,7 @@ AddressService.prototype.__getAddressSummary = function(address, options, callba
     txApperances: 0,
   };
 
-  self.getAddressHistory(address, options, function(err, results) {
+  self.__getAddressHistory(address, options, function(err, results) { //old fn
 
     if (err) {
       return callback(err);
@@ -277,15 +303,15 @@ AddressService.prototype.getAddressSummary = function(address, options, streamer
   var self = this;
 
   options = options || {};
-  options.from = options.from || 0;
+  //options.from = options.from || 0; //Deprecated
-  options.to = options.to || 0xffffffff;
+  //options.to = options.to || 0xffffffff; //Deprecated
   options.txNotNeeded = true; //no need to store tx details in result
 
   if (_.isUndefined(options.queryMempool)) {
     options.queryMempool = true;
   }
 
-  if(!callback){ //if only 3 args, then streamer is callback
+  if(typeof callback !== 'function'){ //if only 3 args, then streamer is callback
     callback = streamer;
     streamer = () => null; //NULL fn
   }
@@ -304,15 +330,13 @@ AddressService.prototype.getAddressSummary = function(address, options, streamer
     txApperances: 0,
   };
 
-  self._aggregateAddressSummaryResult(tx, address, result);
-
   self.getAddressHistory(address, options, function(err, tx) {
 
     if(err)
         return log.error(err);
 
     if(tx)
-      self._aggregateAddressSummaryResult(tx, address, result);
+      self._aggregateAddressSummaryResult(tx, address, result, options);
     
     streamer(null, tx);
 
@@ -322,6 +346,13 @@ AddressService.prototype.getAddressSummary = function(address, options, streamer
       return callback(err);
     }
 
+    result.balanceSat = parseInt(result.balanceSat.toFixed());
+    result.totalReceivedSat = parseInt(result.totalReceivedSat.toFixed());
+    result.totalSentSat = parseInt(result.totalSentSat.toFixed());
+    result.txApperances = parseInt(result.txApperances.toFixed());
+    result.unconfirmedBalanceSat = parseInt(result.unconfirmedBalanceSat.toFixed());
+    result.unconfirmedTxApperances = parseInt(result.unconfirmedTxApperances.toFixed());
+
     result.balance = Unit.fromSatoshis(result.balanceSat).toBTC();
     result.totalReceived = Unit.fromSatoshis(result.totalReceivedSat).toBTC();
     result.totalSent = Unit.fromSatoshis(result.totalSentSat).toBTC();
@@ -375,10 +406,13 @@ AddressService.prototype._setInputResults = function(tx, address, result) {
   }
 };
 
-AddressService.prototype._aggregateAddressSummaryResult = function (tx, address, result, options){
+AddressService.prototype._getAddressSummaryResult = function(txs, address, result, options) {
 
   var self = this;
 
+  for(var i = 0; i < txs.length; i++) {
+    var tx = txs[i];
+
     self._setOutputResults(tx, address, result);
     self._setInputResults(tx, address, result);
 
@@ -391,17 +425,109 @@ AddressService.prototype._aggregateAddressSummaryResult = function (tx, address,
 
   }
 
-AddressService.prototype._getAddressSummaryResult = function(txs, address, result, options) {
+  return result;
+};
+
+AddressService.prototype._getOccurrenceCount = function(tx, address) {
+  let count = 0;
+
+  for(var i = 0; i < tx.inputs.length; i++) {
+
+    var input = tx.inputs[i];
+
+    if(utils.getAddress(input, this._network) === address)
+      count++;
+
+  }
+
+  for(var j = 0; j < tx.outputs.length; j++) {
+
+    var output = tx.outputs[j];
+
+    if(utils.getAddress(output, this._network) === address)
+      count++;
+
+  }
+
+  return count;
+
+}
+
+AddressService.prototype._getOutputResults = function(tx, address) {
+
+  let value = 0;
+
+  for(var j = 0; j < tx.outputs.length; j++) {
+
+    var output = tx.outputs[j];
+
+    if (utils.getAddress(output, this._network) === address) 
+      value += output.value;
+
+  }
+
+  return value;
+
+};
+
+AddressService.prototype._getInputResults = function(tx, address) {
+  
+  let value = 0;
+
+  for(var i = 0; i < tx.inputs.length; i++) {
+
+    var input = tx.inputs[i];
+
+    if (utils.getAddress(input, this._network) === address) 
+      value += tx.__inputValues[i];
+
+  }
+
+  return value;
+
+};
+
+AddressService.prototype._aggregateAddressSummaryResult = function (tx, address, result, options) {
   
   var self = this;
 
-  for(var i = 0; i < txs.length; i++) {
+  let output_val = self._getOutputResults(tx, address);
-    var tx = txs[i];
+  let input_val = self._getInputResults(tx, address);
-    self._aggregateAddressSummaryResult(tx, address, result, options);
+  
+  //aggregate the result
+  result.txApperances++;
+
+  result.totalReceivedSat += output_val;
+  result.balanceSat += output_val;
+
+  result.totalSentSat += input_val;
+  result.balanceSat -= input_val;
+
+  if(!tx.confirmations){
+    result.unconfirmedTxApperances++;
+    result.unconfirmedBalanceSat += output_val;
+    result.unconfirmedBalanceSat -= input_val;
   }
 
-  return result;
+    if (!options.noTxList) {
-};
+
+      if (!result.transactions)  {
+        result.transactions = [];
+      }
+
+      let txid = tx.txid();
+      if(!result.transactions.includes(txid)) {  //push txid only if its not in the array
+        
+        result.transactions.unshift(txid); //using unshift, so that recent tx (low confirmation) are at front
+        
+        if(result.transactions.length > MAX_TX_QUERY_LIMIT)
+          result.transactions.pop(); //pop the oldest tx in list (when list limit is maxed out)
+      
+        }
+
+    }
+
+}
 
 AddressService.prototype.getAddressUnspentOutputs = function(address, options, callback) {
 
@@ -565,18 +691,28 @@ AddressService.prototype.stop = function(callback) {
 
 AddressService.prototype._getTxidStream = function(address, options) {
 
-  var start = this._encoding.encodeAddressIndexKey(address);
+  var start;
+  if(options.after) {
+    start = this._encoding.encodeAddressIndexKey(address, options.start, options.after, 0xffffffff, 1, 0xffffffff); //0xffffffff is for getting after the txid
+  } else {
+    start = this._encoding.encodeAddressIndexKey(address, options.start);
+  }
+
+  var endHeightBuf = new Buffer(4);
+  endHeightBuf.writeUInt32BE(options.end);
+
   var end = Buffer.concat([
     start.slice(0, address.length + 4),
-    options.endHeightBuf,
+    endHeightBuf,
     new Buffer(new Array(83).join('f'), 'hex')
   ]);
 
   var criteria = {
     gte: start,
-    lte: end,
+    lte: end
-    reverse: true // txids stream from low confirmations to high confirmations
+    //reverse: true // txids stream from low confirmations to high confirmations
   };
+  //NOTE: commentted reverse to keep the order in asc when reading to preserve continuity when using `after` option
 
   // txid stream
   var txidStream = this._db.createKeyStream(criteria);
@@ -588,6 +724,7 @@ AddressService.prototype._getTxidStream = function(address, options) {
   return txidStream;
 };
 
+//(used by old fn)
 AddressService.prototype._getAddressTxHistory = function(options, callback) {
 
   var self = this;
@@ -616,6 +753,7 @@ AddressService.prototype._getAddressTxHistory = function(options, callback) {
 
 };
 
+//(used by old fn)
 AddressService.prototype._getAddressTxidHistory = function(address, options, callback) {
   var self = this;
 
@@ -625,9 +763,6 @@ AddressService.prototype._getAddressTxidHistory = function(address, options, cal
 
   var results = [];
 
-  options.endHeightBuf = new Buffer(4);
-  options.endHeightBuf.writeUInt32BE(options.end);
-
   if (_.isUndefined(options.queryMempool)) {
     options.queryMempool = true;
   }
@@ -682,7 +817,9 @@ AddressService.prototype._getAddressTxidHistory = function(address, options, cal
           return;
         }
 
+        if(!results.some(r => r.txid == txInfo.txid)) //add txid to array only if its not already there
           results.push({ txid: txInfo.txid, height: txInfo.height });
+
         callback();
       };
 
@@ -701,25 +838,29 @@ AddressService.prototype._streamAddressSummary = function(address, options, stre
   options.start = options.start || 0;
   options.end = options.end || 0xffffffff;
 
-  options.from = options.from || 0; //TODO: check from/to options are working or not
+  //options.from = options.from || 0; //Deprecated, use `after` option
-  options.to = options.to || 0xffffffff;
+  //options.to = options.to || 0xffffffff; //Deprecated, use `after` option
-
-  if (_.isUndefined(options.queryMempool)) {
-    options.queryMempool = true;
-  }
-
-  options.endHeightBuf = new Buffer(4);
-  options.endHeightBuf.writeUInt32BE(options.end);
 
   if (_.isUndefined(options.queryMempool)) {
     options.queryMempool = true;
   }
 
   //declare the queue to process tx data
+  var tmpTxList = {}; //store processed txid temporarily to ignore duplication
 
-  var q = async.queue(function(task, cb) {
+  var q = async.queue(function(id, cb) {
 
-    let {id, options} = task;
+    //duplication finding
+    if(id.txid in tmpTxList){
+
+      tmpTxList[id.txid][0]++;
+      
+      if(tmpTxList[id.txid][1] !== null && tmpTxList[id.txid][0] >= tmpTxList[id.txid][1]) //all duplications are found for this txid
+        delete tmpTxList[id.txid];
+      
+      return cb();
+
+    } else tmpTxList[id.txid] = [1, null];
 
     if (id.height === 0xffffffff) {
 
@@ -738,14 +879,29 @@ AddressService.prototype._streamAddressSummary = function(address, options, stre
 
   }, 4);
 
-  q.pause(); //pause and wait until queue is set
+  //q.pause(); //pause and wait until queue is set (not needed)
 
-  function process_chunk(err, tx){
+  function chunkCallback(err, tx){
+
+    if(q.killed || (!err && !tx)) //no error or tx data (duplicate calls will have empty tx value) 
+      return; 
+
+    if(tx){
+      let txid = tx.txid();
+      tmpTxList[txid][1] = self._getOccurrenceCount(tx, address);
+
+      if(tmpTxList[txid][0] >= tmpTxList[txid][1]) //all duplications are found for this txid
+        delete tmpTxList[txid];
+    }
 
     streamer(err, tx);
 
-    if(err){
+    if((err || options.flag_stop) && !q.killed){
+
+      console.debug("KILL", options.flag_stop)
+
       q.kill();
+      q.killed = true;
 
       return callback();
     }
@@ -754,6 +910,68 @@ AddressService.prototype._streamAddressSummary = function(address, options, stre
 
   async.waterfall([
 
+    //Find start height if `after` option is passed
+    function(next){
+
+      if(_.isUndefined(options.after)) {
+        return next();
+      }
+      
+      self._transaction.getTransaction(options.after, options, function(err, tx) {
+
+        if(tx && tx.confirmations && tx.height >= options.start) {
+  
+          options.start = tx.height;
+
+        } else { 
+
+          delete options.after;
+
+        } 
+
+        next();
+        
+      });
+
+    },
+
+    // stream the rest of the confirmed txids out of the address index
+    function(next) {
+
+      var txIdTransformStream = new Transform({ objectMode: true });
+
+      txIdTransformStream._flush = function(cb) {
+        txIdTransformStream.emit('end');
+        cb();
+      };
+
+      txIdTransformStream.on('error', function(err) {
+        log.error('Address Service: txstream err: ' + err);
+        txIdTransformStream.unpipe();
+      });
+
+      txIdTransformStream.on('end', function() {
+        next();
+      });
+
+      txIdTransformStream._transform = function(chunk, enc, cb) {
+
+        if(options.flag_stop){//stop data query 
+          console.debug("FLAG STOP:", options.flag_stop)  
+          return txIdTransformStream.unpipe();
+        }
+
+        var txInfo = self._encoding.decodeAddressIndexKey(chunk);
+        q.push({ txid: txInfo.txid, height: txInfo.height }, chunkCallback);
+        
+        cb();
+      };
+
+      var txidStream = self._getTxidStream(address, options);
+      txidStream.pipe(txIdTransformStream);
+      
+    },
+
     // query the mempool for relevant txs for this address
     function(next) {
 
@@ -771,41 +989,23 @@ AddressService.prototype._streamAddressSummary = function(address, options, stre
         return next();
       }
 
-      mempoolTxids.map(id => q.push(id, process_chunk));
+      mempoolTxids.map(id => q.push(id, chunkCallback));
       next();
     },
-    // stream the rest of the confirmed txids out of the address index
+
+    //wait for queue to complete
     function(next) {
 
-      var txIdTransformStream = new Transform({ objectMode: true });
+      console.debug("WAITING FOR QUEUE TO COMPLETE", q.started)
       
-      txIdTransformStream._flush = function(cb) {
+      if(!q.started) //No tx in query
-        txIdTransformStream.emit('end');
+        return next();
-        cb();
-      };
 
-      txIdTransformStream.on('error', function(err) {
+      else
-        log.error('Address Service: txstream err: ' + err);
+        q.drain = () => next();
-        txIdTransformStream.unpipe();
-      });
-
-      txIdTransformStream.on('end', function() {
-        q.resume();
-      });
-
-      txIdTransformStream._transform = function(chunk, enc, cb) {
-        var txInfo = self._encoding.decodeAddressIndexKey(chunk);
-        q.push({ txid: txInfo.txid, height: txInfo.height }, process_chunk);
-        
-        cb();
-      };
-
-      var txidStream = self._getTxidStream(address, options);
-      txidStream.pipe(txIdTransformStream);
-
-      q.drain(next);
 
     }
+
   ], callback);
 
 }

lib/services/web/index.js

@@ -100,13 +100,15 @@ WebService.prototype.stop = function(callback) {
  * all of the exposed HTTP routes.
  */
 WebService.prototype.setupAllRoutes = function() {
-  for(var key in this.node.services) {
+  const self = this;
+
+  for(var key in self.node.services) {
     var subApp = new express();
-    var service = this.node.services[key];
+    var service = self.node.services[key];
 
     if(service.getRoutePrefix && service.setupRoutes) {
-      this.app.use('/' + this.node.services[key].getRoutePrefix(), subApp);
+      self.app.use('/' + self.node.services[key].getRoutePrefix(), subApp);
-      this.node.services[key].setupRoutes(subApp, express, express_ws);
+      self.node.services[key].setupRoutes(subApp, express, a => express_ws(a, self.server));
     } else {
       log.debug('No routes defined for: ' + key);
     }