feat: Add multi-queue namespaces and PostgreSQL storage backend

.github/workflows/ci.yml

@@ -29,6 +29,19 @@ jobs:
           --health-interval 1s
           --health-timeout 3s
           --health-retries 5
+      postgres:
+        image: postgres:17-alpine
+        ports:
+          - 5432:5432
+        env:
+          POSTGRES_DB: job_queue_test
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 1s
+          --health-timeout 3s
+          --health-retries 5
     steps:
       - name: Checkout
         uses: actions/checkout@v4

docker-compose.yml

@@ -18,3 +18,17 @@ services:
       interval: 1s
       timeout: 3s
       retries: 5
+
+  postgres:
+    image: postgres:17-alpine
+    ports:
+      - "127.0.0.1:5432:5432"
+    environment:
+      POSTGRES_DB: job_queue_test
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 1s
+      timeout: 3s
+      retries: 5

package.json

@@ -47,7 +47,8 @@
   },
   "optionalDependencies": {
     "fast-write-atomic": "^0.4.0",
-    "iovalkey": "^0.2.0"
+    "iovalkey": "^0.2.0",
+    "pg": "^8.20.0"
   },
   "devDependencies": {
     "@platformatic/flame": "^1.6.0",
@@ -64,4 +65,4 @@
   "engines": {
     "node": ">=22.19.0"
   }
-}
+}

src/index.ts

@@ -36,6 +36,7 @@ export type { Storage } from './storage/types.ts'
 export { MemoryStorage } from './storage/memory.ts'
 export { RedisStorage } from './storage/redis.ts'
 export { FileStorage } from './storage/file.ts'
+export { PgStorage } from './storage/pg.ts'
 
 // Queue
 export { Queue } from './queue.ts'

src/queue.ts

@@ -44,7 +44,7 @@ export class Queue<TPayload, TResult = void> extends EventEmitter<QueueEvents<TR
   constructor (config: QueueConfig<TPayload, TResult>) {
     super()
 
-    this.#storage = config.storage
+    this.#storage = config.name ? config.storage.createNamespace(config.name) : config.storage
     this.#workerId = config.workerId ?? randomUUID()
     this.#payloadSerde = config.payloadSerde ?? createJsonSerde<TPayload>()
     this.#resultSerde = config.resultSerde ?? createJsonSerde<TResult>()

src/reaper.ts

@@ -17,6 +17,7 @@ interface LeaderElectionConfig {
 
 interface ReaperConfig<TPayload> {
   storage: Storage
+  name?: string
   payloadSerde?: Serde<TPayload>
   visibilityTimeout?: number
   leaderElection?: LeaderElectionConfig
@@ -62,7 +63,7 @@ export class Reaper<TPayload> extends EventEmitter<ReaperEvents> {
 
   constructor (config: ReaperConfig<TPayload>) {
     super()
-    this.#storage = config.storage
+    this.#storage = config.name ? config.storage.createNamespace(config.name) : config.storage
     this.#payloadSerde = config.payloadSerde ?? createJsonSerde<TPayload>()
     this.#visibilityTimeout = config.visibilityTimeout ?? 30000
     this.#leaderElection = config.leaderElection ?? { enabled: false }

src/storage/file.ts

@@ -846,6 +846,10 @@ export class FileStorage implements Storage {
     }
   }
 
+  createNamespace (name: string): Storage {
+    return new FileStorage({ basePath: join(this.#basePath, name) })
+  }
+
   /**
    * Clear all data (useful for testing)
    */

src/storage/memory.ts

@@ -394,6 +394,10 @@ export class MemoryStorage implements Storage {
     }
   }
 
+  createNamespace (_name: string): Storage {
+    return new MemoryStorage()
+  }
+
   /**
    * Clear all data (useful for testing)
    */