From 3b9f9f273c338be208f8aa39f90e02a0afdd8c0c Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Fri, 24 Aug 2018 04:28:08 -0700 Subject: [PATCH] Convert db.coffee to JS --- .gitignore | 1 - lib/db.coffee | 652 ----------------------------------- lib/db.js | 917 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 917 insertions(+), 653 deletions(-) delete mode 100644 lib/db.coffee create mode 100644 lib/db.js diff --git a/.gitignore b/.gitignore index 1536944..a4e333f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,6 @@ *~ lib/*.map lib/cwmp.js -lib/db.js lib/session.js node_modules debug diff --git a/lib/db.coffee b/lib/db.coffee deleted file mode 100644 index 6416718..0000000 --- a/lib/db.coffee +++ /dev/null @@ -1,652 +0,0 @@ -### -# Copyright 2013-2017 Zaid Abdulla -# -# This file is part of GenieACS. -# -# GenieACS is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# GenieACS is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with GenieACS. If not, see . -### - -config = require './config' -mongodb = require 'mongodb' -common = require './common' - -tasksCollection = null -devicesCollection = null -presetsCollection = null -objectsCollection = null -provisionsCollection = null -virtualParametersCollection = null -faultsCollection = null -filesCollection = null -operationsCollection = null - - -connect = (callback) -> - callbackCounter = 9 - mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), (err, db) -> - return callback(err) if err - exports.mongoDb = db - db.collection('tasks', (err, collection) -> - exports.tasksCollection = tasksCollection = collection - collection?.ensureIndex({device: 1, timestamp: 1}, (err) -> - ) - - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('devices', (err, collection) -> - exports.devicesCollection = devicesCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('presets', (err, collection) -> - exports.presetsCollection = presetsCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('objects', (err, collection) -> - exports.objectsCollection = objectsCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('fs.files', (err, collection) -> - exports.filesCollection = filesCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('provisions', (err, collection) -> - exports.provisionsCollection = provisionsCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('virtualParameters', (err, collection) -> - exports.virtualParametersCollection = virtualParametersCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('faults', (err, collection) -> - exports.faultsCollection = faultsCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - db.collection('operations', (err, collection) -> - exports.operationsCollection = operationsCollection = collection - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) - - ) - - -disconnect = () -> - exports.mongoDb?.close() - - -# Optimize projection by removing overlaps -# This can modify the object -optimizeProjection = (obj) -> - if obj[''] - return {'': obj['']} - - keys = Object.keys(obj).sort() - return obj if keys.length <= 1 - i = 1 - while i < keys.length - a = keys[i-1] - b = keys[i] - if b.startsWith(a) - if b.charAt(a.length) == '.' or b.charAt(a.length - 1) == '.' - delete obj[b] - keys.splice(i--, 1) - ++ i - return obj - - -fetchDevice = (id, timestamp, patterns, callback) -> - MAX_DEPTH = config.get('MAX_DEPTH', id) - - if not patterns?.length - patterns = [[[], (1 << MAX_DEPTH) - 1]] - - projection = {} - projectionTree = {} - func = (path, pats, projTree) -> - children = {} - for pat in pats - fragment = pat[0][path.length] or '*' - - if fragment == '*' - projection[path.join('.')] = 1 if pat[1] << path.length - return - - projTree[fragment] ?= {} - - if pat[1] & (1 << path.length) - projection[path.concat('_timestamp').join('.')] = 1 - projection[path.concat('_object').join('.')] = 1 - projection[path.concat('_instance').join('.')] = 1 - projection[path.concat([fragment, '_timestamp']).join('.')] = 1 - projection[path.concat([fragment, '_value']).join('.')] = 1 - projection[path.concat([fragment, '_type']).join('.')] = 1 - projection[path.concat([fragment, '_object']).join('.')] = 1 - projection[path.concat([fragment, '_instance']).join('.')] = 1 - projection[path.concat([fragment, '_writable']).join('.')] = 1 - projection[path.concat([fragment, '_orig']).join('.')] = 1 - - if pat[1] >> (path.length + 1) - children[fragment] ?= [] - children[fragment].push(pat) - - for k, v of children - func(path.concat(k), v, projTree[k]) - - func([], patterns, projectionTree) - - delete projectionTree['DeviceID'] - delete projectionTree['Events'] - delete projectionTree['Tags'] - - res = [] - loaded = [] - - projection = optimizeProjection(projection) - - for k, v of projection - if k == '' or k == 'Events' or k.startsWith('Events.') - if k == '' or k == 'Events' or k == 'Events._writable' - res.push([['Events'], timestamp, - {object: [timestamp, 1], writable: [timestamp, 0]}]) - loaded.push([['Events'], 1]) if k == 'Events._writable' - if k == 'Events' - projection['_lastInform'] = 1 - projection['_lastBoot'] = 1 - projection['_lastBootstrap'] = 1 - projection['_registered'] = 1 - loaded.push([['Events'], ((1 << MAX_DEPTH) - 1)]) - else if k == 'Events.Inform._writable' or k == 'Events.Inform' - projection['_lastInform'] = 1 - loaded.push([['Events', 'Inform'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'Events.1_BOOT._writable' or k == 'Events.1_BOOT' - projection['_lastBoot'] = 1 - loaded.push([['Events', '1_BOOT'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'Events.0_BOOTSTRAP._writable' or k == 'Events.0_BOOTSTRAP' - projection['_lastBootstrap'] = 1 - loaded.push([['Events', '0_BOOTSTRAP'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'Events.Registered._writable' or k == 'Events.Registered' - projection['_registered'] = 1 - loaded.push([['Events', 'Registered'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k.endsWith('._writable') and k != 'Events._writable' - loaded.push([k.split('.').slice(0, 2), 1 ^ ((1 << MAX_DEPTH) - 1)]) - delete projection[k] if k != '' - - if k == '' or k == 'DeviceID' or k.startsWith('DeviceID.') - if k == '' or k == 'DeviceID' or k == 'DeviceID._writable' - res.push([['DeviceID'], timestamp, - {object: [timestamp, 1], writable: [timestamp, 0]}]) - loaded.push([['DeviceID'], 1]) if k == 'DeviceID._writable' - if k == 'DeviceID' - projection['_id'] = 1 - projection['_deviceId'] = 1 - loaded.push([['DeviceID'], ((1 << MAX_DEPTH) - 1)]) - else if k == 'DeviceID.ID._writable' or k == 'DeviceID.ID' - projection['_id'] = 1 - loaded.push([['DeviceID', 'ID'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'DeviceID.Manufacturer._writable' or k == 'DeviceID.DeManufacturer' - projection['_deviceId._Manufacturer'] = 1 - loaded.push([['DeviceID', 'Manufacturer'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'DeviceID.ProductClass._writable' or k == 'DeviceID.ProductClass' - projection['_deviceId._ProductClass'] = 1 - loaded.push([['DeviceID', 'ProductClass'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'DeviceID.OUI._writable' or k == 'DeviceID.OUI' - projection['_deviceId._OUI'] = 1 - loaded.push([['DeviceID', 'ProductClass'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k == 'DeviceID.SerialNumber._writable' or k == 'DeviceID.SerialNumber' - projection['_deviceId._SerialNumber'] = 1 - loaded.push([['DeviceID', 'SerialNumber'], 1 ^ ((1 << MAX_DEPTH) - 1)]) - else if k.endsWith('._writable') and k != 'DeviceID._writable' - loaded.push([k.split('.').slice(0, 2), 1 ^ ((1 << MAX_DEPTH) - 1)]) - delete projection[k] if k != '' - - if k == 'Tags' or k.startsWith('Tags.') - if not projection['_tags'] - projection['_tags'] = 1 - loaded.push([['Tags'], ((1 << MAX_DEPTH) - 1)]) - delete projection[k] - - if projection[''] - proj = {} - else if Object.keys(projection).length == 0 - proj = {'_id': 1} - else - proj = projection - - devicesCollection.findOne({'_id' : id}, proj, (err, device) -> - return callback(err) if err or not device? - - storeParams = (obj, path, timestamp, descendantsFetched, projTree) -> - if descendantsFetched - thisFetched = true - else - if projection[path.join('.')] - descendantsFetched = true - if path.length and projection[path.slice(0, -1).concat("_timestamp").join('.')] - thisFetched = true - loaded.push([path, ((1 << (path.length - 1)) - 1) ^ ((1 << MAX_DEPTH) - 1)]) - else - loaded.push([path, ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1)]) - else if projection[path.concat('_writable').join('.')] - loaded.push([path, 1 << (path.length - 1)]) - thisFetched = true - - if obj['_timestamp']? - obj['_timestamp'] = +obj['_timestamp'] - - # For compatibility with v1.0 database - if obj['_instance'] and not obj['_object']? - obj['_object'] = true - - if thisFetched - attrs = {} - - t = obj['_timestamp'] or 1 - t = timestamp if timestamp > t - - if obj['_value']? - attrs.value = [obj['_timestamp'] or 1, [obj['_value'], obj['_type']]] - attrs.value[1][0] = +attrs.value[1][0] if obj['_type'] == 'xsd:dateTime' - obj['_object'] = false - - if obj['_writable']? - attrs.writable = [timestamp or 1, if obj['_writable'] then 1 else 0] - - if obj['_object']? - attrs.object = [t, if obj['_object'] then 1 else 0] - - res.push([path, t, attrs]) - - for k, v of obj - if not k.startsWith('_') - kk = k - obj['_object'] = true - storeParams(v, path.concat(k), obj['_timestamp'], descendantsFetched, projTree?[k]) - delete projTree[kk] if projTree - - if not descendantsFetched - for k, v of projTree - p = path.concat(k) - loaded.push([p, ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1)]) - if (obj['_object'] or path.length == 0) and obj['_timestamp'] - res.push([p, obj['_timestamp']]) - else if (obj['_object'] or path.length == 0) and obj['_timestamp'] - res.push([path.concat('*'), obj['_timestamp']]) - - for k, v of device - switch k - when '_lastInform' - res.push([['Events', 'Inform'], +v, - {object: [+v, 0], writable: [+v, 0], value: [+v, [+v, 'xsd:dateTime']]}]) - delete device[k] - when '_lastBoot' - res.push([['Events', '1_BOOT'], +v, - {object: [+v, 0], writable: [+v, 0], value: [+v, [+v, 'xsd:dateTime']]}]) - delete device[k] - when '_lastBootstrap' - res.push([['Events', '0_BOOTSTRAP'], +v, - {object: [+v, 0], writable: [+v, 0], value: [+v, [+v, 'xsd:dateTime']]}]) - delete device[k] - when '_registered' - # Use current timestamp for registered event attribute timestamps - res.push([['Events', 'Registered'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [+v, 'xsd:dateTime']]}]) - delete device[k] - when '_id' - if projection[''] or projection['_id'] - res.push([['DeviceID', 'ID'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [v, 'xsd:string']]}]) - delete device[k] - when '_tags' - if v.length - res.push([['Tags'], timestamp, - {object: [timestamp, 1], writable: [timestamp, 0]}]) - for t in v - t = t.replace(/[^a-zA-Z0-9\-]+/g, '_') - res.push([['Tags', t], timestamp, - {object: [timestamp, 0], writable: [timestamp, 1], value: [timestamp, [true, 'xsd:boolean']]}]) - delete device[k] - when '_deviceId' - if v['_Manufacturer']? - res.push([['DeviceID', 'Manufacturer'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [v['_Manufacturer'], 'xsd:string']]}]) - if v['_OUI']? - res.push([['DeviceID', 'OUI'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [v['_OUI'], 'xsd:string']]}]) - if v['_ProductClass']? - res.push([['DeviceID', 'ProductClass'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [v['_ProductClass'], 'xsd:string']]}]) - if v['_SerialNumber']? - res.push([['DeviceID', 'SerialNumber'], timestamp, - {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [v['_SerialNumber'], 'xsd:string']]}]) - delete device[k] - - storeParams(device, [], 0, false, projectionTree) - - return callback(null, res, loaded) - ) - - -saveDevice = (deviceId, deviceData, isNew, sessionTimestamp, callback) -> - update = {'$set' : {}, '$unset' : {}, '$addToSet' : {}, '$pull' : {}} - - iter = deviceData.timestamps.diff() - while diff = iter.next().value - continue if diff[0].wildcard != (1 << (diff[0].length - 1)) - continue if diff[0][0] in ['Events', 'DeviceID', 'Tags'] - - # Param timestamps may be greater than session timestamp to track revisions - if diff[2] > sessionTimestamp - diff[2] = sessionTimestamp - - if not diff[2]? and diff[1]? - update['$unset'][diff[0].slice(0, -1).concat('_timestamp').join('.')] = 1 - else if diff[2] != diff[1] - parent = deviceData.paths.get(diff[0].slice(0, -1)) - if parent and (parent.length == 0 or deviceData.attributes.has(parent)) - update['$set'][diff[0].slice(0, -1).concat('_timestamp').join('.')] = new Date(diff[2]) - - iter = deviceData.attributes.diff() - while diff = iter.next().value - continue if diff[1] == diff[2] - path = diff[0] - switch path[0] - when 'Events' - if diff[0].length == 2 and diff[2]?.value?[1][0] != diff[1]?.value?[1][0] - if not diff[2] - switch path[1] - when 'Inform' - update['$unset']['_lastInform'] = 1 - when '1_BOOT' - update['$unset']['_lastBoot'] = 1 - when '0_BOOTSTRAP' - update['$unset']['_lastBootstrap'] = 1 - when 'Registered' - update['$unset']['_registered'] = 1 - else - t = new Date(diff[2].value[1][0]) - switch path[1] - when 'Inform' - update['$set']['_lastInform'] = t - when '1_BOOT' - update['$set']['_lastBoot'] = t - when '0_BOOTSTRAP' - update['$set']['_lastBootstrap'] = t - when 'Registered' - update['$set']['_registered'] = t - when 'DeviceID' - if diff[2].value?[1]?[0] != diff[1]?.value?[1]?[0] - v = diff[2].value[1][0] - switch path[1] - when 'ID' - update['$set']['_id'] = v - when 'Manufacturer' - update['$set']['_deviceId._Manufacturer'] = v - when 'OUI' - update['$set']['_deviceId._OUI'] = v - when 'ProductClass' - update['$set']['_deviceId._ProductClass'] = v - when 'SerialNumber' - update['$set']['_deviceId._SerialNumber'] = v - when 'Tags' - if diff[2]?.value?[1]?[0] != diff[1]?.value?[1]?[0] - v = diff[2]?.value?[1][0] - if v? - update['$addToSet']['_tags'] ?= {'$each' : []} - update['$addToSet']['_tags']['$each'].push(diff[0][1]) - else - update['$pull']['_tags'] ?= {'$in' : []} - update['$pull']['_tags']['$in'].push(diff[0][1]) - else - if not diff[2] - update['$unset'][diff[0].join('.')] = 1 - continue - - for attrName of diff[2] - # Param timestamps may be greater than session timestamp to track revisions - if diff[2][attrName][0] > sessionTimestamp - diff[2][attrName][0] = sessionTimestamp - - if diff[2][attrName][1]? - switch attrName - when 'value' - if diff[2].value[1][0] != diff[1]?.value?[1]?[0] - if diff[2].value[1][1] == 'xsd:dateTime' and Number.isInteger(diff[2].value[1][0]) - update['$set'][path.concat('_value').join('.')] = new Date(diff[2].value[1][0]) - else - update['$set'][path.concat('_value').join('.')] = diff[2].value[1][0] - if diff[2].value[1][1] != diff[1]?.value?[1]?[1] - update['$set'][path.concat('_type').join('.')] = diff[2].value[1][1] - if diff[2].value[0] != diff[1]?.value?[0] - update['$set'][path.concat('_timestamp').join('.')] = new Date(diff[2].value[0]) - when 'object' - if not diff[1]?.object or diff[2].object[1] != diff[1].object?[1] - update['$set'][path.concat('_object').join('.')] = !!diff[2].object[1] - when 'writable' - if not diff[1]?.writable or diff[2].writable[1] != diff[1].writable?[1] - update['$set'][path.concat('_writable').join('.')] = !!diff[2].writable[1] - - if diff[1] - for attrName of diff[1] - if diff[1][attrName][1]? and not diff[2]?[attrName]?[1]? - update['$unset'][path.concat("_#{attrName}").join('.')] = 1 - if attrName is 'value' - update['$unset'][path.concat('_type').join('.')] = 1 - update['$unset'][path.concat('_timestamp').join('.')] = 1 - - update['$unset'] = optimizeProjection(update['$unset']) - - # Remove overlap possibly caused by parameters changing from objects - # to regular parameters or vice versa. Reason being that _timestamp - # represents two different things depending on whether the parameter - # is an object or not. - for k of update['$unset'] - delete update['$unset'][k] if update['$set'][k]? - - # Remove empty keys - for k of update - if k == '$addToSet' - for kk of update[k] - delete update[k][kk] if update[k][kk]['$each'].length == 0 - else if k == '$pull' - for kk of update[k] - delete update[k][kk] if update[k][kk]['$in'].length == 0 - delete update[k] if Object.keys(update[k]).length == 0 - - return callback() if Object.keys(update).length == 0 - - if update['$addToSet'] and update['$pull'] - # Mongo doesn't allow $addToSet and $pull at the same time - update2 = {'$pull': update['$pull']} - delete update['$pull'] - - devicesCollection.update({'_id' : deviceId}, update, {upsert: isNew}, (err, result) -> - if not err and result.result.n != 1 - return callback(new Error("Device #{deviceId} not found in database")) - - if update2 - return devicesCollection.update({'_id': deviceId}, update2, callback) - - return callback(err) - ) - - -getFaults = (deviceId, callback) -> - faultsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> - return callback(err) if err - - faults = {} - for r in res - channel = r._id.slice(deviceId.length + 1) - delete r._id - delete r.channel - delete r.device - r.timestamp = +r.timestamp - r.provisions = JSON.parse(r.provisions) - faults[channel] = r - - callback(err, faults) - ) - - -saveFault = (deviceId, channel, fault, callback) -> - fault = Object.assign({}, fault) - fault._id = "#{deviceId}:#{channel}" - fault.device = deviceId - fault.channel = channel - fault.timestamp = new Date(fault.timestamp) - fault.provisions = JSON.stringify(fault.provisions) - faultsCollection.save(fault, callback) - - -deleteFault = (deviceId, channel, callback) -> - faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, callback) - - -getDueTasks = (deviceId, timestamp, callback) -> - cur = tasksCollection.find({'device' : deviceId}).sort(['timestamp']) - - tasks = [] - - cur.nextObject(f = (err, task) -> - return callback(err) if err - - if not task? - return callback(null, tasks, null) - - task.timestamp = +task.timestamp if task.timestamp? - task.expiry = +task.expiry if task.expiry? - - if task.timestamp >= timestamp - return callback(null, tasks, +task.timestamp) - - task._id = String(task._id) - tasks.push(task) - - # For API compatibility - if task.name is 'download' and task.file? - if mongodb.ObjectID.isValid(task.file) - q = {_id: {'$in' : [task.file, new mongodb.ObjectID(task.file)]}} - else - q = {_id: task.file} - - filesCollection.find(q).toArray((err, res) -> - return callback(err) if err - if res[0]? - task.fileType ?= res[0].metadata.fileType - task.fileName ?= res[0].filename or res[0]._id.toString() - cur.nextObject(f) - ) - else - cur.nextObject(f) - ) - - -clearTasks = (deviceId, taskIds, callback) -> - tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback) - - -getOperations = (deviceId, callback) -> - operationsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> - return callback(err) if err - - operations = {} - for r in res - commandKey = r._id.slice(deviceId.length + 1) - delete r._id - r.timestamp = +r.timestamp - r.args = JSON.parse(r.args) if r.args - r.provisions = JSON.parse(r.provisions) - r.retries = JSON.parse(r.retries) - operations[commandKey] = r - - callback(err, operations) - ) - - -saveOperation = (deviceId, commandKey, operation, callback) -> - operation = Object.assign({}, operation) - operation._id = "#{deviceId}:#{commandKey}" - operation.timestamp = new Date(operation.timestamp) - operation.provisions = JSON.stringify(operation.provisions) - operation.retries = JSON.stringify(operation.retries) - operation.args = JSON.stringify(operation.args) - operationsCollection.save(operation, callback) - - -deleteOperation = (deviceId, commandKey, callback) -> - operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, callback) - - -getPresets = (callback) -> - presetsCollection.find().toArray(callback) - - -getObjects = (callback) -> - objectsCollection.find().toArray(callback) - - -getProvisions = (callback) -> - provisionsCollection.find().toArray(callback) - - -getVirtualParameters = (callback) -> - virtualParametersCollection.find().toArray(callback) - - -getFiles = (callback) -> - filesCollection.find().toArray(callback) - - -exports.connect = connect -exports.disconnect = disconnect -exports.fetchDevice = fetchDevice -exports.saveDevice = saveDevice -exports.getFaults = getFaults -exports.saveFault = saveFault -exports.deleteFault = deleteFault -exports.clearTasks = clearTasks -exports.saveOperation = saveOperation -exports.deleteOperation = deleteOperation -exports.getPresets = getPresets -exports.getObjects = getObjects -exports.getProvisions = getProvisions -exports.getVirtualParameters = getVirtualParameters -exports.getFiles = getFiles -exports.getDueTasks = getDueTasks -exports.getOperations = getOperations diff --git a/lib/db.js b/lib/db.js new file mode 100644 index 0000000..efd3311 --- /dev/null +++ b/lib/db.js @@ -0,0 +1,917 @@ +/** + * Copyright 2013-2018 Zaid Abdulla + * + * This file is part of GenieACS. + * + * GenieACS is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * GenieACS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with GenieACS. If not, see . + */ +"use strict"; + +const mongodb = require("mongodb"); +const config = require("./config"); +const common = require("./common"); + +let tasksCollection, + devicesCollection, + presetsCollection, + objectsCollection, + provisionsCollection, + virtualParametersCollection, + faultsCollection, + filesCollection, + operationsCollection; + +function connect(callback) { + let callbackCounter = 9; + mongodb.MongoClient.connect( + config.get("MONGODB_CONNECTION_URL"), + (err, db) => { + if (err) return callback(err); + + exports.mongoDb = db; + db.collection("tasks", (err, collection) => { + exports.tasksCollection = tasksCollection = collection; + if (collection != null) + collection.ensureIndex({ device: 1, timestamp: 1 }); + + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("devices", (err, collection) => { + exports.devicesCollection = devicesCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("presets", (err, collection) => { + exports.presetsCollection = presetsCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("objects", (err, collection) => { + exports.objectsCollection = objectsCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("fs.files", (err, collection) => { + exports.filesCollection = filesCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("provisions", (err, collection) => { + exports.provisionsCollection = provisionsCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("virtualParameters", (err, collection) => { + exports.virtualParametersCollection = virtualParametersCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + db.collection("faults", (err, collection) => { + exports.faultsCollection = faultsCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + return db.collection("operations", (err, collection) => { + exports.operationsCollection = operationsCollection = collection; + if (--callbackCounter === 0 || err) { + callbackCounter = 0; + return callback(err); + } + }); + } + ); +} + +function disconnect() { + if (exports.mongoDb) exports.mongoDb.close(); +} + +// Optimize projection by removing overlaps +// This can modify the object +function optimizeProjection(obj) { + if (obj[""]) return { "": obj[""] }; + + const keys = Object.keys(obj).sort(); + if (keys.length <= 1) return obj; + + for (let i = 1; i < keys.length; ++i) { + const a = keys[i - 1]; + const b = keys[i]; + if (b.startsWith(a)) { + if (b.charAt(a.length) === "." || b.charAt(a.length - 1) === ".") { + delete obj[b]; + keys.splice(i--, 1); + } + } + } + return obj; +} + +function fetchDevice(id, timestamp, patterns, callback) { + const MAX_DEPTH = config.get("MAX_DEPTH", id); + + if (!patterns || !patterns.length) patterns = [[[], (1 << MAX_DEPTH) - 1]]; + + let projection = {}; + const projectionTree = {}; + function func(path, pats, projTree) { + const children = {}; + for (let pat of pats) { + const fragment = pat[0][path.length] || "*"; + if (fragment === "*") { + if (pat[1] << path.length) projection[path.join(".")] = 1; + return; + } + + if (!projTree[fragment]) projTree[fragment] = {}; + + if (pat[1] & (1 << path.length)) { + projection[path.concat("_timestamp").join(".")] = 1; + projection[path.concat("_object").join(".")] = 1; + projection[path.concat("_instance").join(".")] = 1; + projection[path.concat([fragment, "_timestamp"]).join(".")] = 1; + projection[path.concat([fragment, "_value"]).join(".")] = 1; + projection[path.concat([fragment, "_type"]).join(".")] = 1; + projection[path.concat([fragment, "_object"]).join(".")] = 1; + projection[path.concat([fragment, "_instance"]).join(".")] = 1; + projection[path.concat([fragment, "_writable"]).join(".")] = 1; + projection[path.concat([fragment, "_orig"]).join(".")] = 1; + } + + if (pat[1] >> (path.length + 1)) { + if (!children[fragment]) children[fragment] = []; + children[fragment].push(pat); + } + } + + for (let [k, v] of Object.entries(children)) + func(path.concat(k), v, projTree[k]); + } + + func([], patterns, projectionTree); + + delete projectionTree["DeviceID"]; + delete projectionTree["Events"]; + delete projectionTree["Tags"]; + + const res = []; + const loaded = []; + + projection = optimizeProjection(projection); + + for (let k of Object.keys(projection)) { + if (k === "" || k === "Events" || k.startsWith("Events.")) { + if (k === "" || k === "Events" || k === "Events._writable") { + res.push([ + ["Events"], + timestamp, + { object: [timestamp, 1], writable: [timestamp, 0] } + ]); + if (k === "Events._writable") loaded.push([["Events"], 1]); + } + if (k === "Events") { + projection["_lastInform"] = 1; + projection["_lastBoot"] = 1; + projection["_lastBootstrap"] = 1; + projection["_registered"] = 1; + loaded.push([["Events"], (1 << MAX_DEPTH) - 1]); + } else if (k === "Events.Inform._writable" || k === "Events.Inform") { + projection["_lastInform"] = 1; + loaded.push([["Events", "Inform"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if (k === "Events.1_BOOT._writable" || k === "Events.1_BOOT") { + projection["_lastBoot"] = 1; + loaded.push([["Events", "1_BOOT"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if ( + k === "Events.0_BOOTSTRAP._writable" || + k === "Events.0_BOOTSTRAP" + ) { + projection["_lastBootstrap"] = 1; + loaded.push([["Events", "0_BOOTSTRAP"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if ( + k === "Events.Registered._writable" || + k === "Events.Registered" + ) { + projection["_registered"] = 1; + loaded.push([["Events", "Registered"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if (k.endsWith("._writable") && k !== "Events._writable") { + loaded.push([k.split(".").slice(0, 2), 1 ^ ((1 << MAX_DEPTH) - 1)]); + } + if (k !== "") delete projection[k]; + } + + if (k === "" || k === "DeviceID" || k.startsWith("DeviceID.")) { + if (k === "" || k === "DeviceID" || k === "DeviceID._writable") { + res.push([ + ["DeviceID"], + timestamp, + { object: [timestamp, 1], writable: [timestamp, 0] } + ]); + if (k === "DeviceID._writable") loaded.push([["DeviceID"], 1]); + } + if (k === "DeviceID") { + projection["_id"] = 1; + projection["_deviceId"] = 1; + loaded.push([["DeviceID"], (1 << MAX_DEPTH) - 1]); + } else if (k === "DeviceID.ID._writable" || k === "DeviceID.ID") { + projection["_id"] = 1; + loaded.push([["DeviceID", "ID"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if ( + k === "DeviceID.Manufacturer._writable" || + k === "DeviceID.DeManufacturer" + ) { + projection["_deviceId._Manufacturer"] = 1; + loaded.push([["DeviceID", "Manufacturer"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if ( + k === "DeviceID.ProductClass._writable" || + k === "DeviceID.ProductClass" + ) { + projection["_deviceId._ProductClass"] = 1; + loaded.push([["DeviceID", "ProductClass"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if (k === "DeviceID.OUI._writable" || k === "DeviceID.OUI") { + projection["_deviceId._OUI"] = 1; + loaded.push([["DeviceID", "ProductClass"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if ( + k === "DeviceID.SerialNumber._writable" || + k === "DeviceID.SerialNumber" + ) { + projection["_deviceId._SerialNumber"] = 1; + loaded.push([["DeviceID", "SerialNumber"], 1 ^ ((1 << MAX_DEPTH) - 1)]); + } else if (k.endsWith("._writable") && k !== "DeviceID._writable") { + loaded.push([k.split(".").slice(0, 2), 1 ^ ((1 << MAX_DEPTH) - 1)]); + } + if (k !== "") delete projection[k]; + } + + if (k === "Tags" || k.startsWith("Tags.")) { + if (!projection["_tags"]) { + projection["_tags"] = 1; + loaded.push([["Tags"], (1 << MAX_DEPTH) - 1]); + } + delete projection[k]; + } + } + + let proj; + if (projection[""]) proj = {}; + else if (!Object.keys(projection).length) proj = { _id: 1 }; + else proj = projection; + + devicesCollection.findOne({ _id: id }, proj, (err, device) => { + if (err || !device) return callback(err); + + function storeParams(obj, path, ts, descendantsFetched, projTree) { + let thisFetched = false; + if (descendantsFetched) { + thisFetched = true; + } else if (projection[path.join(".")]) { + descendantsFetched = true; + if ( + path.length && + projection[ + path + .slice(0, -1) + .concat("_timestamp") + .join(".") + ] + ) { + thisFetched = true; + loaded.push([ + path, + ((1 << (path.length - 1)) - 1) ^ ((1 << MAX_DEPTH) - 1) + ]); + } else { + loaded.push([ + path, + ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1) + ]); + } + } else if (projection[path.concat("_writable").join(".")]) { + loaded.push([path, 1 << (path.length - 1)]); + thisFetched = true; + } + + if (obj["_timestamp"]) obj["_timestamp"] = +obj["_timestamp"]; + + // For compatibility with v1.0 database + if (obj["_instance"] && obj["_object"] == null) obj["_object"] = true; + + if (thisFetched) { + const attrs = {}; + let t = obj["_timestamp"] || 1; + if (ts > t) t = ts; + + if (obj["_value"] != null) { + attrs.value = [obj["_timestamp"] || 1, [obj["_value"], obj["_type"]]]; + if (obj["_type"] === "xsd:dateTime") + attrs.value[1][0] = +attrs.value[1][0]; + + obj["_object"] = false; + } + if (obj["_writable"] != null) + attrs.writable = [ts || 1, obj["_writable"] ? 1 : 0]; + + if (obj["_object"] != null) attrs.object = [t, obj["_object"] ? 1 : 0]; + + res.push([path, t, attrs]); + } + + for (let [k, v] of Object.entries(obj)) { + if (!k.startsWith("_")) { + obj["_object"] = true; + storeParams( + v, + path.concat(k), + obj["_timestamp"], + descendantsFetched, + projTree ? projTree[k] : null + ); + if (projTree) delete projTree[k]; + } + } + + if (!descendantsFetched) { + for (let k of Object.keys(projTree)) { + const p = path.concat(k); + loaded.push([p, ((1 << path.length) - 1) ^ ((1 << MAX_DEPTH) - 1)]); + if ((obj["_object"] || !path.length) && obj["_timestamp"]) + res.push([p, obj["_timestamp"]]); + } + } else if ((obj["_object"] || !path.length) && obj["_timestamp"]) { + res.push([path.concat("*"), obj["_timestamp"]]); + } + } + + for (let [k, v] of Object.entries(device)) { + switch (k) { + case "_lastInform": + res.push([ + ["Events", "Inform"], + +v, + { + object: [+v, 0], + writable: [+v, 0], + value: [+v, [+v, "xsd:dateTime"]] + } + ]); + delete device[k]; + break; + case "_lastBoot": + res.push([ + ["Events", "1_BOOT"], + +v, + { + object: [+v, 0], + writable: [+v, 0], + value: [+v, [+v, "xsd:dateTime"]] + } + ]); + delete device[k]; + break; + case "_lastBootstrap": + res.push([ + ["Events", "0_BOOTSTRAP"], + +v, + { + object: [+v, 0], + writable: [+v, 0], + value: [+v, [+v, "xsd:dateTime"]] + } + ]); + delete device[k]; + break; + case "_registered": + // Use current timestamp for registered event attribute timestamps + res.push([ + ["Events", "Registered"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [+v, "xsd:dateTime"]] + } + ]); + delete device[k]; + break; + case "_id": + if (projection[""] || projection["_id"]) { + res.push([ + ["DeviceID", "ID"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [v, "xsd:string"]] + } + ]); + } + + delete device[k]; + break; + case "_tags": + if (v.length) { + res.push([ + ["Tags"], + timestamp, + { object: [timestamp, 1], writable: [timestamp, 0] } + ]); + } + + for (let t of v) { + t = t.replace(/[^a-zA-Z0-9-]+/g, "_"); + res.push([ + ["Tags", t], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 1], + value: [timestamp, [true, "xsd:boolean"]] + } + ]); + } + delete device[k]; + break; + case "_deviceId": + if (v["_Manufacturer"] != null) { + res.push([ + ["DeviceID", "Manufacturer"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [v["_Manufacturer"], "xsd:string"]] + } + ]); + } + + if (v["_OUI"] != null) { + res.push([ + ["DeviceID", "OUI"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [v["_OUI"], "xsd:string"]] + } + ]); + } + + if (v["_ProductClass"] != null) { + res.push([ + ["DeviceID", "ProductClass"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [v["_ProductClass"], "xsd:string"]] + } + ]); + } + + if (v["_SerialNumber"] != null) { + res.push([ + ["DeviceID", "SerialNumber"], + timestamp, + { + object: [timestamp, 0], + writable: [timestamp, 0], + value: [timestamp, [v["_SerialNumber"], "xsd:string"]] + } + ]); + } + + delete device[k]; + } + } + + storeParams(device, [], 0, false, projectionTree); + callback(null, res, loaded); + }); +} + +function saveDevice(deviceId, deviceData, isNew, sessionTimestamp, callback) { + const update = { $set: {}, $unset: {}, $addToSet: {}, $pull: {} }; + + for (let diff of deviceData.timestamps.diff()) { + if (diff[0].wildcard !== 1 << (diff[0].length - 1)) continue; + + if ( + diff[0][0] === "Events" || + diff[0][0] === "DeviceID" || + diff[0][0] === "Tags" + ) + continue; + + // Param timestamps may be greater than session timestamp to track revisions + if (diff[2] > sessionTimestamp) diff[2] = sessionTimestamp; + + if (diff[2] == null && diff[1] != null) { + update["$unset"][ + diff[0] + .slice(0, -1) + .concat("_timestamp") + .join(".") + ] = 1; + } else if (diff[2] !== diff[1]) { + const parent = deviceData.paths.get(diff[0].slice(0, -1)); + if (parent && (!parent.length || deviceData.attributes.has(parent))) { + update["$set"][ + diff[0] + .slice(0, -1) + .concat("_timestamp") + .join(".") + ] = new Date(diff[2]); + } + } + } + + for (let diff of deviceData.attributes.diff()) { + if (diff[1] === diff[2]) continue; + + const path = diff[0]; + const value1 = (((diff[1] || {}).value || [])[1] || [])[0]; + const value2 = (((diff[2] || {}).value || [])[1] || [])[0]; + const valueType1 = (((diff[1] || {}).value || [])[1] || [])[1]; + const valueType2 = (((diff[2] || {}).value || [])[1] || [])[1]; + const valueTimestamp1 = ((diff[1] || {}).value || [])[0]; + const valueTimestamp2 = ((diff[2] || {}).value || [])[0]; + const object1 = ((diff[1] || {}).object || [])[1]; + const object2 = ((diff[2] || {}).object || [])[1]; + const writable2 = ((diff[2] || {}).writable || [])[1]; + const writable1 = ((diff[1] || {}).writable || [])[1]; + + switch (path[0]) { + case "Events": + if (diff[0].length === 2 && value2 !== value1) { + if (!diff[2]) { + switch (path[1]) { + case "Inform": + update["$unset"]["_lastInform"] = 1; + break; + case "1_BOOT": + update["$unset"]["_lastBoot"] = 1; + break; + case "0_BOOTSTRAP": + update["$unset"]["_lastBootstrap"] = 1; + break; + case "Registered": + update["$unset"]["_registered"] = 1; + } + } else { + const t = new Date(diff[2].value[1][0]); + switch (path[1]) { + case "Inform": + update["$set"]["_lastInform"] = t; + break; + case "1_BOOT": + update["$set"]["_lastBoot"] = t; + break; + case "0_BOOTSTRAP": + update["$set"]["_lastBootstrap"] = t; + break; + case "Registered": + update["$set"]["_registered"] = t; + } + } + } + + break; + case "DeviceID": + if (value2 !== value1) { + const v = diff[2].value[1][0]; + switch (path[1]) { + case "ID": + update["$set"]["_id"] = v; + break; + case "Manufacturer": + update["$set"]["_deviceId._Manufacturer"] = v; + break; + case "OUI": + update["$set"]["_deviceId._OUI"] = v; + break; + case "ProductClass": + update["$set"]["_deviceId._ProductClass"] = v; + break; + case "SerialNumber": + update["$set"]["_deviceId._SerialNumber"] = v; + } + } + break; + case "Tags": + if (value2 !== value1) { + if (value2 != null) { + if (!update["$addToSet"]["_tags"]) + update["$addToSet"]["_tags"] = { $each: [] }; + update["$addToSet"]["_tags"]["$each"].push(diff[0][1]); + } else { + if (!update["$pull"]["_tags"]) { + update["$pull"]["_tags"] = { + $in: [] + }; + } + update["$pull"]["_tags"]["$in"].push(diff[0][1]); + } + } + + break; + default: + if (!diff[2]) { + update["$unset"][diff[0].join(".")] = 1; + continue; + } + + for (let attrName of Object.keys(diff[2])) { + // Param timestamps may be greater than session timestamp to track revisions + if (diff[2][attrName][0] > sessionTimestamp) + diff[2][attrName][0] = sessionTimestamp; + + if (diff[2][attrName][1] != null) { + switch (attrName) { + case "value": + if (value2 !== value1) { + if ( + valueType2 === "xsd:dateTime" && + Number.isInteger(value2) + ) { + update["$set"][path.concat("_value").join(".")] = new Date( + value2 + ); + } else { + update["$set"][path.concat("_value").join(".")] = value2; + } + } + + if (valueType2 !== valueType1) + update["$set"][path.concat("_type").join(".")] = valueType2; + + if (valueTimestamp2 !== valueTimestamp1) { + update["$set"][ + path.concat("_timestamp").join(".") + ] = new Date(valueTimestamp2); + } + + break; + case "object": + if (!diff[1] || !diff[1].object || object2 !== object1) + update["$set"][path.concat("_object").join(".")] = !!object2; + + break; + case "writable": + if (!diff[1] || !diff[1].writable || writable2 !== writable1) { + update["$set"][ + path.concat("_writable").join(".") + ] = !!writable2; + } + } + } + } + + if (diff[1]) { + for (let attrName of Object.keys(diff[1])) { + if ( + diff[1][attrName][1] != null && + (!diff[2] || !diff[2][attrName] || !diff[2][attrName][1]) + ) { + update["$unset"][path.concat(`_${attrName}`).join(".")] = 1; + if (attrName === "value") { + update["$unset"][path.concat("_type").join(".")] = 1; + update["$unset"][path.concat("_timestamp").join(".")] = 1; + } + } + } + } + } + } + + update["$unset"] = optimizeProjection(update["$unset"]); + + // Remove overlap possibly caused by parameters changing from objects + // to regular parameters or vice versa. Reason being that _timestamp + // represents two different things depending on whether the parameter + // is an object or not. + for (let k of Object.keys(update["$unset"])) + if (update["$set"][k] != null) delete update["$unset"][k]; + + // Remove empty keys + for (let [k, v] of Object.entries(update)) { + if (k === "$addToSet") { + for (let [kk, vv] of Object.entries(v)) + if (!vv["$each"].length) delete v[kk]; + } else if (k === "$pull") { + for (let [kk, vv] of Object.entries(v)) + if (!vv["$in"].length) delete v[kk]; + } + if (!Object.keys(v).length) delete update[k]; + } + + if (!Object.keys(update).length) return callback(); + + // Mongo doesn't allow $addToSet and $pull at the same time + let update2; + if (update["$addToSet"] && update["$pull"]) { + update2 = { $pull: update["$pull"] }; + delete update["$pull"]; + } + + devicesCollection.update( + { _id: deviceId }, + update, + { upsert: isNew }, + (err, result) => { + if (!err && result.result.n !== 1) + return callback(new Error(`Device ${deviceId} not found in database`)); + + if (update2) + return devicesCollection.update({ _id: deviceId }, update2, callback); + + return callback(err); + } + ); +} + +function getFaults(deviceId, callback) { + faultsCollection + .find({ _id: { $regex: `^${common.escapeRegExp(deviceId)}\\:` } }) + .toArray((err, res) => { + if (err) return callback(err); + + const faults = {}; + for (const r of res) { + const channel = r._id.slice(deviceId.length + 1); + delete r._id; + delete r.channel; + delete r.device; + r.timestamp = +r.timestamp; + r.provisions = JSON.parse(r.provisions); + faults[channel] = r; + } + + return callback(err, faults); + }); +} + +function saveFault(deviceId, channel, fault, callback) { + fault = Object.assign({}, fault); + fault._id = `${deviceId}:${channel}`; + fault.device = deviceId; + fault.channel = channel; + fault.timestamp = new Date(fault.timestamp); + fault.provisions = JSON.stringify(fault.provisions); + faultsCollection.save(fault, callback); +} + +function deleteFault(deviceId, channel, callback) { + faultsCollection.remove({ _id: `${deviceId}:${channel}` }, callback); +} + +function getDueTasks(deviceId, timestamp, callback) { + const cur = tasksCollection.find({ device: deviceId }).sort(["timestamp"]); + const tasks = []; + + let f; + cur.nextObject( + (f = (err, task) => { + if (err) return callback(err); + + if (!task) return callback(null, tasks, null); + + if (task.timestamp) task.timestamp = +task.timestamp; + if (task.expiry) task.expiry = +task.expiry; + if (task.timestamp >= timestamp) + return callback(null, tasks, +task.timestamp); + task._id = String(task._id); + + tasks.push(task); + + // For API compatibility + if (task.name === "download" && task.file) { + let q; + if (mongodb.ObjectID.isValid(task.file)) + q = { _id: { $in: [task.file, new mongodb.ObjectID(task.file)] } }; + else q = { _id: task.file }; + + filesCollection.find(q).toArray((err, res) => { + if (err) return callback(err); + + if (res[0]) { + if (!task.fileType) task.fileType = res[0].metadata.fileType; + + if (task.fileName) + task.fileName = res[0].filename || res[0]._id.toString(); + } + + cur.nextObject(f); + }); + } else { + cur.nextObject(f); + } + }) + ); +} + +function clearTasks(deviceId, taskIds, callback) { + tasksCollection.remove( + { _id: { $in: taskIds.map(id => new mongodb.ObjectID(id)) } }, + callback + ); +} + +function getOperations(deviceId, callback) { + operationsCollection + .find({ _id: { $regex: `^${common.escapeRegExp(deviceId)}\\:` } }) + .toArray((err, res) => { + if (err) return callback(err); + + const operations = {}; + for (const r of res) { + const commandKey = r._id.slice(deviceId.length + 1); + delete r._id; + r.timestamp = +r.timestamp; + if (r.args) r.args = JSON.parse(r.args); + r.provisions = JSON.parse(r.provisions); + r.retries = JSON.parse(r.retries); + operations[commandKey] = r; + } + + callback(err, operations); + }); +} + +function saveOperation(deviceId, commandKey, operation, callback) { + operation = Object.assign({}, operation); + operation._id = `${deviceId}:${commandKey}`; + operation.timestamp = new Date(operation.timestamp); + operation.provisions = JSON.stringify(operation.provisions); + operation.retries = JSON.stringify(operation.retries); + operation.args = JSON.stringify(operation.args); + operationsCollection.save(operation, callback); +} + +function deleteOperation(deviceId, commandKey, callback) { + operationsCollection.remove({ _id: `${deviceId}:${commandKey}` }, callback); +} + +function getPresets(callback) { + presetsCollection.find().toArray(callback); +} + +function getObjects(callback) { + objectsCollection.find().toArray(callback); +} + +function getProvisions(callback) { + provisionsCollection.find().toArray(callback); +} + +function getVirtualParameters(callback) { + virtualParametersCollection.find().toArray(callback); +} + +function getFiles(callback) { + filesCollection.find().toArray(callback); +} + +exports.connect = connect; +exports.disconnect = disconnect; +exports.fetchDevice = fetchDevice; +exports.saveDevice = saveDevice; +exports.getFaults = getFaults; +exports.saveFault = saveFault; +exports.deleteFault = deleteFault; +exports.clearTasks = clearTasks; +exports.saveOperation = saveOperation; +exports.deleteOperation = deleteOperation; +exports.getPresets = getPresets; +exports.getObjects = getObjects; +exports.getProvisions = getProvisions; +exports.getVirtualParameters = getVirtualParameters; +exports.getFiles = getFiles; +exports.getDueTasks = getDueTasks; +exports.getOperations = getOperations;