☄️ queue-tool (Node.js Adapter)

The high-performance, cross-language, database-agnostic task queue for Node.js and TypeScript. Automatically supports binary MessagePack serialization and atomic claims across PostgreSQL, Redis, and MongoDB.

📦 Installation

npm install queue-tool
# or
pnpm add queue-tool
# or
yarn add queue-tool

Depending on which adapter you use, install the corresponding peer dependency:

• PostgreSQL: npm install pg @types/pg
• Redis: npm install ioredis
• MongoDB: npm install mongodb

🚀 Basic Usage (Node.js / TypeScript)

1. Initialize Adapter

Use the QueueFactory to automatically instantiate the correct adapter based on your connection string:

import { QueueFactory } from 'queue-tool';

// Automatically instantiates PostgresQueueAdapter
const adapter = await QueueFactory.create('postgresql://postgres:password@localhost:5432/queue_tool', {
  completedJobTtlMs: 3600000, // Retain completed jobs for 1 hour
  failedJobTtlMs: 86400000,    // Retain failed jobs for 24 hours
  lockTimeoutMs: 300000,       // Automatically reclaim stuck jobs after 5 mins
  pruneIntervalMs: 60000,      // Run background pruning check every 1 min
});

2. Enqueue Jobs

const job = await adapter.enqueue('image-processing', {
  imageId: 'img-10293',
  operations: ['resize', 'compress'],
}, { 
  priority: 10,       // Higher priority jobs are claimed first
  maxAttempts: 3,     // Auto-retry up to 3 times before failing
  delayMs: 5000       // Wait 5 seconds before making job available
});

3. Claim and Complete Jobs

const job = await adapter.claimJob('image-processing', 'worker-node-1');

if (job) {
  try {
    console.log(`Processing payload:`, job.payload);
    // ... do processing ...
    
    await adapter.completeJob(job.id);
  } catch (error) {
    // Moves back to pending (rescheduled in 5 seconds) or marks failed if attempts exhausted
    await adapter.failJob(job.id, error.message);
  }
}

🦅 NestJS Integration Guide

To cleanly integrate queue-tool inside a NestJS application, you can create a custom dynamic module.

1. Create the Queue Module

Create a file named queue.module.ts:

import { Module, DynamicModule, Global } from '@nestjs/common';
import { QueueFactory } from 'queue-tool';
import { QueueAdapterOptions } from 'queue-tool/dist/types';

export const QUEUE_ADAPTER = 'QUEUE_ADAPTER';

@Global()
@Module({})
export class QueueModule {
  static register(connectionString: string, options?: QueueAdapterOptions): DynamicModule {
    const provider = {
      provide: QUEUE_ADAPTER,
      useFactory: async () => {
        const adapter = await QueueFactory.create(connectionString, options);
        return adapter;
      },
    };

    return {
      module: QueueModule,
      providers: [provider],
      exports: [provider],
    };
  }
}

2. Import Module in App Module

import { Module } from '@nestjs/common';
import { QueueModule } from './queue.module';

@Module({
  imports: [
    QueueModule.register('redis://localhost:6379', {
      lockTimeoutMs: 300000, // 5 minutes
    }),
  ],
})
export class AppModule {}

3. Inject and Use in a Service

import { Injectable, Inject, OnModuleDestroy } from '@nestjs/common';
import { QUEUE_ADAPTER } from './queue.module';
import { IQueueAdapter } from 'queue-tool/dist/types';

@Injectable()
export class TaskService implements OnModuleDestroy {
  constructor(
    @Inject(QUEUE_ADAPTER) private readonly queueAdapter: IQueueAdapter,
  ) {}

  async createProcessTask(data: any) {
    return this.queueAdapter.enqueue('my-task-queue', data, { priority: 5 });
  }

  async onModuleDestroy() {
    await this.queueAdapter.close();
  }
}

4. Create a Background Worker with high-level Worker

import { Injectable, Inject, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { QUEUE_ADAPTER } from './queue.module';
import { IQueueAdapter, Worker } from 'queue-tool';

@Injectable()
export class WorkerService implements OnModuleInit, OnModuleDestroy {
  private worker!: Worker;

  constructor(
    @Inject(QUEUE_ADAPTER) private readonly queueAdapter: IQueueAdapter,
  ) {}

  async onModuleInit() {
    this.worker = new Worker(
      'my-task-queue',
      this.queueAdapter,
      async (job) => {
        console.log('Processing job payload:', job.payload);
        // Your job execution logic here...
      },
      { concurrency: 2, pollIntervalMs: 500 }
    );

    await this.worker.start();
  }

  async onModuleDestroy() {
    await this.worker.stop();
  }
}

🚂 ExpressJS Integration Guide

Create a background worker and API endpoints in your Express application:

import express from 'express';
import { QueueFactory, Worker } from 'queue-tool';

const app = express();
app.use(express.json());

const startApp = async () => {
  // 1. Initialize queue adapter
  const adapter = await QueueFactory.create('redis://localhost:6379');

  // 2. Start worker to process background tasks
  const worker = new Worker('express-tasks', adapter, async (job) => {
    console.log(`Processing background job ${job.id}:`, job.payload);
  });
  await worker.start();

  // 3. Define routes
  app.post('/enqueue', async (req, res) => {
    const job = await adapter.enqueue('express-tasks', req.body);
    res.json({ status: 'enqueued', jobId: job.id });
  });

  app.listen(3000, () => console.log('Server running on port 3000'));
};

startApp();

☄️ Hono Integration Guide

Hono works great with async background workers. Define workers at startup:

import { Hono } from 'hono';
import { QueueFactory, Worker } from 'queue-tool';

const app = new Hono();

// Instantiate adapter
const adapter = await QueueFactory.create('mongodb://localhost:27017/queue_tool');

// Start worker
const worker = new Worker('hono-tasks', adapter, async (job) => {
  console.log(`Hono task running: ${job.id}`);
});
await worker.start();

app.post('/enqueue', async (c) => {
  const body = await c.req.json();
  const job = await adapter.enqueue('hono-tasks', body);
  return c.json({ status: 'enqueued', jobId: job.id });
});

export default app;