Merge pull request #54 from velog-io/feat/post-score-queue

redis queue setup for post score

Author: winverse

apps/cron/src/common/plugins/global/cronPlugin.mts

@@ -4,11 +4,12 @@ import { GenerateFeedJob } from '@jobs/GenerateFeedJob.mjs'
 import { CalculatePostScoreJob } from '@jobs/CalculatePostScoreJob.mjs'
 import { GenerateTrendingWritersJob } from '@jobs/GenerateTrendingWritersJob.mjs'
 import { DeleteFeedJob } from '@jobs/DeleteFeedJob.mjs'
-import { FastifyPluginAsync } from 'fastify'
+import type { FastifyPluginAsync } from 'fastify'
 import { container } from 'tsyringe'
 import { ENV } from '@env'
-import { CheckSpamPostJob } from '@jobs/CheckSpamPostJob.mjs'
+import { CheckPostSpamJob } from '@jobs/CheckPostSpamJob.mjs'
 import { DeletePostReadJob } from '@jobs/DeletePostReadJob.mjs'
+import { ScorePostJob } from '@jobs/ScorePostJob.mjs'
 
 const cronPlugin: FastifyPluginAsync = async (fastfiy) => {
   const calculatePostScoreJob = container.resolve(CalculatePostScoreJob)
@@ -18,8 +19,9 @@ const cronPlugin: FastifyPluginAsync = async (fastfiy) => {
   const statsDailyJob = container.resolve(StatsDaily)
   const statsWeeklyJob = container.resolve(StatsWeekly)
   const statsMonthlyJob = container.resolve(StatsMonthly)
-  const checkSpamPostJob = container.resolve(CheckSpamPostJob)
+  const checkPostSpamJob = container.resolve(CheckPostSpamJob)
   const deleteReadPostJob = container.resolve(DeletePostReadJob)
+  const scorePostJob = container.resolve(ScorePostJob)
 
   // 덜 실행하면서, 실행되는 순서로 정렬
   // crontime은 UTC 기준으로 작성되기 때문에 KST에서 9시간을 빼줘야함
@@ -51,10 +53,15 @@ const cronPlugin: FastifyPluginAsync = async (fastfiy) => {
       jobService: calculatePostScoreJob,
       param: 0.5,
     },
+    {
+      name: 'score post in every 1 minutes',
+      cronTime: '*/1 * * * *', // every 1 minutes
+      jobService: scorePostJob,
+    },
     {
       name: 'check post spam in every 2 minutes',
       cronTime: '*/2 * * * *', // every 2 minutes
-      jobService: checkSpamPostJob,
+      jobService: checkPostSpamJob,
     },
     {
       name: 'delete post read in every 2 minutes',
@@ -151,8 +158,9 @@ type JobService =
   | StatsDaily
   | StatsWeekly
   | StatsMonthly
-  | CheckSpamPostJob
+  | CheckPostSpamJob
   | DeletePostReadJob
+  | ScorePostJob
 
 type BaseJobService = {
   name: string

apps/cron/src/jobs/CheckSpamPostJob.mts renamed to apps/cron/src/jobs/CheckPostSpamJob.mts

@@ -1,12 +1,12 @@
 import { injectable, singleton } from 'tsyringe'
 import { Job, JobProgress } from './JobProgress.mjs'
 import { PostService } from '@services/PostService/index.mjs'
-import { CheckPostSpamArgs, RedisService } from '@lib/redis/RedisService.mjs'
+import { type CheckPostSpamQueueData, RedisService } from '@lib/redis/RedisService.mjs'
 import { DiscordService } from '@lib/discord/DiscordService.mjs'
 
 @injectable()
 @singleton()
-export class CheckSpamPostJob extends JobProgress implements Job {
+export class CheckPostSpamJob extends JobProgress implements Job {
   constructor(
     private readonly redis: RedisService,
     private readonly discord: DiscordService,
@@ -23,7 +23,7 @@ export class CheckSpamPostJob extends JobProgress implements Job {
     while (true) {
       const item = await this.redis.lindex(spamQueueName, 0)
       if (!item) break
-      const data: CheckPostSpamArgs = JSON.parse(item)
+      const data: CheckPostSpamQueueData = JSON.parse(item)
       try {
         await this.postService.checkSpam(data)
       } catch (error) {

apps/cron/src/jobs/ScorePostJob.mts

@@ -0,0 +1,44 @@
+import { injectable, singleton } from 'tsyringe'
+import { Job, JobProgress } from './JobProgress.mjs'
+import { RedisService } from '@lib/redis/RedisService.mjs'
+import { PostService } from '@services/PostService/index.mjs'
+import { ScorePostQueueData } from '@packages/database/velog-redis'
+import { DiscordService } from '@lib/discord/DiscordService.mjs'
+
+@singleton()
+@injectable()
+export class ScorePostJob extends JobProgress implements Job {
+  constructor(
+    private readonly postService: PostService,
+    private readonly redis: RedisService,
+    private readonly discord: DiscordService,
+  ) {
+    super()
+  }
+  public async runner(): Promise<void> {
+    console.log('ScorePostJob start...')
+    console.time('ScorePostJob')
+
+    const scorePostQueueName = this.redis.queueName.scorePost
+    let handledQueueCount = 0
+
+    while (true) {
+      const item = await this.redis.lindex(scorePostQueueName, 0)
+      if (!item) break
+      const data: ScorePostQueueData = JSON.parse(item)
+      try {
+        await this.postService.scoreCalculator(data.post_id)
+      } catch (error) {
+        console.log('ScorePostJob error', error)
+        const message = { message: 'ScorePostJob error', payload: item, error: error }
+        this.discord.sendMessage('error', JSON.stringify(message))
+      } finally {
+        await this.redis.lpop(scorePostQueueName)
+        handledQueueCount++
+      }
+    }
+
+    console.log(`handled Queue count: ${handledQueueCount}`)
+    console.timeEnd('ScorePostJob')
+  }
+}

apps/cron/src/lib/redis/RedisService.mts

@@ -1,9 +1,11 @@
 import { injectable, singleton } from 'tsyringe'
 import { ENV } from '@env'
 import { RedisService as Redis } from '@packages/database/velog-redis'
-export type { ChangeEmailArgs } from '@packages/database/velog-redis'
-export type { CheckPostSpamArgs } from '@packages/database/velog-redis'
-export type { CreateFeedArgs } from '@packages/database/velog-redis'
+export type {
+  ChangeEmailArgs,
+  CheckPostSpamQueueData,
+  CreateFeedQueueData,
+} from '@packages/database/velog-redis'
 
 @injectable()
 @singleton()

apps/cron/src/routes/posts/v1/index.mts

@@ -19,11 +19,13 @@ const v1: FastifyPluginCallback = (fastify, opts, done) => {
     },
   )
 
+  // dev 환경에서만 사용 가능
   fastify.patch('/score', async (_, reply) => {
     const processedPostsCount = await postController.calculateRecentPostScore()
     reply.status(HttpStatus.OK).send({ processedPostsCount })
   })
 
+  // dev 환경에서만 사용 가능
   fastify.post('/test/spam-filter', async (_, reply) => {
     await postController.spamFilterTestRunner()
     reply.status(HttpStatus.OK).send(HttpStatusMessage.Ok)

apps/cron/src/services/FeedService/index.mts

@@ -12,7 +12,7 @@ interface Service {
 @injectable()
 @singleton()
 export class FeedService implements Service {
-  constructor(
+constructor(
     private readonly db: DbService,
     private readonly utils: UtilsService,
     private readonly followUserService: FollowUserService,

apps/cron/src/services/PostService/index.mts

@@ -4,12 +4,12 @@ import { injectable, singleton } from 'tsyringe'
 import geoip from 'geoip-country'
 import { subMonths } from 'date-fns'
 import { DiscordService } from '@lib/discord/DiscordService.mjs'
-import { CheckPostSpamArgs } from '@lib/redis/RedisService.mjs'
+import type { CheckPostSpamQueueData } from '@lib/redis/RedisService.mjs'
 
 interface Service {
   findById(postId: string): Promise<Post | null>
   scoreCalculator(postId: string): Promise<void>
-  checkSpam(args: CheckPostSpamArgs): Promise<void>
+  checkSpam(data: CheckPostSpamQueueData): Promise<void>
 }
 
 @singleton()
@@ -68,7 +68,7 @@ export class PostService implements Service {
     })
   }
 
-  public async checkSpam({ post_id, user_id, ip }: CheckPostSpamArgs): Promise<void> {
+  public async checkSpam({ post_id, user_id, ip }: CheckPostSpamQueueData): Promise<void> {
     const post = await this.db.post.findUnique({
       where: {
         id: post_id,

apps/server/src/lib/redis/RedisService.ts

@@ -3,8 +3,8 @@ import { injectable, singleton } from 'tsyringe'
 import { RedisService as Redis } from '@packages/database/velog-redis'
 export type {
   ChangeEmailArgs,
-  CheckPostSpamArgs,
-  CreateFeedArgs,
+  CheckPostSpamQueueData,
+  CreateFeedQueueData,
 } from '@packages/database/velog-redis'
 
 @injectable()

apps/server/src/services/PostApiService/index.mts

@@ -17,7 +17,11 @@ import { SeriesService } from '@services/SeriesService/index.mjs'
 import { SearchService } from '@services/SearchService/index.js'
 import { ExternalIntegrationService } from '@services/ExternalIntegrationService/index.js'
 import { PostService } from '@services/PostService/index.js'
-import { CreateFeedArgs, RedisService, CheckPostSpamArgs } from '@lib/redis/RedisService.js'
+import {
+  RedisService,
+  type CreateFeedQueueData,
+  type CheckPostSpamQueueData,
+} from '@lib/redis/RedisService.js'
 import { GraphcdnService } from '@lib/graphcdn/GraphcdnService.js'
 import { ImageService } from '@services/ImageService/index.js'
 import { UserService } from '@services/UserService/index.js'
@@ -315,24 +319,24 @@ export class PostApiService implements Service {
       // create feed
       setTimeout(() => {
         if (!post) return
-        const queueData: CreateFeedArgs = {
+        const queueData: CreateFeedQueueData = {
           fk_following_id: signedUserId,
           fk_post_id: post.id,
         }
-        this.redis.createFeedQueue(queueData)
+        this.redis.addToCreateFeedQueue(queueData)
       }, 0)
 
       // check spam
       setTimeout(() => {
         if (!post) return
         if (isSpam) return
         if (isTusted) return
-        const queueData: CheckPostSpamArgs = {
+        const queueData: CheckPostSpamQueueData = {
           post_id: post.id,
           user_id: signedUserId,
           ip,
         }
-        this.redis.addToSpamCheckQueue(queueData)
+        this.redis.addToCheckPostSpamQueue(queueData)
       }, 0)
     }
 

apps/server/src/services/PostService/index.ts

@@ -405,17 +405,7 @@ export class PostService implements Service {
     }
   }
   public async updatePostScore(postId: string) {
-    try {
-      await axios.patch(
-        `${ENV.cronHost}/api/posts/v1/score/${postId}`,
-        {},
-        {
-          headers: {
-            'Cron-Api-Key': ENV.cronApiKey,
-          },
-        },
-      )
-    } catch (_) {}
+    await this.redis.addToScorePostQueue({ post_id: postId })
   }
   public shortDescription(post: Post): string {
     if (post.short_description) return post.short_description

packages/database/src/velog-redis/index.mts

@@ -4,7 +4,9 @@ interface Service {
   connection(): Promise<void>
   get generateKey(): GenerateRedisKey
   get queueName(): Record<QueueName, string>
-  createFeedQueue(data: CreateFeedArgs): Promise<number>
+  addToCreateFeedQueue(data: CreateFeedQueueData): Promise<number>
+  addToCheckPostSpamQueue(data: CheckPostSpamQueueData): Promise<number>
+  addToScorePostQueue(data: ScorePostQueueData): Promise<number>
 }
 
 type RedisOptions = {
@@ -53,17 +55,23 @@ export class RedisService extends Redis.default implements Service {
     return {
       createFeed: 'queue:feed',
       checkPostSpam: 'queue:checkPostSpam',
+      scorePost: 'queue:scorePost',
     }
   }
 
-  public async createFeedQueue(data: CreateFeedArgs): Promise<number> {
+  public addToCreateFeedQueue(data: CreateFeedQueueData) {
     const queueName = this.queueName.createFeed
-    return await this.lpush(queueName, JSON.stringify(data))
+    return this.lpush(queueName, JSON.stringify(data))
   }
 
-  public async addToSpamCheckQueue(data: CheckPostSpamArgs): Promise<number> {
+  public addToCheckPostSpamQueue(data: CheckPostSpamQueueData): Promise<number> {
     const queueName = this.queueName.checkPostSpam
-    return await this.lpush(queueName, JSON.stringify(data))
+    return this.lpush(queueName, JSON.stringify(data))
+  }
+
+  public addToScorePostQueue(data: ScorePostQueueData): Promise<number> {
+    const queueName = this.queueName.scorePost
+    return this.lpush(queueName, JSON.stringify(data))
   }
 }
 
@@ -81,20 +89,24 @@ type GenerateRedisKey = {
   deployBook: (bookId: string) => string
 }
 
-type QueueName = 'createFeed' | 'checkPostSpam'
+type QueueName = 'createFeed' | 'checkPostSpam' | 'scorePost'
 
 export type ChangeEmailArgs = {
   email: string
   userId: string
 }
 
-export type CreateFeedArgs = {
+export type CreateFeedQueueData = {
   fk_following_id: string
   fk_post_id: string
 }
 
-export type CheckPostSpamArgs = {
+export type CheckPostSpamQueueData = {
   post_id: string
   user_id: string
   ip: string
 }
+
+export type ScorePostQueueData = {
+  post_id: string
+}