bitcoin transaction tracking

gistfile1.js

var config = require('../config')
, bitcoin = new (require('bitcoin').Client)(config('BTC'))
, async = require('async')
, db = require('./db')
, num = require('num')
, lastHeight
, clientHeight
, blockHash
, minConf = 3
, txConcurrency = 2
, txOutputConcurrency = 5
, block
, debug = require('debug')('snow:btc-sync')
// returns true when there might be more work immediately
, self = module.exports = function(cb) {
    async.series({
        'find last height in db': function(next) {
            var client = db()
            client.query('SELECT MAX(height) height FROM btc_block', function(err, res) {
                client.end()
                if (err) return next(err)
                lastHeight = res.rows[0].height
                debug('height of database = ' + lastHeight)
                next()
            })
        },

        'find current height': function(next) {
            bitcoin.getBlockCount(function(err, h) {
                if (err) return next(err)
                clientHeight = h

                debug('bitcoind\'s last block height = ' + h)

                // example: database has done block height 5
                if (clientHeight <= lastHeight + minConf) {
                    debug('nothing to do (client height=' + clientHeight + '; internal=' + lastHeight + '; minConf=' + minConf)
                        return cb(null, false)
                }

                next()
            })
        },

        'find the block hash': function(next) {
            debug('looking up hash of height ' + lastHeight + 1)

            bitcoin.getBlockHash(lastHeight + 1, function(err, h) {
                if (err) return next(err)
                blockHash = h
                debug('block at height ' + (lastHeight + 1) + ' found with hash ' + blockHash)
                next()
            })
        },

        'get the block': function(next) {
            debug('retrieving block ' + blockHash)

            bitcoin.getBlock(blockHash, function(err, b) {
                if (err) return next(err)
                block = b
                next()
            })
        },

        'enumerate the transactions': function(next) {
            debug('processing ' + block.tx.length + ' transactions in the block')

            async.forEachLimit(block.tx, 10, function(txid, next) {
                var rawtxt, tx

                async.series({
                    'get the raw tx': function(next) {
                        bitcoin.getRawTransaction(txid, function(err, r) {
                            if (err) return next(err)
                            rawtx = r
                            next()
                        })
                    },

                    'decode the raw transaction': function(next) {
                        bitcoin.decodeRawTransaction(rawtx, function(err, t) {
                            if (err) return next(err)
                            tx = t
                            next()
                        })
                    },

                    'enumerate outputs': function(next) {
                        async.forEachLimit(tx.vout, 10, function(o, next) {
                            if (!o.scriptPubKey) throw new Error('scriptPubKey missing')
                            if (!o.scriptPubKey.addresses) throw new Error('addresses missing')
                            if (o.scriptPubKey.addresses.length !== 1) throw new Error(o.scriptPubKey.addresses.length + ' addresses')
                            var address = o.scriptPubKey.addresses[0]

                            var client = db()
                            client.query({
                                text: 'SELECT COUNT(*) count FROM btc_deposit_address WHERE address = $1',
                                values: [address]
                            }, function(err, res) {
                                client.end()

                                if (err) return next(err)

                                if (res.rows[0].count === 0) {
                                    return next()
                                }

                                var satoshi = o.value * 1e8

                                client = db()

                                client.query({
                                    text: 'SELECT btc_credit($1, $2, $3) tid',
                                    values: [txid, address, satoshi]
                                }, function(err, res) {
                                    client.end()

                                    if (err) {
                                        if (err.code === '23505') {
                                            console.log('*** duplicate credit, probably from resuming')
                                            return next()
                                        }

                                        return next(err)
                                    }

                                    console.log('credited transaction (internal id ' + res.rows[0].tid + ')')
                                    next()
                                })
                            })
                        }, next)
                    }
                }, next)
            }, next)
        }
    }, function(err) {
        if (err) return cb(err)

        debug('inserting processed block into db')

        var client = db()
        client.query({
            text: 'INSERT INTO btc_block (hash, height) VALUES ($1, $2)',
            values: [blockHash, lastHeight + 1]
        }, function(err, res)  {
            client.end()
            if (err) return cb(err)
            debug('block processing completed')
            cb(null, true)
        })
    })
}

if (~process.argv[1].indexOf('bitcoinedge.sync.js')) {
    var cb = function(err, res) {
        if (err) console.error(err)
        if (res) return self(cb)
        setTimeout(process.exit, 500)
    }
    self(cb)
}