feat: ci

app/infra/MQAdapter.ts

@@ -42,7 +42,7 @@ export class MQAdapter implements MQAdapterType {
   async addJobs(key: string, taskId: string, options?: JobsOptions): Promise<boolean> {
     try {
       const queue = this.initQueue(key);
-      await queue.add(key, { jobId: taskId },
+      await queue.add(key, { taskId },
         {
           removeOnComplete: true,
           removeOnFail: true,
@@ -51,6 +51,8 @@ export class MQAdapter implements MQAdapterType {
             type: 'exponential',
             delay: 1000,
           },
+          // remove duplicate job
+          jobId: taskId,
           ...options,
         },
       );

test/infra/MQAdapter.test.ts

@@ -0,0 +1,61 @@
+import assert from 'assert';
+import { setTimeout } from 'node:timers/promises';
+import { Job, Worker } from 'bullmq';
+import { app } from 'egg-mock/bootstrap';
+import { MQAdapter } from '../../app/infra/MQAdapter';
+
+describe('test/infra/MQAdapter.test.ts', () => {
+  let mqAdapter: MQAdapter;
+
+  beforeEach(async () => {
+    mqAdapter = await app.getEggObject(MQAdapter);
+  });
+
+  it('should remove duplicate task', async () => {
+    const queue = mqAdapter.initQueue('banana');
+
+    await mqAdapter.addJobs('banana', '1');
+    await mqAdapter.addJobs('banana', '2');
+    await mqAdapter.addJobs('banana', '1');
+
+    const len = await queue.count();
+    assert.equal(len, 2);
+
+  });
+
+  it('should retry failed task', async () => {
+
+    const queue = mqAdapter.initQueue('apple');
+
+    // retry 1 time;
+    await mqAdapter.addJobs('apple', '1', {
+      attempts: 2,
+      backoff: {
+        type: 'exponential',
+        delay: 1,
+      },
+    });
+
+    let failed = 0;
+
+    const worker = new Worker(queue.name, async (job: Job) => {
+      // console.log(job.data);
+      throw new Error(`${job.data.taskId} error`);
+    });
+
+    worker.on('failed', job => {
+      // console.log('failed', job?.data?.taskId);
+      job && failed++;
+    });
+
+    let len = await queue.count();
+    assert.equal(len, 1);
+
+    // retry triggered
+    await setTimeout(50);
+
+    assert.equal(failed, 2);
+    len = await queue.count();
+    assert.equal(len, 0);
+  });
+});