X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fhandlers%2Fvideo-import.ts;h=2a063282cf2aef546c587bc012ff2b75a17488fe;hb=0c9668f77901e7540e2c7045eb0f2974a4842a69;hp=2f74e9fbd6c9548142f24582088f2c7870194a14;hpb=d17c7b4e8c52317bdc874917387b7a49f6cf8b01;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts index 2f74e9fbd..2a063282c 100644 --- a/server/lib/job-queue/handlers/video-import.ts +++ b/server/lib/job-queue/handlers/video-import.ts @@ -1,22 +1,26 @@ -import { Job } from 'bull' +import { Job } from 'bullmq' import { move, remove, stat } from 'fs-extra' import { retryTransactionWrapper } from '@server/helpers/database-utils' import { YoutubeDLWrapper } from '@server/helpers/youtube-dl' +import { CONFIG } from '@server/initializers/config' import { isPostImportVideoAccepted } from '@server/lib/moderation' import { generateWebTorrentVideoFilename } from '@server/lib/paths' import { Hooks } from '@server/lib/plugins/hooks' import { ServerConfigManager } from '@server/lib/server-config-manager' +import { createOptimizeOrMergeAudioJobs } from '@server/lib/transcoding/create-transcoding-job' import { isAbleToUploadVideo } from '@server/lib/user' -import { addMoveToObjectStorageJob, addOptimizeOrMergeAudioJob } from '@server/lib/video' +import { buildMoveToObjectStorageJob } from '@server/lib/video' import { VideoPathManager } from '@server/lib/video-path-manager' import { buildNextVideoState } from '@server/lib/video-state' import { ThumbnailModel } from '@server/models/video/thumbnail' -import { MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/types/models/video/video-import' +import { MUserId, MVideoFile, MVideoFullLight } from '@server/types/models' +import { MVideoImport, MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/types/models/video/video-import' import { getLowercaseExtension } from '@shared/core-utils' -import { isAudioFile } from '@shared/extra-utils' +import { ffprobePromise, getVideoStreamDimensionsInfo, getVideoStreamDuration, getVideoStreamFPS, isAudioFile } from '@shared/ffmpeg' import { ThumbnailType, VideoImportPayload, + VideoImportPreventExceptionResult, VideoImportState, VideoImportTorrentPayload, VideoImportTorrentPayloadType, @@ -25,25 +29,42 @@ import { VideoResolution, VideoState } from '@shared/models' -import { ffprobePromise, getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffprobe-utils' import { logger } from '../../../helpers/logger' import { getSecureTorrentName } from '../../../helpers/utils' import { createTorrentAndSetInfoHash, downloadWebTorrentVideo } from '../../../helpers/webtorrent' -import { VIDEO_IMPORT_TIMEOUT } from '../../../initializers/constants' +import { JOB_TTL } from '../../../initializers/constants' import { sequelizeTypescript } from '../../../initializers/database' import { VideoModel } from '../../../models/video/video' import { VideoFileModel } from '../../../models/video/video-file' import { VideoImportModel } from '../../../models/video/video-import' -import { MThumbnail } from '../../../types/models/video/thumbnail' import { federateVideoIfNeeded } from '../../activitypub/videos' import { Notifier } from '../../notifier' import { generateVideoMiniature } from '../../thumbnail' +import { JobQueue } from '../job-queue' -async function processVideoImport (job: Job) { +async function processVideoImport (job: Job): Promise { const payload = job.data as VideoImportPayload - if (payload.type === 'youtube-dl') return processYoutubeDLImport(job, payload) - if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') return processTorrentImport(job, payload) + const videoImport = await getVideoImportOrDie(payload) + if (videoImport.state === VideoImportState.CANCELLED) { + logger.info('Do not process import since it has been cancelled', { payload }) + return { resultType: 'success' } + } + + videoImport.state = VideoImportState.PROCESSING + await videoImport.save() + + try { + if (payload.type === 'youtube-dl') await processYoutubeDLImport(job, videoImport, payload) + if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') await processTorrentImport(job, videoImport, payload) + + return { resultType: 'success' } + } catch (err) { + if (!payload.preventException) throw err + + logger.warn('Catch error in video import to send value to parent job.', { payload, err }) + return { resultType: 'error' } + } } // --------------------------------------------------------------------------- @@ -54,44 +75,40 @@ export { // --------------------------------------------------------------------------- -async function processTorrentImport (job: Job, payload: VideoImportTorrentPayload) { - logger.info('Processing torrent video import in job %d.', job.id) +async function processTorrentImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportTorrentPayload) { + logger.info('Processing torrent video import in job %s.', job.id) - const videoImport = await getVideoImportOrDie(payload.videoImportId) + const options = { type: payload.type, videoImportId: payload.videoImportId } - const options = { - type: payload.type, - videoImportId: payload.videoImportId - } const target = { torrentName: videoImport.torrentName ? getSecureTorrentName(videoImport.torrentName) : undefined, uri: videoImport.magnetUri } - return processFile(() => downloadWebTorrentVideo(target, VIDEO_IMPORT_TIMEOUT), videoImport, options) + return processFile(() => downloadWebTorrentVideo(target, JOB_TTL['video-import']), videoImport, options) } -async function processYoutubeDLImport (job: Job, payload: VideoImportYoutubeDLPayload) { - logger.info('Processing youtubeDL video import in job %d.', job.id) +async function processYoutubeDLImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportYoutubeDLPayload) { + logger.info('Processing youtubeDL video import in job %s.', job.id) - const videoImport = await getVideoImportOrDie(payload.videoImportId) - const options = { - type: payload.type, - videoImportId: videoImport.id - } + const options = { type: payload.type, videoImportId: videoImport.id } - const youtubeDL = new YoutubeDLWrapper(videoImport.targetUrl, ServerConfigManager.Instance.getEnabledResolutions('vod')) + const youtubeDL = new YoutubeDLWrapper( + videoImport.targetUrl, + ServerConfigManager.Instance.getEnabledResolutions('vod'), + CONFIG.TRANSCODING.ALWAYS_TRANSCODE_ORIGINAL_RESOLUTION + ) return processFile( - () => youtubeDL.downloadVideo(payload.fileExt, VIDEO_IMPORT_TIMEOUT), + () => youtubeDL.downloadVideo(payload.fileExt, JOB_TTL['video-import']), videoImport, options ) } -async function getVideoImportOrDie (videoImportId: number) { - const videoImport = await VideoImportModel.loadAndPopulateVideo(videoImportId) - if (!videoImport || !videoImport.Video) { - throw new Error('Cannot import video %s: the video import or video linked to this import does not exist anymore.') +async function getVideoImportOrDie (payload: VideoImportPayload) { + const videoImport = await VideoImportModel.loadAndPopulateVideo(payload.videoImportId) + if (!videoImport?.Video) { + throw new Error(`Cannot import video ${payload.videoImportId}: the video import or video linked to this import does not exist anymore.`) } return videoImport @@ -120,10 +137,10 @@ async function processFile (downloader: () => Promise, videoImport: MVid const { resolution } = await isAudioFile(tempVideoPath, probe) ? { resolution: VideoResolution.H_NOVIDEO } - : await getVideoFileResolution(tempVideoPath) + : await getVideoStreamDimensionsInfo(tempVideoPath, probe) - const fps = await getVideoFileFPS(tempVideoPath, probe) - const duration = await getDurationFromVideoFile(tempVideoPath, probe) + const fps = await getVideoStreamFPS(tempVideoPath, probe) + const duration = await getVideoStreamDuration(tempVideoPath, probe) // Prepare video file object for creation in database const fileExt = getLowercaseExtension(tempVideoPath) @@ -161,121 +178,157 @@ async function processFile (downloader: () => Promise, videoImport: MVid } // Video is accepted, resuming preparation - const videoWithFiles = Object.assign(videoImport.Video, { VideoFiles: [ videoFile ], VideoStreamingPlaylists: [] }) - // To clean files if the import fails - const videoImportWithFiles: MVideoImportDefaultFiles = Object.assign(videoImport, { Video: videoWithFiles }) - - // Move file - const videoDestFile = VideoPathManager.Instance.getFSVideoFileOutputPath(videoImportWithFiles.Video, videoFile) - await move(tempVideoPath, videoDestFile) - tempVideoPath = null // This path is not used anymore - - // Generate miniature if the import did not created it - let thumbnailModel: MThumbnail - let thumbnailSave: object - if (!videoImportWithFiles.Video.getMiniature()) { - thumbnailModel = await generateVideoMiniature({ - video: videoImportWithFiles.Video, - videoFile, - type: ThumbnailType.MINIATURE - }) - thumbnailSave = thumbnailModel.toJSON() - } + const videoFileLockReleaser = await VideoPathManager.Instance.lockFiles(videoImport.Video.uuid) - // Generate preview if the import did not created it - let previewModel: MThumbnail - let previewSave: object - if (!videoImportWithFiles.Video.getPreview()) { - previewModel = await generateVideoMiniature({ - video: videoImportWithFiles.Video, - videoFile, - type: ThumbnailType.PREVIEW - }) - previewSave = previewModel.toJSON() - } + try { + const videoImportWithFiles = await refreshVideoImportFromDB(videoImport, videoFile) - // Create torrent - await createTorrentAndSetInfoHash(videoImportWithFiles.Video, videoFile) + // Move file + const videoDestFile = VideoPathManager.Instance.getFSVideoFileOutputPath(videoImportWithFiles.Video, videoFile) + await move(tempVideoPath, videoDestFile) - const videoFileSave = videoFile.toJSON() + tempVideoPath = null // This path is not used anymore - const { videoImportUpdated, video } = await retryTransactionWrapper(() => { - return sequelizeTypescript.transaction(async t => { - const videoImportToUpdate = videoImportWithFiles as MVideoImportVideo + let { + miniatureModel: thumbnailModel, + miniatureJSONSave: thumbnailSave + } = await generateMiniature(videoImportWithFiles, videoFile, ThumbnailType.MINIATURE) - // Refresh video - const video = await VideoModel.load(videoImportToUpdate.videoId, t) - if (!video) throw new Error('Video linked to import ' + videoImportToUpdate.videoId + ' does not exist anymore.') + let { + miniatureModel: previewModel, + miniatureJSONSave: previewSave + } = await generateMiniature(videoImportWithFiles, videoFile, ThumbnailType.PREVIEW) - const videoFileCreated = await videoFile.save({ transaction: t }) + // Create torrent + await createTorrentAndSetInfoHash(videoImportWithFiles.Video, videoFile) - // Update video DB object - video.duration = duration - video.state = buildNextVideoState(video.state) - await video.save({ transaction: t }) + const videoFileSave = videoFile.toJSON() - if (thumbnailModel) await video.addAndSaveThumbnail(thumbnailModel, t) - if (previewModel) await video.addAndSaveThumbnail(previewModel, t) + const { videoImportUpdated, video } = await retryTransactionWrapper(() => { + return sequelizeTypescript.transaction(async t => { + // Refresh video + const video = await VideoModel.load(videoImportWithFiles.videoId, t) + if (!video) throw new Error('Video linked to import ' + videoImportWithFiles.videoId + ' does not exist anymore.') - // Now we can federate the video (reload from database, we need more attributes) - const videoForFederation = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t) - await federateVideoIfNeeded(videoForFederation, true, t) + await videoFile.save({ transaction: t }) - // Update video import object - videoImportToUpdate.state = VideoImportState.SUCCESS - const videoImportUpdated = await videoImportToUpdate.save({ transaction: t }) as MVideoImportVideo - videoImportUpdated.Video = video + // Update video DB object + video.duration = duration + video.state = buildNextVideoState(video.state) + await video.save({ transaction: t }) - videoImportToUpdate.Video = Object.assign(video, { VideoFiles: [ videoFileCreated ] }) + if (thumbnailModel) await video.addAndSaveThumbnail(thumbnailModel, t) + if (previewModel) await video.addAndSaveThumbnail(previewModel, t) - logger.info('Video %s imported.', video.uuid) + // Now we can federate the video (reload from database, we need more attributes) + const videoForFederation = await VideoModel.loadFull(video.uuid, t) + await federateVideoIfNeeded(videoForFederation, true, t) - return { videoImportUpdated, video: videoForFederation } - }).catch(err => { - // Reset fields - if (thumbnailModel) thumbnailModel = new ThumbnailModel(thumbnailSave) - if (previewModel) previewModel = new ThumbnailModel(previewSave) + // Update video import object + videoImportWithFiles.state = VideoImportState.SUCCESS + const videoImportUpdated = await videoImportWithFiles.save({ transaction: t }) as MVideoImport - videoFile = new VideoFileModel(videoFileSave) + logger.info('Video %s imported.', video.uuid) - throw err - }) - }) + return { videoImportUpdated, video: videoForFederation } + }).catch(err => { + // Reset fields + if (thumbnailModel) thumbnailModel = new ThumbnailModel(thumbnailSave) + if (previewModel) previewModel = new ThumbnailModel(previewSave) - Notifier.Instance.notifyOnFinishedVideoImport({ videoImport: videoImportUpdated, success: true }) + videoFile = new VideoFileModel(videoFileSave) - if (video.isBlacklisted()) { - const videoBlacklist = Object.assign(video.VideoBlacklist, { Video: video }) + throw err + }) + }) - Notifier.Instance.notifyOnVideoAutoBlacklist(videoBlacklist) - } else { - Notifier.Instance.notifyOnNewVideoIfNeeded(video) + await afterImportSuccess({ videoImport: videoImportUpdated, video, videoFile, user: videoImport.User }) + } finally { + videoFileLockReleaser() } + } catch (err) { + await onImportError(err, tempVideoPath, videoImport) - if (video.state === VideoState.TO_MOVE_TO_EXTERNAL_STORAGE) { - return addMoveToObjectStorageJob(videoImportUpdated.Video) - } + throw err + } +} - // Create transcoding jobs? - if (video.state === VideoState.TO_TRANSCODE) { - await addOptimizeOrMergeAudioJob(videoImportUpdated.Video, videoFile, videoImport.User) - } +async function refreshVideoImportFromDB (videoImport: MVideoImportDefault, videoFile: MVideoFile): Promise { + // Refresh video, privacy may have changed + const video = await videoImport.Video.reload() + const videoWithFiles = Object.assign(video, { VideoFiles: [ videoFile ], VideoStreamingPlaylists: [] }) - } catch (err) { - try { - if (tempVideoPath) await remove(tempVideoPath) - } catch (errUnlink) { - logger.warn('Cannot cleanup files after a video import error.', { err: errUnlink }) - } + return Object.assign(videoImport, { Video: videoWithFiles }) +} + +async function generateMiniature (videoImportWithFiles: MVideoImportDefaultFiles, videoFile: MVideoFile, thumbnailType: ThumbnailType) { + // Generate miniature if the import did not created it + const needsMiniature = thumbnailType === ThumbnailType.MINIATURE + ? !videoImportWithFiles.Video.getMiniature() + : !videoImportWithFiles.Video.getPreview() - videoImport.error = err.message - if (videoImport.state !== VideoImportState.REJECTED) { - videoImport.state = VideoImportState.FAILED + if (!needsMiniature) { + return { + miniatureModel: null, + miniatureJSONSave: null } - await videoImport.save() + } - Notifier.Instance.notifyOnFinishedVideoImport({ videoImport, success: false }) + const miniatureModel = await generateVideoMiniature({ + video: videoImportWithFiles.Video, + videoFile, + type: thumbnailType + }) + const miniatureJSONSave = miniatureModel.toJSON() - throw err + return { + miniatureModel, + miniatureJSONSave + } +} + +async function afterImportSuccess (options: { + videoImport: MVideoImport + video: MVideoFullLight + videoFile: MVideoFile + user: MUserId +}) { + const { video, videoFile, videoImport, user } = options + + Notifier.Instance.notifyOnFinishedVideoImport({ videoImport: Object.assign(videoImport, { Video: video }), success: true }) + + if (video.isBlacklisted()) { + const videoBlacklist = Object.assign(video.VideoBlacklist, { Video: video }) + + Notifier.Instance.notifyOnVideoAutoBlacklist(videoBlacklist) + } else { + Notifier.Instance.notifyOnNewVideoIfNeeded(video) + } + + if (video.state === VideoState.TO_MOVE_TO_EXTERNAL_STORAGE) { + await JobQueue.Instance.createJob( + await buildMoveToObjectStorageJob({ video, previousVideoState: VideoState.TO_IMPORT }) + ) + return + } + + if (video.state === VideoState.TO_TRANSCODE) { // Create transcoding jobs? + await createOptimizeOrMergeAudioJobs({ video, videoFile, isNewVideo: true, user }) + } +} + +async function onImportError (err: Error, tempVideoPath: string, videoImport: MVideoImportVideo) { + try { + if (tempVideoPath) await remove(tempVideoPath) + } catch (errUnlink) { + logger.warn('Cannot cleanup files after a video import error.', { err: errUnlink }) } + + videoImport.error = err.message + if (videoImport.state !== VideoImportState.REJECTED) { + videoImport.state = VideoImportState.FAILED + } + await videoImport.save() + + Notifier.Instance.notifyOnFinishedVideoImport({ videoImport, success: false }) }