diff --git a/lib/tx-batching.js b/lib/tx-batching.js index 0e7ce6ee..a2cc3b42 100644 --- a/lib/tx-batching.js +++ b/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 }