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 FormatedPod
  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 getRequestScheduler () {
 246   return requestScheduler
 247 }
 248
 249 function getRequestVideoQaduScheduler () {
 250   return requestVideoQaduScheduler
 251 }
 252
 253 function getRequestVideoEventScheduler () {
 254   return requestVideoEventScheduler
 255 }
 256
 257 // ---------------------------------------------------------------------------
 258
 259 export {
 260   activateSchedulers,
 261   addVideoToFriends,
 262   updateVideoToFriends,
 263   reportAbuseVideoToFriend,
 264   quickAndDirtyUpdateVideoToFriends,
 265   quickAndDirtyUpdatesVideoToFriends,
 266   addEventToRemoteVideo,
 267   addEventsToRemoteVideo,
 268   hasFriends,
 269   makeFriends,
 270   quitFriends,
 271   removeVideoToFriends,
 272   sendOwnedVideosToPod,
 273   getRequestScheduler,
 274   getRequestVideoQaduScheduler,
 275   getRequestVideoEventScheduler,
 276   fetchRemotePreview
 277 }
 278
 279 // ---------------------------------------------------------------------------
 280
 281 function computeForeignPodsList (host: string, podsScore: { [ host: string ]: number }) {
 282   // TODO: type res
 283   return getForeignPodsList(host).then(res => {
 284     const foreignPodsList: { host: string }[] = res.data
 285
 286     // Let's give 1 point to the pod we ask the friends list
 287     foreignPodsList.push({ host })
 288
 289     foreignPodsList.forEach(foreignPod => {
 290       const foreignPodHost = foreignPod.host
 291
 292       if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 293       else podsScore[foreignPodHost] = 1
 294     })
 295
 296     return undefined
 297   })
 298 }
 299
 300 function computeWinningPods (hosts: string[], podsScore: { [ host: string ]: number }) {
 301   // Build the list of pods to add
 302   // Only add a pod if it exists in more than a half base pods
 303   const podsList = []
 304   const baseScore = hosts.length / 2
 305
 306   Object.keys(podsScore).forEach(podHost => {
 307     // If the pod is not me and with a good score we add it
 308     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 309       podsList.push({ host: podHost })
 310     }
 311   })
 312
 313   return podsList
 314 }
 315
 316 function getForeignPodsList (host: string) {
 317   return new Promise< ResultList<FormatedPod> >((res, rej) => {
 318     const path = '/api/' + API_VERSION + '/pods'
 319
 320     request.get(REMOTE_SCHEME.HTTP + '://' + host + path, (err, response, body) => {
 321       if (err) return rej(err)
 322
 323       try {
 324         const json = JSON.parse(body)
 325         return res(json)
 326       } catch (err) {
 327         return rej(err)
 328       }
 329     })
 330   })
 331 }
 332
 333 function makeRequestsToWinningPods (cert: string, podsList: PodInstance[]) {
 334   // Stop pool requests
 335   requestScheduler.deactivate()
 336   // Flush pool requests
 337   requestScheduler.forceSend()
 338
 339   return Promise.map(podsList, pod => {
 340     const params = {
 341       url: REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + API_VERSION + '/pods/',
 342       method: 'POST' as 'POST',
 343       json: {
 344         host: CONFIG.WEBSERVER.HOST,
 345         email: CONFIG.ADMIN.EMAIL,
 346         publicKey: cert
 347       }
 348     }
 349
 350     return makeRetryRequest(params)
 351       .then(({ response, body }) => {
 352         body = body as { cert: string, email: string }
 353
 354         if (response.statusCode === 200) {
 355           const podObj = db.Pod.build({ host: pod.host, publicKey: body.cert, email: body.email })
 356           return podObj.save()
 357             .then(podCreated => {
 358
 359               // Add our videos to the request scheduler
 360               sendOwnedVideosToPod(podCreated.id)
 361             })
 362             .catch(err => {
 363               logger.error('Cannot add friend %s pod.', pod.host, err)
 364             })
 365         } else {
 366           logger.error('Status not 200 for %s pod.', pod.host)
 367         }
 368       })
 369       .catch(err => {
 370         logger.error('Error with adding %s pod.', pod.host, { error: err.stack })
 371         // Don't break the process
 372       })
 373   }, { concurrency: REQUESTS_IN_PARALLEL })
 374   .then(() => logger.debug('makeRequestsToWinningPods finished.'))
 375   .finally(() => {
 376     // Final callback, we've ended all the requests
 377     // Now we made new friends, we can re activate the pool of requests
 378     requestScheduler.activate()
 379   })
 380 }
 381
 382 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 383 type CreateRequestOptions = {
 384   type: string
 385   endpoint: RequestEndpoint
 386   data: Object
 387   toIds?: number[]
 388   transaction: Sequelize.Transaction
 389 }
 390 function createRequest (options: CreateRequestOptions) {
 391   if (options.toIds !== undefined) return requestScheduler.createRequest(options as RequestSchedulerOptions)
 392
 393   // If the "toIds" pods is not specified, we send the request to all our friends
 394   return db.Pod.listAllIds(options.transaction).then(podIds => {
 395     const newOptions = Object.assign(options, { toIds: podIds })
 396     return requestScheduler.createRequest(newOptions)
 397   })
 398 }
 399
 400 function createVideoQaduRequest (options: RequestVideoQaduSchedulerOptions) {
 401   return requestVideoQaduScheduler.createRequest(options)
 402 }
 403
 404 function createVideoEventRequest (options: RequestVideoEventSchedulerOptions) {
 405   return requestVideoEventScheduler.createRequest(options)
 406 }
 407
 408 function isMe (host: string) {
 409   return host === CONFIG.WEBSERVER.HOST
 410 }