server/lib/friends.ts

   1 import { each, eachLimit, eachSeries, series, waterfall } from 'async'
   2 import * as request from 'request'
   3 import * as Sequelize from 'sequelize'
   4
   5 import { database as db } from '../initializers/database'
   6 import {
   7   API_VERSION,
   8   CONFIG,
   9   REQUESTS_IN_PARALLEL,
  10   REQUEST_ENDPOINTS,
  11   REQUEST_ENDPOINT_ACTIONS,
  12   REMOTE_SCHEME
  13 } from '../initializers'
  14 import {
  15   logger,
  16   getMyPublicCert,
  17   makeSecureRequest,
  18   makeRetryRequest,
  19   createEmptyCallback
  20 } from '../helpers'
  21 import {
  22   RequestScheduler,
  23   RequestSchedulerOptions,
  24
  25   RequestVideoQaduScheduler,
  26   RequestVideoQaduSchedulerOptions,
  27
  28   RequestVideoEventScheduler,
  29   RequestVideoEventSchedulerOptions
  30 } from './request'
  31 import {
  32   PodInstance,
  33   VideoInstance
  34 } from '../models'
  35 import {
  36   RequestEndpoint,
  37   RequestVideoEventType,
  38   RequestVideoQaduType
  39 } from '../../shared'
  40
  41 type QaduParam = { videoId: string, type: RequestVideoQaduType }
  42 type EventParam = { videoId: string, type: RequestVideoEventType }
  43
  44 const ENDPOINT_ACTIONS = REQUEST_ENDPOINT_ACTIONS[REQUEST_ENDPOINTS.VIDEOS]
  45
  46 const requestScheduler = new RequestScheduler()
  47 const requestVideoQaduScheduler = new RequestVideoQaduScheduler()
  48 const requestVideoEventScheduler = new RequestVideoEventScheduler()
  49
  50 function activateSchedulers () {
  51   requestScheduler.activate()
  52   requestVideoQaduScheduler.activate()
  53   requestVideoEventScheduler.activate()
  54 }
  55
  56 function addVideoToFriends (videoData: Object, transaction: Sequelize.Transaction, callback: (err: Error) => void) {
  57   const options = {
  58     type: ENDPOINT_ACTIONS.ADD,
  59     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  60     data: videoData,
  61     transaction
  62   }
  63   createRequest(options, callback)
  64 }
  65
  66 function updateVideoToFriends (videoData: Object, transaction: Sequelize.Transaction, callback: (err: Error) => void) {
  67   const options = {
  68     type: ENDPOINT_ACTIONS.UPDATE,
  69     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  70     data: videoData,
  71     transaction
  72   }
  73   createRequest(options, callback)
  74 }
  75
  76 function removeVideoToFriends (videoParams: Object) {
  77   const options = {
  78     type: ENDPOINT_ACTIONS.REMOVE,
  79     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  80     data: videoParams,
  81     transaction: null
  82   }
  83   createRequest(options)
  84 }
  85
  86 function reportAbuseVideoToFriend (reportData: Object, video: VideoInstance) {
  87   const options = {
  88     type: ENDPOINT_ACTIONS.REPORT_ABUSE,
  89     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  90     data: reportData,
  91     toIds: [ video.Author.podId ],
  92     transaction: null
  93   }
  94   createRequest(options)
  95 }
  96
  97 function quickAndDirtyUpdateVideoToFriends (qaduParam: QaduParam, transaction?: Sequelize.Transaction, callback?: (err: Error) => void) {
  98   const options = {
  99     videoId: qaduParam.videoId,
 100     type: qaduParam.type,
 101     transaction
 102   }
 103   return createVideoQaduRequest(options, callback)
 104 }
 105
 106 function quickAndDirtyUpdatesVideoToFriends (
 107   qadusParams: QaduParam[],
 108   transaction: Sequelize.Transaction,
 109   finalCallback: (err: Error) => void
 110 ) {
 111   const tasks = []
 112
 113   qadusParams.forEach(function (qaduParams) {
 114     const fun = function (callback) {
 115       quickAndDirtyUpdateVideoToFriends(qaduParams, transaction, callback)
 116     }
 117
 118     tasks.push(fun)
 119   })
 120
 121   series(tasks, finalCallback)
 122 }
 123
 124 function addEventToRemoteVideo (eventParam: EventParam, transaction?: Sequelize.Transaction, callback?: (err: Error) => void) {
 125   const options = {
 126     videoId: eventParam.videoId,
 127     type: eventParam.type,
 128     transaction
 129   }
 130   createVideoEventRequest(options, callback)
 131 }
 132
 133 function addEventsToRemoteVideo (eventsParams: EventParam[], transaction: Sequelize.Transaction, finalCallback: (err: Error) => void) {
 134   const tasks = []
 135
 136   eventsParams.forEach(function (eventParams) {
 137     const fun = function (callback) {
 138       addEventToRemoteVideo(eventParams, transaction, callback)
 139     }
 140
 141     tasks.push(fun)
 142   })
 143
 144   series(tasks, finalCallback)
 145 }
 146
 147 function hasFriends (callback: (err: Error, hasFriends?: boolean) => void) {
 148   db.Pod.countAll(function (err, count) {
 149     if (err) return callback(err)
 150
 151     const hasFriends = (count !== 0)
 152     callback(null, hasFriends)
 153   })
 154 }
 155
 156 function makeFriends (hosts: string[], callback: (err: Error) => void) {
 157   const podsScore = {}
 158
 159   logger.info('Make friends!')
 160   getMyPublicCert(function (err, cert) {
 161     if (err) {
 162       logger.error('Cannot read public cert.')
 163       return callback(err)
 164     }
 165
 166     eachSeries(hosts, function (host, callbackEach) {
 167       computeForeignPodsList(host, podsScore, callbackEach)
 168     }, function (err: Error) {
 169       if (err) return callback(err)
 170
 171       logger.debug('Pods scores computed.', { podsScore: podsScore })
 172       const podsList = computeWinningPods(hosts, podsScore)
 173       logger.debug('Pods that we keep.', { podsToKeep: podsList })
 174
 175       makeRequestsToWinningPods(cert, podsList, callback)
 176     })
 177   })
 178 }
 179
 180 function quitFriends (callback: (err: Error) => void) {
 181   // Stop pool requests
 182   requestScheduler.deactivate()
 183
 184   waterfall([
 185     function flushRequests (callbackAsync) {
 186       requestScheduler.flush(err => callbackAsync(err))
 187     },
 188
 189     function flushVideoQaduRequests (callbackAsync) {
 190       requestVideoQaduScheduler.flush(err => callbackAsync(err))
 191     },
 192
 193     function getPodsList (callbackAsync) {
 194       return db.Pod.list(callbackAsync)
 195     },
 196
 197     function announceIQuitMyFriends (pods, callbackAsync) {
 198       const requestParams = {
 199         method: 'POST' as 'POST',
 200         path: '/api/' + API_VERSION + '/remote/pods/remove',
 201         sign: true,
 202         toPod: null
 203       }
 204
 205       // Announce we quit them
 206       // We don't care if the request fails
 207       // The other pod will exclude us automatically after a while
 208       eachLimit(pods, REQUESTS_IN_PARALLEL, function (pod, callbackEach) {
 209         requestParams.toPod = pod
 210         makeSecureRequest(requestParams, callbackEach)
 211       }, function (err) {
 212         if (err) {
 213           logger.error('Some errors while quitting friends.', { err: err })
 214           // Don't stop the process
 215         }
 216
 217         return callbackAsync(null, pods)
 218       })
 219     },
 220
 221     function removePodsFromDB (pods, callbackAsync) {
 222       each(pods, function (pod: any, callbackEach) {
 223         pod.destroy().asCallback(callbackEach)
 224       }, callbackAsync)
 225     }
 226   ], function (err: Error) {
 227     // Don't forget to re activate the scheduler, even if there was an error
 228     requestScheduler.activate()
 229
 230     if (err) return callback(err)
 231
 232     logger.info('Removed all remote videos.')
 233     return callback(null)
 234   })
 235 }
 236
 237 function sendOwnedVideosToPod (podId: number) {
 238   db.Video.listOwnedAndPopulateAuthorAndTags(function (err, videosList) {
 239     if (err) {
 240       logger.error('Cannot get the list of videos we own.')
 241       return
 242     }
 243
 244     videosList.forEach(function (video) {
 245       video.toAddRemoteJSON(function (err, remoteVideo) {
 246         if (err) {
 247           logger.error('Cannot convert video to remote.', { error: err })
 248           // Don't break the process
 249           return
 250         }
 251
 252         const options = {
 253           type: 'add',
 254           endpoint: REQUEST_ENDPOINTS.VIDEOS,
 255           data: remoteVideo,
 256           toIds: [ podId ],
 257           transaction: null
 258         }
 259         createRequest(options)
 260       })
 261     })
 262   })
 263 }
 264
 265 function getRequestScheduler () {
 266   return requestScheduler
 267 }
 268
 269 function getRequestVideoQaduScheduler () {
 270   return requestVideoQaduScheduler
 271 }
 272
 273 function getRequestVideoEventScheduler () {
 274   return requestVideoEventScheduler
 275 }
 276
 277 // ---------------------------------------------------------------------------
 278
 279 export {
 280   activateSchedulers,
 281   addVideoToFriends,
 282   updateVideoToFriends,
 283   reportAbuseVideoToFriend,
 284   quickAndDirtyUpdateVideoToFriends,
 285   quickAndDirtyUpdatesVideoToFriends,
 286   addEventToRemoteVideo,
 287   addEventsToRemoteVideo,
 288   hasFriends,
 289   makeFriends,
 290   quitFriends,
 291   removeVideoToFriends,
 292   sendOwnedVideosToPod,
 293   getRequestScheduler,
 294   getRequestVideoQaduScheduler,
 295   getRequestVideoEventScheduler
 296 }
 297
 298 // ---------------------------------------------------------------------------
 299
 300 function computeForeignPodsList (host: string, podsScore: { [ host: string ]: number }, callback: (err: Error) => void) {
 301   getForeignPodsList(host, function (err, res) {
 302     if (err) return callback(err)
 303
 304     const foreignPodsList = res.data
 305
 306     // Let's give 1 point to the pod we ask the friends list
 307     foreignPodsList.push({ host })
 308
 309     foreignPodsList.forEach(function (foreignPod) {
 310       const foreignPodHost = foreignPod.host
 311
 312       if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 313       else podsScore[foreignPodHost] = 1
 314     })
 315
 316     return callback(null)
 317   })
 318 }
 319
 320 function computeWinningPods (hosts: string[], podsScore: { [ host: string ]: number }) {
 321   // Build the list of pods to add
 322   // Only add a pod if it exists in more than a half base pods
 323   const podsList = []
 324   const baseScore = hosts.length / 2
 325
 326   Object.keys(podsScore).forEach(function (podHost) {
 327     // If the pod is not me and with a good score we add it
 328     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 329       podsList.push({ host: podHost })
 330     }
 331   })
 332
 333   return podsList
 334 }
 335
 336 function getForeignPodsList (host: string, callback: (err: Error, foreignPodsList?: any) => void) {
 337   const path = '/api/' + API_VERSION + '/pods'
 338
 339   request.get(REMOTE_SCHEME.HTTP + '://' + host + path, function (err, response, body) {
 340     if (err) return callback(err)
 341
 342     try {
 343       const json = JSON.parse(body)
 344       return callback(null, json)
 345     } catch (err) {
 346       return callback(err)
 347     }
 348   })
 349 }
 350
 351 function makeRequestsToWinningPods (cert: string, podsList: PodInstance[], callback: (err: Error) => void) {
 352   // Stop pool requests
 353   requestScheduler.deactivate()
 354   // Flush pool requests
 355   requestScheduler.forceSend()
 356
 357   eachLimit(podsList, REQUESTS_IN_PARALLEL, function (pod: PodInstance, callbackEach) {
 358     const params = {
 359       url: REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + API_VERSION + '/pods/',
 360       method: 'POST' as 'POST',
 361       json: {
 362         host: CONFIG.WEBSERVER.HOST,
 363         email: CONFIG.ADMIN.EMAIL,
 364         publicKey: cert
 365       }
 366     }
 367
 368     makeRetryRequest(params, function (err, res, body: { cert: string, email: string }) {
 369       if (err) {
 370         logger.error('Error with adding %s pod.', pod.host, { error: err })
 371         // Don't break the process
 372         return callbackEach()
 373       }
 374
 375       if (res.statusCode === 200) {
 376         const podObj = db.Pod.build({ host: pod.host, publicKey: body.cert, email: body.email })
 377         podObj.save().asCallback(function (err, podCreated) {
 378           if (err) {
 379             logger.error('Cannot add friend %s pod.', pod.host, { error: err })
 380             return callbackEach()
 381           }
 382
 383           // Add our videos to the request scheduler
 384           sendOwnedVideosToPod(podCreated.id)
 385
 386           return callbackEach()
 387         })
 388       } else {
 389         logger.error('Status not 200 for %s pod.', pod.host)
 390         return callbackEach()
 391       }
 392     })
 393   }, function endRequests () {
 394     // Final callback, we've ended all the requests
 395     // Now we made new friends, we can re activate the pool of requests
 396     requestScheduler.activate()
 397
 398     logger.debug('makeRequestsToWinningPods finished.')
 399     return callback(null)
 400   })
 401 }
 402
 403 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 404 type CreateRequestOptions = {
 405   type: string
 406   endpoint: RequestEndpoint
 407   data: Object
 408   toIds?: number[]
 409   transaction: Sequelize.Transaction
 410 }
 411 function createRequest (options: CreateRequestOptions, callback?: (err: Error) => void) {
 412   if (!callback) callback = function () { /* empty */ }
 413
 414   if (options.toIds !== undefined) return requestScheduler.createRequest(options as RequestSchedulerOptions, callback)
 415
 416   // If the "toIds" pods is not specified, we send the request to all our friends
 417   db.Pod.listAllIds(options.transaction, function (err, podIds) {
 418     if (err) {
 419       logger.error('Cannot get pod ids', { error: err })
 420       return
 421     }
 422
 423     const newOptions = Object.assign(options, { toIds: podIds })
 424     return requestScheduler.createRequest(newOptions, callback)
 425   })
 426 }
 427
 428 function createVideoQaduRequest (options: RequestVideoQaduSchedulerOptions, callback: (err: Error) => void) {
 429   if (!callback) callback = createEmptyCallback()
 430
 431   requestVideoQaduScheduler.createRequest(options, callback)
 432 }
 433
 434 function createVideoEventRequest (options: RequestVideoEventSchedulerOptions, callback: (err: Error) => void) {
 435   if (!callback) callback = createEmptyCallback()
 436
 437   requestVideoEventScheduler.createRequest(options, callback)
 438 }
 439
 440 function isMe (host: string) {
 441   return host === CONFIG.WEBSERVER.HOST
 442 }