feat: bulk issuance implementation using nest queue (#732)

* feat: implemented bulk issuance with nest queue Signed-off-by: tipusinghaw <[email protected]> * refactor: changed comment message Signed-off-by: tipusinghaw <[email protected]> --------- Signed-off-by: tipusinghaw <[email protected]>
credebl · May 22, 2024 · a5c5906 · a5c5906
1 parent e3dfbd3
commit a5c5906
Show file tree

Hide file tree

Showing 4 changed files with 32 additions and 25 deletions.
diff --git a/apps/api-gateway/src/app.module.ts b/apps/api-gateway/src/app.module.ts
@@ -21,7 +21,6 @@ import { UserModule } from './user/user.module';
 import { ConnectionModule } from './connection/connection.module';
 import { EcosystemModule } from './ecosystem/ecosystem.module';
 import { getNatsOptions } from '@credebl/common/nats.config';
-import { BullModule } from '@nestjs/bull';
 import { CacheModule } from '@nestjs/cache-manager';
 import * as redisStore from 'cache-manager-redis-store';
 import { WebhookModule } from './webhook/webhook.module';
@@ -54,13 +53,7 @@ import { NotificationModule } from './notification/notification.module';
     UtilitiesModule,
     WebhookModule,
     NotificationModule,
-    CacheModule.register({ store: redisStore, host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }),
-    BullModule.forRoot({
-      redis: {
-        host: process.env.REDIS_HOST,
-        port: parseInt(process.env.REDIS_PORT)
-      }
-    })
+    CacheModule.register({ store: redisStore, host: process.env.REDIS_HOST, port: process.env.REDIS_PORT })
   ],
   controllers: [AppController],
   providers: [AppService]

diff --git a/apps/issuance/src/issuance.module.ts b/apps/issuance/src/issuance.module.ts
@@ -27,6 +27,12 @@ import { AwsService } from '@credebl/aws';
     ]),
     CommonModule,
     CacheModule.register({ store: redisStore, host: process.env.REDIS_HOST, port: process.env.REDIS_PORT }),
+    BullModule.forRoot({
+      redis: {
+        host: process.env.REDIS_HOST,
+        port: parseInt(process.env.REDIS_PORT)
+      }
+    }),
     BullModule.registerQueue({
       name: 'bulk-issuance'
     })

diff --git a/apps/issuance/src/issuance.processor.ts b/apps/issuance/src/issuance.processor.ts
@@ -11,11 +11,11 @@ export class BulkIssuanceProcessor {
   @OnQueueActive()
   onActive(job: Job): void {
     this.logger.log(
-      `Processing job ${job.id} of type ${job.name} with data ${JSON.stringify(job.data)}...`
+      `Emitting job status${job.id} of type ${job.name} with data ${JSON.stringify(job.data)}...`
     );
   }
 
-  @Process('issue-credential')
+  @Process()
   async issueCredential(job: Job<unknown>):Promise<void> {
     this.logger.log(
       `Processing job ${job.id} of type ${job.name} with data ${JSON.stringify(job.data)}...`

diff --git a/apps/issuance/src/issuance.service.ts b/apps/issuance/src/issuance.service.ts
@@ -1085,23 +1085,31 @@ async sendEmailForCredentialOffer(sendEmailCredentialOffer: SendEmailCredentialO
       if (!respFile) {
         throw new BadRequestException(ResponseMessages.issuance.error.fileData);
       }
+      // ------------------------ Remove after debugging ---------------------------
+      const queueRunningStatus = await this.bulkIssuanceQueue.isReady();
+      this.logger.log(`respFile::::::`, respFile);
+       // eslint-disable-next-line no-console
+       console.log('queueRunningStatus:::::::::', queueRunningStatus);
+       // ------------------------ Remove after debugging ---------------------------
+
       for (const element of respFile) {
         try {
-          const payload = {
-            data: element.credential_data,
-            fileUploadId: element.fileUploadId,
-            clientId: clientDetails.clientId,
-            cacheId: requestId,
-            credentialDefinitionId: element.credDefId,
-            schemaLedgerId: element.schemaId,
-            isRetry: false,
-            orgId,
-            id: element.id,
-            isLastData: respFile.indexOf(element) === respFile.length - 1
-          };
-
-          await this.delay(500); // Wait for 0.5 secends
-          this.processIssuanceData(payload);
+          this.logger.log(`element log::::::`, element); //Remove after debugging
+          this.bulkIssuanceQueue.add(
+            {
+              data: element.credential_data,
+              fileUploadId: element.fileUploadId,
+              clientId: clientDetails.clientId,
+              cacheId: requestId,
+              credentialDefinitionId: element.credDefId,
+              schemaLedgerId: element.schemaId,
+              isRetry: false,
+              orgId,
+              id: element.id,
+              isLastData: respFile.indexOf(element) === respFile.length - 1
+            },
+            { delay: 5000 }
+          );
         } catch (error) {
           this.logger.error(`Error processing issuance data: ${error}`);
         }