fix: clear pin jobs on error

src/modules/ipfs/pin.processor.ts

@@ -1,36 +1,57 @@
 import {
+  InjectQueue,
   OnQueueError,
   OnQueueFailed,
   OnQueueStalled,
+  OnQueueWaiting,
   Process,
   Processor,
 } from '@nestjs/bull';
-import { Job } from 'bull';
+import { Job, Queue } from 'bull';
 import { DidStore as DidStoreInfura } from 'didStoreInfura';
 import { DidStore as DidStoreCluster } from 'didStoreCluster';
 import { Logger } from '../logger/logger.service';
 import { PIN_CLAIM_JOB_NAME, PIN_CLAIM_QUEUE_NAME } from './ipfs.types';
+import { inspect } from 'util';
 
 @Processor(PIN_CLAIM_QUEUE_NAME)
 export class PinProcessor {
   constructor(
     private readonly logger: Logger,
     private didStoreCluster: DidStoreCluster,
-    private didStoreInfura: DidStoreInfura
+    private didStoreInfura: DidStoreInfura,
+    @InjectQueue(PIN_CLAIM_QUEUE_NAME)
+    private pinQueue: Queue
   ) {
     this.logger.setContext(PinProcessor.name);
   }
 
   @OnQueueError()
-  onError(error: Error) {
+  async onError(error: Error) {
     this.logger.error(`Error pinning claims ${error.message}`);
+    const jobsCounts = await this.pinQueue.getJobCounts();
+    this.logger.debug(inspect(jobsCounts, { depth: 2, colors: true }));
+    const totalCount = Object.values(jobsCounts).reduce(
+      (acc, curr) => acc + curr,
+      0
+    );
+    this.logger.debug(`Removing ${totalCount} jobs`);
+    await this.pinQueue.empty();
   }
 
   @OnQueueStalled()
   onStalled(job: Job) {
     this.logger.warn(`Stalled ${job.name} claim ${JSON.parse(job.data).cid}`);
   }
 
+  @OnQueueWaiting()
+  async OnQueueWaiting(job: Job) {
+    this.logger.debug(`Waiting ${job.name} claim ${job.data}`);
+    this.logger.debug(
+      `Total number of waiting jobs ${await this.pinQueue.count()}`
+    );
+  }
+
   @OnQueueFailed()
   onFailed(job: Job, err: Error) {
     this.logger.error(