Fix/homepage scoring (#251)

* Update video relevance manager query * Remove single video schedules for relevance recalc * Fix namings * Reformat query * Revert "Remove single video schedules for relevance recalc" This reverts commit 3a685b9 * Adjust logic for single channel recalc * Unblock video relevance recalc only after processor reaches last exported block * Increase global recalc interval * Avoid query if set is empty * Introduce smaller interval for scheduled channels * Typo fix Co-authored-by: Zeeshan Akram <[email protected]> * Linter * Add video recalc on channel weight change --------- Co-authored-by: Zeeshan Akram <[email protected]>
Joystream · Dec 6, 2023 · c8b2c2e · c8b2c2e
1 parent 4c6e3af
commit c8b2c2e
Show file tree

Hide file tree

Showing 9 changed files with 109 additions and 61 deletions.
diff --git a/src/mappings/content/commentsAndReactions.ts b/src/mappings/content/commentsAndReactions.ts
@@ -318,7 +318,7 @@ export async function processReactVideoMessage(
     existingReaction
   )
 
-  videoRelevanceManager.scheduleRecalcForVideo(video.id)
+  videoRelevanceManager.scheduleRecalcForChannel(channelId)
 
   return new MetaprotocolTransactionResultOK()
 }
@@ -497,7 +497,7 @@ export async function processCreateCommentMessage(
   // schedule comment counters update
   commentCountersManager.scheduleRecalcForComment(comment.parentCommentId)
   commentCountersManager.scheduleRecalcForVideo(comment.videoId)
-  videoRelevanceManager.scheduleRecalcForVideo(comment.videoId)
+  videoRelevanceManager.scheduleRecalcForChannel(video.channelId)
 
   const event = overlay.getRepository(Event).new({
     ...genericEventFields(overlay, block, indexInBlock, txHash),
@@ -661,7 +661,7 @@ export async function processDeleteCommentMessage(
   // schedule comment counters update
   commentCountersManager.scheduleRecalcForComment(comment.parentCommentId)
   commentCountersManager.scheduleRecalcForVideo(comment.videoId)
-  videoRelevanceManager.scheduleRecalcForVideo(comment.videoId)
+  videoRelevanceManager.scheduleRecalcForChannel(video.channelId)
 
   // update the comment
   comment.text = ''

diff --git a/src/mappings/content/metadata.ts b/src/mappings/content/metadata.ts
@@ -65,7 +65,6 @@ import {
   genericEventFields,
   invalidMetadata,
   metaprotocolTransactionFailure,
-  videoRelevanceManager,
 } from '../utils'
 import {
   AsDecoded,
@@ -582,7 +581,6 @@ export async function processModerateCommentMessage(
   // schedule comment counters updates
   commentCountersManager.scheduleRecalcForComment(comment.parentCommentId)
   commentCountersManager.scheduleRecalcForVideo(comment.videoId)
-  videoRelevanceManager.scheduleRecalcForVideo(comment.videoId)
 
   comment.text = ''
   comment.status = CommentStatus.MODERATED

diff --git a/src/mappings/content/video.ts b/src/mappings/content/video.ts
@@ -58,7 +58,7 @@ export async function processVideoCreatedEvent({
     videoRelevance: 0,
   })
 
-  videoRelevanceManager.scheduleRecalcForVideo(videoId)
+  videoRelevanceManager.scheduleRecalcForChannel(channelId.toString())
 
   // fetch related channel and owner
   const channel = await overlay.getRepository(Channel).getByIdOrFail(channelId.toString())

diff --git a/src/mappings/utils.ts b/src/mappings/utils.ts
@@ -14,8 +14,12 @@ import { NextEntityIdManager } from '../utils/NextEntityIdManager'
 
 export const commentCountersManager = new CommentCountersManager()
 export const videoRelevanceManager = new VideoRelevanceManager()
+// eslint-disable-next-line no-void
+void videoRelevanceManager.init({
+  fullUpdateLoopTime: 1000 * 60 * 60 * 12, // 12 hrs
+  scheduledUpdateLoopTime: 1000 * 60 * 10, // 10 mins
+})
 export const migrateCounters = new NextEntityIdManager()
-videoRelevanceManager.init(1000 * 60 * 60)
 
 export const JOYSTREAM_SS58_PREFIX = 126
 

diff --git a/src/processor.ts b/src/processor.ts
@@ -284,7 +284,6 @@ async function processEvent<EventName extends EventNames>(
 async function afterDbUpdate(em: EntityManager) {
   await commentCountersManager.updateVideoCommentsCounters(em)
   await commentCountersManager.updateParentRepliesCounters(em)
-  await videoRelevanceManager.updateVideoRelevanceValue(em)
 }
 
 processor.run(new TypeormDatabase({ isolationLevel: 'READ COMMITTED' }), async (ctx) => {
@@ -314,9 +313,18 @@ processor.run(new TypeormDatabase({ isolationLevel: 'READ COMMITTED' }), async (
         }
       }
     }
+
+    if (
+      !videoRelevanceManager.isVideoRelevanceEnabled &&
+      block.header.height >= exportBlockNumber
+    ) {
+      videoRelevanceManager.turnOnVideoRelevanceManager()
+    }
+
     // Importing exported offchain state
     if (block.header.height >= exportBlockNumber && !offchainState.isImported) {
       ctx.log.info(`Export block ${exportBlockNumber} reached, importing offchain state...`)
+      // there is no need to recalc video relevance before orion is synced
       await overlay.updateDatabase()
       const em = overlay.getEm()
       await offchainState.import(em)

diff --git a/src/server-extension/resolvers/AdminResolver/index.ts b/src/server-extension/resolvers/AdminResolver/index.ts
@@ -216,6 +216,8 @@ export class AdminResolver {
         }
       )
 
+      await videoRelevanceManager.updateVideoRelevanceValue(em, true)
+
       // Push the result into the results array
       results.push({
         channelId,

diff --git a/src/server-extension/resolvers/AdminResolver/utils.ts b/src/server-extension/resolvers/AdminResolver/utils.ts
@@ -1,18 +1,14 @@
 import { EntityManager, In } from 'typeorm'
 import { CommentCountersManager } from '../../../utils/CommentsCountersManager'
 import { Comment } from '../../../model'
-import { VideoRelevanceManager } from '../../../utils/VideoRelevanceManager'
 
 export async function processCommentsCensorshipStatusUpdate(em: EntityManager, ids: string[]) {
   const manager = new CommentCountersManager()
-  const videoRelevanceManager = new VideoRelevanceManager()
   const comments = await em.getRepository(Comment).find({ where: { id: In(ids) } })
   comments.forEach((c) => {
     manager.scheduleRecalcForComment(c.parentCommentId)
     manager.scheduleRecalcForVideo(c.videoId)
-    videoRelevanceManager.scheduleRecalcForVideo(c.videoId)
   })
   await manager.updateVideoCommentsCounters(em)
   await manager.updateParentRepliesCounters(em)
-  await videoRelevanceManager.updateVideoRelevanceValue(em)
 }
diff --git a/src/server-extension/resolvers/VideosResolver/index.ts b/src/server-extension/resolvers/VideosResolver/index.ts
@@ -248,10 +248,9 @@ export class VideosResolver {
 
       const tick = await config.get(ConfigVariable.VideoRelevanceViewsTick, em)
       if (video.viewsNum % tick === 0) {
-        videoRelevanceManager.scheduleRecalcForVideo(videoId)
+        videoRelevanceManager.scheduleRecalcForChannel(video.channelId)
       }
       await em.save([video, video.channel, newView])
-      await videoRelevanceManager.updateVideoRelevanceValue(em)
       return {
         videoId,
         viewsNum: video.viewsNum,

diff --git a/src/utils/VideoRelevanceManager.ts b/src/utils/VideoRelevanceManager.ts
@@ -5,11 +5,35 @@ import { globalEm } from './globalEm'
 // constant used to parse seconds from creation
 export const NEWNESS_SECONDS_DIVIDER = 60 * 60 * 24
 
+type VideoRelevanceManagerLoops = {
+  fullUpdateLoopTime: number
+  scheduledUpdateLoopTime: number
+}
+
 export class VideoRelevanceManager {
-  private videosToUpdate: Set<string> = new Set()
+  private channelsToUpdate: Set<string> = new Set()
+  private _isVideoRelevanceEnabled = false
+
+  public get isVideoRelevanceEnabled(): boolean {
+    return this._isVideoRelevanceEnabled
+  }
+
+  async init({
+    fullUpdateLoopTime,
+    scheduledUpdateLoopTime,
+  }: VideoRelevanceManagerLoops): Promise<void> {
+    const em = await globalEm
+
+    this.updateScheduledLoop(em, scheduledUpdateLoopTime)
+      .then(() => {
+        /* Do nothing */
+      })
+      .catch((err) => {
+        console.error(err)
+        process.exit(-1)
+      })
 
-  init(intervalMs: number): void {
-    this.updateLoop(intervalMs)
+    this.updateFullUpdateLoop(em, fullUpdateLoopTime)
       .then(() => {
         /* Do nothing */
       })
@@ -19,68 +43,85 @@ export class VideoRelevanceManager {
       })
   }
 
-  scheduleRecalcForVideo(id: string | null | undefined) {
+  turnOnVideoRelevanceManager() {
+    this._isVideoRelevanceEnabled = true
+  }
+
+  scheduleRecalcForChannel(id: string | null | undefined) {
     if (id) {
-      this.videosToUpdate.add(id)
+      this.channelsToUpdate.add(id)
     }
   }
 
   async updateVideoRelevanceValue(em: EntityManager, forceUpdateAll?: boolean) {
-    if (this.videosToUpdate.size || forceUpdateAll) {
-      const [
-        newnessWeight,
-        viewsWeight,
-        commentsWeight,
-        reactionsWeight,
-        [joystreamTimestampWeight, ytTimestampWeight] = [7, 3],
-        defaultChannelWeight,
-      ] = await config.get(ConfigVariable.RelevanceWeights, em)
-      const channelWeight = defaultChannelWeight ?? 1
-      await em.query(`
-        WITH weighted_timestamp AS (
-    SELECT 
-        "video"."id",
-        (
+    if (!this._isVideoRelevanceEnabled || !(this.channelsToUpdate.size || forceUpdateAll)) {
+      return
+    }
+
+    const [
+      newnessWeight,
+      viewsWeight,
+      commentsWeight,
+      reactionsWeight,
+      [joystreamTimestampWeight, ytTimestampWeight] = [7, 3],
+      defaultChannelWeight,
+    ] = await config.get(ConfigVariable.RelevanceWeights, em)
+    const channelWeight = defaultChannelWeight ?? 1
+    const wtEpoch = `((
           extract(epoch from video.created_at)*${joystreamTimestampWeight} +
           COALESCE(extract(epoch from video.published_before_joystream), extract(epoch from video.created_at))*${ytTimestampWeight}
-        ) / ${joystreamTimestampWeight} + ${ytTimestampWeight} as wtEpoch,
-        "channel"."channel_weight" as CW
-    FROM 
-        "video" 
-        INNER JOIN
-          "channel" ON "video"."channel_id" = "channel"."id"
+        ) / ${joystreamTimestampWeight} + ${ytTimestampWeight})`
+
+    await em.query(`
+        WITH videos_with_weight AS (
+        SELECT 
+        video.id as videoId,
+        channel.id as channelId,
+        (ROUND((
+        (extract(epoch from now()) - ${wtEpoch})
+        / ${NEWNESS_SECONDS_DIVIDER} * ${newnessWeight * -1} 
+        + (views_num * ${viewsWeight}) 
+        + (comments_count * ${commentsWeight}) 
+        + (reactions_count * ${reactionsWeight})) 
+        * COALESCE(channel.channel_weight, ${channelWeight}),2)) as videoRelevance
+        FROM video
+        INNER JOIN channel  ON video.channel_id = channel.id
         ${
           forceUpdateAll
             ? ''
-            : `WHERE "video"."id" IN (${[...this.videosToUpdate.values()]
+            : `WHERE video.channel_id in (${[...this.channelsToUpdate.values()]
                 .map((id) => `'${id}'`)
                 .join(', ')})`
         }
-        )
-    UPDATE 
-        "video"
-    SET
-        "video_relevance" = ROUND(
-      (
-        (extract(epoch from now()) - wtEpoch) / ${NEWNESS_SECONDS_DIVIDER} * ${newnessWeight * -1} +
-        (views_num * ${viewsWeight}) +
-        (comments_count * ${commentsWeight}) +
-        (reactions_count * ${reactionsWeight})
-      ) * COALESCE(CW, ${channelWeight}),
-            2)
-    FROM
-        weighted_timestamp
-    WHERE
-        "video".id = weighted_timestamp.id;
+        ORDER BY video.id),
+        
+        top_channel_score as (
+        SELECT 
+        channel.id as channelId,
+        MAX(videoCte.videoRelevance) as maxChannelRelevance
+        FROM channel
+        INNER JOIN videos_with_weight as videoCte on videoCte.channelId = channel.id
+        GROUP BY channel.id)
+        
+        UPDATE video
+        SET video_relevance = COALESCE(topChannelVideo.maxChannelRelevance, 1) 
+        FROM videos_with_weight as videoCte
+        LEFT JOIN top_channel_score as topChannelVideo on topChannelVideo.channelId = videoCte.channelId and topChannelVideo.maxChannelRelevance = videoCte.videoRelevance
+        WHERE video.id = videoCte.videoId;
         `)
-      this.videosToUpdate.clear()
+    this.channelsToUpdate.clear()
+  }
+
+  private async updateScheduledLoop(em: EntityManager, intervalMs: number): Promise<void> {
+    while (true) {
+      await this.updateVideoRelevanceValue(em)
+      await new Promise((resolve) => setTimeout(resolve, intervalMs))
     }
   }
 
-  private async updateLoop(intervalMs: number): Promise<void> {
-    const em = await globalEm
+  private async updateFullUpdateLoop(em: EntityManager, intervalMs: number): Promise<void> {
     while (true) {
-      await this.updateVideoRelevanceValue(em, true)
+      await this.updateVideoRelevanceValue(em)
       await new Promise((resolve) => setTimeout(resolve, intervalMs))
     }
   }