/**

* @module NoSQL Embedded Database

* @author Peter Širka <petersirka@gmail.com>

* @copyright Peter Širka 2012-2016

* @version 3.1.2

*/

'use strict';

var fs = require('fs');

var path = require('path');

var events = require('events');

var VERSION = 'v3.1.2';

var STATUS_UNKNOWN = 0;

var STATUS_READING = 1;

var STATUS_WRITING = 2;

var STATUS_LOCKING = 3;

var STATUS_PENDING = 4;

var EXTENSION = '.nosql';

var EXTENSION_VIEW = '.nosql';

var EXTENSION_BINARY = '.nosql-binary';

var EXTENSION_TMP = '.nosql-tmp';

var EXTENSION_CHANGES = '.changelog';

var EXTENSION_STORED = '.nosql-stored';

var EXTENSION_META = '.meta';

var MAX_WRITESTREAM = 2;

var MAX_READSTREAM = 4;

var MAX_BUFFER_SIZE = 1024 * 40;

var BINARY_HEADER_LENGTH = 2000;

var NEWLINE = '\n';

var STRING = 'string';

var FUNCTION = 'function';

var UNDEFINED = 'undefined';

var BOOLEAN = 'boolean';

if (typeof(setImmediate) === UNDEFINED) {

global.setImmediate = function(cb) {

process.nextTick(cb);

};

}

/*

Database

@filename {String}

@directory {String} :: for storing binary data

@changes {Boolean} :: optional, default true

*/

function Database(filename, directory, changes) {

if (typeof(directory) === BOOLEAN) {

changes = directory;

directory = null;

}

this.isReady = false;

this.status_prev = STATUS_UNKNOWN;

this.status = STATUS_UNKNOWN;

this.changes = changes === undefined ? true : changes === true;

this.countRead = 0;

this.countWrite = 0;

this.pendingRead = [];

this.pendingEach = [];

this.pendingLock = [];

this.pendingDrop = [];

this.pendingWrite = [];

this.pendingClear = [];

this.isPending = false;

if (filename.indexOf(EXTENSION) !== -1)

filename = filename.replace(EXTENSION, '');

this.filename = filename + EXTENSION;

this.filenameTemp = filename + EXTENSION_TMP;

this.filenameChanges = filename + EXTENSION_CHANGES;

this.filenameStored = filename + EXTENSION_STORED;

this.filenameMeta = filename + EXTENSION_META;

this.name = path.basename(filename);

this.directory = path.dirname(filename);

this.views = new Views(this);

this.meta = {

version: VERSION,

views: {},

stored: {},

description: '',

created: new Date(),

custom: null

};

this.binary = (directory || '').length ? new Binary(this, directory) : null;

this.changelog = new Changelog(this, this.filenameChanges);

this.file = new FileReader(this);

this.stored = new Stored(this, this.filenameStored);

this._metaLoad();

}

/*

@db {Database}

*/

function Views(db) {

this.views = {};

this.db = db;

this.directory = db.directory;

this.emit = db.emit;

}

/*

@db {Database}

@name {String}

@filename {String} :: database filename

*/

function View(db, name, filename) {

this.db = db;

this.status = STATUS_UNKNOWN;

this.countRead = 0;

this.pendingRead = [];

this.pendingOperation = [];

this.filename = filename;

this.name = name;

this.emit = db.emit;

this.file = new FileReader(db);

}

/*

@db {Database}

*/

function Stored(db, filename) {

this.filename = filename;

this.db = db;

this.stored = {};

this.cache = {};

this.isReaded = false;

}

/*

@db {Database}

@directory {String}

*/

function Binary(db, directory) {

this.db = db;

this.directory = directory;

this.exists = false;

}

/*

@db {Database}

@filename {String}

*/

function Changelog(db, filename) {

this.filename = filename;

this.db = db;

}

/*

@db {Database}

*/

function FileReader(db) {

this.db = db;

}

/*

PROTOTYPES

*/

Database.prototype = {

get created() {

var dt = this.meta.created;

if (!(dt instanceof Date))

return dt;

if (!dt.getTime())

return null;

return new Date(Date.parse(dt.toString()));

}

};

Database.prototype.__proto__ = Object.create(events.EventEmitter.prototype, {

constructor: {

value: Database,

enumberable: false

}

});

/*

Insert data into database

@arr {Array of Object}

@fnCallback {Function(err, count)} callback

@changes {String} :: optional, insert description

return {Database}

*/

Database.prototype.insert = function(arr, fnCallback, changes) {

var self = this;

if (typeof(fnCallback) === STRING) {

changes = fnCallback;

fnCallback = null;

}

if (!(arr instanceof Array))

arr = [arr];

var length = arr.length;

if (self.status === STATUS_LOCKING|| self.status === STATUS_PENDING || self.countWrite >= MAX_WRITESTREAM) {

for (var i = 0; i < length; i++)

self.pendingWrite.push({ json: arr[i], changes: changes });

if (fnCallback)

fnCallback(null, -1);

return self;

}

var builder = [];

var builderChanges = [];

for (var i = 0; i < length; i++) {

var doc = arr[i];

if (doc.json === undefined) {

builder.push(doc);

if (changes)

builderChanges.push(changes);

continue;

}

builder.push(doc.json);

if (doc.changes)

builderChanges.push(doc.changes);

}

if (!builder.length) {

self.next();

return;

}

self.emit('insert', true, builder.length);

self.status = STATUS_WRITING;

self.countWrite++;

appendFile(self.filename, builder, function() {

self.countWrite--;

self.emit('insert', false, builder.length);

self.next();

self.changelog.insert(builderChanges);

if (fnCallback) {

var length = builder.length;

setImmediate(function() { fnCallback(null, length); });

}

builder = null;

builderChanges = null;

arr = null;

var keys = Object.keys(self.meta.views);

var length = keys.length;

for (var i = 0; i < length; i++)

self.views.refresh(keys[i]);

}, self);

return self;

};

Database.prototype.$$insert = function(arr, changes) {

var self = this;

return function(callback) {

self.insert(arr, callback, changes);

};

};

/*

Read data from database

@fnMap {Function} :: params: @doc {Object}, IMPORTANT: you must return {Object}

@fnCallback {Function(err, array/number)}

@itemSkip {Number} :: optional, default 0

@itemTake {Number} :: optional, defualt 0

@isScalar {Boolean} :: optional, default is false

return {Database}

*/

Database.prototype.read = function(fnMap, fnCallback, itemSkip, itemTake, isScalar, name) {

var self = this;

var skip = itemSkip || 0;

var take = itemTake || 0;

if (self.status === STATUS_LOCKING || self.status === STATUS_PENDING || self.countRead >= MAX_READSTREAM) {

self.pendingRead.push(function() {

self.read(fnMap, fnCallback, itemSkip, itemTake, isScalar);

});

return self;

}

if (!fnCallback) {

fnCallback = fnMap;

fnMap = function(doc) { return doc; };

}

if (typeof(fnMap) === STRING)

fnMap = filterPrepare(fnMap);

if (!fnMap)

fnMap = function(doc) { return doc; };

self.emit(name || 'read', true, 0);

// opened streams

self.countRead++;

self.status = STATUS_READING;

var selected = [];

var current = '';

var count = 0;

var resume = true;

var fnBuffer = function(buffer) {

current += buffer;

return current;

};

var fnItem = function(err, doc) {

if (!resume)

return;

// clear buffer;

current = '';

if (err)

return;

var item = fnMap(doc);

if (!item)

return;

count++;

if (skip > 0 && count <= skip)

return;

if (!isScalar)

selected.push(item === true ? doc : item);

if (take > 0 && selected.length === take)

resume = false;

};

self.file.open(self.filename, MAX_BUFFER_SIZE, function(buffer) {

onBuffer(buffer, fnItem, fnBuffer);

return resume;

}, function() {

self.countRead--;

self.next();

setImmediate(function() {

self.emit(name || 'read', false, isScalar ? count : selected.length);

fnCallback(null, isScalar ? count : selected);

});

});

return self;

};

/*

Read all documents from database

@fnMap {Function} :: IMPORTANT: you must return {Object}

@fnCallback {Function(err, array)}

@itemSkip {Number} :: optional, default 0

@itemTake {Number} :: optional, default 0

return {Database}

*/

Database.prototype.all = function(fnMap, fnCallback, itemSkip, itemTake) {

return this.read(fnMap, fnCallback, itemSkip, itemTake, false, 'all');

};

Database.prototype.$$all = function(fnMap, itemSkip, itemTake) {

var self = this;

return function(callback) {

return self.all(fnMap, callback, itemSkip, itemTake);

};

};

/*

Read one document from database

@fnMap {Function} :: must return {Object}

@fnCallback {Function(err, doc)}

return {Database}

*/

Database.prototype.one = function(fnMap, fnCallback) {

var cb = function(err, selected) {

fnCallback(err, selected ? selected[0] || null : null);

};

return this.read(fnMap, cb, 0, 1, false, 'one');

};

Database.prototype.$$one = function(fnMap) {

var self = this;

return function(callback) {

return self.one(fnMap, callback);

};

};

/*

Read TOP "x" documents from database

@fnMap {Function} :: IMPORTANT: you must return {Object}

@fnCallback {Function} :: params: @doc {Array of Object}

return {Database}

*/

Database.prototype.top = function(max, fnMap, fnCallback) {

return this.read(fnMap, fnCallback, 0, max, false, 'top');

};

Database.prototype.$$top = function(max, fnMap) {

var self = this;

return function(callback) {

return self.top(max, fnMap, callback);

};

};

/*

Count documents

@fnFilter {Function} :: params: @doc {Object}, IMPORTANT: you must return {Boolean}

@fnCallback {Function} :: params: @count {Number}

return {Database}

*/

Database.prototype.count = function(fnFilter, fnCallback) {

return this.read(fnFilter, fnCallback, 0, 0, true, 'count');

};

Database.prototype.$$count = function(fnFilter) {

var self = this;

return function(callback) {

return self.count(fnFilter, fnMap, callback);

};

};

/*

Read each document from database

@fnDocument {Function} :: params: @doc {Object}, @offset {Number}

@fnCallback {Function} :: optional

return {Database}

*/

Database.prototype.each = function(fnDocument, fnCallback) {

var self = this;

if (self.status === STATUS_LOCKING || self.status === STATUS_PENDING || self.countRead >= MAX_READSTREAM) {

if (fnDocument)

self.pendingEach.push({ item: fnDocument, callback: fnCallback });

return self;

}

var operation = [];

if (fnDocument)

operation.push({ item: fnDocument, callback: fnCallback });

var length = self.pendingEach.length;

for (var i = 0; i < length; i++)

operation.push(self.pendingEach[i]);

if (!operation.length) {

self.next();

return self;

}

// opened streams

self.countRead++;

self.status = STATUS_READING;

var current = '';

var count = 0;

self.pendingEach = [];

var fnBuffer = function(buffer) {

current += buffer;

return current;

};

var fnItem = function(err, doc) {

// clear buffer;

current = '';

if (err) {

self.emit('error', err, 'each-buffer');

return;

}

var length = operation.length;

for (var i = 0; i < length; i++) {

try

{

operation[i].item(doc, count, 'each-buffer');

} catch (e) {

self.emit('error', e);

}

}

count++;

};

self.emit('each', true, 0);

self.file.open(self.filename, MAX_BUFFER_SIZE, function(buffer) {

onBuffer(buffer, fnItem, fnBuffer);

return true;

}, function() {

self.countRead--;

self.next();

setImmediate(function() {

self.emit('each', false, count);

var length = operation.length;

for (var i = 0; i < length; i++) {

var fn = operation[i];

if (fn.callback)

fn.callback();

}

});

});

return self;

};

Database.prototype.$$each = function(fnDocument) {

var self = this;

return function(callback) {

return self.each(fnDocument, callback);

};

};

/*

Read and sort documents from database (SLOWLY)

@fnMap {Function} :: IMPORTANT: you must return {Object}

@fnSort {Function} :: ---> array.sort()

@itemSkip {Number}, default 0 (if itemSkip = 0 and itemTake = 0 then return all documents)

@itemTake {Number}, default 0 (if itemSkip = 0 and itemTake = 0 then return all documents)

@fnCallback {Function(err, selected, count)} :: params: @doc {Object}, @count {Number}

return {Database}

*/

Database.prototype.sort = function(fnMap, fnSort, fnCallback, itemSkip, itemTake) {

var self = this;

var selected = [];

var count = 0;

if (typeof(fnMap) === STRING)

fnMap = filterPrepare(fnMap);

itemTake = itemTake || 0;

itemSkip = itemSkip || 0;

var onCallback = function() {

selected.sort(fnSort);

if (itemSkip || itemTake)

selected = selected.slice(itemSkip, itemSkip + itemTake);

fnCallback(null, selected, count);

};

var onItem = function(doc) {

var item = fnMap(doc);

if (!item)

return;

count++;

selected.push(item === true ? doc : item);

};

self.each(onItem, onCallback);

return self;

};

Database.prototype.$$sort = function(fnMap, fnSort, itemSkip, itemTake) {

var self = this;

return function(callback) {

self.sort(fnMap, fnSort, callback, itemSkip, itemTake);

};

};

/*

Clear database

@fnCallback {Function} :: optional

@changes {String} :: optional, clear description

return {Database}

*/

Database.prototype.clear = function(fnCallback, changes) {

var self = this;

var type = typeof(fnCallback);

if (type === STRING) {

changes = fnCallback;

fnCallback = null;

}

self.pendingClear.push(function() {

if (changes)

self.changelog.insert(changes);

if (fnCallback)

fnCallback();

});

if (self.status !== STATUS_UNKNOWN)

return self;

self.status = STATUS_LOCKING;

var operation = [];

var length = self.pendingClear.length;

for (var i = 0; i < length; i++) {

var fn = self.pendingClear[i];

if (fn)

operation.push(fn);

}

self.emit('clear', true, false);

self.pendingClear = [];

fs.exists(self.filename, function(exists) {

if (!exists) {

self.next();

setImmediate(function() {

self.emit('clear', false, true);

var length = operation.length;

for (var i = 0; i < length; i++) {

var fn = operation[i];

if (fn)

fn();

}

});

return;

}

fs.unlink(self.filename, function(err) {

if (err)

self.emit('error', err, 'clear');

self.next();

setImmediate(function() {

self.emit('clear', false, err === null);

var length = operation.length;

for (var i = 0; i < length; i++) {

var fn = operation[i];

if (fn)

fn();

}

});

});

});

return self;

};

Database.prototype.$$clear = function(changes) {

var self = this;

return function(callback) {

self.clear(callback, changes);

};

};

/*

Drop database

@fnCallback {Function} :: optional, params: @dropped {Boolean}

return {Database}

*/

Database.prototype.drop = function(fnCallback) {

var self = this;

if (fnCallback === undefined)

fnCallback = null;

self.pendingDrop.push(fnCallback);

if (self.status !== STATUS_UNKNOWN)

return self;

self.status = STATUS_LOCKING;

var operation = [];

self.pendingDrop.forEach(function(o) {

if (o)

operation.push(o);

});

self.emit('drop', true, false);

self.pendingDrop = [];

self._drop();

fs.exists(self.filename, function(exists) {

if (!exists) {

self.next();

setImmediate(function() {

self.emit('drop', false, true);

operation.forEach(function(fn) {

if (fn)

fn(null);

});

});

return;

}

fn.unlink(self.filenameMeta, noop);

fs.unlink(self.filename, function(err) {

if (err)

self.emit('error', err, 'drop');

self.next();

setImmediate(function() {

self.emit('drop', false, err === null);

operation.forEach(function(fn) {

if (fn)

fn(null, err === null);

});

});

});

});

return self;

};

Database.prototype.$$drop = function() {

var self = this;

return function(callback) {

self.drop(callback);

};

};

function noop() {};

/*

Internal function :: remove all files (views, binary)

!!! SYNC !!!

*/

Database.prototype._drop = function() {

var self = this;

fs.readdirSync(self.directory).forEach(function(filename) {

var isView = filename.indexOf(self.name + '#') !== -1 && filename.indexOf(EXTENSION_VIEW) !== -1;

if (isView) {

fs.unlink(path.join(self.directory, filename), noop);

return;

}

var isChange = self.name + EXTENSION_CHANGES === filename;

if (isChange) {

fs.unlink(path.join(self.directory, filename), noop);

return;

}

});

if (!self.binary)

return self;

fs.readdirSync(self.binary.directory).forEach(function(filename) {

if (filename.indexOf(self.name + '#') === -1 || filename.indexOf(EXTENSION_BINARY) === -1)

return;

fs.unlink(path.join(self.binary.directory, filename), noop);

});

return self;

};

/*

Update multiple documents

@fnUpdate {Function} :: params: @doc {Object} and IMPORTANT: you must return updated @doc;

@fnCallback {Function} :: optional, params: @count {Number}

@changes {String} :: optional, changes description

@type {String} :: internal, optional

return {Database}

*/

Database.prototype.update = function(fnUpdate, fnCallback, changes, type) {

var self = this;

if (typeof(fnCallback) === STRING) {

changes = fnCallback;

fnCallback = null;

}

if (fnUpdate)

self.pendingLock.push(updatePrepare(fnUpdate, fnCallback, changes, type || 'update'));

if (self.status !== STATUS_UNKNOWN)

return self;

var operation = [];

self.pendingLock.forEach(function(fn) {

operation.push(fn);

});

if (!operation.length) {

self.next();

return self;

}

self.status = STATUS_LOCKING;

var current = '';

var operationLength = operation.length;

var lines = [];

var countRemove = 0;

var countUpdate = 0;

var countWrite = 0;

var completed = false;

self.emit('update/remove', true, 0, 0);

self.pendingLock = [];

// rename updated file

var fnRename = function() {

operation.forEach(function(o) {

if (o.type === 'update') {

o.count = countUpdate;

return;

}

if (o.type === 'remove')

o.count = countRemove;

});

fs.rename(self.filenameTemp, self.filename, function(err) {

if (err)

self.emit('error', err, 'update/rename-file');

self.emit('update/remove', false, countUpdate, countRemove);

var changes = [];

operation.forEach(function(o) {

if (o.changes !== undefined)

changes.push(o.changes);

if (o.callback)

(function(cb,count) { setImmediate(function() { cb(null, count); }); })(o.callback, o.count);

});

if (changes.length)

self.changelog.insert(changes);

self.next();

});

};

var can = true;

// write to temporary

var fnWrite = function(json, valid) {

if (can) {

can = false;

fs.appendFile(self.filenameTemp, '');

}

if (lines.length > 25 || valid) {

if (lines.length === 0) {

if (completed && countWrite <= 0)

fnRename();

return;

}

countWrite++;

fs.appendFile(self.filenameTemp, lines.join(NEWLINE) + NEWLINE, function() {

countWrite--;

if (completed && countWrite <= 0)

fnRename();

});

lines = [];

}

if (typeof(json) === STRING)

lines.push(json);

};

var fnBuffer = function(buffer) {

current += buffer;

return current;

};

var fnItem = function(err, doc, json) {

// clear buffer;

current = '';

var value = null;

for (var i = 0; i < operationLength; i++) {

var fn = operation[i];

value = fn.filter(doc);

if (!value)

break;

}

if (!value) {

self.emit('remove', doc);

countRemove++;

return;

}

var updated;

if (value && typeof(value.$clean) === FUNCTION)

updated = value.$clean();

else

updated = JSON.stringify(value);

if (updated !== json) {

self.emit('update', value);

countUpdate++;

}

fnWrite(updated);

};

self.file.open(self.filename, MAX_BUFFER_SIZE, function(buffer) {

onBuffer(buffer.toString(), fnItem, fnBuffer);

return true;

}, function(success) {

if (!success) {

self.emit('update/remove', false, countUpdate, countRemove);

var changes = [];

operation.forEach(function(o) {

if (o.changes)

changes.push(o.changes);

if (o.callback)

(function(cb,count) { setImmediate(function() { cb(null, count); }); })(o.callback, o.count);

});

if (changes.length)

self.changelog.insert(changes);

self.next();

return;

}

completed = true;

fnWrite(null, true);

});