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, transaction: Sequelize.Transaction) {
  84   const options = {
  85     type: ENDPOINT_ACTIONS.REMOVE,
  86     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  87     data: videoParams,
  88     transaction
  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         return pods
 194       })
 195     })
 196     .then(pods => {
 197       const tasks = []
 198       pods.forEach(pod => tasks.push(pod.destroy()))
 199
 200       return Promise.all(pods)
 201     })
 202     .then(() => {
 203       logger.info('Removed all remote videos.')
 204       // Don't forget to re activate the scheduler, even if there was an error
 205       return requestScheduler.activate()
 206     })
 207     .finally(() => requestScheduler.activate())
 208 }
 209
 210 function sendOwnedVideosToPod (podId: number) {
 211   db.Video.listOwnedAndPopulateAuthorAndTags()
 212     .then(videosList => {
 213       const tasks = []
 214       videosList.forEach(video => {
 215         const promise = video.toAddRemoteJSON()
 216           .then(remoteVideo => {
 217             const options = {
 218               type: 'add',
 219               endpoint: REQUEST_ENDPOINTS.VIDEOS,
 220               data: remoteVideo,
 221               toIds: [ podId ],
 222               transaction: null
 223             }
 224             return createRequest(options)
 225           })
 226           .catch(err => {
 227             logger.error('Cannot convert video to remote.', err)
 228             // Don't break the process
 229             return undefined
 230           })
 231
 232         tasks.push(promise)
 233       })
 234
 235       return Promise.all(tasks)
 236     })
 237 }
 238
 239 function fetchRemotePreview (pod: PodInstance, video: VideoInstance) {
 240   const host = video.Author.Pod.host
 241   const path = join(STATIC_PATHS.PREVIEWS, video.getPreviewName())
 242
 243   return request.get(REMOTE_SCHEME.HTTP + '://' + host + path)
 244 }
 245
 246 function removeFriend (pod: PodInstance) {
 247   const requestParams = {
 248     method: 'POST' as 'POST',
 249     path: '/api/' + API_VERSION + '/remote/pods/remove',
 250     toPod: pod
 251   }
 252
 253   return makeSecureRequest(requestParams)
 254     .catch(err => logger.warn('Cannot notify friends %s we are quitting him.', pod.host, err))
 255     .then(() => pod.destroy())
 256     .then(() => logger.info('Removed friend %s.', pod.host))
 257     .catch(err => logger.error('Cannot destroy friend %s.', pod.host, err))
 258 }
 259
 260 function getRequestScheduler () {
 261   return requestScheduler
 262 }
 263
 264 function getRequestVideoQaduScheduler () {
 265   return requestVideoQaduScheduler
 266 }
 267
 268 function getRequestVideoEventScheduler () {
 269   return requestVideoEventScheduler
 270 }
 271
 272 // ---------------------------------------------------------------------------
 273
 274 export {
 275   activateSchedulers,
 276   addVideoToFriends,
 277   updateVideoToFriends,
 278   reportAbuseVideoToFriend,
 279   quickAndDirtyUpdateVideoToFriends,
 280   quickAndDirtyUpdatesVideoToFriends,
 281   addEventToRemoteVideo,
 282   addEventsToRemoteVideo,
 283   hasFriends,
 284   makeFriends,
 285   quitFriends,
 286   removeFriend,
 287   removeVideoToFriends,
 288   sendOwnedVideosToPod,
 289   getRequestScheduler,
 290   getRequestVideoQaduScheduler,
 291   getRequestVideoEventScheduler,
 292   fetchRemotePreview
 293 }
 294
 295 // ---------------------------------------------------------------------------
 296
 297 function computeForeignPodsList (host: string, podsScore: { [ host: string ]: number }) {
 298   // TODO: type res
 299   return getForeignPodsList(host).then(res => {
 300     const foreignPodsList: { host: string }[] = res.data
 301
 302     // Let's give 1 point to the pod we ask the friends list
 303     foreignPodsList.push({ host })
 304
 305     foreignPodsList.forEach(foreignPod => {
 306       const foreignPodHost = foreignPod.host
 307
 308       if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 309       else podsScore[foreignPodHost] = 1
 310     })
 311
 312     return undefined
 313   })
 314 }
 315
 316 function computeWinningPods (hosts: string[], podsScore: { [ host: string ]: number }) {
 317   // Build the list of pods to add
 318   // Only add a pod if it exists in more than a half base pods
 319   const podsList = []
 320   const baseScore = hosts.length / 2
 321
 322   Object.keys(podsScore).forEach(podHost => {
 323     // If the pod is not me and with a good score we add it
 324     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 325       podsList.push({ host: podHost })
 326     }
 327   })
 328
 329   return podsList
 330 }
 331
 332 function getForeignPodsList (host: string) {
 333   return new Promise< ResultList<FormattedPod> >((res, rej) => {
 334     const path = '/api/' + API_VERSION + '/remote/pods/list'
 335
 336     request.post(REMOTE_SCHEME.HTTP + '://' + host + path, (err, response, body) => {
 337       if (err) return rej(err)
 338
 339       try {
 340         const json = JSON.parse(body)
 341         return res(json)
 342       } catch (err) {
 343         return rej(err)
 344       }
 345     })
 346   })
 347 }
 348
 349 function makeRequestsToWinningPods (cert: string, podsList: PodInstance[]) {
 350   // Stop pool requests
 351   requestScheduler.deactivate()
 352   // Flush pool requests
 353   requestScheduler.forceSend()
 354
 355   return Promise.map(podsList, pod => {
 356     const params = {
 357       url: REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + API_VERSION + '/remote/pods/add',
 358       method: 'POST' as 'POST',
 359       json: {
 360         host: CONFIG.WEBSERVER.HOST,
 361         email: CONFIG.ADMIN.EMAIL,
 362         publicKey: cert
 363       }
 364     }
 365
 366     return makeRetryRequest(params)
 367       .then(({ response, body }) => {
 368         body = body as { cert: string, email: string }
 369
 370         if (response.statusCode === 200) {
 371           const podObj = db.Pod.build({ host: pod.host, publicKey: body.cert, email: body.email })
 372           return podObj.save()
 373             .then(podCreated => {
 374
 375               // Add our videos to the request scheduler
 376               sendOwnedVideosToPod(podCreated.id)
 377             })
 378             .catch(err => {
 379               logger.error('Cannot add friend %s pod.', pod.host, err)
 380             })
 381         } else {
 382           logger.error('Status not 200 for %s pod.', pod.host)
 383         }
 384       })
 385       .catch(err => {
 386         logger.error('Error with adding %s pod.', pod.host, { error: err.stack })
 387         // Don't break the process
 388       })
 389   }, { concurrency: REQUESTS_IN_PARALLEL })
 390   .then(() => logger.debug('makeRequestsToWinningPods finished.'))
 391   .finally(() => {
 392     // Final callback, we've ended all the requests
 393     // Now we made new friends, we can re activate the pool of requests
 394     requestScheduler.activate()
 395   })
 396 }
 397
 398 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 399 type CreateRequestOptions = {
 400   type: string
 401   endpoint: RequestEndpoint
 402   data: Object
 403   toIds?: number[]
 404   transaction: Sequelize.Transaction
 405 }
 406 function createRequest (options: CreateRequestOptions) {
 407   if (options.toIds !== undefined) return requestScheduler.createRequest(options as RequestSchedulerOptions)
 408
 409   // If the "toIds" pods is not specified, we send the request to all our friends
 410   return db.Pod.listAllIds(options.transaction).then(podIds => {
 411     const newOptions = Object.assign(options, { toIds: podIds })
 412     return requestScheduler.createRequest(newOptions)
 413   })
 414 }
 415
 416 function createVideoQaduRequest (options: RequestVideoQaduSchedulerOptions) {
 417   return requestVideoQaduScheduler.createRequest(options)
 418 }
 419
 420 function createVideoEventRequest (options: RequestVideoEventSchedulerOptions) {
 421   return requestVideoEventScheduler.createRequest(options)
 422 }
 423
 424 function isMe (host: string) {
 425   return host === CONFIG.WEBSERVER.HOST
 426 }