// Copyright 2012-2017 (c) Peter Širka <petersirka@gmail.com>

//

// Permission is hereby granted, free of charge, to any person obtaining a

// copy of this software and associated documentation files (the

// "Software"), to deal in the Software without restriction, including

// without limitation the rights to use, copy, modify, merge, publish,

// distribute, sublicense, and/or sell copies of the Software, and to permit

// persons to whom the Software is furnished to do so, subject to the

// following conditions:

//

// The above copyright notice and this permission notice shall be included

// in all copies or substantial portions of the Software.

//

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS

// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN

// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,

// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR

// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE

// USE OR OTHER DEALINGS IN THE SOFTWARE.

/**

* @module NoSQL

* @version 2.4.0

*/

'use strict';

const Fs = require('fs');

const Path = require('path');

const Events = require('events');

if (!global.framework_utils)

global.framework_utils = require('./utils');

if (!global.framework_image)

global.framework_image = require('./image');

if (!global.framework_nosql)

global.framework_nosql = exports;

if (!global.framework_builders)

global.framework_builders = require('./builders');

const EXTENSION = '.nosql';

const EXTENSION_BINARY = '.nosql-binary';

const EXTENSION_TMP = '.nosql-tmp';

const EXTENSION_META = '.meta';

const BINARY_HEADER_LENGTH = 2000;

const NEWLINE = '\n';

const EMPTYARRAY = [];

const REG_CLEAN = /^[\s]+|[\s]+$/g;

const INMEMORY = {};

Object.freeze(EMPTYARRAY);

function Database(name, filename) {

this.filename = filename + EXTENSION;

this.filenameTemp = filename + EXTENSION_TMP;

this.filenameMeta = filename + EXTENSION_META;

this.directory = Path.dirname(filename);

this.filenameBackup = framework_utils.join(this.directory, name + '_backup' + EXTENSION);

this.name = name;

this.pending_update = [];

this.pending_append = [];

this.pending_reader = [];

this.pending_reader_view = [];

this.pending_remove = [];

this.views = {};

this.step = 0;

this.pending_drops = false;

this.pending_views = false;

this.binary = new Binary(this, this.directory + '/' + this.name + '-binary/');

this.counter = new Counter(this);

this.inmemory = {};

this.inmemorylastusage;

this.metadata;

this.$meta();

this.$timeoutmeta;

}

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

constructor: {

value: Database,

enumberable: false

}

});

exports.load = function(name, filename) {

return new Database(name, filename);

};

exports.memory = exports.inmemory = function(name, view) {

if (view)

name += '#' + view;

return INMEMORY[name] = true;

};

Database.prototype.meta = function(name, value) {

var self = this;

if (value === undefined)

return self.metadata ? self.metadata[name] : undefined;

if (!self.metadata)

self.metadata = {};

self.metadata[name] = value;

clearTimeout(self.timeoutmeta);

self.timeoutmeta = setTimeout(() => self.$meta(true), 500);

return self;

};

Database.prototype.insert = function(doc, unique) {

var self = this;

var builder;

if (unique) {

builder = self.one();

var callback;

builder.callback(function(err, d) {

if (d)

callback && callback(null, 0);

else

self.insert(doc).callback(callback);

});

builder.callback = function(fn) {

callback = fn;

return builder;

};

return builder;

}

builder = new DatabaseBuilder2();

var json = framework_builders.isSchema(doc) ? doc.$clean() : doc;

self.pending_append.push({ doc: JSON.stringify(json), builder: builder });

setImmediate(() => self.next(1));

self.emit('insert', json);

return builder;

};

Database.prototype.upsert = function(doc) {

return this.insert(doc, true);

};

Database.prototype.update = function(doc, insert) {

var self = this;

var builder = new DatabaseBuilder();

self.pending_update.push({ builder: builder, doc: framework_builders.isSchema(doc) ? doc.$clean() : doc, count: 0, insert: insert });

setImmediate(() => self.next(2));

return builder;

};

Database.prototype.modify = function(doc, insert) {

var self = this;

var builder = new DatabaseBuilder();

var data = framework_builders.isSchema(doc) ? doc.$clean() : doc;

var keys = Object.keys(data);

if (!keys.length)

return builder;

self.pending_update.push({ builder: builder, doc: data, count: 0, keys: keys, insert: insert });

setImmediate(() => self.next(2));

return builder;

};

Database.prototype.backup = function(filename, remove) {

if (typeof(filename) === 'boolean') {

remove = filename;

filename = '';

}

var self = this;

if (remove)

return self.remove(filename || '');

var builder = new DatabaseBuilder2();

var stream = Fs.createReadStream(self.filename);

stream.pipe(Fs.createWriteStream(filename || self.filenameBackup));

stream.on('error', function(err) {

builder.$callback && builder.$callback(errorhandling(err, builder));

builder.$callback = null;

});

stream.on('end', function() {

builder.$callback && builder.$callback(errorhandling(null, builder, true), true);

builder.$callback = null;

});

return builder;

};

Database.prototype.drop = function() {

var self = this;

self.pending_drops = true;

setImmediate(() => self.next(7));

return self;

};

Database.prototype.free = function() {

var self = this;

self.removeAllListeners();

delete framework.databases[self.name];

return self;

};

Database.prototype.release = function() {

var self = this;

self.inmemory = {};

self.inmemorylastusage = undefined;

return self;

};

Database.prototype.clear = Database.prototype.remove = function(filename) {

var self = this;

var builder = new DatabaseBuilder();

var backup = filename === undefined ? undefined : filename || self.filenameBackup;

if (backup)

backup = new Backuper(backup);

self.pending_remove.push({ builder: builder, count: 0, backup: backup });

setImmediate(() => self.next(3));

return builder;

};

Database.prototype.find = function(view) {

var self = this;

var builder = new DatabaseBuilder();

if (view) {

self.pending_reader_view.push({ builder: builder, count: 0, counter: 0, view: view });

setImmediate(() => self.next(6));

} else {

self.pending_reader.push({ builder: builder, count: 0, counter: 0, view: view });

setImmediate(() => self.next(4));

}

return builder;

};

Database.prototype.scalar = function(type, field, view) {

return this.find(view).scalar(type, field);

};

Database.prototype.count = function(view) {

var self = this;

var builder = new DatabaseBuilder();

if (view) {

self.pending_reader_view.push({ builder: builder, count: 0, view: view, type: 1 });

setImmediate(() => self.next(6));

} else {

self.pending_reader.push({ builder: builder, count: 0, view: view, type: 1 });

setImmediate(() => self.next(4));

}

return builder;

};

Database.prototype.one = function(view) {

var self = this;

var builder = new DatabaseBuilder();

builder.first();

if (view) {

self.pending_reader_view.push({ builder: builder, count: 0, view: view });

setImmediate(() => self.next(6));

} else {

self.pending_reader.push({ builder: builder, count: 0, view: view });

setImmediate(() => self.next(4));

}

return builder;

};

Database.prototype.top = function(max, view) {

var self = this;

var builder = new DatabaseBuilder();

builder.take(max);

if (view) {

self.pending_reader_view.push({ builder: builder, count: 0, counter: 0, view: view });

setImmediate(() => self.next(6));

} else {

self.pending_reader.push({ builder: builder, count: 0, counter: 0, view: view });

setImmediate(() => self.next(4));

}

return builder;

};

Database.prototype.view = function(name) {

var builder = new DatabaseBuilder();

this.views[name] = {};

this.views[name] = builder;

this.views[name].$filename = this.filename.replace(/\.nosql/, '#' + name + '.nosql');

return builder;

};

Database.prototype.next = function(type) {

var self = this;

if (type && self.step)

return self;

if (self.step !== 6 && self.pending_reader_view.length) {

self.$readerview();

return self;

}

if (self.step !== 4 && self.pending_reader.length) {

self.$reader();

return self;

}

if (self.step !== 1 && self.pending_append.length) {

if (INMEMORY[self.name])

self.$append_inmemory();

else

self.$append();

return self;

}

if (self.step !== 2 && self.pending_update.length) {

if (INMEMORY[self.name])

self.$update_inmemory();

else

self.$update();

return self;

}

if (self.step !== 3 && self.pending_remove.length) {

if (INMEMORY[self.name])

self.$remove_inmemory();

else

self.$remove();

return self;

}

if (self.step !== 5 && self.pending_views) {

if (INMEMORY[self.name])

self.$views_inmemory();

else

self.$views();

return self;

}

if (self.step !== 7 && self.pending_drops) {

self.$drop();

return self;

}

if (self.step !== type) {

self.step = 0;

setImmediate(() => self.next(0));

}

return self;

};

Database.prototype.refresh = function() {

var self = this;

if (!self.views)

return self;

self.pending_views = true;

setImmediate(() => self.next(5));

return self;

};

// ======================================================================

// FILE OPERATIONS

// ======================================================================

// InMemory saving

Database.prototype.$save = function(view) {

var self = this;

setTimeout2('nosql.' + self.name + '.' + view, function() {

var data = self.inmemory[view] || EMPTYARRAY;

var builder = [];

for (var i = 0, length = data.length; i < length; i++)

builder.push(JSON.stringify(data[i]));

var filename = self.filename;

if (view !== '#')

filename = filename.replace(/\.nosql/, '#' + view + '.nosql');

Fs.writeFile(filename, builder.join(NEWLINE) + NEWLINE, NOOP);

}, 50);

return self;

};

Database.prototype.$inmemory = function(view, callback) {

var self = this;

// Last usage

self.inmemorylastusage = global.framework ? global.framework.datetime : undefined;

if (self.inmemory[view])

return callback();

var filename = self.filename;

if (view !== '#')

filename = filename.replace(/\.nosql/, '#' + view + '.nosql');

self.inmemory[view] = [];

Fs.readFile(filename, function(err, data) {

if (err)

return callback();

var arr = data.toString('utf8').split('\n');

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

var item = arr[i];

if (!item)

continue;

try {

item = JSON.parse(item.trim(), jsonparser);

item && self.inmemory[view].push(item);

} catch (e) {};

}

callback();

});

return self;

};

Database.prototype.$meta = function(write) {

var self = this;

if (write) {

Fs.writeFile(self.filenameMeta, JSON.stringify(self.metadata), NOOP);

return self;

}

try {

self.metadata = JSON.parse(Fs.readFileSync(self.filenameMeta).toString('utf8'), jsonparser);

} catch (err) {}

return self;

};

Database.prototype.$append = function() {

var self = this;

self.step = 1;

if (!self.pending_append.length) {

self.next(0);

return self;

}

self.pending_append.splice(0).limit(20, function(items, next) {

var json = [];

for (var i = 0, length = items.length; i < length; i++)

json.push(items[i].doc);

Fs.appendFile(self.filename, json.join(NEWLINE) + NEWLINE, function(err) {

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

var callback = items[i].builder.$callback;

callback && callback(err, 1);

}

next();

});

}, function() {

setImmediate(function() {

self.next(0);

setImmediate(() => self.refresh());

});

});

return self;

};

Database.prototype.$append_inmemory = function() {

var self = this;

self.step = 1;

if (!self.pending_append.length) {

self.next(0);

return self;

}

var items = self.pending_append.splice(0);

return self.$inmemory('#', function() {

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

self.inmemory['#'].push(JSON.parse(items[i].doc, jsonparser));

var callback = items[i].builder.$callback;

callback && callback(null, 1);

}

self.$save('#');

setImmediate(function() {

self.next(0);

setImmediate(() => self.refresh());

});

});

};

Database.prototype.$update = function() {

var self = this;

self.step = 2;

if (!self.pending_update.length) {

self.next(0);

return self;

}

var reader = Fs.createReadStream(self.filename);

var writer = Fs.createWriteStream(self.filenameTemp);

var filter = self.pending_update.splice(0);

var length = filter.length;

var change = false;

reader.on('data', framework_utils.streamer(NEWLINE, function(value, index) {

var doc = JSON.parse(value.trim());

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

var item = filter[i];

var builder = item.builder;

var output = builder.compare(doc, index);

if (output) {

if (item.keys) {

for (var j = 0, jl = item.keys.length; j < jl; j++) {

var val = item.doc[item.keys[j]];

if (val === undefined)

continue;

if (typeof(val) === 'function')

doc[item.keys[j]] = val(doc[item.keys[j]], doc);

else

doc[item.keys[j]] = val;

}

} else

doc = typeof(item.doc) === 'function' ? item.doc(doc) : item.doc;

self.emit(item.keys ? 'modify' : 'update', doc);

item.count++;

change = true;

}

}

writer.write(JSON.stringify(doc) + NEWLINE);

}));

CLEANUP(writer, function() {

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

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

var item = filter[i];

if (item.insert && !item.count)

self.insert(item.insert).$callback = item.builder.$callback;

else

item.builder.$callback && item.builder.$callback(errorhandling(err, item.builder, item.count), item.count);

}

setImmediate(function() {

self.next(0);

change && setImmediate(() => self.refresh());

});

});

});

CLEANUP(reader, () => writer.end());

return self;

};

Database.prototype.$update_inmemory = function() {

var self = this;

self.step = 2;

if (!self.pending_update.length) {

self.next(0);

return self;

}

var filter = self.pending_update.splice(0);

var length = filter.length;

var change = false;

return self.$inmemory('#', function() {

var data = self.inmemory['#'];

for (var j = 0, jl = data.length; j < jl; j++) {

var doc = data[j];

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

var item = filter[i];

var builder = item.builder;

var output = builder.compare(doc, j);

if (output) {

if (item.keys) {

for (var j = 0, jl = item.keys.length; j < jl; j++) {

var val = item.doc[item.keys[j]];

if (val === undefined)

continue;

if (typeof(val) === 'function')

doc[item.keys[j]] = val(doc[item.keys[j]], doc);

else

doc[item.keys[j]] = val;

}

} else

doc = typeof(item.doc) === 'function' ? item.doc(doc) : item.doc;

self.emit(item.keys ? 'modify' : 'update', doc);

item.count++;

change = true;

}

}

}

self.$save('#');

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

var item = filter[i];

if (item.insert && !item.count)

self.insert(item.insert).$callback = item.builder.$callback;

else {

item.count && self.emit(item.keys ? 'modify' : 'update', item.doc);

item.builder.$callback && item.builder.$callback(errorhandling(null, item.builder, item.count), item.count);

}

}

setImmediate(function() {

self.next(0);

change && setImmediate(() => self.refresh());

});

});

};

Database.prototype.$reader = function() {

var self = this;

self.step = 4;

if (!self.pending_reader.length) {

self.next(0);

return self;

}

var list = self.pending_reader.splice(0);

if (INMEMORY[self.name])

self.$reader2_inmemory('#', list, () => self.next(0));

else

self.$reader2(self.filename, list, () => self.next(0));

return self;

};

Database.prototype.$readerview = function() {

var self = this;

self.step = 6;

if (!self.pending_reader_view.length) {

self.next(0);

return self;

}

var group = {};

var skip = true;

for (var i = 0, length = self.pending_reader_view.length; i < length; i++) {

var item = self.pending_reader_view[i];

var name = INMEMORY[self.name] || INMEMORY[self.name + '#' + item.view] ? item.view : self.views[item.view].$filename;

skip = false;

if (group[name])

group[name].push(item);

else

group[name] = [item];

}

self.pending_reader_view = [];

if (skip) {

self.next(0);

return self;

}

Object.keys(group).wait(function(item, next) {

if (INMEMORY[self.name] || INMEMORY[self.name + '#' + item])

self.$reader2_inmemory(item, group[item], next);

else

self.$reader2(item, group[item], next);

}, () => self.next(0), 5);

return self;

};

Database.prototype.$reader2 = function(filename, items, callback) {

var self = this;

var reader = Fs.createReadStream(filename);

var filter = items;

var length = filter.length;

reader.on('data', framework_utils.streamer(NEWLINE, function(value, index) {

var json = JSON.parse(value.trim(), jsonparser);

var val;

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

var item = filter[i];

var builder = item.builder;

var output = builder.compare(json, index);

if (!output)

continue;

item.count++;

if (!builder.$sort && ((builder.$skip && builder.$skip >= item.count) || (builder.$take && builder.$take <= item.counter)))

continue;

item.counter++;

if (item.type)

continue;

switch (builder.$scalar) {

case 'count':

item.scalar = item.scalar ? item.scalar + 1 : 1;

break;

case 'sum':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? item.scalar + val : val;

break;

case 'min':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? Math.min(item.scalar, val) : val;

break;

case 'max':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? Math.max(item.scalar, val) : val;

break;

case 'avg':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? item.scalar + val : val;

break;

case 'group':

val = json[builder.$scalarfield];

if (!item.scalar)

item.scalar = {};

if (item.scalar[val])

item.scalar[val]++;

else

item.scalar[val] = 1;

break;

default:

if (item.response)

item.response.push(output);

else

item.response = [output];

break;

}

}

}));

CLEANUP(reader, function() {

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

var item = filter[i];

var builder = item.builder;

var output;

if (builder.$scalar || !builder.$sort) {

if (builder.$scalar)

output = builder.$scalar === 'avg' ? item.scalar / item.counter : item.scalar;

else if (builder.$first)

output = item.response ? item.response[0] : undefined;

else

output = item.response || EMPTYARRAY;

builder.$callback2(errorhandling(null, builder, output), item.type === 1 ? item.count : output, item.count);

continue;

}

if (item.count) {

if (builder.$sort.name)

item.response.quicksort(builder.$sort.name, builder.$sort.asc);

else if (builder.$sort === EMPTYOBJECT)

item.response.random();

else

item.response.sort(builder.$sort);

if (builder.$skip && builder.$take)

item.response = item.response.splice(builder.$skip, builder.$take);

else if (builder.$skip)

item.response = item.response.splice(builder.$skip);

else if (builder.$take)

item.response = item.response.splice(0, builder.$take);

}

if (builder.$first)

output = item.response ? item.response[0] : undefined;

else

output = item.response || EMPTYARRAY;

builder.$callback2(errorhandling(null, builder, output), item.type === 1 ? item.count : output, item.count);

builder.done();

}

callback();

});

return self;

};

Database.prototype.$reader2_inmemory = function(name, items, callback) {

var self = this;

var filter = items;

var length = filter.length;

return self.$inmemory(name, function() {

var data = self.inmemory[name];

var val;

for (var j = 0, jl = data.length; j < jl; j++) {

var json = data[j];

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

var item = filter[i];

var builder = item.builder;

var output = builder.compare(U.clone(json), j);

if (!output)

continue;

item.count++;

if (!builder.$sort && ((builder.$skip && builder.$skip >= item.count) || (builder.$take && builder.$take <= item.counter)))

continue;

item.counter++;

if (item.type)

continue;

switch (builder.$scalar) {

case 'count':

item.scalar = item.scalar ? item.scalar + 1 : 1;

break;

case 'sum':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? item.scalar + val : val;

break;

case 'min':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? Math.min(item.scalar, val) : val;

break;

case 'max':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? Math.max(item.scalar, val) : val;

break;

case 'avg':

val = json[builder.$scalarfield] || 0;

item.scalar = item.scalar ? item.scalar + val : 0;

break;

case 'group':

val = json[builder.$scalarfield];

if (!item.scalar)

item.scalar = {};

if (item.scalar[val])

item.scalar[val]++;

else

item.scalar[val] = 1;

break;

default:

if (item.response)

item.response.push(output);

else

item.response = [output];

break;

}

}

}

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

var item = filter[i];

var builder = item.builder;

var output;

if (builder.$scalar || !builder.$sort) {

if (builder.$scalar)

output = builder.$scalar === 'avg' ? item.scalar / item.counter : item.scalar;

else if (builder.$first)

output = item.response ? item.response[0] : undefined;

else

output = item.response || EMPTYARRAY;

builder.$callback2(errorhandling(null, builder, output), item.type === 1 ? item.count : output, item.count);

continue;

}

if (item.count) {

if (builder.$sort.name)

item.response.quicksort(builder.$sort.name, builder.$sort.asc);

else if (builder.$sort === EMPTYOBJECT)

item.response.random();

else

item.response.sort(builder.$sort);

if (builder.$skip && builder.$take)

item.response = item.response.splice(builder.$skip, builder.$take);

else if (builder.$skip)

item.response = item.response.splice(builder.$skip);

else if (builder.$take)

item.response = item.response.splice(0, builder.$take);

}

if (builder.$first)

output = item.response ? item.response[0] : undefined;

else

output = item.response || EMPTYARRAY;

builder.$callback2(errorhandling(null, builder, output), item.type === 1 ? item.count : output, item.count);

builder.done();

}

callback();

});

};

Database.prototype.$views = function() {

var self = this;

self.step = 5;

if (!self.pending_views) {

self.next(0);

return self;

}

self.pending_views = false;

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

var length = views.length;

if (!length) {

self.next(0);

return self;

}

var response = [];

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

response.push({ response: [], name: views[i], builder: self.views[views[i]], count: 0, counter: 0 });

var reader = Fs.createReadStream(self.filename);

reader.on('data', framework_utils.streamer(NEWLINE, function(value, index) {

var json = JSON.parse(value.trim());

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

var item = self.views[views[j]];

var output = item.compare(json, index);

if (!output)

continue;

response[j].count++;

if (!item.$sort && ((item.$skip && item.$skip >= response[j].count) || (item.$take && item.$take < response[j].counter)))

continue;

response[j].counter++;

!item.type && response[j].response.push(output);

}

}));

CLEANUP(reader, function() {

response.wait(function(item, next) {

var builder = item.builder;

if (builder.$sort) {

if (builder.$sort.name)

item.response.quicksort(builder.$sort.name, builder.$sort.asc);

else if (builder.$sort === EMPTYOBJECT)

item.response.random();

else

item.response.sort(builder.$sort);

if (builder.$skip && builder.$take)

item.response = item.response.splice(builder.$skip, builder.$take);