Better file structure and naming

lib/api-functions.coffee

@@ -0,0 +1,170 @@
+config = require './config'
+db = require './db'
+http = require 'http'
+common = require './common'
+util = require 'util'
+URL = require 'url'
+auth = require './auth'
+
+
+connectionRequest = (deviceId, callback) ->
+  conReq = (url, authString, callback) ->
+    options = URL.parse(url)
+
+    if authString
+      options.headers = {'Authorization' : authString}
+
+    request = http.get(options, (res) ->
+      if res.statusCode == 401
+        authHeader = auth.parseAuthHeader(res.headers['www-authenticate'])
+      callback(res.statusCode, authHeader)
+    )
+
+    request.on('error', (err) ->
+      # error event when request is aborted
+      request.abort()
+      callback(0)
+    )
+
+    request.on('socket', (socket) ->
+      socket.setTimeout(2000)
+      socket.on('timeout', () ->
+        request.abort()
+      )
+    )
+
+  db.devicesCollection.findOne({_id : deviceId}, {'InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value' : 1}, (err, device)->
+    if err
+      callback(err)
+      return
+    connectionRequestUrl = device.InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value
+    # for testing
+    #connectionRequestUrl = connectionRequestUrl.replace(/^(http:\/\/)([0-9\.]+)(\:[0-9]+\/[a-zA-Z0-9]+\/?$)/, '$110.1.1.254$3')
+    conReq(connectionRequestUrl, null, (statusCode, authHeader) ->
+      if statusCode == 401
+        [username, password] = config.auth.connectionRequest(deviceId)
+        if authHeader.method is 'Basic'
+          authString = auth.basic(username, password)
+        else if authHeader.method is 'Digest'
+          uri = URL.parse(connectionRequestUrl)
+          authString = auth.digest(username, password, uri.path, 'GET', null, authHeader)
+
+        conReq(connectionRequestUrl, authString, (statusCode, authHeader) ->
+          callback()
+        )
+      else
+        callback()
+    )
+  )
+
+
+watchTask = (taskId, timeout, callback) ->
+  setTimeout( () ->
+    db.tasksCollection.findOne({_id : taskId}, {'_id' : 1}, (err, task) ->
+      if task
+        timeout -= 500
+        if timeout < 0
+          callback('timeout')
+        else
+          watchTask(taskId, timeout, callback)
+      else
+        callback(err)
+    )
+  , 500)
+
+
+expandParam = (param, aliases) ->
+  params = [param]
+  for a,aa of aliases
+    if a == param or common.startsWith(a, "#{param}.")
+      for p in aa
+        params.push(p) if p[p.lastIndexOf('.') + 1] != '_'
+
+  return params
+
+
+sanitizeTask = (task, aliases, callback) ->
+  task.timestamp = new Date(task.timestamp ? Date.now())
+  if task.expiry?
+    if common.typeOf(task.expiry) is common.DATE_TYPE or isNaN(task.expiry)
+      task.expiry = new Date(task.expiry)
+    else
+      task.expiry = new Date(task.timestamp.getTime() + +task.expiry * 1000)
+
+  switch task.name
+    when 'getParameterValues'
+      projection = {}
+      for p in task.parameterNames
+        for pp in expandParam(p, aliases)
+          projection[pp] = 1
+      db.devicesCollection.findOne({_id : task.device}, projection, (err, device) ->
+        parameterNames = []
+        for k of projection
+          if common.getParamValueFromPath(device, k)?
+            parameterNames.push(k)
+        task.parameterNames = parameterNames
+        callback(task)
+      )
+    when 'setParameterValues'
+      projection = {}
+      values = {}
+      for p in task.parameterValues
+        for pp in expandParam(p[0], aliases)
+          projection[pp] = 1
+          values[pp] = p[1]
+      db.devicesCollection.findOne({_id : task.device}, projection, (err, device) ->
+        parameterValues = []
+        for k of projection
+          param = common.getParamValueFromPath(device, k)
+          if param?
+            parameterValues.push([k, values[k], param._type])
+        task.parameterValues = parameterValues
+        callback(task)
+      )
+    else
+      # TODO implement setParameterValues
+      callback(task)
+
+
+addAliases = (device, aliases) ->
+  for k,v of aliases
+    for p in v
+      pp = p.split('.')
+      obj = device
+      for i in pp
+        if not obj[i]?
+          obj = null
+          break
+        obj = obj[i]
+
+      device[k] = obj if obj?
+
+
+insertTasks = (tasks, aliases, callback) ->
+  if tasks? and common.typeOf(tasks) isnt common.ARRAY_TYPE
+    tasks = [tasks]
+  else if not tasks? or tasks.length == 0
+    return callback(tasks)
+
+  counter = tasks.length
+
+  for task in tasks
+    sanitizeTask(task, aliases, (t) ->
+      if t.uniqueKey?
+        db.tasksCollection.remove({device : t.device, uniqueKey : t.uniqueKey}, (err, removed) ->
+        )
+
+      --counter
+      if counter == 0
+        db.tasksCollection.insert(tasks, (err, _tasks) ->
+          #util.log("#{_task.device}: Added task #{_task.name}(#{_task._id})") for _task in _tasks
+          callback(err, _tasks)
+        )
+    )
+
+
+exports.addAliases = addAliases
+exports.sanitizeTask = sanitizeTask
+exports.connectionRequest = connectionRequest
+exports.watchTask = watchTask
+exports.insertTasks = insertTasks

lib/auth.coffee

@@ -0,0 +1,63 @@
+crypto = require 'crypto'
+
+
+exports.parseAuthHeader = (authHeader) ->
+  res = {}
+  i = authHeader.indexOf(' ')
+  res['method'] = authHeader.slice(0, i)
+
+  options = authHeader.slice(i + 1)
+  regex = /([a-z]+)="([a-zA-Z0-9\/\.@\-,=]+)"/g
+  while (r = regex.exec(options)) != null
+    res[r[1]] = r[2]
+
+  regex = /([a-z]+)=([a-zA-Z0-9\/\.@\-]+)/g
+  while (r = regex.exec(options)) != null
+    res[r[1]] = r[2]
+  return res
+
+
+exports.basic = (username, password) ->
+  "Basic #{new Buffer("#{username}:#{password}").toString('base64')}"
+
+
+exports.digest = (username, password, uri, httpMethod, body, authHeader) ->
+  cnonce = '0a4f113b'
+  nc = '00000001'
+
+  if authHeader.qop?
+    if authHeader.qop.indexOf(',') != -1
+      qop = 'auth' # either auth or auth-int. prefer auth
+    else
+      qop = authHeader.qop
+
+  ha1 = crypto.createHash('md5')
+  ha1.update(username).update(':').update(authHeader.realm).update(':').update(password)
+  # TODO support "MD5-sess" algorithm directive
+  ha1 = ha1.digest('hex')
+
+  ha2 = crypto.createHash('md5')
+  ha2.update(httpMethod).update(':').update(uri)
+  if qop == 'auth-int'
+    ha2.update(':').update(body)
+  ha2 = ha2.digest('hex')
+
+  response = crypto.createHash('md5')
+  response.update(ha1).update(':').update(authHeader.nonce)
+
+  if qop?
+    response.update(':').update(nc).update(':').update(cnonce).update(':').update(qop)
+  response.update(':').update(ha2)
+  response = response.digest('hex')
+
+  authString = "Digest username=\"#{username}\""
+  authString += ",realm=\"#{authHeader.realm}\""
+  authString += ",nonce=\"#{authHeader.nonce}\""
+  authString += ",uri=\"#{uri}\""
+  authString += ",algorithm=#{authHeader.algorithm}" if authHeader.algorithm?
+  authString += ",qop=#{qop}"
+  authString += ",nc=#{nc}"
+  authString += ",cnonce=\"#{cnonce}\""
+  authString += ",response=\"#{response}\""
+  authString += ",opaque=\"#{authHeader.opaque}\""
+  return authString

lib/common.coffee

@@ -0,0 +1,85 @@
+UNDEFINED_TYPE = '[object Undefined]'
+NULL_TYPE = '[object Null]'
+BOOLEAN_TYPE = '[object Boolean]'
+NUMBER_TYPE = '[object Number]'
+STRING_TYPE = '[object String]'
+ARRAY_TYPE = '[object Array]'
+OBJECT_TYPE = '[object Object]'
+REGEXP_TYPE = '[object RegExp]'
+DATE_TYPE = '[object Date]'
+
+
+typeOf = (obj) ->
+  Object.prototype.toString.call(obj)
+
+
+exports.endsWith = (str, suffix) ->
+  str.indexOf(suffix, str.length - suffix.length) isnt -1
+
+
+exports.startsWith = (str, prefix) ->
+  str.substring(0, prefix.length) == prefix
+
+
+exports.arrayToHash = (arr) ->
+  hash = {}
+  for i in arr
+    hash[i[0]] = i[1]
+  return hash
+
+
+exports.getDeviceId = (deviceIdStruct) ->
+  # Percent escaping function. Escapes everything except alphanumerics and underscore
+  esc = (str) ->
+    str.replace(/[^A-Za-z0-9_]/g, (chr) ->
+      buf = new require('buffer').Buffer(chr)
+      rep = ''
+      rep += "%#{b.toString(16).toUpperCase()}" for b in buf
+      return rep
+    )
+
+  # Guaranteeing globally unique id as defined in TR-069
+  if deviceIdStruct['ProductClass']
+    return "#{esc(deviceIdStruct['OUI'])}-#{esc(deviceIdStruct['ProductClass'])}-#{esc(deviceIdStruct['SerialNumber'])}"
+
+  return "#{esc(deviceIdStruct['OUI'])}-#{esc(deviceIdStruct['SerialNumber'])}"
+
+
+exports.extend = (obj, mixin) ->
+  obj[name] = method for name, method of mixin        
+  obj
+
+
+exports.getParamValueFromPath = (obj, path) ->
+  pp = path.split('.')
+  ref = obj
+  try
+    for p in pp
+      ref = ref[p]
+    return ref
+  catch err
+    return undefined
+
+
+exports.matchType = (src, dst) ->
+  switch typeOf(src)
+    when STRING_TYPE
+      String(dst)
+    when NUMBER_TYPE
+      Number(dst)
+    when BOOLEAN_TYPE
+      v = String(dst).trim().toLowerCase()
+      v == 'true' or v == 'on' or v == 'yes' or v == '1'
+    else
+      dst
+
+
+exports.UNDEFINED_TYPE = UNDEFINED_TYPE
+exports.NULL_TYPE = NULL_TYPE
+exports.NUMBER_TYPE = NUMBER_TYPE
+exports.STRING_TYPE = STRING_TYPE
+exports.ARRAY_TYPE = ARRAY_TYPE
+exports.OBJECT_TYPE = OBJECT_TYPE
+exports.REGEXP_TYPE = REGEXP_TYPE
+
+exports.typeOf = typeOf

lib/config.coffee

@@ -0,0 +1,44 @@
+exports.DATABASE_NAME = 'genieacs'
+exports.ACS_PORT = 7547 # CWMP port as assigned by IANA
+exports.ACS_HTTPS_PORT = 7548
+exports.ACS_INTERFACE = '0.0.0.0'
+exports.ACS_HTTPS_INTERFACE = '0.0.0.0'
+exports.API_PORT = 7557
+exports.API_INTERFACE = '0.0.0.0'
+exports.FILES_PORT = 7567
+exports.FILES_INTERFACE = '0.0.0.0'
+exports.FILES_IP = '172.240.21.2' # Used when sending download requests to devices
+exports.CACHE_DURATION = 60 # in seconds
+exports.PRESETS_CACHE_DURATION = 86400
+exports.PRESETS_TIME_PADDING = 1
+exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock'
+exports.REDIS_SOCKET = '/tmp/redis.sock'
+exports.WORKER_RESPAWN_TIME = 60000
+exports.LOG_INFORMS = false
+exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true}
+exports.DEVICE_ONLINE_THRESHOLD = 4000
+exports.RETRY_DELAY = 300
+
+exports.MONGODB_OPTIONS = {
+  db : {
+    w : 1
+    wtimeout : 60000
+  }
+  server : {
+    auto_reconnect : true
+  }
+}
+
+# load configuration
+c = require('../config/config')
+for k, v of c
+  exports[k] = v
+
+# load parameter configurations
+exports.PARAMETERS = require('../config/parameters')
+
+# load authentication scripts
+exports.auth = require('../config/auth')
+
+exports.CUSTOM_COMMANDS = require('../config/custom_commands')
+

lib/custom-commands.coffee

@@ -0,0 +1,35 @@
+config = require './config'
+common = require './common'
+
+
+exports.execute = (deviceId, command, callback) ->
+  args = command.split(/\s+/)
+  f = args.shift()
+  c = args.shift()
+
+  file = require "./config/custom_commands/#{f}"
+  file[c](deviceId, args.join(' '), callback)
+
+
+exports.getFileCommands = (filename) ->
+  commands = []
+  f = require("./config/custom_commands/#{filename}")
+  for k,v of f
+    commands.push(k)
+  return commands
+
+
+exports.getDeviceCustomCommands = (deviceId) ->
+  commands = {}
+  for k,v of config.CUSTOM_COMMANDS
+    if eval(v).test(deviceId)
+      commands[k] = exports.getFileCommands(k)
+  return commands
+
+
+exports.getDeviceCustomCommandNames = (deviceId) ->
+  commands = []
+  for k,v of config.CUSTOM_COMMANDS
+    if eval(v).test(deviceId)
+      commands.push(k)
+  return commands

lib/cwmp.coffee

@@ -0,0 +1,530 @@
+config = require './config'
+common = require './common'
+util = require 'util'
+http = require 'http'
+https = require 'https'
+soap = require './soap'
+tasks = require './tasks'
+normalize = require('./normalize').normalize
+db = require './db'
+presets = require './presets'
+mongodb = require 'mongodb'
+fs = require 'fs'
+apiFunctions = require './api-functions'
+customCommands = require './custom-commands'
+crypto = require 'crypto'
+parameters = require './parameters'
+
+OBJECT_REGEX = /\.$/
+INSTANCE_REGEX = /\.[\d]+\.$/
+
+
+# Used to reject new TR-069 sessions when under load
+holdUntil = Date.now()
+
+
+writeResponse = (currentRequest, res) ->
+  if res.headers['Content-Length'] == 0
+    # no more requests. terminate TCP connection
+    res.headers['Connection'] = 'close' if currentRequest.httpRequest.httpVersion == '1.1'
+  else
+    res.headers['Connection'] = 'Keep-Alive' if currentRequest.httpRequest.httpVersion == '1.0'
+
+  if config.DEBUG_DEVICES[currentRequest.deviceId]
+    dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n"
+    fs = require('fs').appendFile("debug/#{currentRequest.deviceId}.dump", dump, (err) ->
+      throw err if err
+    )
+
+  currentRequest.httpResponse.writeHead(res.code, res.headers)
+  currentRequest.httpResponse.end(res.data)
+
+
+updateDevice = (currentRequest, actions, callback) ->
+  if not actions?
+    callback() if callback?
+    return
+
+  now = new Date(Date.now())
+  updates = {}
+  deletes = {}
+
+  if actions.parameterValues?
+    for p in actions.parameterValues
+      origValue = p[1]
+      v = normalize(p[0], origValue)
+
+      path = if common.endsWith(p[0], '.') then p[0] else "#{p[0]}."
+      if v == origValue
+        deletes["#{path}_orig"] = 1
+      else
+        updates["#{path}_orig"] = origValue
+      updates["#{path}_value"] = v
+      updates["#{path}_timestamp"] = now
+      if p[2]?
+        updates["#{path}_type"] = p[2]
+      else
+        deletes["#{path}_type"] = 1
+
+  if actions.deletedObjects?
+    for p in actions.deletedObjects
+      deletes[p] = 1
+
+  if actions.instanceName?
+    for i in actions.instanceName
+      updates["#{i[0]}._name"] = i[1]
+
+  if actions.parameterNames?
+    for p in actions.parameterNames
+      if OBJECT_REGEX.test(p[0])
+        path = p[0]
+        if INSTANCE_REGEX.test(p[0])
+          updates["#{path}_instance"] = true
+        else
+          updates["#{path}_object"] = true
+      else
+        path = "#{p[0]}."
+      updates["#{path}_writable"] = p[1] if p[1]?
+      updates["#{path}_timestamp"] = now
+
+      # Add parameter (sans any dot suffix) if has an alias
+      if (a = parameters.getAlias(pp = path.slice(0, -1)))?
+        newAliases = {} if not newAliases?
+        newAliases[pp] = a
+
+  if actions.customCommands?
+    for p in actions.customCommands
+      commandName = p[0]
+      updates["_customCommands.#{commandName}._value"] = p[1]
+      updates["_customCommands.#{commandName}._timestamp"] = now
+
+  if actions.set?
+    common.extend(updates, actions.set)
+
+  if Object.keys(updates).length > 0 or Object.keys(deletes).length > 0
+    db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates, '$unset' : deletes}, {}, (err, count) ->
+      throw err if err
+      # Clear aliases cache if there's a new aliased parameter that has not been included in aliases cache
+      if newAliases?
+        db.getAliases((aliases) ->
+          for p, a of newAliases
+            if not aliases[a]? or p not in aliases[a]
+              db.redisClient.del('aliases', (err, res) ->
+                throw err if err
+              )
+              break
+        )
+      callback?(err)
+    )
+  else
+    callback?()
+
+
+inform = (currentRequest, cwmpRequest) ->
+  # If overloaded, ask CPE to retry in 5 mins
+  if Date.now() < holdUntil
+    res = {code : 503, headers : {'Content-Length' : 0, 'Retry-After' : 300}, data : ''}
+    return writeResponse(currentRequest, res)
+
+  now = new Date(Date.now())
+
+  util.log("#{currentRequest.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") if config.LOG_INFORMS
+
+  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')
+
+  actions = {parameterValues : cwmpRequest.methodRequest.parameterList, set : {}}
+  actions.set._lastInform = now
+  actions.set._lastBoot = now if '1 BOOT' in cwmpRequest.methodRequest.event
+  actions.set._lastBootstrap = lastBootstrap = now if '0 BOOTSTRAP' in cwmpRequest.methodRequest.event
+
+  db.redisClient.get("#{currentRequest.deviceId}_inform_hash", (err, res) ->
+    throw err if err
+
+    if not res?
+      db.redisClient.setex("#{currentRequest.deviceId}_inform_hash", config.PRESETS_CACHE_DURATION, informHash, (err, res) ->
+        throw err if err
+      )
+
+    updateAndRespond = () ->
+      updateDevice(currentRequest, actions, (err) ->
+        throw err if err
+        res = soap.response(cwmpRequest.id, {methodResponse : {type : 'InformResponse'}}, {deviceId : currentRequest.deviceId})
+        writeResponse(currentRequest, res)
+      )
+
+    if res == informHash
+      return updateAndRespond()
+
+    # 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.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.deviceId, name : 'refreshObject', objectName : reference[k]._path})
+          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.deviceId)
+          if not (deviceCustomCommands[cmd]?._timestamp < lastBootstrap)
+            _tasks.push({device: currentRequest.deviceId, name: 'customCommand', command: "#{cmd} init"})
+          delete deviceCustomCommands[cmd]
+
+        for cmd of deviceCustomCommands
+          actions.deletedObjects ?= []
+          actions.deletedObjects.push("_customCommands.#{cmd}")
+
+        apiFunctions.insertTasks(_tasks, {}, () ->
+          updateAndRespond()
+        )
+      else
+        deviceIdDetails = {}
+        for k, v of cwmpRequest.methodRequest.deviceId
+          deviceIdDetails["_#{k}"] = v
+
+        db.devicesCollection.insert({_id : currentRequest.deviceId, _registered : now, _deviceId : deviceIdDetails}, (err) ->
+          throw err if err
+          util.log("#{currentRequest.deviceId}: New device registered")
+          _tasks.push({device: currentRequest.deviceId, name : 'refreshObject', objectName : ''})
+
+          for cmd in customCommands.getDeviceCustomCommandNames(currentRequest.deviceId)
+            _tasks.push({device: currentRequest.deviceId, name: 'customCommand', command: "#{cmd} init"})
+
+          apiFunctions.insertTasks(_tasks, {}, () ->
+            updateAndRespond()
+          )
+        )
+    )
+  )
+
+
+runTask = (currentRequest, task, methodResponse) ->
+  timeDiff = process.hrtime()
+  tasks.task(task, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
+    throw err if err
+    # TODO handle error
+    updateDevice(currentRequest, deviceUpdates, (err) ->
+      throw err if err
+
+      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)
+
+      save = status & tasks.STATUS_SAVE
+      switch status & ~tasks.STATUS_SAVE
+        when tasks.STATUS_OK
+          f = () ->
+            db.redisClient.setex(String(task._id), config.CACHE_DURATION, JSON.stringify(task), (err) ->
+              throw err if err
+              res = soap.response(task._id, cwmpResponse)
+              writeResponse(currentRequest, res)
+            )
+
+          if save
+            db.tasksCollection.update({_id : mongodb.ObjectID(String(task._id))}, {$set : {session : task.session}}, (err) ->
+              throw err if err
+              f()
+            )
+          else
+            f()
+        when tasks.STATUS_COMPLETED
+          util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
+          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.RETRY_DELAY * Math.pow(2, task.retries ? 0)
+          util.log("#{currentRequest.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
+
+          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')
+    )
+  )
+
+
+isTaskExpired = (task) ->
+  task.expiry <= new Date()
+
+
+assertPresets = (currentRequest) ->
+  db.redisClient.mget("#{currentRequest.deviceId}_presets_hash", 'presets_hash', (err, res) ->
+    throw err if err
+    devicePresetsHash = res[0]
+    presetsHash = res[1]
+    if devicePresetsHash? and devicePresetsHash == presetsHash
+      # no discrepancy, return empty response
+      res = soap.response(null, {})
+      writeResponse(currentRequest, res)
+    else
+      db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
+        if not presetsHash?
+          presetsHash = presets.calculatePresetsHash(allPresets, allObjects)
+          db.redisClient.setex('presets_hash', config.PRESETS_CACHE_DURATION, presetsHash, (err, res) ->
+            throw err if err
+          )
+
+        presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, allAliases, (devicePreset) ->
+          presets.processDevicePreset(currentRequest.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) ->
+            db.redisClient.setex("#{currentRequest.deviceId}_presets_hash", Math.floor(Math.max(1, expiry - config.PRESETS_TIME_PADDING)), presetsHash, (err, res) ->
+              throw err if err
+            )
+
+            if addTags.length + deleteTags.length + taskList.length
+              util.log("#{currentRequest.deviceId}: Presets discrepancy found")
+
+            if addTags.length + deleteTags.length
+              util.log("#{currentRequest.deviceId}: Updating tags +(#{addTags}) -(#{deleteTags})")
+
+            if deleteTags.length
+              db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$pull' : {'_tags' : {'$in' : deleteTags}}}, {}, (err, count) ->
+                throw err if err
+              )
+
+            if addTags.length
+              db.devicesCollection.update({'_id' : currentRequest.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.deviceId}: Started task #{task.name}(#{task._id})")
+                runTask(currentRequest, task, {})
+              )
+            else
+              res = soap.response(null, {}, {})
+              writeResponse(currentRequest, res)
+          )
+        )
+      )
+  )
+
+
+nextTask = (currentRequest) ->
+  now = new Date()
+  cur = db.tasksCollection.find({'device' : currentRequest.deviceId, timestamp : {$lte : now}}).sort(['timestamp']).limit(1)
+  cur.nextObject((err, task) ->
+    throw err if err
+    cwmpResponse = {}
+
+    if not task
+      # no more tasks, check presets discrepancy
+      assertPresets(currentRequest)
+    else if isTaskExpired(task)
+      util.log("#{currentRequest.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.deviceId}: Started task #{task.name}(#{task._id})")
+      runTask(currentRequest, task, {})
+  )
+
+
+listener = (httpRequest, httpResponse) ->
+  if httpRequest.method != 'POST'
+    httpResponse.writeHead 405, {'Allow': 'POST'}
+    httpResponse.end('405 Method Not Allowed')
+    return
+
+  chunks = []
+  bytes = 0
+  cookies = {}
+
+  httpRequest.addListener 'data', (chunk) ->
+    chunks.push(chunk)
+    bytes += chunk.length
+
+  httpRequest.getBody = (encoding) ->
+    # Write all chunks into a Buffer
+    body = new Buffer(bytes)
+    offset = 0
+    chunks.forEach((chunk) ->
+      chunk.copy(body, offset, 0, chunk.length)
+      offset += chunk.length
+    )
+
+    #Return encoded (default to UTF8) string
+    return body.toString(encoding || 'utf8', 0, body.byteLength)
+
+  httpRequest.addListener 'end', () ->
+    cwmpResponse = {}
+    cwmpRequest = soap.request(httpRequest)
+
+    currentRequest = {
+      httpRequest : httpRequest
+      httpResponse : httpResponse
+    }
+
+    if cwmpRequest.methodRequest?.deviceId?
+      currentRequest.deviceId = common.getDeviceId(cwmpRequest.methodRequest.deviceId)
+    else
+      currentRequest.deviceId = cwmpRequest.cookies.deviceId
+
+    if config.DEBUG_DEVICES[currentRequest.deviceId]
+      dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n"
+      require('fs').appendFile("debug/#{currentRequest.deviceId}.dump", dump, (err) ->
+        throw err if err
+      )
+
+    if cwmpRequest.methodRequest?
+      if cwmpRequest.methodRequest.type is 'Inform'
+        inform(currentRequest, cwmpRequest)
+      else if cwmpRequest.methodRequest.type is 'TransferComplete'
+        # do nothing
+        util.log("#{currentRequest.deviceId}: Transfer complete")
+        cwmpResponse.methodResponse = {type : 'TransferCompleteResponse'}
+        res = soap.response(cwmpRequest.id, cwmpResponse, cookies)
+        writeResponse(currentRequest, res)
+      else if cwmpRequest.methodRequest.type is 'RequestDownload'
+        requestDownloadResponse = () ->
+          cwmpResponse.methodResponse = {type : 'RequestDownloadResponse'}
+          res = soap.response(cwmpRequest.id, cwmpResponse, cookies)
+          writeResponse(currentRequest, res)
+        fileType = cwmpRequest.methodRequest.fileType
+        util.log("#{currentRequest.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.deviceId, allPresets, allObjects, allAliases, (devicePreset) ->
+            presetSoftwareVersion = devicePreset.softwareVersion?.preset
+            currentSoftwareVersion = devicePreset.softwareVersion?.current._value
+            if presetSoftwareVersion? and presetSoftwareVersion != currentSoftwareVersion
+              projection = {
+                '_deviceId._Manufacturer' : 1,
+                '_deviceId._ProductClass' : 1,
+              }
+              db.devicesCollection.findOne({'_id' : currentRequest.deviceId}, projection, (err, device) ->
+                throw err if err
+                manufacturer = device._deviceId._Manufacturer
+                productClass = device._deviceId._ProductClass
+                db.filesCollection.findOne({'metadata.fileType' : '1 Firmware Upgrade Image', 'metadata.manufacturer' : manufacturer, 'metadata.productClass' : productClass, 'metadata.version' : presetSoftwareVersion}, {_id : 1}, (err, file) ->
+                  throw err if err
+                  if not file?
+                    util.error("#{currentRequest.deviceId}: Firmware image not found (#{presetSoftwareVersion})")
+                    return requestDownloadResponse()
+
+                  task = {
+                    device : currentRequest.deviceId,
+                    name : 'download',
+                    file : file['_id']
+                  }
+                  apiFunctions.insertTasks(task, allAliases, (err, tasks) ->
+                    throw err if err
+                    return requestDownloadResponse()
+                  )
+                )
+              )
+            else
+              return requestDownloadResponse()
+          )
+        )
+      else
+        throw new Error('ACS method not supported')
+    else if cwmpRequest.methodResponse?
+      taskId = cwmpRequest.id
+
+      db.getTask(taskId, (err, task) ->
+        throw err if err
+        if not task
+          nextTask(currentRequest)
+        else
+          runTask(currentRequest, task, cwmpRequest.methodResponse)
+      )
+    else if cwmpRequest.fault?
+      taskId = cwmpRequest.id
+      if not taskId
+        # Fault not related to a task. return empty response.
+        res = soap.response(null, {}, {})
+        writeResponse(currentRequest, res)
+        return
+
+      db.getTask(taskId, (err, task) ->
+        throw err if err
+        if not task
+          nextTask(currentRequest)
+        else
+          runTask(currentRequest, task, cwmpRequest.fault)
+      )
+    else
+      # cpe sent empty response. start sending acs requests
+      nextTask(currentRequest)
+
+
+cluster = require 'cluster'
+numCPUs = require('os').cpus().length
+
+if cluster.isMaster
+  db.redisClient.flushdb()
+  cluster.on('listening', (worker, address) ->
+    util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
+  )
+
+  cluster.on('exit', (worker, code, signal) ->
+    util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
+    setTimeout(()->
+      cluster.fork()
+    , config.WORKER_RESPAWN_TIME)
+  )
+
+  for [1 .. numCPUs]
+    cluster.fork()
+else
+  options = {
+    key: fs.readFileSync('../config/httpscert.key'),
+    cert: fs.readFileSync('../config/httpscert.crt')
+  }
+
+  httpServer = http.createServer(listener)
+  httpsServer = https.createServer(options, listener)
+
+  # wait until DB connections are established
+  setTimeout(() ->
+    httpServer.listen(config.ACS_PORT, config.ACS_INTERFACE)
+    httpsServer.listen(config.ACS_HTTPS_PORT, config.ACS_HTTPS_INTERFACE)
+  , 1000)

lib/db.coffee

@@ -0,0 +1,155 @@
+config = require './config'
+mongodb = require 'mongodb'
+redisClient = require('redis').createClient(config.REDIS_SOCKET)
+parameters = require './parameters'
+
+tasksCollection = null
+devicesCollection = null
+presetsCollection = null
+objectsCollection = null
+
+mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABASE_NAME}", config.MONGODB_OPTIONS, (err, db) ->
+  exports.mongoDb = db
+  db.collection('tasks', (err, collection) ->
+    throw new Error(err) if err?
+    exports.tasksCollection = tasksCollection = collection
+    collection.ensureIndex({device: 1, timestamp: 1}, (err) ->
+    )
+  )
+
+  db.collection('devices', (err, collection) ->
+    throw new Error(err) if err?
+    exports.devicesCollection  = devicesCollection = collection
+  )
+
+  db.collection('presets', (err, collection) ->
+    throw new Error(err) if err?
+    exports.presetsCollection = presetsCollection = collection
+  )
+
+  db.collection('objects', (err, collection) ->
+    throw new Error(err) if err?
+    exports.objectsCollection = objectsCollection = collection
+  )
+
+  db.collection('fs.files', (err, collection) ->
+    throw new Error(err) if err?
+    exports.filesCollection = filesCollection = collection
+  )
+)
+
+
+getTask = (taskId, callback) ->
+  tid = String(taskId)
+  redisClient.get(tid, (err, res) ->
+    return callback(err) if err
+    if res?
+      callback(err, JSON.parse(res))
+    else
+      tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) ->
+        callback(err, task)
+      )
+  )
+
+
+getCached = (name, valueCallback, valueExpiry, callback) ->
+  redisClient.get(name, (err, res) ->
+    return callback(err) if err
+    return callback(null, JSON.parse(res)) if res?
+
+    lockAcquired = () ->
+      valueCallback((err, value) ->
+        return callback(err, value) if err
+        redisClient.setex(name, valueExpiry, JSON.stringify(value), (err) ->
+          return callback(err, value) if err
+          redisClient.del("lock.#{name}", (err) ->
+            callback(err, value)
+          )
+        )
+      )
+
+    redisClient.setnx("lock.#{name}", Date.now() + 30000, (err, res) ->
+      return callback(err) if err
+      if res == 1
+        lockAcquired()
+      else
+        setTimeout(() ->
+          redisClient.get("lock.#{name}", (err, timestamp) ->
+            return callback(err) if err
+            now = Date.now()
+            if not timestamp?
+              getCached(name, valueCallback, valueExpiry, callback)
+            else if timestamp > now
+              setTimeout(() ->
+                getCached(name, valueCallback, valueExpiry, callback)
+              , 1000)
+            else
+              redisClient.getset("lock.#{name}", now + 30000, (err, timestamp) ->
+                return callback(err) if err
+                if timestamp > now
+                  setTimeout(() ->
+                    getCached(name, valueCallback, valueExpiry, callback)
+                  , 1000)
+                else
+                  lockAcquired()
+              )
+          )
+        , 1000)
+    )
+  )
+
+
+getAliases = (callback) ->
+  getCached('aliases', parameters.compileAliases, config.PRESETS_CACHE_DURATION, (err, res) ->
+    throw err if err
+    callback(res)
+  )
+
+
+getPresetsObjectsAliases = (callback) ->
+  redisClient.mget('presets', 'objects', 'aliases', (err, res) ->
+    presets = JSON.parse(res[0])
+    objects = JSON.parse(res[1])
+    aliases = JSON.parse(res[2])
+
+    if presets and objects and aliases
+      return callback(presets, objects, aliases)
+
+    if not presets
+      getCached('presets', (callback) ->
+        presetsCollection.find().toArray((err, res) ->
+          callback(err, res)
+        )
+      , config.PRESETS_CACHE_DURATION, (err, res) ->
+        throw err if err
+        presets = res
+        callback(presets, objects, aliases) if objects and aliases
+      )
+
+    if not objects
+      getCached('objects', (callback) ->
+        objectsCollection.find().toArray((err, res) ->
+          return callback(err) if err
+          objs = {}
+          objs[r._id] = r for r in res
+          callback(null, objs)
+        )
+      , config.PRESETS_CACHE_DURATION, (err, res) ->
+        throw err if err
+        objects = res
+        callback(presets, objects, aliases) if presets and aliases
+      )
+
+    if not aliases
+      getCached('aliases', parameters.compileAliases, config.PRESETS_CACHE_DURATION, (err, res) ->
+        throw err if err
+        aliases = res
+        callback(presets, objects, aliases) if presets and objects
+      )
+  )
+
+
+exports.redisClient = redisClient
+exports.getTask = getTask
+exports.getPresetsObjectsAliases = getPresetsObjectsAliases
+exports.getAliases = getAliases

lib/fs.coffee

@@ -0,0 +1,51 @@
+config = require './config'
+util = require 'util'
+http = require 'http'
+url = require 'url'
+mongodb = require 'mongodb'
+querystring = require 'querystring'
+
+db = null
+mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABASE_NAME}", config.MONGODB_OPTIONS, (err, _db) ->
+  db = _db
+)
+
+
+cluster = require 'cluster'
+numCPUs = require('os').cpus().length
+
+if cluster.isMaster
+  cluster.on('listening', (worker, address) ->
+    util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
+  )
+
+  cluster.on('exit', (worker, code, signal) ->
+    util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
+    setTimeout(()->
+      cluster.fork()
+    , config.WORKER_RESPAWN_TIME)
+  )
+
+  for i in [1 .. numCPUs]
+    cluster.fork()
+else
+  server = http.createServer((request, response) ->
+    urlParts = url.parse(request.url, true)
+    if request.method == 'GET'
+      filename = querystring.unescape(urlParts.pathname.substring(1))
+      gs = new mongodb.GridStore(db, filename, 'r', {})
+      gs.open((err, gs) ->
+        if err
+          response.writeHead(404)
+          response.end()
+          return
+        stream = gs.stream(true)
+        response.writeHead(200, {'Content-Type' : 'application/octet-stream', 'Content-Length' : gs.length})
+        stream.pipe(response)
+      )
+    else
+      response.writeHead(405, {'Allow': 'GET'})
+      response.end('405 Method Not Allowed')
+  )
+
+  server.listen config.FILES_PORT, config.FILES_INTERFACE

lib/mongo-query.coffee

@@ -0,0 +1,100 @@
+common = require './common'
+
+
+test = (obj, query) ->
+  for k,v of query
+    if k.charAt(0) == '$' # this is a logical operator
+      res = testExpressions(obj, v, k)
+    else
+      value = common.getParamValueFromPath(obj, 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
+
+
+testExpressions = (obj, expressions, lop) ->
+  for f in expressions
+    res = test(obj, 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')
+
+
+# generate field projection from a query
+projection = (query, 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
+
+# optimize projection by removing overlaps
+optimizeProjection = (obj) ->
+  keys = Object.keys(obj).sort()
+  return if keys.length <= 1
+  i = 1
+  while i < keys.length
+    a = keys[i-1]
+    b = keys[i]
+    if common.startsWith(b, a)
+      if b.charAt(a.length) == '.' or b.charAt(a.length - 1) == '.'
+        delete obj[b]
+        keys.splice(i--, 1)
+    ++ i
+
+
+exports.test = test
+exports.projection = projection
+exports.optimizeProjection = optimizeProjection

lib/nbi.coffee

@@ -0,0 +1,363 @@
+config = require './config'
+util = require 'util'
+http = require 'http'
+url = require 'url'
+common = require './common'
+db = require './db'
+mongodb = require 'mongodb'
+querystring = require 'querystring'
+query = require './query'
+apiFunctions = require './api-functions'
+presets = require './presets'
+
+# regular expression objects
+DEVICE_TASKS_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/tasks\/?$/
+TASKS_REGEX = /^\/tasks\/([a-zA-Z0-9\-\_\%]+)(\/[a-zA-Z_]*)?$/
+TAGS_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/tags\/([a-zA-Z0-9\-\_\%]+)\/?$/
+PRESETS_REGEX = /^\/presets\/([a-zA-Z0-9\-\_\%]+)\/?$/
+OBJECTS_REGEX = /^\/objects\/([a-zA-Z0-9\-\_\%]+)\/?$/
+FILES_REGEX = /^\/files\/([a-zA-Z0-9\-\_\%\ \.\/\(\)]+)\/?$/
+PING_REGEX = /^\/ping\/([a-zA-Z0-9\-\_\.]+)\/?$/
+QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/
+DEVICE_PRESET_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/preset\/?$/
+
+
+cluster = require 'cluster'
+numCPUs = require('os').cpus().length
+
+if cluster.isMaster
+  cluster.on('listening', (worker, address) ->
+    util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
+  )
+
+  cluster.on('exit', (worker, code, signal) ->
+    util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
+    setTimeout(()->
+      cluster.fork()
+    , config.WORKER_RESPAWN_TIME)
+  )
+
+  for [1 .. numCPUs]
+    cluster.fork()
+else
+  server = http.createServer (request, response) ->
+    chunks = []
+    bytes = 0
+
+    request.addListener 'data', (chunk) ->
+      chunks.push(chunk)
+      bytes += chunk.length
+
+    request.getBody = (encoding) ->
+      # Write all chunks into a Buffer
+      body = new Buffer(bytes)
+      offset = 0
+      chunks.forEach((chunk) ->
+        chunk.copy(body, offset, 0, chunk.length)
+        offset += chunk.length
+      )
+
+      # Return encoded (default to UTF8) string
+      return body.toString(encoding || 'utf8', 0, body.byteLength)
+
+    request.addListener 'end', () ->
+      body = request.getBody()
+      urlParts = url.parse(request.url, true)
+      if PRESETS_REGEX.test(urlParts.pathname)
+        presetName = querystring.unescape(PRESETS_REGEX.exec(urlParts.pathname)[1])
+        if request.method == 'PUT'
+          preset = JSON.parse(body)
+          preset._id = presetName
+
+          db.presetsCollection.save(preset, (err) ->
+            db.redisClient.del('presets', 'presets_hash', (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else if request.method == 'DELETE'
+          db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) ->
+            db.redisClient.del('presets', 'presets_hash', (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else
+          response.writeHead 405, {'Allow': 'PUT, DELETE'}
+          response.end('405 Method Not Allowed')
+      else if OBJECTS_REGEX.test(urlParts.pathname)
+        objectName = querystring.unescape(OBJECTS_REGEX.exec(urlParts.pathname)[1])
+        if request.method == 'PUT'
+          object = JSON.parse(body)
+          object._id = objectName
+
+          db.objectsCollection.save(object, (err) ->
+            db.redisClient.del('objects', 'presets_hash', (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else if request.method == 'DELETE'
+          db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) ->
+            db.redisClient.del('objects', 'presets_hash', (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else
+          response.writeHead 405, {'Allow': 'PUT, DELETE'}
+          response.end('405 Method Not Allowed')
+      else if TAGS_REGEX.test(urlParts.pathname)
+        r = TAGS_REGEX.exec(urlParts.pathname)
+        deviceId = querystring.unescape(r[1])
+        tag = querystring.unescape(r[2])
+        if request.method == 'POST'
+          db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) ->
+            db.redisClient.del("#{deviceId}_presets_hash", (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else if request.method == 'DELETE'
+          db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) ->
+            db.redisClient.del("#{deviceId}_presets_hash", (err) ->
+              throw err if err
+            )
+            if err
+              response.writeHead(500)
+              response.end(err)
+              return
+            response.writeHead(200)
+            response.end()
+          )
+        else
+          response.writeHead 405, {'Allow': 'POST, DELETE'}
+          response.end('405 Method Not Allowed')
+      else if DEVICE_TASKS_REGEX.test(urlParts.pathname)
+        if request.method == 'POST'
+          deviceId = querystring.unescape(DEVICE_TASKS_REGEX.exec(urlParts.pathname)[1])
+          if body
+            task = JSON.parse(body)
+            task.device = deviceId
+            db.getAliases((aliases) ->
+              apiFunctions.insertTasks(task, aliases, (err) ->
+                db.redisClient.del("#{deviceId}_presets_hash", (err) ->
+                  throw err if err
+                )
+                if err
+                  response.writeHead(500)
+                  response.end(err)
+                  return
+
+                if urlParts.query.connection_request?
+                  apiFunctions.connectionRequest(deviceId, (err) ->
+                    if err
+                      response.writeHead(202)
+                      response.end()
+                    else
+                      apiFunctions.watchTask(task._id, config.DEVICE_ONLINE_THRESHOLD, (err) ->
+                        if err
+                          response.writeHead(202)
+                          response.end()
+                          return
+                        response.writeHead(200)
+                        response.end()
+                      )
+                  )
+                else
+                  response.writeHead(202)
+                  response.end()
+              )
+            )
+          else if urlParts.query.connection_request?
+            # no task, send connection request only
+            apiFunctions.connectionRequest(deviceId, (err) ->
+              if err
+                response.writeHead 504
+                response.end()
+                return
+              response.writeHead 200
+              response.end()
+            )
+          else
+            response.writeHead(400)
+            response.end()
+        else
+          response.writeHead 405, {'Allow': 'POST'}
+          response.end('405 Method Not Allowed')
+      else if DEVICE_PRESET_REGEX.test(urlParts.pathname)
+        deviceId = querystring.unescape(DEVICE_PRESET_REGEX.exec(urlParts.pathname)[1])
+        if request.method is 'GET'
+          db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
+            presets.getDevicePreset(deviceId, allPresets, allObjects, allAliases, (devicePreset) ->
+              response.writeHead(200, {'Content-Type' : 'application/json'})
+              response.end(JSON.stringify(devicePreset))
+            )
+          )
+        else
+          response.writeHead 405, {'Allow': 'GET'}
+          response.end('405 Method Not Allowed')
+      else if TASKS_REGEX.test(urlParts.pathname)
+        r = TASKS_REGEX.exec(urlParts.pathname)
+        taskId = mongodb.ObjectID(querystring.unescape(r[1]))
+        action = try querystring.unescape(r[2])
+        if not action? or action is '/'
+          if request.method == 'DELETE'
+            db.tasksCollection.remove({'_id' : taskId}, (err, removedCount) ->
+              if err
+                response.writeHead(500)
+                response.end(err)
+                return
+              response.writeHead(200)
+              response.end()
+            )
+          else
+            response.writeHead 405, {'Allow': 'PUT DELETE'}
+            response.end('405 Method Not Allowed')
+        else if action is '/retry'
+          if request.method == 'POST'
+            db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) ->
+              # TODO need to invalidate presets hash for the device
+              response.writeHead(200)
+              response.end()
+            )
+          else
+            response.writeHead 405, {'Allow': 'POST'}
+            response.end('405 Method Not Allowed')
+        else
+          response.writeHead(404)
+          response.end()
+      else if FILES_REGEX.test(urlParts.pathname)
+        filename = querystring.unescape(FILES_REGEX.exec(urlParts.pathname)[1])
+        if request.method == 'PUT'
+          metadata = {
+            fileType : request.headers.filetype,
+            manufacturer : request.headers.manufacturer,
+            productClass : request.headers.productclass,
+            version : request.headers.version,
+          }
+
+          gs = new mongodb.GridStore(db.mongoDb, filename, 'w', {metadata : metadata})
+          gs.open((err, gs) ->
+            gs.write(request.getBody('binary'), (err, res) ->
+              throw err if err
+              gs.close((err) ->
+                throw err if err
+              )
+              response.writeHead(201)
+              response.end()
+            )
+          )
+        else if request.method == 'DELETE'
+          mongodb.GridStore.unlink(db.mongoDb, filename, (err) ->
+            response.writeHead(200)
+            response.end()
+          )
+        else
+          response.writeHead 405, {'Allow': 'PUT, DELETE'}
+          response.end('405 Method Not Allowed')
+      else if PING_REGEX.test(urlParts.pathname)
+        host = querystring.unescape(PING_REGEX.exec(urlParts.pathname)[1])
+        require('child_process').exec("ping -w 1 -i 0.2 -c 3 #{host}", (err, stdout, stderr) ->
+          if err?
+            response.writeHead(404, {'Cache-Control' : 'no-cache'})
+            response.end()
+            return
+          response.writeHead(200, {'Content-Type' : 'text/plain', 'Cache-Control' : 'no-cache'})
+          response.end(stdout)
+        )
+      else if QUERY_REGEX.test(urlParts.pathname)
+        collectionName = QUERY_REGEX.exec(urlParts.pathname)[1]
+        if request.method not in ['GET', 'HEAD']
+          response.writeHead 405, {'Allow' : 'GET, HEAD'}
+          response.end('405 Method Not Allowed')
+          return
+        collection = db["#{collectionName}Collection"]
+        if not collection?
+          response.writeHead 404
+          response.end('404 Not Found')
+          return
+
+        func = (aliases) ->
+          if urlParts.query.query?
+            q = JSON.parse(urlParts.query.query)
+          else
+            q = {}
+          q = query.expand(q, aliases) if collectionName is 'devices'
+
+          if urlParts.query.projection?
+            projection = {}
+            for p in urlParts.query.projection.split(',')
+              p = p.trim()
+              projection[p] = 1
+              if collectionName is 'devices'
+                for k,v of aliases
+                  if k == p or common.startsWith(k, "#{p}.")
+                    projection[a] = 1 for a in v
+
+          cur = collection.find(q, projection, {batchSize : 50})
+          if urlParts.query.sort?
+            s = JSON.parse(urlParts.query.sort)
+            sort = {}
+            for k,v of s
+              if aliases[k]?
+                sort[a] = v for a in aliases[k]
+              else
+                sort[k] = v
+            cur.sort(sort)
+
+          cur.skip(parseInt(urlParts.query.skip)) if urlParts.query.skip?
+          cur.limit(parseInt(urlParts.query.limit)) if urlParts.query.limit?
+          cur.count((err, total) ->
+            response.writeHead(200, {'Content-Type' : 'application/json', 'total' : total})
+            if request.method is 'HEAD'
+              response.end()
+              return
+            response.write("[\n")
+            i = 0
+            cur.each((err, item) ->
+              if item is null
+                response.end("\n]")
+              else
+                response.write(",\n") if i++
+                apiFunctions.addAliases(item, aliases) if collectionName is 'devices'
+                response.write(JSON.stringify(item))
+            )
+          )
+
+        if collectionName is 'devices'
+          db.getAliases((aliases) ->
+            func(aliases)
+          )
+        else
+          func({})
+      else
+        response.writeHead 404
+        response.end('404 Not Found')
+
+  server.listen config.API_PORT, config.API_INTERFACE

lib/normalize.coffee

@@ -0,0 +1,109 @@
+common = require './common'
+parameters = require './parameters'
+
+
+stringToRegexp = (input) ->
+  if (input.indexOf('*') == -1)
+    return false
+
+  output = input.replace(/[\[\]\\\^\$\.\|\?\+\(\)]/, "\\$&")
+  if output[0] == '*'
+    prefix = '/'
+    output = output.replace(/^\*+/g, '')
+  else
+    prefix = '/^'
+
+  if output[output.length - 1] == '*'
+    suffix = '/'
+    output = output.replace(/\*+$/g, '')
+  else
+    suffix = '$/'
+
+  output = output.replace(/[\*]/, '.*')
+
+  return eval(prefix + output + suffix)
+
+
+normalizers = {}
+
+normalizers.default = (input, normType) ->
+  if normType is 'query' and common.typeOf(input) is common.STRING_TYPE
+    vals = []
+    str = normalizers.string(input, normType)
+    if common.typeOf(str) == common.ARRAY_TYPE
+      vals = vals.concat(str)
+    else
+      vals.push(str)
+
+    f = parseFloat(input)
+    if not isNaN(f)
+      vals.push(f)
+
+    d = new Date(input)
+    if input.length >= 8 and d.getFullYear() > 1983
+      vals.push(d)
+
+    r = stringToRegexp(input)
+    if r isnt false
+      vals.push({'$regex' : r})
+
+    return vals
+  return input
+
+
+FLOAT_REGEX = /^([\-\+]?[0-9\.]+)/
+normalizers.float = (input, normType) ->
+  res = FLOAT_REGEX.exec(input)
+  if res != null
+    parseFloat(res[1])
+  else
+    null
+
+
+normalizers.date = (input, normType) ->
+  return new Date(input)
+
+
+normalizers.string = (input, normType) ->
+  if normType is 'query'
+    if /^\/(.*?)\/(g?i?m?y?)$/.test(input)
+      return [{'$regex' : eval(input)}, input]
+  input
+
+
+colonizeMac = (input) ->
+  if (i = input.indexOf(':')) != -1
+    i = if (i % 2) == 0 then 0 else 1
+    while i < input.length
+      if input[i-1] != ':' and input[i-2] != ':'
+        if input[i] != ':' and i != 0
+          input = input.substring(0, i) + ':' + input.substring(i)
+          ++i
+        ++i
+      ++ i
+    return input
+  else
+    return input.replace(/[0-9A-F]{1}(?!$)/g, '$&:?')
+
+
+normalizers.mac = (input, normType) ->
+  if normType isnt 'query'
+    return input
+
+  input = input.trim().toUpperCase().replace(/[^0-9A-F\*\-\:]/g, '').replace(/[\-\:]/g, ':')
+
+  if input.indexOf('*') != -1
+    input = (colonizeMac(a) for a in input.split('*')).join('*')
+    return stringToRegexp(input, false)
+
+  input = colonizeMac(input)
+
+  if input.length == 17
+    return input
+
+  return {'$regex' : eval('/' + input + '/')}
+
+
+exports.normalize = (path, value, normType) ->
+  type = parameters.getType(path) ? 'default'
+  normalizers[type](value, normType)

lib/parameters.coffee

@@ -0,0 +1,191 @@
+config = require './config'
+db = require './db'
+
+
+PATH_REGEX = /\w+|(\/.+?\/\w*)/g
+
+# Javascript represents numbers as 64-bit floating point numbers
+# Integers up to 2^53 can be encoded precisely
+MAX_PARAMETERS = 52
+
+
+# get the index of the least significant bit
+getLsb = (num) ->
+  i = -1
+  while num > 0
+    ++ i
+    break if num & 1
+    num >>= 1
+  return i
+
+
+createParameterAttributeFinder = (attributeName) ->
+  # compile attributes into data types optimized for quick lookup
+  attributes = []
+  stringSegments = {}
+  regexSegments = {}
+  paramRegex = {}
+
+  for k, v of config.PARAMETERS
+    continue if not v[attributeName]?
+    attributeIndex = attributes.push(v[attributeName]) - 1
+    throw new Error("There cannot be more than #{MAX_PARAMETERS} configured parameters") if attributeIndex >= MAX_PARAMETERS
+    segmentIndex = 0
+    parts = k.match(PATH_REGEX)
+    for i in [0 ... parts.length]
+      part = parts[i]
+      positionHash = i + parts.length * MAX_PARAMETERS
+
+      if part[0] == '/' # this is a regex part
+        j = part.lastIndexOf('/')
+        regExp = new RegExp(part[1...j], part[j+1..])
+        regexSegments[positionHash] ?= 0
+        regexSegments[positionHash] |= Math.pow(2, attributeIndex)
+        paramRegex[attributeIndex] ?= {}
+        paramRegex[attributeIndex][i] = regExp
+      else
+        stringSegments[positionHash] ?= {}
+        stringSegments[positionHash][part] ?= 0
+        stringSegments[positionHash][part] |= Math.pow(2, attributeIndex)
+
+  # Finder function
+  return (param) ->
+    parts = param.split('.')
+    positionHash = parts.length * MAX_PARAMETERS
+
+    stringIndices = 0 | stringSegments[positionHash]?[parts[0]]
+    allIndices = stringIndices | regexSegments[positionHash]
+
+    for i in [1...parts.length]
+      positionHash = i + parts.length * MAX_PARAMETERS
+      strIdx = stringSegments[positionHash]?[parts[i]]
+      rgxIdx = regexSegments[positionHash]
+      stringIndices &= strIdx
+      allIndices &= (strIdx | rgxIdx)
+
+    if stringIndices > 0
+      return attributes[getLsb(stringIndices)]
+
+    attributeIndex = 0
+    while allIndices > 0
+      if allIndices & 1
+        match = true
+        for i, r of paramRegex[attributeIndex]
+          if not r.test(parts[i])
+            match = false
+            break
+
+        if match
+          return attributes[attributeIndex]
+
+      ++ attributeIndex
+      allIndices >>= 1
+
+    return null
+
+
+splitIds = (batches, callback) ->
+  ids = [{}]
+  return callback(ids) if batches == 1
+  db.devicesCollection.count((err, count) ->
+    return callback(ids) if count <= 1000
+    batchSize = Math.floor(count / batches)
+    for i in [1 ... batches]
+      ids[i] = {}
+      do (i) ->
+        cursor = db.devicesCollection.find({}, {_id:1}).sort({'_id':1}).skip(i * batchSize).limit(1)
+        cursor.nextObject((err, obj) ->
+          ids[i]['$gte'] = obj._id
+          ids[i-1]['$lt'] = obj._id
+          if i == batches - 1
+            callback(ids)
+        )
+  )
+
+
+compileAliases = (callback) ->
+  map = () ->
+    recurse = (partIndex, ref) ->
+      part = parts[partIndex]
+      res = []
+      if part.test?
+        for p of ref
+          if part.test(p)
+            if partIndex == parts.length - 1
+              res.push(p)
+            else
+              for r in recurse(partIndex + 1, ref[p])
+                res.push("#{p}.#{r}")
+      else
+        if ref[part]?
+          if partIndex == parts.length - 1
+            return [part]
+          else
+            for r in recurse(partIndex + 1, ref[part])
+              res.push("#{part}.#{r}")
+
+      return res
+
+    for key, parts of keys
+      for alias in recurse(0, this)
+        emit(alias, key)
+    return
+
+  reduce = (k, v) ->
+    return v[0]
+
+  aliases = {}
+  keys = {}
+  for k, v of config.PARAMETERS
+    alias = v.alias
+    continue if not alias?
+    isStatic = true
+    ar = []
+    for p in k.match(/\w+|(\/.+?\/\w*)/g)
+      if p[0] == '/' # this is a regex
+        isStatic = false
+        i = p.lastIndexOf('/')
+        r = new RegExp(p[1...i], p[i+1..])
+        ar.push(r)
+      else
+        ar.push(p)
+
+    if isStatic
+      aliases[alias] ?= []
+      aliases[alias].push(k)
+    else
+      keys[k] = ar
+
+  counter = 0
+  splitIds(4, (batches) ->
+    for batch in batches
+      options = {
+        out : {inline : 1}
+        jsMode : true
+        sort : {$natural : 1}
+        scope : {keys : keys, flags : {}}
+        query : {_id : batch} if batches.length > 1
+      }
+
+      db.devicesCollection.mapReduce(map, reduce, options, (err, out) ->
+        if err
+          callback?(err)
+          return callback = null
+
+        for o in out
+          alias = config.PARAMETERS[o.value].alias
+          aliases[alias] ?= []
+          id = String(o._id)
+          if id not in aliases[alias]
+            aliases[alias].push(id)
+
+        if ++counter == batches.length
+          callback?(null, aliases, config.ALIASES_CACHE)
+      )
+    return
+  )
+
+
+exports.getType = createParameterAttributeFinder('type')
+exports.getAlias = createParameterAttributeFinder('alias')
+exports.compileAliases = compileAliases

lib/presets.coffee

@@ -0,0 +1,282 @@
+config = require './config'
+common = require './common'
+db = require './db'
+mongoQuery = require './mongo-query'
+query = require './query'
+log = require('util').log
+customCommands = require './custom-commands'
+
+
+exports.calculatePresetsHash = (presets, objects) ->
+  crypto = require('crypto')
+  hash = crypto.createHash('md5').update(JSON.stringify(presets) + JSON.stringify(objects)).digest('hex')
+  return hash
+
+
+matchObject = (object, param) ->
+  return false if not object._keys? or object._keys.length == 0
+  for k in object._keys
+    return false if not param[k]?
+    v = common.matchType(param[k]._value, object[k])
+    if param[k]._value != v
+      return false
+  return true
+
+
+exports.getDevicePreset = (deviceId, presets, objects, aliases, callback) ->
+  deviceCustomCommands = customCommands.getDeviceCustomCommands(deviceId)
+  # only fetch relevant params
+  projection = {_id : 1}
+  for p in presets
+    p.precondition = query.expand(p.precondition, aliases)
+    mongoQuery.projection(p.precondition, projection)
+
+    for c in p.configurations
+      switch c.type
+        when 'value', 'age'
+          projection[c.name] = 1
+        when 'custom_command_value', 'custom_command_age'
+          commandfile = c.command.split(' ', 1)
+          if commandfile of deviceCustomCommands
+            projection["_customCommands.#{c.command.split(' ', 1)}"] = 1
+        when 'software_version'
+          projection['InternetGatewayDevice.DeviceInfo.SoftwareVersion'] = 1
+        when 'add_tag', 'delete_tag'
+          projection['_tags'] = 1
+        when 'add_object', 'delete_object'
+          projection[c.name] = 1
+        else
+          throw new Error('Unknown configuration type')
+
+  mongoQuery.optimizeProjection(projection)
+
+  db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) ->
+    throw err if err
+    devicePresets = []
+    for p in presets
+      if mongoQuery.test(device, p.precondition)
+        devicePresets.push(p)
+
+    configurations = accumulateConfigurations(devicePresets, objects)
+
+    getParamDetails = (device, path) ->
+      p = common.getParamValueFromPath(device, path)
+      return undefined if not p?
+      details = {}
+      for k,v of p
+        details[k] = v if k[0] == '_'
+      return details
+
+    devicePreset = {}
+    for c in configurations
+      switch c.type
+        when 'value'
+          devicePreset.parameters ?= {}
+          devicePreset.parameters[c.name] ?= {}
+          devicePreset.parameters[c.name].preset ?= {}
+          devicePreset.parameters[c.name].preset.value = c.value
+          devicePreset.parameters[c.name].current = getParamDetails(device, c.name)
+
+        when 'age'
+          devicePreset.parameters ?= {}
+          devicePreset.parameters[c.name] ?= {}
+          devicePreset.parameters[c.name].preset ?= {}
+          devicePreset.parameters[c.name].preset.expiry = parseInt(c.age)
+          devicePreset.parameters[c.name].current = getParamDetails(device, c.name)
+
+        when 'add_tag'
+          devicePreset.tags ?= {}
+          devicePreset.tags[c.tag] = 2 | c.tag in (device._tags ? [])
+
+        when 'delete_tag'
+          devicePreset.tags ?= {}
+          devicePreset.tags[c.tag] = 0 | c.tag in (device._tags ? [])
+
+        when 'add_object'
+          devicePreset.objects ?= {}
+          devicePreset.objects[c.name] ?= {}
+          devicePreset.objects[c.name][c.object] ?= {}
+          devicePreset.objects[c.name][c.object].preset = objects[c.object]
+          continue if not param = common.getParamValueFromPath(device, c.name)
+          devicePreset.objects[c.name][c.object].current ?= {}
+          for k,p of param
+            continue if k[0] == '_'
+            if p._name?
+              if p._name == c.object
+                devicePreset.objects[c.name][c.object].current[k] = p
+            else if matchObject(objects[c.object], p)
+              u = {}
+              u["#{c.name}.#{k}._name"] = c.object
+              db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) ->
+                throw err if err
+              )
+              devicePreset.objects[c.name][c.object].current[k] = p
+
+        when 'delete_object'
+          devicePreset.objects ?= {}
+          devicePreset.objects[c.name] ?= {}
+          devicePreset.objects[c.name][c.object] ?= {}
+          continue if not param = common.getParamValueFromPath(device, c.name)
+          devicePreset.objects[c.name][c.object].current ?= {}
+          for k,p of param
+            continue if k[0] == '_'
+            if p._name?
+              if p._name == c.object
+                devicePreset.objects[c.name][c.object].current[k] = p
+            else if matchObject(objects[c.object], p)
+              u = {}
+              u["#{c.name}.#{k}._name"] = c.object
+              db.devicesCollection.update({'_id' : deviceId}, {'$set' : u}, {}, (err, count) ->
+                throw err if err
+              )
+              devicePreset.objects[c.name][c.object].current[k] = p
+
+        when 'custom_command_value'
+          [filename, commandName] = c.command.split(' ', 2)
+          devicePreset.customCommands ?= {}
+          devicePreset.customCommands[filename] ?= {}
+          devicePreset.customCommands[filename].preset ?= {}
+          devicePreset.customCommands[filename].preset.value = c.value
+          devicePreset.customCommands[filename].preset.valueCommand = c.command
+          continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
+          if cmd = common.getParamValueFromPath(device, "_customCommands.#{filename}")
+            devicePreset.customCommands[filename].current = cmd
+          else
+            devicePreset.customCommands[filename].current ?= {}
+
+        when 'custom_command_age'
+          [filename, commandName] = c.command.split(' ', 2)
+          devicePreset.customCommands ?= {}
+          devicePreset.customCommands[filename] ?= {}
+          devicePreset.customCommands[filename].preset ?= {}
+          devicePreset.customCommands[filename].preset.expiry = parseInt(c.age)
+          devicePreset.customCommands[filename].preset.expiryCommand = c.command
+          continue if not deviceCustomCommands[filename]? or commandName not in deviceCustomCommands[filename]
+          if cmd = common.getParamValueFromPath(device, "_customCommands.#{filename}")
+            devicePreset.customCommands[filename].current = cmd
+          else
+            devicePreset.customCommands[filename].current ?= {}
+
+        when 'software_version'
+          devicePreset.softwareVersion ?= {}
+          devicePreset.softwareVersion.preset = c.software_version
+          devicePreset.softwareVersion.current = common.getParamValueFromPath(device, 'InternetGatewayDevice.DeviceInfo.SoftwareVersion')
+
+        else
+          throw new Error('Unknown configuration type')
+
+    callback(devicePreset)
+  )
+
+
+exports.processDevicePreset = (deviceId, devicePreset, callback) ->
+  now = new Date()
+  expiry = config.PRESETS_CACHE_DURATION
+  getParameterValues = []
+  setParameterValues = []
+  addTags = []
+  deleteTags = []
+  taskList = []
+
+  # Parameters
+  for parameterPath, parameterDetails of devicePreset.parameters ? {}
+    continue if not parameterDetails.current?
+    expiry = Math.min(expiry, parameterDetails.preset.expiry) if parameterDetails.preset.expiry?
+    presetValue = parameterDetails.preset.value
+    currentValue = parameterDetails.current._value
+    if presetValue? and currentValue != (presetValue = common.matchType(currentValue, presetValue))
+      setParameterValues.push([parameterPath, presetValue, parameterDetails.current._type])
+    else if parameterDetails.preset.expiry?
+      diff = parameterDetails.preset.expiry - (now - (parameterDetails.current._timestamp ? 0)) / 1000
+      if diff <= config.PRESETS_TIME_PADDING
+        if parameterDetails.current._value?
+          getParameterValues.push(parameterPath)
+        else
+          taskList.push({device : deviceId, name : 'refreshObject', objectName : parameterPath})
+      else
+        expiry = Math.min(expiry, diff)
+
+  # Objects
+  for parameterPath, object of devicePreset.objects ? {}
+    for objectName, objectDetails of object
+      continue if not objectDetails.current?
+      if objectDetails.preset?
+        if Object.keys(objectDetails.current).length > 0
+          for i, obj of objectDetails.current
+            for paramName, paramValue of objectDetails.preset
+              continue if paramName[0] == '_'
+              if obj[paramName]
+                currentValue = obj[paramName]._value
+                presetValue = common.matchType(currentValue, objectDetails.preset[paramName])
+                if currentValue != presetValue
+                  setParameterValues.push(["#{parameterPath}.#{i}.#{paramName}", presetValue, obj[paramName]._type])
+        else
+          vals = []
+          for k,v of objectDetails.preset
+            vals.push([k, v]) if k[0] != '_'
+          taskList.push({device : deviceId, name : 'addObject', objectName : parameterPath, parameterValues : vals, instanceName : objectName})
+      else if Object.keys(objectDetails.current).length > 0
+        for i, obj of objectDetails.current
+          taskList.push({device : deviceId, name : 'deleteObject', objectName : "#{parameterPath}.#{i}"})
+
+  # Tags
+  for tag, v of devicePreset.tags ? {}
+    if v == 1
+      deleteTags.push(tag)
+    else if v == 2
+      addTags.push(tag)
+
+  # Custom commands
+  for commandName, commandDetails of devicePreset.customCommands ? {}
+    continue if not commandDetails.current?
+    expiry = Math.min(expiry, commandDetails.preset.expiry) if commandDetails.preset.expiry?
+    currentValue = commandDetails.current._value
+    presetValue = common.matchType(currentValue, commandDetails.preset.value)
+    if commandDetails.preset.value? and currentValue != presetValue
+      taskList.push({device : deviceId, name : 'customCommand', command : commandDetails.preset.valueCommand})
+    else if commandDetails.preset.expiry?
+      diff = commandDetails.preset.expiry - (now - (commandDetails.current._timestamp ? 0)) / 1000
+      if diff <= config.PRESETS_TIME_PADDING
+        taskList.push({device : deviceId, name : 'customCommand', command : commandDetails.preset.expiryCommand})
+      else
+        expiry = Math.min(expiry, diff)
+
+  if setParameterValues.length
+    taskList.push {device : deviceId, name : 'setParameterValues', parameterValues: setParameterValues, timestamp : now}
+
+  if getParameterValues.length
+    taskList.push {device : deviceId, name : 'getParameterValues', parameterNames: getParameterValues, timestamp : now}
+
+  callback(taskList, addTags, deleteTags, expiry)
+
+
+getObjectHash = (object) ->
+  return object._id if not object._keys? or object._keys.length == 0
+  hash = ''
+  for k in object._keys
+    hash += "#{k}=#{object[k]}"
+  return hash
+
+
+accumulateConfigurations = (presets, objects) ->
+  maxWeights = {}
+  configurations = {}
+  for p in presets
+    for c in p.configurations
+      configurationHash = switch c.type
+        when 'add_tag', 'delete_tag'
+          "tag_#{c.tag}"
+        when 'add_object', 'delete_object'
+          objectHash = getObjectHash(objects[c.object])
+          "object_#{c.name}_#{objectHash}"
+        when 'software_version'
+          'software_version'
+        else
+          "#{c.type}_#{c.name}"
+
+      if not maxWeights[configurationHash]? or p.weight > maxWeights[configurationHash]
+        configurations[configurationHash] = c
+        maxWeights[configurationHash] = p.weight
+
+  configurationsList = (configurations[c] for c of configurations)
+  return configurationsList

lib/query.coffee

@@ -0,0 +1,85 @@
+common = require './common'
+normalize = require './normalize'
+
+
+expandValue = (param, value) ->
+  if common.typeOf(value) is common.ARRAY_TYPE
+    a = []
+    for j in value
+      a = a.concat(expandValue(param, j))
+    return [a]
+  else if common.typeOf(value) isnt common.OBJECT_TYPE
+    n = normalize.normalize(param, value, 'query')
+    if common.typeOf(n) isnt common.ARRAY_TYPE
+      return [n]
+    else
+      return n
+
+  objs = []
+  indices = []
+  keys = []
+  values = []
+  for k,v of value
+    keys.push(k)
+    values.push(expandValue(param, v))
+    indices.push(0)
+
+  i = 0
+  while i < indices.length
+    obj = {}
+    for i in [0...keys.length]
+      obj[keys[i]] = values[i][indices[i]]
+    objs.push(obj)
+
+    for i in [0...indices.length]
+      indices[i] += 1
+      if indices[i] < values[i].length
+        break
+      indices[i] = 0
+  return objs
+
+
+permute = (param, val, aliases) ->
+  keys = []
+  if aliases[param]?
+    for p in aliases[param]
+      keys.push(p)
+  else
+    keys.push(param)
+
+  conditions = []
+  for k in keys
+    values = expandValue(k, val)
+    if k[k.lastIndexOf('.') + 1] != '_'
+      k += '._value'
+  
+    for v in values
+      obj = {}
+      obj[k] = v
+      conditions.push(obj)
+
+  return conditions
+
+
+expand = (query, aliases) ->
+  new_query = {}
+  for k,v of query
+    if k[0] == '$' # operator
+      expressions = []
+      for e in v
+        expressions.push(expand(e, aliases))
+      new_query[k] = expressions
+    else
+      conditions = permute(k, v, aliases)
+      if conditions.length > 1
+        if new_query['$and']?
+          new_query['$and'].push({'$or' : conditions})
+        else
+          new_query['$and'] = [{'$or' : conditions}]
+      else
+        common.extend(new_query, conditions[0])
+
+  return new_query
+
+
+exports.expand = expand

lib/soap.coffee

@@ -0,0 +1,335 @@
+libxmljs = require 'libxmljs'
+
+SERVER_NAME = "GenieACS/#{require('../package.json').version}"
+
+NAMESPACES = {
+  'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/',
+  'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/',
+  'xsd' : 'http://www.w3.org/2001/XMLSchema',
+  'xsi' : 'http://www.w3.org/2001/XMLSchema-instance',
+  'cwmp' : 'urn:dslforum-org:cwmp-1-0',
+}
+
+
+cookiesToObj = (cookieLine) ->
+  cookies = {}
+  for c in cookieLine.trim().split ','
+    nv = c.split '='
+    continue if nv.length < 2
+    cookies[nv[0].trim()] = nv[1].trim()
+  return cookies
+
+
+cookiesToStr = (obj) ->
+  l = []
+  for cn, cv of obj
+    l.push("#{cn}=#{cv}")
+  return l# + 'path=/'
+
+
+event = (xml) ->
+  e.text().trim() for e in xml.find('EventStruct/EventCode', NAMESPACES)
+
+
+parameterInfoList = (xml) ->
+  infoList = []
+  for e in xml.find('ParameterInfoStruct', NAMESPACES)
+    name = e.get('Name').text().trim()
+    writable = Boolean(JSON.parse(e.get('Writable').text()))
+    infoList.push([name, writable])
+  return infoList
+
+
+parameterValueList = (xml) ->
+  valueList = []
+  for e in xml.find('ParameterValueStruct', NAMESPACES)
+    valueType = e.get('Value').attr('type').value().trim()
+    name = e.get('Name').text().trim()
+    value = e.get('Value').text().trim()
+    try
+      switch valueType
+        when 'xsd:boolean'
+          value = Boolean(JSON.parse(value))
+        when 'xsd:unsignedInt'
+          value = JSON.parse(value)
+        when 'xsd:int'
+          value = JSON.parse(value)
+    catch err
+      value = null
+    valueList.push([name, value, valueType])
+  return valueList
+
+
+cpeGetParameterNames = (xml, methodRequest) ->
+  el = xml.node('cwmp:GetParameterNames')
+  el.node('ParameterPath').text(methodRequest.parameterPath)
+  el.node('NextLevel').text(+methodRequest.nextLevel)
+
+
+cpeGetParameterNamesResponse = (xml) ->
+  {parameterList : parameterInfoList(xml.get('ParameterList'))}
+
+
+cpeGetParameterValues = (xml, methodRequest) ->
+  el = xml.node('cwmp:GetParameterValues').node('ParameterNames')
+  el.attr({'soap-enc:arrayType' : "xsd:string[#{methodRequest.parameterNames.length}]"})
+  for p in methodRequest.parameterNames
+    el.node('xsd:string').text(p)
+
+
+cpeGetParameterValuesResponse = (xml) ->
+  {parameterList : parameterValueList(xml.get('ParameterList'))}
+
+
+cpeSetParameterValues = (xml, methodRequest) ->
+  el = xml.node('cwmp:SetParameterValues')
+  paramList = el.node('ParameterList')
+  paramList.attr({'soap-enc:arrayType' : "cwmp:ParameterValueStruct[#{methodRequest.parameterList.length}]"})
+  for i in methodRequest.parameterList
+    pvs = paramList.node('ParameterValueStruct')
+    pvs.node('Name').text(i[0])
+    v = pvs.node('Value')
+    v.text(i[1])
+    v.attr({'xsi:type' : i[2]}) if i[2]?
+  el.node('ParameterKey').text(methodRequest.parameterKey ? '')
+
+
+cpeSetParameterValuesResponse = (xml) ->
+  {status : JSON.parse(xml.get('Status').text())}
+
+
+cpeAddObject = (xml, methodRequest) ->
+  el = xml.node('cwmp:AddObject')
+  el.node('ObjectName').text(methodRequest.objectName)
+  el.node('ParameterKey').text(methodRequest.parameterKey ? '')
+
+
+cpeAddObjectResponse = (xml) ->
+  {
+    instanceNumber : parseInt(xml.get('InstanceNumber').text()),
+    status : parseInt(xml.get('Status').text())
+  }
+
+
+cpeDeleteObject = (xml, methodRequest) ->
+  el = xml.node('cwmp:DeleteObject')
+  el.node('ObjectName').text(methodRequest.objectName)
+  el.node('ParameterKey').text(methodRequest.parameterKey ? '')
+
+
+cpeDeleteObjectResponse = (xml) ->
+  {status : parseInt(xml.get('Status').text())}
+
+
+traverseXml = (xml) ->
+  obj = {}
+  for n in xml.childNodes()
+    if n.type() == 'element'
+      obj[n.name()] = traverseXml(n)
+
+  if Object.keys(obj).length == 0
+    xml.text()
+  else
+    obj
+
+fault = (xml) ->
+  return traverseXml(xml)
+
+
+acsInform = (xml) ->
+  {
+    parameterList : parameterValueList(xml.get('ParameterList')),
+    deviceId : traverseXml(xml.get('DeviceId')),
+    event : event(xml.get('Event')),
+    retryCount : JSON.parse(xml.get('RetryCount').text())
+  }
+
+
+acsInformResponse = (xml) ->
+  xml.node('cwmp:InformResponse').node('MaxEnvelopes').text(1)
+
+
+cpeReboot = (xml, methodRequest) ->
+  el = xml.node('cwmp:Reboot')
+  el.node('CommandKey').text(methodRequest.commandKey ? '')
+
+
+cpeRebootResponse = (xml) ->
+  {}
+
+
+cpeFactoryReset = (xml, methodRequest) ->
+  el = xml.node('cwmp:FactoryReset')
+
+
+cpeFactoryResetResponse = (xml, methodRequest) ->
+  {}
+
+
+cpeDownload = (xml, methodRequest) ->
+  el = xml.node('cwmp:Download')
+  el.node('CommandKey').text(methodRequest.commandKey ? '')
+  el.node('FileType').text(methodRequest.fileType)
+  el.node('URL').text(methodRequest.url)
+  el.node('Username').text(methodRequest.username ? '')
+  el.node('Password').text(methodRequest.password ? '')
+  el.node('FileSize').text(methodRequest.fileSize ? '0')
+  el.node('TargetFileName').text(methodRequest.TargetFileName ? '')
+  el.node('DelaySeconds').text(methodRequest.delaySeconds ? '0')
+  el.node('SuccessURL').text(methodRequest.successUrl ? '')
+  el.node('FailureURL').text(methodRequest.failureUrl ? '')
+
+
+cpeDownloadResponse = (xml) ->
+  {status : JSON.parse(xml.get('Status').text())}
+
+
+acsTransferComplete = (xml) ->
+  {
+    commandKey : xml.get('CommandKey').text(),
+    faultStruct : traverseXml(xml.get('FaultStruct')),
+    startTime : xml.get('StartTime').text(), # TODO convert to datetime
+    completeTime : xml.get('CompleteTime').text() # TODO convert to datetime
+  }
+
+
+acsTransferCompleteResponse = (xml, methodRequest) ->
+  xml.node('cwmp:TransferCompleteResponse').text('')
+
+
+acsRequestDownload = (xml) ->
+  # TODO FileTypeArg
+  {
+    fileType : xml.get('FileType').text()
+  }
+
+
+acsRequestDownloadResponse = (xml, methodRequest) ->
+  xml.node('cwmp:RequestDownloadResponse').text('')
+
+
+exports.request = (httpRequest) ->
+  cwmpRequest = {cookies: {}}
+  cwmpRequest.cookies = cookiesToObj(httpRequest.headers.cookie) if httpRequest.headers.cookie
+
+  data = httpRequest.getBody()
+
+  if +httpRequest.headers['Content-Length'] > 0 || data.length > 0
+    try
+      xml = libxmljs.parseXml data
+    catch err
+      # some devices send invalid utf8 characters
+      xml = libxmljs.parseXml httpRequest.getBody('binary')
+
+    try
+      cwmpRequest.id = xml.get('//soap-env:Envelope/soap-env:Header/cwmp:ID', NAMESPACES).text()
+    catch err
+      cwmpRequest.id = null
+
+    methodElement = xml.get('/soap-env:Envelope/soap-env:Body/cwmp:*', NAMESPACES)
+    if methodElement?
+      switch methodElement.name()
+        when 'Inform'
+          cwmpRequest.methodRequest = acsInform(methodElement)
+          cwmpRequest.methodRequest.type = 'Inform'
+        when 'TransferComplete'
+          cwmpRequest.methodRequest = acsTransferComplete(methodElement)
+          cwmpRequest.methodRequest.type = 'TransferComplete'
+        when 'RequestDownload'
+          cwmpRequest.methodRequest = acsRequestDownload(methodElement)
+          cwmpRequest.methodRequest.type = 'RequestDownload'
+        when 'GetParameterNamesResponse'
+          cwmpRequest.methodResponse = cpeGetParameterNamesResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'GetParameterNamesResponse'
+        when 'GetParameterValuesResponse'
+          cwmpRequest.methodResponse = cpeGetParameterValuesResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'GetParameterValuesResponse'
+        when 'SetParameterValuesResponse'
+          cwmpRequest.methodResponse = cpeSetParameterValuesResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'SetParameterValuesResponse'
+        when 'AddObjectResponse'
+          cwmpRequest.methodResponse = cpeAddObjectResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'AddObjectResponse'
+        when 'DeleteObjectResponse'
+          cwmpRequest.methodResponse = cpeDeleteObjectResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'DeleteObjectResponse'
+        when 'RebootResponse'
+          cwmpRequest.methodResponse = cpeRebootResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'RebootResponse'
+        when 'FactoryResetResponse'
+          cwmpRequest.methodResponse = cpeFactoryResetResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'FactoryResetResponse'
+        when 'DownloadResponse'
+          cwmpRequest.methodResponse = cpeDownloadResponse(methodElement)
+          cwmpRequest.methodResponse.type = 'DownloadResponse'
+        else
+          throw Error('8000 Method not supported ' + methodElement.name())
+    else
+      faultElement = xml.get('/soap-env:Envelope/soap-env:Body/soap-env:Fault', NAMESPACES)
+      cwmpRequest.fault = fault(faultElement)
+    
+  return cwmpRequest
+
+
+exports.response = (id, cwmpResponse, cookies = null) ->
+  headers = {
+    'Content-Type' : 'text/xml; charset="utf-8"',
+    'Server' : SERVER_NAME,
+    'SOAPServer' : SERVER_NAME,
+  }
+
+  if cookies? and Object.keys(cookies).length > 0
+    headers['Set-Cookie'] = cookiesToStr(cookies)
+
+  if not cwmpResponse? or Object.keys(cwmpResponse).length == 0
+    #console.log '>>> EMPTY RESPONSE'
+    # send empty response
+    headers['Content-Length'] = 0
+    return {code: 204, headers: headers, data: ''}
+
+  xml = libxmljs.Document()
+  env = xml.node('soap-env:Envelope')
+  for prefix, url of NAMESPACES
+    env.defineNamespace prefix, url
+
+  header = env.node('soap-env:Header')
+  header.node('cwmp:ID').attr({'soap-env:mustUnderstand' : 1}).text(id)
+  body = env.node('soap-env:Body')
+
+  if cwmpResponse.methodResponse?
+    switch cwmpResponse.methodResponse.type
+      when 'InformResponse'
+        acsInformResponse(body)
+      when 'TransferCompleteResponse'
+        acsTransferCompleteResponse(body)
+      when 'RequestDownloadResponse'
+        acsRequestDownloadResponse(body)
+      else
+        throw Error("Unknown method response type #{cwmpResponse.methodResponse.type}")
+  else if cwmpResponse.methodRequest?
+    switch cwmpResponse.methodRequest.type
+      when 'GetParameterNames'
+        cpeGetParameterNames(body, cwmpResponse.methodRequest)
+      when 'GetParameterValues'
+        cpeGetParameterValues(body, cwmpResponse.methodRequest)
+      when 'SetParameterValues'
+        cpeSetParameterValues(body, cwmpResponse.methodRequest)
+      when 'AddObject'
+        cpeAddObject(body, cwmpResponse.methodRequest)
+      when 'DeleteObject'
+        cpeDeleteObject(body, cwmpResponse.methodRequest)
+      when 'Reboot'
+        cpeReboot(body, cwmpResponse.methodRequest)
+      when 'FactoryReset'
+        cpeFactoryReset(body, cwmpResponse.methodRequest)
+      when 'Download'
+        cpeDownload(body, cwmpResponse.methodRequest)
+      else
+        throw Error("Unknown method request #{cwmpResponse.methodRequest.type}")
+  else
+    throw Error('Invalid response arguments')
+  
+  data = xml.toString()
+  headers['Content-Length'] = data.length
+
+  return {code: 200, headers: headers, data: data}

lib/tasks.coffee

@@ -0,0 +1,349 @@
+config = require './config'
+common = require './common'
+db = require './db'
+mongodb = require 'mongodb'
+customCommands = require './custom-commands'
+BATCH_SIZE = 16
+
+exports.STATUS_OK = STATUS_OK = 1
+exports.STATUS_FAULT = STATUS_FAULT = 2
+exports.STATUS_COMPLETED = STATUS_COMPLETED = 4
+exports.STATUS_SAVE = STATUS_SAVE = 128 # not mutually exclusive with the rest
+
+
+this.refreshObject = (task, methodResponse, callback) ->
+  task.session = {} if not task.session?
+
+  if not task.session.subtask?
+    path = task.objectName
+    path += '.' if path != '' and not common.endsWith(path, '.')
+    task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : path, nextLevel : false}
+
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if task.session.subtask.name is 'getParameterNames'
+    this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
+      if status & STATUS_COMPLETED
+        parameterNames = []
+        for p in deviceUpdates.parameterNames
+          if not common.endsWith(p[0], '.')
+            parameterNames.push(p[0])
+        task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames}
+        this.getParameterValues(task.session.subtask, {}, (err, status, cwmpResponse) ->
+          # ignore deviceUpdates returned by firt call to getParameterValues
+          callback(err, status, cwmpResponse, deviceUpdates)
+        )
+      else if status & STATUS_OK
+        callback(err, STATUS_OK, cwmpResponse, deviceUpdates)
+      else
+        throw Error('Unexpected subtask status')
+    )
+  else if task.session.subtask.name is 'getParameterValues'
+    if methodResponse.faultcode?
+      task.fault = methodResponse
+      return callback(null, STATUS_FAULT)
+
+    this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) ->
+      callback(err, status, cwmpResponse, deviceUpdates)
+    )
+  else
+    throw Error('Unexpected subtask name')
+
+
+findMissingParameters = (device, parameterList, prefix) ->
+  missingParameters = []
+  paths = {}
+  for param in parameterList
+    p = param[0]
+    paths[p] = true
+    i = p.indexOf('.')
+    while i != -1
+      paths[p[0...i]] = true
+      i = p.indexOf('.', i + 1)
+
+  recursive = (obj, prefix) ->
+    for k,v of obj
+      continue if k[0] == '_'
+      p = prefix + k
+
+      if paths[p]
+        recursive(v, "#{p}.")
+      else
+        missingParameters.push(p)
+
+  recursive(device, prefix)
+  return missingParameters
+
+
+this.getParameterNames = (task, methodResponse, callback) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if methodResponse.type is 'GetParameterNamesResponse'
+    # Make sure that for each parameters, all its parents are included
+    found = {}
+    for p in methodResponse.parameterList
+      param = p[0]
+      i = param.length
+      while (i = param.lastIndexOf('.', i-1)) > task.parameterPath.length
+        pp = param.slice(0, i + 1)
+        break if found[pp]?
+        found[pp] = 0
+      found[p[0]] = 1
+
+    for k, v of found
+      methodResponse.parameterList.push([k]) if v == 0
+
+    deviceUpdates = {parameterNames : methodResponse.parameterList}
+    path = if common.endsWith(task.parameterPath, '.') then task.parameterPath.slice(0, -1) else task.parameterPath
+    projection = {}
+    projection[path] = 1 if !!task.parameterPath
+
+    # delete nonexisting params
+    db.devicesCollection.findOne({_id : task.device}, projection, (err, device) ->
+      if device
+        root = device
+        rootPath = ''
+        if !!task.parameterPath
+          ps = path.split('.')
+          for p in ps
+            rootPath += "#{p}."
+            root = root[p]
+            break if not root?
+
+        if root
+          deviceUpdates.deletedObjects = findMissingParameters(root, methodResponse.parameterList, rootPath)
+        else
+          # avoid adding and deleting the same param
+          if rootPath isnt task.parameterPath
+            deviceUpdates.deletedObjects = [rootPath.slice(0, -1)]
+
+      # some devices don't return the root object as described in the standard. add manually to update timestamp
+      deviceUpdates.parameterNames.push([task.parameterPath]) if !!task.parameterPath
+
+      callback(null, STATUS_COMPLETED, null, deviceUpdates)
+    )
+  else
+    methodRequest = {
+      type : 'GetParameterNames',
+      parameterPath : task.parameterPath,
+      nextLevel : if task.nextLevel? then task.nextLevel else false
+    }
+    callback(null, STATUS_OK, {methodRequest : methodRequest})
+
+
+this.getParameterValues = (task, methodResponse, callback) ->
+  task.session = {} if not task.session?
+
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if not task.session.currentIndex?
+    task.session.currentIndex = 0
+  else if methodResponse.parameterList?
+    task.session.currentIndex = task.session.nextIndex
+
+  task.session.nextIndex = Math.min(task.session.currentIndex + BATCH_SIZE, task.parameterNames.length)
+  names = task.parameterNames.slice(task.session.currentIndex, task.session.nextIndex)
+
+  if methodResponse.type is 'GetParameterValuesResponse'
+    deviceUpdates = {parameterValues : methodResponse.parameterList}
+
+  if names.length == 0
+    callback(null, STATUS_COMPLETED, null, deviceUpdates)
+  else
+    methodRequest = {
+      type : 'GetParameterValues',
+      parameterNames : names
+    }
+    callback(null, STATUS_OK, {methodRequest : methodRequest}, deviceUpdates)
+
+
+this.setParameterValues = (task, methodResponse, callback) ->
+  task.session = {} if not task.session?
+
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if not task.session.currentIndex?
+    task.session.currentIndex = 0
+  else if methodResponse.type is 'SetParameterValuesResponse'
+    prevValues = task.parameterValues.slice(task.session.currentIndex, task.session.nextIndex)
+    task.session.currentIndex = task.session.nextIndex
+
+  task.session.nextIndex = Math.min(task.session.currentIndex + BATCH_SIZE, task.parameterValues.length)
+  values = task.parameterValues.slice(task.session.currentIndex, task.session.nextIndex)
+
+  if prevValues?
+    deviceUpdates = {parameterValues : prevValues}
+
+  if values.length == 0
+    callback(null, STATUS_COMPLETED, null, deviceUpdates)
+  else
+    callback(null, STATUS_OK, {methodRequest : {type : 'SetParameterValues', parameterList : values}}, deviceUpdates)
+
+
+this.addObject = (task, methodResponse, callback) ->
+  task.session = {} if not task.session?
+
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  allDeviceUpdates = {}
+
+  if not task.session.instanceNumber?
+    if methodResponse.type is 'AddObjectResponse'
+      status_save = STATUS_SAVE
+      task.session.instanceNumber = methodResponse.instanceNumber
+      task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.#{task.session.instanceNumber}.", nextLevel : false}
+      task.session.appliedParameterValues = []
+      task.session.parameterNames = []
+      allDeviceUpdates.instanceName = [["#{task.objectName}.#{task.session.instanceNumber}", task.instanceName]] if task.instanceName?
+    else
+      callback(null, STATUS_OK, {methodRequest : {type : 'AddObject', objectName : "#{task.objectName}."}})
+      return
+
+  subtask = () =>
+    switch task.session.subtask.name
+      when 'getParameterNames'
+        this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
+          common.extend(allDeviceUpdates, deviceUpdates)
+          if deviceUpdates and deviceUpdates.parameterNames
+            for p in deviceUpdates.parameterNames
+              task.session.parameterNames.push(p[0]) if not common.endsWith(p[0], '.')
+
+          if status & STATUS_COMPLETED
+            task.session.subtask = {name : 'getParameterValues', parameterNames : task.session.parameterNames}
+            subtask()
+          else if status & STATUS_OK
+            # Use STATUS_SAVE to avoid adding duplicate object in case of error
+            callback(err, STATUS_OK | status_save, cwmpResponse, allDeviceUpdates)
+          else
+            throw Error('Unexpected subtask status')
+        )
+      when 'getParameterValues'
+        this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
+          common.extend(allDeviceUpdates, deviceUpdates)
+          # if values are given, compare with default values
+          if task.parameterValues?
+            if deviceUpdates and deviceUpdates.parameterValues?
+              for p1 in deviceUpdates.parameterValues
+                for p2 in task.parameterValues
+                  if common.endsWith(p1[0], ".#{p2[0]}")
+                    t = if p2[2] then p2[2] else p1[2]
+                    v = common.matchType(p1[1], p2[1])
+                    # TODO only include if writable
+                    task.session.appliedParameterValues.push([p1[0], v, t])
+
+          if methodResponse.faultcode?
+            task.fault = methodResponse
+            return callback(null, STATUS_FAULT)
+
+          if status & STATUS_COMPLETED and task.session.appliedParameterValues.length > 0
+            task.session.subtask = {name : 'setParameterValues', parameterValues : task.session.appliedParameterValues}
+            subtask()
+          else
+            callback(err, status, cwmpResponse, allDeviceUpdates)
+        )
+      when 'setParameterValues'
+        this.setParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) =>
+          common.extend(allDeviceUpdates, deviceUpdates)
+          callback(err, status, cwmpResponse, allDeviceUpdates)
+        )
+  subtask()
+
+
+this.deleteObject = (task, methodResponse, callback) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if methodResponse.type is 'DeleteObjectResponse'
+    callback(null, STATUS_COMPLETED, null, {deletedObjects : [task.objectName]})
+  else
+    methodRequest = {
+      type : 'DeleteObject',
+      objectName : "#{task.objectName}."
+    }
+    callback(null, STATUS_OK, {methodRequest : methodRequest})
+
+
+this.reboot = (task, methodResponse, callback) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+  
+  if methodResponse.type isnt 'RebootResponse'
+    callback(null, STATUS_OK, {methodRequest : {type : 'Reboot'}})
+  else
+    callback(null, STATUS_COMPLETED)
+
+
+this.factoryReset = (task, methodResponse, callback) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if methodResponse.type isnt 'FactoryResetResponse'
+    callback(null, STATUS_OK, {methodRequest : {type : 'FactoryReset'}})
+  else
+    callback(null, STATUS_COMPLETED)
+
+
+this.download = (task, methodResponse, callback) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    callback(null, STATUS_FAULT)
+    return
+
+  if methodResponse.type isnt 'DownloadResponse'
+    db.filesCollection.findOne({_id : mongodb.ObjectID(String(task.file))}, (err, file) ->
+      if not file?
+        callback('File not found')
+        return
+      else if err?
+        callback(err)
+        return
+
+      methodRequest = {
+        type : 'Download',
+        fileType : file.metadata.fileType,
+        fileSize : file.length,
+        url : "http://#{config.FILES_IP}:#{config.FILES_PORT}/#{encodeURIComponent(file.filename)}",
+        successUrl : task.successUrl,
+        failureUrl : task.failureUrl
+      }
+      callback(null, STATUS_OK, {methodRequest : methodRequest})
+    )
+  else
+    callback(null, STATUS_COMPLETED)
+
+
+this.customCommand = (task, methodResponse, callback) ->
+  # TODO implement timeout
+  customCommands.execute(task.device, task.command, (err, value) ->
+    if err?
+      task.fault = err
+      callback(null, STATUS_FAULT)
+    else
+      commandName = task.command.split(' ', 2)[0]
+      callback(null, STATUS_COMPLETED, null, {customCommands : [[commandName, value]]})
+  )
+
+
+exports.task = (task, methodResponse, callback) ->
+  this[task.name](task, methodResponse, callback)