Fix overall viewers stats with start/end dates

2 files changed, 141 insertions, 45 deletions

diff --git a/server/models/view/local-video-viewer.ts b/server/models/view/local-video-viewer.ts
index 12350861b..9d0d89a59 100644
--- a/server/models/view/local-video-viewer.ts
+++ b/server/models/view/local-video-viewer.ts
@@ -112,58 +112,88 @@ export class LocalVideoViewerModel extends Model<Partial<AttributesOnly<LocalVid
       replacements: { videoId: video.id } as any
     }
 
-    let dateWhere = ''
+    if (startDate) queryOptions.replacements.startDate = startDate
+    if (endDate) queryOptions.replacements.endDate = endDate
 
-    if (startDate) {
-      dateWhere += ' AND "localVideoViewer"."startDate" >= :startDate'
-      queryOptions.replacements.startDate = startDate
+    const buildWatchTimePromise = () => {
+      let watchTimeDateWhere = ''
+
+      if (startDate) watchTimeDateWhere += ' AND "localVideoViewer"."startDate" >= :startDate'
+      if (endDate) watchTimeDateWhere += ' AND "localVideoViewer"."endDate" <= :endDate'
+
+      const watchTimeQuery = `SELECT ` +
+        `COUNT("localVideoViewer"."id") AS "totalViewers", ` +
+        `SUM("localVideoViewer"."watchTime") AS "totalWatchTime", ` +
+        `AVG("localVideoViewer"."watchTime") AS "averageWatchTime" ` +
+        `FROM "localVideoViewer" ` +
+        `INNER JOIN "video" ON "video"."id" = "localVideoViewer"."videoId" ` +
+        `WHERE "videoId" = :videoId ${watchTimeDateWhere}`
+
+      return LocalVideoViewerModel.sequelize.query<any>(watchTimeQuery, queryOptions)
     }
 
-    if (endDate) {
-      dateWhere += ' AND "localVideoViewer"."endDate" <= :endDate'
-      queryOptions.replacements.endDate = endDate
+    const buildWatchPeakPromise = () => {
+      let watchPeakDateWhereStart = ''
+      let watchPeakDateWhereEnd = ''
+
+      if (startDate) {
+        watchPeakDateWhereStart += ' AND "localVideoViewer"."startDate" >= :startDate'
+        watchPeakDateWhereEnd += ' AND "localVideoViewer"."endDate" >= :startDate'
+      }
+
+      if (endDate) {
+        watchPeakDateWhereStart += ' AND "localVideoViewer"."startDate" <= :endDate'
+        watchPeakDateWhereEnd += ' AND "localVideoViewer"."endDate" <= :endDate'
+      }
+
+      // Add viewers that were already here, before our start date
+      const beforeWatchersQuery = startDate
+        // eslint-disable-next-line max-len
+        ? `SELECT COUNT(*) AS "total" FROM "localVideoViewer" WHERE "localVideoViewer"."startDate" < :startDate AND "localVideoViewer"."endDate" >= :startDate`
+        : `SELECT 0 AS "total"`
+
+      const watchPeakQuery = `WITH
+        "beforeWatchers" AS (${beforeWatchersQuery}),
+        "watchPeakValues" AS (
+          SELECT "startDate" AS "dateBreakpoint", 1 AS "inc"
+          FROM "localVideoViewer"
+          WHERE "videoId" = :videoId ${watchPeakDateWhereStart}
+          UNION ALL
+          SELECT "endDate" AS "dateBreakpoint", -1 AS "inc"
+          FROM "localVideoViewer"
+          WHERE "videoId" = :videoId ${watchPeakDateWhereEnd}
+        )
+        SELECT "dateBreakpoint", "concurrent"
+        FROM (
+          SELECT "dateBreakpoint", SUM(SUM("inc")) OVER (ORDER BY "dateBreakpoint") + (SELECT "total" FROM "beforeWatchers") AS "concurrent"
+          FROM "watchPeakValues"
+          GROUP BY "dateBreakpoint"
+        ) tmp
+        ORDER BY "concurrent" DESC
+        FETCH FIRST 1 ROW ONLY`
+
+      return LocalVideoViewerModel.sequelize.query<any>(watchPeakQuery, queryOptions)
     }
 
-    const watchTimeQuery = `SELECT ` +
-      `COUNT("localVideoViewer"."id") AS "totalViewers", ` +
-      `SUM("localVideoViewer"."watchTime") AS "totalWatchTime", ` +
-      `AVG("localVideoViewer"."watchTime") AS "averageWatchTime" ` +
-      `FROM "localVideoViewer" ` +
-      `INNER JOIN "video" ON "video"."id" = "localVideoViewer"."videoId" ` +
-      `WHERE "videoId" = :videoId ${dateWhere}`
-
-    const watchTimePromise = LocalVideoViewerModel.sequelize.query<any>(watchTimeQuery, queryOptions)
-
-    const watchPeakQuery = `WITH "watchPeakValues" AS (
-        SELECT "startDate" AS "dateBreakpoint", 1 AS "inc"
-        FROM "localVideoViewer"
-        WHERE "videoId" = :videoId ${dateWhere}
-        UNION ALL
-        SELECT "endDate" AS "dateBreakpoint", -1 AS "inc"
-        FROM "localVideoViewer"
-        WHERE "videoId" = :videoId ${dateWhere}
-      )
-      SELECT "dateBreakpoint", "concurrent"
-      FROM (
-        SELECT "dateBreakpoint", SUM(SUM("inc")) OVER (ORDER BY "dateBreakpoint") AS "concurrent"
-        FROM "watchPeakValues"
-        GROUP BY "dateBreakpoint"
-      ) tmp
-      ORDER BY "concurrent" DESC
-      FETCH FIRST 1 ROW ONLY`
-    const watchPeakPromise = LocalVideoViewerModel.sequelize.query<any>(watchPeakQuery, queryOptions)
-
-    const countriesQuery = `SELECT country, COUNT(country) as viewers ` +
-      `FROM "localVideoViewer" ` +
-      `WHERE "videoId" = :videoId AND country IS NOT NULL ${dateWhere} ` +
-      `GROUP BY country ` +
-      `ORDER BY viewers DESC`
-    const countriesPromise = LocalVideoViewerModel.sequelize.query<any>(countriesQuery, queryOptions)
+    const buildCountriesPromise = () => {
+      let countryDateWhere = ''
+
+      if (startDate) countryDateWhere += ' AND "localVideoViewer"."endDate" >= :startDate'
+      if (endDate) countryDateWhere += ' AND "localVideoViewer"."startDate" <= :endDate'
+
+      const countriesQuery = `SELECT country, COUNT(country) as viewers ` +
+        `FROM "localVideoViewer" ` +
+        `WHERE "videoId" = :videoId AND country IS NOT NULL ${countryDateWhere} ` +
+        `GROUP BY country ` +
+        `ORDER BY viewers DESC`
+
+      return LocalVideoViewerModel.sequelize.query<any>(countriesQuery, queryOptions)
+    }
 
     const [ rowsWatchTime, rowsWatchPeak, rowsCountries ] = await Promise.all([
-      watchTimePromise,
-      watchPeakPromise,
-      countriesPromise
+      buildWatchTimePromise(),
+      buildWatchPeakPromise(),
+      buildCountriesPromise()
     ])
 
     const viewersPeak = rowsWatchPeak.length !== 0
diff --git a/server/tests/api/views/video-views-overall-stats.ts b/server/tests/api/views/video-views-overall-stats.ts
index 3aadc9689..ac636961e 100644
--- a/server/tests/api/views/video-views-overall-stats.ts
+++ b/server/tests/api/views/video-views-overall-stats.ts
@@ -4,6 +4,56 @@ import { expect } from 'chai'
 import { FfmpegCommand } from 'fluent-ffmpeg'
 import { prepareViewsServers, prepareViewsVideos, processViewersStats } from '@server/tests/shared'
 import { cleanupTests, PeerTubeServer, stopFfmpeg, waitJobs } from '@shared/server-commands'
+import { wait } from '@shared/core-utils'
+import { VideoStatsOverall } from '@shared/models'
+
+/**
+ *
+ * Simulate 5 sections of viewers
+ *  * user0 started and ended before start date
+ *  * user1 started before start date and ended in the interval
+ *  * user2 started started in the interval and ended after end date
+ *  * user3 started and ended in the interval
+ *  * user4 started and ended after end date
+ */
+async function simulateComplexViewers (servers: PeerTubeServer[], videoUUID: string) {
+  const user0 = '8.8.8.8,127.0.0.1'
+  const user1 = '8.8.8.8,127.0.0.1'
+  const user2 = '8.8.8.9,127.0.0.1'
+  const user3 = '8.8.8.10,127.0.0.1'
+  const user4 = '8.8.8.11,127.0.0.1'
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 0, xForwardedFor: user0 }) // User 0 starts
+  await wait(500)
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 0, xForwardedFor: user1 }) // User 1 starts
+  await servers[0].views.view({ id: videoUUID, currentTime: 2, xForwardedFor: user0 }) // User 0 ends
+  await wait(500)
+
+  const startDate = new Date().toISOString()
+  await servers[0].views.view({ id: videoUUID, currentTime: 0, xForwardedFor: user2 }) // User 2 starts
+  await wait(500)
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 0, xForwardedFor: user3 }) // User 3 starts
+  await wait(500)
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 4, xForwardedFor: user1 }) // User 1 ends
+  await wait(500)
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 3, xForwardedFor: user3 }) // User 3 ends
+  await wait(500)
+
+  const endDate = new Date().toISOString()
+  await servers[0].views.view({ id: videoUUID, currentTime: 0, xForwardedFor: user4 }) // User 4 starts
+  await servers[0].views.view({ id: videoUUID, currentTime: 5, xForwardedFor: user2 }) // User 2 ends
+  await wait(500)
+
+  await servers[0].views.view({ id: videoUUID, currentTime: 1, xForwardedFor: user4 }) // User 4 ends
+
+  await processViewersStats(servers)
+
+  return { startDate, endDate }
+}
 
 describe('Test views overall stats', function () {
   let servers: PeerTubeServer[]
@@ -237,6 +287,22 @@ describe('Test views overall stats', function () {
         expect(new Date(stats.viewersPeakDate)).to.be.below(before2Watchers)
       }
     })
+
+    it('Should complex filter peak viewers by date', async function () {
+      this.timeout(60000)
+
+      const { startDate, endDate } = await simulateComplexViewers(servers, videoUUID)
+
+      const expectCorrect = (stats: VideoStatsOverall) => {
+        expect(stats.viewersPeak).to.equal(3)
+        expect(new Date(stats.viewersPeakDate)).to.be.above(new Date(startDate)).and.below(new Date(endDate))
+      }
+
+      expectCorrect(await servers[0].videoStats.getOverallStats({ videoId: videoUUID, startDate, endDate }))
+      expectCorrect(await servers[0].videoStats.getOverallStats({ videoId: videoUUID, startDate }))
+      expectCorrect(await servers[0].videoStats.getOverallStats({ videoId: videoUUID, endDate }))
+      expectCorrect(await servers[0].videoStats.getOverallStats({ videoId: videoUUID }))
+    })
   })
 
   describe('Test countries', function () {