power-queues

A lightweight, scalable, and high-performance queue engine for Node.js built on Redis Streams + Lua scripts.

The library is designed for real-world distributed systems that require high throughput, idempotent task execution, automatic recovery, and predictable performance under heavy load.

Unlike traditional Redis-based queues that rely on lists or complex abstractions, power-queues focuses on low-level control, atomic operations, and minimal overhead, making it ideal for high-load backends, microservices, schedulers, telemetry pipelines, and data-processing clusters.

Extends power-redis.

📚 Documentation

Full documentation is available here:
👉 https://power-queues.docs.ihor.bielchenko.com

📦 Installation

npm install power-queues

OR

yarn add power-queues

🧪 Basic usage

const queue = new PowerQueues({
    stream: 'mysql',
    group: 'workers',
});

await queue.loadScripts(true);

await queue.addTasks('ws', [
    { body: 'welcome', userId: 42 },
    { body: 'hello', userId: 51 }
]);

Example of queue worker for sending message to client via WebSocket and executing a MySQL insert transaction:

import express from 'express';
import http from 'http';
import { Server } from 'socket.io';
import mysql from 'mysql2/promise';
import Redis from 'ioredis';
import type { IORedisLike } from 'power-redis';
import { type Task, PowerQueues, } from 'power-queues';
import { 
    isArrFilled,
    isObjFilled,
} from 'full-utils';

const pool = mysql.createPool({
    host: '127.0.0.1',
    port: 3306,
    user: 'user',
    password: 'password',
    database: 'example',
    waitForConnections: true,
    connectionLimit: 10,
});
const redis = new Redis('redis://127.0.0.1:6379');

const app = express();
const server = http.createServer(app);
const io = new Server(server);

export class WebSocketAndMysqlCreateQueue extends PowerQueues {
    public readonly selectStuckCount: number = 256;
    public readonly selectCount: number = 256;
    public readonly retryCount: number = 3;
    public readonly executeSync: boolean = true;
    public readonly removeOnExecuted: boolean = true;
    public redis!: IORedisLike;

    constructor() {
        super();

        this.redis = redis;
    }

    async onExecute(queueName: string, task: Task) {
        const id = uuid();

        io.to(`user:${task.payload.userId}`).emit('alerts', {
            body: task.payload.body,
            id,
        });
        return {
            ...task,
            payload: {
                ...task.payload,
                id,
            },
        };
    }

    async onBatchReady(queueName: string, tasks: Task[]) {
        const values = tasks.map((task) => task.payload);
        const conn = await pool.getConnection();
    
        try {
            await conn.beginTransaction();

            const cols = Object.keys(values[0]);
            const placeholder = `(${cols.map(() => '?').join(',')})`;
            const sql = `INSERT INTO \`alerts\` (${cols.map((c) => `\`${c}\``).join(',')}) VALUES ${values.map(() => placeholder).join(',')}`;
            const params = [];

            for (const row of values) {
                for (const c of cols) {
                    params.push(row[c]);
                }
            }
            await conn.execute(sql, params);
            await conn.commit();
        }
        catch (err) {
            await queryRunner.rollbackTransaction();
            throw err;
        }
        finally {
            await queryRunner.release();
        }
    }

    async onError(err: any, queueName: string, task: Task): Promise<Task> {
        console.error('Alert error', queueName, task, (process.env.NODE_ENV === 'production')
            ? err.message
            : err);
        return task;
    }

    async onBatchError(err: any, queueName: string, tasks: Array<[ string, any, number, string, string, number ]>) {
        console.error('Batch error', queueName, tasks.length, (process.env.NODE_ENV === 'production')
            ? err.message
            : err);
    }
}

const exampleQueue = new ExampleQueue();

exampleQueue.runQueue('ws');

⚖️ power-queues vs Existing Solutions

🚀 Key Features & Advantages

✔ Ultra‑Fast Bulk XADD (Lua‑Powered)

• Adds thousands of messages per second using optimized Lua scripts.
• Minimizes round‑trips to Redis.
• Supports batching based on:
  • number of tasks
  • number of Redis arguments (safe upper bound)
• Outperforms typical list‑based queues and generic abstractions.

✔ Built‑in Idempotent Workers

Every task can carry an idemKey, guaranteeing exactly‑once execution even under:

• worker crashes
• network interruptions
• duplicate task submissions
• process restarts

Idempotency includes:

• Lock key
• Start key
• Done key
• TTL‑managed execution lock
• Automatic release on failure
• Heartbeat mechanism
• Waiting on TTL for contended executions

This makes the engine ideal for:

• payment processing
• external API calls
• high‑value jobs
• distributed pipelines

✔ Stuck Task Recovery (Advanced Stream Scanning)

If a worker crashes mid‑execution, power-queues automatically detects:

• abandoned tasks
• stalled locks
• unfinished start keys

The engine then recovers these tasks back to active processing safely and efficiently.

✔ High‑Throughput Workers

• Batch execution support
• Parallel or sequential processing mode
• Configurable worker loop interval
• Individual and batch‑level error hooks
• Safe retry flow with per‑task attempt counters

✔ Native DLQ (Dead‑Letter Queue)

When retries reach the configured limit:

• the task is moved into ${stream}:dlq
• includes: payload, attempt count, job, timestamp, error text

Perfect for monitoring or later re‑processing.

✔ Zero‑Overhead Serialization

power-queues uses:

• safe JSON encoding
• predictable and optimized payload transformation

This keeps Redis memory layout clean and eliminates overhead.

✔ Complete Set of Lifecycle Hooks

You can extend any part of the execution flow:

• onExecute
• onBatchReady
• onSuccess
• onError
• onBatchError
• onRetry

This allows full integration with:

• monitoring systems
• logging pipelines
• external APM tools
• domain logic

✔ Atomic Script Loading + NOSCRIPT Recovery

Scripts are:

• loaded once
• cached
• auto‑reloaded if Redis restarts
• executed safely via SHA‑based calls

Ensures resilience in failover scenarios.

✔ Job Progress Tracking

Optional per‑job counters:

• job:total
• job:ok
• job:err
• job:ready

Useful for UI dashboards and real‑time job progress visualization.

🧩 Extensibility

power-queues is ideal for building:

• task schedulers
• distributed cron engines
• ETL pipelines
• telemetry processors
• notification workers
• device monitoring systems
• AI job pipelines
• high-frequency background jobs

🧱 Reliability First

Every part of the engine is designed to prevent:

• double execution
• stuck tasks
• orphan locks
• lost messages
• zombie workers
• script desynchronization

The heartbeat + TTL strategy guarantees that no task is "lost" even in chaotic cluster environments.

📜 License

MIT - free for commercial and private use.