Merge pull request quirrel-dev#771 from quirrel-dev/remove-all-jobs-on-queue

New endpoint to remove all jobs on a queue

Author: Skn0tt

src/api/scheduler/jobs-repo.ts

@@ -145,6 +145,23 @@ export class JobsRepo implements Closable {
     );
   }
 
+  public async emptyQueue(tokenId: string, endpoint: string) {
+    let cursor = 0;
+    const allPromises: Promise<any>[] = [];
+    do {
+      const { cursor: newCursor, jobs } = await this.find(tokenId, endpoint, {
+        cursor,
+      });
+      cursor = newCursor;
+
+      for (const job of jobs) {
+        allPromises.push(this.delete(tokenId, endpoint, job.id));
+      }
+    } while (cursor !== 0);
+
+    await Promise.all(allPromises);
+  }
+
   public async delete(tokenId: string, endpoint: string, id: string) {
     return await this.producer.delete(
       encodeQueueDescriptor(tokenId, endpoint),

src/api/scheduler/routes/queues.ts

@@ -114,6 +114,27 @@ const jobs: FastifyPluginCallback = (fastify, opts, done) => {
     }
   );
 
+  fastify.delete<{ Body: EnqueueJob; Params: QueuesEndpointParams }>(
+    "/:endpoint",
+    {
+      schema: {
+        ...baseSchema,
+        params: EndpointParamsSchema,
+        summary: "Empty a Queue",
+      },
+    },
+    async (request, reply) => {
+      fastify.telemetrist?.dispatch("empty_queue");
+
+      const { tokenId } = request;
+      const { endpoint } = request.params;
+
+      await jobsRepo.emptyQueue(tokenId, endpoint);
+
+      reply.status(204).send();
+    }
+  );
+
   fastify.post<{
     Body: EnqueueJob[];
     Params: QueuesEndpointParams;

src/api/test/jobs.test.ts

@@ -338,6 +338,26 @@ describeAcrossBackends("Jobs", (backend) => {
     expect(lastBody).not.toEqual('{"iWill":"beDeleted"}');
   });
 
+  test("delete all jobs on a queue", async () => {
+    await request(quirrel)
+      .post("/queues/" + endpoint + "/batch")
+      .send([
+        {
+          body: JSON.stringify({ iWill: "beDeleted_1" }),
+          runAt: new Date(Date.now() + 300).toISOString(),
+        },
+        {
+          body: JSON.stringify({ iWill: "beDeleted_2" }),
+          runAt: new Date(Date.now() + 300).toISOString(),
+        },
+      ])
+      .expect(201);
+
+    await request(quirrel).delete(`/queues/${endpoint}`).expect(204);
+
+    await request(quirrel).get(`/queues/${endpoint}`).expect(200, { jobs: [], cursor: null });
+  });
+
   test("idempotent jobs", async () => {
     const id = "sameIdAcrossBothJobs";