Merge pull request #24 from AthennaIO/develop

chore(queue): set status to pending it not acked

Author: jlenon7

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@athenna/queue",
-  "version": "5.8.0",
+  "version": "5.9.0",
   "description": "The Athenna queue handler.",
   "license": "MIT",
   "author": "João Lenon <[email protected]>",

package.json

@@ -16,6 +16,11 @@ import type { DatabaseImpl } from '@athenna/database'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
 
 export class DatabaseDriver extends Driver<DatabaseImpl> {
+  /**
+   * Set the acked ids of the driver.
+   */
+  private ackedIds = new Set<string>()
+
   /**
    * The `connection` database that is being used.
    */
@@ -183,7 +188,13 @@ export class DatabaseDriver extends Driver<DatabaseImpl> {
    * ```
    */
   public async ack(id: string) {
-    await this.client.table(this.table).where('id', id).delete()
+    this.ackedIds.add(id)
+
+    await this.client
+      .table(this.table)
+      .where('id', id)
+      .where('status', 'processing')
+      .delete()
   }
 
   /**
@@ -242,10 +253,27 @@ export class DatabaseDriver extends Driver<DatabaseImpl> {
       return
     }
 
+    this.ackedIds.delete(job.id)
+
     job.attemptsLeft--
+    job.status = 'processing'
+
+    await this.client.table(this.table).where('id', job.id).update({
+      status: 'processing',
+      attemptsLeft: job.attemptsLeft
+    })
 
     try {
       await processor(job)
+
+      /**
+       * If the job still exists after processing, it means that the job was
+       * not processed for some reason, so we need to put it back the pending
+       * status.
+       */
+      if (!this.ackedIds.has(job.id)) {
+        job.status = 'pending'
+      }
     } catch (err) {
       const shouldRetry = job.attemptsLeft > 0
 

src/drivers/DatabaseDriver.ts

@@ -12,10 +12,13 @@ import { Options, Uuid } from '@athenna/common'
 import { Driver } from '#src/drivers/Driver'
 import type { ConnectionOptions } from '#src/types'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
-import { NotFoundJobException } from '#src/exceptions/NotFoundJobException'
-import { JobNotProcessingException } from '#src/exceptions/JobNotProcessingException'
 
 export class VanillaDriver extends Driver {
+  /**
+   * Set the acked ids of the driver.
+   */
+  private ackedIds = new Set<string>()
+
   private defineQueue() {
     if (!this.client.queues[this.queueName]) {
       this.client.queues[this.queueName] = []
@@ -171,6 +174,20 @@ export class VanillaDriver extends Driver {
     return this.client.queues[this.queueName].length
   }
 
+  /**
+   * Find a job by its id.
+   *
+   * @example
+   * ```ts
+   * const job = await Queue.getJobById(id)
+   * ```
+   */
+  public async getJobById(id: string) {
+    this.defineQueue()
+
+    return this.client.queues[this.queueName].find(j => j.id === id)
+  }
+
   /**
    * Acknowledge the job removing it from the queue.
    *
@@ -182,14 +199,16 @@ export class VanillaDriver extends Driver {
   public async ack(id: string) {
     this.defineQueue()
 
-    const job = this.client.queues[this.queueName].find(j => j.id === id)
+    this.ackedIds.add(id)
+
+    const job = await this.getJobById(id)
 
     if (!job) {
-      throw new NotFoundJobException(id)
+      return
     }
 
     if (job.status !== 'processing') {
-      throw new JobNotProcessingException(id)
+      return
     }
 
     this.client.queues[this.queueName] = this.client.queues[
@@ -231,11 +250,22 @@ export class VanillaDriver extends Driver {
       return
     }
 
+    this.ackedIds.delete(job.id)
+
     job.attemptsLeft--
     job.status = 'processing'
 
     try {
       await processor(job)
+
+      /**
+       * If the job still exists after processing, it means that the job was
+       * not processed for some reason, so we need to put it back the pending
+       * status.
+       */
+      if (!this.ackedIds.has(job.id)) {
+        job.status = 'pending'
+      }
     } catch (err) {
       const shouldRetry = job.attemptsLeft > 0