Glide Mq Migrate Bee

Migrate from Bee-Queue to glide-mq

When to Apply

Use this skill when:

• Replacing bee-queue with glide-mq in an existing project
• Converting Bee-Queue's chained job API to glide-mq's options API
• Updating connection configuration from ioredis to valkey-glide
• Upgrading from bee-queue due to Node.js compatibility or maintenance issues

Step-by-step guide for converting Bee-Queue projects to glide-mq. Bee-Queue uses a chained job builder pattern - this migration requires rewriting job creation and separating producer/consumer concerns.

Why Migrate

• Unmaintained - last release 2021, accumulating Node.js compatibility issues
• No cluster support - cannot scale beyond a single Redis instance
• No TLS - requires manual ioredis workarounds for encrypted connections
• No native TypeScript - community @types/bee-queue only, often outdated
• No priority queues - workaround is multiple queues
• No workflows - no parent-child jobs, no DAGs, no repeatable/cron jobs
• No rate limiting, batch processing, or broadcast
• glide-mq provides all Bee-Queue features plus 35%+ higher throughput

Breaking Changes Summary

Queue Settings Mapping

Queue Method Mapping

Event Mapping

Per-job events (job.on(...)) do not exist in glide-mq. Use QueueEvents and filter by jobId, or use queue.addAndWait() for request-reply patterns.

Step-by-Step Conversion

1. Connection

// BEFORE (Bee-Queue)
const Queue = require('bee-queue');
const queue = new Queue('tasks', {
  redis: { host: 'localhost', port: 6379 }
});

// AFTER (glide-mq)
import { Queue, Worker } from 'glide-mq';
const connection = { addresses: [{ host: 'localhost', port: 6379 }] };
const queue = new Queue('tasks', { connection });

2. Job Creation (Biggest Change)

Bee-Queue uses chained builder with no job name. glide-mq uses a single call with a required name.

// BEFORE (Bee-Queue) - chained builder, no name
const job = await queue.createJob({ email: 'user@example.com' })
  .retries(3)
  .backoff('exponential', 1000)
  .delayUntil(Date.now() + 60000)
  .setId('unique-123')
  .save();

// AFTER (glide-mq) - options object, name required
await queue.add('send-email',
  { email: 'user@example.com' },
  {
    attempts: 3,  // NOT "retries" - different name!
    backoff: { type: 'exponential', delay: 1000 },
    delay: 60000,
    jobId: 'unique-123',
  }
);

3. Worker

// BEFORE (Bee-Queue)
queue.process(10, async (job) => {
  return { processed: true };
});
queue.on('succeeded', (job, result) => console.log('Done:', result));

// AFTER (glide-mq) - separate Worker class
const worker = new Worker('tasks', async (job) => {
  return { processed: true };
}, { connection, concurrency: 10 });
worker.on('completed', (job) => console.log('Done:', job.returnValue));

4. Batch Save

// BEFORE (Bee-Queue)
const jobs = items.map(item => queue.createJob(item));
await queue.saveAll(jobs);

// AFTER (glide-mq) - each entry needs a name
await queue.addBulk(items.map(item => ({
  name: 'process',
  data: item
})));

5. Producer-Only

// BEFORE (Bee-Queue) - disable worker mode
const queue = new Queue('tasks', {
  isWorker: false, getEvents: false, sendEvents: false,
  redis: { host: 'localhost', port: 6379 }
});

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

6. Progress Reporting

// BEFORE (Bee-Queue) - arbitrary JSON
queue.process(async (job) => {
  job.reportProgress({ percent: 50, message: 'halfway' });
  return result;
});

// AFTER (glide-mq) - number (0-100) or object
const worker = new Worker('tasks', async (job) => {
  await job.updateProgress(50);
  await job.updateProgress({ page: 3, total: 10 });  // objects also supported
  await job.log('halfway done');  // structured info goes to job.log()
  return result;
}, { connection });

7. Stall Detection

// BEFORE (Bee-Queue) - manual setup required
const queue = new Queue('tasks', { stallInterval: 5000 });
queue.checkStalledJobs(5000);  // must call manually!

// AFTER (glide-mq) - automatic on Worker
const worker = new Worker('tasks', processor, {
  connection,
  lockDuration: 30000,
  stalledInterval: 30000,
  maxStalledCount: 2
});
// Stall detection runs automatically - no manual call

8. Health Check

// BEFORE (Bee-Queue)
const health = await queue.checkHealth();
// { waiting, active, succeeded, failed, delayed, newestJob }

// AFTER (glide-mq)
const counts = await queue.getJobCounts();
// { waiting, active, completed, failed, delayed }

9. Web UI (Arena to Dashboard)

// BEFORE (Bee-Queue) - Arena
const Arena = require('bull-arena');
app.use('/', Arena({ Bee: require('bee-queue'), queues: [{ name: 'tasks' }] }));

// AFTER (glide-mq) - Dashboard
import { createDashboard } from '@glidemq/dashboard';
app.use('/dashboard', createDashboard([queue]));

What You Gain

Features Bee-Queue does not have that are available after migration:

Migration Checklist

- [ ] Install glide-mq, uninstall bee-queue and @types/bee-queue
- [ ] Create connection config (addresses array format)
- [ ] Convert queue.createJob().save() to queue.add(name, data, opts)
- [ ] Add job names to every queue.add() call (Bee-Queue had none)
- [ ] Convert .retries(n) to { attempts: n } (different name!)
- [ ] Convert .backoff(strategy, delay) to { backoff: { type, delay } }
- [ ] Convert .delayUntil(date) to { delay: ms }
- [ ] Convert .setId(id) to { jobId: id }
- [ ] Convert queue.process() to new Worker()
- [ ] Convert queue.saveAll() to queue.addBulk()
- [ ] Separate producer queues (isWorker:false to Producer class)
- [ ] Convert job.reportProgress(json) to job.updateProgress(number | object)
- [ ] Remove manual checkStalledJobs() calls (automatic on Worker)
- [ ] Convert checkHealth() to getJobCounts()
- [ ] Update event listeners (queue.on to worker.on or QueueEvents)
- [ ] Convert per-job events (job.on) to QueueEvents
- [ ] Keep the project's existing module system (CommonJS or ESM)
- [ ] Run full test suite
- [ ] Confirm queue counts: await queue.getJobCounts()
- [ ] Confirm no jobs stuck in active state
- [ ] Smoke-test QueueEvents or SSE listeners if the app exposes them
- [ ] Confirm workers, queues, and connections close cleanly

Troubleshooting

Quick Start Commands

npm uninstall bee-queue @types/bee-queue
npm install glide-mq

References

Bee-Queue to glide-mq - Complete API Mapping

Method-by-method reference for converting every Bee-Queue API call to its glide-mq equivalent.

Constructor

// BEFORE
const Queue = require('bee-queue');
const queue = new Queue('tasks', {
  redis: { host: 'localhost', port: 6379 },
  prefix: 'bq',
  isWorker: true,
  getEvents: true,
  sendEvents: true,
  storeJobs: true,
  removeOnSuccess: false,
  removeOnFailure: false,
  stallInterval: 5000,
  activateDelayedJobs: true,
});

// AFTER - split into Queue + Worker
import { Queue, Worker, QueueEvents } from 'glide-mq';
const connection = { addresses: [{ host: 'localhost', port: 6379 }] };

const queue = new Queue('tasks', {
  connection,
  prefix: 'glide',
});

const worker = new Worker('tasks', processor, {
  connection,
  lockDuration: 30000,
  stalledInterval: 30000,
});

const events = new QueueEvents('tasks', { connection });

Job Creation Methods

createJob + save -> add

// BEFORE - chained builder (no job name)
const job = await queue.createJob({ x: 1 }).save();
console.log(job.id);

// AFTER - single call (name required)
const job = await queue.add('compute', { x: 1 });
console.log(job.id);

setId -> jobId option

// BEFORE
queue.createJob(data).setId('unique-key').save();

// AFTER
await queue.add('task', data, { jobId: 'unique-key' });

retries -> attempts

Name change: `retries` becomes `attempts`.

// BEFORE
queue.createJob(data).retries(3).save();

// AFTER
await queue.add('task', data, { attempts: 3 });

backoff -> backoff option

// BEFORE - immediate (default)
queue.createJob(data).retries(3).backoff('immediate').save();

// AFTER
await queue.add('task', data, {
  attempts: 3,
  backoff: { type: 'fixed', delay: 0 },
});

// BEFORE - fixed
queue.createJob(data).retries(3).backoff('fixed', 1000).save();

// AFTER
await queue.add('task', data, {
  attempts: 3,
  backoff: { type: 'fixed', delay: 1000 },
});

// BEFORE - exponential
queue.createJob(data).retries(3).backoff('exponential', 1000).save();

// AFTER
await queue.add('task', data, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
});

delayUntil -> delay

// BEFORE - absolute timestamp
queue.createJob(data).delayUntil(Date.now() + 60000).save();

// AFTER - relative milliseconds
await queue.add('task', data, { delay: 60000 });

timeout -> timeout job option

// BEFORE - per-job timeout
queue.createJob(data).timeout(30000).save();

// AFTER - per-job timeout option
await queue.add('task', data, { timeout: 30000 });

Full chained builder conversion

// BEFORE - all options chained
const job = await queue.createJob({ email: 'user@example.com' })
  .setId('email-123')
  .retries(3)
  .backoff('exponential', 1000)
  .delayUntil(Date.now() + 60000)
  .timeout(30000)
  .save();

// AFTER - single options object
const job = await queue.add('send-email',
  { email: 'user@example.com' },
  {
    jobId: 'email-123',
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    delay: 60000,
    timeout: 30000,
  }
);

Processing Methods

process -> Worker

// BEFORE - promise-based
queue.process(async (job) => {
  return { result: job.data.x * 2 };
});

// AFTER
const worker = new Worker('tasks', async (job) => {
  return { result: job.data.x * 2 };
}, { connection });

// BEFORE - with concurrency
queue.process(10, async (job) => {
  return await processJob(job);
});

// AFTER
const worker = new Worker('tasks', async (job) => {
  return await processJob(job);
}, { connection, concurrency: 10 });

// BEFORE - callback-based (deprecated pattern)
queue.process(function(job, done) {
  done(null, { result: job.data.x * 2 });
});

// AFTER - always promise-based
const worker = new Worker('tasks', async (job) => {
  return { result: job.data.x * 2 };
}, { connection });

reportProgress -> updateProgress

// BEFORE - any JSON value
queue.process(async (job) => {
  job.reportProgress({ page: 3, total: 10 });
  job.reportProgress(50);
  job.reportProgress('halfway');
  return result;
});

// AFTER - number (0-100) or object, use job.log() for text messages
const worker = new Worker('tasks', async (job) => {
  await job.updateProgress(30);
  await job.updateProgress({ page: 3, total: 10 });  // objects also supported
  await job.log('Processing page 3 of 10');
  await job.updateProgress(50);
  return result;
}, { connection });

Bulk Operations

saveAll -> addBulk

// BEFORE
const jobs = [
  queue.createJob({ x: 1 }),
  queue.createJob({ x: 2 }),
  queue.createJob({ x: 3 }),
];
const errors = await queue.saveAll(jobs);
// errors is Map<Job, Error>

// AFTER
const results = await queue.addBulk([
  { name: 'compute', data: { x: 1 } },
  { name: 'compute', data: { x: 2 } },
  { name: 'compute', data: { x: 3 } },
]);

Query Methods

getJob

// BEFORE
const job = await queue.getJob('42');

// AFTER - same API
const job = await queue.getJob('42');

getJobs

// BEFORE - type + page object
const waiting = await queue.getJobs('waiting', { start: 0, end: 25 });
const failed = await queue.getJobs('failed', { size: 100 });

// AFTER - type + start + end
const waiting = await queue.getJobs('waiting', 0, 25);
const failed = await queue.getJobs('failed', 0, 100);

removeJob

// BEFORE - by ID on queue
await queue.removeJob('42');

// AFTER - via Job instance
const job = await queue.getJob('42');
await job.remove();

checkHealth -> getJobCounts

// BEFORE
const health = await queue.checkHealth();
// { waiting: 5, active: 2, succeeded: 100, failed: 3, delayed: 1, newestJob: '108' }

// AFTER
const counts = await queue.getJobCounts();
// { waiting: 5, active: 2, completed: 100, failed: 3, delayed: 1 }
// Note: "succeeded" renamed to "completed", no "newestJob"

Lifecycle Methods

close

// BEFORE
await queue.close(30000);

// AFTER - close individual components
await worker.close();
await queue.close();
await events.close();

// OR - graceful shutdown (registers SIGTERM/SIGINT, blocks until signal)
import { gracefulShutdown } from 'glide-mq';
const handle = gracefulShutdown([worker, queue, events]);
// For programmatic shutdown: await handle.shutdown();

destroy -> obliterate

// BEFORE
await queue.destroy();

// AFTER
await queue.obliterate();

ready

// BEFORE
await queue.ready();

// AFTER
await worker.waitUntilReady();

isRunning

// BEFORE
queue.isRunning();

// AFTER
worker.isRunning();

Stall Detection

// BEFORE - manual setup, repeated call required
queue.checkStalledJobs(5000, (err, numStalled) => {
  console.log('Stalled:', numStalled);
});

// AFTER - automatic, configured on Worker
const worker = new Worker('tasks', processor, {
  connection,
  lockDuration: 30000,     // how long a job can run before considered stalled
  stalledInterval: 30000,  // how often to check for stalled jobs
  maxStalledCount: 2,      // re-queue up to 2 times before failing
});

worker.on('stalled', (jobId) => {
  console.log('Stalled:', jobId);
});

Event Migration

Local events (Queue -> Worker)

// BEFORE
queue.on('succeeded', (job, result) => {});
queue.on('failed', (job, err) => {});
queue.on('retrying', (job, err) => {});
queue.on('stalled', (jobId) => {});
queue.on('error', (err) => {});

// AFTER
worker.on('completed', (job, result) => {});
worker.on('failed', (job, err) => {});
// No separate 'retrying' event - failed fires for all failures
worker.on('stalled', (jobId) => {});
worker.on('error', (err) => {});

PubSub events (Queue -> QueueEvents)

// BEFORE
queue.on('job succeeded', (jobId, result) => {});
queue.on('job failed', (jobId, err) => {});
queue.on('job progress', (jobId, data) => {});

// AFTER
const events = new QueueEvents('tasks', { connection });
events.on('completed', ({ jobId, returnvalue }) => {});
events.on('failed', ({ jobId, failedReason }) => {});
events.on('progress', ({ jobId, data }) => {});

Per-job events (Job -> QueueEvents)

// BEFORE
const job = await queue.createJob(data).save();
job.on('succeeded', (result) => console.log('Done:', result));
job.on('failed', (err) => console.error('Failed:', err));
job.on('progress', (p) => console.log('Progress:', p));

// AFTER - filter by jobId in QueueEvents
const job = await queue.add('task', data);
const events = new QueueEvents('tasks', { connection });
events.on('completed', ({ jobId, returnvalue }) => {
  if (jobId === job.id) console.log('Done:', returnvalue);
});

// OR - use addAndWait for request-reply
const result = await queue.addAndWait('task', data, { waitTimeout: 30000 });

Custom Backoff Strategies

// BEFORE
queue.backoffStrategies.set('linear', (job) => {
  return job.options.backoff.delay * (job.options.retries + 1);
});
queue.createJob(data).retries(5).backoff('linear', 1000).save();

// AFTER
const worker = new Worker('tasks', processor, {
  connection,
  backoffStrategies: {
    linear: (attemptsMade) => attemptsMade * 1000,
  },
});
await queue.add('task', data, {
  attempts: 5,
  backoff: { type: 'linear', delay: 1000 },
});

Connection Formats

// BEFORE - object
new Queue('tasks', { redis: { host: 'redis.example.com', port: 6380 } });

// BEFORE - URL string
new Queue('tasks', { redis: 'redis://user:pass@host:6379/0' });

// BEFORE - existing ioredis client
const Redis = require('ioredis');
new Queue('tasks', { redis: new Redis() });

// AFTER - always addresses array
const connection = { addresses: [{ host: 'redis.example.com', port: 6380 }] };

// AFTER - with TLS
const connection = { addresses: [{ host: 'redis.example.com', port: 6380 }], useTLS: true };

// AFTER - cluster mode
const connection = {
  addresses: [
    { host: 'node1', port: 7000 },
    { host: 'node2', port: 7001 },
  ],
  clusterMode: true,
};

Graceful Shutdown

// BEFORE
async function shutdown() {
  await queue.close(30000);
  process.exit(0);
}
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

// AFTER - gracefulShutdown registers SIGTERM/SIGINT automatically
import { gracefulShutdown } from 'glide-mq';
const handle = gracefulShutdown([worker, queue, events]);
// Blocks until signal fires. For programmatic: await handle.shutdown()

New Features Available After Migration

Everything Bee-Queue cannot do that glide-mq provides out of the box.

Priority Queues

Bee-Queue has no priority support. glide-mq uses numeric priority where lower = higher priority (0 is the highest, default).

// High priority (processed first)
await queue.add('urgent-alert', data, { priority: 0 });

// Normal priority
await queue.add('report', data, { priority: 5 });

// Low priority (processed last)
await queue.add('cleanup', data, { priority: 20 });

Processing order: priority > LIFO > FIFO.

Job Workflows (FlowProducer)

Parent-child job trees and DAG workflows. The parent waits for all children to complete.

import { FlowProducer } from 'glide-mq';

const flow = new FlowProducer({ connection });
await flow.add({
  name: 'assemble-report',
  queueName: 'reports',
  data: { reportId: 42 },
  children: [
    { name: 'fetch-users', queueName: 'data', data: { source: 'users' } },
    { name: 'fetch-orders', queueName: 'data', data: { source: 'orders' } },
    { name: 'fetch-metrics', queueName: 'data', data: { source: 'metrics' } },
  ],
});

Broadcast (Fan-Out)

Bee-Queue is point-to-point only. glide-mq supports fan-out where every subscriber receives every message.

import { Broadcast, BroadcastWorker } from 'glide-mq';

const broadcast = new Broadcast('events', { connection, maxMessages: 1000 });

// Every subscriber gets the message
const inventory = new BroadcastWorker('events', async (job) => {
  await updateInventory(job.data);
}, { connection, subscription: 'inventory-service' });

const email = new BroadcastWorker('events', async (job) => {
  await sendNotification(job.data);
}, { connection, subscription: 'email-service' });

await broadcast.publish('orders', { event: 'order.placed', orderId: 42 });

Batch Processing

Process multiple jobs in a single handler call for I/O-bound operations.

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

const worker = new Worker('bulk-insert', async (jobs) => {
  // jobs is Job[] when batch is enabled
  const results = await db.insertMany(jobs.map(j => j.data));
  return results; // must return R[] with length === jobs.length
}, {
  connection,
  batch: { size: 50, timeout: 1000 },
});

Deduplication

Prevent duplicate job processing with three modes.

// Simple - reject if job with same deduplication ID exists
await queue.add('task', data, {
  deduplication: { id: 'unique-key' },
});

// Throttle - reject duplicates within a time window
await queue.add('task', data, {
  deduplication: { id: 'user-123', ttl: 60000 },
});

Schedulers (Cron and Interval)

Bee-Queue has no repeatable jobs. glide-mq supports cron patterns and fixed intervals.

// Cron - run every day at midnight
await queue.upsertJobScheduler(
  'daily-report',
  { pattern: '0 0 * * *' },
  { name: 'daily-report', data: {} },
);

// Interval - run every 5 minutes
await queue.upsertJobScheduler(
  'health-check',
  { every: 300000 },
  { name: 'health-check', data: {} },
);

Rate Limiting

Global and per-group rate limits on workers.

const worker = new Worker('api-calls', processor, {
  connection,
  limiter: {
    max: 100,       // max 100 jobs
    duration: 60000, // per minute
  },
});

Dead Letter Queue

Route permanently-failed jobs to a separate queue for inspection.

const worker = new Worker('tasks', processor, {
  connection,
  deadLetterQueue: { name: 'failed-jobs' },
});

LIFO Mode

Process newest jobs first instead of FIFO.

await queue.add('urgent-report', data, { lifo: true });

Job TTL

Automatically fail jobs that are not processed within a time window.

await queue.add('time-sensitive', data, { ttl: 300000 }); // 5 min expiry

Per-Key Ordering

Process jobs sequentially per ordering key while maintaining parallelism across keys.

await queue.add('process-order', data, { ordering: { key: 'customer-123' } });
await queue.add('process-order', data, { ordering: { key: 'customer-456' } });
// Jobs for customer-123 run sequentially; customer-456 runs in parallel

Request-Reply

Wait for a worker result in the producer without polling.

const result = await queue.addAndWait('inference', { prompt: 'Hello' }, {
  waitTimeout: 30000,
});
console.log(result); // processor return value

Step Jobs (Pause and Resume)

Pause a job and resume it later without completing.

const worker = new Worker('drip-campaign', async (job) => {
  if (job.data.step === 'send') {
    await sendEmail(job.data);
    return job.moveToDelayed(Date.now() + 86400000, 'check');
  }
  if (job.data.step === 'check') {
    return await checkOpened(job.data) ? 'done' : job.moveToDelayed(Date.now() + 3600000, 'followup');
  }
  await sendFollowUp(job.data);
  return 'done';
}, { connection });

UnrecoverableError

Skip all retries and fail permanently.

import { UnrecoverableError } from 'glide-mq';

const worker = new Worker('tasks', async (job) => {
  if (!job.data.requiredField) {
    throw new UnrecoverableError('missing required field');
  }
  return processJob(job);
}, { connection });

Serverless Producer

Lightweight producer with no EventEmitter overhead for Lambda/Edge.

import { Producer } from 'glide-mq';

export async function handler(event) {
  const producer = new Producer('queue', { connection });
  await producer.add('process', event.body);
  await producer.close();
  return { statusCode: 200 };
}

Testing Without Valkey

In-memory queue and worker for unit tests.

import { TestQueue, TestWorker } from 'glide-mq/testing';

const queue = new TestQueue('tasks');
await queue.add('test-job', { key: 'value' });
const worker = new TestWorker(queue, async (job) => {
  return { processed: true };
});
await worker.run();

Cluster Support

Native Valkey/Redis Cluster with hash-tagged keys.

const connection = {
  addresses: [
    { host: 'node1', port: 7000 },
    { host: 'node2', port: 7001 },
  ],
  clusterMode: true,
  readFrom: 'AZAffinity',
  clientAz: 'us-east-1a',
};

TLS and IAM Authentication

// TLS
const connection = {
  addresses: [{ host: 'redis.example.com', port: 6380 }],
  useTLS: true,
};

// AWS IAM
const connection = {
  addresses: [{ host: 'cluster.cache.amazonaws.com', port: 6379 }],
  clusterMode: true,
  credentials: {
    type: 'iam',
    serviceType: 'elasticache',
    region: 'us-east-1',
    userId: 'my-iam-user',
    clusterName: 'my-cluster',
  },
};

QueueEvents (Real-Time Stream)

Centralized job lifecycle events via Valkey Streams - replaces Bee-Queue's PubSub model.

import { QueueEvents } from 'glide-mq';

const events = new QueueEvents('tasks', { connection });
events.on('added', ({ jobId }) => console.log('added', jobId));
events.on('completed', ({ jobId, returnvalue }) => console.log('done', jobId));
events.on('failed', ({ jobId, failedReason }) => console.log('failed', jobId));
events.on('progress', ({ jobId, data }) => console.log('progress', jobId, data));
events.on('stalled', ({ jobId }) => console.log('stalled', jobId));

Time-Series Metrics

Per-minute throughput and latency data with zero extra round trips.

const metrics = await queue.getMetrics('completed');
// { count, data: [{ timestamp, count, avgDuration }], meta: { resolution: 'minute' } }

Queue Management

// Pause/resume all workers
await queue.pause();
await queue.resume();

// Drain waiting jobs
await queue.drain();

// Clean old completed/failed jobs
await queue.clean(3600000, 1000, 'completed'); // older than 1 hour

// Obliterate all queue data
await queue.obliterate({ force: true });

Dashboard

Web UI for monitoring and managing queues.

import { createDashboard } from '@glidemq/dashboard';
import express from 'express';

const app = express();
app.use('/dashboard', createDashboard([queue]));

Framework Integrations

Native integrations for Hono, Fastify, NestJS, and Hapi.

OpenTelemetry

Automatic span emission for distributed tracing.

Pluggable Serializers

Custom serialization for job data (e.g., MessagePack, Protocol Buffers).

const queue = new Queue('tasks', { connection, serializer: customSerializer });
const worker = new Worker('tasks', processor, { connection, serializer: customSerializer });

AI-Native Primitives

glide-mq is purpose-built for LLM/AI orchestration. None of these exist in Bee-Queue.

Usage Metadata

Track model, tokens, cost, and latency per job.

await job.reportUsage({
  model: 'gpt-5.4',
  provider: 'openai',
  tokens: { input: 500, output: 200 },
  costs: { total: 0.003 },
  costUnit: 'usd',
  latencyMs: 800,
});

Token Streaming

Stream LLM output tokens in real-time via per-job Valkey Streams.

// Worker: emit chunks
await job.stream({ token: 'Hello' });

// Consumer: read chunks (supports long-polling)
const entries = await queue.readStream(jobId, { block: 5000 });

Suspend / Resume (Human-in-the-Loop)

Pause a job for external approval, resume with signals.

await job.suspend({ reason: 'Needs review', timeout: 86_400_000 });
// Externally:
await queue.signal(jobId, 'approve', { reviewer: 'alice' });

Flow Budget

Cap total tokens/cost across all jobs in a workflow flow.

await flow.add(flowTree, {
  budget: { maxTotalTokens: 50_000, maxTotalCost: 0.50, costUnit: 'usd' },
});

Fallback Chains

Ordered model/provider alternatives on retryable failure.

await queue.add('inference', data, {
  attempts: 4,
  fallbacks: [
    { model: 'gpt-5.4', provider: 'openai' },
    { model: 'claude-sonnet-4-20250514', provider: 'anthropic' },
    { model: 'llama-3-70b', provider: 'groq' },
  ],
});

Dual-Axis Rate Limiting (RPM + TPM)

Rate-limit by both requests and tokens per minute for LLM API compliance.

const worker = new Worker('inference', processor, {
  connection,
  limiter: { max: 60, duration: 60_000 },
  tokenLimiter: { maxTokens: 100_000, duration: 60_000 },
});

Flow Usage Aggregation

Aggregate AI usage across all jobs in a flow.

const usage = await queue.getFlowUsage(parentJobId);
// { tokens, totalTokens, costs, totalCost, costUnit, jobCount, models }

Vector Search

KNN similarity search over job hashes via Valkey Search.

await queue.createJobIndex({
  vectorField: { name: 'embedding', dimensions: 1536 },
});
const job = await queue.add('document', { text: 'Hello world' });
if (job) {
  await job.storeVector('embedding', queryEmbedding);
}
const results = await queue.vectorSearch(queryEmbedding, { k: 10 });