server/lib/friends.js

   1 'use strict'
   2
   3 const each = require('async/each')
   4 const eachLimit = require('async/eachLimit')
   5 const eachSeries = require('async/eachSeries')
   6 const series = require('async/series')
   7 const request = require('request')
   8 const waterfall = require('async/waterfall')
   9
  10 const constants = require('../initializers/constants')
  11 const db = require('../initializers/database')
  12 const logger = require('../helpers/logger')
  13 const peertubeCrypto = require('../helpers/peertube-crypto')
  14 const requests = require('../helpers/requests')
  15 const utils = require('../helpers/utils')
  16 const RequestScheduler = require('./request-scheduler')
  17 const RequestVideoQaduScheduler = require('./request-video-qadu-scheduler')
  18 const RequestVideoEventScheduler = require('./request-video-event-scheduler')
  19
  20 const ENDPOINT_ACTIONS = constants.REQUEST_ENDPOINT_ACTIONS[constants.REQUEST_ENDPOINTS.VIDEOS]
  21
  22 const requestScheduler = new RequestScheduler()
  23 const requestVideoQaduScheduler = new RequestVideoQaduScheduler()
  24 const requestVideoEventScheduler = new RequestVideoEventScheduler()
  25
  26 const friends = {
  27   activate,
  28   addVideoToFriends,
  29   updateVideoToFriends,
  30   reportAbuseVideoToFriend,
  31   quickAndDirtyUpdateVideoToFriends,
  32   quickAndDirtyUpdatesVideoToFriends,
  33   addEventToRemoteVideo,
  34   addEventsToRemoteVideo,
  35   hasFriends,
  36   makeFriends,
  37   quitFriends,
  38   removeVideoToFriends,
  39   sendOwnedVideosToPod,
  40   getRequestScheduler,
  41   getRequestVideoQaduScheduler,
  42   getRequestVideoEventScheduler
  43 }
  44
  45 function activate () {
  46   requestScheduler.activate()
  47   requestVideoQaduScheduler.activate()
  48   requestVideoEventScheduler.activate()
  49 }
  50
  51 function addVideoToFriends (videoData, transaction, callback) {
  52   const options = {
  53     type: ENDPOINT_ACTIONS.ADD,
  54     endpoint: constants.REQUEST_ENDPOINTS.VIDEOS,
  55     data: videoData,
  56     transaction
  57   }
  58   createRequest(options, callback)
  59 }
  60
  61 function updateVideoToFriends (videoData, transaction, callback) {
  62   const options = {
  63     type: ENDPOINT_ACTIONS.UPDATE,
  64     endpoint: constants.REQUEST_ENDPOINTS.VIDEOS,
  65     data: videoData,
  66     transaction
  67   }
  68   createRequest(options, callback)
  69 }
  70
  71 function removeVideoToFriends (videoParams) {
  72   const options = {
  73     type: ENDPOINT_ACTIONS.REMOVE,
  74     endpoint: constants.REQUEST_ENDPOINTS.VIDEOS,
  75     data: videoParams
  76   }
  77   createRequest(options)
  78 }
  79
  80 function reportAbuseVideoToFriend (reportData, video) {
  81   const options = {
  82     type: ENDPOINT_ACTIONS.REPORT_ABUSE,
  83     endpoint: constants.REQUEST_ENDPOINTS.VIDEOS,
  84     data: reportData,
  85     toIds: [ video.Author.podId ]
  86   }
  87   createRequest(options)
  88 }
  89
  90 function quickAndDirtyUpdateVideoToFriends (qaduParams, transaction, callback) {
  91   const options = {
  92     videoId: qaduParams.videoId,
  93     type: qaduParams.type,
  94     transaction
  95   }
  96   return createVideoQaduRequest(options, callback)
  97 }
  98
  99 function quickAndDirtyUpdatesVideoToFriends (qadusParams, transaction, finalCallback) {
 100   const tasks = []
 101
 102   qadusParams.forEach(function (qaduParams) {
 103     const fun = function (callback) {
 104       quickAndDirtyUpdateVideoToFriends(qaduParams, transaction, callback)
 105     }
 106
 107     tasks.push(fun)
 108   })
 109
 110   series(tasks, finalCallback)
 111 }
 112
 113 function addEventToRemoteVideo (eventParams, transaction, callback) {
 114   const options = {
 115     videoId: eventParams.videoId,
 116     type: eventParams.type,
 117     transaction
 118   }
 119   createVideoEventRequest(options, callback)
 120 }
 121
 122 function addEventsToRemoteVideo (eventsParams, transaction, finalCallback) {
 123   const tasks = []
 124
 125   eventsParams.forEach(function (eventParams) {
 126     const fun = function (callback) {
 127       addEventToRemoteVideo(eventParams, transaction, callback)
 128     }
 129
 130     tasks.push(fun)
 131   })
 132
 133   series(tasks, finalCallback)
 134 }
 135
 136 function hasFriends (callback) {
 137   db.Pod.countAll(function (err, count) {
 138     if (err) return callback(err)
 139
 140     const hasFriends = (count !== 0)
 141     callback(null, hasFriends)
 142   })
 143 }
 144
 145 function makeFriends (hosts, callback) {
 146   const podsScore = {}
 147
 148   logger.info('Make friends!')
 149   peertubeCrypto.getMyPublicCert(function (err, cert) {
 150     if (err) {
 151       logger.error('Cannot read public cert.')
 152       return callback(err)
 153     }
 154
 155     eachSeries(hosts, function (host, callbackEach) {
 156       computeForeignPodsList(host, podsScore, callbackEach)
 157     }, function (err) {
 158       if (err) return callback(err)
 159
 160       logger.debug('Pods scores computed.', { podsScore: podsScore })
 161       const podsList = computeWinningPods(hosts, podsScore)
 162       logger.debug('Pods that we keep.', { podsToKeep: podsList })
 163
 164       makeRequestsToWinningPods(cert, podsList, callback)
 165     })
 166   })
 167 }
 168
 169 function quitFriends (callback) {
 170   // Stop pool requests
 171   requestScheduler.deactivate()
 172
 173   waterfall([
 174     function flushRequests (callbackAsync) {
 175       requestScheduler.flush(err => callbackAsync(err))
 176     },
 177
 178     function flushVideoQaduRequests (callbackAsync) {
 179       requestVideoQaduScheduler.flush(err => callbackAsync(err))
 180     },
 181
 182     function getPodsList (callbackAsync) {
 183       return db.Pod.list(callbackAsync)
 184     },
 185
 186     function announceIQuitMyFriends (pods, callbackAsync) {
 187       const requestParams = {
 188         method: 'POST',
 189         path: '/api/' + constants.API_VERSION + '/pods/remove',
 190         sign: true
 191       }
 192
 193       // Announce we quit them
 194       // We don't care if the request fails
 195       // The other pod will exclude us automatically after a while
 196       eachLimit(pods, constants.REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
 197         requestParams.toPod = pod
 198         requests.makeSecureRequest(requestParams, callbackEach)
 199       }, function (err) {
 200         if (err) {
 201           logger.error('Some errors while quitting friends.', { err: err })
 202           // Don't stop the process
 203         }
 204
 205         return callbackAsync(null, pods)
 206       })
 207     },
 208
 209     function removePodsFromDB (pods, callbackAsync) {
 210       each(pods, function (pod, callbackEach) {
 211         pod.destroy().asCallback(callbackEach)
 212       }, callbackAsync)
 213     }
 214   ], function (err) {
 215     // Don't forget to re activate the scheduler, even if there was an error
 216     requestScheduler.activate()
 217
 218     if (err) return callback(err)
 219
 220     logger.info('Removed all remote videos.')
 221     return callback(null)
 222   })
 223 }
 224
 225 function sendOwnedVideosToPod (podId) {
 226   db.Video.listOwnedAndPopulateAuthorAndTags(function (err, videosList) {
 227     if (err) {
 228       logger.error('Cannot get the list of videos we own.')
 229       return
 230     }
 231
 232     videosList.forEach(function (video) {
 233       video.toAddRemoteJSON(function (err, remoteVideo) {
 234         if (err) {
 235           logger.error('Cannot convert video to remote.', { error: err })
 236           // Don't break the process
 237           return
 238         }
 239
 240         const options = {
 241           type: 'add',
 242           endpoint: constants.REQUEST_ENDPOINTS.VIDEOS,
 243           data: remoteVideo,
 244           toIds: [ podId ]
 245         }
 246         createRequest(options)
 247       })
 248     })
 249   })
 250 }
 251
 252 function getRequestScheduler () {
 253   return requestScheduler
 254 }
 255
 256 function getRequestVideoQaduScheduler () {
 257   return requestVideoQaduScheduler
 258 }
 259
 260 function getRequestVideoEventScheduler () {
 261   return requestVideoEventScheduler
 262 }
 263
 264 // ---------------------------------------------------------------------------
 265
 266 module.exports = friends
 267
 268 // ---------------------------------------------------------------------------
 269
 270 function computeForeignPodsList (host, podsScore, callback) {
 271   getForeignPodsList(host, function (err, res) {
 272     if (err) return callback(err)
 273
 274     const foreignPodsList = res.data
 275
 276     // Let's give 1 point to the pod we ask the friends list
 277     foreignPodsList.push({ host })
 278
 279     foreignPodsList.forEach(function (foreignPod) {
 280       const foreignPodHost = foreignPod.host
 281
 282       if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 283       else podsScore[foreignPodHost] = 1
 284     })
 285
 286     return callback()
 287   })
 288 }
 289
 290 function computeWinningPods (hosts, podsScore) {
 291   // Build the list of pods to add
 292   // Only add a pod if it exists in more than a half base pods
 293   const podsList = []
 294   const baseScore = hosts.length / 2
 295
 296   Object.keys(podsScore).forEach(function (podHost) {
 297     // If the pod is not me and with a good score we add it
 298     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 299       podsList.push({ host: podHost })
 300     }
 301   })
 302
 303   return podsList
 304 }
 305
 306 function getForeignPodsList (host, callback) {
 307   const path = '/api/' + constants.API_VERSION + '/pods'
 308
 309   request.get(constants.REMOTE_SCHEME.HTTP + '://' + host + path, function (err, response, body) {
 310     if (err) return callback(err)
 311
 312     try {
 313       const json = JSON.parse(body)
 314       return callback(null, json)
 315     } catch (err) {
 316       return callback(err)
 317     }
 318   })
 319 }
 320
 321 function makeRequestsToWinningPods (cert, podsList, callback) {
 322   // Stop pool requests
 323   requestScheduler.deactivate()
 324   // Flush pool requests
 325   requestScheduler.forceSend()
 326
 327   eachLimit(podsList, constants.REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
 328     const params = {
 329       url: constants.REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + constants.API_VERSION + '/pods/',
 330       method: 'POST',
 331       json: {
 332         host: constants.CONFIG.WEBSERVER.HOST,
 333         email: constants.CONFIG.ADMIN.EMAIL,
 334         publicKey: cert
 335       }
 336     }
 337
 338     requests.makeRetryRequest(params, function (err, res, body) {
 339       if (err) {
 340         logger.error('Error with adding %s pod.', pod.host, { error: err })
 341         // Don't break the process
 342         return callbackEach()
 343       }
 344
 345       if (res.statusCode === 200) {
 346         const podObj = db.Pod.build({ host: pod.host, publicKey: body.cert, email: body.email })
 347         podObj.save().asCallback(function (err, podCreated) {
 348           if (err) {
 349             logger.error('Cannot add friend %s pod.', pod.host, { error: err })
 350             return callbackEach()
 351           }
 352
 353           // Add our videos to the request scheduler
 354           sendOwnedVideosToPod(podCreated.id)
 355
 356           return callbackEach()
 357         })
 358       } else {
 359         logger.error('Status not 200 for %s pod.', pod.host)
 360         return callbackEach()
 361       }
 362     })
 363   }, function endRequests () {
 364     // Final callback, we've ended all the requests
 365     // Now we made new friends, we can re activate the pool of requests
 366     requestScheduler.activate()
 367
 368     logger.debug('makeRequestsToWinningPods finished.')
 369     return callback()
 370   })
 371 }
 372
 373 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 374 // { type, endpoint, data, toIds, transaction }
 375 function createRequest (options, callback) {
 376   if (!callback) callback = function () {}
 377   if (options.toIds) return requestScheduler.createRequest(options, callback)
 378
 379   // If the "toIds" pods is not specified, we send the request to all our friends
 380   db.Pod.listAllIds(options.transaction, function (err, podIds) {
 381     if (err) {
 382       logger.error('Cannot get pod ids', { error: err })
 383       return
 384     }
 385
 386     const newOptions = Object.assign(options, { toIds: podIds })
 387     return requestScheduler.createRequest(newOptions, callback)
 388   })
 389 }
 390
 391 function createVideoQaduRequest (options, callback) {
 392   if (!callback) callback = utils.createEmptyCallback()
 393
 394   requestVideoQaduScheduler.createRequest(options, callback)
 395 }
 396
 397 function createVideoEventRequest (options, callback) {
 398   if (!callback) callback = utils.createEmptyCallback()
 399
 400   requestVideoEventScheduler.createRequest(options, callback)
 401 }
 402
 403 function isMe (host) {
 404   return host === constants.CONFIG.WEBSERVER.HOST
 405 }