improve cash-in error handling

lib/cash-in-tx.js

@@ -2,13 +2,17 @@ const _ = require('lodash/fp')
 const pgp = require('pg-promise')()
 const db = require('./db')
 const BN = require('./bn')
+const plugins = require('./plugins')
+const logger = require('./logger')
 
 const mapValuesWithKey = _.mapValues.convert({cap: false})
 
-module.exports = {post}
+module.exports = {post, monitorPending}
 
 const UPDATEABLE_FIELDS = ['fee', 'txHash', 'phone', 'error', 'send',
   'cryptoAtoms', 'fiat', 'timedout']
+const PENDING_INTERVAL = '1 day'
+const MAX_PENDING = 10
 
 function post (tx, pi) {
   const TransactionMode = pgp.txMode.TransactionMode
@@ -24,7 +28,10 @@ function post (tx, pi) {
     .then(row => {
       return t.any(sql2, [tx.id])
       .then(billRows => {
-        return upsert(row, tx)
+        const oldTx = toObj(row)
+
+        return preProcess(oldTx, tx, pi)
+        .then(preProcessedTx => upsert(oldTx, preProcessedTx))
         .then(vector => {
           return insertNewBills(billRows, tx)
           .then(newBills => _.concat(vector, [newBills]))
@@ -122,10 +129,7 @@ function insertNewBills (billRows, tx) {
   .then(() => bills)
 }
 
-function upsert (row, tx) {
-  console.log('DEBUG501: %j', row)
-  const oldTx = toObj(row)
-
+function upsert (oldTx, tx) {
   if (!oldTx) {
     console.log('DEBUG500: %j', tx)
     return insert(tx)
@@ -162,24 +166,81 @@ function registerTrades (pi, txVector) {
   _.forEach(bill => pi.buy(bill), newBills)
 }
 
+function logAction (rec, tx) {
+  const action = {
+    tx_id: tx.id,
+    action: 'sendCoins',
+    error: rec.error,
+    error_code: rec.errorCode,
+    tx_hash: rec.txHash,
+    success: rec.sendConfirmed === true
+  }
+
+  const sql = pgp.helpers.insert(action, null, 'cash_in_actions')
+
+  return db.none(sql)
+  .then(_.constant(rec))
+}
+
 function postProcess (txVector, pi) {
   const [oldTx, newTx] = txVector
 
   registerTrades(pi, txVector)
 
-  if (newTx.send && !oldTx.send) {
+  const isClearToSend = newTx.send &&
+  !oldTx.sendPending &&
+  !oldTx.sendConfirmed
+
+  if (isClearToSend) {
     return pi.sendCoins(newTx)
     .then(txHash => ({
       txHash,
       sendConfirmed: true,
-      sendTime: 'now()^'
+      sendTime: 'now()^',
+      sendPending: false,
+      error: null,
+      errorCode: null
     }))
     .catch(err => ({
       sendTime: 'now()^',
       error: err.message,
-      errorCode: err.name
+      errorCode: err.name,
+      sendPending: false
     }))
+    .then(r => logAction(r, newTx))
   }
 
   return Promise.resolve({})
 }
+
+function preProcess (oldTx, newTx, pi) {
+  return new Promise(resolve => {
+    if (!oldTx) return resolve(newTx)
+    if (newTx.send && !oldTx.send) return resolve(_.set('sendPending', true, newTx))
+    return resolve(newTx)
+  })
+}
+
+function monitorPending (settings) {
+  const sql = `select * from cash_in_txs
+  where created > now() - interval $1
+  and send
+  and not send_confirmed
+  and not send_pending
+  and not operator_completed
+  order by created
+  limit $2`
+
+  const processPending = row => {
+    const tx = toObj(row)
+    const pi = plugins(settings, tx.deviceId)
+
+    return post(tx, pi)
+    .catch(logger.error)
+  }
+
+  return db.any(sql, [PENDING_INTERVAL, MAX_PENDING])
+  .then(_.tap(console.log))
+  .then(rows => Promise.all(_.map(processPending, rows)))
+  .catch(logger.error)
+}