2 files changed, 149 insertions, 0 deletions

diff --git a/server/lib/activitypub/inbox-manager.ts b/server/lib/activitypub/inbox-manager.ts
new file mode 100644
index 000000000..19e112f91
--- /dev/null
+++ b/server/lib/activitypub/inbox-manager.ts
@@ -0,0 +1,55 @@
+import { AsyncQueue, queue } 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'
+
+type QueueParam = {
+  activities: Activity[]
+  signatureActor?: MActorSignature
+  inboxActor?: MActorDefault
+}
+
+class InboxManager {
+
+  private static instance: InboxManager
+
+  private readonly inboxQueue: AsyncQueue<QueueParam>
+
+  private messagesProcessed = 0
+
+  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 => {
+          logger.error('Error in process activities.', { err })
+          cb()
+        })
+    })
+
+    setInterval(() => {
+      StatsManager.Instance.updateInboxStats(this.messagesProcessed, this.inboxQueue.length())
+    }, SCHEDULER_INTERVALS_MS.updateInboxStats)
+  }
+
+  addInboxMessage (options: QueueParam) {
+    this.inboxQueue.push(options)
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  InboxManager
+}
diff --git a/server/lib/stat-manager.ts b/server/lib/stat-manager.ts
new file mode 100644
index 000000000..f9d69b0dc
--- /dev/null
+++ b/server/lib/stat-manager.ts
@@ -0,0 +1,94 @@
+import { CONFIG } from '@server/initializers/config'
+import { UserModel } from '@server/models/account/user'
+import { ActorFollowModel } from '@server/models/activitypub/actor-follow'
+import { VideoRedundancyModel } from '@server/models/redundancy/video-redundancy'
+import { VideoModel } from '@server/models/video/video'
+import { VideoCommentModel } from '@server/models/video/video-comment'
+import { VideoFileModel } from '@server/models/video/video-file'
+import { ServerStats, VideoRedundancyStrategyWithManual } from '@shared/models'
+
+class StatsManager {
+
+  private static instance: StatsManager
+
+  private readonly instanceStartDate = new Date()
+
+  private inboxMessagesProcessed = 0
+  private inboxMessagesWaiting = 0
+
+  private constructor () {}
+
+  updateInboxStats (inboxMessagesProcessed: number, inboxMessagesWaiting: number) {
+    this.inboxMessagesProcessed = inboxMessagesProcessed
+    this.inboxMessagesWaiting = inboxMessagesWaiting
+  }
+
+  async getStats () {
+    const { totalLocalVideos, totalLocalVideoViews, totalVideos } = await VideoModel.getStats()
+    const { totalLocalVideoComments, totalVideoComments } = await VideoCommentModel.getStats()
+    const { totalUsers, totalDailyActiveUsers, totalWeeklyActiveUsers, totalMonthlyActiveUsers } = await UserModel.getStats()
+    const { totalInstanceFollowers, totalInstanceFollowing } = await ActorFollowModel.getStats()
+    const { totalLocalVideoFilesSize } = await VideoFileModel.getStats()
+
+    const videosRedundancyStats = await this.buildRedundancyStats()
+
+    const data: ServerStats = {
+      totalLocalVideos,
+      totalLocalVideoViews,
+      totalLocalVideoFilesSize,
+      totalLocalVideoComments,
+      totalVideos,
+      totalVideoComments,
+
+      totalUsers,
+      totalDailyActiveUsers,
+      totalWeeklyActiveUsers,
+      totalMonthlyActiveUsers,
+
+      totalInstanceFollowers,
+      totalInstanceFollowing,
+
+      videosRedundancy: videosRedundancyStats,
+
+      totalActivityPubMessagesProcessed: this.inboxMessagesProcessed,
+      activityPubMessagesProcessedPerSecond: this.buildActivityPubMessagesProcessedPerSecond(),
+      totalActivityPubMessagesWaiting: this.inboxMessagesWaiting
+    }
+
+    return data
+  }
+
+  private buildActivityPubMessagesProcessedPerSecond () {
+    const now = new Date()
+    const startedSeconds = (now.getTime() - this.instanceStartDate.getTime()) / 1000
+
+    return this.inboxMessagesProcessed / startedSeconds
+  }
+
+  private buildRedundancyStats () {
+    const strategies = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES
+                                               .map(r => ({
+                                                 strategy: r.strategy as VideoRedundancyStrategyWithManual,
+                                                 size: r.size
+                                               }))
+
+    strategies.push({ strategy: 'manual', size: null })
+
+    return Promise.all(
+      strategies.map(r => {
+        return VideoRedundancyModel.getStats(r.strategy)
+          .then(stats => Object.assign(stats, { strategy: r.strategy, totalSize: r.size }))
+      })
+    )
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  StatsManager
+}