feat: use jobs service wrapper around queue add

Signed-off-by: mertmit <mertmit99@gmail.com>

packages/nocodb/src/modules/jobs/at-import/at-import.controller.ts

@@ -14,7 +14,7 @@ export class AtImportController {
   @Post('/api/v1/db/meta/import/airtable')
   @HttpCode(200)
   async importAirtable(@Request() req) {
-    const job = await this.jobsService.activeQueue.add(JobTypes.AtImport, {
+    const job = await this.jobsService.add(JobTypes.AtImport, {
       ...req.body,
     });
 
@@ -44,7 +44,7 @@ export class AtImportController {
       baseURL = `http://localhost:${process.env.PORT || 8080}`;
     }
 
-    const job = await this.jobsService.activeQueue.add(JobTypes.AtImport, {
+    const job = await this.jobsService.add(JobTypes.AtImport, {
       syncId: req.params.syncId,
       ...(syncSource?.details || {}),
       projectId: syncSource.project_id,

packages/nocodb/src/modules/jobs/export-import/duplicate.controller.ts

@@ -67,7 +67,7 @@ export class DuplicateController {
       user: { id: req.user.id },
     });
 
-    const job = await this.jobsService.activeQueue.add(JobTypes.DuplicateBase, {
+    const job = await this.jobsService.add(JobTypes.DuplicateBase, {
       projectId: project.id,
       baseId: base.id,
       dupProjectId: dupProject.id,
@@ -116,20 +116,17 @@ export class DuplicateController {
       models.map((p) => p.title),
     );
 
-    const job = await this.jobsService.activeQueue.add(
-      JobTypes.DuplicateModel,
-      {
-        projectId: project.id,
-        baseId: base.id,
-        modelId: model.id,
-        title: uniqueTitle,
-        options,
-        req: {
-          user: req.user,
-          clientIp: req.clientIp,
-        },
+    const job = await this.jobsService.add(JobTypes.DuplicateModel, {
+      projectId: project.id,
+      baseId: base.id,
+      modelId: model.id,
+      title: uniqueTitle,
+      options,
+      req: {
+        user: req.user,
+        clientIp: req.clientIp,
       },
-    );
+    });
 
     return { id: job.id, name: job.name };
   }

packages/nocodb/src/modules/jobs/jobs.module.ts

@@ -34,9 +34,9 @@ import { AtImportProcessor } from './at-import/at-import.processor';
     JobsGateway,
     JobsService,
     JobsEventService,
-    DuplicateProcessor,
     ExportService,
     ImportService,
+    DuplicateProcessor,
     AtImportProcessor,
   ],
 })

packages/nocodb/src/modules/jobs/jobs.service.ts

@@ -2,6 +2,7 @@ import { InjectQueue } from '@nestjs/bull';
 import { Injectable } from '@nestjs/common';
 import { Queue } from 'bull';
 import { JOBS_QUEUE, JobStatus } from '../../interface/Jobs';
+import Noco from '../../Noco';
 import { QueueService } from './fallback-queue.service';
 
 @Injectable()
@@ -11,11 +12,16 @@ export class JobsService {
     @InjectQueue(JOBS_QUEUE) private readonly jobsQueue: Queue,
     private readonly fallbackQueueService: QueueService,
   ) {
-    this.activeQueue = this.fallbackQueueService;
-    /* process.env.NC_REDIS_URL
-        ? this.jobsQueue
-        : this.fallbackQueueService;
-    */
+    this.activeQueue = process.env['NC_REDIS_URL']
+      ? this.jobsQueue
+      : this.fallbackQueueService;
+    if (process.env['NC_REDIS_URL'] && !process.env['NC_WORKER_CONTAINER']) {
+      this.jobsQueue.pause(true);
+    }
+  }
+
+  async add(name: string, data: any) {
+    return this.activeQueue.add(name, data);
   }
 
   async jobStatus(jobId: string) {