Add transaction versioning, tx cancellation (#79)

lib/admin/admin-server.js

@@ -150,6 +150,24 @@ app.get('/api/transactions', (req, res, next) => {
   .catch(next)
 })
 
+app.get('/api/transaction/:id', (req, res, next) => {
+  return transactions.single(req.params.id)
+  .then(r => {
+    if (!r) return res.status(404).send({Error: 'Not found'})
+    return res.send(r)
+  })
+})
+
+app.patch('/api/transaction/:id', (req, res, next) => {
+  if (!req.query.cancel) return res.status(400).send({Error: 'Requires cancel'})
+
+  return transactions.cancel(req.params.id)
+  .then(r => {
+    return res.send(r)
+  })
+  .catch(() => res.status(404).send({Error: 'Not found'}))
+})
+
 app.use((err, req, res, next) => {
   console.error(err)
 
@@ -223,7 +241,6 @@ function establishSocket (ws, token) {
     if (!success) return ws.close(1008, 'Authentication error')
 
     const listener = data => {
-      console.log('DEBUG200: %j', data)
       ws.send(JSON.stringify(data))
     }
 
@@ -239,8 +256,6 @@ function establishSocket (ws, token) {
     }, REAUTHENTICATE_INTERVAL)
 
     socketEmitter.on('message', listener)
-
-    console.log('DEBUG120: %j', token)
     ws.send('Testing123')
   })
 }

lib/admin/transactions.js

@@ -2,6 +2,8 @@ const _ = require('lodash/fp')
 
 const db = require('../db')
 const machineLoader = require('../machine-loader')
+const tx = require('../tx')
+const cashInTx = require('../cash-in-tx')
 
 const NUM_RESULTS = 20
 
@@ -18,21 +20,49 @@ function addNames (txs) {
   })
 }
 
-function batch () {
 const camelize = _.mapKeys(_.camelCase)
+
+function batch () {
   const packager = _.flow(_.flatten, _.orderBy(_.property('created'), ['desc']),
     _.take(NUM_RESULTS), _.map(camelize), addNames)
 
-  const cashInSql = `select 'cashIn' as tx_class, cash_in_txs.*
+  const cashInSql = `select 'cashIn' as tx_class, cash_in_txs.*,
+  ((not send_confirmed) and (created <= now() - interval $1)) as expired
   from cash_in_txs
-  order by created desc limit $1`
+  order by created desc limit $2`
 
   const cashOutSql = `select 'cashOut' as tx_class, cash_out_txs.*
   from cash_out_txs
   order by created desc limit $1`
 
-  return Promise.all([db.any(cashInSql, [NUM_RESULTS]), db.any(cashOutSql, [NUM_RESULTS])])
+  return Promise.all([db.any(cashInSql, [cashInTx.PENDING_INTERVAL, NUM_RESULTS]), db.any(cashOutSql, [NUM_RESULTS])])
   .then(packager)
 }
 
-module.exports = {batch}
+function single (txId) {
+  const packager = _.flow(_.compact, _.map(camelize), addNames)
+
+  const cashInSql = `select 'cashIn' as tx_class,
+  ((not send_confirmed) and (created <= now() - interval $1)) as expired,
+  cash_in_txs.*
+  from cash_in_txs
+  where id=$2`
+
+  const cashOutSql = `select 'cashOut' as tx_class, cash_out_txs.*
+  from cash_out_txs
+  where id=$1`
+
+  return Promise.all([
+    db.oneOrNone(cashInSql, [cashInTx.PENDING_INTERVAL, txId]),
+    db.oneOrNone(cashOutSql, [txId])
+  ])
+  .then(packager)
+  .then(_.head)
+}
+
+function cancel (txId) {
+  return tx.cancel(txId)
+  .then(() => single(txId))
+}
+
+module.exports = {batch, single, cancel}

lib/cash-in-tx.js

@@ -4,13 +4,15 @@ const db = require('./db')
 const BN = require('./bn')
 const plugins = require('./plugins')
 const logger = require('./logger')
-const pp = require('./pp')
+const T = require('./time')
+const E = require('./error')
 
-module.exports = {post, monitorPending}
+const PENDING_INTERVAL = '60 minutes'
-
+const PENDING_INTERVAL_MS = 60 * T.minutes
-const PENDING_INTERVAL = '1 day'
 const MAX_PENDING = 10
 
+module.exports = {post, monitorPending, cancel, PENDING_INTERVAL}
+
 function atomic (machineTx, pi) {
   const TransactionMode = pgp.txMode.TransactionMode
   const isolationLevel = pgp.txMode.isolationLevel
@@ -22,6 +24,8 @@ function atomic (machineTx, pi) {
 
     return t.oneOrNone(sql, [machineTx.id])
     .then(row => {
+      if (row && row.tx_version >= machineTx.txVersion) throw new E.StaleTxError('Stale tx')
+
       return t.any(sql2, [machineTx.id])
       .then(billRows => {
         const dbTx = toObj(row)
@@ -29,10 +33,8 @@ function atomic (machineTx, pi) {
         return preProcess(dbTx, machineTx, pi)
         .then(preProcessedTx => upsert(dbTx, preProcessedTx))
         .then(r => {
-          pp('DEBUG701.5')(r)
           return insertNewBills(billRows, machineTx)
           .then(newBills => _.set('newBills', newBills, r))
-          .then(pp('DEBUG702'))
         })
       })
     })
@@ -44,7 +46,6 @@ function atomic (machineTx, pi) {
 }
 
 function post (machineTx, pi) {
-  console.log('DEBUG700: %j', machineTx)
   return db.tx(atomic(machineTx, pi))
   .then(r => {
     const updatedTx = r.tx
@@ -67,11 +68,11 @@ function isMonotonic (oldField, newField, fieldKey) {
   if (oldField.isBigNumber) return oldField.lte(newField)
   if (_.isNumber(oldField)) return oldField <= newField
 
-  throw new Error(`Unexpected value: ${oldField}`)
+  throw new Error(`Unexpected value [${fieldKey}]: ${oldField}, ${newField}`)
 }
 
 function ensureRatchet (oldField, newField, fieldKey) {
-  const monotonic = ['cryptoAtoms', 'fiat', 'send', 'sendConfirmed', 'operatorCompleted', 'timedout']
+  const monotonic = ['cryptoAtoms', 'fiat', 'cashInFeeCrypto', 'send', 'sendConfirmed', 'operatorCompleted', 'timedout', 'txVersion']
   const free = ['sendPending', 'error', 'errorCode']
 
   if (_.isNil(oldField)) return true
@@ -121,16 +122,11 @@ function toObj (row) {
 
   keys.forEach(key => {
     const objKey = _.camelCase(key)
-    if (key === 'crypto_atoms' || key === 'fiat' || key === 'cash_in_fee') {
+    if (_.includes(key, ['crypto_atoms', 'fiat', 'cash_in_fee', 'cash_in_fee_crypto'])) {
       newObj[objKey] = BN(row[key])
       return
     }
 
-    if (key === 'device_time') {
-      newObj[objKey] = parseInt(row[key], 10)
-      return
-    }
-
     newObj[objKey] = row[key]
   })
 
@@ -200,14 +196,13 @@ function update (tx, changes) {
 }
 
 function registerTrades (pi, newBills) {
-  console.log('DEBUG600: %j', newBills)
   _.forEach(bill => pi.buy(bill), newBills)
 }
 
 function logAction (rec, tx) {
   const action = {
     tx_id: tx.id,
-    action: rec.sendConfirmed ? 'sendCoins' : 'sendCoinsError',
+    action: rec.action || (rec.sendConfirmed ? 'sendCoins' : 'sendCoinsError'),
     error: rec.error,
     error_code: rec.errorCode,
     tx_hash: rec.txHash
@@ -219,13 +214,22 @@ function logAction (rec, tx) {
   .then(_.constant(rec))
 }
 
+function logActionById (action, _rec, txId) {
+  const rec = _.assign(_rec, {action, tx_id: txId})
+  const sql = pgp.helpers.insert(rec, null, 'cash_in_actions')
+
+  return db.none(sql)
+}
+
 function isClearToSend (oldTx, newTx) {
+  const now = Date.now()
+
   return newTx.send &&
-    (!oldTx || (!oldTx.sendPending && !oldTx.sendConfirmed))
+    (!oldTx || (!oldTx.sendPending && !oldTx.sendConfirmed)) &&
+    (newTx.created > now - PENDING_INTERVAL_MS)
 }
 
 function postProcess (r, pi) {
-  console.log('DEBUG701: %j', r)
   registerTrades(pi, r.newBills)
 
   if (isClearToSend(r.dbTx, r.tx)) {
@@ -303,3 +307,22 @@ function monitorPending (settings) {
   .then(rows => Promise.all(_.map(processPending, rows)))
   .catch(logger.error)
 }
+
+function cancel (txId) {
+  const updateRec = {
+    error: 'Operator cancel',
+    error_code: 'operatorCancel',
+    operator_completed: true
+  }
+
+  return Promise.resolve()
+  .then(() => {
+    return pgp.helpers.update(updateRec, null, 'cash_in_txs') +
+      pgp.as.format(' where id=$1', [txId])
+  })
+  .then(sql => db.result(sql, false))
+  .then(res => {
+    if (res.rowCount !== 1) throw new Error('No such tx-id')
+  })
+  .then(() => logActionById('operatorCompleted', {}, txId))
+}

lib/error.js

@@ -22,3 +22,4 @@ function register (errorName) {
 register('BadNumberError')
 register('NoDataError')
 register('InsufficientFundsError')
+register('StaleTxError')

lib/exchange.js

@@ -20,7 +20,6 @@ function fetchExchange (settings, cryptoCode) {
 }
 
 function buy (settings, cryptoAtoms, fiatCode, cryptoCode) {
-  console.log('DEBUG600')
   return fetchExchange(settings, cryptoCode)
   .then(r => r.exchange.buy(r.account, cryptoAtoms, fiatCode, cryptoCode))
 }

lib/plugins.js

@@ -184,7 +184,7 @@ function plugins (settings, deviceId) {
       cryptoCode,
       display: cryptoRec.display,
       minimumTx: BN.max(minimumTx, cashInFee),
-      cashInFee: cashInFee,
+      cashInFee,
       cryptoNetwork
     }
   }
@@ -216,8 +216,6 @@ function plugins (settings, deviceId) {
       const testNets = arr.slice(2 * cryptoCodesCount + 3)
       const coinParams = _.zip(cryptoCodes, testNets)
 
-      console.log('DEBUG200: %j', cryptoCodes)
-      console.log('DEBUG201: %j', coinParams)
       return {
         cassettes,
         rates: buildRates(tickers),
@@ -351,7 +349,6 @@ function plugins (settings, deviceId) {
 
     const market = [fiatCode, cryptoCode].join('')
 
-    console.log('DEBUG505')
     if (!exchange.active(settings, cryptoCode)) return
 
     logger.debug('[%s] Pushing trade: %d', market, cryptoAtoms)
@@ -368,7 +365,6 @@ function plugins (settings, deviceId) {
     const market = [fiatCode, cryptoCode].join('')
 
     const marketTradesQueues = tradesQueues[market]
-    console.log('DEBUG504: %j', marketTradesQueues)
     if (!marketTradesQueues || marketTradesQueues.length === 0) return null
 
     logger.debug('[%s] tradesQueues size: %d', market, marketTradesQueues.length)
@@ -409,7 +405,6 @@ function plugins (settings, deviceId) {
   }
 
   function executeTrades () {
-    console.log('DEBUG500')
     return machineLoader.getMachines()
     .then(devices => {
       const deviceIds = devices.map(device => device.deviceId)
@@ -430,7 +425,6 @@ function plugins (settings, deviceId) {
   }
 
   function executeTradesForMarket (settings, fiatCode, cryptoCode) {
-    console.log('DEBUG501: %s, %j', cryptoCode, exchange.active(settings, cryptoCode))
     if (!exchange.active(settings, cryptoCode)) return
 
     const market = [fiatCode, cryptoCode].join('')
@@ -438,8 +432,6 @@ function plugins (settings, deviceId) {
 
     if (tradeEntry === null || tradeEntry.cryptoAtoms.eq(0)) return
 
-    console.log('DEBUG502')
-
     return executeTradeForType(tradeEntry)
     .catch(err => {
       tradesQueues[market].push(tradeEntry)
@@ -457,8 +449,6 @@ function plugins (settings, deviceId) {
     const tradeEntry = expand(_tradeEntry)
     const execute = tradeEntry.type === 'buy' ? exchange.buy : exchange.sell
 
-    console.log('DEBUG503')
-
     return execute(settings, tradeEntry.cryptoAtoms, tradeEntry.fiatCode, tradeEntry.cryptoCode)
     .then(() => recordTrade(tradeEntry))
   }

lib/plugins/exchange/kraken/kraken.js

@@ -33,7 +33,6 @@ function trade (account, type, cryptoAtoms, fiatCode, cryptoCode) {
     kraken.api('AddOrder', orderInfo, (error, response) => {
       if (error) return reject(error)
 
-      console.log('DEBUG900: %j', response)
       return resolve()
     })
   })

lib/routes.js

@@ -15,6 +15,7 @@ const plugins = require('./plugins')
 const helpers = require('./route-helpers')
 const poller = require('./poller')
 const Tx = require('./tx')
+const E = require('./error')
 
 const argv = require('minimist')(process.argv.slice(2))
 
@@ -108,6 +109,10 @@ function postTx (req, res, next) {
 
     return res.json(tx)
   })
+  .catch(err => {
+    if (err instanceof E.StaleTxError) return res.status(404).json({})
+    throw err
+  })
   .catch(next)
 }
 
@@ -270,7 +275,11 @@ app.get('/tx/:id', getTx)
 app.get('/tx', getPhoneTx)
 
 app.use(errorHandler)
-app.use((req, res) => res.status(404).json({error: 'No such route'}))
+app.use((req, res) => {
+  console.log('DEBUG98')
+  console.log(req.route)
+  res.status(404).json({error: 'No such route'})
+})
 
 localApp.get('/pid', (req, res) => {
   const deviceId = req.query.device_id

lib/tx.js

@@ -23,9 +23,13 @@ function massage (tx) {
   const transformDates = r => mapValuesWithKey(transformDate, r)
 
   const mapBN = r => {
-    const update = r.direction === 'cashIn'
+    const update = {
-    ? {cryptoAtoms: BN(r.cryptoAtoms), fiat: BN(r.fiat), cashInFee: BN(r.cashInFee)}
+      cryptoAtoms: BN(r.cryptoAtoms),
-    : {cryptoAtoms: BN(r.cryptoAtoms), fiat: BN(r.fiat)}
+      fiat: BN(r.fiat),
+      cashInFee: BN(r.cashInFee),
+      cashInFeeCrypto: BN(r.cashInFeeCrypto),
+      minimumTx: BN(r.minimumTx)
+    }
 
     return _.assign(r, update)
   }
@@ -35,4 +39,17 @@ function massage (tx) {
   return mapper(tx)
 }
 
-module.exports = {post}
+function cancel (txId) {
+  const promises = [
+    CashInTx.cancel(txId).then(() => true).catch(() => false),
+    CashOutTx.cancel(txId).then(() => true).catch(() => false)
+  ]
+
+  return Promise.all(promises)
+  .then(r => {
+    if (_.some(r)) return
+    throw new Error('No such transaction')
+  })
+}
+
+module.exports = {post, cancel}

migrations/036-add_customers_table.js

@@ -3,8 +3,7 @@ var anonymous = require('../lib/constants').anonymousCustomer
 
 exports.up = function (next) {
   const sql =
-    [
+    [`create table customers (
-    `create table customers ( 
     id uuid PRIMARY KEY,
     phone text unique,
     phone_at timestamptz,
@@ -23,12 +22,12 @@ exports.up = function(next) {
     created timestamptz NOT NULL DEFAULT now() )`,
     `insert into customers (id, name) VALUES ( '${anonymous.uuid}','${anonymous.name}' )`,
     `alter table cash_in_txs add column customer_id uuid references customers (id) DEFAULT '${anonymous.uuid}'`,
-    `alter table cash_out_txs add column customer_id uuid references customers (id) DEFAULT '${anonymous.uuid}'`,
+    `alter table cash_out_txs add column customer_id uuid references customers (id) DEFAULT '${anonymous.uuid}'`
   ]
 
   db.multi(sql, next)
-};
+}
 
 exports.down = function (next) {
-  next();
+  next()
-};
+}

migrations/037-add_compliance_authorizations_table.js

@@ -2,7 +2,7 @@ var db = require('./db')
 
 exports.up = function (next) {
   const sql =
-    [`create type compliance_types  as enum ('manual', 'sanctions', 'sanctions_override') `,
+    [ "create type compliance_types  as enum ('manual', 'sanctions', 'sanctions_override')",
     `create table compliance_authorizations (
     id uuid PRIMARY KEY,
     customer_id uuid REFERENCES customers (id),
@@ -10,10 +10,9 @@ exports.up = function(next) {
     authorized_at timestamptz NOT NULL,
     authorized_by text REFERENCES user_tokens (token) )` ]
 
-
   db.multi(sql, next)
-};
+}
 
 exports.down = function (next) {
-  next();
+  next()
-};
+}

migrations/1503907708756-drop-device-time.js

@@ -0,0 +1,14 @@
+var db = require('./db')
+
+exports.up = function (next) {
+  const sql = [
+    'alter table cash_in_txs drop column device_time',
+    'alter table cash_out_txs drop column device_time'
+  ]
+
+  db.multi(sql, next)
+}
+
+exports.down = function (next) {
+  next()
+}

migrations/1503945570220-add-tx-version.js

@@ -0,0 +1,14 @@
+var db = require('./db')
+
+exports.up = function (next) {
+  const sql = [
+    'alter table cash_in_txs add column tx_version integer not null',
+    'alter table cash_out_txs add column tx_version integer not null'
+  ]
+
+  db.multi(sql, next)
+}
+
+exports.down = function (next) {
+  next()
+}

public/styles.css

@@ -204,6 +204,12 @@ p {
     background-color: #ffffff;
 }
 
+.lamassuAdminTxTable  a {
+    text-decoration: none;
+    color: #5f5f56;
+    border-bottom: 1px solid #37e8d7;
+}
+
 .lamassuAdminTxTable  .lamassuAdminNumberColumn {
     text-align: right;
     width: 10em;
@@ -215,6 +221,10 @@ p {
     font-size: 90%;
 }
 
+.lamassuAdminTxTable  .lamassuAdminTxCancelled {
+    background-color: #efd1d2;
+}
+
 .lamassuAdminTxTable  tbody {
     font-family: Inconsolata, monospace;
     color: #5f5f56;