server/lib/friends.js

   1 'use strict'
   2
   3 const config = require('config')
   4 const each = require('async/each')
   5 const eachLimit = require('async/eachLimit')
   6 const eachSeries = require('async/eachSeries')
   7 const fs = require('fs')
   8 const mongoose = require('mongoose')
   9 const request = require('request')
  10 const waterfall = require('async/waterfall')
  11
  12 const constants = require('../initializers/constants')
  13 const logger = require('../helpers/logger')
  14 const requests = require('../helpers/requests')
  15
  16 const Pod = mongoose.model('Pod')
  17 const Request = mongoose.model('Request')
  18 const Video = mongoose.model('Video')
  19
  20 const friends = {
  21   addVideoToFriends: addVideoToFriends,
  22   hasFriends: hasFriends,
  23   getMyCertificate: getMyCertificate,
  24   makeFriends: makeFriends,
  25   quitFriends: quitFriends,
  26   removeVideoToFriends: removeVideoToFriends,
  27   sendOwnedVideosToPod: sendOwnedVideosToPod
  28 }
  29
  30 function addVideoToFriends (video) {
  31   createRequest('add', video)
  32 }
  33
  34 function hasFriends (callback) {
  35   Pod.countAll(function (err, count) {
  36     if (err) return callback(err)
  37
  38     const hasFriends = (count !== 0)
  39     callback(null, hasFriends)
  40   })
  41 }
  42
  43 function getMyCertificate (callback) {
  44   fs.readFile(constants.CONFIG.STORAGE.CERT_DIR + 'peertube.pub', 'utf8', callback)
  45 }
  46
  47 function makeFriends (callback) {
  48   const podsScore = {}
  49
  50   logger.info('Make friends!')
  51   getMyCertificate(function (err, cert) {
  52     if (err) {
  53       logger.error('Cannot read public cert.')
  54       return callback(err)
  55     }
  56
  57     const urls = config.get('network.friends')
  58
  59     eachSeries(urls, function (url, callbackEach) {
  60       computeForeignPodsList(url, podsScore, callbackEach)
  61     }, function (err) {
  62       if (err) return callback(err)
  63
  64       logger.debug('Pods scores computed.', { podsScore: podsScore })
  65       const podsList = computeWinningPods(urls, podsScore)
  66       logger.debug('Pods that we keep.', { podsToKeep: podsList })
  67
  68       makeRequestsToWinningPods(cert, podsList, callback)
  69     })
  70   })
  71 }
  72
  73 function quitFriends (callback) {
  74   // Stop pool requests
  75   Request.deactivate()
  76   // Flush pool requests
  77   Request.flush()
  78
  79   waterfall([
  80     function getPodsList (callbackAsync) {
  81       return Pod.list(callbackAsync)
  82     },
  83
  84     function announceIQuitMyFriends (pods, callbackAsync) {
  85       const requestParams = {
  86         method: 'POST',
  87         path: '/api/' + constants.API_VERSION + '/pods/remove',
  88         sign: true
  89       }
  90
  91       // Announce we quit them
  92       // We don't care if the request fails
  93       // The other pod will exclude us automatically after a while
  94       eachLimit(pods, constants.REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
  95         requestParams.toPod = pod
  96         requests.makeSecureRequest(requestParams, callbackEach)
  97       }, function (err) {
  98         if (err) {
  99           logger.error('Some errors while quitting friends.', { err: err })
 100           // Don't stop the process
 101         }
 102
 103         return callbackAsync()
 104       })
 105     },
 106
 107     function removePodsFromDB (callbackAsync) {
 108       Pod.removeAll(function (err) {
 109         return callbackAsync(err)
 110       })
 111     },
 112
 113     function listRemoteVideos (callbackAsync) {
 114       logger.info('Broke friends, so sad :(')
 115
 116       Video.listRemotes(callbackAsync)
 117     },
 118
 119     function removeTheRemoteVideos (videosList, callbackAsync) {
 120       each(videosList, function (video, callbackEach) {
 121         video.remove(callbackEach)
 122       }, callbackAsync)
 123     }
 124   ], function (err) {
 125     // Don't forget to re activate the scheduler, even if there was an error
 126     Request.activate()
 127
 128     if (err) return callback(err)
 129
 130     logger.info('Removed all remote videos.')
 131     return callback(null)
 132   })
 133 }
 134
 135 function removeVideoToFriends (videoParams) {
 136   createRequest('remove', videoParams)
 137 }
 138
 139 function sendOwnedVideosToPod (podId) {
 140   Video.listOwned(function (err, videosList) {
 141     if (err) {
 142       logger.error('Cannot get the list of videos we own.')
 143       return
 144     }
 145
 146     videosList.forEach(function (video) {
 147       video.toRemoteJSON(function (err, remoteVideo) {
 148         if (err) {
 149           logger.error('Cannot convert video to remote.', { error: err })
 150           // Don't break the process
 151           return
 152         }
 153
 154         createRequest('add', remoteVideo, [ podId ])
 155       })
 156     })
 157   })
 158 }
 159
 160 // ---------------------------------------------------------------------------
 161
 162 module.exports = friends
 163
 164 // ---------------------------------------------------------------------------
 165
 166 function computeForeignPodsList (url, podsScore, callback) {
 167   getForeignPodsList(url, function (err, foreignPodsList) {
 168     if (err) return callback(err)
 169
 170     if (!foreignPodsList) foreignPodsList = []
 171
 172     // Let's give 1 point to the pod we ask the friends list
 173     foreignPodsList.push({ url: url })
 174
 175     foreignPodsList.forEach(function (foreignPod) {
 176       const foreignPodUrl = foreignPod.url
 177
 178       if (podsScore[foreignPodUrl]) podsScore[foreignPodUrl]++
 179       else podsScore[foreignPodUrl] = 1
 180     })
 181
 182     callback()
 183   })
 184 }
 185
 186 function computeWinningPods (urls, podsScore) {
 187   // Build the list of pods to add
 188   // Only add a pod if it exists in more than a half base pods
 189   const podsList = []
 190   const baseScore = urls.length / 2
 191   Object.keys(podsScore).forEach(function (pod) {
 192     if (podsScore[pod] > baseScore) podsList.push({ url: pod })
 193   })
 194
 195   return podsList
 196 }
 197
 198 function getForeignPodsList (url, callback) {
 199   const path = '/api/' + constants.API_VERSION + '/pods'
 200
 201   request.get(url + path, function (err, response, body) {
 202     if (err) return callback(err)
 203
 204     callback(null, JSON.parse(body))
 205   })
 206 }
 207
 208 function makeRequestsToWinningPods (cert, podsList, callback) {
 209   // Stop pool requests
 210   Request.deactivate()
 211   // Flush pool requests
 212   Request.forceSend()
 213
 214   eachLimit(podsList, constants.REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
 215     const params = {
 216       url: pod.url + '/api/' + constants.API_VERSION + '/pods/',
 217       method: 'POST',
 218       json: {
 219         url: constants.CONFIG.WEBSERVER.URL,
 220         publicKey: cert
 221       }
 222     }
 223
 224     requests.makeRetryRequest(params, function (err, res, body) {
 225       if (err) {
 226         logger.error('Error with adding %s pod.', pod.url, { error: err })
 227         // Don't break the process
 228         return callbackEach()
 229       }
 230
 231       if (res.statusCode === 200) {
 232         const podObj = new Pod({ url: pod.url, publicKey: body.cert })
 233         podObj.save(function (err, podCreated) {
 234           if (err) {
 235             logger.error('Cannot add friend %s pod.', pod.url, { error: err })
 236             return callbackEach()
 237           }
 238
 239           // Add our videos to the request scheduler
 240           sendOwnedVideosToPod(podCreated._id)
 241
 242           return callbackEach()
 243         })
 244       } else {
 245         logger.error('Status not 200 for %s pod.', pod.url)
 246         return callbackEach()
 247       }
 248     })
 249   }, function endRequests () {
 250     // Final callback, we've ended all the requests
 251     // Now we made new friends, we can re activate the pool of requests
 252     Request.activate()
 253
 254     logger.debug('makeRequestsToWinningPods finished.')
 255     return callback()
 256   })
 257 }
 258
 259 function createRequest (type, data, to) {
 260   const req = new Request({
 261     request: {
 262       type: type,
 263       data: data
 264     }
 265   })
 266
 267   if (to) {
 268     req.to = to
 269   }
 270
 271   req.save(function (err) {
 272     if (err) logger.error('Cannot save the request.', { error: err })
 273   })
 274 }