8 files changed, 80 insertions, 41 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 abbd89b3b..9493945ff 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -5,6 +5,7 @@ import { doRequest } from '../../../helpers/requests'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 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[]
@@ -38,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-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
index d36479032..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 { buildGlobalHeaders, buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
 import { JOB_REQUEST_TIMEOUT } from '../../../initializers'
+import { ActorFollowScoreCache } from '../../cache'
 
 export type ActivitypubHttpUnicastPayload = {
   uri: string
@@ -31,9 +31,9 @@ async function processActivityPubHttpUnicast (job: Bull.Job) {
 
   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
index 7752b3b40..454b975fe 100644
--- a/server/lib/job-queue/handlers/activitypub-refresher.ts
+++ b/server/lib/job-queue/handlers/activitypub-refresher.ts
@@ -1,29 +1,33 @@
 import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { fetchVideoByUrl } from '../../../helpers/video'
-import { refreshVideoIfNeeded } from '../../activitypub'
+import { refreshVideoIfNeeded, refreshActorIfNeeded } from '../../activitypub'
+import { ActorModel } from '../../../models/activitypub/actor'
 
 export type RefreshPayload = {
-  videoUrl: string
-  type: 'video'
+  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.', job.id)
 
-  if (payload.type === 'video') return refreshAPVideo(payload.videoUrl)
+  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 refreshAPVideo (videoUrl: string) {
+async function refreshVideo (videoUrl: string) {
   const fetchType = 'all' as 'all'
   const syncParam = { likes: true, dislikes: true, shares: true, comments: true, thumbnail: true }
 
@@ -38,3 +42,13 @@ async function refreshAPVideo (videoUrl: string) {
     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/video-file.ts b/server/lib/job-queue/handlers/video-file.ts
index adc0a2a15..593e43cc5 100644
--- a/server/lib/job-queue/handlers/video-file.ts
+++ b/server/lib/job-queue/handlers/video-file.ts
@@ -8,7 +8,8 @@ import { retryTransactionWrapper } from '../../../helpers/database-utils'
 import { sequelizeTypescript } from '../../../initializers'
 import * as Bluebird from 'bluebird'
 import { computeResolutionsToTranscode } from '../../../helpers/ffmpeg-utils'
-import { importVideoFile, transcodeOriginalVideofile, optimizeVideofile } from '../../video-transcoding'
+import { importVideoFile, optimizeVideofile, transcodeOriginalVideofile } from '../../video-transcoding'
+import { Notifier } from '../../notifier'
 
 export type VideoFilePayload = {
   videoUUID: string
@@ -67,17 +68,17 @@ async function processVideoFile (job: Bull.Job) {
 async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) {
   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 +86,26 @@ 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 }
   })
+
+  if (videoPublished) {
+    Notifier.Instance.notifyOnNewVideo(videoDatabase)
+    Notifier.Instance.notifyOnPendingVideoPublished(videoDatabase)
+  }
 }
 
-async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boolean) {
-  if (video === undefined) return undefined
+async function onVideoFileOptimizerSuccess (videoArg: VideoModel, isNewVideo: boolean) {
+  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 +116,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 +135,22 @@ 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, isNewVideo, t)
+
+    return { videoDatabase, videoPublished }
   })
+
+  if (isNewVideo) Notifier.Instance.notifyOnNewVideo(videoDatabase)
+  if (videoPublished) Notifier.Instance.notifyOnPendingVideoPublished(videoDatabase)
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index 4de901c0c..12004dcd7 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -7,14 +7,15 @@ import { getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } fro
 import { extname, join } from 'path'
 import { VideoFileModel } from '../../../models/video/video-file'
 import { CONFIG, PREVIEWS_SIZE, sequelizeTypescript, THUMBNAILS_SIZE, VIDEO_IMPORT_TIMEOUT } from '../../../initializers'
-import { doRequestAndSaveToFile, downloadImage } from '../../../helpers/requests'
+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()
@@ -138,14 +140,13 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
 
     // 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 downloadImage(options.thumbnailUrl, destThumbnailPath, THUMBNAILS_SIZE)
+        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 downloadImage(options.thumbnailUrl, destPreviewPath, PREVIEWS_SIZE)
+        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 038ef43e2..fa1fd13b3 100644
--- a/server/lib/job-queue/handlers/video-views.ts
+++ b/server/lib/job-queue/handlers/video-views.ts
@@ -23,9 +23,7 @@ async function processVideosViews () {
   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 (%s).', videoId, hour, views)
-      } else {
+      if (views) {
         logger.debug('Adding %d views to video %d in hour %d.', views, videoId, hour)
 
         try {