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