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