server/lib/activitypub/videos.ts

   1 import * as Bluebird from 'bluebird'
   2 import * as sequelize from 'sequelize'
   3 import * as magnetUtil from 'magnet-uri'
   4 import { join } from 'path'
   5 import * as request from 'request'
   6 import { ActivityIconObject, ActivityUrlObject, ActivityVideoUrlObject, VideoState } from '../../../shared/index'
   7 import { VideoTorrentObject } from '../../../shared/models/activitypub/objects'
   8 import { VideoPrivacy } from '../../../shared/models/videos'
   9 import { sanitizeAndCheckVideoTorrentObject } from '../../helpers/custom-validators/activitypub/videos'
  10 import { isVideoFileInfoHashValid } from '../../helpers/custom-validators/videos'
  11 import { resetSequelizeInstance, retryTransactionWrapper } from '../../helpers/database-utils'
  12 import { logger } from '../../helpers/logger'
  13 import { doRequest, doRequestAndSaveToFile } from '../../helpers/requests'
  14 import { ACTIVITY_PUB, CONFIG, REMOTE_SCHEME, sequelizeTypescript, VIDEO_MIMETYPE_EXT } from '../../initializers'
  15 import { ActorModel } from '../../models/activitypub/actor'
  16 import { TagModel } from '../../models/video/tag'
  17 import { VideoModel } from '../../models/video/video'
  18 import { VideoChannelModel } from '../../models/video/video-channel'
  19 import { VideoFileModel } from '../../models/video/video-file'
  20 import { getOrCreateActorAndServerAndModel } from './actor'
  21 import { addVideoComments } from './video-comments'
  22 import { crawlCollectionPage } from './crawl'
  23 import { sendCreateVideo, sendUpdateVideo } from './send'
  24 import { isArray } from '../../helpers/custom-validators/misc'
  25 import { VideoCaptionModel } from '../../models/video/video-caption'
  26 import { JobQueue } from '../job-queue'
  27 import { ActivitypubHttpFetcherPayload } from '../job-queue/handlers/activitypub-http-fetcher'
  28 import { createRates } from './video-rates'
  29 import { addVideoShares, shareVideoByServerAndChannel } from './share'
  30 import { AccountModel } from '../../models/account/account'
  31 import { fetchVideoByUrl, VideoFetchByUrlType } from '../../helpers/video'
  32
  33 async function federateVideoIfNeeded (video: VideoModel, isNewVideo: boolean, transaction?: sequelize.Transaction) {
  34   // If the video is not private and published, we federate it
  35   if (video.privacy !== VideoPrivacy.PRIVATE && video.state === VideoState.PUBLISHED) {
  36     // Fetch more attributes that we will need to serialize in AP object
  37     if (isArray(video.VideoCaptions) === false) {
  38       video.VideoCaptions = await video.$get('VideoCaptions', {
  39         attributes: [ 'language' ],
  40         transaction
  41       }) as VideoCaptionModel[]
  42     }
  43
  44     if (isNewVideo) {
  45       // Now we'll add the video's meta data to our followers
  46       await sendCreateVideo(video, transaction)
  47       await shareVideoByServerAndChannel(video, transaction)
  48     } else {
  49       await sendUpdateVideo(video, transaction)
  50     }
  51   }
  52 }
  53
  54 async function fetchRemoteVideo (videoUrl: string): Promise<{ response: request.RequestResponse, videoObject: VideoTorrentObject }> {
  55   const options = {
  56     uri: videoUrl,
  57     method: 'GET',
  58     json: true,
  59     activityPub: true
  60   }
  61
  62   logger.info('Fetching remote video %s.', videoUrl)
  63
  64   const { response, body } = await doRequest(options)
  65
  66   if (sanitizeAndCheckVideoTorrentObject(body) === false) {
  67     logger.debug('Remote video JSON is not valid.', { body })
  68     return { response, videoObject: undefined }
  69   }
  70
  71   return { response, videoObject: body }
  72 }
  73
  74 async function fetchRemoteVideoDescription (video: VideoModel) {
  75   const host = video.VideoChannel.Account.Actor.Server.host
  76   const path = video.getDescriptionAPIPath()
  77   const options = {
  78     uri: REMOTE_SCHEME.HTTP + '://' + host + path,
  79     json: true
  80   }
  81
  82   const { body } = await doRequest(options)
  83   return body.description ? body.description : ''
  84 }
  85
  86 function fetchRemoteVideoStaticFile (video: VideoModel, path: string, reject: Function) {
  87   const host = video.VideoChannel.Account.Actor.Server.host
  88
  89   // We need to provide a callback, if no we could have an uncaught exception
  90   return request.get(REMOTE_SCHEME.HTTP + '://' + host + path, err => {
  91     if (err) reject(err)
  92   })
  93 }
  94
  95 function generateThumbnailFromUrl (video: VideoModel, icon: ActivityIconObject) {
  96   const thumbnailName = video.getThumbnailName()
  97   const thumbnailPath = join(CONFIG.STORAGE.THUMBNAILS_DIR, thumbnailName)
  98
  99   const options = {
 100     method: 'GET',
 101     uri: icon.url
 102   }
 103   return doRequestAndSaveToFile(options, thumbnailPath)
 104 }
 105
 106 function getOrCreateVideoChannelFromVideoObject (videoObject: VideoTorrentObject) {
 107   const channel = videoObject.attributedTo.find(a => a.type === 'Group')
 108   if (!channel) throw new Error('Cannot find associated video channel to video ' + videoObject.url)
 109
 110   return getOrCreateActorAndServerAndModel(channel.id)
 111 }
 112
 113 type SyncParam = {
 114   likes: boolean
 115   dislikes: boolean
 116   shares: boolean
 117   comments: boolean
 118   thumbnail: boolean
 119   refreshVideo: boolean
 120 }
 121 async function syncVideoExternalAttributes (video: VideoModel, fetchedVideo: VideoTorrentObject, syncParam: SyncParam) {
 122   logger.info('Adding likes/dislikes/shares/comments of video %s.', video.uuid)
 123
 124   const jobPayloads: ActivitypubHttpFetcherPayload[] = []
 125
 126   if (syncParam.likes === true) {
 127     await crawlCollectionPage<string>(fetchedVideo.likes, items => createRates(items, video, 'like'))
 128       .catch(err => logger.error('Cannot add likes of video %s.', video.uuid, { err }))
 129   } else {
 130     jobPayloads.push({ uri: fetchedVideo.likes, videoId: video.id, type: 'video-likes' as 'video-likes' })
 131   }
 132
 133   if (syncParam.dislikes === true) {
 134     await crawlCollectionPage<string>(fetchedVideo.dislikes, items => createRates(items, video, 'dislike'))
 135       .catch(err => logger.error('Cannot add dislikes of video %s.', video.uuid, { err }))
 136   } else {
 137     jobPayloads.push({ uri: fetchedVideo.dislikes, videoId: video.id, type: 'video-dislikes' as 'video-dislikes' })
 138   }
 139
 140   if (syncParam.shares === true) {
 141     await crawlCollectionPage<string>(fetchedVideo.shares, items => addVideoShares(items, video))
 142       .catch(err => logger.error('Cannot add shares of video %s.', video.uuid, { err }))
 143   } else {
 144     jobPayloads.push({ uri: fetchedVideo.shares, videoId: video.id, type: 'video-shares' as 'video-shares' })
 145   }
 146
 147   if (syncParam.comments === true) {
 148     await crawlCollectionPage<string>(fetchedVideo.comments, items => addVideoComments(items, video))
 149       .catch(err => logger.error('Cannot add comments of video %s.', video.uuid, { err }))
 150   } else {
 151     jobPayloads.push({ uri: fetchedVideo.shares, videoId: video.id, type: 'video-shares' as 'video-shares' })
 152   }
 153
 154   await Bluebird.map(jobPayloads, payload => JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload }))
 155 }
 156
 157 async function getOrCreateVideoAndAccountAndChannel (options: {
 158   videoObject: VideoTorrentObject | string,
 159   syncParam?: SyncParam,
 160   fetchType?: VideoFetchByUrlType
 161 }) {
 162   // Default params
 163   const syncParam = options.syncParam || { likes: true, dislikes: true, shares: true, comments: true, thumbnail: true, refreshVideo: false }
 164   const fetchType = options.fetchType || 'all'
 165
 166   // Get video url
 167   const videoUrl = typeof options.videoObject === 'string' ? options.videoObject : options.videoObject.id
 168
 169   let videoFromDatabase = await fetchVideoByUrl(videoUrl, fetchType)
 170   if (videoFromDatabase) {
 171     const p = retryTransactionWrapper(refreshVideoIfNeeded, videoFromDatabase, fetchType, syncParam)
 172     if (syncParam.refreshVideo === true) videoFromDatabase = await p
 173
 174     return { video: videoFromDatabase }
 175   }
 176
 177   const { videoObject: fetchedVideo } = await fetchRemoteVideo(videoUrl)
 178   if (!fetchedVideo) throw new Error('Cannot fetch remote video with url: ' + videoUrl)
 179
 180   const channelActor = await getOrCreateVideoChannelFromVideoObject(fetchedVideo)
 181   const video = await retryTransactionWrapper(createVideo, fetchedVideo, channelActor, syncParam.thumbnail)
 182
 183   await syncVideoExternalAttributes(video, fetchedVideo, syncParam)
 184
 185   return { video }
 186 }
 187
 188 async function updateVideoFromAP (
 189   video: VideoModel,
 190   videoObject: VideoTorrentObject,
 191   account: AccountModel,
 192   channel: VideoChannelModel,
 193   overrideTo?: string[]
 194 ) {
 195   logger.debug('Updating remote video "%s".', videoObject.uuid)
 196   let videoFieldsSave: any
 197
 198   try {
 199     const updatedVideo: VideoModel = await sequelizeTypescript.transaction(async t => {
 200       const sequelizeOptions = {
 201         transaction: t
 202       }
 203
 204       videoFieldsSave = video.toJSON()
 205
 206       // Check actor has the right to update the video
 207       const videoChannel = video.VideoChannel
 208       if (videoChannel.Account.id !== account.id) {
 209         throw new Error('Account ' + account.Actor.url + ' does not own video channel ' + videoChannel.Actor.url)
 210       }
 211
 212       const to = overrideTo ? overrideTo : videoObject.to
 213       const videoData = await videoActivityObjectToDBAttributes(channel, videoObject, to)
 214       video.set('name', videoData.name)
 215       video.set('uuid', videoData.uuid)
 216       video.set('url', videoData.url)
 217       video.set('category', videoData.category)
 218       video.set('licence', videoData.licence)
 219       video.set('language', videoData.language)
 220       video.set('description', videoData.description)
 221       video.set('support', videoData.support)
 222       video.set('nsfw', videoData.nsfw)
 223       video.set('commentsEnabled', videoData.commentsEnabled)
 224       video.set('waitTranscoding', videoData.waitTranscoding)
 225       video.set('state', videoData.state)
 226       video.set('duration', videoData.duration)
 227       video.set('createdAt', videoData.createdAt)
 228       video.set('publishedAt', videoData.publishedAt)
 229       video.set('views', videoData.views)
 230       video.set('privacy', videoData.privacy)
 231       video.set('channelId', videoData.channelId)
 232
 233       await video.save(sequelizeOptions)
 234
 235       // Don't block on request
 236       generateThumbnailFromUrl(video, videoObject.icon)
 237         .catch(err => logger.warn('Cannot generate thumbnail of %s.', videoObject.id, { err }))
 238
 239       // Remove old video files
 240       const videoFileDestroyTasks: Bluebird<void>[] = []
 241       for (const videoFile of video.VideoFiles) {
 242         videoFileDestroyTasks.push(videoFile.destroy(sequelizeOptions))
 243       }
 244       await Promise.all(videoFileDestroyTasks)
 245
 246       const videoFileAttributes = videoFileActivityUrlToDBAttributes(video, videoObject)
 247       const tasks = videoFileAttributes.map(f => VideoFileModel.create(f, sequelizeOptions))
 248       await Promise.all(tasks)
 249
 250       // Update Tags
 251       const tags = videoObject.tag.map(tag => tag.name)
 252       const tagInstances = await TagModel.findOrCreateTags(tags, t)
 253       await video.$set('Tags', tagInstances, sequelizeOptions)
 254
 255       // Update captions
 256       await VideoCaptionModel.deleteAllCaptionsOfRemoteVideo(video.id, t)
 257
 258       const videoCaptionsPromises = videoObject.subtitleLanguage.map(c => {
 259         return VideoCaptionModel.insertOrReplaceLanguage(video.id, c.identifier, t)
 260       })
 261       await Promise.all(videoCaptionsPromises)
 262     })
 263
 264     logger.info('Remote video with uuid %s updated', videoObject.uuid)
 265
 266     return updatedVideo
 267   } catch (err) {
 268     if (video !== undefined && videoFieldsSave !== undefined) {
 269       resetSequelizeInstance(video, videoFieldsSave)
 270     }
 271
 272     // This is just a debug because we will retry the insert
 273     logger.debug('Cannot update the remote video.', { err })
 274     throw err
 275   }
 276 }
 277
 278 export {
 279   updateVideoFromAP,
 280   federateVideoIfNeeded,
 281   fetchRemoteVideo,
 282   getOrCreateVideoAndAccountAndChannel,
 283   fetchRemoteVideoStaticFile,
 284   fetchRemoteVideoDescription,
 285   generateThumbnailFromUrl,
 286   getOrCreateVideoChannelFromVideoObject
 287 }
 288
 289 // ---------------------------------------------------------------------------
 290
 291 function isActivityVideoUrlObject (url: ActivityUrlObject): url is ActivityVideoUrlObject {
 292   const mimeTypes = Object.keys(VIDEO_MIMETYPE_EXT)
 293
 294   return mimeTypes.indexOf(url.mimeType) !== -1 && url.mimeType.startsWith('video/')
 295 }
 296
 297 async function createVideo (videoObject: VideoTorrentObject, channelActor: ActorModel, waitThumbnail = false) {
 298   logger.debug('Adding remote video %s.', videoObject.id)
 299
 300   const videoCreated: VideoModel = await sequelizeTypescript.transaction(async t => {
 301     const sequelizeOptions = { transaction: t }
 302
 303     const videoData = await videoActivityObjectToDBAttributes(channelActor.VideoChannel, videoObject, videoObject.to)
 304     const video = VideoModel.build(videoData)
 305
 306     const videoCreated = await video.save(sequelizeOptions)
 307
 308     // Process files
 309     const videoFileAttributes = videoFileActivityUrlToDBAttributes(videoCreated, videoObject)
 310     if (videoFileAttributes.length === 0) {
 311       throw new Error('Cannot find valid files for video %s ' + videoObject.url)
 312     }
 313
 314     const videoFilePromises = videoFileAttributes.map(f => VideoFileModel.create(f, { transaction: t }))
 315     await Promise.all(videoFilePromises)
 316
 317     // Process tags
 318     const tags = videoObject.tag.map(t => t.name)
 319     const tagInstances = await TagModel.findOrCreateTags(tags, t)
 320     await videoCreated.$set('Tags', tagInstances, sequelizeOptions)
 321
 322     // Process captions
 323     const videoCaptionsPromises = videoObject.subtitleLanguage.map(c => {
 324       return VideoCaptionModel.insertOrReplaceLanguage(videoCreated.id, c.identifier, t)
 325     })
 326     await Promise.all(videoCaptionsPromises)
 327
 328     logger.info('Remote video with uuid %s inserted.', videoObject.uuid)
 329
 330     videoCreated.VideoChannel = channelActor.VideoChannel
 331     return videoCreated
 332   })
 333
 334   const p = generateThumbnailFromUrl(videoCreated, videoObject.icon)
 335     .catch(err => logger.warn('Cannot generate thumbnail of %s.', videoObject.id, { err }))
 336
 337   if (waitThumbnail === true) await p
 338
 339   return videoCreated
 340 }
 341
 342 async function refreshVideoIfNeeded (videoArg: VideoModel, fetchedType: VideoFetchByUrlType, syncParam: SyncParam): Promise<VideoModel> {
 343   // We need more attributes if the argument video was fetched with not enough joints
 344   const video = fetchedType === 'all' ? videoArg : await VideoModel.loadByUrlAndPopulateAccount(videoArg.url)
 345
 346   if (!video.isOutdated()) return video
 347
 348   try {
 349     const { response, videoObject } = await fetchRemoteVideo(video.url)
 350     if (response.statusCode === 404) {
 351       // Video does not exist anymore
 352       await video.destroy()
 353       return undefined
 354     }
 355
 356     if (videoObject === undefined) {
 357       logger.warn('Cannot refresh remote video: invalid body.')
 358       return video
 359     }
 360
 361     const channelActor = await getOrCreateVideoChannelFromVideoObject(videoObject)
 362     const account = await AccountModel.load(channelActor.VideoChannel.accountId)
 363
 364     await updateVideoFromAP(video, videoObject, account, channelActor.VideoChannel)
 365     await syncVideoExternalAttributes(video, videoObject, syncParam)
 366   } catch (err) {
 367     logger.warn('Cannot refresh video.', { err })
 368     return video
 369   }
 370 }
 371
 372 async function videoActivityObjectToDBAttributes (
 373   videoChannel: VideoChannelModel,
 374   videoObject: VideoTorrentObject,
 375   to: string[] = []
 376 ) {
 377   const privacy = to.indexOf(ACTIVITY_PUB.PUBLIC) !== -1 ? VideoPrivacy.PUBLIC : VideoPrivacy.UNLISTED
 378   const duration = videoObject.duration.replace(/[^\d]+/, '')
 379
 380   let language: string | undefined
 381   if (videoObject.language) {
 382     language = videoObject.language.identifier
 383   }
 384
 385   let category: number | undefined
 386   if (videoObject.category) {
 387     category = parseInt(videoObject.category.identifier, 10)
 388   }
 389
 390   let licence: number | undefined
 391   if (videoObject.licence) {
 392     licence = parseInt(videoObject.licence.identifier, 10)
 393   }
 394
 395   const description = videoObject.content || null
 396   const support = videoObject.support || null
 397
 398   return {
 399     name: videoObject.name,
 400     uuid: videoObject.uuid,
 401     url: videoObject.id,
 402     category,
 403     licence,
 404     language,
 405     description,
 406     support,
 407     nsfw: videoObject.sensitive,
 408     commentsEnabled: videoObject.commentsEnabled,
 409     waitTranscoding: videoObject.waitTranscoding,
 410     state: videoObject.state,
 411     channelId: videoChannel.id,
 412     duration: parseInt(duration, 10),
 413     createdAt: new Date(videoObject.published),
 414     publishedAt: new Date(videoObject.published),
 415     // FIXME: updatedAt does not seems to be considered by Sequelize
 416     updatedAt: new Date(videoObject.updated),
 417     views: videoObject.views,
 418     likes: 0,
 419     dislikes: 0,
 420     remote: true,
 421     privacy
 422   }
 423 }
 424
 425 function videoFileActivityUrlToDBAttributes (videoCreated: VideoModel, videoObject: VideoTorrentObject) {
 426   const fileUrls = videoObject.url.filter(u => isActivityVideoUrlObject(u)) as ActivityVideoUrlObject[]
 427
 428   if (fileUrls.length === 0) {
 429     throw new Error('Cannot find video files for ' + videoCreated.url)
 430   }
 431
 432   const attributes: VideoFileModel[] = []
 433   for (const fileUrl of fileUrls) {
 434     // Fetch associated magnet uri
 435     const magnet = videoObject.url.find(u => {
 436       return u.mimeType === 'application/x-bittorrent;x-scheme-handler/magnet' && u.height === fileUrl.height
 437     })
 438
 439     if (!magnet) throw new Error('Cannot find associated magnet uri for file ' + fileUrl.href)
 440
 441     const parsed = magnetUtil.decode(magnet.href)
 442     if (!parsed || isVideoFileInfoHashValid(parsed.infoHash) === false) {
 443       throw new Error('Cannot parse magnet URI ' + magnet.href)
 444     }
 445
 446     const attribute = {
 447       extname: VIDEO_MIMETYPE_EXT[ fileUrl.mimeType ],
 448       infoHash: parsed.infoHash,
 449       resolution: fileUrl.height,
 450       size: fileUrl.size,
 451       videoId: videoCreated.id,
 452       fps: fileUrl.fps
 453     } as VideoFileModel
 454     attributes.push(attribute)
 455   }
 456
 457   return attributes
 458 }