/**

* @module NoSQL Embedded Database

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

* @copyright Peter Širka 2012-2014

* @version 2.0.7

*/

'use strict';

var fs = require('fs');

var path = require('path');

var util = require('util');

var events = require('events');

var VERSION = 'v2.0.7';

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 = typeof(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 === 0 ? null : new Binary(this, directory);

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;

if (directory.length === 0)

return;

if (fs.existsSync(directory))

return;

fs.mkdirSync(directory);

}

/*

@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 (util.isDate(dt))

return dt;

if (dt === null || typeof(dt) === UNDEFINED)

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} :: optional, params: @count {Number}

@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(-1);

return self;

}

var builder = [];

var builderChanges = [];

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

var doc = arr[i];

if (typeof(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 === 0) {

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(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;

};

/*

Read data from database

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

@fnCallback {Function} :: params: @selected {Array of Object} or {Number} if is scalar

@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 (typeof(fnCallback) === UNDEFINED) {

fnCallback = fnMap;

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

}

if (typeof(fnMap) === STRING)

fnMap = filterPrepare(fnMap);

if (fnMap === null)

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 === false || item === null || typeof(item) === UNDEFINED)

return;

count++;

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

return;

if (!isScalar)

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

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

resume = false;

};

var reader = 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(isScalar ? count : selected);

});

});

return self;

};

/*

Read all documents from database

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

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

@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');

};

/*

Read one document from database

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

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

return {Database}

*/

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

var cb = function(selected) {

fnCallback(selected[0] || null);

};

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

};

/*

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');

};

/*

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');

};

/*

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 === 0) {

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);

var reader = 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;

};

/*

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} :: params: @doc {Object}, @count {Number}

return {Database}

*/

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

var self = this;

var selected = [];

var count = 0;

if (typeof(fnMap) === STRING)

fnMap = filterPrepare(fnMap);

itemTake = itemTake || 30;

itemSkip = itemSkip || 0;

var onCallback = function() {

selected.sort(fnSort);

if (itemSkip > 0 || itemTake > 0)

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

fnCallback(selected, count);

};

var onItem = function(doc) {

var item = fnMap(doc);

if (item === false || item === null || typeof(item) === UNDEFINED)

return;

count++;

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

};

self.each(onItem, onCallback);

return self;

};

/*

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 === UNDEFINED)

fnCallback = null;

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 !== null)

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;

};

/*

Drop database

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

return {Database}

*/

Database.prototype.drop = function(fnCallback) {

var self = this;

if (typeof(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 !== null)

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();

});

});

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(err === null);

});

});

});

});

return self;

};

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 === null)

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 (typeof(fnUpdate) !== UNDEFINED)

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 === 0) {

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 (typeof(o.changes) !== UNDEFINED)

changes.push(o.changes);

if (o.callback)

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

});

if (changes.length > 0)

self.changelog.insert(changes);

self.next();

});

};

// write to temporary

var fnWrite = function(json, valid) {

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 skip = false;

var value = null;

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

var fn = operation[i];

value = fn.filter(doc) || null;

if (value === null)

break;

}

if (value === null) {

self.emit('remove', doc);

countRemove++;

return;

}

var updated = JSON.stringify(value);

if (updated !== json) {

self.emit('update', value);

countUpdate++;

}

fnWrite(updated);

};

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

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

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

return true;

}, function(success) {

if (!success) {

self.next();

return;

}

completed = true;

fnWrite(null, true);

});

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

return {Database}

*/

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

var self = this;

if (typeof(fnUpdate) !== UNDEFINED)

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

return self;

};

/*

Remove data from database

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

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

@changes {String} :: optional, changes description

return {Database}

*/

Database.prototype.remove = function(fnFilter, fnCallback, changes) {

var self = this;

if (typeof(fnFilter) === STRING)

fnFilter = filterPrepare(fnFilter);

var filter = function(item) {

if (fnFilter(item) === true)

return null;

return item;

};

self.update(filter, fnCallback, changes, 'remove');

return self;

};

Database.prototype.pause = function() {

var self = this;

if (self.isPending === true)

return self;

self.isPending = true;

if (self.status === STATUS_UNKNOWN) {

self.status = STATUS_PENDING;

self.emit('pause/resume', true);

}

return self;

};

Database.prototype.resume = function() {