feat: use a separate queue and job for progress (#2170)

Signed-off-by: Pranav C <pranavxc@gmail.com>

packages/nocodb/src/lib/noco-jobs/JobsMgr.ts

@@ -42,7 +42,9 @@ export default abstract class JobsMgr {
   }
 
   protected async invokeSuccessCbks(jobName: string, payload: any) {
-    await Promise.all(this.successCbks?.[jobName]?.map(cb => cb(payload)));
+    await Promise.all(
+      this.successCbks?.[jobName]?.map(cb => cb(payload)) || []
+    );
   }
   protected async invokeFailureCbks(
     jobName: string,
@@ -50,7 +52,7 @@ export default abstract class JobsMgr {
     error?: Error
   ) {
     await Promise.all(
-      this.failureCbks?.[jobName]?.map(cb => cb(payload, error))
+      this.failureCbks?.[jobName]?.map(cb => cb(payload, error)) || []
     );
   }
   protected async invokeProgressCbks(
@@ -59,7 +61,7 @@ export default abstract class JobsMgr {
     data?: any
   ) {
     await Promise.all(
-      this.progressCbks?.[jobName]?.map(cb => cb(payload, data))
+      this.progressCbks?.[jobName]?.map(cb => cb(payload, data)) || []
     );
   }
 }

packages/nocodb/src/lib/noco/meta/api/sync/importApis.ts

@@ -9,6 +9,7 @@ import SyncSource from '../../../../noco-models/SyncSource';
 import Noco from '../../../Noco';
 import * as jwt from 'jsonwebtoken';
 const AIRTABLE_IMPORT_JOB = 'AIRTABLE_IMPORT_JOB';
+const AIRTABLE_PROGRESS_JOB = 'AIRTABLE_PROGRESS_JOB';
 
 enum SyncStatus {
   PROGRESS = 'PROGRESS',
@@ -17,24 +18,45 @@ enum SyncStatus {
 }
 
 export default (router: Router, clients: { [id: string]: Socket }) => {
+  // add importer job handler and progress notification job handler
   NocoJobs.jobsMgr.addJobWorker(AIRTABLE_IMPORT_JOB, job);
+  NocoJobs.jobsMgr.addJobWorker(
+    AIRTABLE_PROGRESS_JOB,
+    ({ payload, progress }) => {
+      clients?.[payload?.id]?.emit('progress', {
+        msg: progress?.msg,
+        level: progress?.level,
+        status: progress?.status
+      });
+    }
+  );
+
   NocoJobs.jobsMgr.addProgressCbk(AIRTABLE_IMPORT_JOB, (payload, progress) => {
-    clients?.[payload?.id]?.emit('progress', {
+    NocoJobs.jobsMgr.add(AIRTABLE_PROGRESS_JOB, {
-      msg: progress?.msg,
+      payload,
-      level: progress?.level,
+      progress: {
-      status: SyncStatus.PROGRESS
+        msg: progress?.msg,
+        level: progress?.level,
+        status: progress?.status
+      }
     });
   });
   NocoJobs.jobsMgr.addSuccessCbk(AIRTABLE_IMPORT_JOB, payload => {
-    clients?.[payload?.id]?.emit('progress', {
+    NocoJobs.jobsMgr.add(AIRTABLE_PROGRESS_JOB, {
-      msg: 'Complete!',
+      payload,
-      status: SyncStatus.COMPLETED
+      progress: {
+        msg: 'Complete!',
+        status: SyncStatus.COMPLETED
+      }
     });
   });
   NocoJobs.jobsMgr.addFailureCbk(AIRTABLE_IMPORT_JOB, (payload, error: any) => {
-    clients?.[payload?.id]?.emit('progress', {
+    NocoJobs.jobsMgr.add(AIRTABLE_PROGRESS_JOB, {
-      msg: error?.message || 'Failed due to some internal error',
+      payload,
-      status: SyncStatus.FAILED
+      progress: {
+        msg: error?.message || 'Failed due to some internal error',
+        status: SyncStatus.FAILED
+      }
     });
   });