Merge pull request #25 from KAMALDEEN333/Job-Queue

created the job-queue for large scale

Author: mijinummi

apps/api/package.json

@@ -0,0 +1,22 @@
+{
+  "name": "gasguard-api",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "ts-node src/index.ts",
+    "typecheck": "tsc --noEmit",
+    "build": "tsc -p tsconfig.json",
+    "start": "node dist/index.js"
+  },
+  "dependencies": {
+    "express": "^4.19.2",
+    "bullmq": "^5.7.11",
+    "ioredis": "^5.3.2"
+  },
+  "devDependencies": {
+    "ts-node": "^10.9.2",
+    "typescript": "^5.6.3",
+    "@types/express": "^4.17.21"
+  }
+}

apps/api/src/index.ts

@@ -0,0 +1,17 @@
+import { createServer } from './server.js'
+import { initQueue } from './queue/index.js'
+
+const port = Number(process.env.PORT || 3000)
+const redisUrl = process.env.REDIS_URL || ''
+const queueName = process.env.SCAN_QUEUE_NAME || 'gasguard:scan'
+
+const { queue, worker, events } = initQueue({ redisUrl, queueName })
+const app = createServer(queue)
+
+app.listen(port, () => {})
+
+process.on('SIGINT', async () => {
+  await worker?.close()
+  await events?.close()
+  process.exit(0)
+})

apps/api/src/queue/index.ts

@@ -0,0 +1,29 @@
+import { Queue, Worker, QueueEvents, JobsOptions } from 'bullmq'
+import { createClient } from 'ioredis'
+import { performScan } from '../scan.js'
+import { createInMemoryQueue } from './memory.js'
+
+type InitOptions = { redisUrl: string; queueName: string }
+
+export function initQueue({ redisUrl, queueName }: InitOptions) {
+  if (!redisUrl) {
+    const { queue, worker, events } = createInMemoryQueue(queueName)
+    return { queue, worker, events }
+  }
+  const connection = new createClient(redisUrl)
+  const queue = new Queue(queueName, { connection })
+  const events = new QueueEvents(queueName, { connection })
+  const worker = new Worker(
+    queueName,
+    async job => {
+      const payload = job.data.payload
+      await job.updateProgress(10)
+      const result = await performScan(payload, p => job.updateProgress(p))
+      return result
+    },
+    { connection }
+  )
+  return { queue, worker, events }
+}
+
+export const defaultJobOptions: JobsOptions = { removeOnComplete: true, removeOnFail: true }

apps/api/src/queue/memory.ts

@@ -0,0 +1,67 @@
+import { Queue } from 'bullmq'
+
+type JobRecord = {
+  id: string
+  state: 'waiting' | 'active' | 'completed' | 'failed'
+  progress: number
+  returnvalue?: unknown
+}
+
+export function createInMemoryQueue(name: string) {
+  const jobs = new Map<string, JobRecord>()
+  const q = new Queue(name, { connection: { host: 'localhost', port: 6379 } as any })
+
+  const queue = {
+    add: async (_name: string, data: any) => {
+      const id = Math.random().toString(36).slice(2)
+      jobs.set(id, { id, state: 'waiting', progress: 0 })
+      setTimeout(async () => {
+        const j = jobs.get(id)
+        if (!j) return
+        j.state = 'active'
+        j.progress = 10
+        try {
+          const result = await simulateScan(data.payload, p => {
+            const jj = jobs.get(id)
+            if (jj) jj.progress = p
+          })
+          j.returnvalue = result
+          j.state = 'completed'
+          j.progress = 100
+        } catch {
+          j.state = 'failed'
+        }
+      }, 0)
+      return { id }
+    },
+    getJob: async (id: string) => {
+      const j = jobs.get(id)
+      if (!j) return null
+      return {
+        id: j.id,
+        progress: j.progress,
+        returnvalue: j.returnvalue,
+        getState: async () => j.state
+      } as any
+    }
+  } as unknown as Queue
+
+  const worker = { close: async () => {} } as any
+  const events = { close: async () => {} } as any
+  return { queue, worker, events }
+}
+
+async function simulateScan(input: any, onProgress: (p: number) => void) {
+  onProgress(25)
+  await sleep(400)
+  onProgress(50)
+  await sleep(400)
+  onProgress(75)
+  await sleep(400)
+  onProgress(100)
+  return { summary: 'ok', issues: [], inputSize: JSON.stringify(input || {}).length }
+}
+
+function sleep(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms))
+}

apps/api/src/scan.ts

@@ -0,0 +1,17 @@
+type ProgressCb = (p: number) => Promise<void> | void
+
+export async function performScan(input: any, onProgress: ProgressCb) {
+  await onProgress(25)
+  await sleep(500)
+  await onProgress(50)
+  await sleep(500)
+  await onProgress(75)
+  await sleep(500)
+  const result = { summary: 'ok', issues: [], inputSize: JSON.stringify(input || {}).length }
+  await onProgress(100)
+  return result
+}
+
+function sleep(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms))
+}

apps/api/src/server.ts

@@ -0,0 +1,44 @@
+import express, { Request, Response } from 'express'
+import { Queue } from 'bullmq'
+
+export function createServer(queue: Queue) {
+  const app = express()
+  app.use(express.json({ limit: '2mb' }))
+
+  app.post('/scan', async (req: Request, res: Response) => {
+    const payload = req.body || {}
+    const isLarge = JSON.stringify(payload).length > 200_000 || payload?.large === true
+    const job = await queue.add('scan', { payload, isLarge }, { removeOnComplete: true, removeOnFail: true })
+    res.status(202).json({ jobId: job.id, statusUrl: `/scan/${job.id}/status`, resultUrl: `/scan/${job.id}/result` })
+  })
+
+  app.get('/scan/:id/status', async (req: Request, res: Response) => {
+    const id = req.params.id
+    const job = await queue.getJob(id)
+    if (!job) {
+      res.status(404).json({ error: 'not_found' })
+      return
+    }
+    const state = await job.getState()
+    const progress = job.progress || 0
+    res.json({ state, progress })
+  })
+
+  app.get('/scan/:id/result', async (req: Request, res: Response) => {
+    const id = req.params.id
+    const job = await queue.getJob(id)
+    if (!job) {
+      res.status(404).json({ error: 'not_found' })
+      return
+    }
+    const state = await job.getState()
+    if (state !== 'completed') {
+      res.status(202).json({ state })
+      return
+    }
+    const result = job.returnvalue
+    res.json({ result })
+  })
+
+  return app
+}

apps/api/tsconfig.json

@@ -0,0 +1,15 @@
+{
+  "compilerOptions": {
+    "target": "ES2020",
+    "module": "ES2020",
+    "moduleResolution": "Node",
+    "outDir": "dist",
+    "rootDir": "src",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "forceConsistentCasingInFileNames": true
+  },
+  "include": ["src/**/*"],
+  "exclude": ["node_modules", "dist"]
+}