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