server/lib/friends.ts

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