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

WIP 1st stage refactor to new DB

Browse source
This commit is contained in:
Josh Harvey 2014-11-20 17:26:17 -05:00
parent dc6740ddf2
commit 0cec3670a9
3 changed files with 96 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

131
lib/postgresql_interface.js
View file

@ -2,7 +2,6 @@
var pg = require('pg');
var async = require('async');
var _ = require('lodash');
var logger = require('./logger');
@ -110,17 +109,16 @@ function silentQuery(client, queryStr, values, cb) {
}
// OPTIMIZE: No need to query bills if tx.fiat and tx.satoshis are set
function billsAndTxs(client, sessionId, currencyCode, deviceFingerprint, cb) {
var billsQuery = 'SELECT COALESCE(SUM(denomination), 0) as fiat, ' +
'COALESCE(SUM(satoshis), 0) AS satoshis ' +
function billsAndTxs(client, sessionId, cb) {
var billsQuery = 'SELECT SUM(denomination) as fiat, ' +
'SUM(satoshis) AS satoshis ' +
'FROM bills ' +
'WHERE transaction_id=$1 AND currency_code=$2 AND device_fingerprint=$3';
var billsValues = [sessionId, currencyCode, deviceFingerprint];
var txQuery = 'SELECT COALESCE(SUM(fiat), 0) AS fiat, ' +
'COALESCE(SUM(satoshis), 0) AS satoshis ' +
'WHERE session_id=$1';
var billsValues = [sessionId];
var txQuery = 'SELECT SUM(fiat) AS fiat, SUM(satoshis) AS satoshis ' +
'FROM transactions ' +
'WHERE session_id=$1 AND currency_code=$2 AND device_fingerprint=$3';
var txValues = billsValues; // They happen to be the same
'WHERE session_id=$1 AND stage=$2';
var txValues = [sessionId, 'partialRequest'];
async.parallel([
async.apply(query, client, billsQuery, billsValues),
@ -155,12 +153,11 @@ function computeSendAmount(tx, totals) {
exports.pendingTxs = function pendingTxs(timeoutMS, cb) {
connect(function(err, client, done) {
var sql = 'SELECT * FROM transactions ' +
'WHERE status=$1 AND ' +
'(NOT incoming OR EXTRACT(EPOCH FROM now() - created > $2) ' +
var sql = 'SELECT * FROM pending_transactions ' +
'WHERE (incoming OR EXTRACT(EPOCH FROM now() - created > $2) ' +
'ORDER BY created ASC';
var timeoutS = timeoutMS / 1000;
var values = ['pending', timeoutS];
var values = [timeoutS];
query(client, sql, values, function(err, results) {
done();
cb(err, results);
@ -169,26 +166,42 @@ exports.pendingTxs = function pendingTxs(timeoutMS, cb) {
};
function removePendingTx(client, tx, cb) {
silentQuery(client, 'DELETE FROM transactions WHERE session_id=$1 AND status=$2',
[tx.txId, 'pending'], cb);
var sql = 'DELETE FROM pending_transactions WHERE session_id=$1';
silentQuery(client, sql, [tx.txId], cb);
}
function maybeInsertTx(client, deviceFingerprint, tx, totals, cb) {
function insertOutgoingTx(client, deviceFingerprint, tx, totals, cb) {
var sendAmount = computeSendAmount(tx, totals);
var status = _.isNumber(tx.fiat) ? 'machineSend' : 'timeout';
var stage = 'partial_request';
var source = tx.fiat ? 'machine' : 'timeout';
var satoshis = sendAmount.satoshis;
var fiat = sendAmount.fiat;
insertTx(client, deviceFingerprint, tx, satoshis, fiat, status, function(err, results) {
// TODO: Don't worry about unique violation
insertTx(client, deviceFingerprint, tx, satoshis, fiat, stage, source,
function(err) {
if (err) return cb(err);
cb(null, {id: results.rows[0].id, satoshisToSend: sendAmount.satoshis});
cb(null, satoshis);
});
}
function insertTx(client, deviceFingerprint, tx, satoshis, fiat, status, cb) {
function insertOutgoingCompleteTx(client, deviceFingerprint, tx, cb) {
// Only relevant for machine source transactions, not timeouts
if (!tx.fiat) return cb();
var stage = 'final_request';
var source = 'machine';
var satoshis = tx.satoshis;
var fiat = tx.fiat;
insertTx(client, deviceFingerprint, tx, satoshis, fiat, stage, source, cb);
}
function insertTx(client, deviceFingerprint, tx, satoshis, fiat, stage,
source, cb) {
var fields = [
'session_id',
'status',
'stage',
'source',
'incoming',
'device_fingerprint',
'to_address',
@ -199,8 +212,9 @@ function insertTx(client, deviceFingerprint, tx, satoshis, fiat, status, cb) {
var values = [
tx.txId,
status,
tx.incoming === false ? false : true,
stage,
source,
tx.incoming,
deviceFingerprint,
tx.toAddress,
satoshis,
@ -211,32 +225,60 @@ function insertTx(client, deviceFingerprint, tx, satoshis, fiat, status, cb) {
query(client, getInsertQuery('transactions', fields, true), values, cb);
}
exports.addPendingTx = function addPendingTx(deviceFingerprint, tx, cb) {
exports.addPendingTx = function addPendingTx(deviceFingerprint, sessionId,
incoming, cb) {
connect(function(err, client, done) {
if (err) return cb(err);
insertTx(client, deviceFingerprint, tx, tx.satoshis, tx.fiat, 'pending',
function(err) {
done();
var fields = ['session_id', 'incoming'];
var sql = getInsertQuery('pending_transactions', fields);
query(client, sql, [sessionId, incoming], function(_err) {
done();
// If pending tx already exists, do nothing
if (err && PG_ERRORS[err.code] !== 'uniqueViolation') {
logger.error(err);
return cb(err);
}
cb();
});
// If pending tx already exists, do nothing
if (_err && PG_ERRORS[err.code] !== 'uniqueViolation')
logger.error(err);
cb(_err);
});
});
};
// Calling function should only send bitcoins if result.satoshisToSend > 0
exports.addTx = function addTx(deviceFingerprint, tx, cb) {
exports.addOutgoingTx = function addOutgoingTx(deviceFingerprint, tx, cb) {
connect(function(err, client, done) {
if (err) return cb(err);
async.waterfall([
async.apply(silentQuery, client, 'BEGIN', null),
async.apply(insertOutgoingCompleteTx, client, deviceFingerprint, tx)
async.apply(removePendingTx, client, tx),
async.apply(billsAndTxs, client, tx.txId, tx.currencyCode, deviceFingerprint),
async.apply(maybeInsertTx, client, deviceFingerprint, tx)
async.apply(insertOutgoingTx, client, deviceFingerprint, tx),
], function(err, satoshisToSend) {
if (err) {
rollback(client, done);
return cb(err);
}
silentQuery(client, 'COMMIT', null, function() {
done();
cb(null, satoshisToSend);
});
});
});
};
function removeIncomingPendingTx(client, tx, status, cb) {
if (status !== 'published') return removePendingTx(client, tx, cb);
cb();
}
exports.addIncomingTx = function addIncomingTx(deviceFingerprint, tx, status,
satoshisReceived, cb) {
connect(function(err, client, done) {
if (err) return cb(err);
async.waterfall([
async.apply(silentQuery, client, 'BEGIN', null),
async.apply(removeOutgoingPendingTx, client, tx, status),
async.apply(insertTx, client, tx, satoshisReceived, 0, 'deposit')
], function(err, result) {
if (err) {
rollback(client, done);
@ -250,19 +292,6 @@ exports.addTx = function addTx(deviceFingerprint, tx, cb) {
});
};
exports.addDigitalTx = function addDigitalTx(dbTxId, err, txHash) {
var keys = ['transaction_id', 'tx_hash', 'error'];
var values = [dbTxId, txHash, err && err.message];
var sql = getInsertQuery('digital_transactions', keys);
connect(function(err, client, done) {
query(client, sql, values, function(_err) {
done(_err);
if (_err) logger.error(_err);
});
});
};
/*
exports.decrementCartridges =
function decrementCartridges(fingerprint, cartridge1, cartridge2, cb) {