13 files changed, 67 insertions, 30 deletions

diff --git a/server/lib/activitypub/process/process-accept.ts b/server/lib/activitypub/process/process-accept.ts
index ebb275e34..72bb1975e 100644
--- a/server/lib/activitypub/process/process-accept.ts
+++ b/server/lib/activitypub/process/process-accept.ts
@@ -2,8 +2,10 @@ import { ActivityAccept } from '../../../../shared/models/activitypub'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 import { addFetchOutboxJob } from '../actor'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processAcceptActivity (activity: ActivityAccept, targetActor: ActorModel, inboxActor?: ActorModel) {
+async function processAcceptActivity (options: APProcessorOptions<ActivityAccept>) {
+  const { byActor: targetActor, inboxActor } = options
   if (inboxActor === undefined) throw new Error('Need to accept on explicit inbox.')
 
   return processAccept(inboxActor, targetActor)
diff --git a/server/lib/activitypub/process/process-announce.ts b/server/lib/activitypub/process/process-announce.ts
index 1fe347506..7a59bb84d 100644
--- a/server/lib/activitypub/process/process-announce.ts
+++ b/server/lib/activitypub/process/process-announce.ts
@@ -8,9 +8,14 @@ import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { Notifier } from '../../notifier'
 import { VideoModel } from '../../../models/video/video'
 import { logger } from '../../../helpers/logger'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processAnnounceActivity (activity: ActivityAnnounce, actorAnnouncer: ActorModel) {
-  return retryTransactionWrapper(processVideoShare, actorAnnouncer, activity)
+async function processAnnounceActivity (options: APProcessorOptions<ActivityAnnounce>) {
+  const { activity, byActor: actorAnnouncer } = options
+  // Only notify if it is not from a fetcher job
+  const notify = options.fromFetch !== true
+
+  return retryTransactionWrapper(processVideoShare, actorAnnouncer, activity, notify)
 }
 
 // ---------------------------------------------------------------------------
@@ -21,7 +26,7 @@ export {
 
 // ---------------------------------------------------------------------------
 
-async function processVideoShare (actorAnnouncer: ActorModel, activity: ActivityAnnounce) {
+async function processVideoShare (actorAnnouncer: ActorModel, activity: ActivityAnnounce, notify: boolean) {
   const objectUri = typeof activity.object === 'string' ? activity.object : activity.object.id
 
   let video: VideoModel
@@ -63,5 +68,5 @@ async function processVideoShare (actorAnnouncer: ActorModel, activity: Activity
     return undefined
   })
 
-  if (videoCreated) Notifier.Instance.notifyOnNewVideoIfNeeded(video)
+  if (videoCreated && notify) Notifier.Instance.notifyOnNewVideoIfNeeded(video)
 }
diff --git a/server/lib/activitypub/process/process-create.ts b/server/lib/activitypub/process/process-create.ts
index 1e893cdeb..a979771b6 100644
--- a/server/lib/activitypub/process/process-create.ts
+++ b/server/lib/activitypub/process/process-create.ts
@@ -12,17 +12,22 @@ import { Notifier } from '../../notifier'
 import { PlaylistObject } from '../../../../shared/models/activitypub/objects/playlist-object'
 import { createOrUpdateVideoPlaylist } from '../playlist'
 import { VideoModel } from '../../../models/video/video'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processCreateActivity (activity: ActivityCreate, byActor: ActorModel) {
+async function processCreateActivity (options: APProcessorOptions<ActivityCreate>) {
+  const { activity, byActor } = options
+
+  // Only notify if it is not from a fetcher job
+  const notify = options.fromFetch !== true
   const activityObject = activity.object
   const activityType = activityObject.type
 
   if (activityType === 'Video') {
-    return processCreateVideo(activity)
+    return processCreateVideo(activity, notify)
   }
 
   if (activityType === 'Note') {
-    return retryTransactionWrapper(processCreateVideoComment, activity, byActor)
+    return retryTransactionWrapper(processCreateVideoComment, activity, byActor, notify)
   }
 
   if (activityType === 'CacheFile') {
@@ -45,12 +50,12 @@ export {
 
 // ---------------------------------------------------------------------------
 
-async function processCreateVideo (activity: ActivityCreate) {
+async function processCreateVideo (activity: ActivityCreate, notify: boolean) {
   const videoToCreateData = activity.object as VideoTorrentObject
 
   const { video, created } = await getOrCreateVideoAndAccountAndChannel({ videoObject: videoToCreateData })
 
-  if (created) Notifier.Instance.notifyOnNewVideoIfNeeded(video)
+  if (created && notify) Notifier.Instance.notifyOnNewVideoIfNeeded(video)
 
   return video
 }
@@ -71,7 +76,7 @@ async function processCreateCacheFile (activity: ActivityCreate, byActor: ActorM
   }
 }
 
-async function processCreateVideoComment (activity: ActivityCreate, byActor: ActorModel) {
+async function processCreateVideoComment (activity: ActivityCreate, byActor: ActorModel, notify: boolean) {
   const commentObject = activity.object as VideoCommentObject
   const byAccount = byActor.Account
 
@@ -99,7 +104,7 @@ async function processCreateVideoComment (activity: ActivityCreate, byActor: Act
     await forwardVideoRelatedActivity(activity, undefined, exceptions, video)
   }
 
-  if (created === true) Notifier.Instance.notifyOnNewComment(comment)
+  if (created && notify) Notifier.Instance.notifyOnNewComment(comment)
 }
 
 async function processCreatePlaylist (activity: ActivityCreate, byActor: ActorModel) {
diff --git a/server/lib/activitypub/process/process-delete.ts b/server/lib/activitypub/process/process-delete.ts
index 6f10a50bd..845a7b249 100644
--- a/server/lib/activitypub/process/process-delete.ts
+++ b/server/lib/activitypub/process/process-delete.ts
@@ -9,8 +9,11 @@ import { VideoChannelModel } from '../../../models/video/video-channel'
 import { VideoCommentModel } from '../../../models/video/video-comment'
 import { forwardVideoRelatedActivity } from '../send/utils'
 import { VideoPlaylistModel } from '../../../models/video/video-playlist'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
+
+async function processDeleteActivity (options: APProcessorOptions<ActivityDelete>) {
+  const { activity, byActor } = options
 
-async function processDeleteActivity (activity: ActivityDelete, byActor: ActorModel) {
   const objectUrl = typeof activity.object === 'string' ? activity.object : activity.object.id
 
   if (activity.actor === objectUrl) {
diff --git a/server/lib/activitypub/process/process-dislike.ts b/server/lib/activitypub/process/process-dislike.ts
index f06269f8b..a457e5f17 100644
--- a/server/lib/activitypub/process/process-dislike.ts
+++ b/server/lib/activitypub/process/process-dislike.ts
@@ -7,8 +7,10 @@ import { ActorModel } from '../../../models/activitypub/actor'
 import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { forwardVideoRelatedActivity } from '../send/utils'
 import { getVideoDislikeActivityPubUrl } from '../url'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processDislikeActivity (activity: ActivityCreate | ActivityDislike, byActor: ActorModel) {
+async function processDislikeActivity (options: APProcessorOptions<ActivityCreate | ActivityDislike>) {
+  const { activity, byActor } = options
   return retryTransactionWrapper(processDislike, activity, byActor)
 }
 
diff --git a/server/lib/activitypub/process/process-flag.ts b/server/lib/activitypub/process/process-flag.ts
index 8faab051e..532545e58 100644
--- a/server/lib/activitypub/process/process-flag.ts
+++ b/server/lib/activitypub/process/process-flag.ts
@@ -8,8 +8,10 @@ import { VideoAbuseModel } from '../../../models/video/video-abuse'
 import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { Notifier } from '../../notifier'
 import { getAPId } from '../../../helpers/activitypub'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processFlagActivity (activity: ActivityCreate | ActivityFlag, byActor: ActorModel) {
+async function processFlagActivity (options: APProcessorOptions<ActivityCreate | ActivityFlag>) {
+  const { activity, byActor } = options
   return retryTransactionWrapper(processCreateVideoAbuse, activity, byActor)
 }
 
diff --git a/server/lib/activitypub/process/process-follow.ts b/server/lib/activitypub/process/process-follow.ts
index ed16ba172..8fe9975f6 100644
--- a/server/lib/activitypub/process/process-follow.ts
+++ b/server/lib/activitypub/process/process-follow.ts
@@ -9,8 +9,10 @@ import { Notifier } from '../../notifier'
 import { getAPId } from '../../../helpers/activitypub'
 import { getServerActor } from '../../../helpers/utils'
 import { CONFIG } from '../../../initializers/config'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processFollowActivity (activity: ActivityFollow, byActor: ActorModel) {
+async function processFollowActivity (options: APProcessorOptions<ActivityFollow>) {
+  const { activity, byActor } = options
   const activityObject = getAPId(activity.object)
 
   return retryTransactionWrapper(processFollow, byActor, activityObject)
diff --git a/server/lib/activitypub/process/process-like.ts b/server/lib/activitypub/process/process-like.ts
index bba54a19b..706e63c41 100644
--- a/server/lib/activitypub/process/process-like.ts
+++ b/server/lib/activitypub/process/process-like.ts
@@ -7,8 +7,10 @@ import { forwardVideoRelatedActivity } from '../send/utils'
 import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { getVideoLikeActivityPubUrl } from '../url'
 import { getAPId } from '../../../helpers/activitypub'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processLikeActivity (activity: ActivityLike, byActor: ActorModel) {
+async function processLikeActivity (options: APProcessorOptions<ActivityLike>) {
+  const { activity, byActor } = options
   return retryTransactionWrapper(processLikeVideo, byActor, activity)
 }
 
diff --git a/server/lib/activitypub/process/process-reject.ts b/server/lib/activitypub/process/process-reject.ts
index 709a65096..9181906b4 100644
--- a/server/lib/activitypub/process/process-reject.ts
+++ b/server/lib/activitypub/process/process-reject.ts
@@ -2,8 +2,10 @@ import { ActivityReject } from '../../../../shared/models/activitypub/activity'
 import { sequelizeTypescript } from '../../../initializers'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processRejectActivity (activity: ActivityReject, targetActor: ActorModel, inboxActor?: ActorModel) {
+async function processRejectActivity (options: APProcessorOptions<ActivityReject>) {
+  const { byActor: targetActor, inboxActor } = options
   if (inboxActor === undefined) throw new Error('Need to reject on explicit inbox.')
 
   return processReject(inboxActor, targetActor)
diff --git a/server/lib/activitypub/process/process-undo.ts b/server/lib/activitypub/process/process-undo.ts
index 692c51904..7a0f90adf 100644
--- a/server/lib/activitypub/process/process-undo.ts
+++ b/server/lib/activitypub/process/process-undo.ts
@@ -10,8 +10,10 @@ import { forwardVideoRelatedActivity } from '../send/utils'
 import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { VideoShareModel } from '../../../models/video/video-share'
 import { VideoRedundancyModel } from '../../../models/redundancy/video-redundancy'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processUndoActivity (activity: ActivityUndo, byActor: ActorModel) {
+async function processUndoActivity (options: APProcessorOptions<ActivityUndo>) {
+  const { activity, byActor } = options
   const activityToUndo = activity.object
 
   if (activityToUndo.type === 'Like') {
diff --git a/server/lib/activitypub/process/process-update.ts b/server/lib/activitypub/process/process-update.ts
index 71a16dacc..1e11dd1fd 100644
--- a/server/lib/activitypub/process/process-update.ts
+++ b/server/lib/activitypub/process/process-update.ts
@@ -14,8 +14,11 @@ import { createOrUpdateCacheFile } from '../cache-file'
 import { forwardVideoRelatedActivity } from '../send/utils'
 import { PlaylistObject } from '../../../../shared/models/activitypub/objects/playlist-object'
 import { createOrUpdateVideoPlaylist } from '../playlist'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
+
+async function processUpdateActivity (options: APProcessorOptions<ActivityUpdate>) {
+  const { activity, byActor } = options
 
-async function processUpdateActivity (activity: ActivityUpdate, byActor: ActorModel) {
   const objectType = activity.object.type
 
   if (objectType === 'Video') {
diff --git a/server/lib/activitypub/process/process-view.ts b/server/lib/activitypub/process/process-view.ts
index 8f66d3630..0170b74f4 100644
--- a/server/lib/activitypub/process/process-view.ts
+++ b/server/lib/activitypub/process/process-view.ts
@@ -3,8 +3,10 @@ import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { forwardVideoRelatedActivity } from '../send/utils'
 import { Redis } from '../../redis'
 import { ActivityCreate, ActivityView, ViewObject } from '../../../../shared/models/activitypub'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-async function processViewActivity (activity: ActivityView | ActivityCreate, byActor: ActorModel) {
+async function processViewActivity (options: APProcessorOptions<ActivityCreate | ActivityView>) {
+  const { activity, byActor } = options
   return processCreateView(activity, byActor)
 }
 
diff --git a/server/lib/activitypub/process/process.ts b/server/lib/activitypub/process/process.ts
index 9dd241402..f4a92e341 100644
--- a/server/lib/activitypub/process/process.ts
+++ b/server/lib/activitypub/process/process.ts
@@ -15,8 +15,9 @@ import { getOrCreateActorAndServerAndModel } from '../actor'
 import { processDislikeActivity } from './process-dislike'
 import { processFlagActivity } from './process-flag'
 import { processViewActivity } from './process-view'
+import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
 
-const processActivity: { [ P in ActivityType ]: (activity: Activity, byActor: ActorModel, inboxActor?: ActorModel) => Promise<any> } = {
+const processActivity: { [ P in ActivityType ]: (options: APProcessorOptions<Activity>) => Promise<any> } = {
   Create: processCreateActivity,
   Update: processUpdateActivity,
   Delete: processDeleteActivity,
@@ -37,11 +38,15 @@ async function processActivities (
     signatureActor?: ActorModel
     inboxActor?: ActorModel
     outboxUrl?: string
-  } = {}) {
+    fromFetch?: boolean
+  } = {}
+) {
+  const { outboxUrl, signatureActor, inboxActor, fromFetch = false } = options
+
   const actorsCache: { [ url: string ]: ActorModel } = {}
 
   for (const activity of activities) {
-    if (!options.signatureActor && [ 'Create', 'Announce', 'Like' ].includes(activity.type) === false) {
+    if (!signatureActor && [ 'Create', 'Announce', 'Like' ].includes(activity.type) === false) {
       logger.error('Cannot process activity %s (type: %s) without the actor signature.', activity.id, activity.type)
       continue
     }
@@ -49,17 +54,17 @@ async function processActivities (
     const actorUrl = getAPId(activity.actor)
 
     // When we fetch remote data, we don't have signature
-    if (options.signatureActor && actorUrl !== options.signatureActor.url) {
-      logger.warn('Signature mismatch between %s and %s, skipping.', actorUrl, options.signatureActor.url)
+    if (signatureActor && actorUrl !== signatureActor.url) {
+      logger.warn('Signature mismatch between %s and %s, skipping.', actorUrl, signatureActor.url)
       continue
     }
 
-    if (options.outboxUrl && checkUrlsSameHost(options.outboxUrl, actorUrl) !== true) {
-      logger.warn('Host mismatch between outbox URL %s and actor URL %s, skipping.', options.outboxUrl, actorUrl)
+    if (outboxUrl && checkUrlsSameHost(outboxUrl, actorUrl) !== true) {
+      logger.warn('Host mismatch between outbox URL %s and actor URL %s, skipping.', outboxUrl, actorUrl)
       continue
     }
 
-    const byActor = options.signatureActor || actorsCache[actorUrl] || await getOrCreateActorAndServerAndModel(actorUrl)
+    const byActor = signatureActor || actorsCache[actorUrl] || await getOrCreateActorAndServerAndModel(actorUrl)
     actorsCache[actorUrl] = byActor
 
     const activityProcessor = processActivity[activity.type]
@@ -69,7 +74,7 @@ async function processActivities (
     }
 
     try {
-      await activityProcessor(activity, byActor, options.inboxActor)
+      await activityProcessor({ activity, byActor, inboxActor: inboxActor, fromFetch })
     } catch (err) {
       logger.warn('Cannot process activity %s.', activity.type, { err })
     }