Bufferize videos views in redis

4 files changed, 144 insertions, 8 deletions

diff --git a/server/lib/activitypub/process/process-create.ts b/server/lib/activitypub/process/process-create.ts
index 75f07d131..16f426e23 100644
--- a/server/lib/activitypub/process/process-create.ts
+++ b/server/lib/activitypub/process/process-create.ts
@@ -7,11 +7,11 @@ import { sequelizeTypescript } from '../../../initializers'
 import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { VideoAbuseModel } from '../../../models/video/video-abuse'
-import { VideoCommentModel } from '../../../models/video/video-comment'
 import { getOrCreateActorAndServerAndModel } from '../actor'
 import { addVideoComment, resolveThread } from '../video-comments'
 import { getOrCreateVideoAndAccountAndChannel } from '../videos'
 import { forwardActivity, forwardVideoRelatedActivity } from '../send/utils'
+import { Redis } from '../../redis'
 
 async function processCreateActivity (activity: ActivityCreate) {
   const activityObject = activity.object
@@ -88,7 +88,7 @@ async function processCreateView (byActor: ActorModel, activity: ActivityCreate)
   const actor = await ActorModel.loadByUrl(view.actor)
   if (!actor) throw new Error('Unknown actor ' + view.actor)
 
-  await video.increment('views')
+  await Redis.Instance.addVideoView(video.id)
 
   if (video.isOwned()) {
     // Don't resend the activity to the sender
diff --git a/server/lib/job-queue/handlers/video-views.ts b/server/lib/job-queue/handlers/video-views.ts
new file mode 100644
index 000000000..875d8ab88
--- /dev/null
+++ b/server/lib/job-queue/handlers/video-views.ts
@@ -0,0 +1,40 @@
+import { Redis } from '../../redis'
+import { logger } from '../../../helpers/logger'
+import { VideoModel } from '../../../models/video/video'
+import { VideoViewModel } from '../../../models/video/video-views'
+
+async function processVideosViewsViews () {
+  const hour = new Date().getHours()
+  const startDate = new Date().setMinutes(0, 0, 0)
+  const endDate = new Date().setMinutes(59, 59, 999)
+
+  const videoIds = await Redis.Instance.getVideosIdViewed(hour)
+  if (videoIds.length === 0) return
+
+  logger.info('Processing videos views in job for hour %d.', hour)
+
+  for (const videoId of videoIds) {
+    const views = await Redis.Instance.getVideoViews(videoId, hour)
+    if (isNaN(views)) {
+      logger.error('Cannot process videos views of video %s in hour %d: views number is NaN.', videoId, hour)
+    } else {
+      logger.debug('Adding %d views to video %d in hour %d.', views, videoId, hour)
+
+      await VideoModel.incrementViews(videoId, views)
+      await VideoViewModel.create({
+        startDate,
+        endDate,
+        views,
+        videoId
+      })
+    }
+
+    await Redis.Instance.deleteVideoViews(videoId, hour)
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processVideosViewsViews
+}
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index ddb357db5..0696ba43c 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -2,7 +2,7 @@ import * as Bull from 'bull'
 import { JobState, JobType } from '../../../shared/models'
 import { logger } from '../../helpers/logger'
 import { Redis } from '../redis'
-import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL } from '../../initializers'
+import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS } from '../../initializers'
 import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
 import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
 import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
@@ -10,6 +10,7 @@ import { EmailPayload, processEmail } from './handlers/email'
 import { processVideoFile, processVideoFileImport, VideoFileImportPayload, VideoFilePayload } from './handlers/video-file'
 import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
 import { processVideoImport, VideoImportPayload } from './handlers/video-import'
+import { processVideosViewsViews } from './handlers/video-views'
 
 type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -19,7 +20,8 @@ type CreateJobArgument =
   { type: 'video-file-import', payload: VideoFileImportPayload } |
   { type: 'video-file', payload: VideoFilePayload } |
   { type: 'email', payload: EmailPayload } |
-  { type: 'video-import', payload: VideoImportPayload }
+  { type: 'video-import', payload: VideoImportPayload } |
+  { type: 'videos-views', payload: {} }
 
 const handlers: { [ id in JobType ]: (job: Bull.Job) => Promise<any>} = {
   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
@@ -29,7 +31,8 @@ const handlers: { [ id in JobType ]: (job: Bull.Job) => Promise<any>} = {
   'video-file-import': processVideoFileImport,
   'video-file': processVideoFile,
   'email': processEmail,
-  'video-import': processVideoImport
+  'video-import': processVideoImport,
+  'videos-views': processVideosViewsViews
 }
 
 const jobTypes: JobType[] = [
@@ -40,7 +43,8 @@ const jobTypes: JobType[] = [
   'email',
   'video-file',
   'video-file-import',
-  'video-import'
+  'video-import',
+  'videos-views'
 ]
 
 class JobQueue {
@@ -85,6 +89,8 @@ class JobQueue {
 
       this.queues[handlerName] = queue
     }
+
+    this.addRepeatableJobs()
   }
 
   terminate () {
@@ -163,6 +169,12 @@ class JobQueue {
     }
   }
 
+  private addRepeatableJobs () {
+    this.queues['videos-views'].add({}, {
+      repeat: REPEAT_JOBS['videos-views']
+    })
+  }
+
   static get Instance () {
     return this.instance || (this.instance = new this())
   }
diff --git a/server/lib/redis.ts b/server/lib/redis.ts
index 941f7d557..0b4b41e4e 100644
--- a/server/lib/redis.ts
+++ b/server/lib/redis.ts
@@ -60,11 +60,11 @@ class Redis {
     return this.getValue(this.generateResetPasswordKey(userId))
   }
 
-  setView (ip: string, videoUUID: string) {
+  setIPVideoView (ip: string, videoUUID: string) {
     return this.setValue(this.buildViewKey(ip, videoUUID), '1', VIDEO_VIEW_LIFETIME)
   }
 
-  async isViewExists (ip: string, videoUUID: string) {
+  async isVideoIPViewExists (ip: string, videoUUID: string) {
     return this.exists(this.buildViewKey(ip, videoUUID))
   }
 
@@ -85,6 +85,52 @@ class Redis {
     return this.setObject(this.buildCachedRouteKey(req), cached, lifetime)
   }
 
+  addVideoView (videoId: number) {
+    const keyIncr = this.generateVideoViewKey(videoId)
+    const keySet = this.generateVideosViewKey()
+
+    return Promise.all([
+      this.addToSet(keySet, videoId.toString()),
+      this.increment(keyIncr)
+    ])
+  }
+
+  async getVideoViews (videoId: number, hour: number) {
+    const key = this.generateVideoViewKey(videoId, hour)
+
+    const valueString = await this.getValue(key)
+    return parseInt(valueString, 10)
+  }
+
+  async getVideosIdViewed (hour: number) {
+    const key = this.generateVideosViewKey(hour)
+
+    const stringIds = await this.getSet(key)
+    return stringIds.map(s => parseInt(s, 10))
+  }
+
+  deleteVideoViews (videoId: number, hour: number) {
+    const keySet = this.generateVideosViewKey(hour)
+    const keyIncr = this.generateVideoViewKey(videoId, hour)
+
+    return Promise.all([
+      this.deleteFromSet(keySet, videoId.toString()),
+      this.deleteKey(keyIncr)
+    ])
+  }
+
+  generateVideosViewKey (hour?: number) {
+    if (!hour) hour = new Date().getHours()
+
+    return `videos-view-h${hour}`
+  }
+
+  generateVideoViewKey (videoId: number, hour?: number) {
+    if (!hour) hour = new Date().getHours()
+
+    return `video-view-${videoId}-h${hour}`
+  }
+
   generateResetPasswordKey (userId: number) {
     return 'reset-password-' + userId
   }
@@ -107,6 +153,34 @@ class Redis {
     })
   }
 
+  private getSet (key: string) {
+    return new Promise<string[]>((res, rej) => {
+      this.client.smembers(this.prefix + key, (err, value) => {
+        if (err) return rej(err)
+
+        return res(value)
+      })
+    })
+  }
+
+  private addToSet (key: string, value: string) {
+    return new Promise<string[]>((res, rej) => {
+      this.client.sadd(this.prefix + key, value, err => err ? rej(err) : res())
+    })
+  }
+
+  private deleteFromSet (key: string, value: string) {
+    return new Promise<void>((res, rej) => {
+      this.client.srem(this.prefix + key, value, err => err ? rej(err) : res())
+    })
+  }
+
+  private deleteKey (key: string) {
+    return new Promise<void>((res, rej) => {
+      this.client.del(this.prefix + key, err => err ? rej(err) : res())
+    })
+  }
+
   private setValue (key: string, value: string, expirationMilliseconds: number) {
     return new Promise<void>((res, rej) => {
       this.client.set(this.prefix + key, value, 'PX', expirationMilliseconds, (err, ok) => {
@@ -145,6 +219,16 @@ class Redis {
     })
   }
 
+  private increment (key: string) {
+    return new Promise<number>((res, rej) => {
+      this.client.incr(this.prefix + key, (err, value) => {
+        if (err) return rej(err)
+
+        return res(value)
+      })
+    })
+  }
+
   private exists (key: string) {
     return new Promise<boolean>((res, rej) => {
       this.client.exists(this.prefix + key, (err, existsNumber) => {