clean all redis before running any jobs (#40)

src/processors/index.ts

@@ -42,6 +42,9 @@ export class Processors implements ProcessorInterface {
   }
 
   async reset() {
+    // Reset all the queues first
+    await Promise.all(_.map(this.processors, async(p) => await p.clean()));
+
     const modes: Mode[] = await this.getRunningModes();
     // Depending on the mode, we choose different processor to work on
     modes.map(async (m) => {

src/processors/lite.ts

@@ -37,6 +37,10 @@ export class Lite implements ProcessorInterface {
     return this;
   };
 
+  async clean(): Promise<void> {
+    await this.queue.obliterate({ force: true });
+  }
+
   // In lite mode, the reset does nothing other than just trigger the jobs. We can trigger it multiple time, it has no effect
   async reset(): Promise<void> {
     await this.queue.add({}, { jobId: NAME, repeat: { cron: config.cronJob.liteJobExpression}});

src/processors/standard.ts

@@ -46,6 +46,10 @@ export class Standard implements ProcessorInterface {
     return this;
   }
 
+  async clean(): Promise<void> {
+    await this.queue.obliterate({ force: true });
+  }
+
   // Reset and start the state sync until success
   async reset() {
     try {

src/processors/zero.ts

@@ -42,6 +42,10 @@ export class Zero implements ProcessorInterface {
   async reset(): Promise<void> {
     await this.queue.add({}, REPEAT_JOB_OPT);
   }
+
+  async clean(): Promise<void> {
+    await this.queue.obliterate({ force: true });
+  }
 
   async processEvent() {
     const payloads = await this.zeroService.getPayloads();