server/lib/friends.js

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