<commit_before>var pg = require('pg');
var step = require('step');
var ltx = require('ltx');
var errors = require('./errors');

/* ready DB connections */
var pool = [];
/* waiting transaction requests */
var queue = [];

/* at start and when connection died */
function connectDB(config) {
    var db = new pg.Client(config);
    db.connect();

    /* Reconnect in up to 5s */
    db.on('error', function(err) {
	console.error('Postgres: ' + err.message);
	setTimeout(function() {
	    connectDB(config);
	}, Math.ceil(Math.random() * 5000));
	try { db.end(); } catch (e) { /* Alright */ }
    });

    /* wait until connected & authed */
    db.connection.once('readyForQuery', function() {
	dbIsAvailable(db);
    });
}

function dbIsAvailable(db) {
    var cb;
    if ((cb = queue.shift())) {
	/* request was waiting in queue */
	cb(db);
    } else {
	/* no request, put into pool */
	pool.push(db);
    }
}

/* config: { user, database, host, port, poolSize: 4 } */
exports.start = function(config) {
    for(var i = 0; i < (config.poolSize || 4); i++) {
	connectDB(config);
    }
};


exports.transaction = function(cb) {
    var db;
    if ((db = pool.shift())) {
	/* Got one from pool */
	new Transaction(db, cb);
    } else {
	/* Pool was empty, waiting... TODO: limit length, shift first */
	queue.push(function() {
		       exports.transaction(cb);
		   });
    }
};

/**
 * Wraps the postgres-js transaction with our model operations.
 */
function Transaction(db, cb) {
    var that = this;
    this.db = db;

    db.query("BEGIN", [], function(err, res) {
	cb(err, that);
    });
}

Transaction.prototype.commit = function(cb) {
    var db = this.db;
    db.query("COMMIT", [], function(err, res) {
	cb(err);

	process.nextTick(function() {
	    dbIsAvailable(db);
	});
    });
};

Transaction.prototype.rollback = function(cb) {
    var db = this.db;

    db.query("ROLLBACK", [], function(err, res) {
	cb(err);

	process.nextTick(function() {
	    dbIsAvailable(db);
	});
    });
};


/**
 * Actual data model
 */

Transaction.prototype.createNode = function(node, cb) {
    var db = this.db;

    step(function() {
	db.query("SELECT node FROM nodes WHERE node=$1",
		 [node], this);
    }, function(err, res) {
	if (err) throw err;

	if (res.rowCount > 0)
	    throw new errors.Conflict('Node already exists');

	db.query("INSERT INTO nodes (node) VALUES ($1)", [node], this);
    }, cb);
};

/**
 * cb(err, [{ node: String, title: String }])
 */
Transaction.prototype.listNodes = function(cb) {
    var db = this.db;<commit_msg>model_postgres: move cb() past process.nextTick/dbIsAvailable for safety<commit_after>var pg = require('pg');
var step = require('step');
var ltx = require('ltx');
var errors = require('./errors');

/* ready DB connections */
var pool = [];
/* waiting transaction requests */
var queue = [];

/* at start and when connection died */
function connectDB(config) {
    var db = new pg.Client(config);
    db.connect();

    /* Reconnect in up to 5s */
    db.on('error', function(err) {
	console.error('Postgres: ' + err.message);
	setTimeout(function() {
	    connectDB(config);
	}, Math.ceil(Math.random() * 5000));
	try { db.end(); } catch (e) { /* Alright */ }
    });

    /* wait until connected & authed */
    db.connection.once('readyForQuery', function() {
	dbIsAvailable(db);
    });
}

function dbIsAvailable(db) {
    var cb;
    if ((cb = queue.shift())) {
	/* request was waiting in queue */
	cb(db);
    } else {
	/* no request, put into pool */
	pool.push(db);
    }
}

/* config: { user, database, host, port, poolSize: 4 } */
exports.start = function(config) {
    for(var i = 0; i < (config.poolSize || 4); i++) {
	connectDB(config);
    }
};


exports.transaction = function(cb) {
    var db;
    if ((db = pool.shift())) {
	/* Got one from pool */
	new Transaction(db, cb);
    } else {
	/* Pool was empty, waiting... TODO: limit length, shift first */
	queue.push(function() {
		       exports.transaction(cb);
		   });
    }
};

/**
 * Wraps the postgres-js transaction with our model operations.
 */
function Transaction(db, cb) {
    var that = this;
    this.db = db;

    db.query("BEGIN", [], function(err, res) {
	cb(err, that);
    });
}

Transaction.prototype.commit = function(cb) {
    var db = this.db;
    db.query("COMMIT", [], function(err, res) {
	process.nextTick(function() {
	    dbIsAvailable(db);
	});

	cb(err);
    });
};

Transaction.prototype.rollback = function(cb) {
    var db = this.db;

    db.query("ROLLBACK", [], function(err, res) {
	process.nextTick(function() {
	    dbIsAvailable(db);
	});

	cb(err);
    });
};


/**
 * Actual data model
 */

Transaction.prototype.createNode = function(node, cb) {
    var db = this.db;

    step(function() {
	db.query("SELECT node FROM nodes WHERE node=$1",
		 [node], this);
    }, function(err, res) {
	if (err) throw err;

	if (res.rowCount > 0)
	    throw new errors.Conflict('Node already exists');

	db.query("INSERT INTO nodes (node) VALUES ($1)", [node], this);
    }, cb);
};

/**
 * cb(err, [{ node: String, title: String }])
 */
Transaction.prototype.listNodes = function(cb) {
    var db = this.db;
{'fraction_non_alphanumeric': 0.1843076325500097, 'fraction_numerical': 0.004272674305690425, 'mean_word_length': 3.026583268178264, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4180403', 'n_tokens_mistral': 1781, 'n_tokens_neox': 1711, 'n_words': 439}