server/lib/friends.ts

   1 import * as request from 'request'
   2 import * as Sequelize from 'sequelize'
   3 import * as Bluebird 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   for (const eventParams of eventsParams) {
 192     tasks.push(addEventToRemoteVideo(eventParams, transaction))
 193   }
 194
 195   return Promise.all(tasks)
 196 }
 197
 198 async function hasFriends () {
 199   const count = await db.Pod.countAll()
 200
 201   return count !== 0
 202 }
 203
 204 async function makeFriends (hosts: string[]) {
 205   const podsScore = {}
 206
 207   logger.info('Make friends!')
 208   const cert = await getMyPublicCert()
 209
 210   for (const host of hosts) {
 211     await computeForeignPodsList(host, podsScore)
 212   }
 213
 214   logger.debug('Pods scores computed.', { podsScore: podsScore })
 215
 216   const podsList = computeWinningPods(hosts, podsScore)
 217   logger.debug('Pods that we keep.', { podsToKeep: podsList })
 218
 219   return makeRequestsToWinningPods(cert, podsList)
 220 }
 221
 222 async function quitFriends () {
 223   // Stop pool requests
 224   requestScheduler.deactivate()
 225
 226   try {
 227     await requestScheduler.flush()
 228
 229     await requestVideoQaduScheduler.flush()
 230
 231     const pods = await db.Pod.list()
 232     const requestParams = {
 233       method: 'POST' as 'POST',
 234       path: '/api/' + API_VERSION + '/remote/pods/remove',
 235       toPod: null
 236     }
 237
 238     // Announce we quit them
 239     // We don't care if the request fails
 240     // The other pod will exclude us automatically after a while
 241     try {
 242       await Bluebird.map(pods, pod => {
 243         requestParams.toPod = pod
 244
 245         return makeSecureRequest(requestParams)
 246       }, { concurrency: REQUESTS_IN_PARALLEL })
 247     } catch (err) { // Don't stop the process
 248       logger.error('Some errors while quitting friends.', err)
 249     }
 250
 251     const tasks = []
 252     for (const pod of pods) {
 253       tasks.push(pod.destroy())
 254     }
 255     await Promise.all(pods)
 256
 257     logger.info('Removed all remote videos.')
 258
 259     requestScheduler.activate()
 260   } catch (err) {
 261     // Don't forget to re activate the scheduler, even if there was an error
 262     requestScheduler.activate()
 263
 264     throw err
 265   }
 266 }
 267
 268 async function sendOwnedDataToPod (podId: number) {
 269   // First send authors
 270   await sendOwnedAuthorsToPod(podId)
 271   await sendOwnedChannelsToPod(podId)
 272   await sendOwnedVideosToPod(podId)
 273 }
 274
 275 async function sendOwnedChannelsToPod (podId: number) {
 276   const videoChannels = await db.VideoChannel.listOwned()
 277
 278   const tasks: Promise<any>[] = []
 279   for (const videoChannel of videoChannels) {
 280     const remoteVideoChannel = videoChannel.toAddRemoteJSON()
 281     const options = {
 282       type: 'add-channel' as 'add-channel',
 283       endpoint: REQUEST_ENDPOINTS.VIDEOS,
 284       data: remoteVideoChannel,
 285       toIds: [ podId ],
 286       transaction: null
 287     }
 288
 289     const p = createRequest(options)
 290     tasks.push(p)
 291   }
 292
 293   await Promise.all(tasks)
 294 }
 295
 296 async function sendOwnedAuthorsToPod (podId: number) {
 297   const authors = await db.Author.listOwned()
 298   const tasks: Promise<any>[] = []
 299
 300   for (const author of authors) {
 301     const remoteAuthor = author.toAddRemoteJSON()
 302     const options = {
 303       type: 'add-author' as 'add-author',
 304       endpoint: REQUEST_ENDPOINTS.VIDEOS,
 305       data: remoteAuthor,
 306       toIds: [ podId ],
 307       transaction: null
 308     }
 309
 310     const p = createRequest(options)
 311     tasks.push(p)
 312   }
 313
 314   await Promise.all(tasks)
 315 }
 316
 317 async function sendOwnedVideosToPod (podId: number) {
 318   const videosList = await db.Video.listOwnedAndPopulateAuthorAndTags()
 319   const tasks: Bluebird<any>[] = []
 320
 321   for (const video of videosList) {
 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   await Promise.all(tasks)
 343 }
 344
 345 function fetchRemotePreview (video: VideoInstance) {
 346   const host = video.VideoChannel.Author.Pod.host
 347   const path = join(STATIC_PATHS.PREVIEWS, video.getPreviewName())
 348
 349   return request.get(REMOTE_SCHEME.HTTP + '://' + host + path)
 350 }
 351
 352 async function removeFriend (pod: PodInstance) {
 353   const requestParams = {
 354     method: 'POST' as 'POST',
 355     path: '/api/' + API_VERSION + '/remote/pods/remove',
 356     toPod: pod
 357   }
 358
 359   try {
 360     await makeSecureRequest(requestParams)
 361   } catch (err) {
 362     logger.warn('Cannot notify friends %s we are quitting him.', pod.host, err)
 363   }
 364
 365   try {
 366     await pod.destroy()
 367
 368     logger.info('Removed friend %s.', pod.host)
 369   } catch (err) {
 370     logger.error('Cannot destroy friend %s.', pod.host, err)
 371   }
 372 }
 373
 374 function getRequestScheduler () {
 375   return requestScheduler
 376 }
 377
 378 function getRequestVideoQaduScheduler () {
 379   return requestVideoQaduScheduler
 380 }
 381
 382 function getRequestVideoEventScheduler () {
 383   return requestVideoEventScheduler
 384 }
 385
 386 // ---------------------------------------------------------------------------
 387
 388 export {
 389   activateSchedulers,
 390   addVideoToFriends,
 391   removeVideoAuthorToFriends,
 392   updateVideoToFriends,
 393   addVideoAuthorToFriends,
 394   reportAbuseVideoToFriend,
 395   quickAndDirtyUpdateVideoToFriends,
 396   quickAndDirtyUpdatesVideoToFriends,
 397   addEventToRemoteVideo,
 398   addEventsToRemoteVideo,
 399   hasFriends,
 400   makeFriends,
 401   quitFriends,
 402   removeFriend,
 403   removeVideoToFriends,
 404   sendOwnedDataToPod,
 405   getRequestScheduler,
 406   getRequestVideoQaduScheduler,
 407   getRequestVideoEventScheduler,
 408   fetchRemotePreview,
 409   addVideoChannelToFriends,
 410   updateVideoChannelToFriends,
 411   removeVideoChannelToFriends
 412 }
 413
 414 // ---------------------------------------------------------------------------
 415
 416 async function computeForeignPodsList (host: string, podsScore: { [ host: string ]: number }) {
 417   const result = await getForeignPodsList(host)
 418   const foreignPodsList: { host: string }[] = result.data
 419
 420   // Let's give 1 point to the pod we ask the friends list
 421   foreignPodsList.push({ host })
 422
 423   for (const foreignPod of foreignPodsList) {
 424     const foreignPodHost = foreignPod.host
 425
 426     if (podsScore[foreignPodHost]) podsScore[foreignPodHost]++
 427     else podsScore[foreignPodHost] = 1
 428   }
 429
 430   return undefined
 431 }
 432
 433 function computeWinningPods (hosts: string[], podsScore: { [ host: string ]: number }) {
 434   // Build the list of pods to add
 435   // Only add a pod if it exists in more than a half base pods
 436   const podsList = []
 437   const baseScore = hosts.length / 2
 438
 439   for (const podHost of Object.keys(podsScore)) {
 440     // If the pod is not me and with a good score we add it
 441     if (isMe(podHost) === false && podsScore[podHost] > baseScore) {
 442       podsList.push({ host: podHost })
 443     }
 444   }
 445
 446   return podsList
 447 }
 448
 449 function getForeignPodsList (host: string) {
 450   return new Promise< ResultList<FormattedPod> >((res, rej) => {
 451     const path = '/api/' + API_VERSION + '/remote/pods/list'
 452
 453     request.post(REMOTE_SCHEME.HTTP + '://' + host + path, (err, response, body) => {
 454       if (err) return rej(err)
 455
 456       try {
 457         const json: ResultList<FormattedPod> = JSON.parse(body)
 458         return res(json)
 459       } catch (err) {
 460         return rej(err)
 461       }
 462     })
 463   })
 464 }
 465
 466 async function makeRequestsToWinningPods (cert: string, podsList: PodInstance[]) {
 467   // Stop pool requests
 468   requestScheduler.deactivate()
 469   // Flush pool requests
 470   requestScheduler.forceSend()
 471
 472   try {
 473     await Bluebird.map(podsList, async pod => {
 474       const params = {
 475         url: REMOTE_SCHEME.HTTP + '://' + pod.host + '/api/' + API_VERSION + '/remote/pods/add',
 476         method: 'POST' as 'POST',
 477         json: {
 478           host: CONFIG.WEBSERVER.HOST,
 479           email: CONFIG.ADMIN.EMAIL,
 480           publicKey: cert
 481         }
 482       }
 483
 484       const { response, body } = await makeRetryRequest(params)
 485       const typedBody = body as { cert: string, email: string }
 486
 487       if (response.statusCode === 200) {
 488         const podObj = db.Pod.build({ host: pod.host, publicKey: typedBody.cert, email: typedBody.email })
 489
 490         let podCreated: PodInstance
 491         try {
 492           podCreated = await podObj.save()
 493         } catch (err) {
 494           logger.error('Cannot add friend %s pod.', pod.host, err)
 495         }
 496
 497         // Add our videos to the request scheduler
 498         sendOwnedDataToPod(podCreated.id)
 499           .catch(err => logger.warn('Cannot send owned data to pod %d.', podCreated.id, err))
 500       } else {
 501         logger.error('Status not 200 for %s pod.', pod.host)
 502       }
 503     }, { concurrency: REQUESTS_IN_PARALLEL })
 504
 505     logger.debug('makeRequestsToWinningPods finished.')
 506
 507     requestScheduler.activate()
 508   } catch (err) {
 509     // Final callback, we've ended all the requests
 510     // Now we made new friends, we can re activate the pool of requests
 511     requestScheduler.activate()
 512   }
 513 }
 514
 515 // Wrapper that populate "toIds" argument with all our friends if it is not specified
 516 type CreateRequestOptions = {
 517   type: RemoteVideoRequestType
 518   endpoint: RequestEndpoint
 519   data: Object
 520   toIds?: number[]
 521   transaction: Sequelize.Transaction
 522 }
 523 async function createRequest (options: CreateRequestOptions) {
 524   if (options.toIds !== undefined) {
 525     await requestScheduler.createRequest(options as RequestSchedulerOptions)
 526     return undefined
 527   }
 528
 529   // If the "toIds" pods is not specified, we send the request to all our friends
 530   const podIds = await db.Pod.listAllIds(options.transaction)
 531
 532   const newOptions = Object.assign(options, { toIds: podIds })
 533   await requestScheduler.createRequest(newOptions)
 534
 535   return undefined
 536 }
 537
 538 function createVideoQaduRequest (options: RequestVideoQaduSchedulerOptions) {
 539   return requestVideoQaduScheduler.createRequest(options)
 540 }
 541
 542 function createVideoEventRequest (options: RequestVideoEventSchedulerOptions) {
 543   return requestVideoEventScheduler.createRequest(options)
 544 }
 545
 546 function isMe (host: string) {
 547   return host === CONFIG.WEBSERVER.HOST
 548 }