fix: batching db transactions

lib/tx-batching.js

@@ -6,88 +6,81 @@ const BN = require('./bn')
 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 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)
+      if (_.isNil(batch)) {
+        return db.tx(t => {
+          const newBatchId = uuid.v4()
+          const q1 = t.none(`INSERT INTO transaction_batches (id, crypto_code) VALUES ($1, $2)`, [newBatchId, tx.cryptoCode])
+          const q2 = t.none(sql2, [newBatchId, tx.id])
+
+          return t.batch([q1, q2])
+        })
+      }
+      return db.none(sql2, [batch.id, tx.id])
     })
-    .then(batch => db.none(sql2, [batch.id, tx.id]))
 }
 
-function closeTransactionBatch (t, batch) {
+function closeTransactionBatch (batch) {
   const sql = `UPDATE transaction_batches SET status='ready', closed_at=now() WHERE id=$1`
 
-  return t.none(sql, [batch.id])
+  return db.none(sql, [batch.id])
 }
 
-function confirmSentBatch (t, batch, tx) {
-  return t.none(`UPDATE transaction_batches SET status='sent', error_message=NULL WHERE id=$1`, [batch.id])
-    .then(() =>
-      t.none(`UPDATE cash_in_txs SET tx_hash=$1, fee=$2, send=true, send_confirmed=true, send_time=now(), send_pending=false, error=NULL, error_code=NULL WHERE batch_id=$3`, [tx.txid, tx.fee.toString(), batch.id])
-    )
+function confirmSentBatch (batch, tx) {
+  return db.tx(t => {
+    const q1 = t.none(`UPDATE transaction_batches SET status='sent', error_message=NULL WHERE id=$1`, [batch.id])
+    const q2 = t.none(`UPDATE cash_in_txs SET tx_hash=$1, fee=$2, send=true, send_confirmed=true, send_time=now(), send_pending=false, error=NULL, error_code=NULL WHERE batch_id=$3`, [tx.txid, tx.fee.toString(), batch.id])
+
+    return t.batch([q1, q2])
+  })
 }
 
-function setErroredBatch (t, batch, errorMsg) {
+function setErroredBatch (batch, errorMsg) {
   const sql = `UPDATE transaction_batches SET status='failed', error_message=$1 WHERE id=$2`
 
-  return t.none(sql, [errorMsg, batch.id])
+  return db.none(sql, [errorMsg, batch.id])
 }
 
-function getBatchTransactions (t, batch) {
+function getBatchTransactions (batch) {
   const sql = `SELECT * FROM cash_in_txs WHERE batch_id=$1`
-  return t.manyOrNone(sql, [batch.id])
+  return db.manyOrNone(sql, [batch.id])
 }
 
-function getBatchesByStatus (t, statuses) {
+function getBatchesByStatus (statuses) {
   const sql = `SELECT *, EXTRACT(EPOCH FROM (now() - created_at)) as time_elapsed FROM transaction_batches WHERE status in ($1^)`
 
-  return t.manyOrNone(sql, [_.map(pgp.as.text, statuses).join(',')])
+  return db.manyOrNone(sql, [_.map(pgp.as.text, statuses).join(',')])
 }
 
-function submitBatch (t, settings, batch) {
-  getBatchTransactions(t, batch)
+function submitBatch (settings, batch) {
+  getBatchTransactions(batch)
     .then(txs => {
       wallet.sendCoinsBatch(settings, txs, batch.crypto_code)
-        .then(res => confirmSentBatch(t, batch, res))
-        .catch(err => setErroredBatch(t, batch, err.message))
+        .then(res => confirmSentBatch(batch, res))
+        .catch(err => setErroredBatch(batch, err.message))
     })
 }
 
 function processBatches (settings, lifecycle) {
-  const transaction = t => {
-    getBatchesByStatus(t, ['open', 'failed', 'ready'])
-      .then(batches => {
-        _.each(batch => {
-          const elapsedMS = batch.time_elapsed * 1000
+  return getBatchesByStatus(['open'])
+    .then(batches => {
+      _.each(batch => {
+        const elapsedMS = batch.time_elapsed * 1000
 
-          if (elapsedMS >= lifecycle) {
-            return closeTransactionBatch(t, batch)
-              .then(() => submitBatch(t, settings, batch))
-          }
-        }, batches)
-      })
-  }
-
-  return db.tx(
-    new pgp.txMode.TransactionMode({ tiLevel: pgp.txMode.isolationLevel }),
-    transaction
-  )
+        if (elapsedMS >= lifecycle) {
+          return closeTransactionBatch(batch)
+            .then(() => submitBatch(settings, batch))
+        }
+      }, batches)
+    })
 }
 
 module.exports = {
-  createTransactionBatch,
-  closeTransactionBatch,
   addTransactionToBatch,
   processBatches
 }