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