Merge pull request #8 from noxify/master

feat(queue): support custom key in `queue.enqueue` even when `runAt` is not specified

.env.example

@@ -0,0 +1,7 @@
+DATABASE_ENGINE="postgres"
+DATABASE_HOST="localhost"
+DATABASE_PORT=5432
+DATABASE_USERNAME="postgres"
+DATABASE_PASSWORD="postgres"
+DATABASE_DBNAME="prisma_queue"
+DATABASE_URL=${DATABASE_ENGINE}://${DATABASE_USERNAME}:${DATABASE_PASSWORD}@${DATABASE_HOST}:${DATABASE_PORT}/${DATABASE_DBNAME}

package.json

@@ -39,7 +39,8 @@
     "test": "npm run lint && npm run prettycheck && npm run typecheck && npm run spec",
     "prepare": "prisma generate",
     "reset": "prisma db push --force-reset && prisma generate",
-    "prepublishOnly": "npm run build"
+    "prepublishOnly": "npm run build",
+    "with-env": "dotenv -e ./.env --"
   },
   "dependencies": {
     "croner": "^8.0.2",
@@ -63,6 +64,7 @@
     "tsx": "^4.15.7",
     "typescript": "^5.5.2",
     "vite-tsconfig-paths": "^4.3.2",
-    "vitest": "^1.6.0"
+    "vitest": "^1.6.0",
+    "dotenv-cli": "^7.4.2"
   }
 }

src/PrismaQueue.spec.ts

@@ -54,9 +54,24 @@ describe("PrismaQueue", () => {
         ]
       `);
       const record = await job.fetch();
+      expect(record.key).toBeNull();
       expect(record?.payload).toEqual({ email: "[email protected]" });
       expect(record?.runAt).toBeInstanceOf(Date);
     });
+
+    it("should properly enqueue a job with a custom key", async () => {
+      const job = await queue.enqueue({ email: "[email protected]" }, { key: "custom-key" });
+      expect(job).toBeInstanceOf(PrismaJob);
+      expect(Object.keys(job)).toMatchInlineSnapshot(`
+        [
+          "id",
+        ]
+      `);
+      const record = await job.fetch();
+      expect(record?.payload).toEqual({ email: "[email protected]" });
+      expect(record?.runAt).toBeInstanceOf(Date);
+      expect(record.key).toBe("custom-key");
+    });
   });
 
   describe("schedule", () => {

src/PrismaQueue.ts

@@ -176,11 +176,11 @@ export class PrismaQueue<
   ): Promise<PrismaJob<T, U>> {
     debug(`enqueue`, this.name, payloadOrFunction, options);
     const { name: queueName, config } = this;
-    const { key, cron = null, maxAttempts = config.maxAttempts, priority = 0, runAt } = options;
+    const { key = null, cron = null, maxAttempts = config.maxAttempts, priority = 0, runAt } = options;
     const record = await this.#prisma.$transaction(async (client) => {
       const payload =
         payloadOrFunction instanceof Function ? await payloadOrFunction(client) : payloadOrFunction;
-      const data = { queue: queueName, cron, payload, maxAttempts, priority };
+      const data = { queue: queueName, cron, payload, maxAttempts, priority, key };
       if (key && runAt) {
         const { count } = await this.model.deleteMany({
           where: {
@@ -198,7 +198,7 @@ export class PrismaQueue<
         const update = { ...data, ...(runAt ? { runAt } : {}) };
         return await this.model.upsert({
           where: { key_runAt: { key, runAt } },
-          create: { key, ...update },
+          create: { ...update },
           update,
         });
       }