server/lib/views/shared/video-viewer-counters.ts

   1
   2 import { isTestInstance } from '@server/helpers/core-utils'
   3 import { logger, loggerTagsFactory } from '@server/helpers/logger'
   4 import { VIEW_LIFETIME } from '@server/initializers/constants'
   5 import { sendView } from '@server/lib/activitypub/send/send-view'
   6 import { PeerTubeSocket } from '@server/lib/peertube-socket'
   7 import { getServerActor } from '@server/models/application/application'
   8 import { VideoModel } from '@server/models/video/video'
   9 import { MVideo } from '@server/types/models'
  10 import { buildUUID, sha256 } from '@shared/extra-utils'
  11
  12 const lTags = loggerTagsFactory('views')
  13
  14 type Viewer = {
  15   expires: number
  16   id: string
  17   lastFederation?: number
  18 }
  19
  20 export class VideoViewerCounters {
  21
  22   // expires is new Date().getTime()
  23   private readonly viewersPerVideo = new Map<number, Viewer[]>()
  24   private readonly idToViewer = new Map<string, Viewer>()
  25
  26   private readonly salt = buildUUID()
  27
  28   private processingViewerCounters = false
  29
  30   constructor () {
  31     setInterval(() => this.cleanViewerCounters(), VIEW_LIFETIME.VIEWER_COUNTER)
  32   }
  33
  34   // ---------------------------------------------------------------------------
  35
  36   async addLocalViewer (options: {
  37     video: MVideo
  38     ip: string
  39   }) {
  40     const { video, ip } = options
  41
  42     logger.debug('Adding local viewer to video viewers counter %s.', video.uuid, { ...lTags(video.uuid) })
  43
  44     const viewerId = this.generateViewerId(ip, video.uuid)
  45     const viewer = this.idToViewer.get(viewerId)
  46
  47     if (viewer) {
  48       viewer.expires = this.buildViewerExpireTime()
  49       await this.federateViewerIfNeeded(video, viewer)
  50
  51       return false
  52     }
  53
  54     const newViewer = await this.addViewerToVideo({ viewerId, video })
  55     await this.federateViewerIfNeeded(video, newViewer)
  56
  57     return true
  58   }
  59
  60   async addRemoteViewer (options: {
  61     video: MVideo
  62     viewerId: string
  63     viewerExpires: Date
  64   }) {
  65     const { video, viewerExpires, viewerId } = options
  66
  67     logger.debug('Adding remote viewer to video %s.', video.uuid, { ...lTags(video.uuid) })
  68
  69     await this.addViewerToVideo({ video, viewerExpires, viewerId })
  70
  71     return true
  72   }
  73
  74   // ---------------------------------------------------------------------------
  75
  76   getViewers (video: MVideo) {
  77     const viewers = this.viewersPerVideo.get(video.id)
  78     if (!viewers) return 0
  79
  80     return viewers.length
  81   }
  82
  83   buildViewerExpireTime () {
  84     return new Date().getTime() + VIEW_LIFETIME.VIEWER_COUNTER
  85   }
  86
  87   // ---------------------------------------------------------------------------
  88
  89   private async addViewerToVideo (options: {
  90     video: MVideo
  91     viewerId: string
  92     viewerExpires?: Date
  93   }) {
  94     const { video, viewerExpires, viewerId } = options
  95
  96     let watchers = this.viewersPerVideo.get(video.id)
  97
  98     if (!watchers) {
  99       watchers = []
 100       this.viewersPerVideo.set(video.id, watchers)
 101     }
 102
 103     const expires = viewerExpires
 104       ? viewerExpires.getTime()
 105       : this.buildViewerExpireTime()
 106
 107     const viewer = { id: viewerId, expires }
 108     watchers.push(viewer)
 109
 110     this.idToViewer.set(viewerId, viewer)
 111
 112     await this.notifyClients(video.id, watchers.length)
 113
 114     return viewer
 115   }
 116
 117   private async cleanViewerCounters () {
 118     if (this.processingViewerCounters) return
 119     this.processingViewerCounters = true
 120
 121     if (!isTestInstance()) logger.info('Cleaning video viewers.', lTags())
 122
 123     try {
 124       for (const videoId of this.viewersPerVideo.keys()) {
 125         const notBefore = new Date().getTime()
 126
 127         const viewers = this.viewersPerVideo.get(videoId)
 128
 129         // Only keep not expired viewers
 130         const newViewers: Viewer[] = []
 131
 132         // Filter new viewers
 133         for (const viewer of viewers) {
 134           if (viewer.expires > notBefore) {
 135             newViewers.push(viewer)
 136           } else {
 137             this.idToViewer.delete(viewer.id)
 138           }
 139         }
 140
 141         if (newViewers.length === 0) this.viewersPerVideo.delete(videoId)
 142         else this.viewersPerVideo.set(videoId, newViewers)
 143
 144         await this.notifyClients(videoId, newViewers.length)
 145       }
 146     } catch (err) {
 147       logger.error('Error in video clean viewers scheduler.', { err, ...lTags() })
 148     }
 149
 150     this.processingViewerCounters = false
 151   }
 152
 153   private async notifyClients (videoId: string | number, viewersLength: number) {
 154     const video = await VideoModel.loadImmutableAttributes(videoId)
 155     if (!video) return
 156
 157     PeerTubeSocket.Instance.sendVideoViewsUpdate(video, viewersLength)
 158
 159     logger.debug('Video viewers update for %s is %d.', video.url, viewersLength, lTags())
 160   }
 161
 162   private generateViewerId (ip: string, videoUUID: string) {
 163     return sha256(this.salt + '-' + ip + '-' + videoUUID)
 164   }
 165
 166   private async federateViewerIfNeeded (video: MVideo, viewer: Viewer) {
 167     // Federate the viewer if it's been a "long" time we did not
 168     const now = new Date().getTime()
 169     const federationLimit = now - (VIEW_LIFETIME.VIEWER_COUNTER / 2)
 170
 171     if (viewer.lastFederation && viewer.lastFederation > federationLimit) return
 172
 173     await sendView({ byActor: await getServerActor(), video, type: 'viewer', viewerIdentifier: viewer.id })
 174     viewer.lastFederation = now
 175   }
 176 }