server/lib/friends.ts

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