Lots of development

lib/cash-out-tx.js

@@ -1,15 +1,29 @@
 const _ = require('lodash/fp')
 const pgp = require('pg-promise')()
+
 const db = require('./db')
 const BN = require('./bn')
 const billMath = require('./bill-math')
+const T = require('./time')
+const logger = require('./logger')
+const plugins = require('./plugins')
 
-module.exports = {post}
+module.exports = {
+  post,
+  monitorLiveIncoming,
+  monitorStaleIncoming,
+  monitorUnnotified
+}
 
 const mapValuesWithKey = _.mapValues.convert({cap: false})
 
 const UPDATEABLE_FIELDS = ['txHash', 'status', 'dispensed', 'notified', 'redeem',
-  'phone', 'error', 'confirmationTime']
+  'phone', 'error', 'confirmationTime', 'swept']
+
+const STALE_INCOMING_TX_AGE = T.week
+const STALE_LIVE_INCOMING_TX_AGE = 10 * T.minutes
+const MAX_NOTIFY_AGE = 2 * T.days
+const MIN_NOTIFY_AGE = 5 * T.minutes
 
 function post (tx, pi) {
   const TransactionMode = pgp.txMode.TransactionMode
@@ -47,7 +61,6 @@ function diff (oldTx, newTx) {
   let updatedTx = {}
 
   UPDATEABLE_FIELDS.forEach(fieldKey => {
-    console.log('DEBUG80: %j', [oldTx[fieldKey], newTx[fieldKey]])
     if (oldTx && _.isEqualWith(nilEqual, oldTx[fieldKey], newTx[fieldKey])) return
 
     // We never null out an existing field
@@ -122,8 +135,12 @@ function convertBigNumFields (obj) {
   return _.mapKeys(convertKey, mapValuesWithKey(convert, obj))
 }
 
+function convertField (key) {
+  return _.snakeCase(key)
+}
+
 function toDb (tx) {
-  const massager = _.flow(convertBigNumFields, mapDispense, _.omit(['direction', 'bills']), _.mapKeys(_.snakeCase))
+  const massager = _.flow(convertBigNumFields, mapDispense, _.omit(['direction', 'bills']), _.mapKeys(convertField))
   return massager(tx)
 }
 
@@ -148,17 +165,32 @@ function update (tx, changes) {
   .then(() => newTx)
 }
 
+function nextHd (isHd, tx) {
+  console.log('DEBUG160: %s', isHd)
+  if (!isHd) return Promise.resolve(tx)
+  console.log('DEBUG161: %s', isHd)
+
+  return db.one("select nextval('hd_indices_seq') as hd_index")
+  .then(row => _.set('hdIndex', row.hd_index, tx))
+}
+
 function preProcess (tx, newTx, pi) {
   if (!tx) {
-    return pi.newAddress(newTx)
-    .then(_.set('toAddress', _, newTx))
+    return pi.isHd(newTx)
+    .then(isHd => nextHd(isHd, newTx))
+    .then(newTxHd => {
+      return pi.newAddress(newTxHd)
+      .then(_.set('toAddress', _, newTxHd))
+    })
   }
 
-  return Promise.resolve(newTx)
+  return Promise.resolve(updateStatus(tx, newTx))
 }
 
 function postProcess (txVector, pi) {
-  const [, newTx] = txVector
+  const [oldTx, newTx] = txVector
+
+  if (!oldTx) pi.sell(newTx)
 
   if (newTx.dispensed && !newTx.bills) {
     return pi.buildCartridges()
@@ -169,3 +201,75 @@ function postProcess (txVector, pi) {
 
   return Promise.resolve(newTx)
 }
+
+function updateStatus (oldTx, newTx) {
+  const tx = _.set('status', ratchetStatus(oldTx.status, newTx.status), newTx)
+  const isConfirmed = _.includes(tx.status, ['instant', 'confirmed'])
+
+  if (tx.status === oldTx.status || !isConfirmed) return tx
+
+  return _.set('confirmationTime', 'now()^', tx)
+}
+
+function ratchetStatus (oldStatus, newStatus) {
+  const statusOrder = ['notSeen', 'published', 'rejected',
+    'authorized', 'instant', 'confirmed']
+
+  if (oldStatus === newStatus) return oldStatus
+  if (newStatus === 'insufficientFunds') return newStatus
+
+  const idx = Math.max(statusOrder.indexOf(oldStatus), statusOrder.indexOf(newStatus))
+  return statusOrder[idx]
+}
+
+function fetchOpenTxs (statuses, age) {
+  const sql = `select *
+  from cash_out_txs
+  where ((extract(epoch from (now() - created))) * 1000)<$1
+  and status in ($2^)`
+
+  const statusClause = _.map(pgp.as.text, statuses).join(',')
+
+  return db.any(sql, [age, statusClause])
+  .then(rows => rows.map(toObj))
+}
+
+function processTxStatus (tx, settings) {
+  const pi = plugins(settings, tx.deviceId)
+
+  return pi.getStatus(tx)
+  .then(res => _.set('status', res.status, tx))
+  .then(_tx => post(_tx, pi))
+}
+
+function monitorLiveIncoming (settings) {
+  const statuses = ['notSeen', 'published', 'insufficientFunds']
+
+  return fetchOpenTxs(statuses, STALE_LIVE_INCOMING_TX_AGE)
+  .then(txs => Promise.all(txs.map(tx => processTxStatus(tx, settings))))
+  .catch(logger.error)
+}
+
+function monitorStaleIncoming (settings) {
+  const statuses = ['notSeen', 'published', 'authorized', 'instant', 'rejected', 'insufficientFunds']
+
+  return fetchOpenTxs(statuses, STALE_INCOMING_TX_AGE)
+  .then(txs => Promise.all(txs.map(tx => processTxStatus(tx, settings))))
+  .catch(logger.error)
+}
+
+function monitorUnnotified (settings) {
+  const sql = `select *
+  from cash_out_txs
+  where ((extract(epoch from (now() - created))) * 1000)<$1
+  and notified=$2 and dispensed=$3
+  and phone is not null
+  and status in ('instant', 'confirmed')
+  and (redeem=$4 or ((extract(epoch from (now() - created))) * 1000)>$5)`
+
+  const notify = tx => plugins(settings, tx.deviceId).notifyConfirmation(tx)
+  return db.any(sql, [MAX_NOTIFY_AGE, false, false, true, MIN_NOTIFY_AGE])
+  .then(rows => _.map(toObj, rows))
+  .then(txs => Promise.all(txs.map(notify)))
+  .catch(logger.error)
+}