QueueKit

Runtime-neutral TypeScript infrastructure for job queues, message queues, pub/sub, and streams—without erasing provider-native capabilities and types.

QueueKit supports Node.js 20+, Bun, and Deno 2+ for its runtime-neutral core. Version 0.1 ships adapters for BullMQ, Kafka, RabbitMQ, Redis (Pub/Sub and Streams), NATS Core, and Amazon SQS. Each SDK is an optional peer dependency and loads only when that provider is created.

Install

npm install @mohamedhabibwork/queuekit
# Add only the provider SDK you use, for example:
npm install kafkajs

bun add @mohamedhabibwork/queuekit kafkajs

deno add npm:@mohamedhabibwork/queuekit
deno add npm:kafkajs

Usage

import { createQueue } from '@mohamedhabibwork/queuekit';

const kafka = await createQueue({
  type: 'kafka',
  clientId: 'orders-api',
  brokers: ['localhost:9092'],
});

await kafka.publish('orders.created', {
  payload: { orderId: 'ord_123' },
}, {
  native: { partition: 2, headers: { source: 'api' } },
});

Provider-native options are intentionally kept under native; BullMQ options cannot accidentally be passed to Kafka and vice versa.

import { createQueueManager } from '@mohamedhabibwork/queuekit';

const queues = createQueueManager({
  default: 'jobs',
  providers: {
    jobs: { type: 'bullmq', connection: { host: 'localhost', port: 6379 } },
    events: { type: 'kafka', clientId: 'api', brokers: ['localhost:9092'] },
  },
});

await (await queues.provider('jobs')).publish('emails', {
  type: 'welcome',
  payload: { userId: 'u_1' },
}, { native: { attempts: 5, backoff: { type: 'exponential', delay: 1_000 } } });

Providers

| Provider | Entrypoint | Family | | --- | --- | --- | | BullMQ | @mohamedhabibwork/queuekit/bullmq | Job queue | | Kafka | @mohamedhabibwork/queuekit/kafka | Event stream | | RabbitMQ | @mohamedhabibwork/queuekit/rabbitmq | Message queue | | Redis | @mohamedhabibwork/queuekit/redis | Pub/Sub or stream | | NATS | @mohamedhabibwork/queuekit/nats | Pub/Sub; JetStream publishing | | Amazon SQS | @mohamedhabibwork/queuekit/sqs | Message queue |

See the documentation site for capabilities and acknowledgement semantics. BullMQ no longer bundles a Redis client: when connection is a URL string, also install ioredis (npm install bullmq ioredis). The Redis provider works with both node-redis (default) and ioredis (client: 'ioredis'), against Redis and Valkey servers, and Streams consumers can dead-letter rejected entries with native.deadLetter.

Custom providers and tests

Tests can use the built-in memory fake driver, which supports store-and-forward delivery, delays, retries, acknowledgements, and dead letters with deterministic controls:

import { createFakeQueue } from '@mohamedhabibwork/queuekit/testing';

const testQueue = createFakeQueue();
await testQueue.publish('emails', { type: 'welcome', payload: { email: '[email protected]' } });

await testQueue.waitUntilIdle();                    // await all in-flight handler work
await testQueue.flush();                            // force delayed messages out immediately
testQueue.pause(); testQueue.resume();              // hold and release delivery
testQueue.pending('emails');                        // queued, unacknowledged messages
testQueue.deadLetters('emails');                    // rejected / exhausted messages
testQueue.failNext(new Error('broker down'));       // inject the next publish failure

The fake is also a first-class driver, so createQueue and createQueueManager can point at it with the same config shape used in production:

import { createQueueManager } from '@mohamedhabibwork/queuekit';

const manager = createQueueManager({ providers: { jobs: { type: 'memory' } }, default: 'jobs' });

Custom providers are declared with defineQueueProvider:

import { defineQueueProvider } from '@mohamedhabibwork/queuekit/custom';
import { createMemoryQueue } from '@mohamedhabibwork/queuekit/testing';

const testQueue = createMemoryQueue();

const provider = defineQueueProvider({
  name: 'internal' as const,
  capabilities: { kind: 'queue', publish: true, consume: false },
  async create(config: { endpoint: string }) {
    // Return a QueueProvider implemented entirely against public QueueKit types.
    return testQueue;
  },
});

Guarantees

QueueKit does not claim universal exactly-once delivery. Delivery, retry, ordering, acknowledgement, and dead-letter behavior are broker-specific; use capabilities, provider-native configuration, and idempotent consumers. message.metadata is application-only and is never implicitly sent to a broker.

Development

npm install --legacy-peer-deps
npm run check
npm pack --dry-run

The CI matrix tests Node 20, 22, 24, and 26; Bun; Deno; and TypeScript 5.9, 6, and 7. Local Node 22 and Bun checks are run before release; Deno is covered in GitHub Actions when it is not installed locally.

End-to-end tests

tests/e2e runs every real driver — Redis (pub/sub and streams), RabbitMQ, Kafka, NATS, SQS, and BullMQ — through a produce → consume → acknowledge round-trip against live brokers:

docker compose up -d   # redis, rabbitmq, kafka, nats, localstack (SQS)
npm test               # e2e suites run when a broker is reachable and skip otherwise

Broker endpoints can be overridden with the QUEUEKIT_E2E_REDIS_URL, QUEUEKIT_E2E_RABBITMQ_URL, QUEUEKIT_E2E_KAFKA_BROKER, QUEUEKIT_E2E_NATS_URL, and QUEUEKIT_E2E_SQS_ENDPOINT environment variables. The compose stack uses non-default local ports (Redis 6390, RabbitMQ 5673) so it never collides with an already-running native broker.

Publishing

Publishing runs only from the Release workflow. Add an npm automation token as the repository Actions secret NPM_TOKEN; a local .env file cannot be read by GitHub-hosted runners. Details are in the publishing guide.

License

MIT