blockHandler for wallet-api service

lib/encoding.js

@@ -55,7 +55,6 @@ Encoding.prototype.decodeAddressIndexKey = function(buffer) {
 };
 
 Encoding.prototype.encodeUtxoIndexKey = function(address, txid, outputIndex) {
-console.log('encodeUtxoIndexKey');
   var prefix = new Buffer('01', 'hex');
   var buffers = [this.servicePrefix, prefix];
 
@@ -372,11 +371,13 @@ Encoding.prototype.decodeWalletUtxoSatoshisValue = function(buffer) {
 };
 
 Encoding.prototype.encodeWalletKey = function(walletId) {
-  return new Buffer(walletId, 'hex');
+  var prefix = new Buffer('00', 'hex');
+  var walletIdBuffer = new Buffer(walletId, 'hex');
+  return Buffer.concat([this.servicePrefix, prefix, walletIdBuffer]);
 };
 
 Encoding.prototype.decodeWalletKey = function(buffer) {
-  return buffer.toString('hex');
+  return buffer.slice(3).toString('hex');
 };
 
 Encoding.prototype.encodeWalletValue = function(addresses, balance) {

lib/services/address/index.js

@@ -171,7 +171,7 @@ AddressService.prototype.concurrentBlockHandler = function(block, connectBlock,
         continue;
       }
 
-      var address = self._getAddressString(script);
+      var address = self.getAddressString(script);
 
       if(!address) {
         continue;
@@ -210,7 +210,7 @@ AddressService.prototype.concurrentBlockHandler = function(block, connectBlock,
         continue;
       }
 
-      var inputAddress = self._getAddressString(input.script);
+      var inputAddress = self.getAddressString(input.script);
 
       if(!inputAddress) {
         continue;
@@ -261,17 +261,17 @@ AddressService.prototype.blockHandler = function(block, connectBlock, callback)
 
       if(!script) {
         log.debug('Invalid script');
-        return next();
+        continue;
       }
 
-      var address = self._getAddressString(script);
+      var address = self.getAddressString(script);
 
       if(!address) {
-        return next();
+        continue;
       }
 
       var key = self.encoding.encodeUtxoIndexKey(address, txid, outputIndex);
-      var value = self.encoding.encodeUtxoIndexValue(output.satoshis, output.script);
+      var value = self.encoding.encodeUtxoIndexValue(output.satoshis, output._scriptBuffer);
       operations.push({
         type: action,
         key: key,
@@ -291,7 +291,7 @@ AddressService.prototype.blockHandler = function(block, connectBlock, callback)
         return next();
       }
 
-      var inputAddress = self._getAddressString(input.script);
+      var inputAddress = self.getAddressString(input.script);
 
       if(!inputAddress) {
         return next();
@@ -307,7 +307,7 @@ AddressService.prototype.blockHandler = function(block, connectBlock, callback)
       } else { // uncommon and slower, this happens during a reorg
         self.node.services.transaction.getTransaction(input.prevTxId, {}, function(err, tx) {
           var utxo = tx.outputs[input.outputIndex];
-          var inputValue = self.encoding.encodeUtxoIndexValue(utxo.satoshis, utxo.script);
+          var inputValue = self.encoding.encodeUtxoIndexValue(utxo.satoshis, utxo._scriptBuffer);
           operations.push({
             type: 'put',
             key: inputKey,
@@ -323,13 +323,15 @@ AddressService.prototype.blockHandler = function(block, connectBlock, callback)
     });
   }, function(err) {
     //we are aync predicated on reorg sitch
-    setImmediate(function() {
+    if(err) {
-      callback(null, operations);
+      return callback(err);
-    });
+    }
+
+    callback(null, operations);
   });
 };
 
-AddressService.prototype._getAddressString = function(script, output) {
+AddressService.prototype.getAddressString = function(script, output) {
   var address = script.toAddress();
   if(address) {
     return address.toString();

lib/services/wallet-api/index.js

@@ -18,14 +18,12 @@ var utils = require('./utils');
 var _ = require('lodash');
 var bodyParser = require('body-parser');
 var LRU = require('lru-cache');
+var Encoding = require('../../encoding'); // TODO this needs to be split out by service
+
 
 var WalletService = function(options) {
   BaseService.call(this, options);
-  this._dbOptions = {
+
-    keyEncoding: 'string',
-    valueEncoding: 'json'
-  };
-  this._db = levelup(options.dbPath, this._dbOptions);
   this._cache = LRU({
     max: 500 * 1024 * 1024,
     length: function(n, key) {
@@ -33,6 +31,8 @@ var WalletService = function(options) {
     },
     maxAge: 30 * 60 * 1000
   });
+
+  this._addressMap = {};
 };
 
 inherits(WalletService, BaseService);
@@ -46,7 +46,18 @@ WalletService.prototype.getAPIMethods = function() {
   return [];
 };
 WalletService.prototype.start = function(callback) {
-  setImmediate(callback);
+  var self = this;
+
+  self.node.services.db.getPrefix(self.name, function(err, servicePrefix) {
+    if(err) {
+      return callback(err);
+    }
+
+    self.servicePrefix = servicePrefix;
+    self._encoding = new Encoding(self.servicePrefix);
+
+    self._loadAllAddresses(callback);
+  });
 };
 
 WalletService.prototype.stop = function(callback) {
@@ -57,6 +68,224 @@ WalletService.prototype.getPublishEvents = function() {
   return [];
 };
 
+WalletService.prototype.blockHandler = function(block, connectBlock, callback) {
+  var self = this;
+
+  var txs = block.transactions;
+
+  var action = 'put';
+  var reverseAction = 'del';
+  if (!connectBlock) {
+    action = 'del';
+    reverseAction = 'put';
+  }
+
+  var operations = [];
+
+  async.eachSeries(txs, function(tx, next) {
+    var inputs = tx.inputs;
+    var outputs = tx.outputs;
+
+    for (var outputIndex = 0; outputIndex < outputs.length; outputIndex++) {
+      var output = outputs[outputIndex];
+
+      var script = output.script;
+
+      if(!script) {
+        log.debug('Invalid script');
+        continue;
+      }
+
+      var address = self.services.address.getAddressString(script);
+
+      if(!address || !self._addressMap[address]) {
+        continue;
+      }
+
+      var walletIds = self._addressMap[address];
+
+      walletIds.forEach(function(walletId) {
+        operations.push({
+          type: action,
+          key: self._encoding.encodeWalletUtxoKey(walletId, tx.id, outputIndex),
+          value: self._encoding.encodeWalletUtxoValue(block.__height, output.satoshis, address)
+        });
+
+        operations.push({
+          type: action,
+          key: self._encoding.encodeWalletUtxoSatoshisKey(walletId, output.satoshis, tx.id, outputIndex),
+          value: self._encoding.encodeWalletUtxoValue(block.__height, address)
+        });
+      });
+    }
+
+    if(tx.isCoinbase()) {
+      return next();
+    }
+
+    //TODO deal with P2PK
+    async.each(inputs, function(input, next) {
+      if(!input.script) {
+        log.debug('Invalid script');
+        return next();
+      }
+
+      var inputAddress = self.services.address.getAddressString(input.script);
+
+      if(!inputAddress || !self._addressMap[inputAddress]) {
+        return next();
+      }
+
+      var walletIds = self._addressMap[inputAddress];
+
+      async.each(walletIds, function(walletId, next) {
+        self.node.services.transaction.getTransaction(input.prevTxId, {}, function(err, tx) {
+          if(err) {
+            return next(err);
+          }
+
+          var utxo = tx.outputs[input.outputIndex];
+
+          operations.push({
+            type: reverseAction,
+            key: self._encoding.encodeWalletUtxoKey(walletId, input.prevTxId, input.outputIndex),
+            value: self._encoding.encodeWalletUtxoValue(tx.__height, utxo.satoshis, inputAddress)
+          });
+
+          operations.push({
+            type: reverseAction,
+            key: self._encoding.encodeWalletUtxoSatoshisKey(walletId, utxo.satoshis, tx.id, input.outputIndex),
+            value: self._encoding.encodeWalletUtxoSatoshisValue(tx.__height, inputAddress)
+          });
+
+          next();
+        });
+      }, next);
+    }, next);
+  }, function(err) {
+    if(err) {
+      return callback(err);
+    }
+
+    callback(null, operations);
+  });
+};
+
+WalletService.prototype.concurrentBlockHandler = function(block, connectBlock, callback) {
+  var self = this;
+
+  var txs = block.transactions;
+  var height = block.__height;
+
+  var action = 'put';
+  if (!connectBlock) {
+    action = 'del';
+  }
+
+  var operations = [];
+
+  for(var i = 0; i < txs.length; i++) {
+    var tx = txs[i];
+    var inputs = tx.inputs;
+    var outputs = tx.outputs;
+
+    for (var outputIndex = 0; outputIndex < outputs.length; outputIndex++) {
+      var output = outputs[outputIndex];
+
+      var script = output.script;
+
+      if(!script) {
+        log.debug('Invalid script');
+        continue;
+      }
+
+      var address = self.node.services.address.getAddressString(script);
+
+      if(!address || !self._addressMap[address]) {
+        continue;
+      }
+
+      var walletIds = self._addressMap[address];
+
+      walletIds.forEach(function(walletId) {
+        operations.push({
+          type: action,
+          key: self._encoding.encodeWalletTransactionKey(walletId, block.__height, i),
+          value: self._encoding.encodeWalletTransactionValue(tx.id)
+        })
+      });
+    }
+
+    if(tx.isCoinbase()) {
+      continue;
+    }
+
+    //TODO deal with P2PK
+    for(var inputIndex = 0; inputIndex < inputs.length; inputIndex++) {
+      var input = inputs[inputIndex];
+
+      if(!input.script) {
+        log.debug('Invalid script');
+        continue;
+      }
+
+      var inputAddress = self._getAddressString(input.script);
+
+      if(!inputAddress || !self._addressMap[inputAddress]) {
+        continue;
+      }
+
+      var walletIds = self._addressMap[inputAddress];
+
+      walletIds.forEach(function(walletId) {
+        operations.push({
+          type: action,
+          key: self._encoding.encodeWalletTransactionKey(walletId, block.__height, i),
+          value: self._encoding.encodeWalletTransactionValue(tx.id)
+        });
+      });
+    }
+  }
+  setImmediate(function() {
+    callback(null, operations);
+  });
+};
+
+WalletService.prototype._loadAllAddresses = function(callback) {
+  var self = this;
+
+  var start = self._encoding.encodeWalletKey('00');
+  var end = self._encoding.encodeWalletKey(Array(65).join('f'));
+
+  var stream = self.db.createReadStream({
+    gte: start,
+    lt: end
+  });
+
+  var streamErr = null;
+
+  stream.on('data', function(data) {
+    var key = self._encoding.decodeWalletKey(data.key);
+    var value = self._encoding.decodeWalletValue(data.value);
+
+    value.addresses.forEach(function(address) {
+      if(!self._addressMap[address]) {
+        self._addressMap[address] = [];
+      }
+
+      self._addressMap[address].push(key);
+    });
+  });
+
+  stream.on('error', function(err) {
+    streamErr = err;
+  });
+
+  stream.on('end', function() {
+    callback(streamErr);
+  });
+};
+
 WalletService.prototype._endpointUTXOs = function() {
   var self = this;
   return function(req, res) {