lamassu/lamassu-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

WIPP

Browse source
This commit is contained in:
Josh Harvey 2016-12-08 17:56:50 +02:00
parent df5d9cac89
commit a375adb8b9
3 changed files with 516 additions and 496 deletions
Download patch file Download diff file Expand all files Collapse all files

158
lib/plugins.js
View file

@ -10,7 +10,6 @@ const logger = require('./logger')
const notifier = require('./notifier') const notifier = require('./notifier')
const T = require('./time') const T = require('./time')
const configManager = require('./config-manager') const configManager = require('./config-manager')
const settingsLoader = require('./settings-loader')
const ticker = require('./ticker') const ticker = require('./ticker')
const wallet = require('./wallet') const wallet = require('./wallet')
const exchange = require('./exchange') const exchange = require('./exchange')
@ -38,8 +37,8 @@ const coins = {
let alertFingerprint = null let alertFingerprint = null
let lastAlertTime = null let lastAlertTime = null
function buildRates (deviceId, tickers) { function plugins (settings) {
const settings = settingsLoader.settings() function buildRates (deviceId, tickers) {
const config = configManager.machineScoped(deviceId, settings.config) const config = configManager.machineScoped(deviceId, settings.config)
const cryptoCodes = config.cryptoCurrencies const cryptoCodes = config.cryptoCurrencies
const cashOut = config.cashOutEnabled const cashOut = config.cashOutEnabled
@ -60,10 +59,9 @@ function buildRates (deviceId, tickers) {
}) })
return rates return rates
} }
function buildBalances (deviceId, balanceRecs) { function buildBalances (deviceId, balanceRecs) {
const settings = settingsLoader.settings()
const config = configManager.machineScoped(deviceId, settings.config) const config = configManager.machineScoped(deviceId, settings.config)
const cryptoCodes = config.cryptoCurrencies const cryptoCodes = config.cryptoCurrencies
@ -78,9 +76,9 @@ function buildBalances (deviceId, balanceRecs) {
}) })
return balances return balances
} }
function buildCartridges (cartridges, virtualCartridges, rec) { function buildCartridges (cartridges, virtualCartridges, rec) {
return { return {
cartridges: [ cartridges: [
{ {
@ -94,10 +92,9 @@ function buildCartridges (cartridges, virtualCartridges, rec) {
], ],
virtualCartridges virtualCartridges
} }
} }
function pollQueries (deviceTime, deviceId, deviceRec) { function pollQueries (deviceTime, deviceId, deviceRec) {
const settings = settingsLoader.settings()
const config = configManager.machineScoped(deviceId, settings.config) const config = configManager.machineScoped(deviceId, settings.config)
const fiatCode = config.fiatCurrency const fiatCode = config.fiatCurrency
const cryptoCodes = config.cryptoCurrencies const cryptoCodes = config.cryptoCurrencies
@ -123,11 +120,11 @@ function pollQueries (deviceTime, deviceId, deviceRec) {
balances: buildBalances(deviceId, balances) balances: buildBalances(deviceId, balances)
} }
}) })
} }
// NOTE: This will fail if we have already sent coins because there will be // NOTE: This will fail if we have already sent coins because there will be
// a dbm unique dbm record in the table already. // a dbm unique dbm record in the table already.
function executeTx (deviceId, tx) { function executeTx (deviceId, tx) {
return dbm.addOutgoingTx(deviceId, tx) return dbm.addOutgoingTx(deviceId, tx)
.then(() => wallet.sendCoins(tx.toAddress, tx.cryptoAtoms, tx.cryptoCode)) .then(() => wallet.sendCoins(tx.toAddress, tx.cryptoAtoms, tx.cryptoCode))
.then(txHash => { .then(txHash => {
@ -141,9 +138,9 @@ function executeTx (deviceId, tx) {
txId: tx.id txId: tx.id
})) }))
}) })
} }
function trade (deviceId, rawTrade) { function trade (deviceId, rawTrade) {
// TODO: move this to dbm, too // TODO: move this to dbm, too
// add bill to trader queue (if trader is enabled) // add bill to trader queue (if trader is enabled)
const cryptoCode = rawTrade.cryptoCode const cryptoCode = rawTrade.cryptoCode
@ -165,9 +162,9 @@ function trade (deviceId, rawTrade) {
timestamp: Date.now() timestamp: Date.now()
}) })
}) })
} }
function stateChange (deviceId, deviceTime, rec) { function stateChange (deviceId, deviceTime, rec) {
const event = { const event = {
id: rec.uuid, id: rec.uuid,
deviceId: deviceId, deviceId: deviceId,
@ -176,9 +173,9 @@ function stateChange (deviceId, deviceTime, rec) {
deviceTime: deviceTime deviceTime: deviceTime
} }
return dbm.machineEvent(event) return dbm.machineEvent(event)
} }
function recordPing (deviceId, deviceTime, rec) { function recordPing (deviceId, deviceTime, rec) {
const event = { const event = {
id: uuid.v4(), id: uuid.v4(),
deviceId: deviceId, deviceId: deviceId,
@ -187,13 +184,13 @@ function recordPing (deviceId, deviceTime, rec) {
deviceTime: deviceTime deviceTime: deviceTime
} }
return dbm.machineEvent(event) return dbm.machineEvent(event)
} }
function sendCoins (deviceId, rawTx) { function sendCoins (deviceId, rawTx) {
return executeTx(deviceId, rawTx) return executeTx(deviceId, rawTx)
} }
function cashOut (deviceId, tx) { function cashOut (deviceId, tx) {
const cryptoCode = tx.cryptoCode const cryptoCode = tx.cryptoCode
const serialPromise = wallet.supportsHD const serialPromise = wallet.supportsHD
@ -216,19 +213,17 @@ function cashOut (deviceId, tx) {
.then(() => address) .then(() => address)
}) })
}) })
} }
function dispenseAck (deviceId, tx) { function dispenseAck (deviceId, tx) {
const settings = settingsLoader.settings()
const config = configManager.machineScoped(deviceId, settings.config) const config = configManager.machineScoped(deviceId, settings.config)
const cartridges = [ config.topCashOutDenomination, const cartridges = [ config.topCashOutDenomination,
config.bottomCashOutDenomination ] config.bottomCashOutDenomination ]
return dbm.addDispense(deviceId, tx, cartridges) return dbm.addDispense(deviceId, tx, cartridges)
} }
function fiatBalance (fiatCode, cryptoCode, deviceId) { function fiatBalance (fiatCode, cryptoCode, deviceId) {
const settings = settingsLoader.settings()
const config = configManager.scoped(cryptoCode, deviceId, settings.config) const config = configManager.scoped(cryptoCode, deviceId, settings.config)
return Promise.all([ticker.getRates(fiatCode, cryptoCode), wallet.balance(cryptoCode)]) return Promise.all([ticker.getRates(fiatCode, cryptoCode), wallet.balance(cryptoCode)])
@ -251,14 +246,14 @@ function fiatBalance (fiatCode, cryptoCode, deviceId) {
return {timestamp: balanceRec.timestamp, balance: fiatTransferBalance.round(3).toNumber()} return {timestamp: balanceRec.timestamp, balance: fiatTransferBalance.round(3).toNumber()}
}) })
} }
function processTxStatus (tx) { function processTxStatus (tx) {
return wallet.getStatus(tx.toAddress, tx.cryptoAtoms, tx.cryptoCode) return wallet.getStatus(tx.toAddress, tx.cryptoAtoms, tx.cryptoCode)
.then(res => dbm.updateTxStatus(tx, res.status)) .then(res => dbm.updateTxStatus(tx, res.status))
} }
function notifyConfirmation (tx) { function notifyConfirmation (tx) {
logger.debug('notifyConfirmation') logger.debug('notifyConfirmation')
const phone = tx.phone const phone = tx.phone
@ -271,49 +266,49 @@ function notifyConfirmation (tx) {
return sms.sendMessage(rec) return sms.sendMessage(rec)
.then(() => dbm.updateNotify(tx)) .then(() => dbm.updateNotify(tx))
} }
function monitorLiveIncoming () { function monitorLiveIncoming () {
const statuses = ['notSeen', 'published', 'insufficientFunds'] const statuses = ['notSeen', 'published', 'insufficientFunds']
return dbm.fetchOpenTxs(statuses, STALE_LIVE_INCOMING_TX_AGE) return dbm.fetchOpenTxs(statuses, STALE_LIVE_INCOMING_TX_AGE)
.then(txs => Promise.all(txs.map(processTxStatus))) .then(txs => Promise.all(txs.map(processTxStatus)))
.catch(logger.error) .catch(logger.error)
} }
function monitorIncoming () { function monitorIncoming () {
const statuses = ['notSeen', 'published', 'authorized', 'instant', 'rejected', 'insufficientFunds'] const statuses = ['notSeen', 'published', 'authorized', 'instant', 'rejected', 'insufficientFunds']
return dbm.fetchOpenTxs(statuses, STALE_INCOMING_TX_AGE) return dbm.fetchOpenTxs(statuses, STALE_INCOMING_TX_AGE)
.then(txs => Promise.all(txs.map(processTxStatus))) .then(txs => Promise.all(txs.map(processTxStatus)))
.catch(logger.error) .catch(logger.error)
} }
function monitorUnnotified () { function monitorUnnotified () {
dbm.fetchUnnotifiedTxs(MAX_NOTIFY_AGE, MIN_NOTIFY_AGE) dbm.fetchUnnotifiedTxs(MAX_NOTIFY_AGE, MIN_NOTIFY_AGE)
.then(txs => Promise.all(txs.map(notifyConfirmation))) .then(txs => Promise.all(txs.map(notifyConfirmation)))
.catch(logger.error) .catch(logger.error)
} }
function pong () { function pong () {
db.none('insert into server_events (event_type) values ($1)', ['ping']) db.none('insert into server_events (event_type) values ($1)', ['ping'])
.catch(logger.error) .catch(logger.error)
} }
function pongClear () { function pongClear () {
const sql = `delete from server_events const sql = `delete from server_events
where event_type=$1 where event_type=$1
and created < now() - interval $2` and created < now() - interval $2`
db.none(sql, ['ping', PONG_TTL]) db.none(sql, ['ping', PONG_TTL])
.catch(logger.error) .catch(logger.error)
} }
/* /*
* Trader functions * Trader functions
*/ */
function consolidateTrades (cryptoCode, fiatCode) { function consolidateTrades (cryptoCode, fiatCode) {
const market = [fiatCode, cryptoCode].join('') const market = [fiatCode, cryptoCode].join('')
const marketTradesQueues = tradesQueues[market] const marketTradesQueues = tradesQueues[market]
@ -355,11 +350,9 @@ function consolidateTrades (cryptoCode, fiatCode) {
logger.debug('[%s] consolidated: %j', market, consolidatedTrade) logger.debug('[%s] consolidated: %j', market, consolidatedTrade)
return consolidatedTrade return consolidatedTrade
} }
function executeTrades () {
const settings = settingsLoader.settings()
function executeTrades () {
return dbm.devices() return dbm.devices()
.then(devices => { .then(devices => {
const deviceIds = devices.map(device => device.device_id) const deviceIds = devices.map(device => device.device_id)
@ -377,9 +370,9 @@ function executeTrades () {
return Promise.all(tradesPromises) return Promise.all(tradesPromises)
}) })
.catch(logger.error) .catch(logger.error)
} }
function executeTradesForMarket (settings, fiatCode, cryptoCode) { function executeTradesForMarket (settings, fiatCode, cryptoCode) {
if (!exchange.active(cryptoCode)) return if (!exchange.active(cryptoCode)) return
const market = [fiatCode, cryptoCode].join('') const market = [fiatCode, cryptoCode].join('')
@ -402,10 +395,9 @@ function executeTradesForMarket (settings, fiatCode, cryptoCode) {
if (err.name === 'NoExchangeError') return logger.debug(err.message) if (err.name === 'NoExchangeError') return logger.debug(err.message)
logger.error(err) logger.error(err)
}) })
} }
function sendMessage (rec) { function sendMessage (rec) {
const settings = settingsLoader.settings()
const config = configManager.unscoped(settings.config) const config = configManager.unscoped(settings.config)
let promises = [] let promises = []
@ -413,9 +405,9 @@ function sendMessage (rec) {
if (config.notificationsSMSEnabled) promises.push(sms.sendMessage(rec)) if (config.notificationsSMSEnabled) promises.push(sms.sendMessage(rec))
return Promise.all(promises) return Promise.all(promises)
} }
function sendNoAlerts () { function sendNoAlerts () {
const subject = '[Lamassu] All clear' const subject = '[Lamassu] All clear'
const rec = { const rec = {
sms: { sms: {
@ -427,9 +419,9 @@ function sendNoAlerts () {
} }
} }
return sendMessage(rec) return sendMessage(rec)
} }
function checkNotification () { function checkNotification () {
return notifier.checkStatus() return notifier.checkStatus()
.then(alertRec => { .then(alertRec => {
const currentAlertFingerprint = notifier.alertFingerprint(alertRec) const currentAlertFingerprint = notifier.alertFingerprint(alertRec)
@ -463,10 +455,9 @@ function checkNotification () {
if (results && results.length > 0) logger.debug('Successfully sent alerts') if (results && results.length > 0) logger.debug('Successfully sent alerts')
}) })
.catch(logger.error) .catch(logger.error)
} }
function checkDeviceBalances (deviceId) { function checkDeviceBalances (deviceId) {
const settings = settingsLoader.settings()
const config = configManager.machineScoped(deviceId, settings.config) const config = configManager.machineScoped(deviceId, settings.config)
const cryptoCodes = config.cryptoCurrencies const cryptoCodes = config.cryptoCurrencies
const fiatCode = config.fiatCurrency const fiatCode = config.fiatCurrency
@ -481,9 +472,9 @@ function checkDeviceBalances (deviceId) {
deviceId deviceId
})) }))
}) })
} }
function checkBalances () { function checkBalances () {
return dbm.devices() return dbm.devices()
.then(devices => { .then(devices => {
const deviceIds = devices.map(r => r.device_id) const deviceIds = devices.map(r => r.device_id)
@ -496,19 +487,19 @@ function checkBalances () {
return R.values(R.reduceBy(min, Infinity, toMarket, R.flatten(arr))) return R.values(R.reduceBy(min, Infinity, toMarket, R.flatten(arr)))
}) })
}) })
} }
function startCheckingNotification (config) { function startCheckingNotification (config) {
notifier.init(checkBalances) notifier.init(checkBalances)
checkNotification() checkNotification()
setInterval(checkNotification, CHECK_NOTIFICATION_INTERVAL) setInterval(checkNotification, CHECK_NOTIFICATION_INTERVAL)
} }
function randomCode () { function randomCode () {
return new BigNumber(crypto.randomBytes(3).toString('hex'), 16).shift(-6).toFixed(6).slice(-6) return new BigNumber(crypto.randomBytes(3).toString('hex'), 16).shift(-6).toFixed(6).slice(-6)
} }
function getPhoneCode (phone) { function getPhoneCode (phone) {
const code = argv.mockSms const code = argv.mockSms
? '123' ? '123'
: randomCode() : randomCode()
@ -522,9 +513,9 @@ function getPhoneCode (phone) {
return sms.sendMessage(rec) return sms.sendMessage(rec)
.then(() => code) .then(() => code)
} }
function fetchPhoneTx (phone) { function fetchPhoneTx (phone) {
return dbm.fetchPhoneTxs(phone, TRANSACTION_EXPIRATION) return dbm.fetchPhoneTxs(phone, TRANSACTION_EXPIRATION)
.then(txs => { .then(txs => {
const confirmedTxs = txs.filter(tx => R.contains(tx.status, ['instant', 'confirmed'])) const confirmedTxs = txs.filter(tx => R.contains(tx.status, ['instant', 'confirmed']))
@ -539,9 +530,9 @@ function fetchPhoneTx (phone) {
if (txs.length > 0) return {pending: true} if (txs.length > 0) return {pending: true}
return {} return {}
}) })
} }
function sweepHD (row) { function sweepHD (row) {
const cryptoCode = row.crypto_code const cryptoCode = row.crypto_code
return wallet.sweep(row.hd_serial) return wallet.sweep(row.hd_serial)
@ -552,21 +543,21 @@ function sweepHD (row) {
} }
}) })
.catch(err => logger.error('[%s] Sweep error: %s', cryptoCode, err.message)) .catch(err => logger.error('[%s] Sweep error: %s', cryptoCode, err.message))
} }
function sweepLiveHD () { function sweepLiveHD () {
return dbm.fetchLiveHD() return dbm.fetchLiveHD()
.then(rows => Promise.all(rows.map(sweepHD))) .then(rows => Promise.all(rows.map(sweepHD)))
.catch(err => logger.error(err)) .catch(err => logger.error(err))
} }
function sweepOldHD () { function sweepOldHD () {
return dbm.fetchOldHD() return dbm.fetchOldHD()
.then(rows => Promise.all(rows.map(sweepHD))) .then(rows => Promise.all(rows.map(sweepHD)))
.catch(err => logger.error(err)) .catch(err => logger.error(err))
} }
module.exports = { return {
pollQueries, pollQueries,
trade, trade,
stateChange, stateChange,
@ -584,4 +575,7 @@ module.exports = {
monitorUnnotified, monitorUnnotified,
sweepLiveHD, sweepLiveHD,
sweepOldHD sweepOldHD
}
} }
module.exports = plugins

12
lib/poller.js
View file

@ -10,8 +10,14 @@ const TRADE_INTERVAL = 10 * T.seconds
const PONG_INTERVAL = 10 * T.seconds const PONG_INTERVAL = 10 * T.seconds
const PONG_CLEAR_INTERVAL = 1 * T.day const PONG_CLEAR_INTERVAL = 1 * T.day
function start () { let pi
let pi = plugins
function reload (settings) {
pi = plugins(settings)
}
function start (settings) {
reload(settings)
pi.executeTrades() pi.executeTrades()
pi.pong() pi.pong()
@ -32,4 +38,4 @@ function start () {
setInterval(() => pi.pongClear(), PONG_CLEAR_INTERVAL) setInterval(() => pi.pongClear(), PONG_CLEAR_INTERVAL)
} }
module.exports = {start} module.exports = {start, reload}

20
lib/settings-loader.js
View file

@ -12,6 +12,14 @@ function load (versionId) {
})) }))
} }
function loadLatest (versionId) {
return Promise.all([loadLatestConfig(), loadAccounts()])
.then(([config, accounts]) => ({
config,
accounts
}))
}
function loadConfig (versionId) { function loadConfig (versionId) {
const sql = `select data const sql = `select data
from user_config from user_config
@ -21,6 +29,17 @@ function loadConfig (versionId) {
.then(row => row ? row.data.config : []) .then(row => row ? row.data.config : [])
} }
function loadLatestConfig () {
const sql = `select data
from user_config
where type=$1
order by versionId desc
limit 1`
return db.oneOrNone(sql, ['config'])
.then(row => row ? row.data.config : [])
}
function loadAccounts () { function loadAccounts () {
const toFields = fieldArr => R.fromPairs(R.map(r => [r.code, r.value], fieldArr)) const toFields = fieldArr => R.fromPairs(R.map(r => [r.code, r.value], fieldArr))
const toPairs = r => [r.code, toFields(r.fields)] const toPairs = r => [r.code, toFields(r.fields)]
@ -45,5 +64,6 @@ module.exports = {
settings, settings,
loadConfig, loadConfig,
load, load,
loadLatest,
save save
} }