feat: add transaction batching module

feat: plugin sendCoins batching support
feat: batching processing on poller
feat: mock-wallet batching
fix: bitcoin tx batching
fix: transaction batching db table

lib/plugins.js

@@ -22,6 +22,7 @@ const machineLoader = require('./machine-loader')
 const customers = require('./customers')
 const commissionMath = require('./commission-math')
 const loyalty = require('./loyalty')
+const transactionBatching = require('./tx-batching')
 
 const { cassetteMaxCapacity, CASH_OUT_DISPENSE_READY, CONFIRMATION_CODE } = require('./constants')
 
@@ -277,6 +278,15 @@ function plugins (settings, deviceId) {
   }
 
   function sendCoins (tx) {
+    if (wallet.supportsBatching(settings, tx.cryptoCode)) {
+      return transactionBatching.addTransactionToBatch(tx)
+        .then(() => ({
+          batched: true,
+          sendPending: false,
+          error: null,
+          errorCode: null
+        }))
+    }
     return wallet.sendCoins(settings, tx)
   }
 

lib/plugins/wallet/bitcoind/bitcoind.js

@@ -81,13 +81,18 @@ function sendCoins (account, tx, settings, operatorId, feeMultiplier) {
 
 function sendCoinsBatch (account, txs, cryptoCode) {
   return checkCryptoCode(cryptoCode)
+    .then(() => calculateFeeDiscount(feeMultiplier))
+    .then(newFee => fetch('settxfee', [newFee]))
     .then(() => {
       const txAddressAmountPairs = _.map(tx => [tx.address, tx.cryptoAtoms.shift(-unitScale).toFixed(8)], txs)
       return Promise.all([JSON.stringify(_.fromPairs(txAddressAmountPairs))])
     })
     .then(([obj]) => fetch('sendmany', ['', obj]))
-    .then(res => ({
+    .then((txId) => fetch('gettransaction', [txId]))
-      txid: res.txid
+    .then((res) => _.pick(['fee', 'txid'], res))
+    .then((pickedObj) => ({
+      fee: BN(pickedObj.fee).abs().shift(unitScale).round(),
+      txid: pickedObj.txid
     }))
     .catch(err => {
       if (err.code === -6) throw new E.InsufficientFundsError()
@@ -173,5 +178,6 @@ module.exports = {
   newFunding,
   cryptoNetwork,
   fetchRBF,
-  estimateFee
+  estimateFee,
+  sendCoinsBatch
 }

lib/plugins/wallet/mock-wallet/mock-wallet.js

@@ -1,3 +1,5 @@
+const _ = require('lodash/fp')
+
 const BN = require('../../../bn')
 const E = require('../../../error')
 const { utils: coinUtils } = require('lamassu-coins')
@@ -57,7 +59,29 @@ function sendCoins (account, tx, settings, operatorId) {
   })
 }
 
+<<<<<<< HEAD
 function newAddress (account, info, tx, settings, operatorId) {
+=======
+function sendCoinsBatch (account, txs, cryptoCode) {
+  sendCount = sendCount + txs.length
+  return new Promise((resolve, reject) => {
+    setTimeout(() => {
+      const cryptoSum = _.reduce((acc, value) => acc.add(value.crypto_atoms), BN(0), txs)
+      if (isInsufficient(cryptoSum, cryptoCode)) {
+        console.log('[%s] DEBUG: Mock wallet insufficient funds: %s',
+          cryptoCode, cryptoSum.toString())
+        return reject(new E.InsufficientFundsError())
+      }
+
+      console.log('[%s] DEBUG: Mock wallet sending %s cryptoAtoms in a batch',
+        cryptoCode, cryptoSum.toString())
+      return resolve({ txid: '<txHash>', fee: BN(0) })
+    }, 2000)
+  })
+}
+
+function newAddress () {
+>>>>>>> feat: add transaction batching module
   t0 = Date.now()
   return Promise.resolve('<Fake address, don\'t send>')
 }
@@ -93,6 +117,7 @@ function getStatus (account, tx, requested, settings, operatorId) {
 module.exports = {
   NAME,
   balance,
+  sendCoinsBatch,
   sendCoins,
   newAddress,
   getStatus,

lib/poller.js

@@ -17,6 +17,7 @@ const NodeCache = require('node-cache')
 const util = require('util')
 const db = require('./db')
 const state = require('./middlewares/state')
+const batching = require('./tx-batching')
 
 const INCOMING_TX_INTERVAL = 30 * T.seconds
 const LIVE_INCOMING_TX_INTERVAL = 5 * T.seconds
@@ -31,6 +32,7 @@ const SANCTIONS_INITIAL_DOWNLOAD_INTERVAL = 5 * T.minutes
 const SANCTIONS_UPDATE_INTERVAL = 1 * T.day
 const RADAR_UPDATE_INTERVAL = 5 * T.minutes
 const PRUNE_MACHINES_HEARTBEAT = 1 * T.day
+const TRANSACTION_BATCH_LIFECYCLE = 20 * T.minutes
 
 const CHECK_NOTIFICATION_INTERVAL = 20 * T.seconds
 const PENDING_INTERVAL = 10 * T.seconds
@@ -214,6 +216,7 @@ function doPolling (schema) {
   }
   addToQueue(cashOutTx.monitorUnnotified, UNNOTIFIED_INTERVAL, schema, QUEUE.FAST, settings)
   addToQueue(cashInTx.monitorPending, PENDING_INTERVAL, schema, QUEUE.FAST, settings)
+  addToQueue(batching.processBatches, UNNOTIFIED_INTERVAL, schema, QUEUE.FAST, settings, TRANSACTION_BATCH_LIFECYCLE)
   addToQueue(pi().sweepHd, SWEEP_HD_INTERVAL, schema, QUEUE.FAST, settings)
   addToQueue(pi().pong, PONG_INTERVAL, schema, QUEUE.FAST)
   addToQueue(pi().clearOldLogs, LOGS_CLEAR_INTERVAL, schema, QUEUE.SLOW)

lib/tx-batching.js

@@ -0,0 +1,84 @@
+const _ = require('lodash/fp')
+const pgp = require('pg-promise')()
+const uuid = require('uuid')
+
+const db = require('./db')
+const wallet = require('./wallet')
+
+function createTransactionBatch (cryptoCode) {
+  const sql = `INSERT INTO transaction_batches (id, crypto_code) VALUES ($1, $2) RETURNING *`
+
+  return db.one(sql, [uuid.v4(), cryptoCode])
+}
+
+function closeTransactionBatch (batch) {
+  const sql = `UPDATE transaction_batches SET status='ready', closed_at=now() WHERE id=$1`
+
+  return db.none(sql, [batch.id])
+}
+
+function confirmSentBatch (batch) {
+  const sql = `UPDATE transaction_batches SET status='sent', error_message=NULL WHERE id=$1`
+
+  return db.none(sql, [batch.id])
+}
+
+function setErroredBatch (batch, errorMsg) {
+  const sql = `UPDATE transaction_batches SET status='failed', error_message=$1 WHERE id=$2`
+
+  return db.none(sql, [errorMsg, batch.id])
+}
+
+function addTransactionToBatch (tx) {
+  const sql = `SELECT * FROM transaction_batches WHERE crypto_code=$1 AND status='open' ORDER BY created_at DESC LIMIT 1`
+  const sql2 = `UPDATE cash_in_txs SET batch_id=$1 WHERE id=$2`
+
+  return db.oneOrNone(sql, [tx.cryptoCode])
+    .then(batch => {
+      if (_.isNil(batch))
+        return createTransactionBatch(tx.cryptoCode)
+      return Promise.resolve(batch)
+    })
+    .then(batch => db.none(sql2, [batch.id, tx.id]))
+}
+
+function getBatchTransactions (batch) {
+  const sql = `SELECT * FROM cash_in_txs WHERE batch_id=$1`
+  return db.manyOrNone(sql, [batch.id])
+}
+
+function getBatchesByStatus (statuses) {
+  const sql = `SELECT *, EXTRACT(EPOCH FROM (now() - created_at)) as time_elapsed FROM transaction_batches WHERE status in ($1^)`
+
+  return db.manyOrNone(sql, [_.map(pgp.as.text, statuses).join(',')])
+}
+
+function submitBatch (settings, batch) {
+  getBatchTransactions(batch)
+    .then(txs => {
+      wallet.sendCoinsBatch(settings, txs, batch.crypto_code)
+        .then(() => confirmSentBatch(batch))
+        .catch(err => setErroredBatch(batch, err.message))
+    })
+}
+
+function processBatches (settings, lifecycle) {
+  getBatchesByStatus(['open', 'failed'])
+    .then(batches => {
+      _.each(batch => {
+        const elapsedMS = batch.time_elapsed * 1000
+
+        if (elapsedMS >= lifecycle) {
+          return closeTransactionBatch(batch)
+            .then(() => submitBatch(settings, batch))
+        }
+      }, batches)
+    })
+}
+
+module.exports = {
+  createTransactionBatch,
+  closeTransactionBatch,
+  addTransactionToBatch,
+  processBatches
+}

lib/wallet.js

@@ -75,6 +75,24 @@ function sendCoins (settings, tx) {
     })
 }
 
+function sendCoinsBatch (settings, txs, cryptoCode) {
+  return fetchWallet(settings, cryptoCode)
+    .then(r => {
+      return r.wallet.sendCoinsBatch(r.account, txs, cryptoCode)
+        .then(res => {
+          mem.clear(module.exports.balance)
+          return res
+        })
+    })
+    .catch(err => {
+      if (err.name === INSUFFICIENT_FUNDS_NAME) {
+        throw httpError(INSUFFICIENT_FUNDS_NAME, INSUFFICIENT_FUNDS_CODE)
+      }
+
+      throw err
+    })
+}
+
 function newAddress (settings, info, tx) {
   const walletAddressPromise = fetchWallet(settings, info.cryptoCode)
     .then(r => r.wallet.newAddress(r.account, info, tx, settings, r.operatorId))
@@ -235,11 +253,13 @@ const balanceFiltered = mem(_balance, {
 module.exports = {
   balance,
   sendCoins,
+  sendCoinsBatch,
   newAddress,
   getStatus,
   isStrictAddress,
   sweep,
   isHd,
   newFunding,
-  cryptoNetwork
+  cryptoNetwork,
+  supportsBatching
 }

migrations/1621556014244-add-btc-tx-batching.js

@@ -2,7 +2,7 @@ var db = require('./db')
 
 exports.up = function (next) {
   var sql = [
-    `CREATE TYPE transaction_batch_status AS ENUM('open', 'failed', 'sent')`,
+    `CREATE TYPE transaction_batch_status AS ENUM('open', 'ready', 'failed', 'sent')`,
     `CREATE TABLE transaction_batches (
       id UUID PRIMARY KEY,
       crypto_code TEXT NOT NULL,
@@ -11,7 +11,7 @@ exports.up = function (next) {
       closed_at TIMESTAMPTZ,
       error_message TEXT
     )`,
-    `ALTER TABLE cash_in_txs ADD COLUMN batch_id REFERENCES transaction_batches(id)`
+    `ALTER TABLE cash_in_txs ADD COLUMN batch_id UUID REFERENCES transaction_batches(id)`
   ]
 
   db.multi(sql, next)