refactor: pruning worker processors (#346)

IntersectMBO · Sep 11, 2024 · dec6507 · dec6507
1 parent 9ade333
commit dec6507
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 34 deletions.
diff --git a/worker-service/src/common/common-service.ts b/worker-service/src/common/common-service.ts
@@ -105,8 +105,8 @@ export abstract class CommonService {
       };
       return govActionProposal;
     } catch (e) {
-      this.logger.log(
-        `There has been an exception when fetching data for governance action proposal: ${e}`,
+      this.logger.warn(
+        `Error when fetching GAP metadata from url ${url}; Message: ${e}`,
       );
       return null;
     }

diff --git a/worker-service/src/governance/queues/processors/gov-action-proposal.processor.ts b/worker-service/src/governance/queues/processors/gov-action-proposal.processor.ts
@@ -7,30 +7,32 @@ import {
 import { Logger } from '@nestjs/common';
 import { GovActionProposalService } from '../../services/gov-action-proposal.service';
 
+
 @Processor(QUEUE_NAME_DB_SYNC_GOV_ACTIONS)
 export class GovActionsProposalProcessor extends WorkerHost {
   protected readonly logger = new Logger(GovActionsProposalProcessor.name);
   constructor(
-    private readonly govActionProposalService: GovActionProposalService,
+    private readonly govActionProposalService: GovActionProposalService
   ) {
     super();
   }
   async process(job: Job<any>): Promise<any> {
     switch (job.name) {
       case JOB_NAME_GOV_ACTIONS_SYNC: {
-        this.logger.debug('Data from db-sync for gov action proposals job');
-        return job.data;
+        try {
+          this.logger.debug(
+            `Processing GAP - amount of GAPs to be processed, ${job.data?.length}`,
+          );
+          await this.govActionProposalService.storeGovActionProposalData(
+            job.data,
+          );
+        } catch (error) {
+          this.logger.error(
+            `Error processing job id: ${job.id}, name: ${job.name}. - Error: ${error}`,
+          );
+          throw error;
+        }
       }
     }
   }
-
-  @OnWorkerEvent('completed')
-  onCompleted(job: Job) {
-    const { id, name, queueName, finishedOn, returnvalue } = job;
-    const completionTime = finishedOn ? new Date(finishedOn).toISOString() : '';
-    this.govActionProposalService.storeGovActionProposalData(returnvalue);
-    this.logger.log(
-      `Job id: ${id}, name: ${name} completed in queue ${queueName} on ${completionTime}.`,
-    );
-  }
 }
diff --git a/worker-service/src/governance/queues/processors/vote.processor.ts b/worker-service/src/governance/queues/processors/vote.processor.ts
@@ -6,6 +6,7 @@ import {
 } from '../../../common/constants/bullmq.constants';
 import { Logger } from '@nestjs/common';
 import { VoteService } from '../../services/vote.service';
+import { VoteRequest } from 'src/governance/dto/vote.request';
 
 @Processor(QUEUE_NAME_DB_SYNC_VOTES)
 export class VoteProcessor extends WorkerHost {
@@ -17,19 +18,22 @@ export class VoteProcessor extends WorkerHost {
   async process(job: Job<any>): Promise<any> {
     switch (job.name) {
       case JOB_NAME_VOTE_SYNC: {
-        this.logger.debug('Data from db-sync for votes job');
-        return job.data;
+        try {
+          const voteRequests: VoteRequest[] = job.data;
+          const addresses = voteRequests.map(
+            (voteRequest) => voteRequest.hotAddress,
+          );
+          this.logger.debug(
+            `Processing votes for addresses:, ${JSON.stringify(addresses)}`,
+          );
+          await this.voteService.storeVoteData(voteRequests);
+        } catch (error) {
+          this.logger.error(
+            `Error processing job id: ${job.id}, name: ${job.name}. - Error: ${error}`,
+          );
+          throw error;
+        }
       }
     }
   }
-
-  @OnWorkerEvent('completed')
-  onCompleted(job: Job) {
-    const { id, name, queueName, finishedOn, returnvalue } = job;
-    const completionTime = finishedOn ? new Date(finishedOn).toISOString() : '';
-    this.voteService.storeVoteData(returnvalue);
-    this.logger.log(
-      `Job id: ${id}, name: ${name} completed in queue ${queueName} on ${completionTime}.`,
-    );
-  }
 }
diff --git a/worker-service/src/governance/services/gov-action-proposal.service.ts b/worker-service/src/governance/services/gov-action-proposal.service.ts
@@ -33,16 +33,9 @@ export class GovActionProposalService extends CommonService {
     private readonly configService: ConfigService,
   ) {
     super(dataSource);
-    // this.cronInterval =
-    //   this.configService.get<string>('GOV_ACTION_PROPOSALS_JOB_FREQUENCY') ||
-    //   '0 * * * * *';
     this.logger = new Logger(GovActionProposalService.name);
   }
 
-  // getCronExpression(): string {
-  //   return this.cronInterval;
-  // }
-
   async storeGovActionProposalData(
     govActionProposalRequests: GovActionProposalRequest[],
   ): Promise<void> {

diff --git a/worker-service/src/governance/services/vote.service.ts b/worker-service/src/governance/services/vote.service.ts
@@ -162,6 +162,7 @@ export class VoteService extends CommonService {
     try {
       const prefix = '\\x'; // prefix for each hot address
       const addresses = [...mapHotAddresses.keys()].map((key) => prefix + key);
+      this.logger.debug(`Addresses for fetching from db sync: ${addresses}`);
       const dbData = await this.getDataFromSqlFile(
         SQL_FILE_PATH.GET_VOTES,
         addresses,