3 const each = require('async/each')
4 const eachLimit = require('async/eachLimit')
5 const waterfall = require('async/waterfall')
6 const values = require('lodash/values')
8 const constants = require('../initializers/constants')
9 const logger = require('../helpers/logger')
10 const requests = require('../helpers/requests')
13 let lastRequestTimestamp = 0
15 // ---------------------------------------------------------------------------
17 module.exports = function (sequelize, DataTypes) {
18 const Request = sequelize.define('Request',
25 type: DataTypes.ENUM(values(constants.REQUEST_ENDPOINTS)),
46 // ------------------------------ STATICS ------------------------------
48 function associate (models) {
49 this.belongsToMany(models.Pod, {
54 through: models.RequestToPod,
59 function activate () {
60 logger.info('Requests scheduler activated.')
61 lastRequestTimestamp = Date.now()
64 timer = setInterval(function () {
65 lastRequestTimestamp = Date.now()
66 makeRequests.call(self)
67 }, constants.REQUESTS_INTERVAL)
70 function countTotalRequests (callback) {
72 include: [ this.sequelize.models.Pod ]
75 return this.count(query).asCallback(callback)
78 function deactivate () {
79 logger.info('Requests scheduler deactivated.')
84 function flush (callback) {
85 removeAll.call(this, function (err) {
86 if (err) logger.error('Cannot flush the requests.', { error: err })
92 function forceSend () {
93 logger.info('Force requests scheduler sending.')
94 makeRequests.call(this)
97 function remainingMilliSeconds () {
98 if (timer === null) return -1
100 return constants.REQUESTS_INTERVAL - (Date.now() - lastRequestTimestamp)
103 // ---------------------------------------------------------------------------
105 // Make a requests to friends of a certain type
106 function makeRequest (toPod, requestEndpoint, requestsToMake, callback) {
107 if (!callback) callback = function () {}
111 sign: true, // Prove our identity
113 path: '/api/' + constants.API_VERSION + '/remote/' + requestEndpoint,
114 data: requestsToMake // Requests we need to make
117 // Make multiple retry requests to all of pods
118 // The function fire some useful callbacks
119 requests.makeSecureRequest(params, function (err, res) {
120 if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
122 'Error sending secure request to %s pod.',
125 error: err || new Error('Status code not 20x : ' + res.statusCode)
129 return callback(false)
132 return callback(true)
136 // Make all the requests of the scheduler
137 function makeRequests () {
139 const RequestToPod = this.sequelize.models.RequestToPod
141 // We limit the size of the requests (REQUESTS_LIMIT)
142 // We don't want to stuck with the same failing requests so we get a random list
143 listWithLimitAndRandom.call(self, constants.REQUESTS_LIMIT, function (err, requests) {
145 logger.error('Cannot get the list of requests.', { err: err })
149 // If there are no requests, abort
150 if (requests.length === 0) {
151 logger.info('No requests to make.')
155 logger.info('Making requests to friends.')
157 // We want to group requests by destinations pod and endpoint
158 const requestsToMakeGrouped = {}
160 requests.forEach(function (request) {
161 request.Pods.forEach(function (toPod) {
162 const hashKey = toPod.id + request.endpoint
163 if (!requestsToMakeGrouped[hashKey]) {
164 requestsToMakeGrouped[hashKey] = {
166 endpoint: request.endpoint,
167 ids: [], // request ids, to delete them from the DB in the future
168 datas: [] // requests data,
172 requestsToMakeGrouped[hashKey].ids.push(request.id)
173 requestsToMakeGrouped[hashKey].datas.push(request.request)
180 eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, function (hashKey, callbackEach) {
181 const requestToMake = requestsToMakeGrouped[hashKey]
183 // FIXME: SQL request inside a loop :/
184 self.sequelize.models.Pod.load(requestToMake.toPodId, function (err, toPod) {
186 logger.error('Error finding pod by id.', { err: err })
187 return callbackEach()
190 // Maybe the pod is not our friend anymore so simply remove it
192 const requestIdsToDelete = requestToMake.ids
194 logger.info('Removing %d requests of unexisting pod %s.', requestIdsToDelete.length, requestToMake.toPodId)
195 RequestToPod.removePodOf.call(self, requestIdsToDelete, requestToMake.toPodId)
196 return callbackEach()
199 makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, function (success) {
200 if (success === true) {
201 logger.debug('Removing requests for pod %s.', requestToMake.toPodId, { requestsIds: requestToMake.ids })
203 goodPods.push(requestToMake.toPodId)
205 // Remove the pod id of these request ids
206 RequestToPod.removePodOf(requestToMake.ids, requestToMake.toPodId, callbackEach)
208 badPods.push(requestToMake.toPodId)
214 // All the requests were made, we update the pods score
215 updatePodsScore.call(self, goodPods, badPods)
216 // Flush requests with no pod
217 removeWithEmptyTo.call(self, function (err) {
218 if (err) logger.error('Error when removing requests with no pods.', { error: err })
224 // Remove pods with a score of 0 (too many requests where they were unreachable)
225 function removeBadPods () {
229 function findBadPods (callback) {
230 self.sequelize.models.Pod.listBadPods(function (err, pods) {
232 logger.error('Cannot find bad pods.', { error: err })
236 return callback(null, pods)
240 function removeTheseBadPods (pods, callback) {
241 each(pods, function (pod, callbackEach) {
242 pod.destroy().asCallback(callbackEach)
244 return callback(err, pods.length)
247 ], function (err, numberOfPodsRemoved) {
249 logger.error('Cannot remove bad pods.', { error: err })
250 } else if (numberOfPodsRemoved) {
251 logger.info('Removed %d pods.', numberOfPodsRemoved)
253 logger.info('No need to remove bad pods.')
258 function updatePodsScore (goodPods, badPods) {
260 const Pod = this.sequelize.models.Pod
262 logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
264 if (goodPods.length !== 0) {
265 Pod.incrementScores(goodPods, constants.PODS_SCORE.BONUS, function (err) {
266 if (err) logger.error('Cannot increment scores of good pods.', { error: err })
270 if (badPods.length !== 0) {
271 Pod.incrementScores(badPods, constants.PODS_SCORE.MALUS, function (err) {
272 if (err) logger.error('Cannot decrement scores of bad pods.', { error: err })
273 removeBadPods.call(self)
278 function listWithLimitAndRandom (limit, callback) {
281 self.count().asCallback(function (err, count) {
282 if (err) return callback(err)
285 if (count === 0) return callback(null, [])
287 let start = Math.floor(Math.random() * count) - limit
288 if (start < 0) start = 0
296 include: [ this.sequelize.models.Pod ]
299 self.findAll(query).asCallback(callback)
303 function removeAll (callback) {
304 // Delete all requests
305 this.truncate({ cascade: true }).asCallback(callback)
308 function removeWithEmptyTo (callback) {
309 if (!callback) callback = function () {}
315 this.sequelize.literal('SELECT "requestId" FROM "RequestToPods"')
321 this.destroy(query).asCallback(callback)