feat: implemented loop logic for bulk-issuance

Signed-off-by: tipusinghaw <tipu.singh@ayanworks.com>

apps/api-gateway/src/app.module.ts

@@ -53,12 +53,6 @@ import * as redisStore from 'cache-manager-redis-store';
         host: process.env.REDIS_HOST,
         port: parseInt(process.env.REDIS_PORT)
       }
-    }),
-    BullModule.registerQueue({
-      name: 'bulk-issuance',
-      redis: {
-        port: parseInt(process.env.REDIS_PORT)
-      }
     })
   ],
   controllers: [AppController],

apps/api-gateway/src/issuance/issuance.controller.ts

@@ -249,25 +249,25 @@ export class IssuanceController {
             throw err;
           }
         }
-      );
-
-      const reqPayload: RequestPayload = {
-        credDefId: credentialDefinitionId,
-        filePath: `${process.env.PWD}/uploadedFiles/import/${newFilename}`,
-        fileName: newFilename
-      };
-      this.logger.log(`reqPayload::::::${JSON.stringify(reqPayload)}`);
-      const importCsvDetails = await this.issueCredentialService.importCsv(
-        reqPayload
-      );
-      const finalResponse: IResponseType = {
-        statusCode: HttpStatus.CREATED,
-        message: ResponseMessages.issuance.success.importCSV,
-        data: importCsvDetails.response
-      };
-      return res.status(HttpStatus.CREATED).json(finalResponse);
-
-    } 
+        const reqPayload: RequestPayload = {
+          credDefId: credentialDefinitionId,
+          fileKey,
+          fileName:file?.originalname
+        };
+        this.logger.log(`reqPayload::::::${JSON.stringify(reqPayload)}`);
+        const importCsvDetails = await this.issueCredentialService.importCsv(
+          reqPayload
+        );
+        const finalResponse: IResponseType = {
+          statusCode: HttpStatus.CREATED,
+          message: ResponseMessages.issuance.success.importCSV,
+          data: importCsvDetails.response
+        };
+        return res.status(HttpStatus.CREATED).json(finalResponse);
+      }
+    } catch (error) {
+      throw new RpcException(error.response ? error.response : error);
+    }
   }
 
 

apps/issuance/interfaces/issuance.interfaces.ts

@@ -87,7 +87,7 @@ export interface FileUpload {
     name?: string;
     upload_type?: string;
     status?: string;
-    orgId?: number | string;
+    orgId?:  string;
     createDateTime?: Date;
     lastChangedDateTime?: Date;
   }

apps/issuance/src/issuance.module.ts

@@ -26,17 +26,8 @@ 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',
-      redis: {
-        port: parseInt(process.env.REDIS_PORT)
-      }
+      name: 'bulk-issuance'
     })
   ],
   controllers: [IssuanceController],

apps/issuance/src/issuance.processor.ts

@@ -21,6 +21,6 @@ export class BulkIssuanceProcessor {
       `Processing job ${job.id} of type ${job.name} with data ${JSON.stringify(job.data)}...`
     );
 
-    this.issuanceService.processIssuanceData(job.data);
+    await this.issuanceService.processIssuanceData(job.data);
   }
 }

apps/issuance/src/issuance.repository.ts

@@ -4,7 +4,7 @@ import { PrismaService } from '@credebl/prisma-service';
 // eslint-disable-next-line camelcase
 import { agent_invitations, credentials, file_data, file_upload, org_agents, organisation, platform_config, shortening_url } from '@prisma/client';
 import { ResponseMessages } from '@credebl/common/response-messages';
-import { FileUpload, FileUploadData, PreviewRequest, SchemaDetails } from '../interfaces/issuance.interfaces';
+import {  FileUploadData, PreviewRequest, SchemaDetails } from '../interfaces/issuance.interfaces';
 @Injectable()
 export class IssuanceRepository {
 
@@ -198,13 +198,13 @@ export class IssuanceRepository {
         }
     }
 
-    async saveFileUploadDetails(fileUploadPayload: FileUpload): Promise<file_upload> {
+    async saveFileUploadDetails(fileUploadPayload): Promise<file_upload> {
         try {
-            const { name, orgId, status, upload_type } = fileUploadPayload;
+            const { name, status, upload_type, orgId } = fileUploadPayload;
             return this.prisma.file_upload.create({
                 data: {
                     name,
-                    orgId: `${orgId}`,
+                    orgId: String(orgId),
                     status,
                     upload_type
                 }
@@ -216,18 +216,15 @@ export class IssuanceRepository {
         }
     }
 
-    async updateFileUploadDetails(fileId: string, fileUploadPayload: FileUpload): Promise<file_upload> {
+    async updateFileUploadDetails(fileId: string, fileUploadPayload): Promise<file_upload> {
         try {
-            const { name, orgId, status, upload_type } = fileUploadPayload;
+            const { status } = fileUploadPayload;
             return this.prisma.file_upload.update({
                 where: {
                     id: fileId
                 },
                 data: {
-                    name,
-                    orgId: `${orgId}`,
-                    status,
-                    upload_type
+                    status
                 }
             });
 
@@ -338,18 +335,22 @@ export class IssuanceRepository {
         }
     }
 
-    async getOrgAgentType(orgAgentId: string): Promise<string> {
+    // eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types, @typescript-eslint/explicit-function-return-type, @typescript-eslint/no-unused-vars
+    async saveFileDetails(fileData) {
         try {
+            // const { fileUpload, isError, referenceId, error, detailError } = fileData;
+            // return this.prisma.file_data.create({
+            //     data: {
+            //         detailError,
+            //         error,
+            //         isError,
+            //         referenceId,
+            //         fileUploadId: fileUpload
+            //     }
+            // });
 
-            const { agent } = await this.prisma.org_agents_type.findFirst({
-                where: {
-                    id: orgAgentId
-                }
-            });
-
-            return agent;
         } catch (error) {
-            this.logger.error(`[getOrgAgentType] - error: ${JSON.stringify(error)}`);
+            this.logger.error(`[saveFileUploadData] - error: ${JSON.stringify(error)}`);
             throw error;
         }
     }

apps/issuance/src/issuance.service.ts

@@ -741,10 +741,9 @@ export class IssuanceService {
 
       const parsedData = JSON.parse(cachedData as string).fileData.data;
       const parsedPrimeDetails = JSON.parse(cachedData as string);
-
       fileUpload.upload_type = FileUploadType.Issuance;
       fileUpload.status = FileUploadStatus.started;
-      fileUpload.orgId = orgId;
+      fileUpload.orgId = String(orgId);
       fileUpload.createDateTime = new Date();
 
       if (parsedPrimeDetails && parsedPrimeDetails.fileName) {
@@ -753,29 +752,30 @@ export class IssuanceService {
 
       respFileUpload = await this.issuanceRepository.saveFileUploadDetails(fileUpload);
 
-      this.logger.log(`respFileUpload----${JSON.stringify(respFileUpload)}`);
-      await parsedData.forEach(async (element, index) => {
-        this.logger.log(`element----${JSON.stringify(element)}`);
-        try {
-          this.bulkIssuanceQueue.add(
-            'issue-credential',
-            {
-              data: element,
-              fileUploadId: respFileUpload.id,
-              cacheId: requestId,
-              credentialDefinitionId: parsedPrimeDetails.credentialDefinitionId,
-              schemaLedgerId: parsedPrimeDetails.schemaLedgerId,
-              orgId,
-              isLastData: index === parsedData.length - 1
-            },
-            { delay: 5000 }
-          );
-        } catch (error) {
-          this.logger.error('Error adding item to the queue:', error);
-        }
+
+      await parsedData.forEach(async (element) => {
 
+        await this.issuanceRepository.saveFileDetails(element);
       });
 
+      // this.logger.log(`respFileUpload----${JSON.stringify(respFileUpload)}`);
+      // await parsedData.forEach(async (element, index) => {
+      //   this.logger.log(`element11----${JSON.stringify(element)}`);
+      //   const payload =
+      //   {
+      //     data: element,
+      //     fileUploadId: respFileUpload.id,
+      //     cacheId: requestId,
+      //     credentialDefinitionId: parsedPrimeDetails.credentialDefinitionId,
+      //     schemaLedgerId: parsedPrimeDetails.schemaLedgerId,
+      //     orgId,
+      //     isLastData: index === parsedData.length - 1
+      //   };
+
+      //   this.processIssuanceData(payload);
+
+      // });
+
       return 'Process initiated for bulk issuance';
     } catch (error) {
       fileUpload.status = FileUploadStatus.interrupted;