postgres-index.js

#!/usr/bin/env node -r panik -r dotenv/config
const assert = require('assert');
const { URL } = require('url');
const redis = require('redis');
const bitcoin = require('bitcoin');
const safync = require('./safync');
const { urlToBitcoinOptions } = require('./utils');
const Promise = require('bluebird');
const { pMemoize } = require('./pmr');
const pgp = require('pg-promise');
const createRedisMemCache = require('p-memoize-redis');

Promise.promisifyAll(redis);

const { BITCOIND_RPC_URL, REDIS_URL, DATABASE_URL } = process.env;

const bitcoinRpc = new bitcoin.Client(urlToBitcoinOptions(new URL(BITCOIND_RPC_URL)));
safync.applyTo(bitcoinRpc, 'cmd');

const memBitcoinRpcCmdAsync = pMemoize(bitcoinRpc.cmdAsync, { cache: createRedisMemCache(REDIS_URL, 'drivenetRpc') });
const hexToBuffer = _ => new Buffer(_, 'hex');

function* applyVout(t, block, tx, vout, index) {
  yield t.none(`insert into vout (tx_hash, n, script_pub_key) values ($/txHash/, $/n/, $/scriptPubKey/)`, {
    txHash: hexToBuffer(tx.hash),
    n: index,
    scriptPubKey: vout.scriptPubKey,
  });
}

function* undoVout(t, block, tx, vout, index) {
  yield t.none(`delete from vout where tx_hash = $/txHash/ and n = $/n/`, {
    txHash: hexToBuffer(tx.hash),
    n: index,
  });
}

function* applyVin(t, block, tx, vin, index) {
  yield t.none(
    `insert into vin (tx_hash, n, coinbase, txid, vout, script_sig) values ($/txHash/, $/n/, $/coinbase/, $/txid/, $/vout/, $/scriptSig/)`,
    {
      txHash: hexToBuffer(tx.hash),
      n: index,
      coinbase: vin.coinbase ? hexToBuffer(vin.coinbase) : null,
      txid: vin.txid,
      vout: vin.vout,
      scriptSig: vin.scriptSig,
    }
  );
}

function* undoVin(t, block, tx, vin, index) {
  yield t.none(`delete from vin where tx_hash = $/txHash/ and n = $/n/`, {
    txHash: hexToBuffer(tx.hash),
    n: index,
  });
}

function* applyTx(t, block, tx) {
  yield t.none(`insert into tx (hash) values ($/hash/)`, { hash: hexToBuffer(tx.hash) });

  for (let index = 0; index < tx.vin.length; index++) {
    const vin = tx.vin[index];
    yield* applyVin(t, block, tx, vin, index);
  }

  for (let index = 0; index < tx.vout.length; index++) {
    const vout = tx.vout[index];
    yield* applyVout(t, block, tx, vout, index);
  }

  yield t.none(`insert into block_tx (tx_hash, block_hash) values ($/txHash/, $/blockHash/)`, {
    txHash: hexToBuffer(tx.hash),
    blockHash: hexToBuffer(block.hash),
  });
}

function* undoTx(t, block, tx) {
  yield t.none(`delete from block_tx where tx_hash = $/txHash/ and block_hash = $/blockHash/`, {
    txHash: hexToBuffer(tx.hash),
    blockHash: hexToBuffer(block.hash),
  });

  for (let index = tx.vout.length - 1; index >= 0; index--) {
    const vout = tx.vout[index];
    yield* undoVout(t, block, tx, vout, index);
  }

  for (let index = tx.vin.length - 1; index >= 0; index--) {
    const vin = tx.vin[index];
    yield* undoVin(t, block, tx, vin, index);
  }

  yield t.none(`delete from tx where hash = $/hash/`, {
    hash: hexToBuffer(tx.hash),
  });
}

function* applyBlock(t, block) {
  yield t.none(`insert into block (hash, height) values ($/hash/, $/height/)`, {
    hash: hexToBuffer(block.hash),
    height: block.height,
  });

  for (const tx of block.tx) {
    yield* applyTx(t, block, tx);
  }
}

function* undoBlock(t, block) {
  for (const tx of block.tx.slice().reverse()) {
    yield* undoTx(t, block, tx);
  }

  yield t.none(`delete from block where hash = $/hash/`, {
    hash: hexToBuffer(block.hash),
  });
}

const main = async () => {
  // TODO: detect re-org
  console.log('Starting');

  const db = pgp()(DATABASE_URL);

  const dbBatchFromIterator = iterator => {
    const jobs = Array.from(iterator);
    console.log(`Executing ${jobs.length} jobs...`);
    return db.task(t => t.batch(jobs));
  };

  const { height: localHeight } = await db.one('select coalesce(max(height), -1) height from block');
  const { blocks: remoteHeight } = await bitcoinRpc.cmdAsync('getblockchaininfo');
  console.log({ localHeight, remoteHeight });

  // TODO: ensure that the remote hash at height is same as local hash
  // otherwise need to reorg back until thats the case

  for (let height = 10000; height <= 10010; height++) {
    console.log({ height });
    const blockHash = await bitcoinRpc.cmdAsync('getblockhash', height);
    const block = await memBitcoinRpcCmdAsync('getblock', blockHash, 2);

    await db.task(t => t.batch(Array.from(applyBlock(t, block))));
  }

  for (let height = 10010; height >= 10000; height--) {
    console.log({ height });
    const blockHash = await bitcoinRpc.cmdAsync('getblockhash', height);
    const block = await memBitcoinRpcCmdAsync('getblock', blockHash, 2);

    await db.task(t => t.batch(Array.from(undoBlock(t, block))));
  }

  console.log({ localHeight, remoteHeight });
};

main().then(process.exit);