src/poolRequests.js

   1 ;(function () {
   2   'use strict'
   3
   4   var async = require('async')
   5
   6   var logger = require('./logger')
   7   var database = require('./database')
   8   var PoolRequestsDB = database.PoolRequestsDB
   9   var PodsDB = database.PodsDB
  10   var utils = require('./utils')
  11
  12   var poolRequests = {}
  13
  14   // ----------- Constants -----------
  15
  16   // Time to wait between requests to the friends
  17   var INTERVAL = utils.isTestInstance() ? 10000 : 60000
  18   var PODS_SCORE = {
  19     MALUS: -10,
  20     BONUS: 10
  21   }
  22
  23   // ----------- Private -----------
  24   var timer = null
  25
  26   function removePoolRequestsFromDB (ids) {
  27     PoolRequestsDB.remove({ _id: { $in: ids } }, function (err) {
  28       if (err) {
  29         logger.error('Cannot remove requests from the pool requests database.', { error: err })
  30         return
  31       }
  32
  33       logger.info('Pool requests flushed.')
  34     })
  35   }
  36
  37   function makePoolRequests () {
  38     logger.info('Making pool requests to friends.')
  39
  40     PoolRequestsDB.find({}, { _id: 1, type: 1, request: 1 }, function (err, pool_requests) {
  41       if (err) throw err
  42
  43       if (pool_requests.length === 0) return
  44
  45       var requests = {
  46         add: {
  47           ids: [],
  48           requests: []
  49         },
  50         remove: {
  51           ids: [],
  52           requests: []
  53         }
  54       }
  55
  56       async.each(pool_requests, function (pool_request, callback_each) {
  57         if (pool_request.type === 'add') {
  58           requests.add.requests.push(pool_request.request)
  59           requests.add.ids.push(pool_request._id)
  60         } else if (pool_request.type === 'remove') {
  61           requests.remove.requests.push(pool_request.request)
  62           requests.remove.ids.push(pool_request._id)
  63         } else {
  64           throw new Error('Unkown pool request type.')
  65         }
  66
  67         callback_each()
  68       }, function () {
  69         // Send the add requests
  70         if (requests.add.requests.length !== 0) {
  71           makePoolRequest('add', requests.add.requests, function (err) {
  72             if (err) logger.error('Errors when sent add pool requests.', { error: err })
  73
  74             removePoolRequestsFromDB(requests.add.ids)
  75           })
  76         }
  77
  78         // Send the remove requests
  79         if (requests.remove.requests.length !== 0) {
  80           makePoolRequest('remove', requests.remove.requests, function (err) {
  81             if (err) logger.error('Errors when sent remove pool requests.', { error: err })
  82
  83             removePoolRequestsFromDB(requests.remove.ids)
  84           })
  85         }
  86       })
  87     })
  88   }
  89
  90   function updatePodsScore (good_pods, bad_pods) {
  91     logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
  92
  93     PodsDB.update({ _id: { $in: good_pods } }, { $inc: { score: PODS_SCORE.BONUS } }, { multi: true }).exec()
  94     PodsDB.update({ _id: { $in: bad_pods } }, { $inc: { score: PODS_SCORE.MALUS } }, { multi: true }, function (err) {
  95       if (err) throw err
  96       removeBadPods()
  97     })
  98   }
  99
 100   function removeBadPods () {
 101     PodsDB.remove({ score: 0 }, function (err, result) {
 102       if (err) throw err
 103
 104       var number_removed = result.result.n
 105       if (number_removed !== 0) logger.info('Removed %d pod.', number_removed)
 106     })
 107   }
 108
 109   function makePoolRequest (type, requests, callback) {
 110     if (!callback) callback = function () {}
 111
 112     PodsDB.find({}, { _id: 1, url: 1, publicKey: 1 }).exec(function (err, pods) {
 113       if (err) throw err
 114
 115       var params = {
 116         encrypt: true,
 117         sign: true,
 118         method: 'POST',
 119         path: null,
 120         data: requests
 121       }
 122
 123       if (type === 'add') {
 124         params.path = '/api/' + global.API_VERSION + '/remotevideos/add'
 125       } else if (type === 'remove') {
 126         params.path = '/api/' + global.API_VERSION + '/remotevideos/remove'
 127       } else {
 128         throw new Error('Unkown pool request type.')
 129       }
 130
 131       var bad_pods = []
 132       var good_pods = []
 133
 134       utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
 135
 136       function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
 137         if (err || response.statusCode !== 200) {
 138           bad_pods.push(pod._id)
 139           logger.error('Error sending secure request to %s pod.', url, { error: err })
 140         } else {
 141           good_pods.push(pod._id)
 142         }
 143
 144         return callback_each_pod_finished()
 145       }
 146
 147       function callbackAllPodsFinished (err) {
 148         if (err) return callback(err)
 149
 150         updatePodsScore(good_pods, bad_pods)
 151         callback(null)
 152       }
 153     })
 154   }
 155
 156   // ----------- Public -----------
 157   poolRequests.activate = function () {
 158     logger.info('Pool requests activated.')
 159     timer = setInterval(makePoolRequests, INTERVAL)
 160   }
 161
 162   poolRequests.addToPoolRequests = function (id, type, request) {
 163     logger.debug('Add request to the pool requests.', { id: id, type: type, request: request })
 164
 165     PoolRequestsDB.findOne({ id: id }, function (err, entity) {
 166       if (err) logger.error(err)
 167
 168       if (entity) {
 169         if (entity.type === type) {
 170           logger.error(new Error('Cannot insert two same requests.'))
 171           return
 172         }
 173
 174         // Remove the request of the other type
 175         PoolRequestsDB.remove({ id: id }, function (err) {
 176           if (err) logger.error(err)
 177         })
 178       } else {
 179         PoolRequestsDB.create({ id: id, type: type, request: request }, function (err) {
 180           if (err) logger.error(err)
 181         })
 182       }
 183     })
 184   }
 185
 186   poolRequests.deactivate = function () {
 187     logger.info('Pool requests deactivated.')
 188     clearInterval(timer)
 189   }
 190
 191   module.exports = poolRequests
 192 })()