glide-mq

High-performance AI-native message queue for Node.js on Valkey/Redis Streams with a Rust NAPI core.

Quick Start

import { Queue, Worker } from 'glide-mq';

const connection = { addresses: [{ host: 'localhost', port: 6379 }] };

const queue = new Queue('tasks', { connection });
await queue.add('send-email', { to: 'user@example.com', subject: 'Hello' });

const worker = new Worker('tasks', async (job) => {
  console.log(`Processing ${job.name}:`, job.data);
  return { sent: true };
}, { connection, concurrency: 10 });

worker.on('completed', (job) => console.log(`Done: ${job.id}`));
worker.on('failed', (job, err) => console.error(`Failed: ${job.id}`, err.message));

When to Apply

Use this skill when:

• Creating or configuring queues, workers, or producers
• Adding jobs (single, bulk, delayed, priority)
• Setting up retries, backoff, or dead-letter queues
• Building job workflows (parent-child, DAGs, chains)
• Implementing fan-out broadcast patterns
• Configuring cron/interval schedulers
• Setting up connection options (TLS, IAM, AZ-affinity)
• Working with batch processing or rate limiting
• Tracking AI/LLM usage (tokens, cost, model) per job or flow
• Streaming LLM output tokens in real-time
• Implementing human-in-the-loop approval with suspend/resume
• Setting budget caps (tokens, cost) on workflow flows
• Configuring fallback chains for model/provider failover
• Dual-axis rate limiting (RPM + TPM) for LLM API compliance
• Aggregating rolling usage/cost summaries across queues
• Searching jobs by vector similarity (KNN) with Valkey Search
• Exposing queues or broadcasts over the HTTP proxy, including SSE endpoints
• Integrating with frameworks (Hono, Fastify, NestJS, Hapi)
• Deploying in serverless environments (Lambda, Vercel Edge)

Core API by Priority

Key Patterns

Delayed & Priority Jobs

// Delayed: run after 5 minutes
await queue.add('reminder', data, { delay: 300_000 });

// Priority: lower number = higher priority (default: 0)
await queue.add('urgent', data, { priority: 0 });
await queue.add('low-priority', data, { priority: 10 });

// Retries with exponential backoff
await queue.add('webhook', data, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 1000 }
});

Bulk Ingestion (10,000 jobs in ~350ms)

const jobs = items.map(item => ({
  name: 'process',
  data: item,
  opts: { jobId: `item-${item.id}` }
}));
await queue.addBulk(jobs);

Batch Worker (Process Multiple Jobs at Once)

const worker = new Worker('analytics', async (jobs) => {
  // jobs is Job[] when batch is enabled
  await db.insertMany('events', jobs.map(j => j.data));
}, {
  connection,
  batch: { size: 50, timeout: 5000 }
});

Request-Reply (addAndWait)

const result = await queue.addAndWait('compute', { input: 42 }, {
  waitTimeout: 30_000
});
console.log(result); // processor return value

Serverless Producer (No EventEmitter Overhead)

import { Producer } from 'glide-mq';
const producer = new Producer('queue', { connection });
await producer.add('job-name', data);
await producer.close();

Graceful Shutdown

import { gracefulShutdown } from 'glide-mq';

// Registers SIGTERM/SIGINT handlers and returns a handle.
// await blocks until a signal fires - use as last line of your program.
const handle = gracefulShutdown([worker1, worker2, queue, events]);

// For programmatic shutdown (e.g., in tests):
await handle.shutdown();

// To remove signal handlers without closing:
handle.dispose();

Testing Without Valkey

import { TestQueue, TestWorker } from 'glide-mq/testing';
const queue = new TestQueue('tasks');
await queue.add('test-job', { key: 'value' });
const worker = new TestWorker(queue, processor);
await worker.run();

Problem-to-Reference Mapping

Critical Notes

• Node.js 20+ and Valkey 7.0+ (or Redis 7.0+) required
• At-least-once delivery - make processors idempotent
• Priority: lower number = higher priority (0 is default, highest)
• Cluster-native - hash-tagged keys (glide:{queueName}:*) work out of the box
• All queue logic runs as a single Valkey Server Function (FCALL) - 1 round-trip per job
• Connection format uses addresses: [{ host, port }] array, NOT { host, port } object
• Never use customCommand - use typed API methods with dummy keys for cluster routing

Done When

• npm test or the project-equivalent test command passes
• await queue.getJobCounts() matches the expected queue state
• no jobs are left unexpectedly stuck in active
• any QueueEvents or SSE behavior touched by the change has been smoke-tested
• temporary queues, workers, and listeners are closed cleanly

Full Documentation

https://www.glidemq.dev/