// 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.7.0

*/

'use strict';

const Fs = require('fs');

const Path = require('path');

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 = {};

const CLUSTER_TIMEOUT = 150;

const CLUSTER_LOCK = { TYPE: 'nosql-lock' };

const CLUSTER_UNLOCK = { TYPE: 'nosql-unlock' };

const CLUSTER_META = { TYPE: 'nosql-meta' };

const CLUSTER_LOCK_COUNTER = { TYPE: 'nosql-counter-lock' };

const CLUSTER_UNLOCK_COUNTER = { TYPE: 'nosql-counter-unlock' };

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;

this.$events = {};

this.$free = true;

this.locked = false;

}

Database.prototype.emit = function(name, a, b, c, d, e, f, g) {

var evt = this.$events[name];

if (evt) {

var clean = false;

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

if (evt[i].$once)

clean = true;

evt[i].call(this, a, b, c, d, e, f, g);

}

if (clean) {

evt = evt.remove(n => n.$once);

if (evt.length)

this.$events[name] = evt;

else

this.$events[name] = undefined;

}

}

return this;

};

Database.prototype.on = function(name, fn) {

if (!fn.$once)

this.$free = false;

if (this.$events[name])

this.$events[name].push(fn);

else

this.$events[name] = [fn];

return this;

};

Database.prototype.once = function(name, fn) {

fn.$once = true;

return this.on(name, fn);

};

Database.prototype.removeListener = function(name, fn) {

var evt = this.$events[name];

if (evt) {

evt = evt.remove(n => n === fn);

if (evt.length)

this.$events[name] = evt;

else

this.$events[name] = undefined;

}

return this;

};

Database.prototype.removeAllListeners = function(name) {

if (name === true)

this.$events = EMPTYOBJECT;

else if (name)

this.$events[name] = undefined;

else

this.$events[name] = {};

return this;

};

exports.Database = Database;

exports.DatabaseBuilder = DatabaseBuilder;

exports.DatabaseBuilder2 = DatabaseBuilder2;

exports.DatabaseCounter = Counter;

exports.DatabaseBinary = Binary;

exports.load = function(name, filename) {

CLUSTER_LOCK.id = F.id;

CLUSTER_UNLOCK.id = F.id;

CLUSTER_META.id = F.id;

CLUSTER_LOCK_COUNTER.id = F.id;

CLUSTER_UNLOCK_COUNTER.id = F.id;

return new Database(name, filename);

};

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

if (view)

name += '#' + view;

return INMEMORY[name] = true;

};

Database.prototype.get = function(name) {

return this.meta(name);

};

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

return this.meta(name, value);

};

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;

};

function next_operation(self, type, builder) {

builder && builder.$sortinline();

self.next(type);

}

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(next_operation, self, 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();

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

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

setImmediate(next_operation, self, 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 === true ? data : insert });

setImmediate(next_operation, self, 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(next_operation, self, 7);

return self;

};

Database.prototype.free = function(force) {

var self = this;

if (!force && !self.$free)

return self;

self.counter.removeAllListeners(true);

self.binary.removeAllListeners(true);

self.removeAllListeners(true);

delete F.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(next_operation, self, 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(next_operation, self, 6, builder);

} else {

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

setImmediate(next_operation, self, 4, builder);

}

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(next_operation, self, 6);

} else {

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

setImmediate(next_operation, self, 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(next_operation, self, 6, builder);

} else {

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

setImmediate(next_operation, self, 4, builder);

}

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(next_operation, self, 6, builder);

} else {

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

setImmediate(next_operation, self, 4, builder);

}

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.lock = function() {

this.locked = true;

return this;

};

Database.prototype.unlock = function() {

this.locked = false;

next_operation(this, 0);

return this;

};

Database.prototype.next = function(type) {

if (this.locked || (type && this.step))

return this;

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

this.$readerview();

return this;

}

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

this.$reader();

return this;

}

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

if (INMEMORY[this.name])

this.$append_inmemory();

else

this.$append();

return this;

}

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

if (INMEMORY[this.name])

this.$update_inmemory();

else

this.$update();

return this;

}

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

if (INMEMORY[this.name])

this.$remove_inmemory();

else

this.$remove();

return this;

}

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

if (INMEMORY[this.name])

this.$views_inmemory();

else

this.$views();

return this;

}

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

this.$drop();

return this;

}

if (this.step !== type) {

this.step = 0;

setImmediate(next_operation, this, 0);

}

return this;

};

Database.prototype.refresh = function() {

if (this.views) {

this.pending_views = true;

setImmediate(next_operation, this, 5);

} else if (F.cluster)

this.$unlock();

return this;

};

Database.prototype.$lock = function() {

if (this.lockwait === 2)

return false;

CLUSTER_LOCK.name = this.name;

process.send(CLUSTER_LOCK);

if (this.lockwait === 1)

return true;

if (this.lockwait === 2)

return false;

this.lockwait = 1;

setTimeout(locker_timeout, CLUSTER_TIMEOUT, this);

return true;

};

function locker_timeout(self) {

self.lockwait = 2;

self.next(0);

}

Database.prototype.$unlock = function() {

if (!this.lockwait)

return;

this.lockwait = 0;

CLUSTER_UNLOCK.name = this.name;

process.send(CLUSTER_UNLOCK);

};

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

// 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, 100);

return self;

};

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

var self = this;

// Last usage

self.inmemorylastusage = global.F ? global.F.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), function() {

if (F.isCluster) {

CLUSTER_META.name = self.name;

process.send(CLUSTER_META);

}

});

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.locked || !self.pending_append.length) {

self.next(0);

return;

}

if (F.isCluster && self.$lock())

return;

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

});

}, () => setImmediate(next_append, self));

};

function next_append(self) {

self.next(0);

// $unlock() is in refresh()

setImmediate(views_refresh, self);

}

Database.prototype.$append_inmemory = function() {

var self = this;

self.step = 1;

if (self.locked || !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(next_append, self);

});

};

Database.prototype.$update = function() {

var self = this;

self.step = 2;

if (self.locked || !self.pending_update.length) {

self.next(0);

return self;

}

if (F.isCluster && self.$lock())

return;

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

if (change)

setImmediate(views_refresh, self);

else if (F.isCluster)

self.$unlock();

});

});

});

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

return self;

};

function views_refresh(self) {

self.refresh();

}

Database.prototype.$update_inmemory = function() {

var self = this;

self.step = 2;

if (self.locked || !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(views_refresh, self);

});

});

};

Database.prototype.$reader = function() {

var self = this;

self.step = 4;

if (self.locked || !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.locked || !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.$inlinesort && ((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 (builder.$inlinesort)

nosqlinlinesorter(item, builder, output);

else 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) {

if (!builder.$inlinesort || builder.$take !== item.response.length)

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.$inlinesort && 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;

};

function nosqlinlinesorter(item, builder, doc) {

if (!item.response) {

item.response = [doc];

return;

}

var length = item.response.length;

if (length < builder.$limit) {

item.response.push(doc);

length + 1 >= builder.$limit && item.response.quicksort(builder.$sort.name, builder.$sort.asc);

} else

nosqlresort(item.response, builder, doc);

}

function nosqlsortvalue(a, b, sorter) {

var type = typeof(a);

if (type === 'number')

return sorter.asc ? a > b : a < b;

else if (type === 'string') {

a = (a.length > 5 ? a.substring(0, 5) : a).toLowerCase().removeDiacritics();

var c = a.localeCompare(b);

return sorter.asc ? c === 1 : c === -1;

} else if (a instanceof Date)

return sorter.asc ? a > b : a < b;

return false;

}

function nosqlresort(arr, builder, doc) {

var b = doc[builder.$sort.name];

if (typeof(b) === 'string')

b = (b.length > 5 ? b.substring(0, 5) : b).toLowerCase().removeDiacritics();

var beg = 0;