11 files changed, 220 insertions, 65 deletions

diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts
index 36d0f237b..b4d381062 100644
--- a/server/lib/job-queue/handlers/activitypub-follow.ts
+++ b/server/lib/job-queue/handlers/activitypub-follow.ts
@@ -8,6 +8,7 @@ import { getOrCreateActorAndServerAndModel } from '../../activitypub/actor'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 import { ActorModel } from '../../../models/activitypub/actor'
+import { Notifier } from '../../notifier'
 
 export type ActivitypubFollowPayload = {
   followerActorId: number
@@ -42,7 +43,7 @@ export {
 
 // ---------------------------------------------------------------------------
 
-function follow (fromActor: ActorModel, targetActor: ActorModel) {
+async function follow (fromActor: ActorModel, targetActor: ActorModel) {
   if (fromActor.id === targetActor.id) {
     throw new Error('Follower is the same than target actor.')
   }
@@ -50,7 +51,7 @@ function follow (fromActor: ActorModel, targetActor: ActorModel) {
   // Same server, direct accept
   const state = !fromActor.serverId && !targetActor.serverId ? 'accepted' : 'pending'
 
-  return sequelizeTypescript.transaction(async t => {
+  const actorFollow = await sequelizeTypescript.transaction(async t => {
     const [ actorFollow ] = await ActorFollowModel.findOrCreate({
       where: {
         actorId: fromActor.id,
@@ -68,5 +69,9 @@ function follow (fromActor: ActorModel, targetActor: ActorModel) {
 
     // Send a notification to remote server if our follow is not already accepted
     if (actorFollow.state !== 'accepted') await sendFollow(actorFollow)
+
+    return actorFollow
   })
+
+  if (actorFollow.state === 'accepted') Notifier.Instance.notifyOfNewFollow(actorFollow)
 }
diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index 03a9e12a4..9493945ff 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -3,8 +3,9 @@ import * as Bluebird from 'bluebird'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
-import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
+import { buildGlobalHeaders, buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
 import { BROADCAST_CONCURRENCY, JOB_REQUEST_TIMEOUT } from '../../../initializers'
+import { ActorFollowScoreCache } from '../../cache'
 
 export type ActivitypubHttpBroadcastPayload = {
   uris: string[]
@@ -25,7 +26,8 @@ async function processActivityPubHttpBroadcast (job: Bull.Job) {
     uri: '',
     json: body,
     httpSignature: httpSignatureOptions,
-    timeout: JOB_REQUEST_TIMEOUT
+    timeout: JOB_REQUEST_TIMEOUT,
+    headers: buildGlobalHeaders(body)
   }
 
   const badUrls: string[] = []
@@ -37,7 +39,7 @@ async function processActivityPubHttpBroadcast (job: Bull.Job) {
       .catch(() => badUrls.push(uri))
   }, { concurrency: BROADCAST_CONCURRENCY })
 
-  return ActorFollowModel.updateActorFollowsScore(goodUrls, badUrls, undefined)
+  return ActorFollowScoreCache.Instance.updateActorFollowsScore(goodUrls, badUrls)
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
index 42217c27c..67ccfa995 100644
--- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
@@ -23,7 +23,7 @@ async function processActivityPubHttpFetcher (job: Bull.Job) {
   if (payload.videoId) video = await VideoModel.loadAndPopulateAccountAndServerAndTags(payload.videoId)
 
   const fetcherType: { [ id in FetchType ]: (items: any[]) => Promise<any> } = {
-    'activity': items => processActivities(items),
+    'activity': items => processActivities(items, { outboxUrl: payload.uri }),
     'video-likes': items => createRates(items, video, 'like'),
     'video-dislikes': items => createRates(items, video, 'dislike'),
     'video-shares': items => addVideoShares(items, video),
diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
index c90d735f6..3973dcdc8 100644
--- a/server/lib/job-queue/handlers/activitypub-http-unicast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
@@ -1,9 +1,9 @@
 import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
-import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
-import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
+import { buildGlobalHeaders, buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
 import { JOB_REQUEST_TIMEOUT } from '../../../initializers'
+import { ActorFollowScoreCache } from '../../cache'
 
 export type ActivitypubHttpUnicastPayload = {
   uri: string
@@ -25,14 +25,15 @@ async function processActivityPubHttpUnicast (job: Bull.Job) {
     uri,
     json: body,
     httpSignature: httpSignatureOptions,
-    timeout: JOB_REQUEST_TIMEOUT
+    timeout: JOB_REQUEST_TIMEOUT,
+    headers: buildGlobalHeaders(body)
   }
 
   try {
     await doRequest(options)
-    ActorFollowModel.updateActorFollowsScore([ uri ], [], undefined)
+    ActorFollowScoreCache.Instance.updateActorFollowsScore([ uri ], [])
   } catch (err) {
-    ActorFollowModel.updateActorFollowsScore([], [ uri ], undefined)
+    ActorFollowScoreCache.Instance.updateActorFollowsScore([], [ uri ])
 
     throw err
   }
diff --git a/server/lib/job-queue/handlers/activitypub-refresher.ts b/server/lib/job-queue/handlers/activitypub-refresher.ts
new file mode 100644
index 000000000..454b975fe
--- /dev/null
+++ b/server/lib/job-queue/handlers/activitypub-refresher.ts
@@ -0,0 +1,54 @@
+import * as Bull from 'bull'
+import { logger } from '../../../helpers/logger'
+import { fetchVideoByUrl } from '../../../helpers/video'
+import { refreshVideoIfNeeded, refreshActorIfNeeded } from '../../activitypub'
+import { ActorModel } from '../../../models/activitypub/actor'
+
+export type RefreshPayload = {
+  type: 'video' | 'actor'
+  url: string
+}
+
+async function refreshAPObject (job: Bull.Job) {
+  const payload = job.data as RefreshPayload
+
+  logger.info('Processing AP refresher in job %d for %s.', job.id, payload.url)
+
+  if (payload.type === 'video') return refreshVideo(payload.url)
+  if (payload.type === 'actor') return refreshActor(payload.url)
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  refreshActor,
+  refreshAPObject
+}
+
+// ---------------------------------------------------------------------------
+
+async function refreshVideo (videoUrl: string) {
+  const fetchType = 'all' as 'all'
+  const syncParam = { likes: true, dislikes: true, shares: true, comments: true, thumbnail: true }
+
+  const videoFromDatabase = await fetchVideoByUrl(videoUrl, fetchType)
+  if (videoFromDatabase) {
+    const refreshOptions = {
+      video: videoFromDatabase,
+      fetchedType: fetchType,
+      syncParam
+    }
+
+    await refreshVideoIfNeeded(refreshOptions)
+  }
+}
+
+async function refreshActor (actorUrl: string) {
+  const fetchType = 'all' as 'all'
+  const actor = await ActorModel.loadByUrlAndPopulateAccountAndChannel(actorUrl)
+
+  if (actor) {
+    await refreshActorIfNeeded(actor, fetchType)
+  }
+
+}
diff --git a/server/lib/job-queue/handlers/email.ts b/server/lib/job-queue/handlers/email.ts
index 73d98ae54..220d0af32 100644
--- a/server/lib/job-queue/handlers/email.ts
+++ b/server/lib/job-queue/handlers/email.ts
@@ -6,13 +6,14 @@ export type EmailPayload = {
   to: string[]
   subject: string
   text: string
+  from?: string
 }
 
 async function processEmail (job: Bull.Job) {
   const payload = job.data as EmailPayload
   logger.info('Processing email in job %d.', job.id)
 
-  return Emailer.Instance.sendMail(payload.to, payload.subject, payload.text)
+  return Emailer.Instance.sendMail(payload.to, payload.subject, payload.text, payload.from)
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
index 36092665e..4961d4502 100644
--- a/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
+++ b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
@@ -1,8 +1,12 @@
 import { buildSignedActivity } from '../../../../helpers/activitypub'
 import { getServerActor } from '../../../../helpers/utils'
 import { ActorModel } from '../../../../models/activitypub/actor'
+import { sha256 } from '../../../../helpers/core-utils'
+import { HTTP_SIGNATURE } from '../../../../initializers'
 
-async function computeBody (payload: { body: any, signatureActorId?: number }) {
+type Payload = { body: any, signatureActorId?: number }
+
+async function computeBody (payload: Payload) {
   let body = payload.body
 
   if (payload.signatureActorId) {
@@ -14,7 +18,7 @@ async function computeBody (payload: { body: any, signatureActorId?: number }) {
   return body
 }
 
-async function buildSignedRequestOptions (payload: { signatureActorId?: number }) {
+async function buildSignedRequestOptions (payload: Payload) {
   let actor: ActorModel | null
   if (payload.signatureActorId) {
     actor = await ActorModel.load(payload.signatureActorId)
@@ -26,14 +30,29 @@ async function buildSignedRequestOptions (payload: { signatureActorId?: number }
 
   const keyId = actor.getWebfingerUrl()
   return {
-    algorithm: 'rsa-sha256',
-    authorizationHeaderName: 'Signature',
+    algorithm: HTTP_SIGNATURE.ALGORITHM,
+    authorizationHeaderName: HTTP_SIGNATURE.HEADER_NAME,
     keyId,
-    key: actor.privateKey
+    key: actor.privateKey,
+    headers: HTTP_SIGNATURE.HEADERS_TO_SIGN
+  }
+}
+
+function buildGlobalHeaders (body: any) {
+  return {
+    'Digest': buildDigest(body)
   }
 }
 
+function buildDigest (body: any) {
+  const rawBody = typeof body === 'string' ? body : JSON.stringify(body)
+
+  return 'SHA-256=' + sha256(rawBody, 'base64')
+}
+
 export {
+  buildDigest,
+  buildGlobalHeaders,
   computeBody,
   buildSignedRequestOptions
 }
diff --git a/server/lib/job-queue/handlers/video-file.ts b/server/lib/job-queue/handlers/video-file.ts
index 1463c93fc..04983155c 100644
--- a/server/lib/job-queue/handlers/video-file.ts
+++ b/server/lib/job-queue/handlers/video-file.ts
@@ -5,16 +5,18 @@ import { VideoModel } from '../../../models/video/video'
 import { JobQueue } from '../job-queue'
 import { federateVideoIfNeeded } from '../../activitypub'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
-import { sequelizeTypescript } from '../../../initializers'
+import { sequelizeTypescript, CONFIG } from '../../../initializers'
 import * as Bluebird from 'bluebird'
 import { computeResolutionsToTranscode } from '../../../helpers/ffmpeg-utils'
-import { importVideoFile, transcodeOriginalVideofile, optimizeOriginalVideofile } from '../../video-transcoding'
+import { generateHlsPlaylist, importVideoFile, optimizeVideofile, transcodeOriginalVideofile } from '../../video-transcoding'
+import { Notifier } from '../../notifier'
 
 export type VideoFilePayload = {
   videoUUID: string
-  isNewVideo?: boolean
   resolution?: VideoResolution
+  isNewVideo?: boolean
   isPortraitMode?: boolean
+  generateHlsPlaylist?: boolean
 }
 
 export type VideoFileImportPayload = {
@@ -50,34 +52,51 @@ async function processVideoFile (job: Bull.Job) {
     return undefined
   }
 
-  // Transcoding in other resolution
-  if (payload.resolution) {
+  if (payload.generateHlsPlaylist) {
+    await generateHlsPlaylist(video, payload.resolution, payload.isPortraitMode || false)
+
+    await retryTransactionWrapper(onHlsPlaylistGenerationSuccess, video)
+  } else if (payload.resolution) { // Transcoding in other resolution
     await transcodeOriginalVideofile(video, payload.resolution, payload.isPortraitMode || false)
 
-    await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, video)
+    await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, video, payload)
   } else {
-    await optimizeOriginalVideofile(video)
+    await optimizeVideofile(video)
 
-    await retryTransactionWrapper(onVideoFileOptimizerSuccess, video, payload.isNewVideo)
+    await retryTransactionWrapper(onVideoFileOptimizerSuccess, video, payload)
   }
 
   return video
 }
 
-async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) {
+async function onHlsPlaylistGenerationSuccess (video: VideoModel) {
+  if (video === undefined) return undefined
+
+  await sequelizeTypescript.transaction(async t => {
+    // Maybe the video changed in database, refresh it
+    let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
+    // Video does not exist anymore
+    if (!videoDatabase) return undefined
+
+    // If the video was not published, we consider it is a new one for other instances
+    await federateVideoIfNeeded(videoDatabase, false, t)
+  })
+}
+
+async function onVideoFileTranscoderOrImportSuccess (video: VideoModel, payload?: VideoFilePayload) {
   if (video === undefined) return undefined
 
-  return sequelizeTypescript.transaction(async t => {
+  const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => {
     // Maybe the video changed in database, refresh it
     let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
     // Video does not exist anymore
     if (!videoDatabase) return undefined
 
-    let isNewVideo = false
+    let videoPublished = false
 
     // We transcoded the video file in another format, now we can publish it
     if (videoDatabase.state !== VideoState.PUBLISHED) {
-      isNewVideo = true
+      videoPublished = true
 
       videoDatabase.state = VideoState.PUBLISHED
       videoDatabase.publishedAt = new Date()
@@ -85,21 +104,29 @@ async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) {
     }
 
     // If the video was not published, we consider it is a new one for other instances
-    await federateVideoIfNeeded(videoDatabase, isNewVideo, t)
+    await federateVideoIfNeeded(videoDatabase, videoPublished, t)
 
-    return undefined
+    return { videoDatabase, videoPublished }
   })
+
+  // don't notify prior to scheduled video update
+  if (videoPublished && !videoDatabase.ScheduleVideoUpdate) {
+    Notifier.Instance.notifyOnNewVideo(videoDatabase)
+    Notifier.Instance.notifyOnPendingVideoPublished(videoDatabase)
+  }
+
+  await createHlsJobIfEnabled(payload)
 }
 
-async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boolean) {
-  if (video === undefined) return undefined
+async function onVideoFileOptimizerSuccess (videoArg: VideoModel, payload: VideoFilePayload) {
+  if (videoArg === undefined) return undefined
 
   // Outside the transaction (IO on disk)
-  const { videoFileResolution } = await video.getOriginalFileResolution()
+  const { videoFileResolution } = await videoArg.getOriginalFileResolution()
 
-  return sequelizeTypescript.transaction(async t => {
+  const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => {
     // Maybe the video changed in database, refresh it
-    const videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
+    let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoArg.uuid, t)
     // Video does not exist anymore
     if (!videoDatabase) return undefined
 
@@ -110,8 +137,10 @@ async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boole
       { resolutions: resolutionsEnabled }
     )
 
+    let videoPublished = false
+
     if (resolutionsEnabled.length !== 0) {
-      const tasks: Bluebird<any>[] = []
+      const tasks: Bluebird<Bull.Job<any>>[] = []
 
       for (const resolution of resolutionsEnabled) {
         const dataInput = {
@@ -127,15 +156,27 @@ async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boole
 
       logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled })
     } else {
+      videoPublished = true
+
       // No transcoding to do, it's now published
-      video.state = VideoState.PUBLISHED
-      video = await video.save({ transaction: t })
+      videoDatabase.state = VideoState.PUBLISHED
+      videoDatabase = await videoDatabase.save({ transaction: t })
 
-      logger.info('No transcoding jobs created for video %s (no resolutions).', video.uuid)
+      logger.info('No transcoding jobs created for video %s (no resolutions).', videoDatabase.uuid, { privacy: videoDatabase.privacy })
     }
 
-    return federateVideoIfNeeded(video, isNewVideo, t)
+    await federateVideoIfNeeded(videoDatabase, payload.isNewVideo, t)
+
+    return { videoDatabase, videoPublished }
   })
+
+  // don't notify prior to scheduled video update
+  if (!videoDatabase.ScheduleVideoUpdate) {
+    if (payload.isNewVideo) Notifier.Instance.notifyOnNewVideo(videoDatabase)
+    if (videoPublished) Notifier.Instance.notifyOnPendingVideoPublished(videoDatabase)
+  }
+
+  await createHlsJobIfEnabled(Object.assign({}, payload, { resolution: videoDatabase.getOriginalFile().resolution }))
 }
 
 // ---------------------------------------------------------------------------
@@ -144,3 +185,20 @@ export {
   processVideoFile,
   processVideoFileImport
 }
+
+// ---------------------------------------------------------------------------
+
+function createHlsJobIfEnabled (payload?: VideoFilePayload) {
+  // Generate HLS playlist?
+  if (payload && CONFIG.TRANSCODING.HLS.ENABLED) {
+    const hlsTranscodingPayload = {
+      videoUUID: payload.videoUUID,
+      resolution: payload.resolution,
+      isPortraitMode: payload.isPortraitMode,
+
+      generateHlsPlaylist: true
+    }
+
+    return JobQueue.Instance.createJob({ type: 'video-file', payload: hlsTranscodingPayload })
+  }
+}
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index e3f2a276c..12004dcd7 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -6,15 +6,16 @@ import { VideoImportState } from '../../../../shared/models/videos'
 import { getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffmpeg-utils'
 import { extname, join } from 'path'
 import { VideoFileModel } from '../../../models/video/video-file'
-import { CONFIG, sequelizeTypescript, VIDEO_IMPORT_TIMEOUT } from '../../../initializers'
-import { doRequestAndSaveToFile } from '../../../helpers/requests'
+import { CONFIG, PREVIEWS_SIZE, sequelizeTypescript, THUMBNAILS_SIZE, VIDEO_IMPORT_TIMEOUT } from '../../../initializers'
+import { downloadImage } from '../../../helpers/requests'
 import { VideoState } from '../../../../shared'
 import { JobQueue } from '../index'
 import { federateVideoIfNeeded } from '../../activitypub'
 import { VideoModel } from '../../../models/video/video'
 import { downloadWebTorrentVideo } from '../../../helpers/webtorrent'
 import { getSecureTorrentName } from '../../../helpers/utils'
-import { remove, rename, stat } from 'fs-extra'
+import { remove, move, stat } from 'fs-extra'
+import { Notifier } from '../../notifier'
 
 type VideoImportYoutubeDLPayload = {
   type: 'youtube-dl'
@@ -109,6 +110,7 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
   let tempVideoPath: string
   let videoDestFile: string
   let videoFile: VideoFileModel
+
   try {
     // Download video from youtubeDL
     tempVideoPath = await downloader()
@@ -133,19 +135,18 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
       videoId: videoImport.videoId
     }
     videoFile = new VideoFileModel(videoFileData)
-    // Import if the import fails, to clean files
+    // To clean files if the import fails
     videoImport.Video.VideoFiles = [ videoFile ]
 
     // Move file
     videoDestFile = join(CONFIG.STORAGE.VIDEOS_DIR, videoImport.Video.getVideoFilename(videoFile))
-    await rename(tempVideoPath, videoDestFile)
+    await move(tempVideoPath, videoDestFile)
     tempVideoPath = null // This path is not used anymore
 
     // Process thumbnail
     if (options.downloadThumbnail) {
       if (options.thumbnailUrl) {
-        const destThumbnailPath = join(CONFIG.STORAGE.THUMBNAILS_DIR, videoImport.Video.getThumbnailName())
-        await doRequestAndSaveToFile({ method: 'GET', uri: options.thumbnailUrl }, destThumbnailPath)
+        await downloadImage(options.thumbnailUrl, CONFIG.STORAGE.THUMBNAILS_DIR, videoImport.Video.getThumbnailName(), THUMBNAILS_SIZE)
       } else {
         await videoImport.Video.createThumbnail(videoFile)
       }
@@ -156,8 +157,7 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
     // Process preview
     if (options.downloadPreview) {
       if (options.thumbnailUrl) {
-        const destPreviewPath = join(CONFIG.STORAGE.PREVIEWS_DIR, videoImport.Video.getPreviewName())
-        await doRequestAndSaveToFile({ method: 'GET', uri: options.thumbnailUrl }, destPreviewPath)
+        await downloadImage(options.thumbnailUrl, CONFIG.STORAGE.PREVIEWS_DIR, videoImport.Video.getPreviewName(), PREVIEWS_SIZE)
       } else {
         await videoImport.Video.createPreview(videoFile)
       }
@@ -180,7 +180,7 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
       // Update video DB object
       video.duration = duration
       video.state = CONFIG.TRANSCODING.ENABLED ? VideoState.TO_TRANSCODE : VideoState.PUBLISHED
-      const videoUpdated = await video.save({ transaction: t })
+      await video.save({ transaction: t })
 
       // Now we can federate the video (reload from database, we need more attributes)
       const videoForFederation = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
@@ -192,10 +192,13 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
 
       logger.info('Video %s imported.', video.uuid)
 
-      videoImportUpdated.Video = videoUpdated
+      videoImportUpdated.Video = videoForFederation
       return videoImportUpdated
     })
 
+    Notifier.Instance.notifyOnNewVideo(videoImportUpdated.Video)
+    Notifier.Instance.notifyOnFinishedVideoImport(videoImportUpdated, true)
+
     // Create transcoding jobs?
     if (videoImportUpdated.Video.state === VideoState.TO_TRANSCODE) {
       // Put uuid because we don't have id auto incremented for now
@@ -218,6 +221,8 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
     videoImport.state = VideoImportState.FAILED
     await videoImport.save()
 
+    Notifier.Instance.notifyOnFinishedVideoImport(videoImport, false)
+
     throw err
   }
 }
diff --git a/server/lib/job-queue/handlers/video-views.ts b/server/lib/job-queue/handlers/video-views.ts
index cf180a11a..fa1fd13b3 100644
--- a/server/lib/job-queue/handlers/video-views.ts
+++ b/server/lib/job-queue/handlers/video-views.ts
@@ -3,8 +3,9 @@ import { logger } from '../../../helpers/logger'
 import { VideoModel } from '../../../models/video/video'
 import { VideoViewModel } from '../../../models/video/video-views'
 import { isTestInstance } from '../../../helpers/core-utils'
+import { federateVideoIfNeeded } from '../../activitypub'
 
-async function processVideosViewsViews () {
+async function processVideosViews () {
   const lastHour = new Date()
 
   // In test mode, we run this function multiple times per hour, so we don't want the values of the previous hour
@@ -22,13 +23,9 @@ async function processVideosViewsViews () {
   for (const videoId of videoIds) {
     try {
       const views = await Redis.Instance.getVideoViews(videoId, hour)
-      if (isNaN(views)) {
-        logger.error('Cannot process videos views of video %d in hour %d: views number is NaN.', videoId, hour)
-      } else {
+      if (views) {
         logger.debug('Adding %d views to video %d in hour %d.', views, videoId, hour)
 
-        await VideoModel.incrementViews(videoId, views)
-
         try {
           await VideoViewModel.create({
             startDate,
@@ -36,6 +33,16 @@ async function processVideosViewsViews () {
             views,
             videoId
           })
+
+          const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
+          if (video.isOwned()) {
+            // If this is a remote video, the origin instance will send us an update
+            await VideoModel.incrementViews(videoId, views)
+
+            // Send video update
+            video.views += views
+            await federateVideoIfNeeded(video, false)
+          }
         } catch (err) {
           logger.debug('Cannot create video views for video %d in hour %d. Maybe the video does not exist anymore?', videoId, hour)
         }
@@ -51,5 +58,5 @@ async function processVideosViewsViews () {
 // ---------------------------------------------------------------------------
 
 export {
-  processVideosViewsViews
+  processVideosViews
 }
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 0696ba43c..ba9cbe0d9 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -10,7 +10,8 @@ import { EmailPayload, processEmail } from './handlers/email'
 import { processVideoFile, processVideoFileImport, VideoFileImportPayload, VideoFilePayload } from './handlers/video-file'
 import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
 import { processVideoImport, VideoImportPayload } from './handlers/video-import'
-import { processVideosViewsViews } from './handlers/video-views'
+import { processVideosViews } from './handlers/video-views'
+import { refreshAPObject, RefreshPayload } from './handlers/activitypub-refresher'
 
 type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -21,6 +22,7 @@ type CreateJobArgument =
   { type: 'video-file', payload: VideoFilePayload } |
   { type: 'email', payload: EmailPayload } |
   { type: 'video-import', payload: VideoImportPayload } |
+  { type: 'activitypub-refresher', payload: RefreshPayload } |
   { type: 'videos-views', payload: {} }
 
 const handlers: { [ id in JobType ]: (job: Bull.Job) => Promise<any>} = {
@@ -32,7 +34,8 @@ const handlers: { [ id in JobType ]: (job: Bull.Job) => Promise<any>} = {
   'video-file': processVideoFile,
   'email': processEmail,
   'video-import': processVideoImport,
-  'videos-views': processVideosViewsViews
+  'videos-views': processVideosViews,
+  'activitypub-refresher': refreshAPObject
 }
 
 const jobTypes: JobType[] = [
@@ -44,7 +47,8 @@ const jobTypes: JobType[] = [
   'video-file',
   'video-file-import',
   'video-import',
-  'videos-views'
+  'videos-views',
+  'activitypub-refresher'
 ]
 
 class JobQueue {
@@ -84,7 +88,6 @@ class JobQueue {
 
       queue.on('error', err => {
         logger.error('Error in job queue %s.', handlerName, { err })
-        process.exit(-1)
       })
 
       this.queues[handlerName] = queue
@@ -162,10 +165,10 @@ class JobQueue {
     return total
   }
 
-  removeOldJobs () {
+  async removeOldJobs () {
     for (const key of Object.keys(this.queues)) {
       const queue = this.queues[key]
-      queue.clean(JOB_COMPLETED_LIFETIME, 'completed')
+      await queue.clean(JOB_COMPLETED_LIFETIME, 'completed')
     }
   }