manage log lines using db directly

app.js

@@ -181,8 +181,7 @@ Steppy(
 		logger.log('Server config:', JSON.stringify(app.config, null, 4));
 
 		db.init(app.config.paths.db, {
-			db: require(app.config.storage.backend),
+			db: require(app.config.storage.backend)
-			valueEncoding: 'json'
 		}, this.slot());
 	},
 	function() {

db.js

@@ -4,14 +4,17 @@ var Steppy = require('twostep').Steppy,
 	_ = require('underscore'),
 	nlevel = require('nlevel'),
 	path = require('path'),
-	utils = require('./lib/utils');
+	utils = require('./lib/utils'),
+	through = require('through');
 
 
 exports.init = function(dbPath, params, callback) {
 	callback = _.after(2, callback);
 
 	var maindbPath = path.join(dbPath, 'main'),
-		mainDb = nlevel.db(maindbPath, params, callback);
+		mainDb = nlevel.db(maindbPath, _({
+			valueEncoding: 'json'
+		}).defaults(params), callback);
 
 	exports.builds = new nlevel.DocsSection(mainDb, 'builds', {
 		projections: [
@@ -82,22 +85,98 @@ exports.init = function(dbPath, params, callback) {
 	var buildLogsDbPath = path.join(dbPath, 'buildLogs'),
 		buildLogsDb = nlevel.db(buildLogsDbPath, params, callback);
 
-	exports.logLines = new nlevel.DocsSection(buildLogsDb, 'logLines', {
+	// custom optimized emplementation for storing log lines
-		projections: [
+	exports.logLines = {};
-			{
+
-				key: {
+	exports.logLines.separator = '~';
-					buildId: 1,
+	exports.logLines.end = '\xff';
-					numberStr: function(logLine) {
+
-						return utils.toNumberStr(logLine.number);
+	exports.logLines._getStrKey = function(line) {
-					}
+		return line.buildId + this.separator + (
-				},
+			_(line).has('number') ? utils.toNumberStr(line.number) : ''
-				value: function(logLine) {
+		);
-					return _(logLine).pick('number', 'text');
+	};
-				}
+
-			}
+	exports.logLines._parseData = function(data) {
-		],
+		var keyParts = data.key.split(this.separator),
-		withUniqueId: false
+			buildId = Number(keyParts[0]),
-	});
+			number = Number(keyParts[1]);
+
+		return {buildId: buildId, number: number, text: data.value};
+	};
+
+	exports.logLines.put = function(lines, callback) {
+		var self = this;
+		lines = _(lines).isArray() ? lines : [lines];
+
+		var operations = _(lines).map(function(line) {
+			return {
+				type: 'put',
+				key: self._getStrKey(line),
+				value: line.text
+			};
+		});
+
+		buildLogsDb.batch(operations, callback);
+	};
+
+	exports.logLines.createReadStream = function(params) {
+		var self = this;
+		if (!params.start && params.end) {
+			new Error('`end` selected without `start`');
+		}
+
+		params.start = self._getStrKey(params.start);
+		params.end = params.end ? self._getStrKey(params.end) : params.start;
+		// add end character
+		params.end += self.end;
+		// swap `start` `end` conditions when reverse is set
+		if (params.reverse) {
+			var prevStart = params.start;
+			params.start = params.end;
+			params.end = prevStart;
+		}
+
+		var resultStream = through(function(data) {
+			this.emit('data', _(data).isObject() ? self._parseData(data) : data);
+		});
+
+		return buildLogsDb.createReadStream(params)
+			.on('error', function(err) {
+				resultStream.emit('error', err);
+			})
+			.pipe(resultStream)
+	};
+
+	exports.logLines.find = function(params, callback) {
+		var self = this;
+		callback = _(callback).once();
+
+		var lines = [];
+		self.createReadStream(params)
+			.on('error', callback)
+			.on('data', function(line) {
+				lines.push(line);
+			})
+			.on('end', function() {
+				callback(null, lines);
+			})
+	};
+
+	exports.logLines.remove = function(params, callback) {
+		var self = this;
+		callback = _(callback).once();
+
+		var operations = [];
+		self.createReadStream(_({values: false}).extend(params))
+			.on('error', callback)
+			.on('data', function(key) {
+				operations.push({type: 'del', key: key});
+			})
+			.on('end', function() {
+				buildLogsDb.batch(operations, callback);
+			});
+	};
 };
 
 /*

distributor.js

@@ -57,7 +57,7 @@ exports.init = function(app, callback) {
 			Steppy(
 				function() {
 					db.logLines.find({
-						start: {buildId: buildId, numberStr: ''},
+						start: {buildId: buildId},
 					}, this.slot());
 				},
 				function(err, lines) {

package.json

@@ -50,7 +50,8 @@
     "node-static": "0.7.6",
     "socket.io": "1.3.5",
     "twostep": "0.4.1",
-    "underscore": "1.8.3"
+    "underscore": "1.8.3",
+    "through": "2.3.6"
   },
   "devDependencies": {
     "bower": "1.4.1",

resources/builds.js

@@ -65,7 +65,7 @@ module.exports = function(app) {
 			function() {
 				var findParams = {
 					reverse: true,
-					start: {buildId: req.data.buildId, numberStr: ''},
+					start: {buildId: req.data.buildId},
 					limit: req.data.length
 				};
 
@@ -90,8 +90,8 @@ module.exports = function(app) {
 					count = to - from;
 
 				db.logLines.find({
-					start: {buildId: buildId, numberStr: utils.toNumberStr(from)},
+					start: {buildId: buildId, number: from},
-					end: {buildId: buildId, numberStr: utils.toNumberStr(to)}
+					end: {buildId: buildId, number: to}
 				}, this.slot());
 
 				this.pass(count);

static/js/app/components/buildLog/index.js

@@ -14,11 +14,12 @@ define([
 	return React.createClass({
 		mixins: [
 			Reflux.connectFilter(buildLogStore, 'data', function(data) {
-				data.output = _(data.lines).pluck('text').join('');
+				data.output = _(data.lines).pluck('text').join('<br>');
 				data.output = data.output.replace(
 					/(.*)\n/gi,
 					'<span class="terminal_code_newline">$1</span>'
 				);
+
 				data.output = ansiUp.ansi_to_html(data.output);
 				return data;
 			})