8 files changed, 272 insertions, 210 deletions

diff --git a/server/lib/job-queue/handlers/activitypub-cleaner.ts b/server/lib/job-queue/handlers/activitypub-cleaner.ts
index 84c0a2de2..a25f00b0a 100644
--- a/server/lib/job-queue/handlers/activitypub-cleaner.ts
+++ b/server/lib/job-queue/handlers/activitypub-cleaner.ts
@@ -88,7 +88,7 @@ async function updateObjectIfNeeded <T> (options: {
     const { body } = await doJSONRequest<any>(url, { activityPub: true })
 
     // If not same id, check same host and update
-    if (!body || !body.id || !bodyValidator(body)) throw new Error(`Body or body id of ${url} is invalid`)
+    if (!body?.id || !bodyValidator(body)) throw new Error(`Body or body id of ${url} is invalid`)
 
     if (body.type === 'Tombstone') {
       return on404OrTombstone()
diff --git a/server/lib/job-queue/handlers/manage-video-torrent.ts b/server/lib/job-queue/handlers/manage-video-torrent.ts
index 03aa414c9..cef93afda 100644
--- a/server/lib/job-queue/handlers/manage-video-torrent.ts
+++ b/server/lib/job-queue/handlers/manage-video-torrent.ts
@@ -1,5 +1,7 @@
 import { Job } from 'bullmq'
+import { extractVideo } from '@server/helpers/video'
 import { createTorrentAndSetInfoHash, updateTorrentMetadata } from '@server/helpers/webtorrent'
+import { VideoPathManager } from '@server/lib/video-path-manager'
 import { VideoModel } from '@server/models/video/video'
 import { VideoFileModel } from '@server/models/video/video-file'
 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
@@ -30,17 +32,23 @@ async function doCreateAction (payload: ManageVideoTorrentPayload & { action: 'c
 
   if (!video || !file) return
 
-  await createTorrentAndSetInfoHash(video, file)
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
 
-  // Refresh videoFile because the createTorrentAndSetInfoHash could be long
-  const refreshedFile = await VideoFileModel.loadWithVideo(file.id)
-  // File does not exist anymore, remove the generated torrent
-  if (!refreshedFile) return file.removeTorrent()
+  try {
+    await createTorrentAndSetInfoHash(video, file)
 
-  refreshedFile.infoHash = file.infoHash
-  refreshedFile.torrentFilename = file.torrentFilename
+    // Refresh videoFile because the createTorrentAndSetInfoHash could be long
+    const refreshedFile = await VideoFileModel.loadWithVideo(file.id)
+    // File does not exist anymore, remove the generated torrent
+    if (!refreshedFile) return file.removeTorrent()
 
-  return refreshedFile.save()
+    refreshedFile.infoHash = file.infoHash
+    refreshedFile.torrentFilename = file.torrentFilename
+
+    await refreshedFile.save()
+  } finally {
+    fileMutexReleaser()
+  }
 }
 
 async function doUpdateMetadataAction (payload: ManageVideoTorrentPayload & { action: 'update-metadata' }) {
@@ -52,9 +60,16 @@ async function doUpdateMetadataAction (payload: ManageVideoTorrentPayload & { ac
 
   if ((!video && !streamingPlaylist) || !file) return
 
-  await updateTorrentMetadata(video || streamingPlaylist, file)
+  const extractedVideo = extractVideo(video || streamingPlaylist)
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(extractedVideo.uuid)
 
-  await file.save()
+  try {
+    await updateTorrentMetadata(video || streamingPlaylist, file)
+
+    await file.save()
+  } finally {
+    fileMutexReleaser()
+  }
 }
 
 async function loadVideoOrLog (videoId: number) {
@@ -82,7 +97,7 @@ async function loadStreamingPlaylistOrLog (streamingPlaylistId: number) {
 async function loadFileOrLog (videoFileId: number) {
   if (!videoFileId) return undefined
 
-  const file = await VideoFileModel.loadWithVideo(videoFileId)
+  const file = await VideoFileModel.load(videoFileId)
 
   if (!file) {
     logger.debug('Do not process torrent for file %d: does not exist anymore.', videoFileId)
diff --git a/server/lib/job-queue/handlers/move-to-object-storage.ts b/server/lib/job-queue/handlers/move-to-object-storage.ts
index 25bdebeea..a1530cc57 100644
--- a/server/lib/job-queue/handlers/move-to-object-storage.ts
+++ b/server/lib/job-queue/handlers/move-to-object-storage.ts
@@ -3,10 +3,10 @@ import { remove } from 'fs-extra'
 import { join } from 'path'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { updateTorrentMetadata } from '@server/helpers/webtorrent'
-import { CONFIG } from '@server/initializers/config'
 import { P2P_MEDIA_LOADER_PEER_VERSION } from '@server/initializers/constants'
-import { storeHLSFile, storeWebTorrentFile } from '@server/lib/object-storage'
+import { storeHLSFileFromFilename, storeWebTorrentFile } from '@server/lib/object-storage'
 import { getHLSDirectory, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
+import { VideoPathManager } from '@server/lib/video-path-manager'
 import { moveToFailedMoveToObjectStorageState, moveToNextState } from '@server/lib/video-state'
 import { VideoModel } from '@server/models/video/video'
 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
@@ -28,6 +28,8 @@ export async function processMoveToObjectStorage (job: Job) {
 
   const lTags = lTagsBase(video.uuid, video.url)
 
+  const fileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
   try {
     if (video.VideoFiles) {
       logger.debug('Moving %d webtorrent files for video %s.', video.VideoFiles.length, video.uuid, lTags)
@@ -49,6 +51,10 @@ export async function processMoveToObjectStorage (job: Job) {
     }
   } catch (err) {
     await onMoveToObjectStorageFailure(job, err)
+
+    throw err
+  } finally {
+    fileMutexReleaser()
   }
 
   return payload.videoUUID
@@ -72,9 +78,9 @@ async function moveWebTorrentFiles (video: MVideoWithAllFiles) {
   for (const file of video.VideoFiles) {
     if (file.storage !== VideoStorage.FILE_SYSTEM) continue
 
-    const fileUrl = await storeWebTorrentFile(file.filename)
+    const fileUrl = await storeWebTorrentFile(video, file)
 
-    const oldPath = join(CONFIG.STORAGE.VIDEOS_DIR, file.filename)
+    const oldPath = VideoPathManager.Instance.getFSVideoFileOutputPath(video, file)
     await onFileMoved({ videoOrPlaylist: video, file, fileUrl, oldPath })
   }
 }
@@ -88,10 +94,10 @@ async function moveHLSFiles (video: MVideoWithAllFiles) {
 
       // Resolution playlist
       const playlistFilename = getHlsResolutionPlaylistFilename(file.filename)
-      await storeHLSFile(playlistWithVideo, playlistFilename)
+      await storeHLSFileFromFilename(playlistWithVideo, playlistFilename)
 
       // Resolution fragmented file
-      const fileUrl = await storeHLSFile(playlistWithVideo, file.filename)
+      const fileUrl = await storeHLSFileFromFilename(playlistWithVideo, file.filename)
 
       const oldPath = join(getHLSDirectory(video), file.filename)
 
@@ -113,9 +119,9 @@ async function doAfterLastJob (options: {
     const playlistWithVideo = playlist.withVideo(video)
 
     // Master playlist
-    playlist.playlistUrl = await storeHLSFile(playlistWithVideo, playlist.playlistFilename)
+    playlist.playlistUrl = await storeHLSFileFromFilename(playlistWithVideo, playlist.playlistFilename)
     // Sha256 segments file
-    playlist.segmentsSha256Url = await storeHLSFile(playlistWithVideo, playlist.segmentsSha256Filename)
+    playlist.segmentsSha256Url = await storeHLSFileFromFilename(playlistWithVideo, playlist.segmentsSha256Filename)
 
     playlist.storage = VideoStorage.OBJECT_STORAGE
 
diff --git a/server/lib/job-queue/handlers/video-channel-import.ts b/server/lib/job-queue/handlers/video-channel-import.ts
index 600292844..035f88e96 100644
--- a/server/lib/job-queue/handlers/video-channel-import.ts
+++ b/server/lib/job-queue/handlers/video-channel-import.ts
@@ -5,7 +5,7 @@ import { synchronizeChannel } from '@server/lib/sync-channel'
 import { VideoChannelModel } from '@server/models/video/video-channel'
 import { VideoChannelSyncModel } from '@server/models/video/video-channel-sync'
 import { MChannelSync } from '@server/types/models'
-import { VideoChannelImportPayload, VideoChannelSyncState } from '@shared/models'
+import { VideoChannelImportPayload } from '@shared/models'
 
 export async function processVideoChannelImport (job: Job) {
   const payload = job.data as VideoChannelImportPayload
@@ -32,17 +32,12 @@ export async function processVideoChannelImport (job: Job) {
 
   const videoChannel = await VideoChannelModel.loadAndPopulateAccount(payload.videoChannelId)
 
-  try {
-    logger.info(`Starting importing videos from external channel "${payload.externalChannelUrl}" to "${videoChannel.name}" `)
-
-    await synchronizeChannel({
-      channel: videoChannel,
-      externalChannelUrl: payload.externalChannelUrl,
-      channelSync
-    })
-  } catch (err) {
-    logger.error(`Failed to import channel ${videoChannel.name}`, { err })
-    channelSync.state = VideoChannelSyncState.FAILED
-    await channelSync.save()
-  }
+  logger.info(`Starting importing videos from external channel "${payload.externalChannelUrl}" to "${videoChannel.name}" `)
+
+  await synchronizeChannel({
+    channel: videoChannel,
+    externalChannelUrl: payload.externalChannelUrl,
+    channelSync,
+    videosCountLimit: CONFIG.IMPORT.VIDEO_CHANNEL_SYNCHRONIZATION.FULL_SYNC_VIDEOS_LIMIT
+  })
 }
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index 9901b878c..4d361c7b9 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -12,7 +12,8 @@ import { buildMoveToObjectStorageJob, buildOptimizeOrMergeAudioJob } from '@serv
 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 {
@@ -36,7 +37,6 @@ 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'
@@ -107,7 +107,7 @@ async function processYoutubeDLImport (job: Job, videoImport: MVideoImportDefaul
 
 async function getVideoImportOrDie (payload: VideoImportPayload) {
   const videoImport = await VideoImportModel.loadAndPopulateVideo(payload.videoImportId)
-  if (!videoImport || !videoImport.Video) {
+  if (!videoImport?.Video) {
     throw new Error(`Cannot import video ${payload.videoImportId}: the video import or video linked to this import does not exist anymore.`)
   }
 
@@ -178,125 +178,159 @@ async function processFile (downloader: () => Promise<string>, 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.loadFull(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) {
-      await JobQueue.Instance.createJob(
-        await buildMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT })
-      )
-    }
+    throw err
+  }
+}
 
-    // Create transcoding jobs?
-    if (video.state === VideoState.TO_TRANSCODE) {
-      await JobQueue.Instance.createJob(
-        await buildOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User })
-      )
-    }
+async function refreshVideoImportFromDB (videoImport: MVideoImportDefault, videoFile: MVideoFile): Promise<MVideoImportDefaultFiles> {
+  // 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 })
+}
 
-    videoImport.error = err.message
-    if (videoImport.state !== VideoImportState.REJECTED) {
-      videoImport.state = VideoImportState.FAILED
+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()
+
+  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 JobQueue.Instance.createJob(
+      await buildOptimizeOrMergeAudioJob({ video, videoFile, 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 })
 }
diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts
index 8a3ee09a2..c6263f55a 100644
--- a/server/lib/job-queue/handlers/video-live-ending.ts
+++ b/server/lib/job-queue/handlers/video-live-ending.ts
@@ -4,7 +4,7 @@ import { join } from 'path'
 import { ffprobePromise, getAudioStream, getVideoStreamDimensionsInfo } from '@server/helpers/ffmpeg'
 import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url'
 import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
-import { cleanupPermanentLive, cleanupTMPLiveFiles, cleanupUnsavedNormalLive } from '@server/lib/live'
+import { cleanupAndDestroyPermanentLive, cleanupTMPLiveFiles, cleanupUnsavedNormalLive } from '@server/lib/live'
 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getLiveReplayBaseDirectory } from '@server/lib/paths'
 import { generateVideoMiniature } from '@server/lib/thumbnail'
 import { generateHlsPlaylistResolutionFromTS } from '@server/lib/transcoding/transcoding'
@@ -18,6 +18,7 @@ import { VideoStreamingPlaylistModel } from '@server/models/video/video-streamin
 import { MVideo, MVideoLive, MVideoLiveSession, MVideoWithAllFiles } from '@server/types/models'
 import { ThumbnailType, VideoLiveEndingPayload, VideoState } from '@shared/models'
 import { logger, loggerTagsFactory } from '../../../helpers/logger'
+import { VideoPathManager } from '@server/lib/video-path-manager'
 
 const lTags = loggerTagsFactory('live', 'job')
 
@@ -34,13 +35,13 @@ async function processVideoLiveEnding (job: Job) {
   const live = await VideoLiveModel.loadByVideoId(payload.videoId)
   const liveSession = await VideoLiveSessionModel.load(payload.liveSessionId)
 
-  const permanentLive = live.permanentLive
-
   if (!video || !live || !liveSession) {
     logError()
     return
   }
 
+  const permanentLive = live.permanentLive
+
   liveSession.endingProcessed = true
   await liveSession.save()
 
@@ -141,23 +142,22 @@ async function replaceLiveByReplay (options: {
 }) {
   const { video, liveSession, live, permanentLive, replayDirectory } = options
 
-  await cleanupTMPLiveFiles(video)
+  const videoWithFiles = await VideoModel.loadFull(video.id)
+  const hlsPlaylist = videoWithFiles.getHLSPlaylist()
+
+  await cleanupTMPLiveFiles(videoWithFiles, hlsPlaylist)
 
   await live.destroy()
 
-  video.isLive = false
-  video.waitTranscoding = true
-  video.state = VideoState.TO_TRANSCODE
+  videoWithFiles.isLive = false
+  videoWithFiles.waitTranscoding = true
+  videoWithFiles.state = VideoState.TO_TRANSCODE
 
-  await video.save()
+  await videoWithFiles.save()
 
-  liveSession.replayVideoId = video.id
+  liveSession.replayVideoId = videoWithFiles.id
   await liveSession.save()
 
-  // Remove old HLS playlist video files
-  const videoWithFiles = await VideoModel.loadFull(video.id)
-
-  const hlsPlaylist = videoWithFiles.getHLSPlaylist()
   await VideoFileModel.removeHLSFilesOfVideoId(hlsPlaylist.id)
 
   // Reset playlist
@@ -206,18 +206,27 @@ async function assignReplayFilesToVideo (options: {
   const concatenatedTsFiles = await readdir(replayDirectory)
 
   for (const concatenatedTsFile of concatenatedTsFiles) {
+    const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
     const concatenatedTsFilePath = join(replayDirectory, concatenatedTsFile)
 
     const probe = await ffprobePromise(concatenatedTsFilePath)
     const { audioStream } = await getAudioStream(concatenatedTsFilePath, probe)
     const { resolution } = await getVideoStreamDimensionsInfo(concatenatedTsFilePath, probe)
 
-    await generateHlsPlaylistResolutionFromTS({
-      video,
-      concatenatedTsFilePath,
-      resolution,
-      isAAC: audioStream?.codec_name === 'aac'
-    })
+    try {
+      await generateHlsPlaylistResolutionFromTS({
+        video,
+        inputFileMutexReleaser,
+        concatenatedTsFilePath,
+        resolution,
+        isAAC: audioStream?.codec_name === 'aac'
+      })
+    } catch (err) {
+      logger.error('Cannot generate HLS playlist resolution from TS files.', { err })
+    }
+
+    inputFileMutexReleaser()
   }
 
   return video
@@ -234,7 +243,7 @@ async function cleanupLiveAndFederate (options: {
 
   if (streamingPlaylist) {
     if (permanentLive) {
-      await cleanupPermanentLive(video, streamingPlaylist)
+      await cleanupAndDestroyPermanentLive(video, streamingPlaylist)
     } else {
       await cleanupUnsavedNormalLive(video, streamingPlaylist)
     }
diff --git a/server/lib/job-queue/handlers/video-transcoding.ts b/server/lib/job-queue/handlers/video-transcoding.ts
index b0e92acf7..3e6d23363 100644
--- a/server/lib/job-queue/handlers/video-transcoding.ts
+++ b/server/lib/job-queue/handlers/video-transcoding.ts
@@ -13,7 +13,6 @@ import {
   MergeAudioTranscodingPayload,
   NewWebTorrentResolutionTranscodingPayload,
   OptimizeTranscodingPayload,
-  VideoResolution,
   VideoTranscodingPayload
 } from '@shared/models'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
@@ -94,15 +93,24 @@ async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MV
 
   const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
 
-  await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
-    return generateHlsPlaylistResolution({
-      video,
-      videoInputPath,
-      resolution: payload.resolution,
-      copyCodecs: payload.copyCodecs,
-      job
+  const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
+
+  try {
+    await videoFileInput.getVideo().reload()
+
+    await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
+      return generateHlsPlaylistResolution({
+        video,
+        videoInputPath,
+        inputFileMutexReleaser,
+        resolution: payload.resolution,
+        copyCodecs: payload.copyCodecs,
+        job
+      })
     })
-  })
+  } finally {
+    inputFileMutexReleaser()
+  }
 
   logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
@@ -177,38 +185,44 @@ async function onVideoFirstWebTorrentTranscoding (
   transcodeType: TranscodeVODOptionsType,
   user: MUserId
 ) {
-  const { resolution, audioStream } = await videoArg.probeMaxQualityFile()
-
-  // Maybe the video changed in database, refresh it
-  const videoDatabase = await VideoModel.loadFull(videoArg.uuid)
-  // Video does not exist anymore
-  if (!videoDatabase) return undefined
-
-  // Generate HLS version of the original file
-  const originalFileHLSPayload = {
-    ...payload,
-
-    hasAudio: !!audioStream,
-    resolution: videoDatabase.getMaxQualityFile().resolution,
-    // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
-    copyCodecs: transcodeType !== 'quick-transcode',
-    isMaxQuality: true
-  }
-  const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
-  const hasNewResolutions = await createLowerResolutionsJobs({
-    video: videoDatabase,
-    user,
-    videoFileResolution: resolution,
-    hasAudio: !!audioStream,
-    type: 'webtorrent',
-    isNewVideo: payload.isNewVideo ?? true
-  })
-
-  await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
-
-  // Move to next state if there are no other resolutions to generate
-  if (!hasHls && !hasNewResolutions) {
-    await retryTransactionWrapper(moveToNextState, { video: videoDatabase, isNewVideo: payload.isNewVideo })
+  const mutexReleaser = await VideoPathManager.Instance.lockFiles(videoArg.uuid)
+
+  try {
+    // Maybe the video changed in database, refresh it
+    const videoDatabase = await VideoModel.loadFull(videoArg.uuid)
+    // Video does not exist anymore
+    if (!videoDatabase) return undefined
+
+    const { resolution, audioStream } = await videoDatabase.probeMaxQualityFile()
+
+    // Generate HLS version of the original file
+    const originalFileHLSPayload = {
+      ...payload,
+
+      hasAudio: !!audioStream,
+      resolution: videoDatabase.getMaxQualityFile().resolution,
+      // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
+      copyCodecs: transcodeType !== 'quick-transcode',
+      isMaxQuality: true
+    }
+    const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
+    const hasNewResolutions = await createLowerResolutionsJobs({
+      video: videoDatabase,
+      user,
+      videoFileResolution: resolution,
+      hasAudio: !!audioStream,
+      type: 'webtorrent',
+      isNewVideo: payload.isNewVideo ?? true
+    })
+
+    await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
+
+    // Move to next state if there are no other resolutions to generate
+    if (!hasHls && !hasNewResolutions) {
+      await retryTransactionWrapper(moveToNextState, { video: videoDatabase, isNewVideo: payload.isNewVideo })
+    }
+  } finally {
+    mutexReleaser()
   }
 }
 
@@ -266,7 +280,7 @@ async function createLowerResolutionsJobs (options: {
 
   // Create transcoding jobs if there are enabled resolutions
   const resolutionsEnabled = await Hooks.wrapObject(
-    computeResolutionsToTranscode({ input: videoFileResolution, type: 'vod', includeInput: false, strictLower: true }),
+    computeResolutionsToTranscode({ input: videoFileResolution, type: 'vod', includeInput: false, strictLower: true, hasAudio }),
     'filter:transcoding.auto.resolutions-to-transcode.result',
     options
   )
@@ -274,8 +288,6 @@ async function createLowerResolutionsJobs (options: {
   const resolutionCreated: string[] = []
 
   for (const resolution of resolutionsEnabled) {
-    if (resolution === VideoResolution.H_NOVIDEO && hasAudio === false) continue
-
     let dataInput: VideoTranscodingPayload
 
     if (CONFIG.TRANSCODING.WEBTORRENT.ENABLED && type === 'webtorrent') {
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 655be6568..6bc59732f 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -63,6 +63,7 @@ import { processVideoLiveEnding } from './handlers/video-live-ending'
 import { processVideoStudioEdition } from './handlers/video-studio-edition'
 import { processVideoTranscoding } from './handlers/video-transcoding'
 import { processVideosViewsStats } from './handlers/video-views-stats'
+import { Redis } from '../redis'
 
 export type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -183,7 +184,7 @@ class JobQueue {
     }
 
     this.flowProducer = new FlowProducer({
-      connection: this.getRedisConnection(),
+      connection: Redis.getRedisClientOptions('FlowProducer'),
       prefix: this.jobRedisPrefix
     })
     this.flowProducer.on('error', err => { logger.error('Error in flow producer', { err }) })
@@ -196,7 +197,7 @@ class JobQueue {
       autorun: false,
       concurrency: this.getJobConcurrency(handlerName),
       prefix: this.jobRedisPrefix,
-      connection: this.getRedisConnection()
+      connection: Redis.getRedisClientOptions('Worker')
     }
 
     const handler = function (job: Job) {
@@ -236,7 +237,7 @@ class JobQueue {
 
   private buildQueue (handlerName: JobType) {
     const queueOptions: QueueOptions = {
-      connection: this.getRedisConnection(),
+      connection: Redis.getRedisClientOptions('Queue'),
       prefix: this.jobRedisPrefix
     }
 
@@ -249,7 +250,7 @@ class JobQueue {
   private buildQueueScheduler (handlerName: JobType) {
     const queueSchedulerOptions: QueueSchedulerOptions = {
       autorun: false,
-      connection: this.getRedisConnection(),
+      connection: Redis.getRedisClientOptions('QueueScheduler'),
       prefix: this.jobRedisPrefix,
       maxStalledCount: 10
     }
@@ -263,7 +264,7 @@ class JobQueue {
   private buildQueueEvent (handlerName: JobType) {
     const queueEventsOptions: QueueEventsOptions = {
       autorun: false,
-      connection: this.getRedisConnection(),
+      connection: Redis.getRedisClientOptions('QueueEvent'),
       prefix: this.jobRedisPrefix
     }
 
@@ -273,16 +274,6 @@ class JobQueue {
     this.queueEvents[handlerName] = queueEvents
   }
 
-  private getRedisConnection () {
-    return {
-      password: CONFIG.REDIS.AUTH,
-      db: CONFIG.REDIS.DB,
-      host: CONFIG.REDIS.HOSTNAME,
-      port: CONFIG.REDIS.PORT,
-      path: CONFIG.REDIS.SOCKET
-    }
-  }
-
   // ---------------------------------------------------------------------------
 
   async terminate () {