5 files changed, 71 insertions, 56 deletions

diff --git a/server/lib/job-queue/handlers/activitypub-cleaner.ts b/server/lib/job-queue/handlers/activitypub-cleaner.ts
index d5e4508fe..509dd1cb5 100644
--- a/server/lib/job-queue/handlers/activitypub-cleaner.ts
+++ b/server/lib/job-queue/handlers/activitypub-cleaner.ts
@@ -8,36 +8,35 @@ import {
 } from '@server/helpers/custom-validators/activitypub/activity'
 import { sanitizeAndCheckVideoCommentObject } from '@server/helpers/custom-validators/activitypub/video-comments'
 import { doJSONRequest, PeerTubeRequestError } from '@server/helpers/requests'
-import { AP_CLEANER_CONCURRENCY } from '@server/initializers/constants'
+import { AP_CLEANER } from '@server/initializers/constants'
+import { Redis } from '@server/lib/redis'
 import { VideoModel } from '@server/models/video/video'
 import { VideoCommentModel } from '@server/models/video/video-comment'
 import { VideoShareModel } from '@server/models/video/video-share'
 import { HttpStatusCode } from '@shared/models'
-import { logger } from '../../../helpers/logger'
+import { logger, loggerTagsFactory } from '../../../helpers/logger'
 import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
 
+const lTags = loggerTagsFactory('ap-cleaner')
+
 // Job to clean remote interactions off local videos
 
 async function processActivityPubCleaner (_job: Job) {
-  logger.info('Processing ActivityPub cleaner.')
+  logger.info('Processing ActivityPub cleaner.', lTags())
 
   {
     const rateUrls = await AccountVideoRateModel.listRemoteRateUrlsOfLocalVideos()
     const { bodyValidator, deleter, updater } = rateOptionsFactory()
 
     await map(rateUrls, async rateUrl => {
-      try {
-        const result = await updateObjectIfNeeded(rateUrl, bodyValidator, updater, deleter)
+      const result = await updateObjectIfNeeded({ url: rateUrl, bodyValidator, updater, deleter })
 
-        if (result?.status === 'deleted') {
-          const { videoId, type } = result.data
+      if (result?.status === 'deleted') {
+        const { videoId, type } = result.data
 
-          await VideoModel.updateRatesOf(videoId, type, undefined)
-        }
-      } catch (err) {
-        logger.warn('Cannot update/delete remote AP rate %s.', rateUrl, { err })
+        await VideoModel.updateRatesOf(videoId, type, undefined)
       }
-    }, { concurrency: AP_CLEANER_CONCURRENCY })
+    }, { concurrency: AP_CLEANER.CONCURRENCY })
   }
 
   {
@@ -45,12 +44,8 @@ async function processActivityPubCleaner (_job: Job) {
     const { bodyValidator, deleter, updater } = shareOptionsFactory()
 
     await map(shareUrls, async shareUrl => {
-      try {
-        await updateObjectIfNeeded(shareUrl, bodyValidator, updater, deleter)
-      } catch (err) {
-        logger.warn('Cannot update/delete remote AP share %s.', shareUrl, { err })
-      }
-    }, { concurrency: AP_CLEANER_CONCURRENCY })
+      await updateObjectIfNeeded({ url: shareUrl, bodyValidator, updater, deleter })
+    }, { concurrency: AP_CLEANER.CONCURRENCY })
   }
 
   {
@@ -58,12 +53,8 @@ async function processActivityPubCleaner (_job: Job) {
     const { bodyValidator, deleter, updater } = commentOptionsFactory()
 
     await map(commentUrls, async commentUrl => {
-      try {
-        await updateObjectIfNeeded(commentUrl, bodyValidator, updater, deleter)
-      } catch (err) {
-        logger.warn('Cannot update/delete remote AP comment %s.', commentUrl, { err })
-      }
-    }, { concurrency: AP_CLEANER_CONCURRENCY })
+      await updateObjectIfNeeded({ url: commentUrl, bodyValidator, updater, deleter })
+    }, { concurrency: AP_CLEANER.CONCURRENCY })
   }
 }
 
@@ -75,14 +66,16 @@ export {
 
 // ---------------------------------------------------------------------------
 
-async function updateObjectIfNeeded <T> (
-  url: string,
-  bodyValidator: (body: any) => boolean,
-  updater: (url: string, newUrl: string) => Promise<T>,
-  deleter: (url: string) => Promise<T>
+async function updateObjectIfNeeded <T> (options: {
+  url: string
+  bodyValidator: (body: any) => boolean
+  updater: (url: string, newUrl: string) => Promise<T>
+  deleter: (url: string) => Promise<T> }
 ): Promise<{ data: T, status: 'deleted' | 'updated' } | null> {
+  const { url, bodyValidator, updater, deleter } = options
+
   const on404OrTombstone = async () => {
-    logger.info('Removing remote AP object %s.', url)
+    logger.info('Removing remote AP object %s.', url, lTags(url))
     const data = await deleter(url)
 
     return { status: 'deleted' as 'deleted', data }
@@ -104,7 +97,7 @@ async function updateObjectIfNeeded <T> (
         throw new Error(`New url ${newUrl} has not the same host than old url ${url}`)
       }
 
-      logger.info('Updating remote AP object %s.', url)
+      logger.info('Updating remote AP object %s.', url, lTags(url))
       const data = await updater(url, newUrl)
 
       return { status: 'updated', data }
@@ -117,7 +110,15 @@ async function updateObjectIfNeeded <T> (
       return on404OrTombstone()
     }
 
-    throw err
+    logger.debug('Remote AP object %s is unavailable.', url, lTags(url))
+
+    const unavailability = await Redis.Instance.addAPUnavailability(url)
+    if (unavailability >= AP_CLEANER.UNAVAILABLE_TRESHOLD) {
+      logger.info('Removing unavailable AP resource %s.', url, lTags(url))
+      return on404OrTombstone()
+    }
+
+    return null
   }
 }
 
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 54a7c566b..9e39322a8 100644
--- a/server/lib/job-queue/handlers/move-to-object-storage.ts
+++ b/server/lib/job-queue/handlers/move-to-object-storage.ts
@@ -2,16 +2,16 @@ import { Job } from 'bull'
 import { remove } from 'fs-extra'
 import { join } from 'path'
 import { logger } from '@server/helpers/logger'
-import { updateTorrentUrls } from '@server/helpers/webtorrent'
+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 { getHLSDirectory, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
-import { moveToNextState } from '@server/lib/video-state'
+import { moveToFailedMoveToObjectStorageState, moveToNextState } from '@server/lib/video-state'
 import { VideoModel } from '@server/models/video/video'
 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
 import { MStreamingPlaylistVideo, MVideo, MVideoFile, MVideoWithAllFiles } from '@server/types/models'
-import { MoveObjectStoragePayload, VideoStorage } from '../../../../shared'
+import { MoveObjectStoragePayload, VideoStorage } from '@shared/models'
 
 export async function processMoveToObjectStorage (job: Job) {
   const payload = job.data as MoveObjectStoragePayload
@@ -24,18 +24,25 @@ export async function processMoveToObjectStorage (job: Job) {
     return undefined
   }
 
-  if (video.VideoFiles) {
-    await moveWebTorrentFiles(video)
-  }
+  try {
+    if (video.VideoFiles) {
+      await moveWebTorrentFiles(video)
+    }
 
-  if (video.VideoStreamingPlaylists) {
-    await moveHLSFiles(video)
-  }
+    if (video.VideoStreamingPlaylists) {
+      await moveHLSFiles(video)
+    }
+
+    const pendingMove = await VideoJobInfoModel.decrease(video.uuid, 'pendingMove')
+    if (pendingMove === 0) {
+      logger.info('Running cleanup after moving files to object storage (video %s in job %d)', video.uuid, job.id)
+      await doAfterLastJob(video, payload.isNewVideo)
+    }
+  } catch (err) {
+    logger.error('Cannot move video %s to object storage.', video.url, { err })
 
-  const pendingMove = await VideoJobInfoModel.decrease(video.uuid, 'pendingMove')
-  if (pendingMove === 0) {
-    logger.info('Running cleanup after moving files to object storage (video %s in job %d)', video.uuid, job.id)
-    await doAfterLastJob(video, payload.isNewVideo)
+    await moveToFailedMoveToObjectStorageState(video)
+    await VideoJobInfoModel.abortAllTasks(video.uuid, 'pendingMove')
   }
 
   return payload.videoUUID
@@ -113,7 +120,7 @@ async function onFileMoved (options: {
   file.fileUrl = fileUrl
   file.storage = VideoStorage.OBJECT_STORAGE
 
-  await updateTorrentUrls(videoOrPlaylist, file)
+  await updateTorrentMetadata(videoOrPlaylist, file)
   await file.save()
 
   logger.debug('Removing %s because it\'s now on object storage', oldPath)
diff --git a/server/lib/job-queue/handlers/video-file-import.ts b/server/lib/job-queue/handlers/video-file-import.ts
index a91c2ef80..0d9e80cb8 100644
--- a/server/lib/job-queue/handlers/video-file-import.ts
+++ b/server/lib/job-queue/handlers/video-file-import.ts
@@ -1,6 +1,6 @@
 import { Job } from 'bull'
 import { copy, stat } from 'fs-extra'
-import { getLowercaseExtension } from '@server/helpers/core-utils'
+import { getLowercaseExtension } from '@shared/core-utils'
 import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
 import { CONFIG } from '@server/initializers/config'
 import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index 4ce1a6c30..2f74e9fbd 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -1,6 +1,5 @@
 import { Job } from 'bull'
 import { move, remove, stat } from 'fs-extra'
-import { getLowercaseExtension } from '@server/helpers/core-utils'
 import { retryTransactionWrapper } from '@server/helpers/database-utils'
 import { YoutubeDLWrapper } from '@server/helpers/youtube-dl'
 import { isPostImportVideoAccepted } from '@server/lib/moderation'
@@ -13,17 +12,20 @@ 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 { getLowercaseExtension } from '@shared/core-utils'
+import { isAudioFile } from '@shared/extra-utils'
 import {
+  ThumbnailType,
   VideoImportPayload,
+  VideoImportState,
   VideoImportTorrentPayload,
   VideoImportTorrentPayloadType,
   VideoImportYoutubeDLPayload,
   VideoImportYoutubeDLPayloadType,
+  VideoResolution,
   VideoState
-} from '../../../../shared'
-import { VideoImportState } from '../../../../shared/models/videos'
-import { ThumbnailType } from '../../../../shared/models/videos/thumbnail.type'
-import { getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffprobe-utils'
+} 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'
@@ -114,9 +116,14 @@ async function processFile (downloader: () => Promise<string>, videoImport: MVid
       throw new Error('The user video quota is exceeded with this video to import.')
     }
 
-    const { resolution } = await getVideoFileResolution(tempVideoPath)
-    const fps = await getVideoFileFPS(tempVideoPath)
-    const duration = await getDurationFromVideoFile(tempVideoPath)
+    const probe = await ffprobePromise(tempVideoPath)
+
+    const { resolution } = await isAudioFile(tempVideoPath, probe)
+      ? { resolution: VideoResolution.H_NOVIDEO }
+      : await getVideoFileResolution(tempVideoPath)
+
+    const fps = await getVideoFileFPS(tempVideoPath, probe)
+    const duration = await getDurationFromVideoFile(tempVideoPath, probe)
 
     // Prepare video file object for creation in database
     const fileExt = getLowercaseExtension(tempVideoPath)
diff --git a/server/lib/job-queue/handlers/video-transcoding.ts b/server/lib/job-queue/handlers/video-transcoding.ts
index 0edcdcba3..ef3abcbcd 100644
--- a/server/lib/job-queue/handlers/video-transcoding.ts
+++ b/server/lib/job-queue/handlers/video-transcoding.ts
@@ -12,7 +12,7 @@ import {
   NewResolutionTranscodingPayload,
   OptimizeTranscodingPayload,
   VideoTranscodingPayload
-} from '../../../../shared'
+} from '@shared/models'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
 import { computeLowerResolutionsToTranscode } from '../../../helpers/ffprobe-utils'
 import { logger, loggerTagsFactory } from '../../../helpers/logger'