add new getChaintipHeight() method to remote importer and data sources

lib/indexer-rpc/rpc-client.js

@@ -181,14 +181,14 @@ class RpcClient {
                 throw new Error(JSON.stringify(parsed.error))
               // Add the parsed reponse to the array of responses
               if (batched) {
-                responses = parsed.map(p => { return {idxAddr: p.id, txs: p.result} })
+                responses = parsed.map(p => { return {id: p.id, response: p.result} })
               } else {
-                responses.push({idxAddr: parsed.id, txs: parsed.result})
+                responses.push({id: parsed.id, response: parsed.result})
               }
               // Reset the response
               response = ''
               // If all responses have been received
-              // close the connection 
+              // close the connection
               if (responses.length == data.length)
                 conn.end()
             } catch (err) {

lib/remote-importer/bitcoind-wrapper.js

@@ -6,6 +6,7 @@
 
 const bitcoin = require('bitcoinjs-lib')
 const RpcClient = require('../bitcoind-rpc/rpc-client')
+const rpcLatestBlock = require('../bitcoind-rpc/latest-block')
 const Logger = require('../logger')
 const network = require('../bitcoin/network')
 const activeNet = network.network
@@ -77,7 +78,7 @@ class BitcoindWrapper extends Wrapper {
         txids: []
       }
     }
-    
+
     return ret
   }
 
@@ -113,7 +114,7 @@ class BitcoindWrapper extends Wrapper {
     }
 
     const aRet = Object.values(ret)
-    
+
     for (let i in aRet) {
       if (filterAddr && aRet[i].ntx > keys.addrFilterThreshold) {
         Logger.info(`Importer : Import of ${aRet[i].address} rejected (too many transactions - ${aRet[i].ntx})`)
@@ -124,6 +125,15 @@ class BitcoindWrapper extends Wrapper {
     return aRet
   }
 
+  /**
+   * Retrieve the height of the chaintip for the remote source
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    return {'chainTipHeight': rpcLatestBlock.height}
+  }
+
 }
 
 module.exports = BitcoindWrapper

lib/remote-importer/esplora-wrapper.js

@@ -37,7 +37,7 @@ class EsploraWrapper extends Wrapper {
       json: true,
       timeout: 15000
     }
-    
+
     // Sets socks proxy agent if required
     if (keys.indexer.socks5Proxy != null)
       params['agent'] = this.socksProxyAgent
@@ -123,6 +123,19 @@ class EsploraWrapper extends Wrapper {
     return ret
   }
 
+  /**
+   * Retrieve the height of the chaintip for the remote source
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    let chainTipHeight = null
+    const result = await this._get(`/api/blocks/tip/height`)
+    if (result != null)
+      chainTipHeight = parseInt(result)
+    return {'chainTipHeight': chainTipHeight}
+  }
+
 }
 
 // Esplora returns a max of 25 txs per page

lib/remote-importer/local-indexer-wrapper.js

@@ -16,7 +16,7 @@ const Wrapper = require('./wrapper')
 /**
  * Wrapper for a local indexer
  * Currently supports indexers
- * providing a RPC API compliant 
+ * providing a RPC API compliant
  * with a subset of the electrum protocol
  */
 class LocalIndexerWrapper extends Wrapper {
@@ -64,7 +64,7 @@ class LocalIndexerWrapper extends Wrapper {
       scriptHash
     )
 
-    for (let r of results.txs) {
+    for (let r of results.response) {
       ret.txids.push(r.tx_hash)
       ret.ntx++
     }
@@ -77,7 +77,7 @@ class LocalIndexerWrapper extends Wrapper {
         txids: []
       }
     }
-    
+
     return ret
   }
 
@@ -109,8 +109,8 @@ class LocalIndexerWrapper extends Wrapper {
       : await this.client.sendRequests(commands)
 
     for (let r of results) {
-      const addr = addresses[r.idxAddr]
-      const txids = r.txs.map(t => t.tx_hash)
+      const addr = addresses[r.id]
+      const txids = r.response.map(t => t.tx_hash)
 
       ret[addr] = {
         address: addr,
@@ -131,12 +131,31 @@ class LocalIndexerWrapper extends Wrapper {
     return aRet
   }
 
+  /**
+   * Retrieve the height of the chaintip for the remote source
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    let chainTipHeight = null
+    const result = await this.client.sendRequest(
+      LocalIndexerWrapper.HEADERS_SUBSCRIBE_RPC_CMD,
+      null
+    )
+    if (result != null && result['response'] != null && result['response']['height'] != null)
+      chainTipHeight = parseInt(result['response']['height'])
+    return {'chainTipHeight': chainTipHeight}
+  }
 }
 
 /**
  * Get history RPC command (Electrum protocol)
  */
 LocalIndexerWrapper.GET_HISTORY_RPC_CMD = 'blockchain.scripthash.get_history'
+/**
+ * Get history RPC command (Electrum protocol)
+ */
+LocalIndexerWrapper.HEADERS_SUBSCRIBE_RPC_CMD = 'blockchain.headers.subscribe'
 
 
 module.exports = LocalIndexerWrapper

lib/remote-importer/oxt-wrapper.js

@@ -20,7 +20,7 @@ class OxtWrapper extends Wrapper {
    * Constructor
    */
   constructor(url) {
-    super(url, keys.indexer.socks5Proxy)    
+    super(url, keys.indexer.socks5Proxy)
   }
 
   /**
@@ -55,7 +55,7 @@ class OxtWrapper extends Wrapper {
     // Try to retrieve more txs than the 1000 managed by the backend
     const uri = `/addresses/${address}/txids?count=${keys.addrFilterThreshold + 1}`
     const result = await this._get(uri)
-    
+
     const ret = {
       address: address,
       ntx: result.count,
@@ -109,6 +109,19 @@ class OxtWrapper extends Wrapper {
     return ret
   }
 
+  /**
+   * Retrieve the height of the chaintip for the remote source
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    let chainTipHeight = null
+    const result = await this._get(`/lastblock`)
+    if (result != null && result['data'].length == 1)
+      chainTipHeight = parseInt(result['data'][0]['height'])
+    return {'chainTipHeight': chainTipHeight}
+  }
+
 }
 
 module.exports = OxtWrapper

lib/remote-importer/remote-importer.js

@@ -25,7 +25,7 @@ if (network.key == 'bitcoin') {
 
 
 /**
- * A singleton providing tools 
+ * A singleton providing tools
  * for importing HD and loose addresses from remote sources
  */
 class RemoteImporter {
@@ -81,7 +81,7 @@ class RemoteImporter {
 
       if (!txParents[txid])
         txParents[txid] = []
-      
+
       for (let i in tx.inputs) {
         const input = tx.inputs[i]
         let prev = input.outpoint.txid
@@ -109,13 +109,13 @@ class RemoteImporter {
    * Import a list of transactions associated to a list of addresses
    * @param {object[]} addresses - array of addresses objects
    * @param {object[]} txns - array of transaction objects
-   * @returns {Promise} 
+   * @returns {Promise}
    */
   async _importTransactions(addresses, txns) {
     const addrIdMap = await db.getAddressesIds(addresses)
 
-    // The transactions array must be topologically ordered, such that 
-    // entries earlier in the array MUST NOT depend upon any entry later 
+    // The transactions array must be topologically ordered, such that
+    // entries earlier in the array MUST NOT depend upon any entry later
     // in the array.
     const txMaps = this._processTxsRelations(txns)
     const txOrdered = util.topologicalOrdering(txMaps.txParents, txMaps.txChildren)
@@ -161,7 +161,7 @@ class RemoteImporter {
     if (gapLimit
       && ((keys.indexer.active == 'local_bitcoind')
         || (keys.indexer.active == 'local_indexer'))
-    ) { 
+    ) {
       gaps = [gapLimit, gapLimit]
     }
 
@@ -212,12 +212,12 @@ class RemoteImporter {
    * 4. Set u = highest chain index of used address, go to 1
    * 5. Store all in database
    *
-   * @returns {object} returns 
+   * @returns {object} returns
    *  {
    *    addresses: [{address, chain, index}],
    *    transactions: [{
-   *      txid, 
-   *      version, 
+   *      txid,
+   *      version,
    *      locktime,
    *      created,  // if known
    *      block: 'abcdef',  // if confirmed
@@ -331,7 +331,7 @@ class RemoteImporter {
       return true
 
     Logger.info(`Importer : Importing ${addresses.join(',')}`)
-    
+
     try {
       const scanTx = []
       const results = await this.sources.getAddresses(addresses, filterAddr)
@@ -373,7 +373,7 @@ class RemoteImporter {
 
       if (N > 0)
         Logger.info(`Importer :  Imported ${N} addresses in ${ts}s (${(dt/N).toFixed(0)} ms/addr)`)
-      
+
       for (let address of addresses)
         delete this.importing[address]
 
@@ -405,7 +405,7 @@ class RemoteImporter {
         const filteredTxs = txs.filter(tx => (tx.block && tx.block.hash == block.blockHash))
         if (filteredTxs.length > 0) {
           const txids = filteredTxs.map(tx => tx.txid)
-          // Asynchronous confirmations 
+          // Asynchronous confirmations
           db.confirmTransactions(txids, block.blockID)
         }
       }
@@ -430,12 +430,12 @@ class RemoteImporter {
         }
       }
       await db.addOutputs(outputs)
-      
+
       // Store the inputs in db
       const inputs = []
       const spent = {}
 
-      // Get any outputs spent by the inputs of this transaction, 
+      // Get any outputs spent by the inputs of this transaction,
       // add those database outIDs to the corresponding inputs, and store.
       let outpoints = []
       for (let tx of txs)
@@ -465,6 +465,15 @@ class RemoteImporter {
     }
   }
 
+  /**
+   * Retrieve the height of the chaintip for the remote source
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    return this.sources.getChainTipHeight()
+  }
+
 }
 
 module.exports = new RemoteImporter()

lib/remote-importer/sources.js

@@ -38,7 +38,7 @@ class Sources {
 
     try {
       const result = await this.source.getAddress(address, filterAddr)
-      
+
       if (result.ntx)
         ret.ntx = result.ntx
       else if (result.txids)
@@ -81,6 +81,22 @@ class Sources {
     }
   }
 
+  /**
+   * Retrieve the height of the chaintip
+   * @returns {Promise} returns an object
+   *    {chainTipHeight: <chaintip_height>}
+   */
+  async getChainTipHeight() {
+    let ret = {'chainTipHeight': null}
+    try {
+      ret = await this.source.getChainTipHeight()
+    } catch(e) {
+      Logger.error(e, `Importer : Sources.getChainTipHeight() : Error while retrieving the chaintip`)
+    } finally {
+      return ret
+    }
+  }
+
 }
 
 module.exports = Sources