fix: fixed remove not existing job

dappforce · Jan 22, 2024 · 9714c4b · 9714c4b
1 parent 1483436
commit 9714c4b
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 0 deletions.
diff --git a/src/modules/queueProcessor/services/producers/datasourceChunksParallelHandling.producer.ts b/src/modules/queueProcessor/services/producers/datasourceChunksParallelHandling.producer.ts
@@ -44,6 +44,10 @@ export class DatasourceChunksParallelHandlingProducer {
             clearInterval(intervalInst);
             const jobRes =
               await this.datasourceChunksParallelHandlingQueue.getJob(job.id);
+            if (!jobRes) {
+              resolve({ jobResult: { fetchedChunkData: [] } });
+              return;
+            }
             await jobRes.remove();
             resolve({ jobResult: JSON.parse(jobRes.returnvalue) });
             return;

diff --git a/src/modules/queueProcessor/services/producers/datasourceHandling.producer.ts b/src/modules/queueProcessor/services/producers/datasourceHandling.producer.ts
@@ -12,6 +12,8 @@ import { CollectEventDataFromDataSourceResponse } from '../../dto/collectEventDa
 import { CollectEventDataHandlerResponse } from '../../../dataAggregator/dto/collectEventDataHandler.response';
 import { CollectEventDataChunkFromDataSourceInput } from '../../dto/collectEventDataChunkFromDataSource.input';
 import { CollectTransfersChunkHandlerResponseResponse } from '../../../dataAggregator/dto/collectTransfersChunkHandlerResponse.response';
+import { NativeTransactionKind } from '../../../../constants/common';
+import { BlockchainTag } from '../../../../constants/blockchain';
 
 @Injectable()
 export class DatasourceHandlingProducer {
@@ -49,6 +51,17 @@ export class DatasourceHandlingProducer {
           if (jobStatus === 'completed' || jobStatus === 'failed') {
             clearInterval(intervalInst);
             const jobRes = await this.datasourceHandlingQueue.getJob(job.id);
+
+            if (!jobRes) {
+              resolve({
+                jobResult: {
+                  latestProcessedBlock: requestData.latestProcessedBlock,
+                  action: requestData.event,
+                  blockchainTag: requestData.blockchainTag,
+                },
+                requestData,
+              });
+            }
             await jobRes.remove();
             resolve({ jobResult: JSON.parse(jobRes.returnvalue), requestData });
             return;