From 2186386cca113506791583cb07d6ccacba7af4e0 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Tue, 12 Jun 2018 20:04:58 +0200 Subject: Add concept of video state, and add ability to wait transcoding before publishing a video --- server/lib/activitypub/audience.ts | 10 +- server/lib/activitypub/crawl.ts | 2 +- server/lib/activitypub/process/process-update.ts | 27 +++-- server/lib/activitypub/send/send-announce.ts | 14 +-- server/lib/activitypub/send/send-create.ts | 43 ++++---- server/lib/activitypub/send/send-like.ts | 33 +++--- server/lib/activitypub/send/send-undo.ts | 42 ++++---- server/lib/activitypub/send/send-update.ts | 36 +++---- server/lib/activitypub/videos.ts | 80 ++++++++------ server/lib/job-queue/handlers/video-file.ts | 127 ++++++++++++----------- server/lib/job-queue/job-queue.ts | 1 + 11 files changed, 215 insertions(+), 200 deletions(-) (limited to 'server/lib') diff --git a/server/lib/activitypub/audience.ts b/server/lib/activitypub/audience.ts index c1265dbcd..7164135b6 100644 --- a/server/lib/activitypub/audience.ts +++ b/server/lib/activitypub/audience.ts @@ -20,7 +20,7 @@ function getVideoCommentAudience ( isOrigin = false ) { const to = [ ACTIVITY_PUB.PUBLIC ] - const cc = [ ] + const cc = [] // Owner of the video we comment if (isOrigin === false) { @@ -55,7 +55,7 @@ async function getActorsInvolvedInVideo (video: VideoModel, t: Transaction) { return actors } -async function getAudience (actorSender: ActorModel, t: Transaction, isPublic = true) { +function getAudience (actorSender: ActorModel, isPublic = true) { return buildAudience([ actorSender.followersUrl ], isPublic) } @@ -67,14 +67,14 @@ function buildAudience (followerUrls: string[], isPublic = true) { to = [ ACTIVITY_PUB.PUBLIC ] cc = followerUrls } else { // Unlisted - to = [ ] - cc = [ ] + to = [] + cc = [] } return { to, cc } } -function audiencify (object: T, audience: ActivityAudience) { +function audiencify (object: T, audience: ActivityAudience) { return Object.assign(object, audience) } diff --git a/server/lib/activitypub/crawl.ts b/server/lib/activitypub/crawl.ts index 7305b3969..d4fc786f7 100644 --- a/server/lib/activitypub/crawl.ts +++ b/server/lib/activitypub/crawl.ts @@ -28,7 +28,7 @@ async function crawlCollectionPage (uri: string, handler: (items: T[]) => Pr if (Array.isArray(body.orderedItems)) { const items = body.orderedItems - logger.info('Processing %i ActivityPub items for %s.', items.length, nextLink) + logger.info('Processing %i ActivityPub items for %s.', items.length, options.uri) await handler(items) } diff --git a/server/lib/activitypub/process/process-update.ts b/server/lib/activitypub/process/process-update.ts index 2750f48c3..77de8c155 100644 --- a/server/lib/activitypub/process/process-update.ts +++ b/server/lib/activitypub/process/process-update.ts @@ -1,7 +1,6 @@ import * as Bluebird from 'bluebird' import { ActivityUpdate } from '../../../../shared/models/activitypub' import { ActivityPubActor } from '../../../../shared/models/activitypub/activitypub-actor' -import { VideoTorrentObject } from '../../../../shared/models/activitypub/objects' import { retryTransactionWrapper } from '../../../helpers/database-utils' import { logger } from '../../../helpers/logger' import { resetSequelizeInstance } from '../../../helpers/utils' @@ -13,6 +12,7 @@ import { VideoChannelModel } from '../../../models/video/video-channel' import { VideoFileModel } from '../../../models/video/video-file' import { fetchAvatarIfExists, getOrCreateActorAndServerAndModel, updateActorAvatarInstance, updateActorInstance } from '../actor' import { + fetchRemoteVideo, generateThumbnailFromUrl, getOrCreateAccountAndVideoAndChannel, getOrCreateVideoChannel, @@ -51,15 +51,18 @@ function processUpdateVideo (actor: ActorModel, activity: ActivityUpdate) { } async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) { - const videoAttributesToUpdate = activity.object as VideoTorrentObject + const videoUrl = activity.object.id - const res = await getOrCreateAccountAndVideoAndChannel(videoAttributesToUpdate.id) + const videoObject = await fetchRemoteVideo(videoUrl) + if (!videoObject) throw new Error('Cannot fetch remote video with url: ' + videoUrl) + + const res = await getOrCreateAccountAndVideoAndChannel(videoObject.id) // Fetch video channel outside the transaction - const newVideoChannelActor = await getOrCreateVideoChannel(videoAttributesToUpdate) + const newVideoChannelActor = await getOrCreateVideoChannel(videoObject) const newVideoChannel = newVideoChannelActor.VideoChannel - logger.debug('Updating remote video "%s".', videoAttributesToUpdate.uuid) + logger.debug('Updating remote video "%s".', videoObject.uuid) let videoInstance = res.video let videoFieldsSave: any @@ -77,7 +80,7 @@ async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) { throw new Error('Account ' + actor.url + ' does not own video channel ' + videoChannel.Actor.url) } - const videoData = await videoActivityObjectToDBAttributes(newVideoChannel, videoAttributesToUpdate, activity.to) + const videoData = await videoActivityObjectToDBAttributes(newVideoChannel, videoObject, activity.to) videoInstance.set('name', videoData.name) videoInstance.set('uuid', videoData.uuid) videoInstance.set('url', videoData.url) @@ -88,6 +91,8 @@ async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) { videoInstance.set('support', videoData.support) videoInstance.set('nsfw', videoData.nsfw) videoInstance.set('commentsEnabled', videoData.commentsEnabled) + videoInstance.set('waitTranscoding', videoData.waitTranscoding) + videoInstance.set('state', videoData.state) videoInstance.set('duration', videoData.duration) videoInstance.set('createdAt', videoData.createdAt) videoInstance.set('updatedAt', videoData.updatedAt) @@ -98,8 +103,8 @@ async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) { await videoInstance.save(sequelizeOptions) // Don't block on request - generateThumbnailFromUrl(videoInstance, videoAttributesToUpdate.icon) - .catch(err => logger.warn('Cannot generate thumbnail of %s.', videoAttributesToUpdate.id, { err })) + generateThumbnailFromUrl(videoInstance, videoObject.icon) + .catch(err => logger.warn('Cannot generate thumbnail of %s.', videoObject.id, { err })) // Remove old video files const videoFileDestroyTasks: Bluebird[] = [] @@ -108,16 +113,16 @@ async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) { } await Promise.all(videoFileDestroyTasks) - const videoFileAttributes = videoFileActivityUrlToDBAttributes(videoInstance, videoAttributesToUpdate) + const videoFileAttributes = videoFileActivityUrlToDBAttributes(videoInstance, videoObject) const tasks = videoFileAttributes.map(f => VideoFileModel.create(f)) await Promise.all(tasks) - const tags = videoAttributesToUpdate.tag.map(t => t.name) + const tags = videoObject.tag.map(t => t.name) const tagInstances = await TagModel.findOrCreateTags(tags, t) await videoInstance.$set('Tags', tagInstances, sequelizeOptions) }) - logger.info('Remote video with uuid %s updated', videoAttributesToUpdate.uuid) + logger.info('Remote video with uuid %s updated', videoObject.uuid) } catch (err) { if (videoInstance !== undefined && videoFieldsSave !== undefined) { resetSequelizeInstance(videoInstance, videoFieldsSave) diff --git a/server/lib/activitypub/send/send-announce.ts b/server/lib/activitypub/send/send-announce.ts index fa1d47259..dfc099ff2 100644 --- a/server/lib/activitypub/send/send-announce.ts +++ b/server/lib/activitypub/send/send-announce.ts @@ -11,7 +11,7 @@ async function buildVideoAnnounce (byActor: ActorModel, videoShare: VideoShareMo const accountsToForwardView = await getActorsInvolvedInVideo(video, t) const audience = getObjectFollowersAudience(accountsToForwardView) - return announceActivityData(videoShare.url, byActor, announcedObject, t, audience) + return announceActivityData(videoShare.url, byActor, announcedObject, audience) } async function sendVideoAnnounce (byActor: ActorModel, videoShare: VideoShareModel, video: VideoModel, t: Transaction) { @@ -20,16 +20,8 @@ async function sendVideoAnnounce (byActor: ActorModel, videoShare: VideoShareMod return broadcastToFollowers(data, byActor, [ byActor ], t) } -async function announceActivityData ( - url: string, - byActor: ActorModel, - object: string, - t: Transaction, - audience?: ActivityAudience -): Promise { - if (!audience) { - audience = await getAudience(byActor, t) - } +function announceActivityData (url: string, byActor: ActorModel, object: string, audience?: ActivityAudience): ActivityAnnounce { + if (!audience) audience = getAudience(byActor) return { type: 'Announce', diff --git a/server/lib/activitypub/send/send-create.ts b/server/lib/activitypub/send/send-create.ts index 3ef4fcd3b..293947b05 100644 --- a/server/lib/activitypub/send/send-create.ts +++ b/server/lib/activitypub/send/send-create.ts @@ -23,8 +23,8 @@ async function sendCreateVideo (video: VideoModel, t: Transaction) { const byActor = video.VideoChannel.Account.Actor const videoObject = video.toActivityPubObject() - const audience = await getAudience(byActor, t, video.privacy === VideoPrivacy.PUBLIC) - const data = await createActivityData(video.url, byActor, videoObject, t, audience) + const audience = getAudience(byActor, video.privacy === VideoPrivacy.PUBLIC) + const data = createActivityData(video.url, byActor, videoObject, audience) return broadcastToFollowers(data, byActor, [ byActor ], t) } @@ -33,7 +33,7 @@ async function sendVideoAbuse (byActor: ActorModel, videoAbuse: VideoAbuseModel, const url = getVideoAbuseActivityPubUrl(videoAbuse) const audience = { to: [ video.VideoChannel.Account.Actor.url ], cc: [] } - const data = await createActivityData(url, byActor, videoAbuse.toActivityPubObject(), t, audience) + const data = createActivityData(url, byActor, videoAbuse.toActivityPubObject(), audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } @@ -57,7 +57,7 @@ async function sendCreateVideoComment (comment: VideoCommentModel, t: Transactio audience = getObjectFollowersAudience(actorsInvolvedInComment.concat(parentsCommentActors)) } - const data = await createActivityData(comment.url, byActor, commentObject, t, audience) + const data = createActivityData(comment.url, byActor, commentObject, audience) // This was a reply, send it to the parent actors const actorsException = [ byActor ] @@ -82,14 +82,14 @@ async function sendCreateView (byActor: ActorModel, video: VideoModel, t: Transa // Send to origin if (video.isOwned() === false) { const audience = getVideoAudience(video, actorsInvolvedInVideo) - const data = await createActivityData(url, byActor, viewActivityData, t, audience) + const data = createActivityData(url, byActor, viewActivityData, audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } // Send to followers const audience = getObjectFollowersAudience(actorsInvolvedInVideo) - const data = await createActivityData(url, byActor, viewActivityData, t, audience) + const data = createActivityData(url, byActor, viewActivityData, audience) // Use the server actor to send the view const serverActor = await getServerActor() @@ -106,34 +106,31 @@ async function sendCreateDislike (byActor: ActorModel, video: VideoModel, t: Tra // Send to origin if (video.isOwned() === false) { const audience = getVideoAudience(video, actorsInvolvedInVideo) - const data = await createActivityData(url, byActor, dislikeActivityData, t, audience) + const data = createActivityData(url, byActor, dislikeActivityData, audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } // Send to followers const audience = getObjectFollowersAudience(actorsInvolvedInVideo) - const data = await createActivityData(url, byActor, dislikeActivityData, t, audience) + const data = createActivityData(url, byActor, dislikeActivityData, audience) const actorsException = [ byActor ] return broadcastToFollowers(data, byActor, actorsInvolvedInVideo, t, actorsException) } -async function createActivityData (url: string, - byActor: ActorModel, - object: any, - t: Transaction, - audience?: ActivityAudience): Promise { - if (!audience) { - audience = await getAudience(byActor, t) - } - - return audiencify({ - type: 'Create' as 'Create', - id: url + '/activity', - actor: byActor.url, - object: audiencify(object, audience) - }, audience) +function createActivityData (url: string, byActor: ActorModel, object: any, audience?: ActivityAudience): ActivityCreate { + if (!audience) audience = getAudience(byActor) + + return audiencify( + { + type: 'Create' as 'Create', + id: url + '/activity', + actor: byActor.url, + object: audiencify(object, audience) + }, + audience + ) } function createDislikeActivityData (byActor: ActorModel, video: VideoModel) { diff --git a/server/lib/activitypub/send/send-like.ts b/server/lib/activitypub/send/send-like.ts index ddeb1fcd2..37ee7c096 100644 --- a/server/lib/activitypub/send/send-like.ts +++ b/server/lib/activitypub/send/send-like.ts @@ -14,36 +14,31 @@ async function sendLike (byActor: ActorModel, video: VideoModel, t: Transaction) // Send to origin if (video.isOwned() === false) { const audience = getVideoAudience(video, accountsInvolvedInVideo) - const data = await likeActivityData(url, byActor, video, t, audience) + const data = likeActivityData(url, byActor, video, audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } // Send to followers const audience = getObjectFollowersAudience(accountsInvolvedInVideo) - const data = await likeActivityData(url, byActor, video, t, audience) + const data = likeActivityData(url, byActor, video, audience) const followersException = [ byActor ] return broadcastToFollowers(data, byActor, accountsInvolvedInVideo, t, followersException) } -async function likeActivityData ( - url: string, - byActor: ActorModel, - video: VideoModel, - t: Transaction, - audience?: ActivityAudience -): Promise { - if (!audience) { - audience = await getAudience(byActor, t) - } - - return audiencify({ - type: 'Like' as 'Like', - id: url, - actor: byActor.url, - object: video.url - }, audience) +function likeActivityData (url: string, byActor: ActorModel, video: VideoModel, audience?: ActivityAudience): ActivityLike { + if (!audience) audience = getAudience(byActor) + + return audiencify( + { + type: 'Like' as 'Like', + id: url, + actor: byActor.url, + object: video.url + }, + audience + ) } // --------------------------------------------------------------------------- diff --git a/server/lib/activitypub/send/send-undo.ts b/server/lib/activitypub/send/send-undo.ts index 9733e66dc..33c3d2429 100644 --- a/server/lib/activitypub/send/send-undo.ts +++ b/server/lib/activitypub/send/send-undo.ts @@ -27,7 +27,7 @@ async function sendUndoFollow (actorFollow: ActorFollowModel, t: Transaction) { const undoUrl = getUndoActivityPubUrl(followUrl) const object = followActivityData(followUrl, me, following) - const data = await undoActivityData(undoUrl, me, object, t) + const data = undoActivityData(undoUrl, me, object) return unicastTo(data, me, following.inboxUrl) } @@ -37,18 +37,18 @@ async function sendUndoLike (byActor: ActorModel, video: VideoModel, t: Transact const undoUrl = getUndoActivityPubUrl(likeUrl) const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, t) - const object = await likeActivityData(likeUrl, byActor, video, t) + const object = likeActivityData(likeUrl, byActor, video) // Send to origin if (video.isOwned() === false) { const audience = getVideoAudience(video, actorsInvolvedInVideo) - const data = await undoActivityData(undoUrl, byActor, object, t, audience) + const data = undoActivityData(undoUrl, byActor, object, audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } const audience = getObjectFollowersAudience(actorsInvolvedInVideo) - const data = await undoActivityData(undoUrl, byActor, object, t, audience) + const data = undoActivityData(undoUrl, byActor, object, audience) const followersException = [ byActor ] return broadcastToFollowers(data, byActor, actorsInvolvedInVideo, t, followersException) @@ -60,16 +60,16 @@ async function sendUndoDislike (byActor: ActorModel, video: VideoModel, t: Trans const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, t) const dislikeActivity = createDislikeActivityData(byActor, video) - const object = await createActivityData(dislikeUrl, byActor, dislikeActivity, t) + const object = createActivityData(dislikeUrl, byActor, dislikeActivity) if (video.isOwned() === false) { const audience = getVideoAudience(video, actorsInvolvedInVideo) - const data = await undoActivityData(undoUrl, byActor, object, t, audience) + const data = undoActivityData(undoUrl, byActor, object, audience) return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl) } - const data = await undoActivityData(undoUrl, byActor, object, t) + const data = undoActivityData(undoUrl, byActor, object) const followersException = [ byActor ] return broadcastToFollowers(data, byActor, actorsInvolvedInVideo, t, followersException) @@ -80,7 +80,7 @@ async function sendUndoAnnounce (byActor: ActorModel, videoShare: VideoShareMode const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, t) const object = await buildVideoAnnounce(byActor, videoShare, video, t) - const data = await undoActivityData(undoUrl, byActor, object, t) + const data = undoActivityData(undoUrl, byActor, object) const followersException = [ byActor ] return broadcastToFollowers(data, byActor, actorsInvolvedInVideo, t, followersException) @@ -97,21 +97,21 @@ export { // --------------------------------------------------------------------------- -async function undoActivityData ( +function undoActivityData ( url: string, byActor: ActorModel, object: ActivityFollow | ActivityLike | ActivityCreate | ActivityAnnounce, - t: Transaction, audience?: ActivityAudience -): Promise { - if (!audience) { - audience = await getAudience(byActor, t) - } - - return audiencify({ - type: 'Undo' as 'Undo', - id: url, - actor: byActor.url, - object - }, audience) +): ActivityUndo { + if (!audience) audience = getAudience(byActor) + + return audiencify( + { + type: 'Undo' as 'Undo', + id: url, + actor: byActor.url, + object + }, + audience + ) } diff --git a/server/lib/activitypub/send/send-update.ts b/server/lib/activitypub/send/send-update.ts index d64b88343..2fd374ec6 100644 --- a/server/lib/activitypub/send/send-update.ts +++ b/server/lib/activitypub/send/send-update.ts @@ -15,9 +15,9 @@ async function sendUpdateVideo (video: VideoModel, t: Transaction) { const url = getUpdateActivityPubUrl(video.url, video.updatedAt.toISOString()) const videoObject = video.toActivityPubObject() - const audience = await getAudience(byActor, t, video.privacy === VideoPrivacy.PUBLIC) + const audience = getAudience(byActor, video.privacy === VideoPrivacy.PUBLIC) - const data = await updateActivityData(url, byActor, videoObject, t, audience) + const data = updateActivityData(url, byActor, videoObject, audience) const actorsInvolved = await VideoShareModel.loadActorsByShare(video.id, t) actorsInvolved.push(byActor) @@ -30,8 +30,8 @@ async function sendUpdateActor (accountOrChannel: AccountModel | VideoChannelMod const url = getUpdateActivityPubUrl(byActor.url, byActor.updatedAt.toISOString()) const accountOrChannelObject = accountOrChannel.toActivityPubObject() - const audience = await getAudience(byActor, t) - const data = await updateActivityData(url, byActor, accountOrChannelObject, t, audience) + const audience = getAudience(byActor) + const data = updateActivityData(url, byActor, accountOrChannelObject, audience) let actorsInvolved: ActorModel[] if (accountOrChannel instanceof AccountModel) { @@ -56,21 +56,17 @@ export { // --------------------------------------------------------------------------- -async function updateActivityData ( - url: string, - byActor: ActorModel, - object: any, - t: Transaction, - audience?: ActivityAudience -): Promise { - if (!audience) { - audience = await getAudience(byActor, t) - } +function updateActivityData (url: string, byActor: ActorModel, object: any, audience?: ActivityAudience): ActivityUpdate { + if (!audience) audience = getAudience(byActor) - return audiencify({ - type: 'Update' as 'Update', - id: url, - actor: byActor.url, - object: audiencify(object, audience) - }, audience) + return audiencify( + { + type: 'Update' as 'Update', + id: url, + actor: byActor.url, + object: audiencify(object, audience + ) + }, + audience + ) } diff --git a/server/lib/activitypub/videos.ts b/server/lib/activitypub/videos.ts index 907f7e11e..7ec8ca193 100644 --- a/server/lib/activitypub/videos.ts +++ b/server/lib/activitypub/videos.ts @@ -1,8 +1,9 @@ import * as Bluebird from 'bluebird' +import * as sequelize from 'sequelize' import * as magnetUtil from 'magnet-uri' import { join } from 'path' import * as request from 'request' -import { ActivityIconObject } from '../../../shared/index' +import { ActivityIconObject, VideoState } from '../../../shared/index' import { VideoTorrentObject } from '../../../shared/models/activitypub/objects' import { VideoPrivacy, VideoRateType } from '../../../shared/models/videos' import { sanitizeAndCheckVideoTorrentObject } from '../../helpers/custom-validators/activitypub/videos' @@ -21,6 +22,21 @@ import { VideoShareModel } from '../../models/video/video-share' import { getOrCreateActorAndServerAndModel } from './actor' import { addVideoComments } from './video-comments' import { crawlCollectionPage } from './crawl' +import { sendCreateVideo, sendUpdateVideo } from './send' +import { shareVideoByServerAndChannel } from './index' + +async function federateVideoIfNeeded (video: VideoModel, isNewVideo: boolean, transaction?: sequelize.Transaction) { + // If the video is not private and published, we federate it + if (video.privacy !== VideoPrivacy.PRIVATE && video.state === VideoState.PUBLISHED) { + if (isNewVideo === true) { + // Now we'll add the video's meta data to our followers + await sendCreateVideo(video, transaction) + await shareVideoByServerAndChannel(video, transaction) + } else { + await sendUpdateVideo(video, transaction) + } + } +} function fetchRemoteVideoPreview (video: VideoModel, reject: Function) { const host = video.VideoChannel.Account.Actor.Server.host @@ -55,9 +71,11 @@ function generateThumbnailFromUrl (video: VideoModel, icon: ActivityIconObject) return doRequestAndSaveToFile(options, thumbnailPath) } -async function videoActivityObjectToDBAttributes (videoChannel: VideoChannelModel, - videoObject: VideoTorrentObject, - to: string[] = []) { +async function videoActivityObjectToDBAttributes ( + videoChannel: VideoChannelModel, + videoObject: VideoTorrentObject, + to: string[] = [] +) { const privacy = to.indexOf(ACTIVITY_PUB.PUBLIC) !== -1 ? VideoPrivacy.PUBLIC : VideoPrivacy.UNLISTED const duration = videoObject.duration.replace(/[^\d]+/, '') @@ -90,6 +108,8 @@ async function videoActivityObjectToDBAttributes (videoChannel: VideoChannelMode support, nsfw: videoObject.sensitive, commentsEnabled: videoObject.commentsEnabled, + waitTranscoding: videoObject.waitTranscoding, + state: videoObject.state, channelId: videoChannel.id, duration: parseInt(duration, 10), createdAt: new Date(videoObject.published), @@ -185,22 +205,20 @@ async function getOrCreateVideo (videoObject: VideoTorrentObject, channelActor: } async function getOrCreateAccountAndVideoAndChannel (videoObject: VideoTorrentObject | string, actor?: ActorModel) { - if (typeof videoObject === 'string') { - const videoUrl = videoObject - - const videoFromDatabase = await VideoModel.loadByUrlAndPopulateAccount(videoUrl) - if (videoFromDatabase) { - return { - video: videoFromDatabase, - actor: videoFromDatabase.VideoChannel.Account.Actor, - channelActor: videoFromDatabase.VideoChannel.Actor - } + const videoUrl = typeof videoObject === 'string' ? videoObject : videoObject.id + + const videoFromDatabase = await VideoModel.loadByUrlAndPopulateAccount(videoUrl) + if (videoFromDatabase) { + return { + video: videoFromDatabase, + actor: videoFromDatabase.VideoChannel.Account.Actor, + channelActor: videoFromDatabase.VideoChannel.Actor } - - videoObject = await fetchRemoteVideo(videoUrl) - if (!videoObject) throw new Error('Cannot fetch remote video with url: ' + videoUrl) } + videoObject = await fetchRemoteVideo(videoUrl) + if (!videoObject) throw new Error('Cannot fetch remote video with url: ' + videoUrl) + if (!actor) { const actorObj = videoObject.attributedTo.find(a => a.type === 'Person') if (!actorObj) throw new Error('Cannot find associated actor to video ' + videoObject.url) @@ -291,20 +309,6 @@ async function addVideoShares (shareUrls: string[], instance: VideoModel) { } } -export { - getOrCreateAccountAndVideoAndChannel, - fetchRemoteVideoPreview, - fetchRemoteVideoDescription, - generateThumbnailFromUrl, - videoActivityObjectToDBAttributes, - videoFileActivityUrlToDBAttributes, - getOrCreateVideo, - getOrCreateVideoChannel, - addVideoShares -} - -// --------------------------------------------------------------------------- - async function fetchRemoteVideo (videoUrl: string): Promise { const options = { uri: videoUrl, @@ -324,3 +328,17 @@ async function fetchRemoteVideo (videoUrl: string): Promise return body } + +export { + federateVideoIfNeeded, + fetchRemoteVideo, + getOrCreateAccountAndVideoAndChannel, + fetchRemoteVideoPreview, + fetchRemoteVideoDescription, + generateThumbnailFromUrl, + videoActivityObjectToDBAttributes, + videoFileActivityUrlToDBAttributes, + getOrCreateVideo, + getOrCreateVideoChannel, + addVideoShares +} diff --git a/server/lib/job-queue/handlers/video-file.ts b/server/lib/job-queue/handlers/video-file.ts index 85f7dbfc2..f5ad076a6 100644 --- a/server/lib/job-queue/handlers/video-file.ts +++ b/server/lib/job-queue/handlers/video-file.ts @@ -1,17 +1,16 @@ import * as kue from 'kue' -import { VideoResolution } from '../../../../shared' -import { VideoPrivacy } from '../../../../shared/models/videos' +import { VideoResolution, VideoState } from '../../../../shared' import { logger } from '../../../helpers/logger' import { computeResolutionsToTranscode } from '../../../helpers/utils' -import { sequelizeTypescript } from '../../../initializers' import { VideoModel } from '../../../models/video/video' -import { shareVideoByServerAndChannel } from '../../activitypub' -import { sendCreateVideo, sendUpdateVideo } from '../../activitypub/send' import { JobQueue } from '../job-queue' +import { federateVideoIfNeeded } from '../../activitypub' +import { retryTransactionWrapper } from '../../../helpers/database-utils' +import { sequelizeTypescript } from '../../../initializers' export type VideoFilePayload = { videoUUID: string - isNewVideo: boolean + isNewVideo?: boolean resolution?: VideoResolution isPortraitMode?: boolean } @@ -52,10 +51,20 @@ async function processVideoFile (job: kue.Job) { // Transcoding in other resolution if (payload.resolution) { await video.transcodeOriginalVideofile(payload.resolution, payload.isPortraitMode) - await onVideoFileTranscoderOrImportSuccess(video) + + const options = { + arguments: [ video ], + errorMessage: 'Cannot execute onVideoFileTranscoderOrImportSuccess with many retries.' + } + await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, options) } else { await video.optimizeOriginalVideofile() - await onVideoFileOptimizerSuccess(video, payload.isNewVideo) + + const options = { + arguments: [ video, payload.isNewVideo ], + errorMessage: 'Cannot execute onVideoFileOptimizerSuccess with many retries.' + } + await retryTransactionWrapper(onVideoFileOptimizerSuccess, options) } return video @@ -64,68 +73,70 @@ async function processVideoFile (job: kue.Job) { async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) { if (video === undefined) return undefined - // Maybe the video changed in database, refresh it - const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid) - // Video does not exist anymore - if (!videoDatabase) return undefined + return sequelizeTypescript.transaction(async t => { + // Maybe the video changed in database, refresh it + let videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid, t) + // Video does not exist anymore + if (!videoDatabase) return undefined - if (video.privacy !== VideoPrivacy.PRIVATE) { - await sendUpdateVideo(video, undefined) - } + // We transcoded the video file in another format, now we can publish it + const oldState = videoDatabase.state + videoDatabase.state = VideoState.PUBLISHED + videoDatabase = await videoDatabase.save({ transaction: t }) + + // If the video was not published, we consider it is a new one for other instances + const isNewVideo = oldState !== VideoState.PUBLISHED + await federateVideoIfNeeded(videoDatabase, isNewVideo, t) - return undefined + return undefined + }) } async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boolean) { if (video === undefined) return undefined - // Maybe the video changed in database, refresh it - const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid) - // Video does not exist anymore - if (!videoDatabase) return undefined - - if (video.privacy !== VideoPrivacy.PRIVATE) { - if (isNewVideo !== false) { - // Now we'll add the video's meta data to our followers - await sequelizeTypescript.transaction(async t => { - await sendCreateVideo(video, t) - await shareVideoByServerAndChannel(video, t) - }) - } else { - await sendUpdateVideo(video, undefined) - } - } - - const { videoFileResolution } = await videoDatabase.getOriginalFileResolution() - - // Create transcoding jobs if there are enabled resolutions - const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution) - logger.info( - 'Resolutions computed for video %s and origin file height of %d.', videoDatabase.uuid, videoFileResolution, - { resolutions: resolutionsEnabled } - ) + // Outside the transaction (IO on disk) + const { videoFileResolution } = await video.getOriginalFileResolution() + + return sequelizeTypescript.transaction(async t => { + // Maybe the video changed in database, refresh it + const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid, t) + // Video does not exist anymore + if (!videoDatabase) return undefined + + // Create transcoding jobs if there are enabled resolutions + const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution) + logger.info( + 'Resolutions computed for video %s and origin file height of %d.', videoDatabase.uuid, videoFileResolution, + { resolutions: resolutionsEnabled } + ) + + if (resolutionsEnabled.length !== 0) { + const tasks: Promise[] = [] + + for (const resolution of resolutionsEnabled) { + const dataInput = { + videoUUID: videoDatabase.uuid, + resolution + } + + const p = JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput }) + tasks.push(p) + } - if (resolutionsEnabled.length !== 0) { - const tasks: Promise[] = [] + await Promise.all(tasks) - for (const resolution of resolutionsEnabled) { - const dataInput = { - videoUUID: videoDatabase.uuid, - resolution, - isNewVideo - } + logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled }) + } else { + // No transcoding to do, it's now published + video.state = VideoState.PUBLISHED + video = await video.save({ transaction: t }) - const p = JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput }) - tasks.push(p) + logger.info('No transcoding jobs created for video %s (no resolutions).', video.uuid) } - await Promise.all(tasks) - - logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled }) - } else { - logger.info('No transcoding jobs created for video %s (no resolutions enabled).') - return undefined - } + return federateVideoIfNeeded(video, isNewVideo, t) + }) } // --------------------------------------------------------------------------- diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index bdfa19b61..695fe0eea 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -79,6 +79,7 @@ class JobQueue { const res = await handlers[ handlerName ](job) return done(null, res) } catch (err) { + logger.error('Cannot execute job %d.', job.id, { err }) return done(err) } }) -- cgit v1.2.3