lib/poolRequests.js

   1 ;(function () {
   2   'use strict'
   3
   4   var async = require('async')
   5   var pluck = require('lodash-node/compat/collection/pluck')
   6
   7   var constants = require('../initializers/constants')
   8   var logger = require('../helpers/logger')
   9   var Pods = require('../models/pods')
  10   var PoolRequests = require('../models/poolRequests')
  11   var utils = require('../helpers/utils')
  12   var Videos = require('../models/videos')
  13
  14   var timer = null
  15
  16   var poolRequests = {
  17     activate: activate,
  18     deactivate: deactivate,
  19     forceSend: forceSend
  20   }
  21
  22   function deactivate () {
  23     logger.info('Pool requests deactivated.')
  24     clearInterval(timer)
  25   }
  26
  27   function forceSend () {
  28     logger.info('Force pool requests sending.')
  29     makePoolRequests()
  30   }
  31
  32   function activate () {
  33     logger.info('Pool requests activated.')
  34     timer = setInterval(makePoolRequests, constants.INTERVAL)
  35   }
  36
  37   // ---------------------------------------------------------------------------
  38
  39   module.exports = poolRequests
  40
  41   // ---------------------------------------------------------------------------
  42
  43   function makePoolRequest (type, requests, callback) {
  44     if (!callback) callback = function () {}
  45
  46     Pods.list(function (err, pods) {
  47       if (err) throw err
  48
  49       var params = {
  50         encrypt: true,
  51         sign: true,
  52         method: 'POST',
  53         path: null,
  54         data: requests
  55       }
  56
  57       if (type === 'add') {
  58         params.path = '/api/' + constants.API_VERSION + '/remotevideos/add'
  59       } else if (type === 'remove') {
  60         params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove'
  61       } else {
  62         throw new Error('Unkown pool request type.')
  63       }
  64
  65       var bad_pods = []
  66       var good_pods = []
  67
  68       utils.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
  69
  70       function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
  71         if (err || (response.statusCode !== 200 && response.statusCode !== 204)) {
  72           bad_pods.push(pod._id)
  73           logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') })
  74         } else {
  75           good_pods.push(pod._id)
  76         }
  77
  78         return callback_each_pod_finished()
  79       }
  80
  81       function callbackAllPodsFinished (err) {
  82         if (err) return callback(err)
  83
  84         updatePodsScore(good_pods, bad_pods)
  85         callback(null)
  86       }
  87     })
  88   }
  89
  90   function makePoolRequests () {
  91     logger.info('Making pool requests to friends.')
  92
  93     PoolRequests.list(function (err, pool_requests) {
  94       if (err) throw err
  95
  96       if (pool_requests.length === 0) return
  97
  98       var requests = {
  99         add: {
 100           ids: [],
 101           requests: []
 102         },
 103         remove: {
 104           ids: [],
 105           requests: []
 106         }
 107       }
 108
 109       async.each(pool_requests, function (pool_request, callback_each) {
 110         if (pool_request.type === 'add') {
 111           requests.add.requests.push(pool_request.request)
 112           requests.add.ids.push(pool_request._id)
 113         } else if (pool_request.type === 'remove') {
 114           requests.remove.requests.push(pool_request.request)
 115           requests.remove.ids.push(pool_request._id)
 116         } else {
 117           throw new Error('Unkown pool request type.')
 118         }
 119
 120         callback_each()
 121       }, function () {
 122         // Send the add requests
 123         if (requests.add.requests.length !== 0) {
 124           makePoolRequest('add', requests.add.requests, function (err) {
 125             if (err) logger.error('Errors when sent add pool requests.', { error: err })
 126
 127             PoolRequests.removeRequests(requests.add.ids)
 128           })
 129         }
 130
 131         // Send the remove requests
 132         if (requests.remove.requests.length !== 0) {
 133           makePoolRequest('remove', requests.remove.requests, function (err) {
 134             if (err) logger.error('Errors when sent remove pool requests.', { error: err })
 135
 136             PoolRequests.removeRequests(requests.remove.ids)
 137           })
 138         }
 139       })
 140     })
 141   }
 142
 143   function removeBadPods () {
 144     Pods.findBadPods(function (err, pods) {
 145       if (err) throw err
 146
 147       if (pods.length === 0) return
 148
 149       var urls = pluck(pods, 'url')
 150       var ids = pluck(pods, '_id')
 151
 152       Videos.removeAllRemotesOf(urls, function (err, r) {
 153         if (err) logger.error('Cannot remove videos from a pod that we removing.', { error: err })
 154         var videos_removed = r.result.n
 155         logger.info('Removed %d videos.', videos_removed)
 156
 157         Pods.removeAllByIds(ids, function (err, r) {
 158           if (err) logger.error('Cannot remove bad pods.', { error: err })
 159
 160           var pods_removed = r.result.n
 161           logger.info('Removed %d pods.', pods_removed)
 162         })
 163       })
 164     })
 165   }
 166
 167   function updatePodsScore (good_pods, bad_pods) {
 168     logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
 169
 170     Pods.incrementScores(good_pods, constants.PODS_SCORE.BONUS)
 171     Pods.incrementScores(bad_pods, constants.PODS_SCORE.MALUS, function (err) {
 172       if (err) throw err
 173       removeBadPods()
 174     })
 175   }
 176 })()