wire recurrent task to processor/consumer

Author: bryopsida

package-lock.json

@@ -34,6 +34,7 @@
     "@nestjs/core": "^10.0.0",
     "@nestjs/microservices": "^10.0.1",
     "@nestjs/platform-express": "^10.0.0",
+    "@nestjs/schedule": "^3.0.1",
     "@nestjs/swagger": "^7.0.1",
     "@nestjs/terminus": "^10.0.1",
     "bull": "^4.10.4",
@@ -49,6 +50,7 @@
     "@nestjs/cli": "^10.0.0",
     "@nestjs/schematics": "^10.0.0",
     "@nestjs/testing": "^10.0.0",
+    "@types/cron": "^2.0.1",
     "@types/express": "^4.17.13",
     "@types/jest": "29.5.2",
     "@types/node": "18.16.18",

package.json

@@ -1,5 +1,4 @@
 sonar.sources=./src
-sonar.tests=./test
 sonar.javascript.lcov.reportPaths=./coverage/lcov.info
 sonar.projectKey=bryopsida_patchwork
 sonar.organization=bryopsida

sonar-project.properties

@@ -1,8 +1,24 @@
 import { Module } from '@nestjs/common'
 import { ImageService } from './image.service'
 import { RegistryService } from './registry.service'
+import { BullModule } from '@nestjs/bull'
+import { ImageListWorker } from './image-list.consumer'
+import { TaskRegisterService } from './task-register.service'
 
 @Module({
-  providers: [ImageService, RegistryService],
+  imports: [
+    BullModule.registerQueue({
+      name: 'analyzer.fetch.imagelist',
+    }),
+    BullModule.registerQueue({
+      name: 'analyzer.check.updates',
+    }),
+  ],
+  providers: [
+    ImageService,
+    RegistryService,
+    ImageListWorker,
+    TaskRegisterService,
+  ],
 })
 export class AnalyzerModule {}

src/analyzer/analyzer.module.ts

@@ -0,0 +1,14 @@
+import { Processor, Process } from '@nestjs/bull'
+import { Logger } from '@nestjs/common'
+import { Job } from 'bull'
+
+@Processor('analyzer.fetch.imagelist')
+export class ImageListWorker {
+  private readonly logger = new Logger(ImageListWorker.name)
+
+  @Process()
+  async fetchImageList(job: Job<unknown>) {
+    this.logger.log('Fetching list of images')
+    return {}
+  }
+}

src/analyzer/image-list.consumer.ts

@@ -0,0 +1,33 @@
+import { InjectQueue } from '@nestjs/bull'
+import { Injectable, Logger } from '@nestjs/common'
+import { Timeout } from '@nestjs/schedule'
+import { Queue } from 'bull'
+
+@Injectable()
+export class TaskRegisterService {
+  private readonly logger = new Logger(TaskRegisterService.name)
+  private readonly fetchImageListQueue: Queue
+
+  constructor(@InjectQueue('analyzer.fetch.imagelist') fetchImageList: Queue) {
+    this.fetchImageListQueue = fetchImageList
+  }
+
+  @Timeout(5000)
+  async handleTimeout(): Promise<void> {
+    this.logger.log('Ensuring fetch image list job is registered in queue')
+    await this.fetchImageListQueue.add(
+      {},
+      {
+        repeat: {
+          every: 15000,
+        },
+        attempts: -1,
+        backoff: {
+          type: 'exponential',
+          delay: 5000,
+        },
+        jobId: 'fetch-image-list',
+      }
+    )
+  }
+}

src/analyzer/task-register.service.ts

@@ -8,13 +8,15 @@ import { AnalyzerModule } from './analyzer/analyzer.module'
 import { PatcherModule } from './patcher/patcher.module'
 import { BullModule } from '@nestjs/bull'
 import { ConnOptionsService } from './common/conn-options.service'
+import { ScheduleModule } from '@nestjs/schedule'
 
 const connOptionsService = new ConnOptionsService()
 
 @Module({
   imports: [
     ConfigModule.forRoot(),
     LoggerModule.forRoot(),
+    ScheduleModule.forRoot(),
     BullModule.forRoot({
       redis: connOptionsService.getRedisOptions().options,
     }),