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