Support video views/viewers stats in server

 * Add "currentTime" and "event" body params to view endpoint
 * Merge watching and view endpoints
 * Introduce WatchAction AP activity
 * Add tables to store viewer information of local videos
 * Add endpoints to fetch video views/viewers stats of local videos
 * Refactor views/viewers handlers
 * Support "views" and "viewers" counters for both VOD and live videos

5 files changed, 355 insertions, 17 deletions

diff --git a/server/models/video/formatter/video-format-utils.ts b/server/models/video/formatter/video-format-utils.ts
index 611edf0b9..6222107d7 100644
--- a/server/models/video/formatter/video-format-utils.ts
+++ b/server/models/video/formatter/video-format-utils.ts
@@ -1,11 +1,19 @@
 import { generateMagnetUri } from '@server/helpers/webtorrent'
+import { getActivityStreamDuration } from '@server/lib/activitypub/activity'
 import { getLocalVideoFileMetadataUrl } from '@server/lib/video-urls'
-import { VideoViews } from '@server/lib/video-views'
+import { VideoViewsManager } from '@server/lib/views/video-views-manager'
 import { uuidToShort } from '@shared/extra-utils'
-import { VideoFile, VideosCommonQueryAfterSanitize } from '@shared/models'
-import { ActivityTagObject, ActivityUrlObject, VideoObject } from '../../../../shared/models/activitypub/objects'
-import { Video, VideoDetails, VideoInclude } from '../../../../shared/models/videos'
-import { VideoStreamingPlaylist } from '../../../../shared/models/videos/video-streaming-playlist.model'
+import {
+  ActivityTagObject,
+  ActivityUrlObject,
+  Video,
+  VideoDetails,
+  VideoFile,
+  VideoInclude,
+  VideoObject,
+  VideosCommonQueryAfterSanitize,
+  VideoStreamingPlaylist
+} from '@shared/models'
 import { isArray } from '../../../helpers/custom-validators/misc'
 import {
   MIMETYPES,
@@ -97,7 +105,10 @@ function videoModelToFormattedJSON (video: MVideoFormattable, options: VideoForm
 
     isLocal: video.isOwned(),
     duration: video.duration,
+
     views: video.views,
+    viewers: VideoViewsManager.Instance.getViewers(video),
+
     likes: video.likes,
     dislikes: video.dislikes,
     thumbnailPath: video.getMiniatureStaticPath(),
@@ -121,10 +132,6 @@ function videoModelToFormattedJSON (video: MVideoFormattable, options: VideoForm
     pluginData: (video as any).pluginData
   }
 
-  if (video.isLive) {
-    videoObject.viewers = VideoViews.Instance.getViewers(video)
-  }
-
   const add = options.additionalAttributes
   if (add?.state === true) {
     videoObject.state = {
@@ -459,11 +466,6 @@ function videoModelToActivityPubObject (video: MVideoAP): VideoObject {
   }
 }
 
-function getActivityStreamDuration (duration: number) {
-  // https://www.w3.org/TR/activitystreams-vocabulary/#dfn-duration
-  return 'PT' + duration + 'S'
-}
-
 function getCategoryLabel (id: number) {
   return VIDEO_CATEGORIES[id] || 'Misc'
 }
@@ -489,7 +491,6 @@ export {
   videoModelToFormattedDetailsJSON,
   videoFilesModelToFormattedJSON,
   videoModelToActivityPubObject,
-  getActivityStreamDuration,
 
   guessAdditionalAttributesFromQuery,
 
diff --git a/server/models/video/video.ts b/server/models/video/video.ts
index 8bad2a01e..13d81561a 100644
--- a/server/models/video/video.ts
+++ b/server/models/video/video.ts
@@ -106,6 +106,7 @@ import { setAsUpdated } from '../shared'
 import { UserModel } from '../user/user'
 import { UserVideoHistoryModel } from '../user/user-video-history'
 import { buildTrigramSearchIndex, buildWhereIdOrUUID, getVideoSort, isOutdated, throwIfNotValid } from '../utils'
+import { VideoViewModel } from '../view/video-view'
 import {
   videoFilesModelToFormattedJSON,
   VideoFormattingJSONOptions,
@@ -135,7 +136,6 @@ import { VideoPlaylistElementModel } from './video-playlist-element'
 import { VideoShareModel } from './video-share'
 import { VideoStreamingPlaylistModel } from './video-streaming-playlist'
 import { VideoTagModel } from './video-tag'
-import { VideoViewModel } from './video-view'
 
 export enum ScopeNames {
   FOR_API = 'FOR_API',
diff --git a/server/models/view/local-video-viewer-watch-section.ts b/server/models/view/local-video-viewer-watch-section.ts
new file mode 100644
index 000000000..e29bb7847
--- /dev/null
+++ b/server/models/view/local-video-viewer-watch-section.ts
@@ -0,0 +1,63 @@
+import { Transaction } from 'sequelize'
+import { AllowNull, BelongsTo, Column, CreatedAt, ForeignKey, Model, Table } from 'sequelize-typescript'
+import { MLocalVideoViewerWatchSection } from '@server/types/models'
+import { AttributesOnly } from '@shared/typescript-utils'
+import { LocalVideoViewerModel } from './local-video-viewer'
+
+@Table({
+  tableName: 'localVideoViewerWatchSection',
+  updatedAt: false,
+  indexes: [
+    {
+      fields: [ 'localVideoViewerId' ]
+    }
+  ]
+})
+export class LocalVideoViewerWatchSectionModel extends Model<Partial<AttributesOnly<LocalVideoViewerWatchSectionModel>>> {
+  @CreatedAt
+  createdAt: Date
+
+  @AllowNull(false)
+  @Column
+  watchStart: number
+
+  @AllowNull(false)
+  @Column
+  watchEnd: number
+
+  @ForeignKey(() => LocalVideoViewerModel)
+  @Column
+  localVideoViewerId: number
+
+  @BelongsTo(() => LocalVideoViewerModel, {
+    foreignKey: {
+      allowNull: false
+    },
+    onDelete: 'CASCADE'
+  })
+  LocalVideoViewer: LocalVideoViewerModel
+
+  static async bulkCreateSections (options: {
+    localVideoViewerId: number
+    watchSections: {
+      start: number
+      end: number
+    }[]
+    transaction?: Transaction
+  }) {
+    const { localVideoViewerId, watchSections, transaction } = options
+    const models: MLocalVideoViewerWatchSection[] = []
+
+    for (const section of watchSections) {
+      const model = await this.create({
+        watchStart: section.start,
+        watchEnd: section.end,
+        localVideoViewerId
+      }, { transaction })
+
+      models.push(model)
+    }
+
+    return models
+  }
+}
diff --git a/server/models/view/local-video-viewer.ts b/server/models/view/local-video-viewer.ts
new file mode 100644
index 000000000..6f8de53cd
--- /dev/null
+++ b/server/models/view/local-video-viewer.ts
@@ -0,0 +1,274 @@
+import { QueryTypes } from 'sequelize'
+import { AllowNull, BelongsTo, Column, CreatedAt, DataType, Default, ForeignKey, HasMany, IsUUID, Model, Table } from 'sequelize-typescript'
+import { STATS_TIMESERIE } from '@server/initializers/constants'
+import { getActivityStreamDuration } from '@server/lib/activitypub/activity'
+import { MLocalVideoViewer, MLocalVideoViewerWithWatchSections, MVideo } from '@server/types/models'
+import { VideoStatsOverall, VideoStatsRetention, VideoStatsTimeserie, VideoStatsTimeserieMetric, WatchActionObject } from '@shared/models'
+import { AttributesOnly } from '@shared/typescript-utils'
+import { VideoModel } from '../video/video'
+import { LocalVideoViewerWatchSectionModel } from './local-video-viewer-watch-section'
+
+@Table({
+  tableName: 'localVideoViewer',
+  updatedAt: false,
+  indexes: [
+    {
+      fields: [ 'videoId' ]
+    }
+  ]
+})
+export class LocalVideoViewerModel extends Model<Partial<AttributesOnly<LocalVideoViewerModel>>> {
+  @CreatedAt
+  createdAt: Date
+
+  @AllowNull(false)
+  @Column(DataType.DATE)
+  startDate: Date
+
+  @AllowNull(false)
+  @Column(DataType.DATE)
+  endDate: Date
+
+  @AllowNull(false)
+  @Column
+  watchTime: number
+
+  @AllowNull(true)
+  @Column
+  country: string
+
+  @AllowNull(false)
+  @Default(DataType.UUIDV4)
+  @IsUUID(4)
+  @Column(DataType.UUID)
+  uuid: string
+
+  @AllowNull(false)
+  @Column
+  url: string
+
+  @ForeignKey(() => VideoModel)
+  @Column
+  videoId: number
+
+  @BelongsTo(() => VideoModel, {
+    foreignKey: {
+      allowNull: false
+    },
+    onDelete: 'CASCADE'
+  })
+  Video: VideoModel
+
+  @HasMany(() => LocalVideoViewerWatchSectionModel, {
+    foreignKey: {
+      allowNull: false
+    },
+    onDelete: 'cascade'
+  })
+  WatchSections: LocalVideoViewerWatchSectionModel[]
+
+  static loadByUrl (url: string): Promise<MLocalVideoViewer> {
+    return this.findOne({
+      where: {
+        url
+      }
+    })
+  }
+
+  static loadFullById (id: number): Promise<MLocalVideoViewerWithWatchSections> {
+    return this.findOne({
+      include: [
+        {
+          model: VideoModel.unscoped(),
+          required: true
+        },
+        {
+          model: LocalVideoViewerWatchSectionModel.unscoped(),
+          required: true
+        }
+      ],
+      where: {
+        id
+      }
+    })
+  }
+
+  static async getOverallStats (video: MVideo): Promise<VideoStatsOverall> {
+    const options = {
+      type: QueryTypes.SELECT as QueryTypes.SELECT,
+      replacements: { videoId: video.id }
+    }
+
+    const watchTimeQuery = `SELECT ` +
+      `SUM("localVideoViewer"."watchTime") AS "totalWatchTime", ` +
+      `AVG("localVideoViewer"."watchTime") AS "averageWatchTime" ` +
+      `FROM "localVideoViewer" ` +
+      `INNER JOIN "video" ON "video"."id" = "localVideoViewer"."videoId" ` +
+      `WHERE "videoId" = :videoId`
+
+    const watchTimePromise = LocalVideoViewerModel.sequelize.query<any>(watchTimeQuery, options)
+
+    const watchPeakQuery = `WITH "watchPeakValues" AS (
+        SELECT "startDate" AS "dateBreakpoint", 1 AS "inc"
+        FROM "localVideoViewer"
+        WHERE "videoId" = :videoId
+        UNION ALL
+        SELECT "endDate" AS "dateBreakpoint", -1 AS "inc"
+        FROM "localVideoViewer"
+        WHERE "videoId" = :videoId
+      )
+      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, options)
+
+    const commentsQuery = `SELECT COUNT(*) AS comments FROM "videoComment" WHERE "videoId" = :videoId`
+    const commentsPromise = LocalVideoViewerModel.sequelize.query<any>(commentsQuery, options)
+
+    const countriesQuery = `SELECT country, COUNT(country) as viewers ` +
+      `FROM "localVideoViewer" ` +
+      `WHERE "videoId" = :videoId AND country IS NOT NULL ` +
+      `GROUP BY country ` +
+      `ORDER BY viewers DESC`
+    const countriesPromise = LocalVideoViewerModel.sequelize.query<any>(countriesQuery, options)
+
+    const [ rowsWatchTime, rowsWatchPeak, rowsComment, rowsCountries ] = await Promise.all([
+      watchTimePromise,
+      watchPeakPromise,
+      commentsPromise,
+      countriesPromise
+    ])
+
+    return {
+      totalWatchTime: rowsWatchTime.length !== 0
+        ? Math.round(rowsWatchTime[0].totalWatchTime) || 0
+        : 0,
+      averageWatchTime: rowsWatchTime.length !== 0
+        ? Math.round(rowsWatchTime[0].averageWatchTime) || 0
+        : 0,
+
+      viewersPeak: rowsWatchPeak.length !== 0
+        ? parseInt(rowsWatchPeak[0].concurrent) || 0
+        : 0,
+      viewersPeakDate: rowsWatchPeak.length !== 0
+        ? rowsWatchPeak[0].dateBreakpoint || null
+        : null,
+
+      views: video.views,
+      likes: video.likes,
+      dislikes: video.dislikes,
+
+      comments: rowsComment.length !== 0
+        ? parseInt(rowsComment[0].comments) || 0
+        : 0,
+
+      countries: rowsCountries.map(r => ({
+        isoCode: r.country,
+        viewers: r.viewers
+      }))
+    }
+  }
+
+  static async getRetentionStats (video: MVideo): Promise<VideoStatsRetention> {
+    const step = Math.max(Math.round(video.duration / 100), 1)
+
+    const query = `WITH "total" AS (SELECT COUNT(*) AS viewers FROM "localVideoViewer" WHERE "videoId" = :videoId) ` +
+      `SELECT serie AS "second", ` +
+        `(COUNT("localVideoViewer".id)::float / (SELECT GREATEST("total"."viewers", 1) FROM "total")) AS "retention" ` +
+      `FROM generate_series(0, ${video.duration}, ${step}) serie ` +
+      `LEFT JOIN "localVideoViewer" ON "localVideoViewer"."videoId" = :videoId ` +
+        `AND EXISTS (` +
+          `SELECT 1 FROM "localVideoViewerWatchSection" ` +
+          `WHERE "localVideoViewer"."id" = "localVideoViewerWatchSection"."localVideoViewerId" ` +
+          `AND serie >= "localVideoViewerWatchSection"."watchStart" ` +
+          `AND serie <= "localVideoViewerWatchSection"."watchEnd"` +
+        `)` +
+      `GROUP BY serie ` +
+      `ORDER BY serie ASC`
+
+    const queryOptions = {
+      type: QueryTypes.SELECT as QueryTypes.SELECT,
+      replacements: { videoId: video.id }
+    }
+
+    const rows = await LocalVideoViewerModel.sequelize.query<any>(query, queryOptions)
+
+    return {
+      data: rows.map(r => ({
+        second: r.second,
+        retentionPercent: parseFloat(r.retention) * 100
+      }))
+    }
+  }
+
+  static async getTimeserieStats (options: {
+    video: MVideo
+    metric: VideoStatsTimeserieMetric
+  }): Promise<VideoStatsTimeserie> {
+    const { video, metric } = options
+
+    const selectMetrics: { [ id in VideoStatsTimeserieMetric ]: string } = {
+      viewers: 'COUNT("localVideoViewer"."id")',
+      aggregateWatchTime: 'SUM("localVideoViewer"."watchTime")'
+    }
+
+    const query = `WITH days AS ( ` +
+       `SELECT (current_date::timestamp - (serie || ' days')::interval)::timestamptz AS day
+        FROM generate_series(0, ${STATS_TIMESERIE.MAX_DAYS - 1}) serie` +
+      `) ` +
+      `SELECT days.day AS date, COALESCE(${selectMetrics[metric]}, 0) AS value ` +
+      `FROM days ` +
+      `LEFT JOIN "localVideoViewer" ON "localVideoViewer"."videoId" = :videoId ` +
+        `AND date_trunc('day', "localVideoViewer"."startDate") = date_trunc('day', days.day) ` +
+      `GROUP BY day ` +
+      `ORDER BY day `
+
+    const queryOptions = {
+      type: QueryTypes.SELECT as QueryTypes.SELECT,
+      replacements: { videoId: video.id }
+    }
+
+    const rows = await LocalVideoViewerModel.sequelize.query<any>(query, queryOptions)
+
+    return {
+      data: rows.map(r => ({
+        date: r.date,
+        value: parseInt(r.value)
+      }))
+    }
+  }
+
+  toActivityPubObject (this: MLocalVideoViewerWithWatchSections): WatchActionObject {
+    const location = this.country
+      ? {
+        location: {
+          addressCountry: this.country
+        }
+      }
+      : {}
+
+    return {
+      id: this.url,
+      type: 'WatchAction',
+      duration: getActivityStreamDuration(this.watchTime),
+      startTime: this.startDate.toISOString(),
+      endTime: this.endDate.toISOString(),
+
+      object: this.Video.url,
+      uuid: this.uuid,
+      actionStatus: 'CompletedActionStatus',
+
+      watchSections: this.WatchSections.map(w => ({
+        startTimestamp: w.watchStart,
+        endTimestamp: w.watchEnd
+      })),
+
+      ...location
+    }
+  }
+}
diff --git a/server/models/video/video-view.ts b/server/models/view/video-view.ts
index d72df100f..df462e631 100644
--- a/server/models/video/video-view.ts
+++ b/server/models/view/video-view.ts
@@ -1,7 +1,7 @@
 import { literal, Op } from 'sequelize'
 import { AllowNull, BelongsTo, Column, CreatedAt, DataType, ForeignKey, Model, Table } from 'sequelize-typescript'
 import { AttributesOnly } from '@shared/typescript-utils'
-import { VideoModel } from './video'
+import { VideoModel } from '../video/video'
 
 @Table({
   tableName: 'videoView',