6 files changed, 28 insertions, 13 deletions

diff --git a/server/lib/activitypub/local-video-viewer.ts b/server/lib/activitypub/local-video-viewer.ts
index 738083adc..bdd746791 100644
--- a/server/lib/activitypub/local-video-viewer.ts
+++ b/server/lib/activitypub/local-video-viewer.ts
@@ -23,7 +23,7 @@ async function createOrUpdateLocalVideoViewer (watchAction: WatchActionObject, v
       : null,
 
     videoId: video.id
-  })
+  }, { transaction: t })
 
   await LocalVideoViewerWatchSectionModel.bulkCreateSections({
     localVideoViewerId: localVideoViewer.id,
@@ -31,7 +31,9 @@ async function createOrUpdateLocalVideoViewer (watchAction: WatchActionObject, v
     watchSections: watchAction.watchSections.map(s => ({
       start: s.startTimestamp,
       end: s.endTimestamp
-    }))
+    })),
+
+    transaction: t
   })
 }
 
diff --git a/server/lib/activitypub/send/send-view.ts b/server/lib/activitypub/send/send-view.ts
index 25a20ec6d..bf3451603 100644
--- a/server/lib/activitypub/send/send-view.ts
+++ b/server/lib/activitypub/send/send-view.ts
@@ -26,7 +26,7 @@ async function sendView (options: {
     return buildViewActivity({ url, byActor, video, audience, type })
   }
 
-  return sendVideoRelatedActivity(activityBuilder, { byActor, video, transaction, contextType: 'View' })
+  return sendVideoRelatedActivity(activityBuilder, { byActor, video, transaction, contextType: 'View', parallelizable: true })
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/activitypub/send/shared/audience-utils.ts b/server/lib/activitypub/send/shared/audience-utils.ts
index ba4be487c..2f6b0741d 100644
--- a/server/lib/activitypub/send/shared/audience-utils.ts
+++ b/server/lib/activitypub/send/shared/audience-utils.ts
@@ -3,7 +3,7 @@ import { ACTIVITY_PUB } from '@server/initializers/constants'
 import { ActorModel } from '@server/models/actor/actor'
 import { VideoModel } from '@server/models/video/video'
 import { VideoShareModel } from '@server/models/video/video-share'
-import { MActorFollowersUrl, MActorLight, MActorUrl, MCommentOwner, MCommentOwnerVideo, MVideoId } from '@server/types/models'
+import { MActorFollowersUrl, MActorUrl, MCommentOwner, MCommentOwnerVideo, MVideoId } from '@server/types/models'
 import { ActivityAudience } from '@shared/models'
 
 function getOriginVideoAudience (accountActor: MActorUrl, actorsInvolvedInVideo: MActorFollowersUrl[] = []): ActivityAudience {
@@ -51,13 +51,13 @@ function getAudienceFromFollowersOf (actorsInvolvedInObject: MActorFollowersUrl[
 }
 
 async function getActorsInvolvedInVideo (video: MVideoId, t: Transaction) {
-  const actors: MActorLight[] = await VideoShareModel.loadActorsByShare(video.id, t)
+  const actors = await VideoShareModel.listActorIdsAndFollowerUrlsByShare(video.id, t)
 
   const videoAll = video as VideoModel
 
   const videoActor = videoAll.VideoChannel?.Account
     ? videoAll.VideoChannel.Account.Actor
-    : await ActorModel.loadFromAccountByVideoId(video.id, t)
+    : await ActorModel.loadAccountActorFollowerUrlByVideoId(video.id, t)
 
   actors.push(videoActor)
 
diff --git a/server/lib/activitypub/send/shared/send-utils.ts b/server/lib/activitypub/send/shared/send-utils.ts
index dbcde91ee..fcec63991 100644
--- a/server/lib/activitypub/send/shared/send-utils.ts
+++ b/server/lib/activitypub/send/shared/send-utils.ts
@@ -15,17 +15,18 @@ async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAud
   byActor: MActorLight
   video: MVideoImmutable | MVideoAccountLight
   contextType: ContextType
+  parallelizable?: boolean
   transaction?: Transaction
 }) {
-  const { byActor, video, transaction, contextType } = options
-
-  const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, transaction)
+  const { byActor, video, transaction, contextType, parallelizable } = options
 
   // Send to origin
   if (video.isOwned() === false) {
     return sendVideoActivityToOrigin(activityBuilder, options)
   }
 
+  const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, transaction)
+
   // Send to followers
   const audience = getAudienceFromFollowersOf(actorsInvolvedInVideo)
   const activity = activityBuilder(audience)
@@ -38,6 +39,7 @@ async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAud
     toFollowersOf: actorsInvolvedInVideo,
     transaction,
     actorsException,
+    parallelizable,
     contextType
   })
 }
@@ -130,9 +132,10 @@ async function broadcastToFollowers (options: {
   transaction: Transaction
   contextType: ContextType
 
+  parallelizable?: boolean
   actorsException?: MActorWithInboxes[]
 }) {
-  const { data, byActor, toFollowersOf, transaction, contextType, actorsException = [] } = options
+  const { data, byActor, toFollowersOf, transaction, contextType, actorsException = [], parallelizable } = options
 
   const uris = await computeFollowerUris(toFollowersOf, actorsException, transaction)
 
@@ -141,6 +144,7 @@ async function broadcastToFollowers (options: {
       uris,
       data,
       byActor,
+      parallelizable,
       contextType
     })
   })
@@ -173,8 +177,9 @@ function broadcastTo (options: {
   data: any
   byActor: MActorId
   contextType: ContextType
+  parallelizable?: boolean // default to false
 }) {
-  const { uris, data, byActor, contextType } = options
+  const { uris, data, byActor, contextType, parallelizable } = options
 
   if (uris.length === 0) return undefined
 
@@ -200,7 +205,13 @@ function broadcastTo (options: {
       contextType
     }
 
-    JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
+    JobQueue.Instance.createJob({
+      type: parallelizable
+        ? 'activitypub-http-broadcast-parallel'
+        : 'activitypub-http-broadcast',
+
+      payload
+    })
   }
 
   for (const unicastUri of unicastUris) {
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 61e41fb0f..ce24763f1 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -43,6 +43,7 @@ import { processVideosViewsStats } from './handlers/video-views-stats'
 
 type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
+  { type: 'activitypub-http-broadcast-parallel', payload: ActivitypubHttpBroadcastPayload } |
   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
   { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
   { type: 'activitypub-http-cleaner', payload: {} } |
@@ -68,6 +69,7 @@ export type CreateJobOptions = {
 
 const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
+  'activitypub-http-broadcast-parallel': processActivityPubHttpBroadcast,
   'activitypub-http-unicast': processActivityPubHttpUnicast,
   'activitypub-http-fetcher': processActivityPubHttpFetcher,
   'activitypub-cleaner': processActivityPubCleaner,
@@ -93,6 +95,7 @@ const errorHandlers: { [id in JobType]?: (job: Job, err: any) => Promise<any> }
 const jobTypes: JobType[] = [
   'activitypub-follow',
   'activitypub-http-broadcast',
+  'activitypub-http-broadcast-parallel',
   'activitypub-http-fetcher',
   'activitypub-http-unicast',
   'activitypub-cleaner',
diff --git a/server/lib/views/shared/video-viewer-counters.ts b/server/lib/views/shared/video-viewer-counters.ts
index 941b62ed7..999ab7d8d 100644
--- a/server/lib/views/shared/video-viewer-counters.ts
+++ b/server/lib/views/shared/video-viewer-counters.ts
@@ -1,4 +1,3 @@
-
 import { isTestInstance } from '@server/helpers/core-utils'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { VIEW_LIFETIME } from '@server/initializers/constants'