server/models/request.js

   1 'use strict'
   2
   3 const each = require('async/each')
   4 const eachLimit = require('async/eachLimit')
   5 const map = require('lodash/map')
   6 const mongoose = require('mongoose')
   7 const waterfall = require('async/waterfall')
   8
   9 const constants = require('../initializers/constants')
  10 const logger = require('../helpers/logger')
  11 const requests = require('../helpers/requests')
  12
  13 const Pod = mongoose.model('Pod')
  14 const Video = mongoose.model('Video')
  15
  16 let timer = null
  17
  18 // ---------------------------------------------------------------------------
  19
  20 const RequestSchema = mongoose.Schema({
  21   request: mongoose.Schema.Types.Mixed,
  22   to: [ { type: mongoose.Schema.Types.ObjectId, ref: 'Pod' } ]
  23 })
  24
  25 RequestSchema.statics = {
  26   activate,
  27   deactivate,
  28   flush,
  29   forceSend,
  30   list
  31 }
  32
  33 RequestSchema.pre('save', function (next) {
  34   const self = this
  35
  36   if (self.to.length === 0) {
  37     Pod.listAllIds(function (err, podIds) {
  38       if (err) return next(err)
  39
  40       // No friends
  41       if (podIds.length === 0) return
  42
  43       self.to = podIds
  44       return next()
  45     })
  46   } else {
  47     return next()
  48   }
  49 })
  50
  51 mongoose.model('Request', RequestSchema)
  52
  53 // ------------------------------ STATICS ------------------------------
  54
  55 function activate () {
  56   logger.info('Requests scheduler activated.')
  57   timer = setInterval(makeRequests.bind(this), constants.REQUESTS_INTERVAL)
  58 }
  59
  60 function deactivate () {
  61   logger.info('Requests scheduler deactivated.')
  62   clearInterval(timer)
  63 }
  64
  65 function flush () {
  66   removeAll.call(this, function (err) {
  67     if (err) logger.error('Cannot flush the requests.', { error: err })
  68   })
  69 }
  70
  71 function forceSend () {
  72   logger.info('Force requests scheduler sending.')
  73   makeRequests.call(this)
  74 }
  75
  76 function list (callback) {
  77   this.find({ }, callback)
  78 }
  79
  80 // ---------------------------------------------------------------------------
  81
  82 // Make a requests to friends of a certain type
  83 function makeRequest (toPod, requestsToMake, callback) {
  84   if (!callback) callback = function () {}
  85
  86   const params = {
  87     toPod: toPod,
  88     encrypt: true, // Security
  89     sign: true, // To prove our identity
  90     method: 'POST',
  91     path: '/api/' + constants.API_VERSION + '/remote/videos',
  92     data: requestsToMake // Requests we need to make
  93   }
  94
  95   // Make multiple retry requests to all of pods
  96   // The function fire some useful callbacks
  97   requests.makeSecureRequest(params, function (err, res) {
  98     if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
  99       logger.error('Error sending secure request to %s pod.', toPod.url, { error: err || new Error('Status code not 20x') })
 100
 101       return callback(false)
 102     }
 103
 104     return callback(true)
 105   })
 106 }
 107
 108 // Make all the requests of the scheduler
 109 function makeRequests () {
 110   const self = this
 111
 112   listWithLimit.call(self, constants.REQUESTS_LIMIT, function (err, requests) {
 113     if (err) {
 114       logger.error('Cannot get the list of requests.', { err: err })
 115       return // Abort
 116     }
 117
 118     // If there are no requests, abort
 119     if (requests.length === 0) {
 120       logger.info('No requests to make.')
 121       return
 122     }
 123
 124     logger.info('Making requests to friends.')
 125
 126     // Requests by pods id
 127     const requestsToMake = {}
 128
 129     requests.forEach(function (poolRequest) {
 130       poolRequest.to.forEach(function (toPodId) {
 131         if (!requestsToMake[toPodId]) {
 132           requestsToMake[toPodId] = {
 133             ids: [],
 134             datas: []
 135           }
 136         }
 137
 138         requestsToMake[toPodId].ids.push(poolRequest._id)
 139         requestsToMake[toPodId].datas.push(poolRequest.request)
 140       })
 141     })
 142
 143     const goodPods = []
 144     const badPods = []
 145
 146     eachLimit(Object.keys(requestsToMake), constants.REQUESTS_IN_PARALLEL, function (toPodId, callbackEach) {
 147       const requestToMake = requestsToMake[toPodId]
 148
 149       // FIXME: mongodb request inside a loop :/
 150       Pod.load(toPodId, function (err, toPod) {
 151         if (err) {
 152           logger.error('Error finding pod by id.', { err: err })
 153           return callbackEach()
 154         }
 155
 156         // Maybe the pod is not our friend anymore so simply remove them
 157         if (!toPod) {
 158           removePodOf.call(self, requestToMake.ids, toPodId)
 159           return callbackEach()
 160         }
 161
 162         makeRequest(toPod, requestToMake.datas, function (success) {
 163           if (err) {
 164             logger.error('Errors when sent request to %s.', toPod.url, { error: err })
 165             // Do not stop the process just for one error
 166             return callbackEach()
 167           }
 168
 169           if (success === true) {
 170             logger.debug('Removing requests for %s pod.', toPodId, { requestsIds: requestToMake.ids })
 171
 172             // Remove the pod id of these request ids
 173             removePodOf.call(self, requestToMake.ids, toPodId)
 174             goodPods.push(toPodId)
 175           } else {
 176             badPods.push(toPodId)
 177           }
 178
 179           callbackEach()
 180         })
 181       })
 182     }, function () {
 183       // All the requests were made, we update the pods score
 184       updatePodsScore(goodPods, badPods)
 185       // Flush requests with no pod
 186       removeWithEmptyTo.call(self)
 187     })
 188   })
 189 }
 190
 191 // Remove pods with a score of 0 (too many requests where they were unreachable)
 192 function removeBadPods () {
 193   waterfall([
 194     function findBadPods (callback) {
 195       Pod.listBadPods(function (err, pods) {
 196         if (err) {
 197           logger.error('Cannot find bad pods.', { error: err })
 198           return callback(err)
 199         }
 200
 201         return callback(null, pods)
 202       })
 203     },
 204
 205     function listVideosOfTheseBadPods (pods, callback) {
 206       if (pods.length === 0) return callback(null)
 207
 208       const urls = map(pods, 'url')
 209
 210       Video.listByUrls(urls, function (err, videosList) {
 211         if (err) {
 212           logger.error('Cannot list videos urls.', { error: err, urls: urls })
 213           return callback(null, pods, [])
 214         }
 215
 216         return callback(null, pods, videosList)
 217       })
 218     },
 219
 220     function removeVideosOfTheseBadPods (pods, videosList, callback) {
 221       // We don't have to remove pods, skip
 222       if (typeof pods === 'function') {
 223         callback = pods
 224         return callback(null)
 225       }
 226
 227       each(videosList, function (video, callbackEach) {
 228         video.remove(callbackEach)
 229       }, function (err) {
 230         if (err) {
 231           // Don't stop the process
 232           logger.error('Error while removing videos of bad pods.', { error: err })
 233           return
 234         }
 235
 236         return callback(null, pods)
 237       })
 238     },
 239
 240     function removeBadPodsFromDB (pods, callback) {
 241       // We don't have to remove pods, skip
 242       if (typeof pods === 'function') {
 243         callback = pods
 244         return callback(null)
 245       }
 246
 247       each(pods, function (pod, callbackEach) {
 248         pod.remove(callbackEach)
 249       }, function (err) {
 250         if (err) return callback(err)
 251
 252         return callback(null, pods.length)
 253       })
 254     }
 255   ], function (err, numberOfPodsRemoved) {
 256     if (err) {
 257       logger.error('Cannot remove bad pods.', { error: err })
 258     } else if (numberOfPodsRemoved) {
 259       logger.info('Removed %d pods.', numberOfPodsRemoved)
 260     } else {
 261       logger.info('No need to remove bad pods.')
 262     }
 263   })
 264 }
 265
 266 function updatePodsScore (goodPods, badPods) {
 267   logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
 268
 269   Pod.incrementScores(goodPods, constants.PODS_SCORE.BONUS, function (err) {
 270     if (err) logger.error('Cannot increment scores of good pods.')
 271   })
 272
 273   Pod.incrementScores(badPods, constants.PODS_SCORE.MALUS, function (err) {
 274     if (err) logger.error('Cannot decrement scores of bad pods.')
 275     removeBadPods()
 276   })
 277 }
 278
 279 function listWithLimit (limit, callback) {
 280   this.find({ }, { _id: 1, request: 1, to: 1 }).sort({ _id: 1 }).limit(limit).exec(callback)
 281 }
 282
 283 function removeAll (callback) {
 284   this.remove({ }, callback)
 285 }
 286
 287 function removePodOf (requestsIds, podId, callback) {
 288   if (!callback) callback = function () {}
 289
 290   this.update({ _id: { $in: requestsIds } }, { $pull: { to: podId } }, { multi: true }, callback)
 291 }
 292
 293 function removeWithEmptyTo (callback) {
 294   if (!callback) callback = function () {}
 295
 296   this.remove({ to: { $size: 0 } }, callback)
 297 }