wip

lib/services/address/index.js

@@ -398,57 +398,67 @@ AddressService.prototype.onReorg = function(args, callback) {
 
   var removalOps = [];
   // for every tx, remove the address index key for every input and output
+  // for every output record we remove, we need to replace it with the output pointer located in the input that spent it
   // TODO: DRY self up!
-  for(var i = 0; i < oldBlockList.length; i++) {
+  async.eachSeries(oldBlockList, function(block, next) {
 
-    var block = oldBlockList[i];
-    //txs
-    for(var j = 0; j < block.txs.length; j++) {
+    async.eachSeries(block.txs, function(tx, next) {
 
-      var tx = block.txs[j];
+      async.parallelLimit([
 
-      //inputs
-      var address;
+        function(next) {
+          async.eachOfSeries(tx.inputs, function(index, input, next) {
 
-      for(var k = 0; k < tx.inputs.length; k++) {
+            var address = input.getAddress();
 
-        var input = tx.inputs[k];
-        address = input.getAddress();
+            if (!address) {
+              return next();
+            }
 
-        if (!address) {
-          continue;
+            address.network = self._network;
+            address = address.toString();
+
+            removalOps.push({
+              type: 'del',
+              key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 1, block.ts)
+            });
+            // look up prev out here
+            removalOps.push({
+              type: 'put',
+              key: self._encoding.encodeUtxoIndexKey(address, tx.txid(), input.prevout.index),
+              value: self._encoding.encodeUtxoIndexValue(prevTx.height,
+
+          });
+        },
+        function(next) {
+
+          async.eachOfSeries(tx.outputs, function(index, output, next) {
+
+            var address = output.getAddress();
+
+            if (!address) {
+              return next();
+            }
+
+            address.network = self._network;
+            address = address.toString();
+
+            removalOps.push({
+              type: 'del',
+              key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 0, block.ts)
+            });
+
+          });
+        }
+      ], 4, function(err) {
+
+        if (err) {
+          return callback(err);
         }
 
-        address.network = self._network;
-        address = address.toString();
-
-        removalOps.push({
-          type: 'del',
-          key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), k, 1, block.ts)
-        });
-      }
-
-      //outputs
-      for(k = 0; k < tx.outputs.length; k++) {
-
-        var output = tx.outputs[k];
-        address = output.getAddress();
-
-        if (!address) {
-          continue;
-        }
-
-        address.network = self._network;
-        address = address.toString();
-
-        removalOps.push({
-          type: 'del',
-          key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), k, 0, block.ts)
-        });
-
-      }
-    }
-  }
+      });
+    });
+  });
 
   callback(null, removalOps);
 };
@@ -473,7 +483,7 @@ AddressService.prototype.onBlock = function(block, callback) {
   callback(null, operations);
 };
 
-AddressService.prototype._processInput = function(tx, input, opts) {
+AddressService.prototype._processInput = function(tx, input, index, opts) {
 
   var address = input.getAddress();
 
@@ -483,12 +493,14 @@ AddressService.prototype._processInput = function(tx, input, opts) {
 
   address.network = this._network;
   address = address.toString();
+
   var txid = tx.txid();
   var timestamp = this._timestamp.getTimestampSync(opts.block.rhash());
+
   assert(timestamp, 'Must have a timestamp in order to process input.');
 
   // address index
-  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid, opts.inputIndex, 1, timestamp);
+  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid, index, 1, timestamp);
 
   var operations = [{
     type: 'put',
@@ -501,11 +513,6 @@ AddressService.prototype._processInput = function(tx, input, opts) {
     key: this._encoding.encodeUtxoIndexKey(address, input.prevout.txid(), input.prevout.index)
   };
 
-  // In the event where we are reorg'ing,
-  // this is where we are putting a utxo back in, we don't know what the original height, sats, or scriptBuffer
-  // since this only happens on reorg and the utxo that was spent in the chain we are reorg'ing away from will likely
-  // be spent again sometime soon, we will not add the value back in, just the key
-
   operations.push(rec);
 
   return operations;
@@ -521,8 +528,10 @@ AddressService.prototype._processOutput = function(tx, output, index, opts) {
 
   address.network = this._network;
   address = address.toString();
+
   var txid = tx.txid();
   var timestamp = this._timestamp.getTimestampSync(opts.block.rhash());
+
   assert(timestamp, 'Must have a timestamp in order to process output.');
 
   var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid, index, 0, timestamp);
@@ -561,14 +570,21 @@ AddressService.prototype._processTransaction = function(tx, opts) {
   });
 
   outputOperations = _.flatten(_.compact(outputOperations));
+  assert(outputOperations.length % 2 === 0 &&
+    outputOperations.length <= tx.outputs.length * 2,
+    'Output operations count is not reflective of what should be possible.');
+
 
   var inputOperations = tx.inputs.map(function(input, index) {
-    _opts.inputIndex = index;
-    return self._processInput(tx, input, _opts);
+    return self._processInput(tx, input, index, _opts);
   });
 
   inputOperations = _.flatten(_.compact(inputOperations));
 
+  assert(inputOperations.length % 2 === 0 &&
+    inputOperations.length <= tx.inputs.length * 2,
+    'Input operations count is not reflective of what should be possible.');
+
   outputOperations = outputOperations.concat(inputOperations);
   return outputOperations;