diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index c0729c1..61fa687 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -1,5 +1,5 @@ ### -# Copyright 2013, 2014 Zaid Abdulla +# Copyright 2013-2016 Zaid Abdulla # # GenieACS is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as @@ -38,29 +38,22 @@ # THE SOFTWARE. ### +util = require 'util' +zlib = require 'zlib' + config = require './config' common = require './common' -util = require 'util' soap = require './soap' -tasks = require './tasks' db = require './db' -presets = require './presets' -mongodb = require 'mongodb' -apiFunctions = require './api-functions' -customCommands = require './custom-commands' -crypto = require 'crypto' -zlib = require 'zlib' -updateDevice = require './update-device' - - -# Used to reject new TR-069 sessions when under load -holdUntil = Date.now() +session = require './session' +query = require './query' +device = require './device' writeResponse = (currentRequest, res) -> - if config.get('DEBUG', currentRequest.session.deviceId) + if config.get('DEBUG', currentRequest.sessionData.deviceId) dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" - require('fs').appendFile("./debug/#{currentRequest.session.deviceId}.dump", dump, (err) -> + require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> throw err if err ) @@ -86,290 +79,95 @@ writeResponse = (currentRequest, res) -> currentRequest.httpResponse.end(res.data) -endSession = (currentRequest) -> - db.redisClient.del("session_#{currentRequest.sessionId}", (err, res) -> - throw err if err - res = soap.response(null) - writeResponse(currentRequest, res) - ) - - inform = (currentRequest, cwmpRequest) -> - # If overloaded, ask CPE to retry in 5 mins - if Date.now() < holdUntil - res = {code : 503, headers : {'Retry-After' : 300}, data : ''} - return writeResponse(currentRequest, res) + if config.get('LOG_INFORMS', currentRequest.sessionData.deviceId) + util.log("#{currentRequest.sessionData.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") - now = new Date(Date.now()) - - if config.get('LOG_INFORMS', currentRequest.session.deviceId) - util.log("#{currentRequest.session.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") - - parameterNames = (p[0] for p in cwmpRequest.methodRequest.parameterList) - - informHash = crypto.createHash('md5').update(JSON.stringify(parameterNames)).update(JSON.stringify(config.CUSTOM_COMMANDS)).digest('hex') - - deviceUpdates = {parameterValues : cwmpRequest.methodRequest.parameterList, informEvents : cwmpRequest.methodRequest.event} - - # Delete all pending tasks on bootstrap (e.g. factory reset) - if '0 BOOTSTRAP' in cwmpRequest.methodRequest.event - lastBootstrap = now - db.tasksCollection.remove({device : currentRequest.session.deviceId}, (err, removed) -> - throw err if err - ) - - db.redisClient.get("#{currentRequest.session.deviceId}_inform_hash", (err, oldInformHash) -> + session.inform(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse) -> throw err if err + res = soap.response({ + id : cwmpRequest.id, + methodResponse : rpcResponse, + cwmpVersion : currentRequest.sessionData.cwmpVersion + }) - updateAndRespond = () -> - updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, false, (err) -> - throw err if err - db.redisClient.setex("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, JSON.stringify(currentRequest.session), (err) -> - throw err if err - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'InformResponse'}, - cwmpVersion : cwmpRequest.cwmpVersion - }) - - if !!cookiesPath = config.get('COOKIES_PATH', currentRequest.session.deviceId) - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}; Path=#{cookiesPath}" - else - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionId}" - - writeResponse(currentRequest, res) - ) - ) - - if oldInformHash == informHash and not lastBootstrap? - return updateAndRespond() - - db.redisClient.setex("#{currentRequest.session.deviceId}_inform_hash", config.get('PRESETS_CACHE_DURATION', currentRequest.session.deviceId), informHash, (err, res) -> + session.save(currentRequest.sessionData, (err, sessionId) -> throw err if err - ) - - # populate projection and parameterStructure - parameterStructure = {} - projection = {} - projection._customCommands = 1 - projection._timestamp = 1 - projection._lastBootstrap = 1 - for param in parameterNames - ref = parameterStructure - path = '' - for p in param.split('.') - ref[p] ?= {_path : path} - ref = ref[p] - path += p + '.' - projection[path + '_timestamp'] = 1 - - db.devicesCollection.findOne({'_id' : currentRequest.session.deviceId}, projection, (err, device) -> - throw err if err - lastBootstrap ?= device?._lastBootstrap - _tasks = [] - - # For any parameters that aren't in the DB model, issue a refresh task - traverse = (reference, actual) -> - for k of reference - continue if k[0] == '_' - if not actual[k]?._timestamp? or actual[k]._timestamp < lastBootstrap - _tasks.push({device: currentRequest.session.deviceId, name : 'refreshObject', objectName : reference[k]._path, timestamp : now}) - else - traverse(reference[k], actual[k]) - - if device? - traverse(parameterStructure, device) - - # Update custom commands if needed - deviceCustomCommands = {} - deviceCustomCommands[k] = v._timestamp for k,v of device._customCommands - - for cmd in customCommands.getDeviceCustomCommandNames(currentRequest.session.deviceId) - if not deviceCustomCommands[cmd]? or deviceCustomCommands[cmd] < lastBootstrap - # Increment timestamp by one millisecond to ensure it runs after any refresh task - _tasks.push({device: currentRequest.session.deviceId, name: 'customCommand', command: "#{cmd} init", timestamp: new Date(now.getTime() + 1)}) - delete deviceCustomCommands[cmd] - - for cmd of deviceCustomCommands - deviceUpdates.deletedObjects ?= [] - deviceUpdates.deletedObjects.push("_customCommands.#{cmd}") - - # clear presets hash if parameters are potentially modified - if _tasks.length > 0 - db.redisClient.del("#{currentRequest.session.deviceId}_presets_hash", (err) -> - throw err if err - ) - - apiFunctions.insertTasks(_tasks, {}, () -> - updateAndRespond() - ) + if !!cookiesPath = config.get('COOKIES_PATH', currentRequest.sessionData.deviceId) + res.headers['Set-Cookie'] = "session=#{sessionId}; Path=#{cookiesPath}" else - deviceIdDetails = {} - for k, v of cwmpRequest.methodRequest.deviceId - deviceIdDetails["_#{k}"] = v + res.headers['Set-Cookie'] = "session=#{sessionId}" - db.devicesCollection.insert({_id : currentRequest.session.deviceId, _registered : now, _deviceId : deviceIdDetails}, (err) -> - throw err if err - util.log("#{currentRequest.session.deviceId}: New device registered") - _tasks.push({device: currentRequest.session.deviceId, name : 'refreshObject', objectName : '', timestamp : now}) - - for cmd in customCommands.getDeviceCustomCommandNames(currentRequest.session.deviceId) - # Increment timestamp by one millisecond to ensure it runs after any refresh task - _tasks.push({device: currentRequest.session.deviceId, name: 'customCommand', command: "#{cmd} init", timestamp: new Date(now.getTime() + 1)}) - - apiFunctions.insertTasks(_tasks, {}, () -> - updateAndRespond() - ) - ) + writeResponse(currentRequest, res) ) ) -runTask = (currentRequest, task, methodResponse) -> - timeDiff = process.hrtime() - tasks.task(task, methodResponse, (err, status, methodRequest, deviceUpdates) -> - throw err if err +applyPresets = (currentRequest) -> + db.getPresetsObjectsAliases((presets, objects, aliases) -> + parameters = {} + for p in presets + for a in Object.keys(query.queryProjection(p.precondition)) + parameters[a] = common.parsePath(a) - timeDiff = process.hrtime(timeDiff)[0] + 1 - if timeDiff > 3 # in seconds - # Server under load. Hold new sessions temporarily. - holdUntil = Math.max(Date.now() + timeDiff * 2000, holdUntil) + declarations = [] + for k, v of parameters + declarations.push([v, 1, null, null, null, null, null, 1]) - save = status & tasks.STATUS_SAVE - switch status & ~tasks.STATUS_SAVE - when tasks.STATUS_OK - f = () -> - db.redisClient.setex(String(task._id), currentRequest.session.sessionTimeout, JSON.stringify(task), (err) -> - throw err if err - res = soap.response({ - id : task._id, - methodRequest : methodRequest, - cwmpVersion : currentRequest.session.cwmpVersion - }) - writeResponse(currentRequest, res) - updateSessionExpiry(currentRequest) - ) + session.rpcRequest(currentRequest.sessionData, declarations, (err, id, rpcRequest) -> + throw err if err - if save - updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> - throw err if err - db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) -> - throw err if err - f() - ) - ) - else - updateDevice.queueUpdates(currentRequest.session.deviceId, deviceUpdates, (err) -> - f() - ) - when tasks.STATUS_COMPLETED - util.log("#{currentRequest.session.deviceId}: Completed task #{task.name}(#{task._id})") - updateDevice.commitUpdates(currentRequest.session.deviceId, deviceUpdates, true, (err) -> - throw err if err - db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, (err, removed) -> - throw err if err - db.redisClient.del(String(task._id), (err, res) -> - throw err if err - nextTask(currentRequest) - ) - ) - ) - when tasks.STATUS_FAULT - retryAfter = config.get('RETRY_DELAY', currentRequest.session.deviceId) * Math.pow(2, task.retries ? 0) - util.log("#{currentRequest.session.deviceId}: Fault response for task #{task._id}. Retrying after #{retryAfter} seconds.") - taskUpdate = {fault : task.fault, timestamp : new Date(Date.now() + retryAfter * 1000)} - if save - taskUpdate.session = task.session + if rpcRequest? + return sendRpcRequest(currentRequest, id, rpcRequest) - db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : taskUpdate, $inc : {retries : 1}}, (err) -> - throw err if err - nextTask(currentRequest) - ) - else - throw new Error('Unknown task status') + for k, v of parameters + if (vv = device.getAll(currentRequest.sessionData.deviceData, v)[0]?[8]?[0]) + parameters[k] = vv + + for p in presets + if query.test(parameters, p.precondition) + session.addPreset(currentRequest.sessionData, p._id, p) + + session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) -> + throw err if err + sendRpcRequest(currentRequest, id, rpcRequest) + ) + ) ) -isTaskExpired = (task) -> - task.expiry <= new Date() - - -assertPresets = (currentRequest) -> - db.redisClient.mget("#{currentRequest.session.deviceId}_presets_hash", 'presets_hash', (err, res) -> +nextRpc = (currentRequest) -> + session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) -> throw err if err - devicePresetsHash = res[0] - presetsHash = res[1] - if devicePresetsHash? and devicePresetsHash == presetsHash - # no discrepancy, end the session - endSession(currentRequest) - else - PRESETS_CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', currentRequest.session.deviceId) - db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) -> - if not presetsHash? - presetsHash = presets.calculatePresetsHash(allPresets, allObjects) - db.redisClient.setex('presets_hash', PRESETS_CACHE_DURATION, presetsHash, (err, res) -> - throw err if err - ) + if not rpcRequest? + return applyPresets(currentRequest) - presets.getDevicePreset(currentRequest.session.deviceId, allPresets, allObjects, allAliases, (devicePreset) -> - presets.processDevicePreset(currentRequest.session.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) -> - db.redisClient.setex("#{currentRequest.session.deviceId}_presets_hash", Math.floor(Math.max(1, expiry - config.get('PRESETS_TIME_PADDING'))), presetsHash, (err, res) -> - throw err if err - ) - - if addTags.length + deleteTags.length + taskList.length - util.log("#{currentRequest.session.deviceId}: Presets discrepancy found") - - if addTags.length + deleteTags.length - util.log("#{currentRequest.session.deviceId}: Updating tags +(#{addTags}) -(#{deleteTags})") - - if deleteTags.length - db.devicesCollection.update({'_id' : currentRequest.session.deviceId}, {'$pull' : {'_tags' : {'$in' : deleteTags}}}, {}, (err, count) -> - throw err if err - ) - - if addTags.length - db.devicesCollection.update({'_id' : currentRequest.session.deviceId}, {'$addToSet' : {'_tags' : {'$each' : addTags}}}, {}, (err, count) -> - throw err if err - ) - - if taskList.length - t.expiry = expiry for t in taskList - apiFunctions.insertTasks(taskList, allAliases, (err, taskList) -> - throw err if err - task = taskList[0] - util.log("#{currentRequest.session.deviceId}: Started task #{task.name}(#{task._id})") - runTask(currentRequest, task, {}) - ) - else - endSession(currentRequest) - ) - ) - ) + sendRpcRequest(currentRequest, id, rpcRequest) ) -nextTask = (currentRequest) -> - now = new Date() - cur = db.tasksCollection.find({'device' : currentRequest.session.deviceId, timestamp : {$lte : now}}).sort(['timestamp']).limit(1) - cur.nextObject((err, task) -> - throw err if err +sendRpcRequest = (currentRequest, id, rpcRequest) -> + if not rpcRequest? + session.end(currentRequest.sessionData, (err, isNew) -> + throw err if err + if isNew + util.log("#{currentRequest.sessionData.deviceId}: New device registered") - if not task - # no more tasks, check presets discrepancy - assertPresets(currentRequest) - else if isTaskExpired(task) - util.log("#{currentRequest.session.deviceId}: Task is expired #{task.name}(#{task._id})") - db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) -> - throw err if err - nextTask(currentRequest) - ) - else - util.log("#{currentRequest.session.deviceId}: Started task #{task.name}(#{task._id})") - updateDevice.clearUpdatesQueue(currentRequest.session.deviceId, (err) -> - throw err if err - runTask(currentRequest, task, {}) - ) + writeResponse(currentRequest, soap.response(null)) + ) + return + + util.log("#{currentRequest.sessionData.deviceId}: #{rpcRequest.type} (#{id})") + + res = soap.response({ + id : id, + methodRequest : rpcRequest, + cwmpVersion : currentRequest.sessionData.cwmpVersion + }) + + session.save(currentRequest.sessionData, (err) -> + throw err if err + writeResponse(currentRequest, res) ) @@ -381,20 +179,9 @@ getSession = (httpRequest, callback) -> return callback() if not sessionId? - db.redisClient.get("session_#{sessionId}", (err, res) -> + session.load(sessionId, (err, sessionData) -> throw err if err - return callback(sessionId, JSON.parse(res)) - ) - - -updateSessionExpiry = (currentRequest) -> - db.redisClient.expire("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, (err, res) -> - throw err if err - if not res - # Resave session in the very rare case when session cache expires while processing request - db.redisClient.setex("session_#{currentRequest.sessionId}", currentRequest.session.sessionTimeout, JSON.stringify(currentRequest.session), (err) -> - throw err if err - ) + return callback(sessionId, sessionData) ) @@ -436,35 +223,27 @@ listener = (httpRequest, httpResponse) -> return body stream.on('end', () -> - getSession(httpRequest, (sessionId, session) -> + getSession(httpRequest, (sessionId, sessionData) -> cwmpRequest = soap.request(httpRequest, session?.cwmpVersion) - if not session? + if not sessionData? if cwmpRequest.methodRequest?.type isnt 'Inform' httpResponse.writeHead(400) httpResponse.end('Session is expired') return deviceId = common.generateDeviceId(cwmpRequest.methodRequest.deviceId) - sessionId = crypto.randomBytes(8).toString('hex') - - session = { - deviceId : deviceId, - cwmpVersion : cwmpRequest.cwmpVersion, - sessionTimeout : cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId) - } - - httpRequest.connection.setTimeout(session.sessionTimeout * 1000) + sessionData = session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId)) + httpRequest.connection.setTimeout(session.timeout * 1000) currentRequest = { httpRequest : httpRequest, httpResponse : httpResponse, - sessionId : sessionId, - session : session + sessionData : sessionData } - if config.get('DEBUG', currentRequest.session.deviceId) + if config.get('DEBUG', currentRequest.sessionData.deviceId) dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n" - require('fs').appendFile("./debug/#{currentRequest.session.deviceId}.dump", dump, (err) -> + require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> throw err if err ) @@ -472,93 +251,30 @@ listener = (httpRequest, httpResponse) -> if cwmpRequest.methodRequest.type is 'Inform' inform(currentRequest, cwmpRequest) else if cwmpRequest.methodRequest.type is 'GetRPCMethods' - util.log("#{currentRequest.session.deviceId}: GetRPCMethods") + util.log("#{currentRequest.sessionData.deviceId}: GetRPCMethods") res = soap.response({ id : cwmpRequest.id, methodResponse : {type : 'GetRPCMethodsResponse', methodList : ['Inform', 'GetRPCMethods', 'TransferComplete', 'RequestDownload']}, - cwmpVersion : currentRequest.session.cwmpVersion + cwmpVersion : currentRequest.sessionData.cwmpVersion }) - writeResponse(currentRequest, res) - updateSessionExpiry(currentRequest) - else if cwmpRequest.methodRequest.type is 'TransferComplete' - # do nothing - util.log("#{currentRequest.session.deviceId}: Transfer complete") - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'TransferCompleteResponse'}, - cwmpVersion : currentRequest.session.cwmpVersion - }) - writeResponse(currentRequest, res) - updateSessionExpiry(currentRequest) - else if cwmpRequest.methodRequest.type is 'RequestDownload' - requestDownloadResponse = () -> - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'RequestDownloadResponse'}, - cwmpVersion : currentRequest.session.cwmpVersion - }) + session.save(currentRequest.sessionData, (err) -> + throw err if err writeResponse(currentRequest, res) - updateSessionExpiry(currentRequest) - fileType = cwmpRequest.methodRequest.fileType - util.log("#{currentRequest.session.deviceId}: RequestDownload (#{fileType})") - if fileType isnt '1 Firmware Upgrade Image' - # Only supporting firmware upgrade for now - return requestDownloadResponse() - - db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) -> - presets.getDevicePreset(currentRequest.session.deviceId, allPresets, allObjects, allAliases, (devicePreset) -> - presetSoftwareVersion = devicePreset.softwareVersion?.preset - currentSoftwareVersion = devicePreset.softwareVersion?.current._value - if presetSoftwareVersion? and presetSoftwareVersion != currentSoftwareVersion - deviceIdStruct = common.parseDeviceId(currentRequest.session.deviceId) - db.filesCollection.findOne({'metadata.fileType' : '1 Firmware Upgrade Image', 'metadata.oui' : deviceIdStruct.oui, 'metadata.productClass' : deviceIdStruct.productClass, 'metadata.version' : presetSoftwareVersion}, {_id : 1}, (err, file) -> - throw err if err - if not file? - util.error("#{currentRequest.session.deviceId}: Firmware image not found (#{presetSoftwareVersion})") - return requestDownloadResponse() - - task = { - device : currentRequest.session.deviceId, - name : 'download', - file : file['_id'] - } - apiFunctions.insertTasks(task, allAliases, (err, tasks) -> - throw err if err - return requestDownloadResponse() - ) - ) - else - return requestDownloadResponse() - ) ) - else + else if cwmpRequest.methodRequest.type is 'TransferComplete' throw new Error('ACS method not supported') else if cwmpRequest.methodResponse? - taskId = cwmpRequest.id - - db.getTask(taskId, (err, task) -> + session.rpcResponse(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.methodResponse, (err) -> throw err if err - if not task - nextTask(currentRequest) - else - runTask(currentRequest, task, cwmpRequest.methodResponse) + nextRpc(currentRequest) ) else if cwmpRequest.fault? - taskId = cwmpRequest.id - if not taskId - # Fault not related to a task. End the session. - return endSession(currentRequest) - - db.getTask(taskId, (err, task) -> - throw err if err - if not task - nextTask(currentRequest) - else - runTask(currentRequest, task, cwmpRequest.fault) + session.rpcFault(sessionData, cwmpRequest.id, cwmpRequest.fault, (err) -> + nextRpc(currentRequest) ) else - # cpe sent empty response. start sending acs requests - nextTask(currentRequest) + # cpe sent empty response. add presets + nextRpc(currentRequest) ) ) diff --git a/lib/db.coffee b/lib/db.coffee index ab2ecc4..1b4b239 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -193,6 +193,24 @@ getPresetsObjectsAliases = (callback) -> if not presets getCached('presets', (callback) -> presetsCollection.find().toArray((err, res) -> + for r in res + r.precondition = JSON.parse(r.precondition) + + # Generate provisions from the old configuration format + r.provisions = [] + for c in r.configurations + switch c.type + when 'age' + r.provisions.push(['refresh', c.name, c.age * -1000]) + when 'value' + r.provisions.push(['value', c.name, c.value]) + when 'add_tag' + r.provisions.push(['tag', c.tag, true]) + when 'delete_tag' + r.provisions.push(['tag', c.tag, false]) + else + throw new Error("Unknown configuration type #{c.type}") + callback(err, res) ) , config.get('PRESETS_CACHE_DURATION'), (err, res) -> diff --git a/lib/query.coffee b/lib/query.coffee index 5db57ac..5e91654 100644 --- a/lib/query.coffee +++ b/lib/query.coffee @@ -1,4 +1,22 @@ ### +# Copyright 2013-2016 Zaid Abdulla +# +# 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 . +# +# This file incorporates work covered by the following copyright and +# permission notice: +# # Copyright 2013 Fanoos Telecom # # Permission is hereby granted, free of charge, to any person obtaining a copy @@ -75,7 +93,7 @@ permute = (param, val, aliases) -> values = expandValue(k, val) if k[k.lastIndexOf('.') + 1] != '_' k += '._value' - + for v in values obj = {} obj[k] = v @@ -106,6 +124,7 @@ expand = (query, aliases) -> return new_query + # Replace _id string values with ObjectID type substituteObjectId = (query) -> for k, v of query @@ -131,5 +150,90 @@ substituteObjectId = (query) -> return query +# Generate parameter projection from a query +# If second arg is given, it's edited and returned +queryProjection = (query, proj) -> + proj ?= {} + for k,v of query + if k.charAt(0) == '$' # this is a logical operator + for q in v + projection(q, proj) + else + proj[k] = 1 + return proj + + +testExpressions = (params, expressions, lop) -> + for f in expressions + res = test(params, f) + + switch lop + when '$and' + return false if not res + when '$or' + return true if res + when '$nor' + return false if res + + return switch lop + when '$and' then true + when '$or' then false + when '$nor' then true + else throw new Error('Unknown logical operator') + + +test = (params, query) -> + for k,v of query + if k.charAt(0) == '$' # this is a logical operator + res = testExpressions(params, v, k) + else + value = params[k] + + if common.typeOf(v) isnt common.OBJECT_TYPE + # TODO comparing array to regex, array to array, and object to object + if common.typeOf(value) is common.ARRAY_TYPE + res = value.indexOf(v) != -1 + else + if common.typeOf(v) is common.REGEXP_TYPE + res = v.test(value) + else + res = v == value + else + for k2,v2 of v + switch k2 + when '$ne' + if common.typeOf(value) is common.ARRAY_TYPE + res = value.indexOf(v2) == -1 + else + res = value != v2 + when '$lt' + res = value < v2 + when '$lte' + res = value <= v2 + when '$gt' + res = value > v2 + when '$gte' + res = value >= v2 + when '$regex' + res = v2.test(value) + when '$in' + throw new Error('Operator not supported') + when '$nin' + throw new Error('Operator not supported') + when '$all' + throw new Error('Operator not supported') + when '$exists' + throw new Error('Operator not supported') + else + throw new Error('Operator not supported') + + if not res + return false + + return true + + exports.expand = expand exports.substituteObjectId = substituteObjectId +exports.queryProjection = queryProjection +exports.test = test