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 peertubeCrypto = require('../helpers/peertube-crypto')
  15 const requests = require('../helpers/requests')
  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 const Pod = mongoose.model('Pod')
  21 const Request = mongoose.model('Request')
  22 const Video = mongoose.model('Video')
  23
  24 const friends = {
  25   addVideoToFriends: addVideoToFriends,
  26   hasFriends: hasFriends,
  27   getMyCertificate: getMyCertificate,
  28   makeFriends: makeFriends,
  29   quitFriends: quitFriends,
  30   removeVideoToFriends: removeVideoToFriends,
  31   sendOwnedVideosToPod: sendOwnedVideosToPod
  32 }
  33
  34 function addVideoToFriends (video) {
  35   createRequest('add', video)
  36 }
  37
  38 function hasFriends (callback) {
  39   Pod.countAll(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     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   Request.deactivate()
  80   // Flush pool requests
  81   Request.flush()
  82
  83   waterfall([
  84     function getPodsList (callbackAsync) {
  85       return Pod.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       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       Pod.removeAll(function (err) {
 113         return callbackAsync(err)
 114       })
 115     },
 116
 117     function listRemoteVideos (callbackAsync) {
 118       logger.info('Broke friends, so sad :(')
 119
 120       Video.listRemotes(callbackAsync)
 121     },
 122
 123     function removeTheRemoteVideos (videosList, callbackAsync) {
 124       each(videosList, function (video, callbackEach) {
 125         video.remove(callbackEach)
 126       }, callbackAsync)
 127     }
 128   ], function (err) {
 129     // Don't forget to re activate the scheduler, even if there was an error
 130     Request.activate()
 131
 132     if (err) return callback(err)
 133
 134     logger.info('Removed all remote videos.')
 135     return callback(null)
 136   })
 137 }
 138
 139 function removeVideoToFriends (videoParams) {
 140   createRequest('remove', videoParams)
 141 }
 142
 143 function sendOwnedVideosToPod (podId) {
 144   Video.listOwned(function (err, videosList) {
 145     if (err) {
 146       logger.error('Cannot get the list of videos we own.')
 147       return
 148     }
 149
 150     videosList.forEach(function (video) {
 151       video.toRemoteJSON(function (err, remoteVideo) {
 152         if (err) {
 153           logger.error('Cannot convert video to remote.', { error: err })
 154           // Don't break the process
 155           return
 156         }
 157
 158         createRequest('add', remoteVideo, [ podId ])
 159       })
 160     })
 161   })
 162 }
 163
 164 // ---------------------------------------------------------------------------
 165
 166 module.exports = friends
 167
 168 // ---------------------------------------------------------------------------
 169
 170 function computeForeignPodsList (url, podsScore, callback) {
 171   getForeignPodsList(url, function (err, foreignPodsList) {
 172     if (err) return callback(err)
 173
 174     if (!foreignPodsList) foreignPodsList = []
 175
 176     // Let's give 1 point to the pod we ask the friends list
 177     foreignPodsList.push({ url: url })
 178
 179     foreignPodsList.forEach(function (foreignPod) {
 180       const foreignPodUrl = foreignPod.url
 181
 182       if (podsScore[foreignPodUrl]) podsScore[foreignPodUrl]++
 183       else podsScore[foreignPodUrl] = 1
 184     })
 185
 186     callback()
 187   })
 188 }
 189
 190 function computeWinningPods (urls, podsScore) {
 191   // Build the list of pods to add
 192   // Only add a pod if it exists in more than a half base pods
 193   const podsList = []
 194   const baseScore = urls.length / 2
 195   Object.keys(podsScore).forEach(function (pod) {
 196     if (podsScore[pod] > baseScore) podsList.push({ url: pod })
 197   })
 198
 199   return podsList
 200 }
 201
 202 function getForeignPodsList (url, callback) {
 203   const path = '/api/' + constants.API_VERSION + '/pods'
 204
 205   request.get(url + path, function (err, response, body) {
 206     if (err) return callback(err)
 207
 208     callback(null, JSON.parse(body))
 209   })
 210 }
 211
 212 function makeRequestsToWinningPods (cert, podsList, callback) {
 213   // Stop pool requests
 214   Request.deactivate()
 215   // Flush pool requests
 216   Request.forceSend()
 217
 218   eachLimit(podsList, constants.REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
 219     const params = {
 220       url: pod.url + '/api/' + constants.API_VERSION + '/pods/',
 221       method: 'POST',
 222       json: {
 223         url: http + '://' + host + ':' + port,
 224         publicKey: cert
 225       }
 226     }
 227
 228     requests.makeRetryRequest(params, function (err, res, body) {
 229       if (err) {
 230         logger.error('Error with adding %s pod.', pod.url, { error: err })
 231         // Don't break the process
 232         return callbackEach()
 233       }
 234
 235       if (res.statusCode === 200) {
 236         const podObj = new Pod({ url: pod.url, publicKey: body.cert })
 237         podObj.save(function (err, podCreated) {
 238           if (err) {
 239             logger.error('Cannot add friend %s pod.', pod.url, { error: err })
 240             return callbackEach()
 241           }
 242
 243           // Add our videos to the request scheduler
 244           sendOwnedVideosToPod(podCreated._id)
 245
 246           return callbackEach()
 247         })
 248       } else {
 249         logger.error('Status not 200 for %s pod.', pod.url)
 250         return callbackEach()
 251       }
 252     })
 253   }, function endRequests () {
 254     // Final callback, we've ended all the requests
 255     // Now we made new friends, we can re activate the pool of requests
 256     Request.activate()
 257
 258     logger.debug('makeRequestsToWinningPods finished.')
 259     return callback()
 260   })
 261 }
 262
 263 function createRequest (type, data, to) {
 264   const req = new Request({
 265     request: {
 266       type: type,
 267       data: data
 268     }
 269   })
 270
 271   if (to) {
 272     req.to = to
 273   }
 274
 275   req.save(function (err) {
 276     if (err) logger.error('Cannot save the request.', { error: err })
 277   })
 278 }