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   RemoteVideoRequestType,
  46   Pod as FormattedPod,
  47   RemoteVideoChannelCreateData,
  48   RemoteVideoChannelUpdateData,
  49   RemoteVideoChannelRemoveData,
  50   RemoteVideoAuthorCreateData,
  51   RemoteVideoAuthorRemoveData
  52 } from '../../shared'
  53
  54 type QaduParam = { videoId: number, type: RequestVideoQaduType }
  55 type EventParam = { videoId: number, type: RequestVideoEventType }
  56
  57 const ENDPOINT_ACTIONS = REQUEST_ENDPOINT_ACTIONS[REQUEST_ENDPOINTS.VIDEOS]
  58
  59 const requestScheduler = new RequestScheduler()
  60 const requestVideoQaduScheduler = new RequestVideoQaduScheduler()
  61 const requestVideoEventScheduler = new RequestVideoEventScheduler()
  62
  63 function activateSchedulers () {
  64   requestScheduler.activate()
  65   requestVideoQaduScheduler.activate()
  66   requestVideoEventScheduler.activate()
  67 }
  68
  69 function addVideoToFriends (videoData: RemoteVideoCreateData, transaction: Sequelize.Transaction) {
  70   const options = {
  71     type: ENDPOINT_ACTIONS.ADD_VIDEO,
  72     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  73     data: videoData,
  74     transaction
  75   }
  76   return createRequest(options)
  77 }
  78
  79 function updateVideoToFriends (videoData: RemoteVideoUpdateData, transaction: Sequelize.Transaction) {
  80   const options = {
  81     type: ENDPOINT_ACTIONS.UPDATE_VIDEO,
  82     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  83     data: videoData,
  84     transaction
  85   }
  86   return createRequest(options)
  87 }
  88
  89 function removeVideoToFriends (videoParams: RemoteVideoRemoveData, transaction: Sequelize.Transaction) {
  90   const options = {
  91     type: ENDPOINT_ACTIONS.REMOVE_VIDEO,
  92     endpoint: REQUEST_ENDPOINTS.VIDEOS,
  93     data: videoParams,
  94     transaction
  95   }
  96   return createRequest(options)
  97 }
  98
  99 function addVideoAuthorToFriends (authorData: RemoteVideoAuthorCreateData, transaction: Sequelize.Transaction) {
 100   const options = {
 101     type: ENDPOINT_ACTIONS.ADD_AUTHOR,
 102     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 103     data: authorData,
 104     transaction
 105   }
 106   return createRequest(options)
 107 }
 108
 109 function removeVideoAuthorToFriends (authorData: RemoteVideoAuthorRemoveData, transaction: Sequelize.Transaction) {
 110   const options = {
 111     type: ENDPOINT_ACTIONS.REMOVE_AUTHOR,
 112     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 113     data: authorData,
 114     transaction
 115   }
 116   return createRequest(options)
 117 }
 118
 119 function addVideoChannelToFriends (videoChannelData: RemoteVideoChannelCreateData, transaction: Sequelize.Transaction) {
 120   const options = {
 121     type: ENDPOINT_ACTIONS.ADD_CHANNEL,
 122     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 123     data: videoChannelData,
 124     transaction
 125   }
 126   return createRequest(options)
 127 }
 128
 129 function updateVideoChannelToFriends (videoChannelData: RemoteVideoChannelUpdateData, transaction: Sequelize.Transaction) {
 130   const options = {
 131     type: ENDPOINT_ACTIONS.UPDATE_CHANNEL,
 132     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 133     data: videoChannelData,
 134     transaction
 135   }
 136   return createRequest(options)
 137 }
 138
 139 function removeVideoChannelToFriends (videoChannelParams: RemoteVideoChannelRemoveData, transaction: Sequelize.Transaction) {
 140   const options = {
 141     type: ENDPOINT_ACTIONS.REMOVE_CHANNEL,
 142     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 143     data: videoChannelParams,
 144     transaction
 145   }
 146   return createRequest(options)
 147 }
 148
 149 function reportAbuseVideoToFriend (reportData: RemoteVideoReportAbuseData, video: VideoInstance, transaction: Sequelize.Transaction) {
 150   const options = {
 151     type: ENDPOINT_ACTIONS.REPORT_ABUSE,
 152     endpoint: REQUEST_ENDPOINTS.VIDEOS,
 153     data: reportData,
 154     toIds: [ video.VideoChannel.Author.podId ],
 155     transaction
 156   }
 157   return createRequest(options)
 158 }
 159
 160 function quickAndDirtyUpdateVideoToFriends (qaduParam: QaduParam, transaction?: Sequelize.Transaction) {
 161   const options = {
 162     videoId: qaduParam.videoId,
 163     type: qaduParam.type,
 164     transaction
 165   }
 166   return createVideoQaduRequest(options)
 167 }
 168
 169 function quickAndDirtyUpdatesVideoToFriends (qadusParams: QaduParam[], transaction: Sequelize.Transaction) {
 170   const tasks = []
 171
 172   qadusParams.forEach(qaduParams => {
 173     tasks.push(quickAndDirtyUpdateVideoToFriends(qaduParams, transaction))
 174   })
 175
 176   return Promise.all(tasks)
 177 }
 178
 179 function addEventToRemoteVideo (eventParam: EventParam, transaction?: Sequelize.Transaction) {
 180   const options = {
 181     videoId: eventParam.videoId,
 182     type: eventParam.type,
 183     transaction
 184   }
 185   return createVideoEventRequest(options)
 186 }
 187
 188 function addEventsToRemoteVideo (eventsParams: EventParam[], transaction: Sequelize.Transaction) {
 189   const tasks = []
 190
 191   eventsParams.forEach(eventParams => {
 192     tasks.push(addEventToRemoteVideo(eventParams, transaction))
 193   })
 194
 195   return Promise.all(tasks)
 196 }
 197
 198 function hasFriends () {
 199   return db.Pod.countAll().then(count => count !== 0)
 200 }
 201
 202 function makeFriends (hosts: string[]) {
 203   const podsScore = {}
 204
 205   logger.info('Make friends!')
 206   return getMyPublicCert()
 207     .then(cert => {
 208       return Promise.each(hosts, host => computeForeignPodsList(host, podsScore)).then(() => cert)
 209     })
 210     .then(cert => {
 211       logger.debug('Pods scores computed.', { podsScore: podsScore })
 212       const podsList = computeWinningPods(hosts, podsScore)
 213       logger.debug('Pods that we keep.', { podsToKeep: podsList })
 214
 215       return makeRequestsToWinningPods(cert, podsList)
 216     })
 217 }
 218
 219 function quitFriends () {
 220   // Stop pool requests
 221   requestScheduler.deactivate()
 222
 223   return requestScheduler.flush()
 224     .then(() => {
 225       return requestVideoQaduScheduler.flush()
 226     })
 227     .then(() => {
 228       return db.Pod.list()
 229     })
 230     .then(pods => {
 231       const requestParams = {
 232         method: 'POST' as 'POST',
 233         path: '/api/' + API_VERSION + '/remote/pods/remove',
 234         toPod: null
 235       }
 236
 237       // Announce we quit them
 238       // We don't care if the request fails
 239       // The other pod will exclude us automatically after a while
 240       return Promise.map(pods, pod => {
 241         requestParams.toPod = pod
 242
 243         return makeSecureRequest(requestParams)
 244       }, { concurrency: REQUESTS_IN_PARALLEL })
 245       .then(() => pods)
 246       .catch(err => {
 247         logger.error('Some errors while quitting friends.', err)
 248         // Don't stop the process
 249         return pods
 250       })
 251     })
 252     .then(pods => {
 253       const tasks = []
 254       pods.forEach(pod => tasks.push(pod.destroy()))
 255
 256       return Promise.all(pods)
 257     })
 258     .then(() => {
 259       logger.info('Removed all remote videos.')
 260       // Don't forget to re activate the scheduler, even if there was an error
 261       return requestScheduler.activate()
 262     })
 263     .finally(() => requestScheduler.activate())
 264 }
 265
 266 function sendOwnedDataToPod (podId: number) {
 267   // First send authors
 268   return sendOwnedAuthorsToPod(podId)
 269     .then(() => sendOwnedChannelsToPod(podId))
 270     .then(() => sendOwnedVideosToPod(podId))
 271 }
 272
 273 function sendOwnedChannelsToPod (podId: number) {
 274   return db.VideoChannel.listOwned()
 275     .then(videoChannels => {
 276       const tasks = []
 277       videoChannels.forEach(videoChannel => {
 278         const remoteVideoChannel = videoChannel.toAddRemoteJSON()
 279         const options = {
 280           type: 'add-channel' as 'add-channel',
 281           endpoint: REQUEST_ENDPOINTS.VIDEOS,
 282           data: remoteVideoChannel,
 283           toIds: [ podId ],
 284           transaction: null
 285         }
 286
 287         const p = createRequest(options)
 288         tasks.push(p)
 289       })
 290
 291       return Promise.all(tasks)
 292     })
 293 }
 294
 295 function sendOwnedAuthorsToPod (podId: number) {
 296   return db.Author.listOwned()
 297     .then(authors => {
 298       const tasks = []
 299       authors.forEach(author => {
 300         const remoteAuthor = author.toAddRemoteJSON()
 301         const options = {
 302           type: 'add-author' as 'add-author',
 303           endpoint: REQUEST_ENDPOINTS.VIDEOS,
 304           data: remoteAuthor,
 305           toIds: [ podId ],
 306           transaction: null
 307         }
 308
 309         const p = createRequest(options)
 310         tasks.push(p)
 311       })
 312
 313       return Promise.all(tasks)
 314     })
 315 }
 316
 317 function sendOwnedVideosToPod (podId: number) {
 318   return db.Video.listOwnedAndPopulateAuthorAndTags()
 319     .then(videosList => {
 320       const tasks = []
 321       videosList.forEach(video => {
 322         const promise = video.toAddRemoteJSON()
 323           .then(remoteVideo => {
 324             const options = {
 325               type: 'add-video' as 'add-video',
 326               endpoint: REQUEST_ENDPOINTS.VIDEOS,
 327               data: remoteVideo,
 328               toIds: [ podId ],
 329               transaction: null
 330             }
 331             return createRequest(options)
 332           })
 333           .catch(err => {
 334             logger.error('Cannot convert video to remote.', err)
 335             // Don't break the process
 336             return undefined
 337           })
 338
 339         tasks.push(promise)
 340       })
 341
 342       return Promise.all(tasks)
 343     })
 344 }
 345
 346 function fetchRemotePreview (video: VideoInstance) {
 347   const host = video.VideoChannel.Author.Pod.host
 348   const path = join(STATIC_PATHS.PREVIEWS, video.getPreviewName())
 349
 350   return request.get(REMOTE_SCHEME.HTTP + '://' + host + path)
 351 }
 352
 353 function removeFriend (pod: PodInstance) {
 354   const requestParams = {
 355     method: 'POST' as 'POST',
 356     path: '/api/' + API_VERSION + '/remote/pods/remove',
 357     toPod: pod
 358   }
 359
 360   return makeSecureRequest(requestParams)
 361     .catch(err => logger.warn('Cannot notify friends %s we are quitting him.', pod.host, err))
 362     .then(() => pod.destroy())
 363     .then(() => logger.info('Removed friend %s.', pod.host))
 364     .catch(err => logger.error('Cannot destroy friend %s.', pod.host, err))
 365 }
 366
 367 function getRequestScheduler () {
 368   return requestScheduler
 369 }
 370
 371 function getRequestVideoQaduScheduler () {
 372   return requestVideoQaduScheduler
 373 }
 374
 375 function getRequestVideoEventScheduler () {
 376   return requestVideoEventScheduler
 377 }
 378
 379 // ---------------------------------------------------------------------------
 380
 381 export {
 382   activateSchedulers,
 383   addVideoToFriends,
 384   removeVideoAuthorToFriends,
 385   updateVideoToFriends,
 386   addVideoAuthorToFriends,
 387   reportAbuseVideoToFriend,
 388   quickAndDirtyUpdateVideoToFriends,
 389   quickAndDirtyUpdatesVideoToFriends,
 390   addEventToRemoteVideo,
 391   addEventsToRemoteVideo,
 392   hasFriends,
 393   makeFriends,
 394   quitFriends,
 395   removeFriend,
 396   removeVideoToFriends,
 397   sendOwnedDataToPod,
 398   getRequestScheduler,
 399   getRequestVideoQaduScheduler,
 400   getRequestVideoEventScheduler,
 401   fetchRemotePreview,
 402   addVideoChannelToFriends,
 403   updateVideoChannelToFriends,
 404   removeVideoChannelToFriends
 405 }
 406
 407 // ---------------------------------------------------------------------------
 408
 409 function computeForeignPodsList (host: string, podsScore: { [ host: string ]: number }) {
 410   // TODO: type res
 411   return getForeignPodsList(host).then(res => {
 412     const foreignPodsList: { host: string }[] = res.data
 413
 414     // Let's give 1 point to the pod we ask the friends list
 415     foreignPodsList.push({ host })
 416
 417     foreignPodsList.forEach(foreignPod => {
 418       const foreignPodHost = foreignPod.host
 419
 420       if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 421       else podsScore[foreignPodHost] = 1
 422     })
 423
 424     return undefined
 425   })
 426 }
 427
 428 function computeWinningPods (hosts: string[], podsScore: { [ host: string ]: number }) {
 429   // Build the list of pods to add
 430   // Only add a pod if it exists in more than a half base pods
 431   const podsList = []
 432   const baseScore = hosts.length / 2
 433
 434   Object.keys(podsScore).forEach(podHost => {
 435     // If the pod is not me and with a good score we add it
 436     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 437       podsList.push({ host: podHost })
 438     }
 439   })
 440
 441   return podsList
 442 }
 443
 444 function getForeignPodsList (host: string) {
 445   return new Promise< ResultList<FormattedPod> >((res, rej) => {
 446     const path = '/api/' + API_VERSION + '/remote/pods/list'
 447
 448     request.post(REMOTE_SCHEME.HTTP + '://' + host + path, (err, response, body) => {
 449       if (err) return rej(err)
 450
 451       try {
 452         const json = JSON.parse(body)
 453         return res(json)
 454       } catch (err) {
 455         return rej(err)
 456       }
 457     })
 458   })
 459 }
 460
 461 function makeRequestsToWinningPods (cert: string, podsList: PodInstance[]) {
 462   // Stop pool requests
 463   requestScheduler.deactivate()
 464   // Flush pool requests
 465   requestScheduler.forceSend()
 466
 467   return Promise.map(podsList, pod => {
 468     const params = {
 469       url: REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + API_VERSION + '/remote/pods/add',
 470       method: 'POST' as 'POST',
 471       json: {
 472         host: CONFIG.WEBSERVER.HOST,
 473         email: CONFIG.ADMIN.EMAIL,
 474         publicKey: cert
 475       }
 476     }
 477
 478     return makeRetryRequest(params)
 479       .then(({ response, body }) => {
 480         body = body as { cert: string, email: string }
 481
 482         if (response.statusCode === 200) {
 483           const podObj = db.Pod.build({ host: pod.host, publicKey: body.cert, email: body.email })
 484           return podObj.save()
 485             .then(podCreated => {
 486
 487               // Add our videos to the request scheduler
 488               sendOwnedDataToPod(podCreated.id)
 489             })
 490             .catch(err => {
 491               logger.error('Cannot add friend %s pod.', pod.host, err)
 492             })
 493         } else {
 494           logger.error('Status not 200 for %s pod.', pod.host)
 495         }
 496       })
 497       .catch(err => {
 498         logger.error('Error with adding %s pod.', pod.host, { error: err.stack })
 499         // Don't break the process
 500       })
 501   }, { concurrency: REQUESTS_IN_PARALLEL })
 502   .then(() => logger.debug('makeRequestsToWinningPods finished.'))
 503   .finally(() => {
 504     // Final callback, we've ended all the requests
 505     // Now we made new friends, we can re activate the pool of requests
 506     requestScheduler.activate()
 507   })
 508 }
 509
 510 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 511 type CreateRequestOptions = {
 512   type: RemoteVideoRequestType
 513   endpoint: RequestEndpoint
 514   data: Object
 515   toIds?: number[]
 516   transaction: Sequelize.Transaction
 517 }
 518 function createRequest (options: CreateRequestOptions) {
 519   if (options.toIds !== undefined) return requestScheduler.createRequest(options as RequestSchedulerOptions)
 520
 521   // If the "toIds" pods is not specified, we send the request to all our friends
 522   return db.Pod.listAllIds(options.transaction).then(podIds => {
 523     const newOptions = Object.assign(options, { toIds: podIds })
 524     return requestScheduler.createRequest(newOptions)
 525   })
 526 }
 527
 528 function createVideoQaduRequest (options: RequestVideoQaduSchedulerOptions) {
 529   return requestVideoQaduScheduler.createRequest(options)
 530 }
 531
 532 function createVideoEventRequest (options: RequestVideoEventSchedulerOptions) {
 533   return requestVideoEventScheduler.createRequest(options)
 534 }
 535
 536 function isMe (host: string) {
 537   return host === CONFIG.WEBSERVER.HOST
 538 }