From f27b7a750f1876632e84d594608d3d64ce974efc Mon Sep 17 00:00:00 2001
From: Chocobozzz <me@florianbigard.com>
Date: Fri, 17 Jun 2022 14:08:13 +0200
Subject: Send views in a dedicated queue

---
 client/src/app/+admin/system/jobs/jobs.component.ts |  1 +
 server/initializers/constants.ts                    |  3 +++
 server/lib/activitypub/send/send-view.ts            |  2 +-
 server/lib/activitypub/send/shared/send-utils.ts    | 19 +++++++++++++++----
 server/lib/job-queue/job-queue.ts                   |  3 +++
 shared/models/server/job.model.ts                   |  1 +
 6 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/client/src/app/+admin/system/jobs/jobs.component.ts b/client/src/app/+admin/system/jobs/jobs.component.ts
index f7e10fd04..42f503be6 100644
--- a/client/src/app/+admin/system/jobs/jobs.component.ts
+++ b/client/src/app/+admin/system/jobs/jobs.component.ts
@@ -25,6 +25,7 @@ export class JobsComponent extends RestTable implements OnInit {
 
     'activitypub-follow',
     'activitypub-http-broadcast',
+    'activitypub-http-broadcast-parallel',
     'activitypub-http-fetcher',
     'activitypub-http-unicast',
     'activitypub-refresher',
diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts
index 6034da1c6..cd2cc33d3 100644
--- a/server/initializers/constants.ts
+++ b/server/initializers/constants.ts
@@ -139,6 +139,7 @@ const REMOTE_SCHEME = {
 
 const JOB_ATTEMPTS: { [id in JobType]: number } = {
   'activitypub-http-broadcast': 1,
+  'activitypub-http-broadcast-parallel': 1,
   'activitypub-http-unicast': 1,
   'activitypub-http-fetcher': 2,
   'activitypub-follow': 5,
@@ -159,6 +160,7 @@ const JOB_ATTEMPTS: { [id in JobType]: number } = {
 // Excluded keys are jobs that can be configured by admins
 const JOB_CONCURRENCY: { [id in Exclude<JobType, 'video-transcoding' | 'video-import'>]: number } = {
   'activitypub-http-broadcast': 1,
+  'activitypub-http-broadcast-parallel': 30,
   'activitypub-http-unicast': 10,
   'activitypub-http-fetcher': 3,
   'activitypub-cleaner': 1,
@@ -176,6 +178,7 @@ const JOB_CONCURRENCY: { [id in Exclude<JobType, 'video-transcoding' | 'video-im
 }
 const JOB_TTL: { [id in JobType]: number } = {
   'activitypub-http-broadcast': 60000 * 10, // 10 minutes
+  'activitypub-http-broadcast-parallel': 60000 * 10, // 10 minutes
   'activitypub-http-unicast': 60000 * 10, // 10 minutes
   'activitypub-http-fetcher': 1000 * 3600 * 10, // 10 hours
   'activitypub-follow': 60000 * 10, // 10 minutes
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/send-utils.ts b/server/lib/activitypub/send/shared/send-utils.ts
index 72368c762..fcec63991 100644
--- a/server/lib/activitypub/send/shared/send-utils.ts
+++ b/server/lib/activitypub/send/shared/send-utils.ts
@@ -15,9 +15,10 @@ async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAud
   byActor: MActorLight
   video: MVideoImmutable | MVideoAccountLight
   contextType: ContextType
+  parallelizable?: boolean
   transaction?: Transaction
 }) {
-  const { byActor, video, transaction, contextType } = options
+  const { byActor, video, transaction, contextType, parallelizable } = options
 
   // Send to origin
   if (video.isOwned() === false) {
@@ -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 d3776c3bf..f339e9135 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,
@@ -89,6 +91,7 @@ const handlers: { [id in JobType]: (job: Job) => 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/shared/models/server/job.model.ts b/shared/models/server/job.model.ts
index 073f15872..4633ab769 100644
--- a/shared/models/server/job.model.ts
+++ b/shared/models/server/job.model.ts
@@ -9,6 +9,7 @@ export type JobState = 'active' | 'completed' | 'failed' | 'waiting' | 'delayed'
 export type JobType =
   | 'activitypub-http-unicast'
   | 'activitypub-http-broadcast'
+  | 'activitypub-http-broadcast-parallel'
   | 'activitypub-http-fetcher'
   | 'activitypub-cleaner'
   | 'activitypub-follow'
-- 
cgit v1.2.3