feat(adapter): add knex adapter

Author: RomainLanz

compose.yml

@@ -2,4 +2,13 @@ services:
   redis:
     image: redis:alpine
     ports:
-      - "6379:6379"
+      - "6379:6379"
+
+  postgres:
+    image: postgres:alpine
+    ports:
+      - "5432:5432"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: queue_test

examples/config.ts

@@ -1,21 +1,29 @@
-import type { QueueManagerConfig } from '#types/main'
-import { Redis } from 'ioredis'
 import { redis } from '#drivers/redis_adapter'
 import { sync } from '#drivers/sync_adapter'
-
-export const redisConnection = new Redis({
-  host: 'localhost',
-  port: 6379,
-  keyPrefix: 'boringnode::queue::',
-  db: 0,
-})
+import { knex } from '#drivers/knex_adapter'
+import type { QueueManagerConfig } from '#types/main'
 
 export const config: QueueManagerConfig = {
-  default: 'redis',
+  default: 'knex',
 
   adapters: {
     sync: sync(),
-    redis: redis(redisConnection),
+    redis: redis({
+      host: 'localhost',
+      port: 6379,
+      keyPrefix: 'boringnode::queue::',
+      db: 0,
+    }),
+    knex: knex({
+      client: 'pg',
+      connection: {
+        host: process.env.PG_HOST || 'localhost',
+        port: Number.parseInt(process.env.PG_PORT || '5432', 10),
+        user: process.env.PG_USER || 'postgres',
+        password: process.env.PG_PASSWORD || 'postgres',
+        database: process.env.PG_DATABASE || 'queue_test',
+      },
+    }),
   },
 
   worker: {

package.json

@@ -46,12 +46,17 @@
     "@japa/expect-type": "^2.0.3",
     "@japa/file-system": "^2.3.2",
     "@japa/runner": "^4.4.0",
+    "@types/better-sqlite3": "^7.6.13",
     "@types/node": "^24.3.1",
+    "@types/pg": "^8",
+    "better-sqlite3": "^12.5.0",
     "bullmq": "^5.65.1",
     "c8": "^10.1.3",
     "del-cli": "^7.0.0",
     "eslint": "^9.35.0",
     "ioredis": "^5.7.0",
+    "knex": "^3.1.0",
+    "pg": "^8.16.3",
     "prettier": "^3.6.2",
     "release-it": "^19.0.4",
     "testcontainers": "^11.5.1",

src/drivers/knex_adapter.ts

@@ -0,0 +1,288 @@
+import KnexPkg from 'knex'
+import type { Knex } from 'knex'
+import type { Adapter, AcquiredJob } from '#contracts/adapter'
+import type { JobData } from '#types/main'
+
+export interface KnexAdapterOptions {
+  connection: Knex
+  tableName?: string
+  ownsConnection?: boolean
+}
+
+type KnexConfig = Knex | Knex.Config
+
+/**
+ * Create a new Knex adapter factory.
+ * Accepts either a Knex instance or a Knex configuration object.
+ *
+ * When passing a config object, the adapter will create and manage
+ * the connection lifecycle (closing it on destroy).
+ *
+ * When passing a Knex instance, the caller is responsible for
+ * managing the connection lifecycle.
+ */
+export function knex(config: KnexConfig, tableName?: string) {
+  return () => {
+    const isKnexInstance = typeof config === 'function'
+    const connection = isKnexInstance ? config : KnexPkg(config)
+    return new KnexAdapter({ connection, tableName, ownsConnection: !isKnexInstance })
+  }
+}
+
+/**
+ * Knex adapter for the queue system.
+ * Stores jobs in a SQL database using Knex.
+ */
+export class KnexAdapter implements Adapter {
+  readonly #connection: Knex
+  readonly #tableName: string
+  readonly #ownsConnection: boolean
+  #workerId: string = ''
+  #initialized: boolean = false
+
+  constructor(config: KnexAdapterOptions) {
+    this.#connection = config.connection
+    this.#tableName = config.tableName ?? 'queue_jobs'
+    this.#ownsConnection = config.ownsConnection ?? false
+  }
+
+  setWorkerId(workerId: string): void {
+    this.#workerId = workerId
+  }
+
+  /**
+   * Ensure the jobs table exists.
+   * Creates it if not exists, handles race conditions.
+   */
+  async #ensureTableExists(): Promise<void> {
+    if (this.#initialized) return
+
+    try {
+      await this.#connection.schema.createTable(this.#tableName, (table) => {
+        table.string('id', 255).notNullable()
+        table.string('queue', 255).notNullable()
+        table.enu('status', ['pending', 'active', 'delayed']).notNullable()
+        table.text('data').notNullable()
+        table.bigint('score').unsigned().nullable()
+        table.string('worker_id', 255).nullable()
+        table.bigint('acquired_at').unsigned().nullable()
+        table.bigint('execute_at').unsigned().nullable()
+        table.primary(['id', 'queue'])
+        table.index(['queue', 'status', 'score'])
+        table.index(['queue', 'status', 'execute_at'])
+      })
+    } catch {
+      /**
+       * If table creation fails, verify the table actually exists.
+       * This handles race conditions where multiple instances try to create
+       * the table simultaneously.
+       */
+      const hasTable = await this.#connection.schema.hasTable(this.#tableName)
+      if (!hasTable) {
+        throw new Error(`Failed to create table "${this.#tableName}"`)
+      }
+    }
+
+    this.#initialized = true
+  }
+
+  async destroy(): Promise<void> {
+    if (this.#ownsConnection) {
+      await this.#connection.destroy()
+    }
+  }
+
+  async pop(): Promise<AcquiredJob | null> {
+    return this.popFrom('default')
+  }
+
+  async popFrom(queue: string): Promise<AcquiredJob | null> {
+    await this.#ensureTableExists()
+
+    const now = Date.now()
+
+    // First, move ready delayed jobs to pending
+    await this.#processDelayedJobs(queue, now)
+
+    // Use a transaction to atomically pop a job
+    return this.#connection.transaction(async (trx) => {
+      // Select the highest priority job (lowest score)
+      const job = await trx(this.#tableName)
+        .where('queue', queue)
+        .where('status', 'pending')
+        .orderBy('score', 'asc')
+        .first()
+
+      if (!job) {
+        return null
+      }
+
+      // Update job to active status
+      await trx(this.#tableName)
+        .where('id', job.id)
+        .where('queue', queue)
+        .update({
+          status: 'active',
+          worker_id: this.#workerId,
+          acquired_at: now,
+        })
+
+      const jobData: JobData = JSON.parse(job.data)
+
+      return {
+        ...jobData,
+        acquiredAt: now,
+      }
+    })
+  }
+
+  async #processDelayedJobs(queue: string, now: number): Promise<void> {
+    // Get all ready delayed jobs
+    const delayedJobs = await this.#connection(this.#tableName)
+      .where('queue', queue)
+      .where('status', 'delayed')
+      .where('execute_at', '<=', now)
+      .select('id', 'data')
+
+    if (delayedJobs.length === 0) return
+
+    // Move them to pending
+    for (const job of delayedJobs) {
+      const jobData: JobData = JSON.parse(job.data)
+      const priority = jobData.priority ?? 5
+      const score = priority * 1e13 + now
+
+      await this.#connection(this.#tableName)
+        .where('id', job.id)
+        .where('queue', queue)
+        .update({
+          status: 'pending',
+          score,
+          execute_at: null,
+        })
+    }
+  }
+
+  async completeJob(jobId: string, queue: string): Promise<void> {
+    await this.#ensureTableExists()
+
+    await this.#connection(this.#tableName)
+      .where('id', jobId)
+      .where('queue', queue)
+      .delete()
+  }
+
+  async failJob(jobId: string, queue: string, _error?: Error): Promise<void> {
+    await this.#ensureTableExists()
+
+    await this.#connection(this.#tableName)
+      .where('id', jobId)
+      .where('queue', queue)
+      .delete()
+  }
+
+  async retryJob(jobId: string, queue: string, retryAt?: Date): Promise<void> {
+    await this.#ensureTableExists()
+
+    const now = Date.now()
+
+    // Get the active job
+    const activeJob = await this.#connection(this.#tableName)
+      .where('id', jobId)
+      .where('queue', queue)
+      .where('status', 'active')
+      .first()
+
+    if (!activeJob) return
+
+    const jobData: JobData = JSON.parse(activeJob.data)
+    jobData.attempts = (jobData.attempts || 0) + 1
+
+    const updatedData = JSON.stringify(jobData)
+
+    if (retryAt && retryAt.getTime() > now) {
+      // Move to delayed
+      await this.#connection(this.#tableName)
+        .where('id', jobId)
+        .where('queue', queue)
+        .update({
+          status: 'delayed',
+          data: updatedData,
+          worker_id: null,
+          acquired_at: null,
+          score: null,
+          execute_at: retryAt.getTime(),
+        })
+    } else {
+      // Move back to pending
+      const priority = jobData.priority ?? 5
+      const score = priority * 1e13 + now
+
+      await this.#connection(this.#tableName)
+        .where('id', jobId)
+        .where('queue', queue)
+        .update({
+          status: 'pending',
+          data: updatedData,
+          worker_id: null,
+          acquired_at: null,
+          score,
+          execute_at: null,
+        })
+    }
+  }
+
+  async push(jobData: JobData): Promise<void> {
+    return this.pushOn('default', jobData)
+  }
+
+  async pushOn(queue: string, jobData: JobData): Promise<void> {
+    await this.#ensureTableExists()
+
+    const priority = jobData.priority ?? 5
+    const timestamp = Date.now()
+    const score = priority * 1e13 + timestamp
+
+    await this.#connection(this.#tableName).insert({
+      id: jobData.id,
+      queue,
+      status: 'pending',
+      data: JSON.stringify(jobData),
+      score,
+    })
+  }
+
+  async pushLater(jobData: JobData, delay: number): Promise<void> {
+    return this.pushLaterOn('default', jobData, delay)
+  }
+
+  async pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
+    await this.#ensureTableExists()
+
+    const executeAt = Date.now() + delay
+
+    await this.#connection(this.#tableName).insert({
+      id: jobData.id,
+      queue,
+      status: 'delayed',
+      data: JSON.stringify(jobData),
+      execute_at: executeAt,
+    })
+  }
+
+  async size(): Promise<number> {
+    return this.sizeOf('default')
+  }
+
+  async sizeOf(queue: string): Promise<number> {
+    await this.#ensureTableExists()
+
+    const result = await this.#connection(this.#tableName)
+      .where('queue', queue)
+      .where('status', 'pending')
+      .count('* as count')
+      .first()
+
+    return Number(result?.count ?? 0)
+  }
+}