Add: CheckSpamPostJob to cronPlugin

Author: winverse

packages/velog-cron/src/common/plugins/globals/cronPlugin.ts

@@ -6,6 +6,7 @@ import { DeleteFeedJob } from '@jobs/DeleteFeedJob.js'
 import { FastifyPluginCallback } from 'fastify'
 import { container } from 'tsyringe'
 import { ENV } from '@env'
+import { CheckSpamPostJob } from '@jobs/CheckSpamPostJob.js'
 
 const cronPlugin: FastifyPluginCallback = async (fastfiy, opts, done) => {
   const calculatePostScoreJob = container.resolve(CalculatePostScoreJob)
@@ -15,6 +16,7 @@ const cronPlugin: FastifyPluginCallback = async (fastfiy, opts, done) => {
   const statsDailyJob = container.resolve(StatsDaily)
   const statsWeeklyJob = container.resolve(StatsWeekly)
   const statsMonthlyJob = container.resolve(StatsMonthly)
+  const checkSpamPostJob = container.resolve(CheckSpamPostJob)
 
   // 덜 실행하면서, 실행되는 순서로 정렬
   // crontime은 UTC 기준으로 작성되기 때문에 KST에서 9시간을 빼줘야함
@@ -46,6 +48,11 @@ const cronPlugin: FastifyPluginCallback = async (fastfiy, opts, done) => {
       jobService: calculatePostScoreJob,
       param: 0.5,
     },
+    {
+      name: 'check post spam in every 2 minutes',
+      cronTime: '*/2 * * * *', // every 2 minutes
+      jobService: checkSpamPostJob,
+    },
     // Stats Start
     {
       name: 'providing a count of new users and posts from the past 1 day',
@@ -79,6 +86,13 @@ const cronPlugin: FastifyPluginCallback = async (fastfiy, opts, done) => {
     if (ENV.appEnv !== 'production') {
       const immediateRunJobs = jobDescription.filter((job) => !!job.isImmediateExecute)
       await Promise.all(immediateRunJobs.map(createTick))
+      const crons = immediateRunJobs.map(createJob)
+      await Promise.all(
+        crons.map((cron) => {
+          console.log(`${cron.name} is registered`)
+          cron.start()
+        }),
+      )
     }
 
     if (ENV.dockerEnv === 'production') {
@@ -93,7 +107,6 @@ const cronPlugin: FastifyPluginCallback = async (fastfiy, opts, done) => {
   } catch (error) {
     console.error('Error initializing cron jobs:', error)
   } finally {
-    console.log('finally')
     done()
   }
 }
@@ -105,14 +118,7 @@ function isNeedParamJobService(arg: any): arg is NeedParamJobService {
 }
 
 function isNotNeedParamJobService(arg: any): arg is NotNeedParamJobService {
-  return (
-    arg.jobService instanceof GenerateFeedJob ||
-    arg.jobService instanceof GenerateTrendingWritersJob ||
-    arg.jobService instanceof DeleteFeedJob ||
-    arg.jobService instanceof StatsDaily ||
-    arg.jobService instanceof StatsWeekly ||
-    arg.jobService instanceof StatsMonthly
-  )
+  return arg.jobService instanceof CalculatePostScoreJob === false
 }
 
 async function createTick(description: JobDescription) {
@@ -134,24 +140,27 @@ async function createTick(description: JobDescription) {
 
 type JobDescription = NeedParamJobService | NotNeedParamJobService
 
-type NeedParamJobService = {
+type JobService =
+  | CalculatePostScoreJob
+  | GenerateFeedJob
+  | GenerateTrendingWritersJob
+  | DeleteFeedJob
+  | StatsDaily
+  | StatsWeekly
+  | StatsMonthly
+  | CheckSpamPostJob
+
+type BaseJobService = {
   name: string
   cronTime: string
-  param: any
   isImmediateExecute?: boolean
+}
+
+type NeedParamJobService = BaseJobService & {
+  param: any
   jobService: CalculatePostScoreJob
 }
 
-type NotNeedParamJobService = {
-  name: string
-  cronTime: string
-  param?: undefined
-  isImmediateExecute?: boolean
-  jobService:
-    | GenerateFeedJob
-    | GenerateTrendingWritersJob
-    | DeleteFeedJob
-    | StatsDaily
-    | StatsWeekly
-    | StatsMonthly
+type NotNeedParamJobService = Omit<BaseJobService, 'param'> & {
+  jobService: Exclude<JobService, CalculatePostScoreJob>
 }

packages/velog-cron/src/env.ts

@@ -42,6 +42,7 @@ const env = z.object({
   discordBotToken: z.string(),
   discordStatsChannel: z.string(),
   discordSpamChannel: z.string(),
+  discordErrorChannel: z.string(),
 })
 
 export const ENV = env.parse({
@@ -54,4 +55,5 @@ export const ENV = env.parse({
   discordBotToken: process.env.DISCORD_BOT_TOKEN,
   discordStatsChannel: process.env.DISCORD_STATS_CHANNEL,
   discordSpamChannel: process.env.DISCORD_SPAM_CHANNEL,
+  discordErrorChannel: process.env.DISCORD_ERROR_CHANNEL,
 })

packages/velog-cron/src/jobs/CheckSpamPostJob.ts

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

packages/velog-cron/src/lib/discord/DiscordService.ts

@@ -25,7 +25,7 @@ export class DiscordService {
   public async sendMessage(type: MessageType, message: string) {
     this.isSending = true
 
-    const frequentWord = ['connection pool', 'canceling statement', 'Not allow origin']
+    const frequentWord: string[] = []
     const isFrequentWordIncluded = frequentWord.some((word) => message.includes(word))
 
     if (isFrequentWordIncluded) {
@@ -44,6 +44,7 @@ export class DiscordService {
       const channelMapper: Record<MessageType, string> = {
         stats: ENV.discordStatsChannel,
         spam: ENV.discordSpamChannel,
+        error: ENV.discordErrorChannel,
       }
 
       const channelId = channelMapper[type]
@@ -71,4 +72,4 @@ export class DiscordService {
   }
 }
 
-type MessageType = 'stats' | 'spam'
+type MessageType = 'stats' | 'spam' | 'error'

packages/velog-cron/src/services/PostService/index.ts

@@ -1,10 +1,11 @@
 import { DbService } from '@lib/db/DbService.js'
-import { CheckPostSpamArgs } from '@lib/redis/RedisService'
+import { CheckPostSpamArgs } from '@lib/redis/RedisService.js'
 import { Post, Prisma } from '@prisma/client'
 import { injectable, singleton } from 'tsyringe'
 import geoip from 'geoip-country'
 import { subMonths } from 'date-fns'
-import { DiscordService } from '@lib/discord/DiscordService'
+import { DiscordService } from '@lib/discord/DiscordService.js'
+import { NotFoundError } from '@errors/NotfoundError.js'
 
 interface Service {
   findById(postId: string): Promise<Post | null>
@@ -86,7 +87,7 @@ export class PostService implements Service {
     })
 
     if (!post) {
-      throw new Error('Not found Post')
+      throw new NotFoundError('Not found Post')
     }
 
     const user = await this.db.user.findUnique({
@@ -99,7 +100,7 @@ export class PostService implements Service {
     })
 
     if (!user) {
-      throw new Error('Not found User')
+      throw new NotFoundError('Not found User')
     }
 
     const country = geoip.lookup(ip)?.country ?? ''

packages/velog-server/src/services/PostApiService/index.ts

@@ -289,6 +289,7 @@ export class PostApiService implements Service {
       // check spam
       setTimeout(() => {
         if (isSpam) return
+        if (isTusted) return
         const queueData: CheckPostSpamArgs = {
           post_id: post.id,
           user_id: signedUserId,