Bustabit's old source

README.md

@@ -1 +1,2 @@
-bustabit
+
+### Bustabit

depositor/Procfile

@@ -0,0 +1 @@
+worker: node daemon.js

depositor/daemon.js

@@ -0,0 +1,179 @@
+var assert = require('better-assert');
+var async = require('async');
+var bc = require('./src/bitcoin_client');
+var db = require('./src/db');
+var lib = require('./src/lib');
+var fs = require('fs')
+
+
+var client;
+
+// Mapping of deposit_address -> user_id
+var depositAddresses = JSON.parse(fs.readFileSync('.addresses.json', 'utf8'));
+assert(depositAddresses);
+
+
+startBlockLoop();
+
+function processTransactionIds(txids, callback) {
+
+    bc.getTransactionIdsAddresses(txids, function(err, addressToAmountLists) {
+        if (err) return callback(err);
+
+        assert(txids.length === addressToAmountLists.length);
+
+        var tasks = [];
+
+        addressToAmountLists.forEach(function(addressToAmount, i) {
+
+            var txid = txids[i];
+            assert(txid);
+
+            var usersToAmounts = {};
+
+            Object.keys(addressToAmount).forEach(function(address) {
+
+                var userId = depositAddresses[address];
+                if (userId) {
+                    usersToAmounts[userId] = addressToAmount[address];
+                }
+            });
+
+
+            if (Object.keys(usersToAmounts).length > 0) {
+                console.log('Transactions: ', txid, ' matches: ', usersToAmounts);
+
+                Object.keys(usersToAmounts).forEach(function(userId) {
+                    tasks.push(function(callback) {
+                        db.addDeposit(userId, txid, usersToAmounts[userId], callback);
+                    });
+                });
+
+
+            }
+        });
+
+        async.parallelLimit(tasks, 3, callback);
+    });
+}
+
+
+
+// Handling the block...
+
+
+/// block chain loop
+
+var lastBlockCount;
+var lastBlockHash;
+
+function startBlockLoop() {
+    // initialize...
+    db.getLastBlock(function (err, block) {
+        if (err)
+            throw new Error('Unable to get initial last block: ', err);
+
+        lastBlockCount = block.height;
+        lastBlockHash = block.hash;
+
+        console.log('Initialized on block: ', lastBlockCount, ' with hash: ', lastBlockHash);
+
+        blockLoop();
+    });
+}
+
+function scheduleBlockLoop() {
+    setTimeout(blockLoop, 20000);
+}
+
+function blockLoop() {
+    bc.getBlockCount(function(err, num) {
+        if (err) {
+            console.error('Unable to get block count');
+            return scheduleBlockLoop();
+        }
+
+        if (num === lastBlockCount) {
+            console.log('Block chain still ', num, ' length. No need to do anything');
+            return scheduleBlockLoop();
+        }
+
+        bc.getBlockHash(lastBlockCount, function(err, hash) {
+            if (err) {
+                console.error('Could not get block hash, error: ' + err);
+                return scheduleBlockLoop();
+            }
+
+            if (lastBlockHash !== hash) {
+                // There was a block-chain reshuffle. So let's just jump back a block
+                db.getBlock(lastBlockCount - 1, function(err, block) {
+                    if (err) {
+                        console.error('ERROR: Unable jump back ', err);
+                        return scheduleBlockLoop();
+                    }
+
+                    --lastBlockCount;
+                    lastBlockHash = block.hash;
+                    blockLoop();
+                });
+                return;
+            }
+
+            bc.getBlockHash(lastBlockCount+1, function(err, hash) {
+                if (err) {
+                    console.error('Unable to get block hash: ', lastBlockCount+1);
+                    return scheduleBlockLoop();
+                }
+
+                processBlock(hash, function(err) {
+                    if (err) {
+                        console.error('Unable to process block: ', hash, ' because: ', err);
+                        return scheduleBlockLoop();
+                    }
+
+                    ++lastBlockCount;
+                    lastBlockHash = hash;
+
+
+                    db.insertBlock(lastBlockCount, lastBlockHash, function(err) {
+                       if (err)
+                          console.error('Danger, unable to save results in database...');
+
+                        // All good! Loop immediately!
+                        blockLoop();
+                    });
+                });
+            });
+
+        });
+    });
+}
+
+function processBlock(hash, callback) {
+    console.log('Processing block: ', hash);
+
+    var start = new Date();
+
+    bc.getBlock(hash, function(err, blockInfo) {
+        if (err) {
+            console.error('Unable to get block info for: ', hash, ' got error: ', err);
+            return callback(err);
+        }
+
+        var transactionsIds = blockInfo.tx;
+
+
+        processTransactionIds(transactionsIds, function(err) {
+            if (err)
+                console.log('Unable to process block (in ',  (new Date() - start ) / 1000, ' seconds)');
+            else
+                console.log('Processed ', transactionsIds.length, ' transactions in ', (new Date() - start ) / 1000, ' seconds');
+
+            callback(err)
+        });
+    });
+
+}
+
+
+

depositor/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "bv-notifier",
+  "description": "An alerting service for bitcoin transactions.",
+  "version": "0.0.1",
+  "private": true,
+  "dependencies": {
+    "async": "~0.7.0",
+    "better-assert": "~1.0.0",
+    "bitcoin": "~2.1.2",
+    "bitcoinjs-lib": "^2.2.0",
+    "debug": "^2.2.0",
+    "pg": "^4.5.1"
+  },
+  "scripts": {
+    "start-dev": "env $(cat .env | xargs) nodemon daemon.js",
+    "start": "env $(cat .env | xargs) forever start daemon.js",
+    "generate-addresses": "env $(cat .env | xargs) node ./src/generate_addresses > .addresses.json",
+    "generate-wallet": "env $(cat .env | xargs) node ./src/generate_wallet > .wallet.sh"
+  }
+}

depositor/src/bitcoin_client.js

@@ -0,0 +1,102 @@
+var assert = require('better-assert');
+var bitcoin = require('bitcoin');
+
+
+var lib = require('./lib');
+
+var client = new bitcoin.Client({
+    host: process.env.BITCOIND_HOST,
+    port: process.env.BITCOIND_PORT || 8332,
+    user: process.env.BITCOIND_USER,
+    pass: process.env.BITCOIND_PASS,
+    timeout: 240000
+});
+
+assert(client.rpc.opts.host);
+assert(client.rpc.opts.user);
+assert(client.rpc.opts.pass);
+
+function doGetTransactions(txIds, callback) {
+    if (txIds.length === 0)
+        return callback(null, []);
+
+    var batch = txIds.map(function(txId) {
+        return {
+            method: 'getrawtransaction',
+            params: [txId, 1]
+        };
+    });
+
+    var abort = false;
+    var transactions = [];
+    var count = 0;
+
+    client.cmd(batch, function(err, transaction) {
+        if (abort) return;
+
+        if (err) {
+            abort = true;
+            return callback(err);
+        }
+
+        transactions.push(transaction);
+
+        if (++count === txIds.length) {
+            return callback(null, transactions);
+        }
+
+        assert(count < txIds.length);
+    });
+}
+
+client.getTransactions = function(txIds, callback) {
+    return lib.chunkRun(doGetTransactions, txIds, 3, 1, function(err, data) {
+
+        if (err) {
+            console.error('Error: when fetching', txIds.length, ' transactions, got error: ', err);
+            return callback(err);
+        }
+
+        callback(null, data);
+    });
+};
+
+client.getTransaction = function(transactionHash, callback) {
+    client.getRawTransaction(transactionHash, 1, callback);
+};
+
+// returns [{address: amount}])
+function transactionsAddresses(transactions) {
+
+    return transactions.map(function(transaction) {
+        var addressToAmount = {};
+
+        transaction.vout.forEach(function (out) {
+            var addresses = out.scriptPubKey.addresses;
+            if (!addresses || addresses.length !== 1) {
+                return;
+            }
+
+            assert(out.value >= 0);
+            var oldAmount = addressToAmount[addresses[0]] || 0;
+            addressToAmount[addresses[0]] = oldAmount + out.value;
+        });
+
+        return addressToAmount;
+    });
+}
+
+function doGetTransactionIdsAddresses(txids, callback) {
+    doGetTransactions(txids, function(err, transactions) {
+        if (err) return callback(err);
+
+        callback(null, transactionsAddresses(transactions));
+    });
+}
+
+// callback(err, listOfAddressToAmount
+client.getTransactionIdsAddresses = function(txIds, callback) {
+    lib.chunkRun(doGetTransactionIdsAddresses, txIds, 20, 2, callback);
+};
+
+module.exports = client;