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