aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/views/shared
diff options
context:
space:
mode:
authorChocobozzz <me@florianbigard.com>2022-04-06 08:50:43 +0200
committerChocobozzz <chocobozzz@cpy.re>2022-04-15 09:49:35 +0200
commitac907dc7c158056e9b6a5cb58acd27df5c7c2670 (patch)
treef9d8bff22e0543a305c64e1a1808c0df6d512f1e /server/lib/views/shared
parentdfbcefc20dc64f0814b1f2e8e782a4ea1bd24db2 (diff)
downloadPeerTube-ac907dc7c158056e9b6a5cb58acd27df5c7c2670.tar.gz
PeerTube-ac907dc7c158056e9b6a5cb58acd27df5c7c2670.tar.zst
PeerTube-ac907dc7c158056e9b6a5cb58acd27df5c7c2670.zip
Improve viewer counter
More precise, avoid weird decrease, reuse an id to federate viewers
Diffstat (limited to 'server/lib/views/shared')
-rw-r--r--server/lib/views/shared/index.ts3
-rw-r--r--server/lib/views/shared/video-viewer-counters.ts176
-rw-r--r--server/lib/views/shared/video-viewer-stats.ts (renamed from server/lib/views/shared/video-viewers.ts)116
-rw-r--r--server/lib/views/shared/video-views.ts7
4 files changed, 197 insertions, 105 deletions
diff --git a/server/lib/views/shared/index.ts b/server/lib/views/shared/index.ts
index dd510f4e2..139471183 100644
--- a/server/lib/views/shared/index.ts
+++ b/server/lib/views/shared/index.ts
@@ -1,2 +1,3 @@
1export * from './video-viewers' 1export * from './video-viewer-counters'
2export * from './video-viewer-stats'
2export * from './video-views' 3export * from './video-views'
diff --git a/server/lib/views/shared/video-viewer-counters.ts b/server/lib/views/shared/video-viewer-counters.ts
new file mode 100644
index 000000000..941b62ed7
--- /dev/null
+++ b/server/lib/views/shared/video-viewer-counters.ts
@@ -0,0 +1,176 @@
1
2import { isTestInstance } from '@server/helpers/core-utils'
3import { logger, loggerTagsFactory } from '@server/helpers/logger'
4import { VIEW_LIFETIME } from '@server/initializers/constants'
5import { sendView } from '@server/lib/activitypub/send/send-view'
6import { PeerTubeSocket } from '@server/lib/peertube-socket'
7import { getServerActor } from '@server/models/application/application'
8import { VideoModel } from '@server/models/video/video'
9import { MVideo } from '@server/types/models'
10import { buildUUID, sha256 } from '@shared/extra-utils'
11
12const lTags = loggerTagsFactory('views')
13
14type Viewer = {
15 expires: number
16 id: string
17 lastFederation?: number
18}
19
20export 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}
diff --git a/server/lib/views/shared/video-viewers.ts b/server/lib/views/shared/video-viewer-stats.ts
index 4dad1f0e8..fd66fd5c7 100644
--- a/server/lib/views/shared/video-viewers.ts
+++ b/server/lib/views/shared/video-viewer-stats.ts
@@ -6,7 +6,6 @@ import { MAX_LOCAL_VIEWER_WATCH_SECTIONS, VIEW_LIFETIME } from '@server/initiali
6import { sequelizeTypescript } from '@server/initializers/database' 6import { sequelizeTypescript } from '@server/initializers/database'
7import { sendCreateWatchAction } from '@server/lib/activitypub/send' 7import { sendCreateWatchAction } from '@server/lib/activitypub/send'
8import { getLocalVideoViewerActivityPubUrl } from '@server/lib/activitypub/url' 8import { getLocalVideoViewerActivityPubUrl } from '@server/lib/activitypub/url'
9import { PeerTubeSocket } from '@server/lib/peertube-socket'
10import { Redis } from '@server/lib/redis' 9import { Redis } from '@server/lib/redis'
11import { VideoModel } from '@server/models/video/video' 10import { VideoModel } from '@server/models/video/video'
12import { LocalVideoViewerModel } from '@server/models/view/local-video-viewer' 11import { LocalVideoViewerModel } from '@server/models/view/local-video-viewer'
@@ -32,39 +31,15 @@ type LocalViewerStats = {
32 videoId: number 31 videoId: number
33} 32}
34 33
35export class VideoViewers { 34export class VideoViewerStats {
36 35 private processingViewersStats = false
37 // Values are Date().getTime()
38 private readonly viewersPerVideo = new Map<number, number[]>()
39
40 private processingViewerCounters = false
41 private processingViewerStats = false
42 36
43 constructor () { 37 constructor () {
44 setInterval(() => this.cleanViewerCounters(), VIEW_LIFETIME.VIEWER_COUNTER)
45
46 setInterval(() => this.processViewerStats(), VIEW_LIFETIME.VIEWER_STATS) 38 setInterval(() => this.processViewerStats(), VIEW_LIFETIME.VIEWER_STATS)
47 } 39 }
48 40
49 // --------------------------------------------------------------------------- 41 // ---------------------------------------------------------------------------
50 42
51 getViewers (video: MVideo) {
52 const viewers = this.viewersPerVideo.get(video.id)
53 if (!viewers) return 0
54
55 return viewers.length
56 }
57
58 buildViewerExpireTime () {
59 return new Date().getTime() + VIEW_LIFETIME.VIEWER_COUNTER
60 }
61
62 async getWatchTime (videoId: number, ip: string) {
63 const stats: LocalViewerStats = await Redis.Instance.getLocalVideoViewer({ ip, videoId })
64
65 return stats?.watchTime || 0
66 }
67
68 async addLocalViewer (options: { 43 async addLocalViewer (options: {
69 video: MVideo 44 video: MVideo
70 currentTime: number 45 currentTime: number
@@ -73,51 +48,20 @@ export class VideoViewers {
73 }) { 48 }) {
74 const { video, ip, viewEvent, currentTime } = options 49 const { video, ip, viewEvent, currentTime } = options
75 50
76 logger.debug('Adding local viewer to video %s.', video.uuid, { currentTime, viewEvent, ...lTags(video.uuid) }) 51 logger.debug('Adding local viewer to video stats %s.', video.uuid, { currentTime, viewEvent, ...lTags(video.uuid) })
77
78 await this.updateLocalViewerStats({ video, viewEvent, currentTime, ip })
79
80 const viewExists = await Redis.Instance.doesVideoIPViewerExist(ip, video.uuid)
81 if (viewExists) return false
82 52
83 await Redis.Instance.setIPVideoViewer(ip, video.uuid) 53 return this.updateLocalViewerStats({ video, viewEvent, currentTime, ip })
84
85 return this.addViewerToVideo({ video })
86 } 54 }
87 55
88 async addRemoteViewer (options: { 56 // ---------------------------------------------------------------------------
89 video: MVideo
90 viewerExpires: Date
91 }) {
92 const { video, viewerExpires } = options
93 57
94 logger.debug('Adding remote viewer to video %s.', video.uuid, { ...lTags(video.uuid) }) 58 async getWatchTime (videoId: number, ip: string) {
59 const stats: LocalViewerStats = await Redis.Instance.getLocalVideoViewer({ ip, videoId })
95 60
96 return this.addViewerToVideo({ video, viewerExpires }) 61 return stats?.watchTime || 0
97 } 62 }
98 63
99 private async addViewerToVideo (options: { 64 // ---------------------------------------------------------------------------
100 video: MVideo
101 viewerExpires?: Date
102 }) {
103 const { video, viewerExpires } = options
104
105 let watchers = this.viewersPerVideo.get(video.id)
106
107 if (!watchers) {
108 watchers = []
109 this.viewersPerVideo.set(video.id, watchers)
110 }
111
112 const expiration = viewerExpires
113 ? viewerExpires.getTime()
114 : this.buildViewerExpireTime()
115
116 watchers.push(expiration)
117 await this.notifyClients(video.id, watchers.length)
118
119 return true
120 }
121 65
122 private async updateLocalViewerStats (options: { 66 private async updateLocalViewerStats (options: {
123 video: MVideo 67 video: MVideo
@@ -170,45 +114,9 @@ export class VideoViewers {
170 await Redis.Instance.setLocalVideoViewer(ip, video.id, stats) 114 await Redis.Instance.setLocalVideoViewer(ip, video.id, stats)
171 } 115 }
172 116
173 private async cleanViewerCounters () {
174 if (this.processingViewerCounters) return
175 this.processingViewerCounters = true
176
177 if (!isTestInstance()) logger.info('Cleaning video viewers.', lTags())
178
179 try {
180 for (const videoId of this.viewersPerVideo.keys()) {
181 const notBefore = new Date().getTime()
182
183 const viewers = this.viewersPerVideo.get(videoId)
184
185 // Only keep not expired viewers
186 const newViewers = viewers.filter(w => w > notBefore)
187
188 if (newViewers.length === 0) this.viewersPerVideo.delete(videoId)
189 else this.viewersPerVideo.set(videoId, newViewers)
190
191 await this.notifyClients(videoId, newViewers.length)
192 }
193 } catch (err) {
194 logger.error('Error in video clean viewers scheduler.', { err, ...lTags() })
195 }
196
197 this.processingViewerCounters = false
198 }
199
200 private async notifyClients (videoId: string | number, viewersLength: number) {
201 const video = await VideoModel.loadImmutableAttributes(videoId)
202 if (!video) return
203
204 PeerTubeSocket.Instance.sendVideoViewsUpdate(video, viewersLength)
205
206 logger.debug('Video viewers update for %s is %d.', video.url, viewersLength, lTags())
207 }
208
209 async processViewerStats () { 117 async processViewerStats () {
210 if (this.processingViewerStats) return 118 if (this.processingViewersStats) return
211 this.processingViewerStats = true 119 this.processingViewersStats = true
212 120
213 if (!isTestInstance()) logger.info('Processing viewer statistics.', lTags()) 121 if (!isTestInstance()) logger.info('Processing viewer statistics.', lTags())
214 122
@@ -245,7 +153,7 @@ export class VideoViewers {
245 logger.error('Error in video save viewers stats scheduler.', { err, ...lTags() }) 153 logger.error('Error in video save viewers stats scheduler.', { err, ...lTags() })
246 } 154 }
247 155
248 this.processingViewerStats = false 156 this.processingViewersStats = false
249 } 157 }
250 158
251 private async saveViewerStats (video: MVideo, stats: LocalViewerStats, transaction: Transaction) { 159 private async saveViewerStats (video: MVideo, stats: LocalViewerStats, transaction: Transaction) {
diff --git a/server/lib/views/shared/video-views.ts b/server/lib/views/shared/video-views.ts
index 19250f993..275f7a014 100644
--- a/server/lib/views/shared/video-views.ts
+++ b/server/lib/views/shared/video-views.ts
@@ -1,5 +1,8 @@
1import { logger, loggerTagsFactory } from '@server/helpers/logger' 1import { logger, loggerTagsFactory } from '@server/helpers/logger'
2import { sendView } from '@server/lib/activitypub/send/send-view'
3import { getServerActor } from '@server/models/application/application'
2import { MVideo } from '@server/types/models' 4import { MVideo } from '@server/types/models'
5import { buildUUID } from '@shared/extra-utils'
3import { Redis } from '../../redis' 6import { Redis } from '../../redis'
4 7
5const lTags = loggerTagsFactory('views') 8const lTags = loggerTagsFactory('views')
@@ -24,6 +27,8 @@ export class VideoViews {
24 27
25 await this.addView(video) 28 await this.addView(video)
26 29
30 await sendView({ byActor: await getServerActor(), video, type: 'view', viewerIdentifier: buildUUID() })
31
27 return true 32 return true
28 } 33 }
29 34
@@ -39,6 +44,8 @@ export class VideoViews {
39 return true 44 return true
40 } 45 }
41 46
47 // ---------------------------------------------------------------------------
48
42 private async addView (video: MVideo) { 49 private async addView (video: MVideo) {
43 const promises: Promise<any>[] = [] 50 const promises: Promise<any>[] = []
44 51