@queuert/redis

Redis notify adapter for Queuert - a TypeScript library for database-backed job queues.

What does this do?

Queuert separates job storage (state adapter) from worker coordination (notify adapter). This package provides a notify adapter that uses Redis pub/sub.

The notify adapter handles:

• Broadcasting job scheduling events so workers wake up immediately
• Signaling chain completion for awaitJobChain
• Thundering herd optimization - Uses Lua scripts to atomically limit how many workers query the database

When to use Redis

• High-throughput systems - Redis pub/sub is fast and reliable
• Many workers - Thundering herd optimization prevents database overload
• Existing Redis infrastructure - If you already use Redis for caching/sessions

This is a notify adapter only. You still need a state adapter (PostgreSQL or SQLite) to store jobs.

Requirements

• Node.js 22 or later
• TypeScript 5.0+ (recommended)
• Redis 6 or later

Installation

npm install @queuert/redis

Peer dependencies: queuert

Quick Start

import { createClient, createConsoleLog, defineJobTypeRegistry } from "queuert";
import { createPgStateAdapter } from "@queuert/postgres";
import { createRedisNotifyAdapter } from "@queuert/redis";
import { createClient as createRedisClient } from "redis";

const jobTypeRegistry = defineJobTypeRegistry<{
  "send-email": { entry: true; input: { to: string }; output: { sent: true } };
}>();

const stateAdapter = await createPgStateAdapter({ stateProvider: myPgProvider });

// Redis requires two separate connections (subscribe mode is exclusive)
const redis = createRedisClient();
const redisSubscription = createRedisClient();
await redis.connect();
await redisSubscription.connect();

const notifyAdapter = await createRedisNotifyAdapter({
  provider: {
    publish: async (channel, message) => {
      await redis.publish(channel, message);
    },
    subscribe: async (channel, onMessage) => {
      await redisSubscription.subscribe(channel, onMessage);
      return async () => {
        await redisSubscription.unsubscribe(channel);
      };
    },
    eval: async (script, keys, args) => {
      return redis.eval(script, { keys, arguments: args });
    },
  },
});

const client = await createClient({
  stateAdapter,
  notifyAdapter,
  registry: jobTypeRegistry,
  log: createConsoleLog(),
});

Configuration

const notifyAdapter = await createRedisNotifyAdapter({
  provider: myRedisNotifyProvider, // You provide this - see Quick Start
  channelPrefix: "queuert", // Channel prefix (default: "queuert")
});

How it works

• Uses 3 fixed channels with payload-based filtering (same pattern as PostgreSQL LISTEN/NOTIFY)
• When N jobs are scheduled, only N workers query the database (via atomic Lua script decrements)
• Requires two Redis connections because Redis clients in subscribe mode can't run other commands

API Reference

For the full API reference with types and signatures, see the @queuert/redis reference.

Documentation

For full documentation and examples, see the Queuert documentation.