feat: Add ScorePostJob to cron plugin

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 { 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)
@@ -20,6 +21,7 @@ const cronPlugin: FastifyPluginAsync = async (fastfiy) => {
   const statsMonthlyJob = container.resolve(StatsMonthly)
   const checkPostSpamJob = container.resolve(CheckPostSpamJob)
   const deleteReadPostJob = container.resolve(DeletePostReadJob)
+  const scorePostJob = container.resolve(ScorePostJob)
 
   // 덜 실행하면서, 실행되는 순서로 정렬
   // crontime은 UTC 기준으로 작성되기 때문에 KST에서 9시간을 빼줘야함
@@ -51,6 +53,11 @@ 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
@@ -153,6 +160,7 @@ type JobService =
   | StatsMonthly
   | CheckPostSpamJob
   | DeletePostReadJob
+  | ScorePostJob
 
 type BaseJobService = {
   name: string

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/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)