3 const async = require('async')
4 const pluck = require('lodash-node/compat/collection/pluck')
6 const constants = require('../initializers/constants')
7 const logger = require('../helpers/logger')
8 const Pods = require('../models/pods')
9 const PoolRequests = require('../models/poolRequests')
10 const requests = require('../helpers/requests')
11 const Videos = require('../models/videos')
15 const poolRequests = {
17 addRequest: addRequest,
18 deactivate: deactivate,
22 function activate () {
23 logger.info('Pool requests activated.')
24 timer = setInterval(makePoolRequests, constants.INTERVAL)
27 function addRequest (id, type, request) {
28 logger.debug('Add request to the pool requests.', { id: id, type: type, request: request })
30 PoolRequests.findById(id, function (err, entity) {
32 logger.error('Cannot find one pool request.', { error: err })
37 if (entity.type === type) {
38 logger.error('Cannot insert two same requests.')
42 // Remove the request of the other type
43 PoolRequests.removeRequestById(id, function (err) {
45 logger.error('Cannot remove a pool request.', { error: err })
50 PoolRequests.create(id, type, request, function (err) {
51 if (err) logger.error('Cannot create a pool request.', { error: err })
58 function deactivate () {
59 logger.info('Pool requests deactivated.')
63 function forceSend () {
64 logger.info('Force pool requests sending.')
68 // ---------------------------------------------------------------------------
70 module.exports = poolRequests
72 // ---------------------------------------------------------------------------
74 function makePoolRequest (type, requests_to_make, callback) {
75 if (!callback) callback = function () {}
77 Pods.list(function (err, pods) {
78 if (err) return callback(err)
85 data: requests_to_make
89 params.path = '/api/' + constants.API_VERSION + '/remotevideos/add'
90 } else if (type === 'remove') {
91 params.path = '/api/' + constants.API_VERSION + '/remotevideos/remove'
93 return callback(new Error('Unkown pool request type.'))
99 requests.makeMultipleRetryRequest(params, pods, callbackEachPodFinished, callbackAllPodsFinished)
101 function callbackEachPodFinished (err, response, body, url, pod, callback_each_pod_finished) {
102 if (err || (response.statusCode !== 200 && response.statusCode !== 204)) {
103 bad_pods.push(pod._id)
104 logger.error('Error sending secure request to %s pod.', url, { error: err || new Error('Status code not 20x') })
106 good_pods.push(pod._id)
109 return callback_each_pod_finished()
112 function callbackAllPodsFinished (err) {
113 if (err) return callback(err)
115 updatePodsScore(good_pods, bad_pods)
121 function makePoolRequests () {
122 logger.info('Making pool requests to friends.')
124 PoolRequests.list(function (err, pool_requests) {
126 logger.error('Cannot get the list of pool requests.', { err: err })
130 if (pool_requests.length === 0) return
132 const requests_to_make = {
143 async.each(pool_requests, function (pool_request, callback_each) {
144 if (pool_request.type === 'add') {
145 requests_to_make.add.requests.push(pool_request.request)
146 requests_to_make.add.ids.push(pool_request._id)
147 } else if (pool_request.type === 'remove') {
148 requests_to_make.remove.requests.push(pool_request.request)
149 requests_to_make.remove.ids.push(pool_request._id)
151 logger.error('Unkown pool request type.', { request_type: pool_request.type })
157 // Send the add requests
158 if (requests_to_make.add.requests.length !== 0) {
159 makePoolRequest('add', requests_to_make.add.requests, function (err) {
160 if (err) logger.error('Errors when sent add pool requests.', { error: err })
162 PoolRequests.removeRequests(requests_to_make.add.ids)
166 // Send the remove requests
167 if (requests_to_make.remove.requests.length !== 0) {
168 makePoolRequest('remove', requests_to_make.remove.requests, function (err) {
169 if (err) logger.error('Errors when sent remove pool requests.', { error: err })
171 PoolRequests.removeRequests(requests_to_make.remove.ids)
178 function removeBadPods () {
179 Pods.findBadPods(function (err, pods) {
181 logger.error('Cannot find bad pods.', { error: err })
185 if (pods.length === 0) return
187 const urls = pluck(pods, 'url')
188 const ids = pluck(pods, '_id')
190 Videos.removeAllRemotesOf(urls, function (err, r) {
192 logger.error('Cannot remove videos from a pod that we removing.', { error: err })
194 const videos_removed = r.result.n
195 logger.info('Removed %d videos.', videos_removed)
198 Pods.removeAllByIds(ids, function (err, r) {
200 logger.error('Cannot remove bad pods.', { error: err })
202 const pods_removed = r.result.n
203 logger.info('Removed %d pods.', pods_removed)
210 function updatePodsScore (good_pods, bad_pods) {
211 logger.info('Updating %d good pods and %d bad pods scores.', good_pods.length, bad_pods.length)
213 Pods.incrementScores(good_pods, constants.PODS_SCORE.BONUS, function (err) {
214 if (err) logger.error('Cannot increment scores of good pods.')
217 Pods.incrementScores(bad_pods, constants.PODS_SCORE.MALUS, function (err) {
218 if (err) logger.error('Cannot increment scores of bad pods.')