Add more AP stats to stats endpoint

It will help to understand if the federation correctly works or not

2 files changed, 9 insertions, 8 deletions

diff --git a/server/lib/activitypub/inbox-manager.ts b/server/lib/activitypub/inbox-manager.ts
index 18ae49532..282e7ce66 100644
--- a/server/lib/activitypub/inbox-manager.ts
+++ b/server/lib/activitypub/inbox-manager.ts
@@ -1,10 +1,10 @@
-import { AsyncQueue, queue } from 'async'
+import { queue, QueueObject } from 'async'
 import { logger } from '@server/helpers/logger'
 import { SCHEDULER_INTERVALS_MS } from '@server/initializers/constants'
 import { MActorDefault, MActorSignature } from '@server/types/models'
 import { Activity } from '@shared/models'
-import { processActivities } from './process'
 import { StatsManager } from '../stat-manager'
+import { processActivities } from './process'
 
 type QueueParam = {
   activities: Activity[]
@@ -16,16 +16,12 @@ class InboxManager {
 
   private static instance: InboxManager
 
-  private readonly inboxQueue: AsyncQueue<QueueParam>
-
-  private messagesProcessed = 0
+  private readonly inboxQueue: QueueObject<QueueParam>
 
   private constructor () {
     this.inboxQueue = queue<QueueParam, Error>((task, cb) => {
       const options = { signatureActor: task.signatureActor, inboxActor: task.inboxActor }
 
-      this.messagesProcessed++
-
       processActivities(task.activities, options)
         .then(() => cb())
         .catch(err => {
@@ -35,7 +31,7 @@ class InboxManager {
     })
 
     setInterval(() => {
-      StatsManager.Instance.updateInboxStats(this.messagesProcessed, this.getActivityPubMessagesWaiting())
+      StatsManager.Instance.updateInboxWaiting(this.getActivityPubMessagesWaiting())
     }, SCHEDULER_INTERVALS_MS.updateInboxStats)
   }
 
diff --git a/server/lib/activitypub/process/process.ts b/server/lib/activitypub/process/process.ts
index e60dd2a5b..5cef75665 100644
--- a/server/lib/activitypub/process/process.ts
+++ b/server/lib/activitypub/process/process.ts
@@ -16,6 +16,7 @@ import { processFlagActivity } from './process-flag'
 import { processViewActivity } from './process-view'
 import { APProcessorOptions } from '../../../types/activitypub-processor.model'
 import { MActorDefault, MActorSignature } from '../../../types/models'
+import { StatsManager } from '@server/lib/stat-manager'
 
 const processActivity: { [ P in ActivityType ]: (options: APProcessorOptions<Activity>) => Promise<any> } = {
   Create: processCreateActivity,
@@ -75,8 +76,12 @@ async function processActivities (
 
     try {
       await activityProcessor({ activity, byActor, inboxActor, fromFetch })
+
+      StatsManager.Instance.addInboxProcessedSuccess(activity.type)
     } catch (err) {
       logger.warn('Cannot process activity %s.', activity.type, { err })
+
+      StatsManager.Instance.addInboxProcessedError(activity.type)
     }
   }
 }