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