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